pending_writes_loop.go

// SPDX-License-Identifier: Apache-2.0

package git

import (
	"errors"
	"fmt"
	"slices"

	"github.com/ConfigButler/gitops-reverser/internal/manifestanalyzer"
	itypes "github.com/ConfigButler/gitops-reverser/internal/types"
)

// This file is the branch worker's ordered pending writes, and the one driver that turns them into
// commits and publications. See docs/design/gittarget-branch-worker-pending-writes.md.
//
// Every write the loop makes, whatever produced it (a closed window, a resync, a save's empty
// record), takes one path:
//
//   - decide appends it to the pending writes. The write gates and building the write have passed, so the
//     decision stands from here on. Deciding does nothing else: no commit, no push, no settlement.
//   - advance is the one driver, run once at the end of every wake of the loop. It commits what was
//     decided, schedules the push, and closes the outage once no pending writes remain.
//   - materialize makes the checkout the projection of the pending writes: the remote tip the writes were
//     planned on plus one commit for each, in order. It rebuilds the writes committed before when
//     the checkout is behind them, then commits each decided write not committed yet.
//   - materialize classifies each attempt in one place: committed (settleCommitted); failed for
//     good, because the write itself was refused or cannot be made (settleFailed); or left pending,
//     because the remote could not be reached. A write left pending needs no settling, only its
//     waiting resync callers an answer (replyToDeferredResyncs).
//
// Settling an outcome never decides more work directly: a refused write leaves a reapply request
// due to its applier (reapplyForRefusal), which is sent after a flush, outside the
// pending writes.

// errWriteFailed marks a failure of the write itself, as opposed to reaching the remote or reading
// the GitProvider: the plan was refused or could not be applied. It is terminal for the write, while
// any other commit failure leaves a decided write pending for the retry. Match it with errors.Is.
var errWriteFailed = errors.New("the write failed")

// writeFailedError carries errWriteFailed without changing the message of the error it wraps.
type writeFailedError struct{ err error }

func (e writeFailedError) Error() string        { return e.err.Error() }
func (e writeFailedError) Unwrap() error        { return e.err }
func (e writeFailedError) Is(target error) bool { return target == errWriteFailed }

// writeOrigin is what a write's outcome is settled against: the caller waiting on a resync. It is a
// process object, kept beside the write's data rather than in it.
type writeOrigin struct {
	resync *ResyncRequest
	// answered records that the resync's caller was answered already: the remote held it back,
	// and it is applied later with nobody waiting.
	answered bool
}

// errAdmissionClosed answers a resync refused at admission. It is ErrFinalizeQueueFull to every
// caller: the work was not accepted, and the producer that sent it keeps it.
var errAdmissionClosed = fmt.Errorf(
	"%w: the branch has paused intake while the remote cannot take what it holds", ErrFinalizeQueueFull)

// pendingWriteOverheadBytes is what every retained item is charged beyond its payload: each event a
// window or a write keeps (its identity, author, path), and a write that keeps none (its target
// metadata, commit configuration, signer, the save it carries). Without it an item with no payload,
// a delete or a save's empty record, cost nothing to keep, and a branch
// whose remote was down admitted them for as long as the outage lasted.
const pendingWriteOverheadBytes = 1024

// retainedCharge is what keeping a decided write is charged against the retained-byte budget: its
// payload, its message, and one item's overhead for each event it keeps, or for itself when it
// keeps none. That is never more than the window it closed or the save it records was charged, so
// deciding a write never takes what the branch holds past what intake admitted. It is fixed when
// the write is decided, so the refund when it stops being pending matches it whatever the write's later
// rendering changes.
func retainedCharge(pendingWrite *PendingWrite) int64 {
	items := int64(max(len(pendingWrite.Events), 1))
	return pendingWrite.ByteSize + int64(len(pendingWrite.CommitMessage)) + items*pendingWriteOverheadBytes
}

// decide appends a write the loop has decided to make to the pending writes, and charges what
// keeping it costs.
// A request riding it is held from here on, so the controller never fails a save whose write can
// still land. Committing it is advance's job, at the end of the wake.
func (l *branchWorkerEventLoop) decide(pendingWrite PendingWrite) {
	l.decisions++
	pendingWrite.decision = l.decisions
	pendingWrite.charge = retainedCharge(&pendingWrite)
	pendingWrite.decidedAt = l.w.now()
	l.pendingWrites = append(l.pendingWrites, pendingWrite)
	l.pendingWritesBytes += pendingWrite.charge
	if id := pendingWrite.CommitRequest; id != nil {
		l.w.setCommitRequestPhase(*id, PhaseWaitingForPush)
		// Bound to its write from here on: see pendingCommitRequest.attached.
		if pcr := l.pendingCRs[*id]; pcr != nil {
			pcr.attached = true
		}
	}
}

// advance drives the pending writes once the wake's handlers have decided what they decide: it
// commits them, schedules their push, and ends the outage once none remain. It is the only
// caller of materialize outside a publication, and the only place where having no pending writes
// closes the retry and parent recovery, so no exit that settles the last one can leave intake
// paused.
//
// A push that lands settles the writes a replay refused. The loop ends once every decided write is
// committed and its push is scheduled.
func (l *branchWorkerEventLoop) advance() {
	for {
		l.commitDecided()
		if len(l.pendingWrites) == 0 {
			l.stopPushTimer()
			l.endOutage()
			return
		}
		l.maybeSchedulePush()
		if l.retry.pending() || l.materializedPrefix() == len(l.pendingWrites) {
			return
		}
	}
}

// commitDecided commits every decided write not committed yet. Deciding is not committing: a remote
// that cannot be reached leaves the writes pending for the retry. Until a pending retry is due, a
// commit that would need a connection waits for that retry instead of spending one. A resync always
// needs one, and its caller is waiting to hear what it found: it is answered at once with the
// failure the retry is waiting out, and stays pending to be applied when the retry is due
// (replyToDeferredResyncs).
func (l *branchWorkerEventLoop) commitDecided() {
	if l.materializedPrefix() == len(l.pendingWrites) {
		return
	}
	if l.awaitingRetry() && !l.materializeIsLocal() && !l.w.awaitingParentProbe() {
		l.replyToDeferredResyncs(l.retry.cause)
		return
	}
	if err := l.materialize(); err != nil {
		l.noteParentUnavailable(err)
		if !l.retry.pending() {
			l.scheduleRetry(err)
		}
		l.w.Log.Error(err, "Cannot commit the decided writes yet; they stay pending for the retry",
			"pendingWrites", len(l.pendingWrites), "retryAt", l.retry.due)
	}
}

// materialize makes the checkout the projection of the pending writes, committing every one not
// committed yet, including any that settling one decides. It returns the error that stopped it when
// the remote could not be reached; the writes it could not commit stay pending.
func (l *branchWorkerEventLoop) materialize() error {
	for {
		i := l.materializedPrefix()
		// A resync judges its snapshot against the newest remote tree, so it fetches first whatever
		// the checkout holds.
		refetch := ""
		if i < len(l.pendingWrites) && l.pendingWrites[i].Kind == PendingWriteResync {
			refetch = fetchReasonForcedRecheck
		}
		tipBefore := l.w.inbox.lastTip()
		if err := l.materializePrefix(refetch); err != nil {
			l.replyToDeferredResyncs(err)
			l.w.recordMaterializationFailure(err)
			return err
		}
		// Read again: a rebuild settles the writes the new tree refuses out of the prefix.
		i = l.materializedPrefix()
		if i == len(l.pendingWrites) {
			return nil
		}
		if refetch != "" && l.discardIfFetchFoundForeign(i, tipBefore) {
			continue
		}
		// A write is planned under the policy in force when it is committed, exactly as a replay is:
		// an operator who tightened pruning since it was decided is obeyed.
		if err := l.w.tightenPendingPruneModes(l.w.ctx, l.pendingWrites[i:i+1]); err != nil {
			l.replyToDeferredResyncs(err)
			l.w.recordMaterializationFailure(err)
			return err
		}
		err := l.w.commitPendingWrites(l.pendingWrites[i : i+1])
		switch {
		case err == nil:
			l.settleCommitted(i)
		case errors.Is(err, errWriteFailed):
			l.settleFailed(i, err)
		default:
			l.replyToDeferredResyncs(err)
			l.w.recordMaterializationFailure(err)
			return err
		}
	}
}

// discardIfFetchFoundForeign discards the resync at i when the fetch it was planned behind found a
// commit the worker did not push: the cluster does not run it yet, and it holds what the snapshot is
// about to be judged against (docs/design/applier-echoes.md, F3). tipBefore is the tip observed before
// that fetch. The resync is decided and not committed, so nothing is committed on it.
func (l *branchWorkerEventLoop) discardIfFetchFoundForeign(i int, tipBefore string) bool {
	req := l.pendingWrites[i].origin.resync
	if req == nil || l.w.inbox.lastTip() == tipBefore || !l.w.inbox.foreignPending() || !l.snapshotsWait(req) {
		return false
	}
	answered := l.pendingWrites[i].origin.answered
	l.removeAt(i)
	if answered {
		// Its caller heard already that the remote held it back; it is only asked for again.
		req = &ResyncRequest{GitTargetName: req.GitTargetName, GitTargetNamespace: req.GitTargetNamespace,
			Scope: req.Scope}
	}
	l.discardSnapshot(req, snapshotForeignFetch)
	return true
}

// materializePrefix rebuilds the writes committed before when the checkout is behind them: a write
// failed part-way and could not be undone, a reset discarded their commits and the replay did not
// finish, or their root was chosen under an older parent configuration. A parent change is read
// from the generations, not taken from a flag: a rebuild that fails leaves them apart, so the next
// attempt rebuilds again.
//
// refetch, when set, is the fetch series of a caller that needs the remote tip whatever the checkout
// holds. With nothing committed and no refetch there is nothing to project: commitPendingWrites' own
// base check fetches when the base is untrusted or the worktree dirty, which it can do safely
// because nothing is lost.
func (l *branchWorkerEventLoop) materializePrefix(refetch string) error {
	m := l.materializedPrefix()
	if refetch == "" && (m == 0 || l.w.checkoutHolds(m) && !l.w.rootParentStale()) {
		return nil
	}
	// The rebuild fetches, and a parent known to be missing is not fetched again before its probe
	// is due: every commit path comes through here, so without the check each write arriving in the
	// meantime would cost a connection.
	if l.w.awaitingParentProbe() {
		return errAwaitingParentProbe
	}
	if m == 0 {
		// Nothing to replay, so fetch and reset directly. Leaving it to ensureBaseForCycle would
		// work, but that call records `publication`, and a resync is not one.
		return l.w.syncWithRemote(l.w.ctx, refetch)
	}
	reason := refetch
	if reason == "" {
		// The same series the nothing-committed case records in ensureBaseForCycle: this is one
		// event, and which half of it an operator sees must not depend on whether a push happened
		// to be in cooldown at the time.
		reason = fetchReasonRecovery
		l.w.Log.Info("Rebuilding retained writes onto the remote tip",
			"pendingWrites", m,
			"worktreeDirty", l.w.worktreeDirty(),
			"committedWrites", l.w.checkoutApplied.Load())
	}
	if err := l.w.refreshRemoteAndRebuildPendingWrites(l.w.ctx, l.pendingWrites[:m], reason); err != nil {
		return err
	}
	l.settleReplayRefusals(l.takeReplayRefusals())
	return nil
}

// materializedPrefix is how many writes at the head of the pending writes have been committed before. Writes
// not committed yet are always the tail: they are decided in order and committed in order.
func (l *branchWorkerEventLoop) materializedPrefix() int {
	for i := range l.pendingWrites {
		if !l.pendingWrites[i].committedOnce {
			return i
		}
	}
	return len(l.pendingWrites)
}

// materializeIsLocal reports whether committing the uncommitted pending writes now needs no
// connection: the checkout already holds the writes before them, on a base the worker can vouch for,
// and none of them is a resync, which fetches whatever the checkout holds.
func (l *branchWorkerEventLoop) materializeIsLocal() bool {
	m := l.materializedPrefix()
	for i := m; i < len(l.pendingWrites); i++ {
		if l.pendingWrites[i].Kind == PendingWriteResync {
			return false
		}
	}
	if m > 0 {
		return l.w.checkoutHolds(m) && !l.w.rootParentStale()
	}
	return l.w.baseTrusted() && !l.w.worktreeDirty()
}

// checkoutCurrent reports whether the checkout is the projection of every pending write, so it may be
// pushed as it stands.
func (l *branchWorkerEventLoop) checkoutCurrent() bool {
	m := l.materializedPrefix()
	if m < len(l.pendingWrites) {
		return false
	}
	return m == 0 || l.w.checkoutHolds(m) && !l.w.rootParentStale()
}

// removeAt takes the write at i out of the pending writes. The slot it vacates is cleared, so the backing array
// does not keep the write's payload reachable once the budget has refunded it.
func (l *branchWorkerEventLoop) removeAt(i int) PendingWrite {
	pendingWrite := l.pendingWrites[i]
	l.pendingWrites = slices.Delete(l.pendingWrites, i, i+1)
	l.pendingWritesBytes -= pendingWrite.charge
	return pendingWrite
}

// takeReplayRefusals takes out of the pending writes every one the last completed replay refused,
// and returns them in order. The checkout holds none of them (replayPendingWrites), so after this it
// is again the projection of the committed prefix.
func (l *branchWorkerEventLoop) takeReplayRefusals() []PendingWrite {
	var refused []PendingWrite
	for i := 0; i < len(l.pendingWrites); {
		if l.pendingWrites[i].replayRefusal == nil {
			i++
			continue
		}
		refused = append(refused, l.removeAt(i))
	}
	return refused
}

// settleReplayRefusals settles writes a replay refused, the way a refusal at first commit is settled:
// the refusal is reported, a request riding the write fails, and the write is gone. Taking them out
// of the pending writes is a separate step (takeReplayRefusals), because the push path must do it before it
// resolves the writes it published, and settling one can decide more work.
func (l *branchWorkerEventLoop) settleReplayRefusals(refused []PendingWrite) {
	for _, pendingWrite := range refused {
		l.settleDropped(pendingWrite, pendingWrite.replayRefusal)
	}
}

// settleCommitted acts on a write whose commit was made. A resync answers its caller with what it
// found. A write whose plan left no action, a resync or a window alike, stops being pending: it is
// neither retained nor pushed, so a replay can never plan it onto a newer tip
// (docs/design/applier-echoes.md, F2).
func (l *branchWorkerEventLoop) settleCommitted(i int) {
	l.pendingWrites[i].committedOnce = true
	pendingWrite := l.pendingWrites[i]
	switch {
	case pendingWrite.origin.resync != nil:
		l.settleResyncApplied(i)
	case !pendingWrite.retained():
		l.removeAt(i)
	}
}

// settleResyncApplied answers a resync whose plan was accepted.
func (l *branchWorkerEventLoop) settleResyncApplied(i int) {
	req := l.pendingWrites[i].origin.resync
	answered := l.pendingWrites[i].origin.answered
	stats := l.pendingWrites[i].ResyncStats
	committed := l.pendingWrites[i].retained()
	if stats.Created > 0 {
		// An object a resync adds to Git runs on the cluster already, and its applier never applied
		// it: it is adopted (S2 in docs/design/applier-echoes.md).
		l.adopt(req.GitTargetName, req.GitTargetNamespace, fmt.Sprintf("%d objects", stats.Created))
	}
	if !committed {
		l.removeAt(i)
	}
	// The plan was accepted, so the refusal standing over THIS SCOPE is over: the observation the
	// last reapply request covered is forgotten and any request still queued for it is cancelled.
	// This is the recovery path the dedupe depends on: a per-type reconcile runs after the applier
	// reverts the edit, and it is what makes the NEXT refusal, including a re-made byte-identical
	// one, a new trigger rather than a repeat. Scoped to the collection this request evaluated,
	// because a ConfigMap resync succeeding is no evidence about a Deployment that is still refused.
	l.refusalRecovered(
		itypes.NewResourceReference(req.GitTargetName, req.GitTargetNamespace), req.refusalCollection())
	// Recorded here, on the loop and before the caller hears anything, rather than only by the
	// caller once it has the answer: the next item off the queue may be a live event, and it must
	// meet the gate this resync produced. See recordReplayClean.
	l.recordReplayClean(req)
	// Its snapshot holds what the live events of its collection held before it did.
	l.supersedeHeld(req)
	l.w.Log.Info("Resync request applied",
		"committed", committed,
		"created", stats.Created,
		"updated", stats.Updated,
		"deleted", stats.Deleted,
		"skipped", stats.Skipped,
		"placementSkipped", stats.PlacementSkipped,
		"pendingWrites", len(l.pendingWrites))
	if !answered {
		answerResync(req, ResyncResult{Stats: *stats})
		if committed {
			// Its caller has its answer, so a replay that later refuses it reports the refusal
			// the way a live write's is, and does not answer again.
			l.pendingWrites[i].origin.answered = true
		}
	}
}

// answerResync answers a resync's caller. A resync the remote held back was answered then
// (writeOrigin.answered), and is applied later with nobody waiting: its outcome is then reported the
// way a live write's is.
func answerResync(req *ResyncRequest, result ResyncResult) {
	req.reply(result)
}

// settleFailed acts on a write whose commit failed for good: the plan was refused, or the write
// cannot be made, and retrying the same broken state helps nobody. It is no longer pending, a
// request riding it fails, and a refusal is surfaced as GitPathAccepted=False instead of being logged as a
// write fault (a resync's refusal reaches its caller, which classifies it).
func (l *branchWorkerEventLoop) settleFailed(i int, err error) {
	l.settleDropped(l.removeAt(i), err)
}

// settleDropped settles a write already taken out of the pending writes whose commit failed for good.
func (l *branchWorkerEventLoop) settleDropped(pendingWrite PendingWrite, err error) {
	switch {
	case pendingWrite.origin.resync != nil && !pendingWrite.origin.answered:
		req := pendingWrite.origin.resync
		// A refusal reaches the caller on ResyncResult.Err, where the watch layer classifies it
		// and blocks the collection. spec.onRefusal needs it here too, and this is the path that
		// matters: a per-type reconcile evaluates the same objects a live write would, so it is
		// normally what discovers a write-boundary refusal first.
		var refused *manifestanalyzer.AcceptanceRefusedError
		if errors.As(err, &refused) {
			l.reapplyForRefusal(req.GitTargetName, req.GitTargetNamespace, err.Error(), refused,
				refusalObservationForDesired(req.Desired, refused), req.refusalCollection())
		}
		l.w.Log.Error(err, "Resync commit failed; dropping request", "resources", len(req.Desired))
		answerResync(req, ResyncResult{Err: err})
	case pendingWrite.origin.resync != nil:
		// Nobody is waiting any more: report it as a live write's refusal is reported.
		req := pendingWrite.origin.resync
		collection := req.refusalCollection()
		if isRefusal, refused := l.w.reportPathRefusal(err, req.GitTargetName, req.GitTargetNamespace,
			collection); isRefusal {
			l.reapplyForRefusal(req.GitTargetName, req.GitTargetNamespace, err.Error(), refused,
				refusalObservationForDesired(req.Desired, refused), collection)
		}
		l.w.Log.Error(err, "Resync commit failed after its caller was answered; dropping it",
			"resources", len(req.Desired))
	case pendingWrite.Kind == PendingWriteRequestRecord:
		l.w.Log.Error(err, "The empty commit recording a CommitRequest failed",
			"gitTarget", pendingWrite.GitTargetNamespace+"/"+pendingWrite.GitTargetName)
		l.failDecidedRequest(pendingWrite, fmt.Errorf("record the message in an empty commit: %w", err))
	default:
		l.settleFailedChange(pendingWrite, err)
	}
}

// settleFailedChange settles a window whose commit failed for good.
func (l *branchWorkerEventLoop) settleFailedChange(pendingWrite PendingWrite, err error) {
	name, namespace := pendingWrite.windowTarget()
	collection := sourceCollectionForEvents(pendingWrite.Events)
	if isRefusal, refused := l.w.reportPathRefusal(err, name, namespace, collection); isRefusal {
		l.w.recordCommitFailure(commitFailureRefused)
		l.reapplyForRefusal(name, namespace, err.Error(), refused,
			refusalObservationForEvents(pendingWrite.Events, refused), collection)
	} else {
		l.w.recordCommitFailure(commitFailureError)
		l.w.Log.Error(err, "Commit failed; dropping the write",
			"gitTarget", namespace+"/"+name, "events", len(pendingWrite.Events))
	}
	l.failDecidedRequest(pendingWrite, fmt.Errorf("commit failed: %w", err))
}

// replyToDeferredResyncs answers the callers of the resyncs among the writes not committed yet, when
// the remote could not be reached or the parent branch a new write branch is created from is
// missing. The writes themselves all stay pending for the retry, and admission backpressure bounds
// them however long that lasts. A resync's caller is answered now with the error, because it is
// waiting to hear what the resync found and that cannot be known yet; the resync itself stays, in
// its place, and is applied with the rest.
func (l *branchWorkerEventLoop) replyToDeferredResyncs(err error) {
	for i := l.materializedPrefix(); i < len(l.pendingWrites); i++ {
		origin := &l.pendingWrites[i].origin
		if origin.resync == nil || origin.answered {
			continue
		}
		req := origin.resync
		l.w.Log.Error(err, "Cannot refresh the remote before resync; it stays pending",
			"resources", len(req.Desired), "gitTarget", req.GitTargetNamespace+"/"+req.GitTargetName)
		answerResync(req, ResyncResult{Err: fmt.Errorf("%w: refresh remote before resync: %w", ErrResyncDeferred, err)})
		origin.answered = true
	}
}

// failDecidedRequest fails the request a dropped write carried.
func (l *branchWorkerEventLoop) failDecidedRequest(pendingWrite PendingWrite, cause error) {
	if pendingWrite.CommitRequest != nil {
		l.resolveCommitRequest(*pendingWrite.CommitRequest, FinalizeResult{Branch: l.w.Branch, Err: cause})
	}
}