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, an atomic batch, 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, or the
// atomic request. They are process objects, kept beside
// the write's data rather than in it.
type writeOrigin struct {
	resync *ResyncRequest
	atomic *WriteRequest
	// 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)
	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 or an atomic batch whose commit failed for good.
func (l *branchWorkerEventLoop) settleFailedChange(pendingWrite PendingWrite, err error) {
	kind, name, namespace, collection := changeIdentity(pendingWrite)
	if isRefusal, refused := l.w.reportPathRefusal(err, name, namespace, collection); isRefusal {
		l.w.recordCommitFailure(kind, commitFailureRefused)
		l.reapplyForRefusal(name, namespace, err.Error(), refused,
			refusalObservationForEvents(pendingWrite.Events, refused), collection)
	} else {
		l.w.recordCommitFailure(kind, commitFailureError)
		l.w.Log.Error(err, "Commit failed; dropping the write",
			"kind", kind, "gitTarget", namespace+"/"+name, "events", len(pendingWrite.Events))
	}
	l.failDecidedRequest(pendingWrite, fmt.Errorf("commit failed: %w", err))
}

// changeIdentity is a window's or an atomic batch's metric kind, GitTarget, and source collection.
func changeIdentity(pendingWrite PendingWrite) (string, string, string, itypes.CollectionKey) {
	if request := pendingWrite.origin.atomic; request != nil {
		name, namespace := atomicRefusalTarget(request)
		return commitFailureKindAtomic, name, namespace, request.sourceCollection()
	}
	name, namespace := pendingWrite.windowTarget()
	return commitFailureKindWindow, name, namespace, sourceCollectionForEvents(pendingWrite.Events)
}

// 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("refresh remote before resync: %w", 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})
	}
}