From 91d5d45ff2fc18e3fda0cf7cf352da01e01705ff Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Tue, 27 Oct 2026 20:20:11 +0000 Subject: [PATCH 1/8] refactor(git): read the commit windows' timers from the worker's clock The idle and max-duration deadlines, the opening time a window's duration is measured from, and a CommitRequest's attach deadline now read BranchWorker.now(), the clock the retry schedule already uses. A test can step it past a deadline and deliver the timer's wake by hand, with no sleep and no rewound deadline. The timers stay real; unrelated time.Now calls (push cooldown, apply windows, outcome retention) are unchanged. Co-Authored-By: Claude Opus 5.5 --- internal/git/branch_worker.go | 12 ++- internal/git/commit_request_attach_loop.go | 14 +-- internal/git/commit_window_clock_test.go | 117 +++++++++++++++++++++ internal/git/parent_recovery.go | 4 +- 4 files changed, 134 insertions(+), 13 deletions(-) create mode 100644 internal/git/commit_window_clock_test.go diff --git a/internal/git/branch_worker.go b/internal/git/branch_worker.go index ba477525..522fc59c 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. @@ -1327,13 +1328,14 @@ func (l *branchWorkerEventLoop) appendLiveEvent(event Event) { "gitTarget", event.GitTargetNamespace+"/"+event.GitTargetName, "resource", event.Identifier.String()) l.openWindow = newOpenWindow(event, l.w.contentWriter) + l.openWindow.openedAt = 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{ idle: timers.idleTimeout, - maxAt: time.Now().Add(timers.maxDuration), + maxAt: l.w.now().Add(timers.maxDuration), } opened = true } @@ -1341,7 +1343,7 @@ func (l *branchWorkerEventLoop) appendLiveEvent(event Event) { if replaced, ok := l.openWindow.add(event); ok { l.windowBytes -= eventRetainedCharge(&replaced) } - l.openWindow.lastWriteAt = time.Now() + l.openWindow.lastWriteAt = l.w.now() l.windowBytes += eventRetainedCharge(&event) if opened { @@ -1413,7 +1415,7 @@ func (l *branchWorkerEventLoop) closeOrArmWindow() { return } closeAt, reason := l.openWindow.closeAt() - delay := time.Until(closeAt) + delay := closeAt.Sub(l.w.now()) if delay <= 0 { l.finalizeOpenWindowWithReason(reason) return @@ -2906,7 +2908,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..9aae4080 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, @@ -265,7 +265,7 @@ func (l *branchWorkerEventLoop) handleAttachCommitRequest(req *AttachCommitReque 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 @@ -323,7 +323,7 @@ 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) { continue @@ -345,7 +345,7 @@ func (l *branchWorkerEventLoop) attachWaitingCommitRequests() { if l.openWindow == nil || l.openWindow.pendingCR != nil { return } - now := time.Now() + now := l.w.now() var first *pendingCommitRequest for _, pcr := range l.pendingCRs { if !pcr.waiting(now) || !pcr.matchesWindow(l.openWindow) { @@ -365,7 +365,7 @@ func (l *branchWorkerEventLoop) attachWaitingCommitRequests() { // 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() + now := l.w.now() l.openWindow.pendingMessage = pcr.message id := pcr.id l.openWindow.pendingCR = &id @@ -390,7 +390,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 +534,7 @@ func (l *branchWorkerEventLoop) rearmAttachTimer() { l.stopAttachTimer() return } - delay := time.Until(earliest) + delay := earliest.Sub(l.w.now()) if delay < 0 { delay = 0 } diff --git a/internal/git/commit_window_clock_test.go b/internal/git/commit_window_clock_test.go new file mode 100644 index 00000000..c917857f --- /dev/null +++ b/internal/git/commit_window_clock_test.go @@ -0,0 +1,117 @@ +// 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 the timer would have, and the window or the waiting request is due, +// with no sleep and no rewound deadline. + +// timerWake is the wake the commit and attach timers deliver when they fire, as run does. +func timerWake(loop *branchWorkerEventLoop) { + loop.commitTimer = nil + loop.closeOrArmWindow() + 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/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() From d6639e597897aeeef60a3aa831e485608cbb6ea7 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Tue, 27 Oct 2026 20:23:36 +0000 Subject: [PATCH 2/8] refactor(git): hold the commit windows in a set that allows one The window bookkeeping moves behind commitWindows: the open windows in opening order, each numbered by an opening sequence, and the charge each one carries. Every path says which window it means (the one a write joins or must close, the one a request attaches to or carries, the one whose deadline passed) or all of them (a resync, an apply flush, the byte cap, shutdown). The commit timer is armed for the earliest deadline of the open windows. The set allows exactly one window, so close reasons, close order, retained-byte charges and request settlement are what they were. The loop embeds the set, so loop.openWindow and loop.windowBytes still name the one window and its charge, and every existing test compiles and passes unchanged. Co-Authored-By: Claude Opus 5.5 --- .../window-field-ownership-and-apply-hold.md | 2 +- internal/git/applier_claim.go | 19 ++- internal/git/branch_worker.go | 122 ++++++++++-------- internal/git/commit_request_attach_loop.go | 82 ++++++------ internal/git/intake.go | 2 +- internal/git/open_window.go | 13 +- internal/git/refresh.go | 2 +- internal/git/resync_flush.go | 4 +- internal/git/window_set.go | 121 +++++++++++++++++ internal/git/window_set_test.go | 65 ++++++++++ internal/git/write_gate.go | 2 +- 11 files changed, 320 insertions(+), 114 deletions(-) create mode 100644 internal/git/window_set.go create mode 100644 internal/git/window_set_test.go 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/branch_worker.go b/internal/git/branch_worker.go index 522fc59c..eb946726 100644 --- a/internal/git/branch_worker.go +++ b/internal/git/branch_worker.go @@ -1009,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). @@ -1018,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. @@ -1226,7 +1226,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) } @@ -1250,10 +1250,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. @@ -1305,7 +1305,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) @@ -1313,38 +1313,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) - l.openWindow.openedAt = l.w.now() + 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: 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 = l.w.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 @@ -1354,7 +1354,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 } } @@ -1406,21 +1406,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 := closeAt.Sub(l.w.now()) - if delay <= 0 { - l.finalizeOpenWindowWithReason(reason) + if next.IsZero() { + l.stopCommitTimer() return } - l.resetCommitTimer(delay) + l.resetCommitTimer(next.Sub(now)) } func (l *branchWorkerEventLoop) handleShutdown() { @@ -1496,34 +1502,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 } @@ -1532,7 +1542,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) @@ -1545,18 +1555,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", @@ -1590,12 +1599,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 { @@ -1605,11 +1614,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}) } diff --git a/internal/git/commit_request_attach_loop.go b/internal/git/commit_request_attach_loop.go index 9aae4080..d2fda1ac 100644 --- a/internal/git/commit_request_attach_loop.go +++ b/internal/git/commit_request_attach_loop.go @@ -232,34 +232,34 @@ 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( @@ -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 := 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 := l.w.now() - var first *pendingCommitRequest - for _, pcr := range l.pendingCRs { - if !pcr.waiting(now) || !pcr.matchesWindow(l.openWindow) { + 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) { +// 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() - l.openWindow.pendingMessage = pcr.message + 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, 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..fa8152fe 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 @@ -57,6 +56,12 @@ type openWindow struct { // 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. 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_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. From 2c7285e24083050b0530e1e57319c5535a06d72f Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Tue, 27 Oct 2026 20:47:36 +0000 Subject: [PATCH 3/8] test(git): check the branch worker against a reference model of its commit windows A small, independent implementation of today's single-window rules runs beside a worker on a real Git server, fed the same random inputs: writes by several authors (new values, undos, no-ops, deletes), saves (attach_next, zero and non-zero attach timeouts and maxDurations, idle timeouts), outside pushes with a push hint, apply flushes, resyncs, buffer pressure, clock steps and shutdown. After every step the open window, the registered saves, the retained-byte charge, the close reasons and every save's outcome must be the model's; every published commit must have the model's author, message and documents, with no older write landing after a newer one; every save settles exactly once; and drained, Git holds the model's tree with nothing charged. The model is how the next stage, several windows open at once, will be judged (docs/design/concurrent-author-windows.md). A withheld late apply rides the next edit of the object it overwrote, and the whole-object window commits it under that author's name. The failing test is written and skipped (GITOPS_REVERSER_KNOWN_BUGS=1 runs it), and the bug is recorded in docs/TODO.md. Co-Authored-By: Claude Opus 5.5 --- docs/TODO.md | 7 + internal/git/apply_window_carry_test.go | 102 +++ internal/git/window_model_test.go | 965 ++++++++++++++++++++++++ 3 files changed, 1074 insertions(+) create mode 100644 internal/git/apply_window_carry_test.go create mode 100644 internal/git/window_model_test.go 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/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/window_model_test.go b/internal/git/window_model_test.go new file mode 100644 index 00000000..d5c0e227 --- /dev/null +++ b/internal/git/window_model_test.go @@ -0,0 +1,965 @@ +// SPDX-License-Identifier: Apache-2.0 + +package git + +import ( + "fmt" + "maps" + "math/rand" + "os" + "regexp" + "slices" + "strconv" + "strings" + "testing" + "time" + + 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 +// 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, and the documents it changes. 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 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 each deadline in turn and delivers the wake the timer would. +// +// 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"} +) + +// 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. +type modelCommit struct { + author string + message string + resync 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 + + 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 + } + 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 +} + +func (m *windowModel) timerWake() bool { + m.closeOrArm() + return m.endWake() +} + +// 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) + 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") + } +} + +// timerWake is the wake the commit and attach timers deliver. +func (r *windowModelRun) timerWake() { + r.f.loop.lastPushAt = time.Time{} + r.f.loop.commitTimer = nil + r.f.loop.closeOrArmWindow() + r.f.loop.endWake(0) + r.applied(r.m.timerWake()) +} + +// nextDeadline is the earliest timer the worker has armed: a window's close, or a waiting save's +// attach deadline. +func (r *windowModelRun) nextDeadline() (time.Time, bool) { + var next time.Time + for _, window := range r.f.loop.commitWindows.all() { + if at, _ := window.closeAt(); next.IsZero() || at.Before(next) { + next = at + } + } + for _, pcr := range r.f.loop.pendingCRs { + if !pcr.attached && (next.IsZero() || pcr.attachDeadline.Before(next)) { + next = pcr.attachDeadline + } + } + return next, !next.IsZero() +} + +func (r *windowModelRun) advance(d time.Duration) { + r.notef("clock +%s", d) + target := r.clock.Now().Add(d) + for range 1000 { + next, ok := r.nextDeadline() + if !ok || next.After(target) { + r.clock.advance(target.Sub(r.clock.Now())) + r.m.now = target + return + } + if next.After(r.clock.Now()) { + r.clock.advance(next.Sub(r.clock.Now())) + } + r.m.now = r.clock.Now() + r.timerWake() + r.check("timer") + } + r.failf("a deadline 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.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)) + } +} + +// 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) + } + files := map[string]string{} + for _, path := range splitLines(gitIn(r.t, r.f.repoDir, "diff-tree", "-r", "--name-only", "--no-commit-id", + hash)) { + obj, ok := strings.CutPrefix(path, echoTarget+"/default/configmaps/") + if !ok { + continue + } + obj = strings.TrimSuffix(obj, ".yaml") + files[obj] = r.valueAt(hash, obj) + } + if fmt.Sprint(files) != fmt.Sprint(want.files) { + r.failf("%s: commit %s by %s changes %v; the model's changes %v", what, hash[:8], author, files, want.files) + } + 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 + } +} + +// valueAt is the key of the document obj at commit hash, or "" when the commit holds none. +func (r *windowModelRun) valueAt(hash, obj string) string { + listed := gitIn(r.t, r.f.repoDir, "ls-tree", "--name-only", hash, echoFile(obj)) + if listed == "" { + return "" + } + document := gitIn(r.t, r.f.repoDir, "show", hash+":"+echoFile(obj)) + "\n" + for _, line := range strings.Split(document, "\n") { + if value, ok := strings.CutPrefix(line, " key: "); ok { + return value + } + } + r.failf("commit %s holds %s without a key: %s", hash[:8], obj, document) + return "" +} + +// 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()) + } + for _, obj := range modelObjects { + if got := r.valueAt("main", obj); got != r.m.remote[obj] { + r.failf("drained, Git holds %s=%q; the model %q", obj, got, r.m.remote[obj]) + } + } + 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) + } + } +} + +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) + }) + } +} From 445fb93bc2af22cc4a257650b4eb5b359e32d952 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Tue, 27 Oct 2026 22:41:15 +0000 Subject: [PATCH 4/8] test(git): a push hint leaves a save's window to the save Covers the window closeWindowForPushHint now skips per window, the one line Codecov reported uncovered on this PR. Co-Authored-By: Claude Opus 5.5 --- internal/git/applier_claim_test.go | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) 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) { From f7d822b63e32d25fd1d086945535e964f83db10c Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Tue, 27 Oct 2026 22:51:03 +0000 Subject: [PATCH 5/8] test(git): fire only the timers the worker armed, and check every path a commit changes Review of the reference model found two gaps. The driver derived its wakes from the window and request deadlines and ran closeOrArmWindow itself, so a worker that never armed its commit timer still passed. The loop now records the worker-clock time each of its commit and attach timers is armed for. The model checks both after every step, and the driver fires only an armed timer, through that timer's own wake, in either order when two are due at once. The clock tests deliver wakes the same way. The commit check ignored paths outside the ConfigMap folder and read only data.key. It now compares every changed path, allowing the bootstrap files only in the folder's first commit, and each document whole, parsed as YAML 1.2. The drained check compares the whole tree the same way. Co-Authored-By: Claude Opus 5.5 --- internal/git/branch_worker.go | 6 + internal/git/commit_request_attach_loop.go | 1 + internal/git/commit_window_clock_test.go | 20 +- internal/git/window_model_test.go | 250 +++++++++++++++------ 4 files changed, 204 insertions(+), 73 deletions(-) diff --git a/internal/git/branch_worker.go b/internal/git/branch_worker.go index eb946726..21f041e5 100644 --- a/internal/git/branch_worker.go +++ b/internal/git/branch_worker.go @@ -1032,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 @@ -1102,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. @@ -1489,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 diff --git a/internal/git/commit_request_attach_loop.go b/internal/git/commit_request_attach_loop.go index d2fda1ac..a57a53b9 100644 --- a/internal/git/commit_request_attach_loop.go +++ b/internal/git/commit_request_attach_loop.go @@ -536,6 +536,7 @@ func (l *branchWorkerEventLoop) rearmAttachTimer() { l.stopAttachTimer() return } + l.attachTimerAt = earliest delay := earliest.Sub(l.w.now()) if delay < 0 { delay = 0 diff --git a/internal/git/commit_window_clock_test.go b/internal/git/commit_window_clock_test.go index c917857f..5b365dc6 100644 --- a/internal/git/commit_window_clock_test.go +++ b/internal/git/commit_window_clock_test.go @@ -11,14 +11,22 @@ import ( ) // These tests pin that the commit windows' timers read the worker's clock: a test steps it past a -// deadline, delivers the wake the timer would have, and the window or the waiting request is due, -// with no sleep and no rewound deadline. +// 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 is the wake the commit and attach timers deliver when they fire, as run does. +// 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) { - loop.commitTimer = nil - loop.closeOrArmWindow() - loop.endWake(0) + 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) { diff --git a/internal/git/window_model_test.go b/internal/git/window_model_test.go index d5c0e227..01abdc6e 100644 --- a/internal/git/window_model_test.go +++ b/internal/git/window_model_test.go @@ -7,6 +7,7 @@ import ( "maps" "math/rand" "os" + "reflect" "regexp" "slices" "strconv" @@ -14,6 +15,8 @@ import ( "testing" "time" + "gopkg.in/yaml.v3" + configv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3" "github.com/ConfigButler/gitops-reverser/internal/manifestanalyzer" ) @@ -22,18 +25,20 @@ import ( // 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 -// 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. +// - 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, and the documents it changes. Per document, the writes that land do so in the order -// they were accepted, so an older write never lands after a newer one. +// 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 and nothing is charged. +// - 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 each deadline in turn and delivers the wake the timer would. +// 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. @@ -47,6 +52,9 @@ const ( 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 @@ -99,12 +107,14 @@ type modelPending struct { } // 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 - files map[string]string - seqs map[string]int + author string + message string + resync bool + bootstrap bool + files map[string]string + seqs map[string]int } type windowModel struct { @@ -120,6 +130,8 @@ type windowModel struct { history map[string][]string foreignUnseen, applyDue, hinted bool + // bootstrapped is whether Git holds modelBootstrap. + bootstrapped bool pending []modelPending commits []modelCommit @@ -372,6 +384,8 @@ func (m *windowModel) push() { } continue } + commit.bootstrap = !m.bootstrapped + m.bootstrapped = true m.commits = append(m.commits, commit) if p.save != nil { m.resolve(p.save, FinalizeCommitted) @@ -412,11 +426,6 @@ func (m *windowModel) endWake() bool { return true } -func (m *windowModel) timerWake() bool { - m.closeOrArm() - return m.endWake() -} - // 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() { @@ -485,6 +494,7 @@ func newWindowModelRun(t *testing.T, seed int64) *windowModelRun { 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 } @@ -538,50 +548,66 @@ func (r *windowModelRun) applied(modelSent bool) { } } -// timerWake is the wake the commit and attach timers deliver. -func (r *windowModelRun) timerWake() { +// 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.applied(r.m.timerWake()) + r.m.closeOrArm() + r.applied(r.m.endWake()) } -// nextDeadline is the earliest timer the worker has armed: a window's close, or a waiting save's -// attach deadline. -func (r *windowModelRun) nextDeadline() (time.Time, bool) { - var next time.Time - for _, window := range r.f.loop.commitWindows.all() { - if at, _ := window.closeAt(); next.IsZero() || at.Before(next) { - next = at - } - } - for _, pcr := range r.f.loop.pendingCRs { - if !pcr.attached && (next.IsZero() || pcr.attachDeadline.Before(next)) { - next = pcr.attachDeadline - } - } - return next, !next.IsZero() +// 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 { - next, ok := r.nextDeadline() - if !ok || next.After(target) { + at, commit, ok := r.armed() + if !ok || at.After(target) { r.clock.advance(target.Sub(r.clock.Now())) r.m.now = target return } - if next.After(r.clock.Now()) { - r.clock.advance(next.Sub(r.clock.Now())) + if at.After(r.clock.Now()) { + r.clock.advance(at.Sub(r.clock.Now())) } r.m.now = r.clock.Now() - r.timerWake() + if commit { + r.commitWake() + } else { + r.attachWake() + } r.check("timer") } - r.failf("a deadline stayed due after its wake") + r.failf("a timer stayed due after its wake") } func (r *windowModelRun) freshValue(obj string) string { @@ -746,6 +772,7 @@ var closeLine = regexp.MustCompile(`"msg"="Finalizing open commit window" "reaso func (r *windowModelRun) check(what string) { r.t.Helper() r.checkWindow(what) + r.checkTimers(what) r.checkSaves(what) r.checkCloses(what) r.checkCommits(what) @@ -767,6 +794,38 @@ func (r *windowModelRun) checkWindow(what string) { } } +// 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 @@ -871,18 +930,24 @@ func (r *windowModelRun) checkCommit(what, hash string, want modelCommit) { r.failf("%s: commit %s by %q %q; the model's by %q %q", what, hash[:8], author, message, want.author, want.message) } - files := map[string]string{} - for _, path := range splitLines(gitIn(r.t, r.f.repoDir, "diff-tree", "-r", "--name-only", "--no-commit-id", - hash)) { - obj, ok := strings.CutPrefix(path, echoTarget+"/default/configmaps/") - if !ok { - continue + 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) } - obj = strings.TrimSuffix(obj, ".yaml") - files[obj] = r.valueAt(hash, obj) } - if fmt.Sprint(files) != fmt.Sprint(want.files) { - r.failf("%s: commit %s by %s changes %v; the model's changes %v", what, hash[:8], author, files, want.files) + 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] { @@ -892,20 +957,39 @@ func (r *windowModelRun) checkCommit(what, hash string, want modelCommit) { } } -// valueAt is the key of the document obj at commit hash, or "" when the commit holds none. -func (r *windowModelRun) valueAt(hash, obj string) string { - listed := gitIn(r.t, r.f.repoDir, "ls-tree", "--name-only", hash, echoFile(obj)) - if listed == "" { - return "" +// 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) + } } - document := gitIn(r.t, r.f.repoDir, "show", hash+":"+echoFile(obj)) + "\n" - for _, line := range strings.Split(document, "\n") { - if value, ok := strings.CutPrefix(line, " key: "); ok { - return value + if value != "" { + want = map[string]any{ + "apiVersion": "v1", "kind": "ConfigMap", + "metadata": map[string]any{"name": obj, "namespace": "default"}, + "data": map[string]any{"key": value}, } } - r.failf("commit %s holds %s without a key: %s", hash[:8], obj, document) - return "" + 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. @@ -924,11 +1008,7 @@ func (r *windowModelRun) drained() { if held != 0 || r.f.loop.totalRetainedBytes() != 0 { r.failf("drained, the worker still charges %d (retained %d)", held, r.f.loop.totalRetainedBytes()) } - for _, obj := range modelObjects { - if got := r.valueAt("main", obj); got != r.m.remote[obj] { - r.failf("drained, Git holds %s=%q; the model %q", obj, got, r.m.remote[obj]) - } - } + 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) @@ -939,6 +1019,42 @@ func (r *windowModelRun) drained() { } } +// 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 From 520341793df9ecdb2052e8a35b8e6608c293aff1 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Wed, 28 Oct 2026 08:49:03 +0000 Subject: [PATCH 6/8] refactor(git): forget a stopped timer's time, and open a window on one clock stopCommitTimer and stopAttachTimer now clear commitTimerAt and attachTimerAt, so the recorded times hold only while their timer is armed, as their comments say. newOpenWindow no longer stamps openedAt from the wall clock: commitWindows.open sets it from the worker's clock, its only source. Co-Authored-By: Claude Opus 5.5 --- internal/git/branch_worker.go | 1 + internal/git/commit_request_attach_loop.go | 1 + internal/git/open_window.go | 7 +++---- 3 files changed, 5 insertions(+), 4 deletions(-) diff --git a/internal/git/branch_worker.go b/internal/git/branch_worker.go index 21f041e5..f52d0959 100644 --- a/internal/git/branch_worker.go +++ b/internal/git/branch_worker.go @@ -1779,6 +1779,7 @@ func (l *branchWorkerEventLoop) stopCommitTimer() { } } l.commitTimer = nil + l.commitTimerAt = time.Time{} } func (l *branchWorkerEventLoop) stopTimers() { diff --git a/internal/git/commit_request_attach_loop.go b/internal/git/commit_request_attach_loop.go index a57a53b9..2301e621 100644 --- a/internal/git/commit_request_attach_loop.go +++ b/internal/git/commit_request_attach_loop.go @@ -565,4 +565,5 @@ func (l *branchWorkerEventLoop) stopAttachTimer() { } } l.attachTimer = nil + l.attachTimerAt = time.Time{} } diff --git a/internal/git/open_window.go b/internal/git/open_window.go index fa8152fe..5e90fb4e 100644 --- a/internal/git/open_window.go +++ b/internal/git/open_window.go @@ -49,9 +49,9 @@ 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. @@ -96,7 +96,6 @@ func newOpenWindow(e Event, writer eventContentWriter) *openWindow { GitTargetNamespace: e.GitTargetNamespace, pathToEvent: make(map[string]Event), writer: writer, - openedAt: time.Now(), } } From 40fa6e960ed59da8b4963b5edcb8edccceed750b Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Wed, 28 Oct 2026 08:52:52 +0000 Subject: [PATCH 7/8] docs: name each worker decision as a specific fact, starting with the commit windows The event model now asks that a fact be named after the one decision it records, with the identities it is about and its reason, and that a change adding a decision name it in its design before anything persists facts. Concurrent author windows names the facts its stage 2 adds: WindowOpened, WriteJoinedWindow, FootprintReproved, WindowClosed and SingleWindowFallback. Co-Authored-By: Claude Opus 5.5 --- docs/design/branch-worker-event-model.md | 9 +++++++++ docs/design/concurrent-author-windows.md | 22 ++++++++++++++++++++++ 2 files changed, 31 insertions(+) diff --git a/docs/design/branch-worker-event-model.md b/docs/design/branch-worker-event-model.md index b4974e5c..d2e18ef8 100644 --- a/docs/design/branch-worker-event-model.md +++ b/docs/design/branch-worker-event-model.md @@ -324,6 +324,15 @@ 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. +**Name each fact after the one decision it records**, in the product's words: `SaveAttached`, +`WindowClosed` with its reason, not a generic `StateChanged` or `WindowUpdated` that leaves the +reader to compare two states to learn what happened. A fact carries the identities it is about +(request UID, window ID) and the reason for the decision. A change that adds a decision to the +worker names that decision as a fact in its design, even while nothing persists facts yet, so the +vocabulary is settled before the journal is built. +[Concurrent author windows](concurrent-author-windows.md#the-window-decisions-are-named-facts) is +the first design to do so. + ```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..985a6c26 100644 --- a/docs/design/concurrent-author-windows.md +++ b/docs/design/concurrent-author-windows.md @@ -220,6 +220,28 @@ 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 are named facts + +The [event model](branch-worker-event-model.md#commands-facts-and-effects) wants every decision of +the worker to become a fact named after that decision, so that a later journal can replay it +without reading the clock or the tree again. This change adds decisions, so it names them now. +Nothing persists them yet: the names are used in log lines, close reasons, metrics and the +reference model. + +| Fact | Carries | 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. It is unique for the life of one worker; making it +survive a restart belongs to the +[HA and durable delivery plan](../future/ha-gittarget-distribution-plan.md), with the rest of the +journal. Deadlines are not facts of this design: they follow from the opening time, the latest write +and the timers, and the event model's deadline protocol covers them. + ## 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 From 946e39b7a70e4201d86100a6da578df04872499c Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Wed, 28 Oct 2026 09:03:28 +0000 Subject: [PATCH 8/8] docs: keep decision names for diagnostics, and refuse an over-broad write before its commit The event model asked every new decision to be named as a fact before anything persists facts, which settles a journal vocabulary before the recovery contract chooses between durable current state and an event history. It now asks only for specific names and reasons, used in logs, close reasons and tests, and leaves that choice open. Concurrent author windows checked a write against its footprint at materialization but committed it anyway, which cannot stop an older value from landing over a newer one. The check now runs before the commit, over every path the commit changes (writeBatch.touched covers one event only), and refuses the write: worktree undone, save failed, one window from then on, and a resync of the GitTarget. The windows' decisions are names for diagnostics, independent of persistence. Co-Authored-By: Claude Opus 5.5 --- docs/design/branch-worker-event-model.md | 15 ++++----- docs/design/concurrent-author-windows.md | 42 +++++++++++++----------- 2 files changed, 29 insertions(+), 28 deletions(-) diff --git a/docs/design/branch-worker-event-model.md b/docs/design/branch-worker-event-model.md index d2e18ef8..d44f4087 100644 --- a/docs/design/branch-worker-event-model.md +++ b/docs/design/branch-worker-event-model.md @@ -324,14 +324,13 @@ 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. -**Name each fact after the one decision it records**, in the product's words: `SaveAttached`, -`WindowClosed` with its reason, not a generic `StateChanged` or `WindowUpdated` that leaves the -reader to compare two states to learn what happened. A fact carries the identities it is about -(request UID, window ID) and the reason for the decision. A change that adds a decision to the -worker names that decision as a fact in its design, even while nothing persists facts yet, so the -vocabulary is settled before the journal is built. -[Concurrent author windows](concurrent-author-windows.md#the-window-decisions-are-named-facts) is -the first design to do so. +**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 diff --git a/docs/design/concurrent-author-windows.md b/docs/design/concurrent-author-windows.md index 985a6c26..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,15 +224,15 @@ 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 are named facts +### The window decisions have specific names -The [event model](branch-worker-event-model.md#commands-facts-and-effects) wants every decision of -the worker to become a fact named after that decision, so that a later journal can replay it -without reading the clock or the tree again. This change adds decisions, so it names them now. -Nothing persists them yet: the names are used in log lines, close reasons, metrics and the -reference model. +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. -| Fact | Carries | When | +| 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 | @@ -236,11 +240,8 @@ reference model. | `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. It is unique for the life of one worker; making it -survive a restart belongs to the -[HA and durable delivery plan](../future/ha-gittarget-distribution-plan.md), with the rest of the -journal. Deadlines are not facts of this design: they follow from the opening time, the latest write -and the timers, and the event model's deadline protocol covers them. +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 @@ -322,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. @@ -336,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.