diff --git a/docs/TODO.md b/docs/TODO.md index 8e589146..6d89effc 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -11,6 +11,13 @@ Nothing open. The 0.54.0 release PR's changelog was checked against what ships o ## Applier claims and bi-directional editing +- [ ] **A withheld apply rides the next edit.** S1 withholds an apply that lands over an edit decided + after the flush mark, and a `Repair` puts the edit back. But the apply did land on the cluster, so + the next edit of that object carries the apply's value for every field it did not change, and the + window commits the whole object under that author's name: the overwritten edit is lost, and the + `Repair` then applies the loss. The failing test is + `TestApplyWindow_AWithheldApplyIsNotCommittedUnderTheNextEdit` + (`internal/git/apply_window_carry_test.go`, skipped; `GITOPS_REVERSER_KNOWN_BUGS=1` runs it). - [ ] **Reopen the render-fidelity gate after a Git repair, automatically.** Today a target on `RenderMatchesLive=False` clears only on an explicit recheck (the `reconcile.configbutler.ai/requestedAt` annotation, which restarts every scope; the render diff --git a/docs/design/branch-worker-event-model.md b/docs/design/branch-worker-event-model.md index b4974e5c..d44f4087 100644 --- a/docs/design/branch-worker-event-model.md +++ b/docs/design/branch-worker-event-model.md @@ -324,6 +324,14 @@ Represent intent, decisions, and external work separately so recovery knows what These names are illustrative. Use a versioned journal schema when choosing the implementation. Facts describe decisions already made; a duplicate command must find its previous decision. +**Give each decision a specific name and a reason**, in the product's words: `SaveAttached`, +`WindowClosed` with its close reason, rather than a generic `StateChanged` that leaves the reader to +compare two states. Today those names serve log lines, close reasons and tests, with the identities +the decision is about (request UID, window ID). They are not a journal schema. Whether restart +recovery stores durable current state (pending work, deadlines, receipts) or an event history is +still open; the HA plan decides it from the guarantees it needs and their measured cost. Git +already keeps the authored history of every published change, which weighs toward current state. + ```mermaid flowchart TD I[Watch observations and save commands] --> A[Durable admission] diff --git a/docs/design/concurrent-author-windows.md b/docs/design/concurrent-author-windows.md index 159ef353..70816f87 100644 --- a/docs/design/concurrent-author-windows.md +++ b/docs/design/concurrent-author-windows.md @@ -162,10 +162,14 @@ generation change), the worker recomputes every open window's footprint against **before** the next write is admitted beside others. A footprint that is now unknown, or that now overlaps another window's, closes the windows involved, oldest first, with `path_conflict`. -A last check runs at materialization: the files the write plan touched (`writeBatch.touched`) must -be inside the window's footprint. A plan that touched more is a footprint bug. The worker cannot -undo the order of what is already decided, so it logs the paths, counts it in a metric, and falls -back to one window for the rest of its life. The reference model treats it as a test failure. +A last check runs at materialization, **before the commit**: every path the write's commit would +change must be inside the window's footprint. That is the whole batch's changed files; +`writeBatch.touched` covers one event at a time and is not enough. A write that changes more is a +footprint bug, and committing it could put an older value over a newer one, so the worker refuses +it: it undoes the worktree, fails the window's save with the paths in its message, falls back to +one window for the rest of its life, and asks a resync of the `GitTarget`, which brings Git back to +the cluster without depending on the order of the windows. The reference model treats it as a test +failure. ### Every close trigger keeps its exceptions @@ -220,6 +224,25 @@ is the point of the change, and it needs saying in `docs/configuration.md` when - `SaveInProgress` is unchanged: it is already per author and `GitTarget`. - Each request settles exactly once, whichever trigger closes its window. +### The window decisions have specific names + +Concurrent windows do not depend on persistence: they keep explicit identities, deadlines and +decision reasons, and leave how a restart recovers them to the +[event model](branch-worker-event-model.md#commands-facts-and-effects) and the HA plan. Each decision +gets a specific name, used in log lines, close reasons and the reference model. A log line carries +the branch with the window ID, since the ID resets with the worker; the ID is never a metric label. + +| Decision | Logged with | When | +| --- | --- | --- | +| `WindowOpened` | window ID, author, `GitTarget`, footprint, revision, opening time | a write opens a window | +| `WriteJoinedWindow` | window ID, the write, the footprint after it | a write joins an open window | +| `FootprintReproved` | window ID, new revision, the footprint or unknown | the revision changes under an open window | +| `WindowClosed` | window ID, close reason, the writes it holds | a window closes | +| `SingleWindowFallback` | the paths a plan touched outside its footprint | the materialization check fails | + +The window ID is the opening sequence number, unique for the life of one worker. Deadlines follow +from the opening time, the latest write and the timers. + ## How to test it: a reference model "Git equals the cluster at the end" is too weak a check for this change. A later write can repair a @@ -300,8 +323,9 @@ the reference model checks every commit. - **P1. A footprint that is too small.** A write lands somewhere its footprint did not name (an existing document in an unexpected file, a generator source, a removal that edits a shared file), and two windows commit to one file in the wrong order. *Mitigation:* footprints are proven or - unknown, never guessed; the table's "no" rows stay serial; the reference model includes shared - files and generator layouts. + unknown, never guessed; the table's "no" rows stay serial; the check before the commit refuses a + write that changes more than its footprint; the reference model includes shared files and + generator layouts. - **P2. A close trigger loses its exception.** A push hint or an apply flush that closes a window carrying a save splits a save that is protected today. *Mitigation:* the trigger table above is the specification, and each exception has a test in stage 3. @@ -314,8 +338,8 @@ the reference model checks every commit. every step, as well as once drained. - **P6. A stale footprint proof.** A proof made against one tree is used after an own commit, a replay or a `GitTarget` change moved the index. *Mitigation:* proofs carry their revision, are - recomputed before the next concurrent admission, and the materialization check falls back to one - window if a plan ever touches a file outside its footprint. + recomputed before the next concurrent admission, and the check before the commit refuses a write + that changes a file outside its footprint. - **P7. The flush starts waiting again.** A change that makes `apply_due` wait for windows to close on their own reintroduces the starvation S1 removed. *Mitigation:* the flush closes every window at once, as today; only the save exception holds it. diff --git a/docs/design/window-field-ownership-and-apply-hold.md b/docs/design/window-field-ownership-and-apply-hold.md index b10ddf83..a6d1886d 100644 --- a/docs/design/window-field-ownership-and-apply-hold.md +++ b/docs/design/window-field-ownership-and-apply-hold.md @@ -329,7 +329,7 @@ its Kustomization suspended and the `ArgoCDClaim` keeps automated sync off. Window timers cannot bound the hold. A window's `max_duration` is not fixed: attaching a `CommitRequest` replaces the window's timers with the request's, running from the attach -(`attachToOpenWindow`, `internal/git/commit_request_attach_loop.go`). A hold that waited for the +(`attachToWindow`, `internal/git/commit_request_attach_loop.go`). A hold that waited for the windows' own deadlines could be extended by every attach. So the hold has an **absolute deadline, measured from when it began**. It ends at the first of: diff --git a/internal/git/applier_claim.go b/internal/git/applier_claim.go index d956e233..d6fa44df 100644 --- a/internal/git/applier_claim.go +++ b/internal/git/applier_claim.go @@ -365,18 +365,23 @@ func (l *branchWorkerEventLoop) writeBusy() bool { return len(l.pendingCRs) > 0 || l.recovery.active || l.w.worktreeDirty() } -// closeWindowForPushHint closes the open commit window early when a Git host has said the branch may -// have moved, so its writes are published, and replayed onto the move, now rather than when the -// window's timers allow. A window a CommitRequest holds is left to its save. +// closeWindowForPushHint closes the open commit windows early when a Git host has said the branch may +// have moved, so their writes are published, and replayed onto the move, now rather than when the +// windows' timers allow. A window a CommitRequest holds is left to its save. func (l *branchWorkerEventLoop) closeWindowForPushHint() { hinted := l.pushHinted l.pushHinted = false - if !hinted || l.openWindow == nil || l.openWindow.pendingCR != nil || l.w.applyRequester == nil { + if !hinted || l.w.applyRequester == nil { return } - l.w.Log.Info("A Git host reported a push; closing the commit window to publish it now", - "branch", l.w.Branch) - l.finalizeOpenWindowWithReason(windowFinalizeReasonForeignCommit) + for _, window := range l.commitWindows.all() { + if window.pendingCR != nil { + continue + } + l.w.Log.Info("A Git host reported a push; closing the commit window to publish it now", + "branch", l.w.Branch) + l.finalizeWindow(window, windowFinalizeReasonForeignCommit) + } } // serviceAppliers settles the reports that came back, and sends whatever attempts are due, after a diff --git a/internal/git/applier_claim_test.go b/internal/git/applier_claim_test.go index b4a986ee..77ae71de 100644 --- a/internal/git/applier_claim_test.go +++ b/internal/git/applier_claim_test.go @@ -379,6 +379,29 @@ func TestCloseWindowForPushHint_PublishesTheOpenWindowNow(t *testing.T) { assert.Len(t, loop.pendingWrites, 1, "the window's write is decided, to be published") } +// TestCloseWindowForPushHint_LeavesASavesWindowToTheSave: a window a CommitRequest holds closes +// when its save says, not on a push hint. +func TestCloseWindowForPushHint_LeavesASavesWindowToTheSave(t *testing.T) { + worker, _, _ := setupCommitPushSplitWorker(t) + createPlainGitTarget(t, worker, "team-a", "team-a") + worker.applyRequester = &recordingRequester{} + loop := newBranchWorkerEventLoop(worker, time.Hour) + t.Cleanup(loop.stopTimers) + + loop.handleQueueItem(WorkItem{Request: &WriteRequest{ + Events: []Event{configMapTargetEvent("edited", "alice", "team-a")}, + }}) + require.NotNil(t, loop.openWindow, "an hour-long window is still open") + id := commitRequestID{Namespace: "team-a", Name: "save", UID: "uid-save"} + loop.openWindow.pendingCR = &id + + loop.handleRefreshRequest(&RefreshRequest{PushNotice: true}) + loop.closeWindowForPushHint() + + assert.NotNil(t, loop.openWindow, "the save's window is still open") + assert.Empty(t, loop.pendingWrites) +} + // applierLedgerFixture is a real Git server, a GitTarget naming an applier, and an event loop whose // requests are recorded. func applierLedgerFixture(t *testing.T, slug string) (*ledgerFixture, *branchWorkerEventLoop, *recordingRequester) { diff --git a/internal/git/apply_window_carry_test.go b/internal/git/apply_window_carry_test.go new file mode 100644 index 00000000..fa02370d --- /dev/null +++ b/internal/git/apply_window_carry_test.go @@ -0,0 +1,102 @@ +// SPDX-License-Identifier: Apache-2.0 + +package git + +import ( + "os" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + configv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3" + itypes "github.com/ConfigButler/gitops-reverser/internal/types" +) + +// TestApplyWindow_AWithheldApplyIsNotCommittedUnderTheNextEdit is a known bug (docs/TODO.md, "A +// withheld apply rides the next edit"), written now and skipped until it is fixed; set +// GITOPS_REVERSER_KNOWN_BUGS=1 to run it. +// +// S1 withholds an apply that lands over an edit decided after the flush mark, and a Repair puts the +// edit back once it is published. But the apply did land on the cluster, and a window holds whole +// objects: the next edit of that object, by the same author, carries the apply's value for every +// field it did not touch, and its commit writes that value under the author's name. The edit the +// apply overwrote is lost, and the Repair puts the loss back on the cluster. +func TestApplyWindow_AWithheldApplyIsNotCommittedUnderTheNextEdit(t *testing.T) { + if os.Getenv("GITOPS_REVERSER_KNOWN_BUGS") == "" { + t.Skip("known bug, docs/TODO.md: a withheld apply rides the next edit") + } + + // edit is alice's live update of y, holding both of its keys. + edit := func(a, b string) Event { + event := configMapTargetEvent("y", "alice", echoTarget) + event.Operation = string(itypes.OperationUpdate) + event.Object.Object["data"] = map[string]interface{}{"a": a, "b": b} + return event + } + // lateApply is the applier writing y back to what the flushed commit holds, after the edit. + lateApply := func() Event { + event := edit("a1", "b1") + event.UserInfo.Username = "system:serviceaccount:flux-system:kustomize-controller" + event.LastFieldManagers = []string{FluxFieldManager} + event.LastFieldManagedAt = time.Now().Add(time.Hour) + return event + } + // flushOverAnEdit makes an apply due while x2 sits in the open window, and lets the flush publish + // it only after alice's edit a=a2 is decided: the attempt goes for a commit that lacks the edit. + flushOverAnEdit := func(t *testing.T, slug string) (*echoFixture, *recordingRequester) { + f := newEchoFixture(t, slug, configv1alpha3.PruneOnEvent, map[string]string{"x": "x1"}) + f.writeWindow(edit("a1", "b1")) // y as Git holds it before anyone edits + requester := f.withClaim() + f.loop.lastPushAt = time.Now() + + f.writeWindow(echoEvent("x", "x2")) + restore := failPushes(t, f.worker) + f.echoApplier().repair = true + f.loop.endWake(0) + + f.writeWindow(edit("a2", "b1")) // E1, decided after the mark + restore() + f.loop.retry.due = time.Now().Add(-time.Second) + f.loop.runRetry() + f.loop.endWake(0) + require.Len(t, requester.taken(), 1, "the apply is sent without E1") + return f, requester + } + + t.Run("in the window that holds the edit", func(t *testing.T) { + f, _ := flushOverAnEdit(t, "carry-same-window") + require.NotNil(t, f.loop.openWindow, "E1 is collecting in alice's window") + + f.writeWindow(lateApply()) + require.True(t, f.echoApplier().repair, "the apply over E1 is withheld, and a Repair is pending") + f.writeWindow(edit("a1", "b2")) // E2: alice changes b on the object the apply wrote + f.loop.lastPushAt = time.Time{} + require.True(t, f.loop.finalizeOpenWindow()) + f.loop.pushPending() + + require.Equal(t, "alice", gitIn(t, f.repoDir, "log", "-1", "--format=%an", "main")) + assert.Contains(t, f.onMain(echoFile("y")), "a: a2", + "E1 is alice's, and the apply that overwrote it was withheld; Git loses it") + }) + + t.Run("after the edit is published", func(t *testing.T) { + f, _ := flushOverAnEdit(t, "carry-next-window") + f.loop.lastPushAt = time.Time{} + require.True(t, f.loop.finalizeOpenWindow()) + f.loop.pushPending() + require.Contains(t, f.onMain(echoFile("y")), "a: a2", "precondition: E1 is published") + + f.writeWindow(lateApply()) + require.True(t, f.echoApplier().repair, "the apply over E1 is withheld, and a Repair is pending") + f.writeWindow(edit("a1", "b2")) + f.loop.lastPushAt = time.Time{} + require.True(t, f.loop.finalizeOpenWindow()) + f.loop.pushPending() + + require.Equal(t, "alice", gitIn(t, f.repoDir, "log", "-1", "--format=%an", "main")) + assert.Contains(t, f.onMain(echoFile("y")), "a: a2", + "alice's commit of b reverts her own E1 to the withheld apply's a1") + }) +} diff --git a/internal/git/branch_worker.go b/internal/git/branch_worker.go index ba477525..f52d0959 100644 --- a/internal/git/branch_worker.go +++ b/internal/git/branch_worker.go @@ -386,7 +386,8 @@ type BranchWorker struct { // Set by the WorkerManager before Start; nil for a worker built on its own. crOwners *commitRequestOwners - // clock is the time source of the retry schedule; nil is time.Now. See now(). + // clock is the time source of the retry schedule and of the commit windows' timers: the idle and + // max-duration deadlines and a CommitRequest's attach deadline. nil is time.Now. See now(). clock func() time.Time // parentRecoveryOpen, parentRecoveryFound and parentProbeHold mirror the event loop's // recovery latch for readers outside it. See parent_recovery.go. @@ -1008,6 +1009,12 @@ func (w *BranchWorker) processEvents() { // goroutine running run() may touch these fields, so no extra synchronisation // is required. type branchWorkerEventLoop struct { + // commitWindows holds the live commit-shaped event windows and what they are charged. A window is + // finalized when its idle or max-duration timer runs out, and eagerly on author/target changes, + // resync arrivals, byte-cap trips, an attach: Next, and shutdown. Embedded, so loop.openWindow + // names the one window it holds. + commitWindows + w *BranchWorker // barrierSeq numbers the barriers this loop's apply windows take (apply_barrier.go). @@ -1017,12 +1024,6 @@ type branchWorkerEventLoop struct { // for any target that declares no spec.commit.window field and for one that cannot be read. defaultWindow commitWindowDefaults - // openWindow holds the one live commit-shaped event window. It is finalized when its idle or - // max-duration timer runs out, and eagerly on author/target changes, resync arrivals, byte-cap - // trips, an attach: Next, and shutdown. - openWindow *openWindow - windowBytes int64 - // pendingWrites holds the retained durability units for local commits that // have not yet successfully reached the remote. A successful push is the // only thing that clears this slice. @@ -1031,6 +1032,9 @@ type branchWorkerEventLoop struct { lastPushAt time.Time commitTimer *time.Timer + // commitTimerAt is the worker-clock time commitTimer is armed for, while it is armed. The timer + // itself runs on real time; tests that step the clock deliver its wake at this time. + commitTimerAt time.Time // pushTimer fires the next push when the success cooldown ends. pushTimer *time.Timer // retry is the one schedule of the next attempt at work a failed attempt left pending. See @@ -1101,6 +1105,8 @@ type branchWorkerEventLoop struct { // that no window reaches still resolves on time. An attached request needs no timer of its own: // its window's timers close it. attachTimer *time.Timer + // attachTimerAt is the worker-clock time attachTimer is armed for, while it is armed. + attachTimerAt time.Time // recovery is the open task of writing what a missing parent branch held back, and the probe // that drives it. See parent_recovery.go. @@ -1225,7 +1231,7 @@ func (l *branchWorkerEventLoop) endWake(released int64) { // publishing the flag second would let both read empty while a window is open, and a scrape landing // there would report a drained queue for a worker that has work. func (l *branchWorkerEventLoop) publishLoopState(released int64) { - l.w.hasUnpushedWork.Store(l.openWindow != nil || len(l.pendingWrites) > 0 || len(l.held) > 0) + l.w.hasUnpushedWork.Store(l.commitWindows.anyOpen() || len(l.pendingWrites) > 0 || len(l.held) > 0) l.syncAdmission(released) } @@ -1249,10 +1255,10 @@ func (l *branchWorkerEventLoop) timerChannels() ( } // totalRetainedBytes is what the operator-level byte cap is enforced against: -// the open window plus the locally-committed-but-not-yet-pushed events that +// the open windows plus the locally-committed-but-not-yet-pushed events that // we keep around for replay-on-conflict. func (l *branchWorkerEventLoop) totalRetainedBytes() int64 { - return l.windowBytes + l.pendingWritesBytes + return l.commitWindows.windowBytes + l.pendingWritesBytes } // dispatch hands one item to its handler, and returns the intake charge the item was admitted with. @@ -1304,7 +1310,7 @@ func (l *branchWorkerEventLoop) handleLiveEvents(request *WriteRequest) { // finalizing it as identity, the byte cap or a zero window require. func (l *branchWorkerEventLoop) appendLiveEvent(event Event) { { - if l.openWindow != nil && !l.openWindow.canAppend(event) { + for _, other := range l.commitWindows.blocking(event) { // Log the identity that broke the window so an unexpected split is // diagnosable: the common cause is an incoming event whose author is empty // (an unattributed update, e.g. a /status change that carries no audit fact) @@ -1312,37 +1318,38 @@ func (l *branchWorkerEventLoop) appendLiveEvent(event Event) { // can split a CommitRequest collect window. windowAuthor vs eventAuthor makes // that visible at a glance. l.w.Log.Info("Window identity change forces finalize before appending event", - "windowAuthor", l.openWindow.Author, + "windowAuthor", other.Author, "eventAuthor", event.UserInfo.Username, "operation", event.Operation, "resource", event.Identifier.String(), "eventTarget", event.GitTargetNamespace+"/"+event.GitTargetName) - l.finalizeOpenWindowWithReason(windowFinalizeReasonIdentityChange) + l.finalizeWindow(other, windowFinalizeReasonIdentityChange) } + window := l.commitWindows.forWrite(event) opened := false - if l.openWindow == nil { + if window == nil { l.w.Log.Info("Opening commit window", "author", event.UserInfo.Username, "gitTarget", event.GitTargetNamespace+"/"+event.GitTargetName, "resource", event.Identifier.String()) - l.openWindow = newOpenWindow(event, l.w.contentWriter) + window = l.commitWindows.open(event, l.w.contentWriter, l.w.now()) // One window, one GitTarget: snapshot that target's timers now, so a later edit to the // target never moves a deadline already running. timers := l.w.commitWindowFor( l.w.ctx, event.GitTargetName, event.GitTargetNamespace, l.defaultWindow) - l.openWindow.timers = windowTimers{ + window.timers = windowTimers{ idle: timers.idleTimeout, - maxAt: time.Now().Add(timers.maxDuration), + maxAt: l.w.now().Add(timers.maxDuration), } opened = true } // The window keeps one event per path, so a replaced one stops being charged. - if replaced, ok := l.openWindow.add(event); ok { - l.windowBytes -= eventRetainedCharge(&replaced) + if replaced, ok := window.add(event); ok { + l.commitWindows.charge(window, -eventRetainedCharge(&replaced)) } - l.openWindow.lastWriteAt = time.Now() - l.windowBytes += eventRetainedCharge(&event) + window.lastWriteAt = l.w.now() + l.commitWindows.charge(window, eventRetainedCharge(&event)) if opened { // A waiting CommitRequest gets the new window BEFORE anything can close it: before the @@ -1352,7 +1359,7 @@ func (l *branchWorkerEventLoop) appendLiveEvent(event Event) { // waiting on another author, never see the window that refuses it. l.noteForeignWindow() l.attachWaitingCommitRequests() - if l.openWindow == nil { + if !l.commitWindows.isOpen(window) { return // the attached request's maxDuration: 0s closed it already } } @@ -1404,21 +1411,27 @@ func (l *branchWorkerEventLoop) mirrors(event Event) bool { return true } -// closeOrArmWindow finalizes the open window when its first deadline has passed, and otherwise -// arms the commit timer for that deadline. A zero timer therefore closes the window in the same -// step that collected the write. +// closeOrArmWindow finalizes every open window whose first deadline has passed, oldest first, and +// arms the commit timer for the earliest deadline of the windows left open. A zero timer therefore +// closes the window in the same step that collected the write. func (l *branchWorkerEventLoop) closeOrArmWindow() { - if l.openWindow == nil { - l.stopCommitTimer() - return + now := l.w.now() + var next time.Time + for _, window := range l.commitWindows.all() { + closeAt, reason := window.closeAt() + if !closeAt.After(now) { + l.finalizeWindow(window, reason) + continue + } + if next.IsZero() || closeAt.Before(next) { + next = closeAt + } } - closeAt, reason := l.openWindow.closeAt() - delay := time.Until(closeAt) - if delay <= 0 { - l.finalizeOpenWindowWithReason(reason) + if next.IsZero() { + l.stopCommitTimer() return } - l.resetCommitTimer(delay) + l.resetCommitTimer(next.Sub(now)) } func (l *branchWorkerEventLoop) handleShutdown() { @@ -1481,6 +1494,7 @@ func (w *BranchWorker) drainQueue() { } func (l *branchWorkerEventLoop) resetCommitTimer(delay time.Duration) { + l.commitTimerAt = l.w.now().Add(delay) if l.commitTimer == nil { l.commitTimer = time.NewTimer(delay) return @@ -1494,34 +1508,38 @@ func (l *branchWorkerEventLoop) resetCommitTimer(delay time.Duration) { l.commitTimer.Reset(delay) } -// finalizeOpenWindowWithReason closes the live event window into one decided write. The attached -// CommitRequest message overrides the live template. +// finalizeOpenWindowWithReason closes every open window, oldest first: the triggers that are about +// the branch rather than about one window (a resync, an apply flush, the byte cap, shutdown). +func (l *branchWorkerEventLoop) finalizeOpenWindowWithReason(reason windowFinalizeReason) { + for _, window := range l.commitWindows.all() { + l.finalizeWindow(window, reason) + } +} + +// finalizeWindow closes one live event window into one decided write. The attached CommitRequest +// message overrides the live template. // // The closed window is a decided write (decide): a remote that cannot be reached leaves it pending // for the publication retry, and its CommitRequest held. Only a failure of the write itself // drops it, and resolves that request Failed (settleFailed). -func (l *branchWorkerEventLoop) finalizeOpenWindowWithReason(reason windowFinalizeReason) { - if l.openWindow == nil { - return - } - +func (l *branchWorkerEventLoop) finalizeWindow(window *openWindow, reason windowFinalizeReason) { l.stopCommitTimer() // Counted here, before anything can fail or find no diff, so every window is counted once and // its collection time never includes the commit. - l.w.recordWindowClosed(l.openWindow, reason) - events := l.openWindow.orderedEvents() - windowAuthor := l.openWindow.Author - targetName, targetNamespace := l.openWindow.GitTarget, l.openWindow.GitTargetNamespace + l.w.recordWindowClosed(window, reason) + events := window.orderedEvents() + windowAuthor := window.Author + targetName, targetNamespace := window.GitTarget, window.GitTargetNamespace windowTarget := targetNamespace + "/" + targetName - pendingCR := l.openWindow.pendingCR - effectiveMessage := l.openWindow.pendingMessage + pendingCR := window.pendingCR + effectiveMessage := window.pendingMessage // The render-fidelity gate drops the whole window, request or not: its events were planned // against a render nobody has checked. Suspension is read from the write built below, because // a suspended target still scans. See write_gate.go. if err := l.w.targetWriteRefusal(targetName, targetNamespace, false); err != nil { l.w.Log.V(1).Info("Discarding open window while render fidelity is not established", "reason", string(reason), "gitTarget", targetNamespace+"/"+targetName) - l.dropOpenWindow(pendingCR, err) + l.dropWindow(window, pendingCR, err) return } @@ -1530,7 +1548,7 @@ func (l *branchWorkerEventLoop) finalizeOpenWindowWithReason(reason windowFinali "windowAuthor", windowAuthor, "windowTarget", windowTarget, "events", len(events), - "windowBytes", l.windowBytes, + "windowBytes", window.bytes, "pendingWrites", len(l.pendingWrites), "messageOverride", effectiveMessage != "", "attachedCR", pendingCR != nil) @@ -1543,18 +1561,17 @@ func (l *branchWorkerEventLoop) finalizeOpenWindowWithReason(reason windowFinali "windowAuthor", windowAuthor, "windowTarget", windowTarget, "events", len(events)) - l.dropOpenWindow(pendingCR, fmt.Errorf("build pending write: %w", err)) + l.dropWindow(window, pendingCR, fmt.Errorf("build pending write: %w", err)) return } pendingWrite.CommitMessage = effectiveMessage - pendingCR, proceed := l.gateBuiltWindow(pendingCR, pendingWrite.Target(), reason) + pendingCR, proceed := l.gateBuiltWindow(window, pendingCR, pendingWrite.Target(), reason) if !proceed { return } l.carryRequestOnWrite(pendingWrite, pendingCR) - l.openWindow = nil - l.windowBytes = 0 + l.commitWindows.remove(window) l.decide(*pendingWrite) l.w.Log.Info("Open commit window finalized", @@ -1588,12 +1605,12 @@ func (l *branchWorkerEventLoop) carryRequestOnWrite(pendingWrite *PendingWrite, // target's scan stays fresh, but it commits nothing, and the push would report that as // AlreadyPresent, so only the request fails. See write_gate.go. func (l *branchWorkerEventLoop) gateBuiltWindow( - pendingCR *commitRequestID, target ResolvedTargetMetadata, reason windowFinalizeReason, + window *openWindow, pendingCR *commitRequestID, target ResolvedTargetMetadata, reason windowFinalizeReason, ) (*commitRequestID, bool) { if err := l.w.targetWriteRefusal(target.Name, target.Namespace, false); err != nil { l.w.Log.V(1).Info("Discarding built window: render fidelity closed while it was finalized", "reason", string(reason), "gitTarget", target.Namespace+"/"+target.Name) - l.dropOpenWindow(pendingCR, err) + l.dropWindow(window, pendingCR, err) return nil, false } if pendingCR != nil && target.Suspend { @@ -1603,11 +1620,10 @@ func (l *branchWorkerEventLoop) gateBuiltWindow( return pendingCR, true } -// dropOpenWindow discards a window whose finalize failed before it was decided, resolving any +// dropWindow discards a window whose finalize failed before it was decided, resolving any // attached CommitRequest as Failed so the controller does not poll forever. -func (l *branchWorkerEventLoop) dropOpenWindow(pendingCR *commitRequestID, cause error) { - l.openWindow = nil - l.windowBytes = 0 +func (l *branchWorkerEventLoop) dropWindow(window *openWindow, pendingCR *commitRequestID, cause error) { + l.commitWindows.remove(window) if pendingCR != nil { l.resolveCommitRequest(*pendingCR, FinalizeResult{Branch: l.w.Branch, Err: cause}) } @@ -1763,6 +1779,7 @@ func (l *branchWorkerEventLoop) stopCommitTimer() { } } l.commitTimer = nil + l.commitTimerAt = time.Time{} } func (l *branchWorkerEventLoop) stopTimers() { @@ -2906,7 +2923,7 @@ func (w *BranchWorker) recordWindowClosed(window *openWindow, reason windowFinal )...)) } if telemetry.GitCommitWindowDurationSeconds != nil { - telemetry.GitCommitWindowDurationSeconds.Record(ctx, time.Since(window.openedAt).Seconds(), + telemetry.GitCommitWindowDurationSeconds.Record(ctx, w.now().Sub(window.openedAt).Seconds(), metric.WithAttributes(target...)) } } diff --git a/internal/git/commit_request_attach_loop.go b/internal/git/commit_request_attach_loop.go index 9800534b..2301e621 100644 --- a/internal/git/commit_request_attach_loop.go +++ b/internal/git/commit_request_attach_loop.go @@ -210,7 +210,7 @@ func (l *branchWorkerEventLoop) handleAttachCommitRequest(req *AttachCommitReque gitTargetNamespace: req.GitTargetNamespace, message: req.Message, seq: l.crSeq, - attachDeadline: time.Now().Add(req.AttachTimeout), + attachDeadline: l.w.now().Add(req.AttachTimeout), idleTimeout: req.IdleTimeout, maxDuration: req.MaxDuration, commitEmpty: req.CommitEmpty, @@ -232,40 +232,40 @@ func (l *branchWorkerEventLoop) handleAttachCommitRequest(req *AttachCommitReque "attach", string(req.Attach), "attachTimeout", req.AttachTimeout.String()) - if l.openWindow == nil { - return - } - if !pcr.matchesWindow(l.openWindow) { + if l.commitWindows.foreignTo(pcr) != nil { // Noted here and not left to noteForeignWindow, which skips a request whose deadline has // passed: with attachTimeout: 0s that is every request, so a foreign window open at // registration would go unseen and the request would resolve NoWindow — and, with // CommitEmpty, record an empty commit while another author's work is in flight. pcr.sawForeignWindow = true + } + window := l.commitWindows.forRequest(pcr) + if window == nil { return } if req.Attach == v1alpha3.AttachNext { // Start clean: close the author's window under whatever message it already carries, // including one another request attached, and wait for a write to open the next. The // registration above is what makes this happen once per request. - l.finalizeOpenWindowWithReason(windowFinalizeReasonAttachNext) + l.finalizeWindow(window, windowFinalizeReasonAttachNext) return } - if l.openWindow.pendingCR == nil { + if window.pendingCR == nil { // CurrentOrNext attaches to a window already open at registration before any deadline is // consulted, which is what lets attachTimeout: 0s attach at all. - l.attachToOpenWindow(pcr) + l.attachToWindow(window, pcr) } } // saveInProgress returns the earlier request a newly registered one would race for one window, or nil. // That is one of the same author on the same GitTarget (sameSaver) that is still waiting for a window, -// or that is collecting the open one. A request that asks for the Next window does not race one that +// or that is collecting an open window. A request that asks for the Next window does not race one that // is collecting: it closes that window under the earlier request's message first. A request already // waiting for its push holds no window, so it races nothing. func (l *branchWorkerEventLoop) saveInProgress( pcr *pendingCommitRequest, attach v1alpha3.AttachPolicy, ) *pendingCommitRequest { - now := time.Now() + now := l.w.now() for _, earlier := range l.pendingCRs { if !earlier.sameSaver(pcr) { continue @@ -273,8 +273,7 @@ func (l *branchWorkerEventLoop) saveInProgress( if earlier.waiting(now) { return earlier } - collecting := earlier.attached && l.openWindow != nil && l.openWindow.pendingCR != nil && - *l.openWindow.pendingCR == earlier.id + collecting := earlier.attached && l.commitWindows.carrying(earlier.id) != nil if collecting && attach != v1alpha3.AttachNext { return earlier } @@ -320,12 +319,13 @@ func (p *pendingCommitRequest) waiting(now time.Time) bool { // pass. Nothing refused that request; it waited out its whole wait with nothing open, which is the // benign outcome. func (l *branchWorkerEventLoop) noteForeignWindow() { - if l.openWindow == nil { - return - } - now := time.Now() + now := l.w.now() for _, pcr := range l.pendingCRs { - if !pcr.waiting(now) || pcr.sawForeignWindow || pcr.matchesWindow(l.openWindow) { + if !pcr.waiting(now) || pcr.sawForeignWindow { + continue + } + foreign := l.commitWindows.foreignTo(pcr) + if foreign == nil { continue } pcr.sawForeignWindow = true @@ -333,50 +333,52 @@ func (l *branchWorkerEventLoop) noteForeignWindow() { "request", pcr.id.Namespace+"/"+pcr.id.Name, "requestAuthor", pcr.author, "requestTarget", pcr.gitTargetNamespace+"/"+pcr.gitTargetName, - "windowAuthor", l.openWindow.Author, - "windowTarget", l.openWindow.GitTargetNamespace+"/"+l.openWindow.GitTarget) + "windowAuthor", foreign.Author, + "windowTarget", foreign.GitTargetNamespace+"/"+foreign.GitTarget) } } -// attachWaitingCommitRequests attaches the first-registered waiting request that matches the open -// window, when the window carries none yet. A window carries at most one request; a second waits -// for the next window, and a request whose deadline has passed takes none. +// attachWaitingCommitRequests attaches, to each open window that carries no request yet, oldest +// first, the first-registered waiting request that matches it. A window carries at most one request; +// a second waits for the next window, and a request whose deadline has passed takes none. func (l *branchWorkerEventLoop) attachWaitingCommitRequests() { - if l.openWindow == nil || l.openWindow.pendingCR != nil { - return - } - now := time.Now() - var first *pendingCommitRequest - for _, pcr := range l.pendingCRs { - if !pcr.waiting(now) || !pcr.matchesWindow(l.openWindow) { + now := l.w.now() + for _, window := range l.commitWindows.all() { + if window.pendingCR != nil { continue } - if first == nil || pcr.seq < first.seq { - first = pcr + var first *pendingCommitRequest + for _, pcr := range l.pendingCRs { + if !pcr.waiting(now) || !pcr.matchesWindow(window) { + continue + } + if first == nil || pcr.seq < first.seq { + first = pcr + } + } + if first != nil { + l.attachToWindow(window, first) } } - if first != nil { - l.attachToOpenWindow(first) - } } -// attachToOpenWindow binds a request's message to the open window and replaces the window's timers -// with the request's own: the request's maxDuration runs from this attach, and so does its first -// idle interval. A zero maxDuration closes the window right here. The message is the window's to -// keep from now on, so it moves into the window's charge. -func (l *branchWorkerEventLoop) attachToOpenWindow(pcr *pendingCommitRequest) { - now := time.Now() - l.openWindow.pendingMessage = pcr.message +// attachToWindow binds a request's message to an open window and replaces the window's timers with +// the request's own: the request's maxDuration runs from this attach, and so does its first idle +// interval. A zero maxDuration closes the window right here. The message is the window's to keep +// from now on, so it moves into the window's charge. +func (l *branchWorkerEventLoop) attachToWindow(window *openWindow, pcr *pendingCommitRequest) { + now := l.w.now() + window.pendingMessage = pcr.message id := pcr.id - l.openWindow.pendingCR = &id + window.pendingCR = &id pcr.attached = true - l.windowBytes += attachCharge(pcr.message) + l.commitWindows.charge(window, attachCharge(pcr.message)) timers := windowTimers{noIdle: pcr.idleTimeout == nil, maxAt: now.Add(pcr.maxDuration)} if pcr.idleTimeout != nil { timers.idle = *pcr.idleTimeout } - l.openWindow.timers = timers - l.openWindow.lastWriteAt = now + window.timers = timers + window.lastWriteAt = now l.w.setCommitRequestPhase(id, PhaseCollectingWindow) l.w.Log.Info("CommitRequest attached to open window", "request", id.Namespace+"/"+id.Name, @@ -390,7 +392,7 @@ func (l *branchWorkerEventLoop) attachToOpenWindow(pcr *pendingCommitRequest) { // still waiting. An attached request is not here: its window's timers close it, and whichever path // closes the window resolves it. func (l *branchWorkerEventLoop) expireWaitingCommitRequests() { - now := time.Now() + now := l.w.now() for id, pcr := range l.pendingCRs { if pcr.attached || pcr.attachDeadline.After(now) { continue @@ -534,7 +536,8 @@ func (l *branchWorkerEventLoop) rearmAttachTimer() { l.stopAttachTimer() return } - delay := time.Until(earliest) + l.attachTimerAt = earliest + delay := earliest.Sub(l.w.now()) if delay < 0 { delay = 0 } @@ -562,4 +565,5 @@ func (l *branchWorkerEventLoop) stopAttachTimer() { } } l.attachTimer = nil + l.attachTimerAt = time.Time{} } diff --git a/internal/git/commit_window_clock_test.go b/internal/git/commit_window_clock_test.go new file mode 100644 index 00000000..5b365dc6 --- /dev/null +++ b/internal/git/commit_window_clock_test.go @@ -0,0 +1,125 @@ +// SPDX-License-Identifier: Apache-2.0 + +package git + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// These tests pin that the commit windows' timers read the worker's clock: a test steps it past a +// deadline, delivers the wake of each timer the worker armed for it, and the window or the waiting +// request is due, with no sleep and no rewound deadline. + +// timerWake delivers the wakes the commit and attach timers would, as run does: each timer the worker +// has armed whose time on the worker's clock has come. A deadline no timer was armed for passes. +func timerWake(loop *branchWorkerEventLoop) { + now := loop.w.now() + if loop.commitTimer != nil && !loop.commitTimerAt.After(now) { + loop.commitTimer = nil + loop.closeOrArmWindow() + loop.endWake(0) + } + if loop.attachTimer != nil && !loop.attachTimerAt.After(now) { + loop.attachTimer = nil + loop.endWake(0) + } +} + +func TestWindowClock_TheIdleTimeoutIsMeasuredOnTheWorkersClock(t *testing.T) { + worker, _, _ := setupCommitPushSplitWorker(t) + createPlainGitTarget(t, worker, "team-a", "team-a") + clock := useTestClock(worker) + loop := newBranchWorkerEventLoop(worker, 10*time.Second) + defer loop.stopTimers() + loop.lastPushAt = time.Now() // hold the push, so the local commit stays inspectable + + writeTo(loop, "first") + require.NotNil(t, loop.openWindow) + assert.Equal(t, clock.Now(), loop.openWindow.openedAt, "the window opens at the clock's time") + + clock.advance(9 * time.Second) + timerWake(loop) + require.NotNil(t, loop.openWindow, "a second short of the idle timeout") + + writeTo(loop, "second") // restarts the idle interval + clock.advance(9 * time.Second) + timerWake(loop) + require.NotNil(t, loop.openWindow, "the second write restarted it") + + clock.advance(time.Second) + timerWake(loop) + assert.Nil(t, loop.openWindow, "idle for the whole timeout") + require.Len(t, loop.pendingWrites, 1) + assert.Len(t, loop.pendingWrites[0].Events, 2) +} + +func TestWindowClock_MaxDurationIsMeasuredOnTheWorkersClock(t *testing.T) { + worker, _, _ := setupCommitPushSplitWorker(t) + createPlainGitTarget(t, worker, "team-a", "team-a") + clock := useTestClock(worker) + loop := newBranchWorkerEventLoop(worker, 10*time.Second) + defer loop.stopTimers() + loop.defaultWindow.maxDuration = 25 * time.Second + loop.lastPushAt = time.Now() + + for i, name := range []string{"a", "b", "c"} { + writeTo(loop, name) + require.NotNil(t, loop.openWindow, "write %d keeps the window busy", i) + clock.advance(9 * time.Second) + timerWake(loop) + } + assert.Nil(t, loop.openWindow, "27s after it opened, past its 25s maxDuration") + require.Len(t, loop.pendingWrites, 1) + assert.Len(t, loop.pendingWrites[0].Events, 3) +} + +func TestWindowClock_AnAttachedRequestsTimersRunFromTheAttachOnTheWorkersClock(t *testing.T) { + worker, _, _ := setupCommitPushSplitWorker(t) + createPlainGitTarget(t, worker, "team-a", "team-a") + clock := useTestClock(worker) + loop := newBranchWorkerEventLoop(worker, time.Hour) + defer loop.stopTimers() + + writeTo(loop, "first") + clock.advance(time.Minute) + req := attachReq("alice", time.Hour) + req.MaxDuration = 30 * time.Second + serviceAttach(loop, req) + require.NotNil(t, loop.openWindow.pendingCR, "precondition: attached") + assert.Equal(t, clock.Now().Add(30*time.Second), loop.openWindow.timers.maxAt) + + clock.advance(29 * time.Second) + timerWake(loop) + require.NotNil(t, loop.openWindow) + + clock.advance(time.Second) + timerWake(loop) + assert.Nil(t, loop.openWindow, "the request's maxDuration ran out") + res, ok := outcome(t, worker) + require.True(t, ok) + assert.Equal(t, FinalizeCommitted, res.Outcome) +} + +func TestWindowClock_TheAttachDeadlineIsMeasuredOnTheWorkersClock(t *testing.T) { + worker, _, _ := setupCommitPushSplitWorker(t) + createPlainGitTarget(t, worker, "team-a", "team-a") + clock := useTestClock(worker) + loop := newBranchWorkerEventLoop(worker, time.Hour) + defer loop.stopTimers() + + serviceAttach(loop, attachReq("alice", 5*time.Second)) + clock.advance(4 * time.Second) + timerWake(loop) + _, resolved := outcome(t, worker) + require.False(t, resolved, "still waiting a second before its deadline") + + clock.advance(time.Second) + timerWake(loop) + res, resolved := outcome(t, worker) + require.True(t, resolved) + assert.Equal(t, FinalizeNoOpenWindow, res.Outcome) +} diff --git a/internal/git/intake.go b/internal/git/intake.go index f021ff61..6340596a 100644 --- a/internal/git/intake.go +++ b/internal/git/intake.go @@ -173,7 +173,7 @@ func (r *ResyncRequest) charge() int64 { return r.payloadSize() + pendingWriteOv // attached save is in the charge of the window it is attached to, and then of the write that window // became. func (l *branchWorkerEventLoop) heldBytes() int64 { - held := l.windowBytes + l.pendingWritesBytes + l.heldEventBytes + held := l.commitWindows.windowBytes + l.pendingWritesBytes + l.heldEventBytes for _, pcr := range l.pendingCRs { if !pcr.attached { held += attachCharge(pcr.message) diff --git a/internal/git/open_window.go b/internal/git/open_window.go index 8679f247..5e90fb4e 100644 --- a/internal/git/open_window.go +++ b/internal/git/open_window.go @@ -4,10 +4,9 @@ package git import "time" -// openWindow is the one live commit-shaped event window owned by a branch -// worker. It accepts only events with the same author and target; repeated -// writes to the same Git path are last-write-wins while preserving first-seen -// path order. +// openWindow is one live commit-shaped event window of a branch worker (commitWindows). It accepts +// only events with the same author and target; repeated writes to the same Git path are +// last-write-wins while preserving first-seen path order. type openWindow struct { // Author is event.UserInfo.Username verbatim. Author string @@ -50,13 +49,19 @@ type openWindow struct { // timers decide when the window closes. They are the GitTarget's, snapshotted when the window // opened, until a CommitRequest attaches and replaces them with its own. timers windowTimers - // openedAt is when the write that opened the window arrived. Unlike the timers and lastWriteAt it - // never moves, not even when a CommitRequest attaches, so it is what the window's collection time - // is measured from. + // openedAt is when the write that opened the window arrived, on the worker's clock (set by + // commitWindows.open). Unlike the timers and lastWriteAt it never moves, not even when a + // CommitRequest attaches, so it is what the window's collection time is measured from. openedAt time.Time // lastWriteAt is when the idle interval last restarted: the latest collected write, or the // attach of a CommitRequest. lastWriteAt time.Time + + // seq is the window's place in the order windows opened (commitWindows.windowSeq). + seq uint64 + // bytes is what the window is charged: its events, and its attached request's message. The set's + // windowBytes is the sum of every open window's. + bytes int64 } // windowTimers are the two timers that close a commit window, resolved to absolute terms. @@ -91,7 +96,6 @@ func newOpenWindow(e Event, writer eventContentWriter) *openWindow { GitTargetNamespace: e.GitTargetNamespace, pathToEvent: make(map[string]Event), writer: writer, - openedAt: time.Now(), } } diff --git a/internal/git/parent_recovery.go b/internal/git/parent_recovery.go index c96d2f03..a0ca5008 100644 --- a/internal/git/parent_recovery.go +++ b/internal/git/parent_recovery.go @@ -52,7 +52,9 @@ type parentRecovery struct { gen uint64 } -// now is the loop's clock, injectable so the retry schedule can be tested without sleeping. +// now is the loop's clock, injectable so the retry schedule and the commit windows' timers can be +// tested without sleeping. The timers themselves are real: a test that moves the clock past a +// deadline delivers the wake the timer would have, as run does. func (w *BranchWorker) now() time.Time { if w.clock != nil { return w.clock() diff --git a/internal/git/refresh.go b/internal/git/refresh.go index 5f078f68..4c14330d 100644 --- a/internal/git/refresh.go +++ b/internal/git/refresh.go @@ -131,7 +131,7 @@ func (l *branchWorkerEventLoop) handleRefreshRequest(req *RefreshRequest) { // commits, and the worktree may hold a partial write — which is also why this is the one exit // that does not re-read the folder: a layout resolved from a half-written tree is worse than // a slightly old one. A target mid-cycle is writing, and a write publishes its own layout. - if len(l.pendingWrites) > 0 || l.openWindow != nil || w.worktreeDirty() { + if len(l.pendingWrites) > 0 || l.commitWindows.anyOpen() || w.worktreeDirty() { l.noteBusyPushNotice(req, observed, known) w.Log.V(1).Info("Skipping refresh: this branch is mid-cycle", "branch", w.Branch, "gitTarget", req.Target.String()) diff --git a/internal/git/resync_flush.go b/internal/git/resync_flush.go index 1ebfc6b0..5ee78aee 100644 --- a/internal/git/resync_flush.go +++ b/internal/git/resync_flush.go @@ -45,9 +45,9 @@ func (l *branchWorkerEventLoop) applyResync(req *ResyncRequest) { "resourceVersion", req.ResourceVersion, "scope", req.Scope.String(), "gitTarget", req.GitTargetNamespace+"/"+req.GitTargetName, - "openWindow", l.openWindow != nil, + "openWindow", l.commitWindows.anyOpen(), "pendingWrites", len(l.pendingWrites)) - // A resync force-finalizes the open window to preserve arrival order before its mark-and-sweep. + // A resync force-finalizes every open window to preserve arrival order before its mark-and-sweep. l.finalizeOpenWindowWithReason(windowFinalizeReasonResyncBeforeApply) pendingWrite, err := l.w.buildResyncPendingWrite(l.w.ctx, req, &ResyncStats{}) diff --git a/internal/git/window_model_test.go b/internal/git/window_model_test.go new file mode 100644 index 00000000..01abdc6e --- /dev/null +++ b/internal/git/window_model_test.go @@ -0,0 +1,1081 @@ +// SPDX-License-Identifier: Apache-2.0 + +package git + +import ( + "fmt" + "maps" + "math/rand" + "os" + "reflect" + "regexp" + "slices" + "strconv" + "strings" + "testing" + "time" + + "gopkg.in/yaml.v3" + + configv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3" + "github.com/ConfigButler/gitops-reverser/internal/manifestanalyzer" +) + +// The reference model of docs/design/concurrent-author-windows.md ("How to test it"): a small, +// independent implementation of today's single-window rules, fed the same random inputs as a branch +// worker on a real Git server, with the worker checked against it. +// +// - After every step: the open window (author, objects, attached save, deadline, charge), the timers +// armed for it and for the waiting saves, the registered saves (attached or waiting, and until +// when), the retained-byte charge, the close reasons so far, and every save's outcome are the +// model's. +// - Every commit the worker publishes is the one the model's closing window makes: its author, its +// message, every path it changes, and each document whole. Per document, the writes that land do +// so in the order they were accepted, so an older write never lands after a newer one. +// - Every save settles exactly once, or is given back at shutdown. +// - Drained, Git holds the model's tree, path for path, and nothing is charged. +// +// The model shares no window code with the worker: only the event fixtures, the per-event charge +// (eventRetainedCharge, attachCharge) and the close-reason names. The worker's timers are real; the +// driver steps the worker's clock to the time each armed timer is due and delivers that timer's wake, +// so a deadline the worker armed no timer for is never delivered. +// +// WINDOW_MODEL_SEEDS and WINDOW_MODEL_STEPS raise the search; a failure names its seed, and +// WINDOW_MODEL_SEED replays one. + +const ( + // modelIdle and modelMax are the commit window the GitTarget declares none of, so the loop's default. + modelIdle = 10 * time.Second + modelMax = 60 * time.Second +) + +var ( + modelObjects = []string{"x", "y", "z"} + modelAuthors = []string{"alice", "bob", "carol"} + // modelBootstrap is what the worker adds to the folder, from its template, with the first commit + // it makes there. The GitTarget encrypts nothing, so there is no .sops.yaml. + modelBootstrap = []string{gitTargetIgnoreFileName, "README.md"} +) + +// modelEdit is one accepted write: seq is the order the worker accepted it in, and value is the +// document's key, empty for a delete. +type modelEdit struct { + seq int + obj string + value string + op string + charge int64 +} + +type modelSave struct { + name, author, message string + seq int + next bool + deadline time.Time + idle *time.Duration + maxDuration time.Duration + attached bool + sawForeign bool +} + +type modelWindow struct { + author string + objs []string + latest map[string]modelEdit + save *modelSave + idle time.Duration + noIdle bool + maxAt time.Time + lastWriteAt time.Time + charge int64 +} + +func (w *modelWindow) closeAt() (time.Time, windowFinalizeReason) { + if !w.noIdle && w.lastWriteAt.Add(w.idle).Before(w.maxAt) { + return w.lastWriteAt.Add(w.idle), windowFinalizeReasonIdleTimeout + } + return w.maxAt, windowFinalizeReasonMaxDuration +} + +// modelPending is a decided write the next push publishes. +type modelPending struct { + save *modelSave + author string + message string + resync bool + changes []modelEdit +} + +// modelCommit is a commit the worker must publish. A resync's message is not the model's to say. +// bootstrap is the folder's first commit, which also adds modelBootstrap. +type modelCommit struct { + author string + message string + resync bool + bootstrap bool + files map[string]string + seqs map[string]int +} + +type windowModel struct { + now time.Time + window *modelWindow + saves map[string]*modelSave + saveSeq int + editSeq int + + // remote is Git, known what the worker's checkout holds of it, cluster the live objects. + remote, known, cluster map[string]string + // history is the values each object held on the cluster, for undos. + history map[string][]string + + foreignUnseen, applyDue, hinted bool + // bootstrapped is whether Git holds modelBootstrap. + bootstrapped bool + + pending []modelPending + commits []modelCommit + closes []string + outcomes map[string]FinalizeOutcome + given map[string]bool + // landed is, per document, the acceptance seq of the last write a commit landed. + landed map[string]int +} + +func newWindowModel(now time.Time, docs map[string]string) *windowModel { + m := &windowModel{ + now: now, saves: map[string]*modelSave{}, + remote: map[string]string{}, known: map[string]string{}, cluster: map[string]string{}, + history: map[string][]string{}, outcomes: map[string]FinalizeOutcome{}, given: map[string]bool{}, + landed: map[string]int{}, + } + for obj, value := range docs { + m.remote[obj], m.known[obj], m.cluster[obj] = value, value, value + m.history[obj] = []string{value} + } + return m +} + +func (m *windowModel) windowCharge() int64 { + if m.window == nil { + return 0 + } + return m.window.charge +} + +// describeWindow is the open window: its author, how many objects it holds, the save it carries, when +// and why it closes, and what it is charged. +func (m *windowModel) describeWindow() string { + w := m.window + if w == nil { + return "none" + } + saved := "" + if w.save != nil { + saved = w.save.name + } + at, reason := w.closeAt() + return fmt.Sprintf("%s objs=%d save=%q closes in %s (%s) charge %d", w.author, len(w.objs), saved, + at.Sub(m.now), reason, w.charge) +} + +func (m *windowModel) waiting(s *modelSave) bool { return !s.attached && s.deadline.After(m.now) } + +func (m *windowModel) matches(s *modelSave) bool { + return m.window != nil && m.window.author == s.author +} + +// write is appendLiveEvent's rules for one accepted write. +func (m *windowModel) write(author string, edit modelEdit, pressure bool) { + m.editSeq++ + edit.seq = m.editSeq + m.cluster[edit.obj] = edit.value + if edit.value != "" { + m.history[edit.obj] = append(m.history[edit.obj], edit.value) + } + if m.window != nil && m.window.author != author { + m.close(windowFinalizeReasonIdentityChange) + } + opened := m.window == nil + if opened { + m.window = &modelWindow{ + author: author, latest: map[string]modelEdit{}, + idle: modelIdle, maxAt: m.now.Add(modelMax), + } + } + w := m.window + if old, ok := w.latest[edit.obj]; ok { + w.charge -= old.charge + } else { + w.objs = append(w.objs, edit.obj) + } + w.latest[edit.obj] = edit + w.charge += edit.charge + w.lastWriteAt = m.now + if opened { + m.noteForeign() + m.attachWaiting() + if m.window != w { + return + } + } + if pressure { + m.close(windowFinalizeReasonBufferLimit) + return + } + m.closeOrArm() +} + +// register is handleAttachCommitRequest's rules for a save. +func (m *windowModel) register(s *modelSave, attachTimeout time.Duration) { + for _, earlier := range m.saves { + if earlier.author != s.author { + continue + } + collecting := earlier.attached && m.window != nil && m.window.save == earlier + if m.waiting(earlier) || (collecting && !s.next) { + m.outcomes[s.name] = FinalizeSaveInProgress + return + } + } + m.saveSeq++ + s.seq = m.saveSeq + s.deadline = m.now.Add(attachTimeout) + m.saves[s.name] = s + if m.window == nil { + return + } + if !m.matches(s) { + s.sawForeign = true + return + } + if s.next { + m.close(windowFinalizeReasonAttachNext) + return + } + if m.window.save == nil { + m.attach(s) + } +} + +func (m *windowModel) attach(s *modelSave) { + w := m.window + w.save, s.attached = s, true + w.charge += attachCharge(s.message) + w.noIdle = s.idle == nil + if s.idle != nil { + w.idle = *s.idle + } + w.maxAt = m.now.Add(s.maxDuration) + w.lastWriteAt = m.now + m.closeOrArm() +} + +func (m *windowModel) closeOrArm() { + if m.window == nil { + return + } + if at, reason := m.window.closeAt(); !at.After(m.now) { + m.close(reason) + } +} + +func (m *windowModel) noteForeign() { + for _, s := range m.saves { + if m.waiting(s) && m.window != nil && !m.matches(s) { + s.sawForeign = true + } + } +} + +func (m *windowModel) attachWaiting() { + if m.window == nil || m.window.save != nil { + return + } + var first *modelSave + for _, s := range m.saves { + if m.waiting(s) && m.matches(s) && (first == nil || s.seq < first.seq) { + first = s + } + } + if first != nil { + m.attach(first) + } +} + +func (m *windowModel) expire() { + for _, s := range m.saves { + if s.attached || s.deadline.After(m.now) { + continue + } + outcome := FinalizeNoOpenWindow + if s.sawForeign { + outcome = FinalizeWindowMismatch + } + m.resolve(s, outcome) + } +} + +func (m *windowModel) resolve(s *modelSave, outcome FinalizeOutcome) { + delete(m.saves, s.name) + m.outcomes[s.name] = outcome +} + +// close decides the open window: the documents whose latest write differs from the checkout, in the +// order the window first saw them. A window that changes nothing and carries no save leaves no trace. +// A save is never trimmed: it decides every document it captured, so a replay onto a push the worker +// had not seen re-asserts them (API wins for a save). +func (m *windowModel) close(reason windowFinalizeReason) { + w := m.window + m.window = nil + m.closes = append(m.closes, string(reason)+" "+w.author) + var changes []modelEdit + for _, obj := range w.objs { + if edit := w.latest[obj]; edit.value != m.known[obj] || w.save != nil { + changes = append(changes, edit) + m.known[obj] = edit.value + } + } + if len(changes) == 0 && w.save == nil { + return + } + p := modelPending{save: w.save, author: w.author, changes: changes} + if w.save != nil { + p.message = w.save.message + } else { + p.message = generatedMessage(changes) + } + m.pending = append(m.pending, p) +} + +func generatedMessage(changes []modelEdit) string { + var b strings.Builder + plural := "s" + if len(changes) == 1 { + plural = "" + } + fmt.Fprintf(&b, "chore: sync %d resource%s\n\n", len(changes), plural) + for _, edit := range changes { + fmt.Fprintf(&b, "- [%s] v1/configmaps/default/%s\n", edit.op, edit.obj) + } + return b.String() +} + +// push publishes the pending writes. A remote that moved behind the worker's back is found by the +// push, which replays the decided changes onto it, and every applier is due for the foreign commit. +func (m *windowModel) push() { + if len(m.pending) == 0 { + return + } + for _, p := range m.pending { + commit := modelCommit{author: p.author, message: p.message, resync: p.resync, + files: map[string]string{}, seqs: map[string]int{}} + for _, edit := range p.changes { + if edit.value == m.remote[edit.obj] { + continue + } + m.remote[edit.obj] = edit.value + commit.files[edit.obj] = edit.value + commit.seqs[edit.obj] = edit.seq + } + if len(commit.files) == 0 { + if p.save != nil { + m.resolve(p.save, FinalizeAlreadyPresent) + } + continue + } + commit.bootstrap = !m.bootstrapped + m.bootstrapped = true + m.commits = append(m.commits, commit) + if p.save != nil { + m.resolve(p.save, FinalizeCommitted) + } + } + m.pending = nil + if m.foreignUnseen { + m.foreignUnseen, m.applyDue = false, true + } + m.known = maps.Clone(m.remote) +} + +// endWake is the loop's end of every wake. It reports whether an apply was sent: the driver's applier +// then puts the tip on the cluster. +func (m *windowModel) endWake() bool { + if len(m.saves) > 0 { + m.noteForeign() + m.expire() + m.attachWaiting() + } + if m.hinted { + m.hinted = false + if m.window != nil && m.window.save == nil { + m.close(windowFinalizeReasonForeignCommit) + } + } + m.push() + if !m.applyDue || len(m.saves) > 0 { + return false + } + if m.window != nil { + m.close(windowFinalizeReasonApplyDue) + } + m.push() + m.applyDue = false + // The apply is of the tip the worker knows: a push it has not found yet is not in it. + m.cluster = maps.Clone(m.known) + return true +} + +// resync closes the window, then commits, as the operator, every live object Git does not hold as +// it is. prune.mode is onEvent, so nothing is swept. An object it adds is adopted: an apply is due. +func (m *windowModel) resync() { + if m.window != nil { + m.close(windowFinalizeReasonResyncBeforeApply) + } + var changes []modelEdit + for _, obj := range modelObjects { + value := m.cluster[obj] + if value == "" || value == m.known[obj] { + continue + } + if m.known[obj] == "" { + m.applyDue = true + } + m.known[obj] = value + changes = append(changes, modelEdit{obj: obj, value: value}) + } + if len(changes) > 0 { + m.pending = append(m.pending, modelPending{author: DefaultCommitterName, resync: true, changes: changes}) + } +} + +// shutdown closes the window and publishes; a save still waiting for a window is given back. +func (m *windowModel) shutdown() { + if m.window != nil { + m.close(windowFinalizeReasonShutdown) + } + m.push() + for name := range m.saves { + m.given[name] = true + delete(m.saves, name) + } +} + +// windowModelRun drives one worker and one model with the same random inputs. +type windowModelRun struct { + t *testing.T + seed int64 + f *echoFixture + requester *recordingRequester + clock *testClock + logs *infoLines + rng *rand.Rand + m *windowModel + base string + knownTip string + start time.Time + trace []string + checked int + exited bool + values int + saves int +} + +func newWindowModelRun(t *testing.T, seed int64) *windowModelRun { + t.Helper() + docs := map[string]string{"x": "x0", "y": "y0", "z": "z0"} + f := newEchoFixture(t, "window-model-"+strconv.FormatInt(seed, 10), configv1alpha3.PruneOnEvent, docs) + requester := f.withClaim() + r := &windowModelRun{t: t, seed: seed, f: f, requester: requester, rng: rand.New(rand.NewSource(seed))} + r.clock = useTestClock(f.worker) + r.logs = captureInfo(f.worker) + f.loop.defaultWindow = commitWindowDefaults{idleTimeout: modelIdle, maxDuration: modelMax} + r.base = gitIn(t, f.repoDir, "rev-parse", "main") + r.knownTip = r.base + r.start = r.clock.Now() + r.m = newWindowModel(r.start, docs) + r.m.bootstrapped = gitIn(t, f.repoDir, "ls-tree", "--name-only", r.base, echoTarget+"/README.md") != "" + return r +} + +func (r *windowModelRun) failf(format string, args ...any) { + r.t.Helper() + tail := r.trace + if len(tail) > 40 { + tail = tail[len(tail)-40:] + } + r.t.Fatalf("seed %d: %s\nlast steps:\n %s", r.seed, fmt.Sprintf(format, args...), strings.Join(tail, "\n ")) +} + +func (r *windowModelRun) notef(format string, args ...any) { + r.trace = append(r.trace, fmt.Sprintf("t=%s ", r.clock.Now().Sub(r.start))+fmt.Sprintf(format, args...)) +} + +// wake runs one wake of the worker's loop on item, pushing at once whatever it decides, and the model's +// matching wake; then answers an apply sent, as the applier, and settles it in a wake of both. +func (r *windowModelRun) wake(item *WorkItem, model func()) { + r.f.loop.lastPushAt = time.Time{} + var released int64 + if item != nil { + released = r.f.loop.dispatch(*item) + } + r.f.loop.endWake(released) + if model != nil { + model() + } + r.applied(r.m.endWake()) +} + +func (r *windowModelRun) applied(modelSent bool) { + // Every push the worker has made is on main, unless a push from outside came after it unseen. + if !r.m.foreignUnseen { + r.knownTip = gitIn(r.t, r.f.repoDir, "rev-parse", "main") + } + sent := r.requester.taken() + if modelSent != (len(sent) > 0) { + r.failf("the model sent an apply: %v; the worker sent %d", modelSent, len(sent)) + } + for _, req := range sent { + if req.Commit != r.knownTip { + r.failf("an apply was sent for %s, not the tip the worker knows, %s", req.Commit, r.knownTip) + } + req.Report(ApplyResult{ID: req.ID, Outcome: ApplySucceeded, AppliedRevision: req.Commit}) + } + r.f.loop.lastPushAt = time.Time{} + r.f.loop.endWake(0) + if r.m.endWake() { + r.failf("the model sent a second apply in the settling wake") + } +} + +// armed is the timer the worker has armed to fire first, when, and whether it is the commit timer. Two +// due at the same instant fire in either order, as run's select picks. +func (r *windowModelRun) armed() (time.Time, bool, bool) { + loop := r.f.loop + commit, attach := loop.commitTimer != nil, loop.attachTimer != nil + switch { + case commit && attach && loop.commitTimerAt.Equal(loop.attachTimerAt): + return loop.commitTimerAt, r.rng.Intn(2) == 0, true + case commit && (!attach || loop.commitTimerAt.Before(loop.attachTimerAt)): + return loop.commitTimerAt, true, true + case attach: + return loop.attachTimerAt, false, true + } + return time.Time{}, false, false +} + +// commitWake is the wake the commit timer delivers, as run does. +func (r *windowModelRun) commitWake() { + r.notef("commit timer") + r.f.loop.lastPushAt = time.Time{} + r.f.loop.commitTimer = nil + r.f.loop.closeOrArmWindow() + r.f.loop.endWake(0) + r.m.closeOrArm() + r.applied(r.m.endWake()) +} + +// attachWake is the wake the attach timer delivers: endWake's serviceCommitRequests does its work. +func (r *windowModelRun) attachWake() { + r.notef("attach timer") + r.f.loop.lastPushAt = time.Time{} + r.f.loop.attachTimer = nil + r.f.loop.endWake(0) + r.applied(r.m.endWake()) +} + +// advance steps the clock by d, firing each timer the worker armed when it comes due. A deadline the +// worker armed no timer for passes unseen, as it would in run. +func (r *windowModelRun) advance(d time.Duration) { + r.notef("clock +%s", d) + target := r.clock.Now().Add(d) + for range 1000 { + at, commit, ok := r.armed() + if !ok || at.After(target) { + r.clock.advance(target.Sub(r.clock.Now())) + r.m.now = target + return + } + if at.After(r.clock.Now()) { + r.clock.advance(at.Sub(r.clock.Now())) + } + r.m.now = r.clock.Now() + if commit { + r.commitWake() + } else { + r.attachWake() + } + r.check("timer") + } + r.failf("a timer stayed due after its wake") +} + +func (r *windowModelRun) freshValue(obj string) string { + r.values++ + return obj + strconv.Itoa(r.values) +} + +func (r *windowModelRun) event(author, obj, value string) Event { + if value == "" { + event := configMapTargetEvent(obj, author, echoTarget) + event.Operation = "DELETE" + event.Object = nil + return event + } + event := echoEvent(obj, value) + event.UserInfo.Username = author + return event +} + +// write is one accepted live write: a new value, an undo to an earlier one, the value the object +// already holds, or a delete. +func (r *windowModelRun) write(pressure bool) { + author := modelAuthors[r.rng.Intn(len(modelAuthors))] + obj := modelObjects[r.rng.Intn(len(modelObjects))] + current := r.m.cluster[obj] + value, kind := r.freshValue(obj), "write" + switch roll := r.rng.Intn(20); { + case roll < 4 && len(r.m.history[obj]) > 1: + value, kind = r.m.history[obj][r.rng.Intn(len(r.m.history[obj])-1)], "undo" + case roll < 7 && current != "": + value, kind = current, "no-op" + case roll < 10 && current != "": + value, kind = "", "delete" + } + event := r.event(author, obj, value) + r.notef("%s %s %s=%q pressure=%v", author, kind, obj, value, pressure) + edit := modelEdit{obj: obj, value: value, op: event.Operation, charge: eventRetainedCharge(&event)} + item := WorkItem{Request: &WriteRequest{Events: []Event{event}}} + r.f.loop.lastPushAt = time.Time{} + if pressure { + limit := r.f.worker.branchBufferMaxBytes + r.f.worker.branchBufferMaxBytes = 1 + r.f.loop.dispatch(item) + r.f.worker.branchBufferMaxBytes = limit + } else { + r.f.loop.dispatch(item) + } + r.f.loop.endWake(0) + r.m.write(author, edit, pressure) + r.applied(r.m.endWake()) +} + +func (r *windowModelRun) save() { + r.saves++ + author := modelAuthors[r.rng.Intn(len(modelAuthors))] + name := "save-" + strconv.Itoa(r.saves) + req := &AttachCommitRequest{ + Namespace: "default", Name: name, UID: "uid-" + name, + Author: author, GitTargetName: echoTarget, GitTargetNamespace: "default", + Message: fmt.Sprintf("feat: %s by %s", name, author), + Attach: configv1alpha3.AttachCurrentOrNext, + } + if r.rng.Intn(4) == 0 { + req.Attach = configv1alpha3.AttachNext + } + req.AttachTimeout = []time.Duration{0, 5 * time.Second, 20 * time.Second}[r.rng.Intn(3)] + req.MaxDuration = []time.Duration{0, 15 * time.Second, 40 * time.Second}[r.rng.Intn(3)] + if r.rng.Intn(2) == 0 { + req.IdleTimeout = durationPtr([]time.Duration{0, 5 * time.Second}[r.rng.Intn(2)]) + } + r.notef("%s registers %s attach=%s attachTimeout=%s maxDuration=%s idle=%v", author, name, req.Attach, + req.AttachTimeout, req.MaxDuration, req.IdleTimeout) + s := &modelSave{name: name, author: author, message: req.Message, next: req.Attach == configv1alpha3.AttachNext, + idle: req.IdleTimeout, maxDuration: req.MaxDuration} + r.wake(&WorkItem{Attach: req}, func() { r.m.register(s, req.AttachTimeout) }) +} + +// foreign pushes a commit from outside and reports it, as a Git host's push notice does. +func (r *windowModelRun) foreign() { + obj := modelObjects[r.rng.Intn(len(modelObjects))] + r.values++ + value := "outside" + strconv.Itoa(r.values) + r.notef("outside push %s=%q, hinted", obj, value) + r.f.pushFromOutside(obj, value) + r.m.remote[obj] = value + r.m.foreignUnseen = true + r.f.loop.pushHinted = true + r.wake(nil, func() { r.m.hinted = true }) +} + +// apply makes an apply due for the folder's applier (a Repair). +func (r *windowModelRun) apply() { + r.notef("apply due") + r.f.echoApplier().repair = true + r.wake(nil, func() { r.m.applyDue = true }) +} + +func (r *windowModelRun) resync() { + r.notef("resync") + var desired []manifestanalyzer.DesiredResource + for _, obj := range modelObjects { + if value := r.m.cluster[obj]; value != "" { + desired = append(desired, echoDesired(obj, value)) + } + } + req := &ResyncRequest{Desired: desired, GitTargetName: echoTarget, GitTargetNamespace: "default", + Result: make(chan ResyncResult, 1)} + item := resyncItem(req) + r.wake(&item, r.m.resync) + if result := <-req.Result; result.Err != nil { + r.failf("resync: %v", result.Err) + } +} + +func (r *windowModelRun) shutdown() { + r.notef("shutdown") + r.f.loop.lastPushAt = time.Time{} + r.f.loop.handleShutdown() + r.f.worker.settleCommitRequestsAfterExit() + r.exited = true + r.m.shutdown() +} + +func (r *windowModelRun) step() { + switch roll := r.rng.Intn(100); { + case roll < 40: + r.write(false) + case roll < 44: + r.write(true) + case roll < 62: + r.advance(time.Duration(1+r.rng.Intn(25)) * time.Second) + case roll < 76: + r.save() + case roll < 83: + r.foreign() + case roll < 90: + r.apply() + case roll < 96 && !r.m.foreignUnseen && !r.m.applyDue: + r.resync() + default: + r.write(false) + } +} + +func (r *windowModelRun) run(steps int) { + for range steps { + r.step() + r.check("step") + } + if r.rng.Intn(2) == 0 { + r.shutdown() + r.check("shutdown") + } else { + r.advance(3 * modelMax) + } + r.drained() +} + +var closeLine = regexp.MustCompile(`"msg"="Finalizing open commit window" "reason"="([^"]+)" "windowAuthor"="([^"]*)"`) + +// check compares the worker with the model after a step. +func (r *windowModelRun) check(what string) { + r.t.Helper() + r.checkWindow(what) + r.checkTimers(what) + r.checkSaves(what) + r.checkCloses(what) + r.checkCommits(what) +} + +// checkWindow compares the open window and what is retained. +func (r *windowModelRun) checkWindow(what string) { + r.t.Helper() + loop, m := r.f.loop, r.m + if got, want := r.workerWindow(), m.describeWindow(); got != want { + r.failf("%s: the open window differs:\n worker %s\n model %s", what, got, want) + } + if got, want := loop.totalRetainedBytes(), m.windowCharge(); got != want || loop.commitWindows.windowBytes != want { + r.failf("%s: retained %d (window %d), the model's window charge is %d", what, got, + loop.commitWindows.windowBytes, want) + } + if len(loop.pendingWrites) != 0 { + r.failf("%s: %d writes still pending after a wake that pushed at once", what, len(loop.pendingWrites)) + } +} + +// checkTimers compares the timers the worker has armed with the model's deadlines: the commit timer +// for the open window's close, the attach timer for the earliest deadline of a save still waiting. +func (r *windowModelRun) checkTimers(what string) { + r.t.Helper() + if r.exited { + return + } + loop, m := r.f.loop, r.m + var commitAt, attachAt time.Time + if m.window != nil { + commitAt, _ = m.window.closeAt() + } + for _, s := range m.saves { + if !s.attached && (attachAt.IsZero() || s.deadline.Before(attachAt)) { + attachAt = s.deadline + } + } + got := r.describeTimer(loop.commitTimer != nil, loop.commitTimerAt) + ", " + + r.describeTimer(loop.attachTimer != nil, loop.attachTimerAt) + want := r.describeTimer(!commitAt.IsZero(), commitAt) + ", " + r.describeTimer(!attachAt.IsZero(), attachAt) + if got != want { + r.failf("%s: the armed timers (commit, attach) differ:\n worker %s\n model %s", what, got, want) + } +} + +func (r *windowModelRun) describeTimer(armed bool, at time.Time) string { + if !armed { + return "none" + } + return "in " + at.Sub(r.m.now).String() +} + +// workerWindow describes the worker's open window as describeWindow does the model's. +func (r *windowModelRun) workerWindow() string { + w := r.f.loop.openWindow + if w == nil { + return "none" + } + saved := "" + if w.pendingCR != nil { + saved = w.pendingCR.Name + } + at, reason := w.closeAt() + return fmt.Sprintf("%s objs=%d save=%q closes in %s (%s) charge %d", w.Author, len(w.pathOrder), saved, + at.Sub(r.m.now), reason, w.bytes) +} + +// checkSaves compares the registered saves, and every save's outcome and how often it settled. +func (r *windowModelRun) checkSaves(what string) { + r.t.Helper() + loop, m := r.f.loop, r.m + var registered, modelRegistered []string + for id, pcr := range loop.pendingCRs { + registered = append(registered, fmt.Sprintf("%s attached=%v deadline=%s", id.Name, pcr.attached, + pcr.attachDeadline.Sub(m.now))) + } + for name, s := range m.saves { + modelRegistered = append(modelRegistered, fmt.Sprintf("%s attached=%v deadline=%s", name, s.attached, + s.deadline.Sub(m.now))) + } + slices.Sort(registered) + slices.Sort(modelRegistered) + // An exited loop keeps what it held; settleCommitRequestsAfterExit gave the waiting saves back. + if !r.exited && !slices.Equal(registered, modelRegistered) { + r.failf("%s: registered saves differ:\n worker %v\n model %v", what, registered, modelRegistered) + } + for i := 1; i <= r.saves; i++ { + name := "save-" + strconv.Itoa(i) + res, resolved := r.f.worker.LookupCommitRequestOutcome("default", name, "uid-"+name) + want, modelResolved := m.outcomes[name] + if resolved != modelResolved || (resolved && (res.Outcome != want || res.Err != nil)) { + r.failf("%s: %s settled %v as %q (err %v); the model says %v as %q", what, name, resolved, + res.Outcome, res.Err, modelResolved, want) + } + if n := r.settlements(name); n > 1 || (n == 1) != modelResolved { + r.failf("%s: %s settled %d times", what, name, n) + } + } +} + +// checkCloses compares the windows closed so far, by reason and author, in order. +func (r *windowModelRun) checkCloses(what string) { + r.t.Helper() + var closes []string + for _, line := range r.logs.matching("Finalizing open commit window") { + match := closeLine.FindStringSubmatch(line) + if match == nil { + r.failf("unparsed close line %s", line) + } + closes = append(closes, match[1]+" "+match[2]) + } + if !slices.Equal(closes, r.m.closes) { + r.failf("%s: close reasons differ:\n worker %v\n model %v", what, closes, r.m.closes) + } +} + +// settlements counts the times a save was settled: resolved, or refused at registration. +func (r *windowModelRun) settlements(name string) int { + n := 0 + for _, line := range r.logs.matching(`"request"="default/` + name + `"`) { + if strings.Contains(line, `"msg"="CommitRequest resolved"`) || + strings.Contains(line, `"msg"="CommitRequest refused`) { + n++ + } + } + return n +} + +// checkCommits compares the commits published since the last check with the model's. +func (r *windowModelRun) checkCommits(what string) { + r.t.Helper() + var published []string + for _, line := range splitLines(gitIn(r.t, r.f.repoDir, "log", "--reverse", "--first-parent", + "--format=%H %an", r.base+"..main")) { + hash, author, _ := strings.Cut(line, " ") + if author != "Outside" { + published = append(published, hash) + } + } + if len(published) != len(r.m.commits) { + r.failf("%s: the worker published %d commits; the model %d", what, len(published), len(r.m.commits)) + } + for i := r.checked; i < len(published); i++ { + r.checkCommit(what, published[i], r.m.commits[i]) + } + r.checked = len(published) +} + +func (r *windowModelRun) checkCommit(what, hash string, want modelCommit) { + r.t.Helper() + author := gitIn(r.t, r.f.repoDir, "log", "-1", "--format=%an", hash) + message := gitIn(r.t, r.f.repoDir, "log", "-1", "--format=%B", hash) + if author != want.author || (!want.resync && strings.TrimSpace(message) != strings.TrimSpace(want.message)) { + r.failf("%s: commit %s by %q %q; the model's by %q %q", what, hash[:8], author, message, want.author, + want.message) + } + paths := splitLines(gitIn(r.t, r.f.repoDir, "diff-tree", "-r", "--name-only", "--no-commit-id", hash)) + var wantPaths []string + for obj := range want.files { + wantPaths = append(wantPaths, echoFile(obj)) + } + if want.bootstrap { + for _, name := range modelBootstrap { + wantPaths = append(wantPaths, echoTarget+"/"+name) + r.checkBootstrap(what, hash, name) + } + } + slices.Sort(paths) + slices.Sort(wantPaths) + if !slices.Equal(paths, wantPaths) { + r.failf("%s: commit %s by %s changes %v; the model's changes %v", what, hash[:8], author, paths, wantPaths) + } + for obj, value := range want.files { + r.checkDocument(what, hash, obj, value) + } + for obj, seq := range want.seqs { + if seq <= r.m.landed[obj] { + r.failf("%s: commit %s lands write %d of %s after write %d", what, hash[:8], seq, obj, r.m.landed[obj]) + } + r.m.landed[obj] = seq + } +} + +// checkDocument compares the document obj at rev, parsed, with the ConfigMap obj holding value; value +// "" is no document. YAML 1.2, so a bare y is the name y. +func (r *windowModelRun) checkDocument(what, rev, obj, value string) { + r.t.Helper() + var got, want map[string]any + listed := gitIn(r.t, r.f.repoDir, "ls-tree", "--name-only", rev, echoFile(obj)) + if listed != "" { + if err := yaml.Unmarshal([]byte(gitIn(r.t, r.f.repoDir, "show", rev+":"+echoFile(obj))), &got); err != nil { + r.failf("%s: %s at %s: %v", what, obj, rev, err) + } + } + if value != "" { + want = map[string]any{ + "apiVersion": "v1", "kind": "ConfigMap", + "metadata": map[string]any{"name": obj, "namespace": "default"}, + "data": map[string]any{"key": value}, + } + } + if !reflect.DeepEqual(got, want) { + r.failf("%s: %s at %s is %v; the model's is %v", what, obj, rev, got, want) + } +} + +// checkBootstrap compares the bootstrap file name at rev with the worker's template. +func (r *windowModelRun) checkBootstrap(what, rev, name string) { + r.t.Helper() + want, err := readBootstrapTemplateContent(name, pathBootstrapOptions{Enabled: true}) + if err != nil { + r.failf("%s: %v", what, err) + } + if got := gitIn(r.t, r.f.repoDir, "show", rev+":"+echoTarget+"/"+name); got != strings.TrimSpace(string(want)) { + r.failf("%s: %s at %s is not the template's:\n%s", what, name, rev, got) + } +} + +// drained checks the worker once nothing is open, waiting or pending. +func (r *windowModelRun) drained() { + r.t.Helper() + r.check("drained") + if r.m.window != nil || len(r.m.saves) != 0 || len(r.m.pending) != 0 { + r.failf("the model is not drained") + } + // An exited loop is dropped with the saves it gave back still counted in heldBytes; what it retains + // for Git must be nothing. + held := r.f.loop.heldBytes() + if r.exited { + held = 0 + } + if held != 0 || r.f.loop.totalRetainedBytes() != 0 { + r.failf("drained, the worker still charges %d (retained %d)", held, r.f.loop.totalRetainedBytes()) + } + r.checkTree() + for name := range r.m.given { + if _, resolved := r.f.worker.LookupCommitRequestOutcome("default", name, "uid-"+name); resolved { + r.failf("%s was given back at shutdown, and has an outcome", name) + } + if phase := r.f.worker.LookupCommitRequestPhase("default", name, "uid-"+name); phase != "" { + r.failf("%s was given back at shutdown, and is still %s", name, phase) + } + } +} + +// checkTree checks that Git is the base tree with the model's documents and, once the folder is +// bootstrapped, the template's files: nothing else was added, changed or removed. +func (r *windowModelRun) checkTree() { + r.t.Helper() + present := map[string]bool{} + for _, path := range splitLines(gitIn(r.t, r.f.repoDir, "ls-tree", "-r", "--name-only", r.base)) { + present[path] = true + } + for _, obj := range modelObjects { + present[echoFile(obj)] = r.m.remote[obj] != "" + r.checkDocument("drained", "main", obj, r.m.remote[obj]) + } + others := []string{"diff", "--name-only", r.base, "main", "--", ".", + ":(exclude)" + echoTarget + "/default/configmaps"} + for _, name := range modelBootstrap { + others = append(others, ":(exclude)"+echoTarget+"/"+name) + if r.m.bootstrapped { + present[echoTarget+"/"+name] = true + r.checkBootstrap("drained", "main", name) + } + } + var want []string + for path, ok := range present { + if ok { + want = append(want, path) + } + } + slices.Sort(want) + if tree := splitLines(gitIn(r.t, r.f.repoDir, "ls-tree", "-r", "--name-only", "main")); !slices.Equal(tree, want) { + r.failf("drained, Git holds %v; the model %v", tree, want) + } + if changed := gitIn(r.t, r.f.repoDir, others...); changed != "" { + r.failf("drained, Git changed the base's %s", changed) + } +} + +func envInt(name string, fallback int) int { + if v, err := strconv.Atoi(os.Getenv(name)); err == nil && v > 0 { + return v + } + return fallback +} + +// TestWindowModel_TheWorkerFollowsTheReferenceModel runs random input sequences through a worker and +// the model, and checks every step. +func TestWindowModel_TheWorkerFollowsTheReferenceModel(t *testing.T) { + seeds, steps := envInt("WINDOW_MODEL_SEEDS", 6), envInt("WINDOW_MODEL_STEPS", 40) + if testing.Short() { + seeds = 2 + } + first := int64(1) + if seed := envInt("WINDOW_MODEL_SEED", 0); seed > 0 { + first, seeds = int64(seed), 1 + } + for seed := first; seed < first+int64(seeds); seed++ { + t.Run("seed-"+strconv.FormatInt(seed, 10), func(t *testing.T) { + newWindowModelRun(t, seed).run(steps) + }) + } +} diff --git a/internal/git/window_set.go b/internal/git/window_set.go new file mode 100644 index 00000000..71c8aa44 --- /dev/null +++ b/internal/git/window_set.go @@ -0,0 +1,121 @@ +// SPDX-License-Identifier: Apache-2.0 + +package git + +import "time" + +// commitWindows is the set of commit windows a branch worker holds open, and what they are charged. +// Every path that reads a window says which one it means, or all of them. +// +// It allows one window. A write that cannot join the open window closes it first, with close reason +// identity_change, before it opens its own, so commits land in the order their windows opened. +// docs/design/concurrent-author-windows.md lets several stay open side by side; windows that one +// trigger closes then close oldest first, by the sequence number each took when it opened, never by +// map order or by time. +// +// The loop embeds the set, so loop.openWindow still names the one window it holds. +type commitWindows struct { + // openWindow is the open window, or nil. + openWindow *openWindow + // windowBytes is what the open windows are charged: their events, and the message of the request + // attached to each. It is the sum of every open window's charge. + windowBytes int64 + // windowSeq numbers the windows in the order they opened. + windowSeq uint64 +} + +// all returns the open windows, oldest first. +func (s *commitWindows) all() []*openWindow { + if s.openWindow == nil { + return nil + } + return []*openWindow{s.openWindow} +} + +// anyOpen reports whether a window is open. +func (s *commitWindows) anyOpen() bool { + return s.openWindow != nil +} + +// isOpen reports whether w is still one of the open windows. +func (s *commitWindows) isOpen(w *openWindow) bool { + return w != nil && s.openWindow == w +} + +// open opens a window for the write e at now, numbered after every window opened before it. The +// caller has closed every window e cannot be admitted beside (blocking), so the set is empty here. +func (s *commitWindows) open(e Event, writer eventContentWriter, now time.Time) *openWindow { + s.windowSeq++ + w := newOpenWindow(e, writer) + w.seq = s.windowSeq + w.openedAt = now + s.openWindow = w + return w +} + +// forWrite returns the open window the write e joins, or nil when it must open one. +func (s *commitWindows) forWrite(e Event) *openWindow { + for _, w := range s.all() { + if w.canAppend(e) { + return w + } + } + return nil +} + +// blocking returns the open windows the write e cannot be admitted beside, oldest first: every +// window it cannot join. +func (s *commitWindows) blocking(e Event) []*openWindow { + var out []*openWindow + for _, w := range s.all() { + if !w.canAppend(e) { + out = append(out, w) + } + } + return out +} + +// forRequest returns the open window the request may attach to, or nil. +func (s *commitWindows) forRequest(p *pendingCommitRequest) *openWindow { + for _, w := range s.all() { + if p.matchesWindow(w) { + return w + } + } + return nil +} + +// foreignTo returns the oldest open window the request may not attach to, or nil. +func (s *commitWindows) foreignTo(p *pendingCommitRequest) *openWindow { + for _, w := range s.all() { + if !p.matchesWindow(w) { + return w + } + } + return nil +} + +// carrying returns the open window the request is attached to, or nil. +func (s *commitWindows) carrying(id commitRequestID) *openWindow { + for _, w := range s.all() { + if w.pendingCR != nil && *w.pendingCR == id { + return w + } + } + return nil +} + +// charge adds n bytes, which may be negative, to w's charge and to the set's. +func (s *commitWindows) charge(w *openWindow, n int64) { + w.bytes += n + s.windowBytes += n +} + +// remove forgets w, and releases what it was charged. +func (s *commitWindows) remove(w *openWindow) { + if !s.isOpen(w) { + return + } + s.openWindow = nil + s.windowBytes -= w.bytes +} diff --git a/internal/git/window_set_test.go b/internal/git/window_set_test.go new file mode 100644 index 00000000..9c84a4a1 --- /dev/null +++ b/internal/git/window_set_test.go @@ -0,0 +1,65 @@ +// SPDX-License-Identifier: Apache-2.0 + +package git + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestCommitWindows_WindowsAreNumberedInTheOrderTheyOpen(t *testing.T) { + var set commitWindows + opened := time.Now() + + first := set.open(configMapTargetEvent("a", "alice", "team-a"), nil, opened) + assert.Equal(t, opened, first.openedAt) + assert.Equal(t, []*openWindow{first}, set.all()) + set.remove(first) + assert.Empty(t, set.all()) + + second := set.open(configMapTargetEvent("b", "bob", "team-a"), nil, opened) + assert.Greater(t, second.seq, first.seq, "a window opened later is numbered after it") +} + +func TestCommitWindows_ASetSaysWhichWindowAWriteOrARequestBelongsTo(t *testing.T) { + var set commitWindows + window := set.open(configMapTargetEvent("a", "alice", "team-a"), nil, time.Now()) + + alice := configMapTargetEvent("b", "alice", "team-a") + bob := configMapTargetEvent("b", "bob", "team-a") + assert.Same(t, window, set.forWrite(alice)) + assert.Empty(t, set.blocking(alice)) + assert.Nil(t, set.forWrite(bob)) + assert.Equal(t, []*openWindow{window}, set.blocking(bob), "bob's write must close alice's window first") + + aliceSave := &pendingCommitRequest{author: "alice", gitTargetName: "team-a", gitTargetNamespace: "default"} + bobSave := &pendingCommitRequest{author: "bob", gitTargetName: "team-a", gitTargetNamespace: "default"} + assert.Same(t, window, set.forRequest(aliceSave)) + assert.Nil(t, set.foreignTo(aliceSave)) + assert.Nil(t, set.forRequest(bobSave)) + assert.Same(t, window, set.foreignTo(bobSave)) + + id := commitRequestID{Namespace: "default", Name: "save", UID: "uid-save"} + assert.Nil(t, set.carrying(id)) + window.pendingCR = &id + assert.Same(t, window, set.carrying(id)) +} + +func TestCommitWindows_ClosingAWindowReleasesExactlyItsCharge(t *testing.T) { + var set commitWindows + window := set.open(configMapTargetEvent("a", "alice", "team-a"), nil, time.Now()) + set.charge(window, 100) + set.charge(window, -30) + require.Equal(t, int64(70), window.bytes) + require.Equal(t, int64(70), set.windowBytes) + + set.remove(window) + assert.False(t, set.anyOpen()) + assert.Zero(t, set.windowBytes) + + set.remove(window) + assert.Zero(t, set.windowBytes, "a window already closed has nothing left to release") +} diff --git a/internal/git/write_gate.go b/internal/git/write_gate.go index f849c7f6..f9b70b9f 100644 --- a/internal/git/write_gate.go +++ b/internal/git/write_gate.go @@ -16,7 +16,7 @@ import ( // // - a request no window reached (expireWaitingCommitRequests, before the record it may make); // - the record itself (buildRequestRecordWrite); -// - a window a request is attached to (finalizeOpenWindowWithReason before it builds the write, +// - a window a request is attached to (finalizeWindow before it builds the write, // and gateBuiltWindow after, right before the commit). A render-fidelity refusal seen at // either look stops the whole window; suspension fails only the request, and the write still // runs so the target's scan stays fresh.