From 1e3675ee7bf8f26b32bb36ba2ed0eb67f1a22fbb Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Mon, 9 Nov 2026 18:19:06 +0000 Subject: [PATCH 1/5] refactor(git): run the push cooldown on the worker clock maybeSchedulePush and pushPending read BranchWorker.now(), and the loop records pushTimerAt as it does commitTimerAt, so a test that steps the clock delivers the push timer's wake. injectOutage switches push, fetch and sync faults on and off per step. holdPushes starts the cooldown an hour ahead on the worker clock, so a test that steps the clock still holds. Co-Authored-By: Claude Opus 5.5 --- internal/git/branch_worker.go | 13 +++++--- internal/git/commit_request_attach_test.go | 7 +++-- internal/git/parent_recovery.go | 6 ++-- internal/git/write_gate_test.go | 36 ++++++++++++++++++++++ 4 files changed, 52 insertions(+), 10 deletions(-) diff --git a/internal/git/branch_worker.go b/internal/git/branch_worker.go index 55dbeae2..eb264e11 100644 --- a/internal/git/branch_worker.go +++ b/internal/git/branch_worker.go @@ -1093,8 +1093,11 @@ type branchWorkerEventLoop struct { // 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 + // pushTimer fires the next push when the success cooldown ends, and pushTimerAt is the + // worker-clock time it is armed for, while it is armed. Like commitTimerAt, tests that step the + // clock deliver its wake at this time. + pushTimer *time.Timer + pushTimerAt time.Time // retry is the one schedule of the next attempt at work a failed attempt left pending. See // retry.go. retry retrySchedule @@ -1778,13 +1781,14 @@ func (l *branchWorkerEventLoop) maybeSchedulePush() { l.pushPending() return } - elapsed := time.Since(l.lastPushAt) + elapsed := l.w.now().Sub(l.lastPushAt) if elapsed >= PushCooldown { l.pushPending() return } if l.pushTimer == nil { l.pushTimer = time.NewTimer(PushCooldown - elapsed) + l.pushTimerAt = l.lastPushAt.Add(PushCooldown) } } @@ -1850,7 +1854,7 @@ func (l *branchWorkerEventLoop) pushPending() { l.pendingWrites = nil l.pendingWritesBytes = 0 - l.lastPushAt = time.Now() + l.lastPushAt = l.w.now() l.stopPushTimer() l.endOutage() l.settleReplayRefusals(refused) @@ -1902,6 +1906,7 @@ func (l *branchWorkerEventLoop) resolvePushedCommitRequests() { } func (l *branchWorkerEventLoop) stopPushTimer() { + l.pushTimerAt = time.Time{} if l.pushTimer == nil { return } diff --git a/internal/git/commit_request_attach_test.go b/internal/git/commit_request_attach_test.go index ef29f4ad..14d8a038 100644 --- a/internal/git/commit_request_attach_test.go +++ b/internal/git/commit_request_attach_test.go @@ -69,10 +69,11 @@ func forceDue(loop *branchWorkerEventLoop) { } } -// holdPushes starts the push cooldown now and keeps every push behind it, a save's too, so the local -// commits stay inspectable until the test pushes. +// holdPushes keeps every push behind the cooldown, a save's too, so the local commits stay +// inspectable until the test pushes. The cooldown runs on the worker's clock, so it starts an hour +// ahead: a test that steps the clock still holds. func holdPushes(loop *branchWorkerEventLoop) { - loop.lastPushAt = time.Now() + loop.lastPushAt = loop.w.now().Add(time.Hour) loop.cooldownHoldsSaves = true } diff --git a/internal/git/parent_recovery.go b/internal/git/parent_recovery.go index a0ca5008..f175f53f 100644 --- a/internal/git/parent_recovery.go +++ b/internal/git/parent_recovery.go @@ -52,9 +52,9 @@ type parentRecovery struct { gen uint64 } -// 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. +// now is the loop's clock, injectable so the retry schedule, the push cooldown and the commit +// windows' timers can be tested without sleeping. The timers themselves are real: a test that moves +// the clock past a deadline delivers the wake the timer would have, as run does. func (w *BranchWorker) now() time.Time { if w.clock != nil { return w.clock() diff --git a/internal/git/write_gate_test.go b/internal/git/write_gate_test.go index 23c58525..3306def9 100644 --- a/internal/git/write_gate_test.go +++ b/internal/git/write_gate_test.go @@ -108,6 +108,42 @@ func failPushes(t *testing.T, worker *BranchWorker) func() { return restore } +// remoteOutage is a Git host a test takes down and brings back, step by step: while it is down, +// every push, fetch and sync fails on a transport error, so the worker learns nothing of the +// remote, as in an outage. Up, each goes through to the real server. +type remoteOutage struct{ down bool } + +// injectOutage installs an outage, up, until the test ends. +func injectOutage(t *testing.T) *remoteOutage { + t.Helper() + o := &remoteOutage{} + unreachable := errors.New("dial tcp: connection refused") + push, fetch, sync := pushAtomicFn, fetchRemoteBranchHashFn, syncToRemoteFn + pushAtomicFn = func(ctx context.Context, repo *gogit.Repository, root plumbing.Hash, + branch plumbing.ReferenceName, auth []gitclient.Option) (PushOutcome, error) { + if o.down { + return PushOutcome{}, unreachable + } + return push(ctx, repo, root, branch, auth) + } + fetchRemoteBranchHashFn = func(ctx context.Context, repo *gogit.Repository, branch plumbing.ReferenceName, + auth []gitclient.Option) (plumbing.Hash, error) { + if o.down { + return plumbing.ZeroHash, unreachable + } + return fetch(ctx, repo, branch, auth) + } + syncToRemoteFn = func(ctx context.Context, repo *gogit.Repository, branch plumbing.ReferenceName, + parent string, auth []gitclient.Option) (*PullReport, error) { + if o.down { + return nil, unreachable + } + return sync(ctx, repo, branch, parent, auth) + } + t.Cleanup(func() { pushAtomicFn, fetchRemoteBranchHashFn, syncToRemoteFn = push, fetch, sync }) + return o +} + // TestWriteGate_ARequestOnARefusingTargetFails is the whole matrix of a request that ends on a target // that refuses writes: each gate, each way a request can end, each whenNothingToCommit. Every cell // fails with the gate's error and leaves the remote and the retained writes untouched. Before From dcfcbd255f061516917304e52e8b67d8ab49ec11 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Mon, 9 Nov 2026 18:19:07 +0000 Subject: [PATCH 2/5] test(git): keep work pending across wakes in the window reference model The model no longer pushes at once after every wake. It keeps its own pending writes and follows the worker's publication rules (cooldown, failed push and backoff, deferred commits, the apply flush and the resync barrier), checks the pending writes, their charge and all four timers after every step, adds outage, restart, overdue-timer, stale-applier and restructure steps, three profiles and sixteen reach floors, and runs 20 seeds of 60 steps in CI. Co-Authored-By: Claude Opus 5.5 --- internal/git/window_model_test.go | 1030 +++++++++++++++++++++++++---- 1 file changed, 888 insertions(+), 142 deletions(-) diff --git a/internal/git/window_model_test.go b/internal/git/window_model_test.go index 01af0d4b..802ab109 100644 --- a/internal/git/window_model_test.go +++ b/internal/git/window_model_test.go @@ -20,6 +20,7 @@ import ( "time" "github.com/go-git/go-git/v6/plumbing" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "gopkg.in/yaml.v3" @@ -32,9 +33,10 @@ import ( // real Git server, with the worker checked against it. // // - After every step: the open windows, oldest first (each with its number, author, objects, -// attached save, deadline, charge and footprint), the timers armed for them 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. +// attached save, deadline, charge and footprint), the pending writes in order (resync or not, +// the save each carries, its charge, committed or not), the timers (commit, attach, push cooldown +// and retry), the registered saves (attached or waiting, and until when), the retained-byte +// charge, the close reasons so far, and every save's outcome are the model's. // - Every commit the worker publishes is the one the model's closing window makes: its author, its // message, every path it changes, and each file whole, its documents parsed. Per document, the // writes that land do so in the order they were accepted, so an older write never lands after a @@ -51,10 +53,16 @@ import ( // reads a bare y as a boolean, so the render of the root refuses a ConfigMap named y once it lands // there. // +// Work is kept across wakes, as the worker keeps it: a decided write is committed at the end of its +// wake unless it needs a connection while a retry waits, and published by a push the cooldown spaces, +// or at once for a save, a flush and the first push. The Git server can be taken down and brought +// back (injectOutage): a failed push keeps every write for the retry, which backs off from 10s to 5m. +// // The model shares no window code with the worker: only the event fixtures, the per-event charge -// (eventRetainedCharge, attachCharge) and the close-reason names. The worker's timers are real; the -// driver steps the worker's clock to the time each armed timer is due and delivers that timer's wake, -// so a deadline the worker armed no timer for is never delivered. +// (eventRetainedCharge, attachCharge, a resync's charge, the pending-write overhead) and the +// close-reason names. The worker's timers are real; the driver steps the worker's clock to the time +// each armed timer is due and delivers that timer's wake, so a deadline the worker armed no timer for +// is never delivered. // // WINDOW_MODEL_SEEDS and WINDOW_MODEL_STEPS raise the search; a failure names its seed, and // WINDOW_MODEL_SEED replays one. WINDOW_MODEL_LOGS prints the worker's log with a failure. @@ -141,17 +149,26 @@ func (t *modelTree) set(obj, value string) []string { } // setOutside is a write from another clone: it edits a document where it lives, or adds the object at -// its built-in path, unregistered. +// its built-in path, unregistered, after any document a restructuring left in that file. func (t *modelTree) setOutside(obj, value string) string { at := t.loc(obj) if at == "" { at = echoFile(obj) - t.files[at] = []string{obj} + t.files[at] = append(t.files[at], obj) } t.values[obj] = value return at } +// merge is a restructuring from another clone: obj's document moves into the file into's document is +// alone in, after it. It returns the paths it changes. +func (t *modelTree) merge(into, obj string) []string { + from, to := t.loc(obj), t.loc(into) + delete(t.files, from) + t.files[to] = append(t.files[to], obj) + return []string{to, from} +} + // render is the bytes of the file at path. func (t *modelTree) render(path string) string { if path == modelKustFile { @@ -240,6 +257,12 @@ type modelSave struct { sawForeign bool } +// modelOutcome is how a save settled: with an outcome, or failed. +type modelOutcome struct { + outcome FinalizeOutcome + failed bool +} + type modelWindow struct { seq int author string @@ -261,8 +284,10 @@ func (w *modelWindow) closeAt() (time.Time, windowFinalizeReason) { return w.maxAt, windowFinalizeReasonMaxDuration } -// modelPending is a decided write the next push publishes. footprint is the proven footprint of the -// window it closed, when that window had company. +// modelPending is a decided write a later push publishes. footprint is the proven footprint of the +// window it closed, when that window had company. decision numbers it in the order it was decided, +// charge is what keeping it costs, and committed says the checkout holds it. A write that changes +// nothing and carries no save stays pending until it is committed, and then leaves no trace. type modelPending struct { save *modelSave author string @@ -270,6 +295,13 @@ type modelPending struct { resync bool changes []modelEdit footprint *modelFootprint + decision int + charge int64 + committed bool + // creates is a resync that adds an object to Git: committed, its applier is due. changesTree is a + // write that changes the tree it commits on, so its commit adds the folder's bootstrap files: a save + // that changes nothing commits nothing. + creates, changesTree bool } // modelFile is what a commit leaves at one path: the kustomization's entries, the documents, or @@ -309,24 +341,67 @@ type windowModel struct { history map[string][]string foreignUnseen, applyDue, hinted bool - // bootstrapped is whether Git holds modelBootstrap. - bootstrapped bool + // bootstrapped is whether Git holds modelBootstrap, and bootstrapLocal whether the checkout does. + bootstrapped, bootstrapLocal bool + // surprise names a state the model does not cover, which the run reports. + surprise string + // reach counts the situations the search must reach (reachFloors). + reach map[string]int + // applied is what the last apply put on the cluster, and withheld the values of applier writes kept + // out of Git, by object. + applied, withheld map[string]string + + // The publication. lastPushAt is the last push that landed, pushAt the cooldown's timer while it + // is armed, and retryAt and backoff the retry schedule while a failed attempt waits for its next. + lastPushAt, pushAt, retryAt time.Time + backoff time.Duration + // down is an outage: every push, fetch and sync fails. untrusted is a base the worker cannot vouch + // for since a push failed: a commit with nothing committed before it then needs a fetch. + down, untrusted bool + // flushing is an apply that waits for the writes decided up to flushMark to be published. + flushing bool + flushMark, decisions int pending []modelPending commits []modelCommit closes []string - outcomes map[string]FinalizeOutcome + outcomes map[string]modelOutcome given map[string]bool // landed is, per document, the acceptance seq of the last write a commit landed. landed map[string]int } +// The situations the search must reach, across its seeds, for a pass to say anything about them. +const ( + reachShutdown = "a shutdown with several windows open" + reachFlush = "an apply's flush with several windows open" + reachHint = "a push hint with several windows open" + reachAttachNext = "attach: Next beside another author's window" + reachReproof = "a reproof that finds open windows overlap" + reachPathConflict = "a path_conflict close" + reachFailedPush = "a failed push with several windows open" + reachCooldown = "the cooldown holding writes with several windows open" + reachRetryWait = "a retry waiting with several windows open" + reachDeferred = "a decided write waiting for its retry to be committed" + reachReplay = "a replay onto a moved remote with a window open" + reachPressure = "buffer pressure with several windows open" + reachApplyWaits = "an apply's flush waiting for the retry" + reachRestart = "a restart mid-run" + reachOverdue = "an input handled while a timer is overdue" + reachStaleApply = "an applier write over a newer edit, withheld" +) + +func (m *windowModel) reached(what string) { m.reach[what]++ } + +// several reports whether more than one window is open. +func (m *windowModel) several() bool { return len(m.windows) > 1 } + func newWindowModel(now time.Time, tree modelTree) *windowModel { m := &windowModel{ now: now, saves: map[string]*modelSave{}, remote: tree.clone(), known: tree.clone(), cluster: maps.Clone(tree.values), - history: map[string][]string{}, outcomes: map[string]FinalizeOutcome{}, given: map[string]bool{}, - landed: map[string]int{}, + history: map[string][]string{}, outcomes: map[string]modelOutcome{}, given: map[string]bool{}, + landed: map[string]int{}, reach: map[string]int{}, withheld: map[string]string{}, } for obj, value := range tree.values { m.history[obj] = []string{value} @@ -384,12 +459,31 @@ func (m *windowModel) others(own *modelWindow) []*modelWindow { func (m *windowModel) waiting(s *modelSave) bool { return !s.attached && s.deadline.After(m.now) } -// prove is the footprint of a write of obj on the tree the worker's checkout holds. Before the folder -// is bootstrapped every write adds its bootstrap files. +// prove is the footprint of a write of obj admitted now: proven on the tree the checkout holds, once +// it holds every decided write, and unknown while a decided write waits for a connection to commit. func (m *windowModel) prove(obj string) modelFootprint { + if !m.holdsDecided() { + return modelFootprint{unknown: true} + } + return m.proveOn(obj) +} + +// holdsDecided reports whether the checkout holds every decided write. +func (m *windowModel) holdsDecided() bool { + for _, p := range m.pending { + if !p.committed { + return false + } + } + return true +} + +// proveOn is the footprint of a write of obj on the tree that holds every decided write. Before the +// folder is bootstrapped every write adds its bootstrap files. +func (m *windowModel) proveOn(obj string) modelFootprint { at := m.known.loc(obj) switch { - case !m.bootstrapped, at == "", strings.HasPrefix(at, modelKustDir+"/"), len(m.known.files[at]) > 1: + case !m.bootstrapLocal, at == "", strings.HasPrefix(at, modelKustDir+"/"), len(m.known.files[at]) > 1: return modelFootprint{unknown: true} default: return modelFootprint{paths: []string{at}} @@ -404,6 +498,16 @@ func (m *windowModel) proveWindow(w *modelWindow) modelFootprint { return f } +// proveCommit is the footprint a closing window's commit is checked against: proven again on the tree +// the commit is planned on, which holds every write decided before it. +func (m *windowModel) proveCommit(w *modelWindow) modelFootprint { + f := modelFootprint{} + for _, obj := range w.objs { + f = f.union(m.proveOn(obj)) + } + return f +} + // admit closes the windows a write of obj by author cannot sit beside, and returns the footprint its // window takes. func (m *windowModel) admit(author, obj string) *modelFootprint { @@ -451,6 +555,7 @@ func (m *windowModel) reprove() { } for _, w := range slices.Clone(m.windows) { if involved[w] { + m.reached(reachReproof) m.close(w, windowFinalizeReasonPathConflict) } } @@ -492,6 +597,9 @@ func (m *windowModel) write(author string, edit modelEdit, pressure bool) { } } if pressure { + if m.several() { + m.reached(reachPressure) + } m.closeAll(windowFinalizeReasonBufferLimit) return } @@ -509,7 +617,7 @@ func (m *windowModel) register(s *modelSave, attachTimeout time.Duration) { collecting = collecting || (earlier.attached && w.save == earlier) } if m.waiting(earlier) || (collecting && !s.next) { - m.outcomes[s.name] = FinalizeSaveInProgress + m.outcomes[s.name] = modelOutcome{outcome: FinalizeSaveInProgress} return } } @@ -600,7 +708,7 @@ func (m *windowModel) expire() { func (m *windowModel) resolve(s *modelSave, outcome FinalizeOutcome) { delete(m.saves, s.name) - m.outcomes[s.name] = outcome + m.outcomes[s.name] = modelOutcome{outcome: outcome} } // close decides a window: the documents whose latest write differs from the checkout, in the order the @@ -609,33 +717,56 @@ func (m *windowModel) resolve(s *modelSave, outcome FinalizeOutcome) { // seen re-asserts them (API wins for a save). A window once proven carries its footprint, proven again // on the checkout it commits on. func (m *windowModel) close(w *modelWindow, reason windowFinalizeReason) { + switch { + case reason == windowFinalizeReasonPathConflict: + m.reached(reachPathConflict) + case reason == windowFinalizeReasonAttachNext && m.several(): + m.reached(reachAttachNext) + } m.windows = slices.DeleteFunc(m.windows, func(o *modelWindow) bool { return o == w }) m.closes = append(m.closes, fmt.Sprintf("%s %s #%d", reason, w.author, w.seq)) var footprint *modelFootprint if w.footprint != nil { - if f := m.proveWindow(w); !f.unknown { + if f := m.proveCommit(w); !f.unknown { footprint = &f } } var changes []modelEdit + changesTree := false for _, obj := range w.objs { if edit := w.latest[obj]; edit.value != m.known.values[obj] || w.save != nil { changes = append(changes, edit) + changesTree = changesTree || edit.value != m.known.values[obj] m.known.set(obj, edit.value) } } - if len(changes) == 0 && w.save == nil { - return - } - p := modelPending{save: w.save, author: w.author, changes: changes, footprint: footprint} + // The write keeps the window's charge, less the save's own item: its message is the write's. + p := modelPending{save: w.save, author: w.author, changes: changes, footprint: footprint, charge: w.charge, + changesTree: changesTree} if w.save != nil { p.message = w.save.message + p.charge -= pendingWriteOverheadBytes } else { p.message = generatedMessage(changes) } + m.decide(p) +} + +// decide appends a decided write; advance commits it at the end of the wake. +func (m *windowModel) decide(p modelPending) { + m.decisions++ + p.decision = m.decisions m.pending = append(m.pending, p) } +func (m *windowModel) pendingCharge() int64 { + var charge int64 + for _, p := range m.pending { + charge += p.charge + } + return charge +} + func generatedMessage(changes []modelEdit) string { var b strings.Builder plural := "s" @@ -649,13 +780,170 @@ func generatedMessage(changes []modelEdit) string { 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. -// The footprint of a replayed write was proven on another tree, so it holds nothing about the commit. +// The model's publication rules: a decided write is committed at the end of the wake that decided it, +// unless it needs a connection while a failed attempt waits for its retry, and published by a push +// the cooldown spaces after the last one that landed. A push that fails keeps every write for a retry, +// one backoff later, 10s doubling to 5m, whatever arrives meanwhile. +const ( + modelCooldown = 5 * time.Second + modelRetryFirst = 10 * time.Second + modelRetryMax = 5 * time.Minute +) + +func (m *windowModel) awaitingRetry() bool { return !m.retryAt.IsZero() && m.now.Before(m.retryAt) } + +func (m *windowModel) scheduleRetry() { + if m.backoff == 0 { + m.backoff = modelRetryFirst + } else { + m.backoff = min(2*m.backoff, modelRetryMax) + } + m.retryAt = m.now.Add(m.backoff) +} + +func (m *windowModel) endOutage() { m.retryAt, m.backoff = time.Time{}, 0 } + +// pendingUpTo reports whether a write decided up to mark is still pending. +func (m *windowModel) pendingUpTo(mark int) bool { + return slices.ContainsFunc(m.pending, func(p modelPending) bool { return p.decision <= mark }) +} + +func (m *windowModel) carriesSave() bool { + return slices.ContainsFunc(m.pending, func(p modelPending) bool { return p.save != nil }) +} + +// commitIsLocal reports whether committing the decided writes the checkout does not hold needs no +// connection: none is a resync, and the checkout holds a write before them or sits on a base the +// worker can vouch for. +func (m *windowModel) commitIsLocal() bool { + committed := false + for _, p := range m.pending { + if p.committed { + committed = true + } else if p.resync { + return false + } + } + return committed || !m.untrusted +} + +// materialize commits, in order, every decided write the checkout does not hold, and reports whether +// it could: a resync fetches first, and so does a write with nothing committed before it on a base +// the worker cannot vouch for. A write that commits nothing and carries no save leaves the pending +// writes once it is committed; a resync that adds an object makes its applier due. +func (m *windowModel) materialize() bool { + for i := 0; i < len(m.pending); { + p := &m.pending[i] + if p.committed { + i++ + continue + } + if (p.resync || (i == 0 && m.untrusted)) && !m.fetch() { + return false + } + p.committed = true + m.bootstrapLocal = m.bootstrapLocal || p.changesTree + m.applyDue = m.applyDue || (p.resync && p.creates) + if len(p.changes) == 0 && p.save == nil { + m.pending = slices.Delete(m.pending, i, i+1) + continue + } + i++ + } + return true +} + +// fetch is the fetch a commit needs first, and reports whether it could be made. +func (m *windowModel) fetch() bool { + if m.down { + return false + } + if m.foreignUnseen { + m.surprise = "a fetch before a commit would find the push from outside" + } + m.untrusted = false + return true +} + +// commitDecided commits what the wake decided. While a failed attempt waits for its retry, a commit +// that needs a connection waits for that retry too. +func (m *windowModel) commitDecided() { + if m.holdsDecided() || (m.awaitingRetry() && !m.commitIsLocal()) { + return + } + if !m.materialize() && m.retryAt.IsZero() { + m.scheduleRetry() + } +} + +// advance is the end of every wake: commit what was decided, and push it, at once or when the +// cooldown ends; with nothing pending, the outage is over. +func (m *windowModel) advance() { + for range 100 { + m.commitDecided() + if len(m.pending) == 0 { + m.pushAt = time.Time{} + m.endOutage() + return + } + m.maybePush() + if !m.retryAt.IsZero() || m.holdsDecided() { + return + } + } + m.surprise = "advance made no progress" +} + +// maybePush pushes at once the first time, for an apply waiting on its flush, and for a save; any +// other push waits for the cooldown since the last one that landed. While a retry waits, nothing +// pushes before it. +func (m *windowModel) maybePush() { + if len(m.pending) == 0 || m.awaitingRetry() { + return + } + if m.lastPushAt.IsZero() || (m.flushing && m.pendingUpTo(m.flushMark)) || m.carriesSave() || + !m.now.Before(m.lastPushAt.Add(modelCooldown)) { + m.push() + return + } + if m.pushAt.IsZero() { + m.pushAt = m.lastPushAt.Add(modelCooldown) + } +} + +// push commits what is decided and publishes all of it, or keeps it all for the retry. func (m *windowModel) push() { + m.pushAt = time.Time{} + if len(m.pending) == 0 { + return + } + if !m.materialize() { + m.scheduleRetry() + return + } if len(m.pending) == 0 { return } + if m.down { + if m.several() { + m.reached(reachFailedPush) + } + m.untrusted = true + m.scheduleRetry() + return + } + if m.foreignUnseen && len(m.windows) > 0 { + m.reached(reachReplay) + } + m.publish() + m.lastPushAt, m.untrusted = m.now, false + m.endOutage() +} + +// publish 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. +// The footprint of a replayed write was proven on another tree, so it holds nothing about the commit. +func (m *windowModel) publish() { for _, p := range m.pending { commit := modelCommit{author: p.author, message: p.message, resync: p.resync, files: map[string]modelFile{}, seqs: map[string]int{}} @@ -694,6 +982,7 @@ func (m *windowModel) push() { m.foreignUnseen, m.applyDue = false, true } m.known = m.remote.clone() + m.bootstrapLocal = m.bootstrapped } // endWake is the loop's end of every wake. It reports whether an apply was sent: the driver's applier @@ -706,49 +995,100 @@ func (m *windowModel) endWake() bool { } if m.hinted { m.hinted = false + if m.several() { + m.reached(reachHint) + } for _, w := range slices.Clone(m.windows) { if w.save == nil { m.close(w, windowFinalizeReasonForeignCommit) } } } - m.push() - if !m.applyDue || len(m.saves) > 0 { + m.advance() + return m.serviceAppliers() +} + +// serviceAppliers sends a due apply once no save is registered: the open windows close and the writes +// decided so far are published first, at once, cooldown or not. While a retry waits, the apply waits +// for it; windows opened meanwhile stay open. A resync not published yet holds the applier back +// (the barrier), and ends the flush: the next one closes the windows again. +func (m *windowModel) serviceAppliers() bool { + if !m.applyDue || slices.ContainsFunc(m.pending, func(p modelPending) bool { return p.resync }) { + m.flushing = false return false } - m.closeAll(windowFinalizeReasonApplyDue) - m.push() - m.applyDue = false + if len(m.saves) > 0 { + return false + } + if !m.flushing { + if m.several() { + m.reached(reachFlush) + } + m.closeAll(windowFinalizeReasonApplyDue) + m.flushing, m.flushMark = true, m.decisions + if m.pendingUpTo(m.flushMark) { + m.advance() + } + } + if m.pendingUpTo(m.flushMark) { + if !m.retryAt.IsZero() { + m.reached(reachApplyWaits) + } + return false + } + m.flushing, m.applyDue = false, 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.values) + m.applied = maps.Clone(m.known.values) return true } -// resync closes the windows, 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() { +// resync closes the windows, then decides, as the operator, every live object Git does not hold as it +// is, charged what its snapshot is. prune.mode is onEvent, so nothing is swept. An object it adds is +// adopted once it is committed: an apply is due. +func (m *windowModel) resync(charge int64) { m.closeAll(windowFinalizeReasonResyncBeforeApply) - var changes []modelEdit + p := modelPending{author: DefaultCommitterName, resync: true, charge: charge} for _, obj := range modelObjects { value := m.cluster[obj] if value == "" || value == m.known.values[obj] { continue } - if m.known.values[obj] == "" { - m.applyDue = true - } + p.creates = p.creates || m.known.values[obj] == "" m.known.set(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}) + // What the snapshot found on the cluster is the newest write of the object. + m.editSeq++ + p.changes = append(p.changes, modelEdit{seq: m.editSeq, obj: obj, value: value}) + p.changesTree = true } + m.decide(p) +} + +// restart is a new worker after a shutdown: a checkout cloned from Git, so it knows what Git holds, and +// no window, save, pending write or schedule. Window numbers start again. +func (m *windowModel) restart() { + m.pending, m.saves, m.given = nil, map[string]*modelSave{}, map[string]bool{} + m.known, m.bootstrapLocal = m.remote.clone(), m.bootstrapped + m.foreignUnseen, m.applyDue, m.hinted, m.flushing, m.untrusted = false, false, false, false, false + m.lastPushAt, m.pushAt = time.Time{}, time.Time{} + m.endOutage() + m.seq = 0 } -// shutdown closes the windows and publishes; a save still waiting for a window is given back. +// shutdown closes the windows and pushes, cooldown and retry or not. A save riding a write the push +// could not publish fails; a save still waiting for a window is given back. func (m *windowModel) shutdown() { + if m.several() { + m.reached(reachShutdown) + } m.closeAll(windowFinalizeReasonShutdown) m.push() + for _, p := range m.pending { + if p.save != nil { + delete(m.saves, p.save.name) + m.outcomes[p.save.name] = modelOutcome{failed: true} + } + } for name := range m.saves { m.given[name] = true delete(m.saves, name) @@ -773,8 +1113,16 @@ type windowModelRun struct { exited bool values int saves int + // outage takes the Git server down and brings it back. + outage *remoteOutage + profile windowModelProfile // concurrent is how many steps ended with more than one window open. concurrent int + // firstSave is the first save the current worker was sent: a restart leaves the earlier ones with + // the worker before it. + firstSave int + // sent records that an apply was sent since it was last cleared. + sent bool } func newWindowModelRun(t *testing.T, seed int64) *windowModelRun { @@ -794,7 +1142,9 @@ func newWindowModelRun(t *testing.T, seed int64) *windowModelRun { f := newEchoFixtureWith(t, "window-model-"+strconv.FormatInt(seed, 10), configv1alpha3.PruneOnEvent, map[string]string{"x": "x0", "w": "w0", "z": "z0"}, layout) requester := f.withClaim() - r := &windowModelRun{t: t, seed: seed, f: f, requester: requester, rng: rand.New(rand.NewSource(seed))} + r := &windowModelRun{t: t, seed: seed, f: f, requester: requester, rng: rand.New(rand.NewSource(seed)), + outage: injectOutage(t), firstSave: 1} + r.profile = []windowModelProfile{outageProfile, wideProfile, concurrentProfile}[seed%3] r.clock = useTestClock(f.worker) r.logs = captureInfo(f.worker) f.loop.defaultWindow = commitWindowDefaults{idleTimeout: modelIdle, maxDuration: modelMax} @@ -803,15 +1153,21 @@ func newWindowModelRun(t *testing.T, seed int64) *windowModelRun { r.start = r.clock.Now() r.m = newWindowModel(r.start, tree) r.m.bootstrapped = gitIn(t, f.repoDir, "ls-tree", "--name-only", r.base, echoTarget+"/README.md") != "" + r.m.bootstrapLocal = r.m.bootstrapped return r } -// pushFromOutside commits the files at paths, as tree holds them, from the outside clone. +// pushFromOutside commits the files at paths, as tree holds them, from the outside clone: a path the +// tree holds no file at is removed. func (r *windowModelRun) pushFromOutside(tree modelTree, paths []string, message string) { r.t.Helper() clone := r.f.clone gitIn(r.t, clone, "pull", "--quiet", "--ff-only", "origin", "main") for _, path := range paths { + if _, held := tree.files[path]; !held && path != modelKustFile { + gitIn(r.t, clone, "rm", "--quiet", path) + continue + } full := filepath.Join(clone, path) require.NoError(r.t, os.MkdirAll(filepath.Dir(full), 0o750)) require.NoError(r.t, os.WriteFile(full, []byte(tree.render(path)), 0o600)) @@ -840,10 +1196,9 @@ 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. +// wake runs one wake of the worker's loop on item, 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) @@ -861,6 +1216,7 @@ func (r *windowModelRun) applied(modelSent bool) { r.knownTip = gitIn(r.t, r.f.repoDir, "rev-parse", "main") } sent := r.requester.taken() + r.sent = r.sent || modelSent if modelSent != (len(sent) > 0) { r.failf("the model sent an apply: %v; the worker sent %d", modelSent, len(sent)) } @@ -870,46 +1226,87 @@ func (r *windowModelRun) applied(modelSent bool) { } req.Report(ApplyResult{ID: req.ID, Outcome: ApplySucceeded, AppliedRevision: req.Commit}) } - r.f.loop.lastPushAt = time.Time{} r.f.loop.endWake(0) if r.m.endWake() { r.failf("the model sent a second apply in the settling wake") } } -// armed is the timer the worker has armed to fire first, when, and whether it is the commit timer. Two -// due at the same instant fire in either order, as run's select picks. -func (r *windowModelRun) armed() (time.Time, bool, bool) { +// modelTimer names one of the worker's timers the driver delivers. +type modelTimer string + +const ( + timerCommit modelTimer = "commit" + timerAttach modelTimer = "attach" + timerPush modelTimer = "push" + timerRetry modelTimer = "retry" +) + +// armedTimers is every timer the worker has armed, and when it is due on the worker's clock. +func (r *windowModelRun) armedTimers() map[modelTimer]time.Time { loop := r.f.loop - commit, attach := loop.commitTimer != nil, loop.attachTimer != nil - switch { - case commit && attach && loop.commitTimerAt.Equal(loop.attachTimerAt): - return loop.commitTimerAt, r.rng.Intn(2) == 0, true - case commit && (!attach || loop.commitTimerAt.Before(loop.attachTimerAt)): - return loop.commitTimerAt, true, true - case attach: - return loop.attachTimerAt, false, true - } - return time.Time{}, false, false -} - -// commitWake is the wake the commit timer delivers, as run does. -func (r *windowModelRun) commitWake() { - r.notef("commit timer") - r.f.loop.lastPushAt = time.Time{} - r.f.loop.commitTimer = nil - r.f.loop.closeOrArmWindow() - r.f.loop.endWake(0) - r.m.closeOrArm() - r.applied(r.m.endWake()) + armed := make(map[modelTimer]time.Time, 4) + if loop.commitTimer != nil { + armed[timerCommit] = loop.commitTimerAt + } + if loop.attachTimer != nil { + armed[timerAttach] = loop.attachTimerAt + } + if loop.pushTimer != nil { + armed[timerPush] = loop.pushTimerAt + } + if loop.retry.timer != nil { + armed[timerRetry] = loop.retry.due + } + return armed } -// 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) +// armed is the timer the worker has armed to fire first, and when. Two due at the same instant fire in +// either order, as run's select picks. +func (r *windowModelRun) armed() (time.Time, modelTimer, bool) { + var at time.Time + var first []modelTimer + for _, timer := range []modelTimer{timerCommit, timerAttach, timerPush, timerRetry} { + due, ok := r.armedTimers()[timer] + switch { + case !ok: + case len(first) == 0 || due.Before(at): + at, first = due, []modelTimer{timer} + case due.Equal(at): + first = append(first, timer) + } + } + if len(first) == 0 { + return time.Time{}, "", false + } + return at, first[r.rng.Intn(len(first))], true +} + +// deliver is the wake timer delivers, as run does. +func (r *windowModelRun) deliver(timer modelTimer) { + r.notef("%s timer", timer) + loop := r.f.loop + switch timer { + case timerCommit: + loop.commitTimer = nil + loop.closeOrArmWindow() + loop.endWake(0) + r.m.closeOrArm() + case timerAttach: + // endWake's serviceCommitRequests does its work. + loop.attachTimer = nil + loop.endWake(0) + case timerPush: + loop.pushTimer = nil + loop.pushPending() + loop.endWake(0) + r.m.push() + case timerRetry: + loop.retry.timer = nil + loop.runRetry() + loop.endWake(0) + r.m.push() + } r.applied(r.m.endWake()) } @@ -919,7 +1316,7 @@ func (r *windowModelRun) advance(d time.Duration) { r.notef("clock +%s", d) target := r.clock.Now().Add(d) for range 1000 { - at, commit, ok := r.armed() + at, timer, ok := r.armed() if !ok || at.After(target) { r.clock.advance(target.Sub(r.clock.Now())) r.m.now = target @@ -929,16 +1326,17 @@ func (r *windowModelRun) advance(d time.Duration) { r.clock.advance(at.Sub(r.clock.Now())) } r.m.now = r.clock.Now() - if commit { - r.commitWake() - } else { - r.attachWake() - } + r.deliver(timer) r.check("timer") } r.failf("a timer stayed due after its wake") } +// clockStep is how far one step of the clock goes: a second up to the profile's longest. +func (r *windowModelRun) clockStep() time.Duration { + return time.Second + time.Duration(r.rng.Int63n(int64(r.profile.clock/time.Second)))*time.Second +} + func (r *windowModelRun) freshValue(obj string) string { r.values++ return obj + "v" + strconv.Itoa(r.values) @@ -956,6 +1354,61 @@ func (r *windowModelRun) event(author, obj, value string) Event { return event } +// windowModelProfile weighs the inputs a run draws. own is the percentage of writes an author makes of +// an object of its own (alice x, bob w, carol z, each alone in its own file), so their windows can +// stay open side by side; the others are of any object. merge is the percentage of pushes from outside +// that move a document into another's file, and clock the longest step of the clock. +type windowModelProfile struct { + name string + own int + merge int + clock time.Duration + weights map[modelStep]int +} + +type modelStep string + +const ( + stepWrite modelStep = "write" + stepPressure modelStep = "write under buffer pressure" + stepAdvance modelStep = "clock" + stepOverdue modelStep = "overdue" + stepSave modelStep = "save" + stepForeign modelStep = "push from outside" + stepApply modelStep = "apply" + stepResync modelStep = "resync" + stepOutage modelStep = "outage" + stepRestart modelStep = "restart" + stepStale modelStep = "late applier write" +) + +// modelSteps fixes the order the weights are drawn in, so a seed replays. +var modelSteps = []modelStep{stepWrite, stepPressure, stepAdvance, stepOverdue, stepSave, stepForeign, stepApply, + stepResync, stepOutage, stepRestart, stepStale} + +var ( + // wideProfile draws every kind of input. + wideProfile = windowModelProfile{name: "wide", own: 67, merge: 25, clock: 25 * time.Second, + weights: map[modelStep]int{ + stepWrite: 36, stepPressure: 4, stepAdvance: 16, stepOverdue: 3, stepSave: 12, stepForeign: 6, + stepApply: 6, stepResync: 7, stepOutage: 8, stepRestart: 2, stepStale: 3, + }} + // concurrentProfile keeps authors on objects of their own, so several windows stay open through + // cooldowns, outages, replays and flushes. + concurrentProfile = windowModelProfile{name: "concurrent", own: 95, merge: 25, clock: 25 * time.Second, + weights: map[modelStep]int{ + stepWrite: 46, stepPressure: 3, stepAdvance: 14, stepOverdue: 4, stepSave: 9, stepForeign: 5, + stepApply: 5, stepResync: 3, stepOutage: 9, stepRestart: 2, stepStale: 3, + }} + // outageProfile is the concurrent one with short clock steps, so windows stay open across the + // cooldown and the retries, and with pushes from outside that restructure the folder under them. + outageProfile = windowModelProfile{name: "outage", own: 95, merge: 50, clock: 8 * time.Second, + weights: map[modelStep]int{ + stepWrite: 50, stepPressure: 2, stepAdvance: 16, stepOverdue: 3, stepSave: 5, stepForeign: 8, + stepApply: 3, stepResync: 4, stepOutage: 7, stepRestart: 1, stepStale: 2, + }} +) + // write is one accepted live write: a new value, an undo to an earlier one, the value the object // already holds, or a delete. Each author mostly edits an object of its own, so their windows can stay // open side by side, and sometimes any other. @@ -963,7 +1416,7 @@ func (r *windowModelRun) write(pressure bool) { who := r.rng.Intn(len(modelAuthors)) author := modelAuthors[who] obj := modelObjects[r.rng.Intn(len(modelObjects))] - if r.rng.Intn(3) > 0 { + if r.rng.Intn(100) < r.profile.own { obj = modelObjects[who] } current := r.m.cluster[obj] @@ -976,11 +1429,20 @@ func (r *windowModelRun) write(pressure bool) { case roll < 9 && current != "": value, kind = "", "delete" } + if value != "" && value == r.m.withheld[obj] { + // An edit back to the value a withheld apply wrote is carried as the apply's, a decided limit of + // the withheld-field record: the model does not make one. + value, kind = r.freshValue(obj), "write" + } + r.writeAs(author, obj, value, kind, pressure) +} + +// writeAs is the live write of obj to value by author. +func (r *windowModelRun) writeAs(author, obj, value, kind string, pressure bool) { 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 @@ -1012,6 +1474,12 @@ func (r *windowModelRun) save() { if r.rng.Intn(2) == 0 { req.IdleTimeout = durationPtr([]time.Duration{0, 5 * time.Second}[r.rng.Intn(2)]) } + r.register(req) +} + +// register sends the save req, which names the run's next save. +func (r *windowModelRun) register(req *AttachCommitRequest) { + name, author := req.Name, req.Author 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, @@ -1021,6 +1489,9 @@ func (r *windowModelRun) save() { // foreign pushes a commit from outside and reports it, as a Git host's push notice does. func (r *windowModelRun) foreign() { + if r.rng.Intn(100) < r.profile.merge && r.mergeOutside() { + return + } obj := modelObjects[r.rng.Intn(len(modelObjects))] r.values++ value := "outside" + strconv.Itoa(r.values) @@ -1032,6 +1503,61 @@ func (r *windowModelRun) foreign() { r.wake(nil, func() { r.m.hinted = true }) } +// mergeOutside moves, from outside, the document of one of x, w and z into the file another is alone +// in, and reports it, when two of them are each alone in a file of their own: windows proven apart +// before the move overlap once it is found. +func (r *windowModelRun) mergeOutside() bool { + var alone []string + for _, obj := range modelObjects[:3] { + if at := r.m.remote.loc(obj); at != "" && len(r.m.remote.files[at]) == 1 && at == echoFile(obj) { + alone = append(alone, obj) + } + } + if len(alone) < 2 { + return false + } + r.rng.Shuffle(len(alone), func(i, j int) { alone[i], alone[j] = alone[j], alone[i] }) + r.merge(alone[0], alone[1]) + return true +} + +// merge moves, from outside, obj's document into the file into's is alone in, and reports it. +func (r *windowModelRun) merge(into, obj string) { + r.notef("outside push moves %s into %s's file, hinted", obj, into) + r.pushFromOutside(r.m.remote, r.m.remote.merge(into, obj), "outside: move "+obj+" in with "+into) + r.m.foreignUnseen = true + r.f.loop.pushHinted = true + r.wake(nil, func() { r.m.hinted = true }) +} + +// restructure is a push from outside that moves bob's document into alice's file while each has a +// window open, proven apart on the tree before it, and a save keeps the apply the push brings from +// closing them: once the worker finds the push, the next write proves both windows again, finds them +// overlapping, and closes them. It needs x and w each alone in its file, and a recent push, so the +// cooldown holds the push the hint makes; a run starts it with neither. +func (r *windowModelRun) restructure() { + r.writeAs("alice", "x", r.freshValue("x"), "write", false) + r.advance(modelIdle + time.Second) // alice's window closes, and its push is made at once + r.check("restructure: a push") + r.saves++ + r.register(&AttachCommitRequest{ + Namespace: "default", Name: "save-" + strconv.Itoa(r.saves), UID: "uid-save-" + strconv.Itoa(r.saves), + Author: "carol", GitTargetName: echoTarget, GitTargetNamespace: "default", + Message: "feat: carol's save", Attach: configv1alpha3.AttachCurrentOrNext, AttachTimeout: 20 * time.Second, + MaxDuration: 40 * time.Second, + }) + r.check("restructure: a save waits") + r.writeAs("bob", "z", r.freshValue("z"), "write", false) + r.merge("x", "w") // the hint closes bob's window, and the cooldown holds its push + r.check("restructure: a push from outside") + r.writeAs("alice", "x", r.freshValue("x"), "write", false) + r.writeAs("bob", "w", r.freshValue("w"), "write", false) + r.check("restructure: two windows") + r.advance(modelCooldown) // the push replays onto the move; carol's save holds the apply + r.check("restructure: the replay") + r.writeAs("carol", "z", r.freshValue("z"), "write", false) +} + // apply makes an apply due for the folder's applier (a Repair). func (r *windowModelRun) apply() { r.notef("apply due") @@ -1050,55 +1576,194 @@ func (r *windowModelRun) resync() { 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) + var decision int + r.wake(&item, func() { + r.m.resync(req.charge()) + decision = r.m.decisions + }) + // Its caller hears at once: what it found, once it is committed, or the failure a retry waits out. + deferred := slices.ContainsFunc(r.m.pending, func(p modelPending) bool { + return p.decision == decision && !p.committed + }) + if result := <-req.Result; (result.Err != nil) != deferred { + r.failf("resync: %v; the model has it deferred: %v", result.Err, deferred) + } +} + +// staleApply is an applier's write landing over a newer edit: an apply is sent and reported, an author +// edits an object of the apply's commit after it, and then the applier's write of that object, with +// the value the commit holds, arrives inside the apply's window. The worker keeps it out of Git, and +// asks the applier for a Repair, which puts the edit back once it is published. +func (r *windowModelRun) staleApply() { + r.sent = false + r.apply() + who := r.rng.Intn(len(modelAuthors)) + obj := modelObjects[who] + if !r.sent || r.m.applied[obj] == "" { + return } + r.check("apply") + r.writeAs(modelAuthors[who], obj, r.freshValue(obj), "write", false) + r.check("write") + stale := r.m.applied[obj] + r.notef("the applier writes %s=%q late, over the edit", obj, stale) + event := echoEvent(obj, stale) + event.UserInfo.Username = "system:serviceaccount:flux-system:kustomize-controller" + event.LastFieldManagers = []string{FluxFieldManager} + event.LastFieldManagedAt = time.Now().Add(time.Hour) + r.wake(&WorkItem{Request: &WriteRequest{Events: []Event{event}}}, func() { + r.m.reached(reachStaleApply) + r.m.withheld[obj] = stale + r.m.applyDue = true + }) +} + +// toggleOutage takes the Git server down, or brings it back. +func (r *windowModelRun) toggleOutage() { + r.m.down = !r.m.down + r.outage.down = r.m.down + r.notef("outage down=%v", r.m.down) } 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: +// overdue steps the clock past the timers' deadlines without delivering them, handles one input first, +// as run's select may when an item and a timer are both ready, and then delivers the timers. +func (r *windowModelRun) overdue() { + d := r.clockStep() + r.notef("clock +%s, timers held", d) + r.clock.advance(d) + r.m.now = r.clock.Now() + if at, _, ok := r.armed(); ok && !at.After(r.m.now) { + r.m.reached(reachOverdue) + } + if r.rng.Intn(2) == 0 { 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: + } else { 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) } + r.check("overdue input") + r.advance(0) +} + +// restart stops the worker, as a rollout does, and starts a new one on the same Git server with an +// empty checkout, once the server is up: what the old one could not push is lost with its checkout, +// and the new one starts with no window, save or pending write. +func (r *windowModelRun) restart() { + r.shutdown() + r.check("shutdown") + r.checkGiven() + r.m.reached(reachRestart) + if r.m.down { + r.toggleOutage() + } + r.notef("restart") + old := r.f.worker + r.f.loop.stopTimers() + w := NewBranchWorker(old.Client, old.Log, old.GitProviderRef, old.GitProviderNamespace, old.Branch, old.repo, + nil, BranchWorkerLimits{}) + w.Log, w.ctx, w.GitBranch, w.mapper, w.clock = old.Log, old.ctx, old.GitBranch, old.mapper, old.clock + w.applyRequester = old.applyRequester + require.NoError(r.t, os.RemoveAll(w.repoRootPath())) + require.NoError(r.t, w.ensureRepositoryInitialized(w.ctx)) + loop := newBranchWorkerEventLoop(w, time.Hour) + r.t.Cleanup(loop.stopTimers) + loop.defaultWindow = commitWindowDefaults{idleTimeout: modelIdle, maxDuration: modelMax} + loop.followRecordedOperations(time.Now()) + // As at the start: the converges a new worker leaves pending are answered already. + w.inbox.take() + for _, state := range loop.appliers { + state.converge = false + } + r.f.worker, r.f.loop = w, loop + r.exited, r.firstSave = false, r.saves+1 + r.knownTip = gitIn(r.t, r.f.repoDir, "rev-parse", "main") + r.m.restart() +} + +func (r *windowModelRun) step() { + step := r.drawStep() + switch { + case step == stepForeign && r.resyncWaiting(): + // A resync that waits for its retry fetches before it commits, and a push from outside found + // by that fetch discards it: the model does not cover that. + step = stepWrite + case step == stepResync && (r.m.foreignUnseen || r.m.applyDue): + step = stepWrite + } + map[modelStep]func(){ + stepWrite: func() { r.write(false) }, + stepPressure: func() { r.write(true) }, + stepAdvance: func() { r.advance(r.clockStep()) }, + stepOverdue: r.overdue, + stepSave: r.save, + stepForeign: r.foreign, + stepApply: r.apply, + stepResync: r.resync, + stepOutage: r.toggleOutage, + stepRestart: r.restart, + stepStale: r.staleApply, + }[step]() +} + +// drawStep draws a step by the profile's weights. +func (r *windowModelRun) drawStep() modelStep { + total := 0 + for _, s := range modelSteps { + total += r.profile.weights[s] + } + roll := r.rng.Intn(total) + for _, s := range modelSteps { + if roll < r.profile.weights[s] { + return s + } + roll -= r.profile.weights[s] + } + return stepWrite +} + +// resyncWaiting reports whether a decided resync waits for its retry to be committed. +func (r *windowModelRun) resyncWaiting() bool { + return slices.ContainsFunc(r.m.pending, func(p modelPending) bool { return p.resync && !p.committed }) } func (r *windowModelRun) run(steps int) { + if r.profile.name == outageProfile.name { + r.restructure() + r.check("restructure") + } for range steps { r.step() r.check("step") - if len(r.m.windows) > 1 { + m := r.m + if m.several() { r.concurrent++ + if !m.pushAt.IsZero() { + m.reached(reachCooldown) + } + if !m.retryAt.IsZero() { + m.reached(reachRetryWait) + } + } + if !m.holdsDecided() { + m.reached(reachDeferred) } } if r.rng.Intn(2) == 0 { r.shutdown() r.check("shutdown") } else { - r.advance(3 * modelMax) + if r.m.down { + r.toggleOutage() + } + // Every window closes, and the retry, at its longest backoff, comes due. + r.advance(modelMax + modelRetryMax) } r.drained() } @@ -1106,10 +1771,15 @@ func (r *windowModelRun) run(steps int) { var closeLine = regexp.MustCompile(`"msg"="Finalizing open commit window" "reason"="([^"]+)" ` + `"windowAuthor"="([^"]*)" "windowTarget"="[^"]*" "window"=(\d+)`) +// check compares the worker with the model after a step. // check compares the worker with the model after a step. func (r *windowModelRun) check(what string) { r.t.Helper() + if r.m.surprise != "" { + r.failf("%s: the model does not cover this: %s", what, r.m.surprise) + } r.checkWindows(what) + r.checkPending(what) r.checkTimers(what) r.checkSaves(what) r.checkCloses(what) @@ -1119,31 +1789,62 @@ func (r *windowModelRun) check(what string) { } } -// checkWindows compares the open windows and what is retained. +// checkWindows compares the open windows and what they are charged. func (r *windowModelRun) checkWindows(what string) { r.t.Helper() loop, m := r.f.loop, r.m if got, want := r.workerWindows(), m.describeWindows(); got != want { r.failf("%s: the open windows differ:\n worker %s\n model %s", what, got, want) } - if got, want := loop.totalRetainedBytes(), m.windowCharge(); got != want || loop.windows.bytes() != want { - r.failf("%s: retained %d (windows %d), the model's window charge is %d", what, got, - loop.windows.bytes(), want) + if got, want := loop.windows.bytes(), m.windowCharge(); got != want { + r.failf("%s: the windows are charged %d; the model's %d", what, got, want) + } +} + +// checkPending compares the pending writes, in order, what each is charged, which are committed, and +// what everything retained is charged. +func (r *windowModelRun) checkPending(what string) { + r.t.Helper() + loop, m := r.f.loop, r.m + var got, want []string + for i := range loop.pendingWrites { + pw := &loop.pendingWrites[i] + save := "" + if pw.CommitRequest != nil { + save = pw.CommitRequest.Name + } + got = append(got, fmt.Sprintf("resync=%v save=%q charge=%d committed=%v", pw.Kind == PendingWriteResync, + save, pw.charge, pw.committedOnce)) + } + for _, p := range m.pending { + save := "" + if p.save != nil { + save = p.save.name + } + want = append(want, fmt.Sprintf("resync=%v save=%q charge=%d committed=%v", p.resync, save, p.charge, + p.committed)) + } + if !slices.Equal(got, want) { + r.failf("%s: the pending writes differ:\n worker %v\n model %v", what, got, want) } - if len(loop.pendingWrites) != 0 { - r.failf("%s: %d writes still pending after a wake that pushed at once", what, len(loop.pendingWrites)) + if loop.pendingWritesBytes != m.pendingCharge() { + r.failf("%s: the pending writes are charged %d; the model's %d", what, loop.pendingWritesBytes, + m.pendingCharge()) + } + if got, want := loop.totalRetainedBytes(), m.windowCharge()+m.pendingCharge(); got != want { + r.failf("%s: retained %d; the model's %d", what, got, want) } } // checkTimers compares the timers the worker has armed with the model's deadlines: the commit timer // for the earliest close of an open window, the attach timer for the earliest deadline of a save still -// waiting. +// waiting, the push timer for the end of the cooldown, and the retry timer for the next attempt. func (r *windowModelRun) checkTimers(what string) { r.t.Helper() if r.exited { return } - loop, m := r.f.loop, r.m + m := r.m var commitAt, attachAt time.Time for _, w := range m.windows { if at, _ := w.closeAt(); commitAt.IsZero() || at.Before(commitAt) { @@ -1155,11 +1856,17 @@ func (r *windowModelRun) checkTimers(what string) { 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) + want := map[modelTimer]time.Time{timerCommit: commitAt, timerAttach: attachAt, timerPush: m.pushAt, + timerRetry: m.retryAt} + armed := r.armedTimers() + var gotTimers, wantTimers []string + for _, timer := range []modelTimer{timerCommit, timerAttach, timerPush, timerRetry} { + at, ok := armed[timer] + gotTimers = append(gotTimers, string(timer)+" "+r.describeTimer(ok, at)) + wantTimers = append(wantTimers, string(timer)+" "+r.describeTimer(!want[timer].IsZero(), want[timer])) + } + if !slices.Equal(gotTimers, wantTimers) { + r.failf("%s: the armed timers differ:\n worker %v\n model %v", what, gotTimers, wantTimers) } } @@ -1212,17 +1919,23 @@ func (r *windowModelRun) checkSaves(what string) { 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) - } + for i := r.firstSave; i <= r.saves; i++ { + r.checkOutcome(what, "save-"+strconv.Itoa(i)) + } +} + +// checkOutcome compares a save's outcome, or none, and how often it settled. +func (r *windowModelRun) checkOutcome(what, name string) { + r.t.Helper() + res, resolved := r.f.worker.LookupCommitRequestOutcome("default", name, "uid-"+name) + want, modelResolved := r.m.outcomes[name] + if resolved != modelResolved || (resolved && (res.Err != nil) != want.failed) || + (resolved && !want.failed && res.Outcome != want.outcome) { + r.failf("%s: %s settled %v as %q (err %v); the model says %v as %+v", 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) } } @@ -1451,19 +2164,27 @@ func (r *windowModelRun) checkBootstrap(what, rev, name string) { func (r *windowModelRun) drained() { r.t.Helper() r.check("drained") - if len(r.m.windows) != 0 || len(r.m.saves) != 0 || len(r.m.pending) != 0 { + // A shutdown during an outage loses the writes its last push could not publish, with the checkout. + lost := r.exited && r.m.down + if len(r.m.windows) != 0 || len(r.m.saves) != 0 || (len(r.m.pending) != 0 && !lost) { 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. + // An exited loop is dropped with the saves it gave back still counted in heldBytes, and the writes + // it lost; what it retains for Git is nothing else. held := r.f.loop.heldBytes() if r.exited { held = 0 } - if held != 0 || r.f.loop.totalRetainedBytes() != 0 { + if held != 0 || r.f.loop.totalRetainedBytes() != r.m.pendingCharge() { r.failf("drained, the worker still charges %d (retained %d)", held, r.f.loop.totalRetainedBytes()) } r.checkTree() + r.checkGiven() +} + +// checkGiven checks that the saves given back at shutdown have no outcome and no phase. +func (r *windowModelRun) checkGiven() { + r.t.Helper() 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) @@ -1513,11 +2234,21 @@ func envInt(name string, fallback int) int { return fallback } +// reachFloors are the situations the search must reach across its seeds, at the default size or larger, +// or a pass says nothing about them. +var reachFloors = []string{ + reachShutdown, reachFlush, reachHint, reachAttachNext, reachReproof, reachPathConflict, reachFailedPush, + reachCooldown, reachRetryWait, reachDeferred, reachReplay, reachPressure, reachApplyWaits, reachRestart, + reachOverdue, reachStaleApply, +} + // TestWindowModel_TheWorkerFollowsTheReferenceModel runs random input sequences through a worker and -// the model, and checks every step. Across the seeds, some steps must end with several windows open, +// the model, and checks every step. Half the seeds draw the wide profile, half the concurrent one. +// Across the seeds, some steps must end with several windows open, and every reach floor must be met, // or the search says nothing about them. func TestWindowModel_TheWorkerFollowsTheReferenceModel(t *testing.T) { - seeds, steps := envInt("WINDOW_MODEL_SEEDS", 6), envInt("WINDOW_MODEL_STEPS", 40) + const defaultSeeds = 20 + seeds, steps := envInt("WINDOW_MODEL_SEEDS", defaultSeeds), envInt("WINDOW_MODEL_STEPS", 60) if testing.Short() { seeds = 2 } @@ -1526,15 +2257,30 @@ func TestWindowModel_TheWorkerFollowsTheReferenceModel(t *testing.T) { first, seeds = int64(seed), 1 } concurrent := 0 + reach := map[string]int{} for seed := first; seed < first+int64(seeds); seed++ { t.Run("seed-"+strconv.FormatInt(seed, 10), func(t *testing.T) { run := newWindowModelRun(t, seed) run.run(steps) concurrent += run.concurrent + for what, n := range run.m.reach { + reach[what] += n + } }) } t.Logf("%d steps ended with several windows open", concurrent) - if seeds > 1 && !t.Failed() { + for _, what := range reachFloors { + t.Logf("reached %d times: %s", reach[what], what) + } + if t.Failed() { + return + } + if seeds > 1 { require.Positive(t, concurrent, "no step ended with several windows open") } + if seeds >= defaultSeeds && steps >= 60 { + for _, what := range reachFloors { + assert.Positive(t, reach[what], "the search never reached %s", what) + } + } } From d13b5f63a078637a73952b0fbfcde09b6064a2d1 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Mon, 9 Nov 2026 18:19:07 +0000 Subject: [PATCH 3/5] test(git): pin a footprint revision's generation and resets, and two guards no input reaches Co-Authored-By: Claude Opus 5.5 --- internal/git/retry_test.go | 27 ++++++++ internal/git/window_admission_test.go | 93 +++++++++++++++++++++++++++ 2 files changed, 120 insertions(+) diff --git a/internal/git/retry_test.go b/internal/git/retry_test.go index b722a5ea..598cc9d5 100644 --- a/internal/git/retry_test.go +++ b/internal/git/retry_test.go @@ -20,6 +20,8 @@ import ( "github.com/go-logr/logr" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + + "github.com/ConfigButler/gitops-reverser/internal/manifestanalyzer" ) // failNextPushes makes the next n push attempts fail as an unreachable remote would, and counts every @@ -224,3 +226,28 @@ func TestPublicationRetry_AHeldCommitRequestIsCommittedWithoutAnotherWrite(t *te assert.Equal(t, f.ref("feature").String(), res.Commit) assert.Equal(t, int32(2), attempts.Load(), "one failed push, then the retry") } + +// An outage can end with nothing to publish: a resync decided while Git is down waits for the retry, +// and committed then it changes nothing, so no write is left to push. The outage is over all the same: +// the retry closes, and the next failure waits the first backoff again, not twice the last. +func TestPublicationRetry_AnOutageEndsWhenItsWritesSettleWithoutAPush(t *testing.T) { + f := newAdmissionFixture(t, "retry-settles", nil) + f.loop.pushPending() // the bootstrap write + clock := useTestClock(f.worker) + outage := injectOutage(t) + outage.down = true + req := &ResyncRequest{Desired: []manifestanalyzer.DesiredResource{echoDesired("x", "x1")}, + GitTargetName: echoTarget, GitTargetNamespace: "default", Result: make(chan ResyncResult, 1)} + f.loop.handleQueueItem(resyncItem(req)) + require.Error(t, (<-req.Result).Err, "precondition: the resync waits for the retry") + require.Len(t, f.loop.pendingWrites, 1) + require.True(t, f.loop.retry.pending()) + + outage.down = false + clock.advance(retryInitialBackoff) + fireRetry(f.loop) + + require.Empty(t, f.loop.pendingWrites, "the resync changed nothing") + assert.False(t, f.loop.retry.pending(), "the outage is over") + assert.Zero(t, f.loop.retry.backoff, "and the next one starts from the first backoff") +} diff --git a/internal/git/window_admission_test.go b/internal/git/window_admission_test.go index 74481670..a1946330 100644 --- a/internal/git/window_admission_test.go +++ b/internal/git/window_admission_test.go @@ -5,6 +5,8 @@ package git import ( "fmt" "maps" + "os" + "path/filepath" "strings" "testing" "time" @@ -570,3 +572,94 @@ func TestFootprint_OverlapUnionAndOutside(t *testing.T) { assert.Equal(t, "unknown: why", unknown.String()) assert.Equal(t, "unproven", footprintString(nil)) } + +// TestWindowAdmission_AGitTargetChangeProvesTheOpenWindowsAgain pins the generation in a footprint's +// revision: a GitTarget that now places by a template proves nothing, so two windows proven apart +// before the change close once the next write proves them again. With the generation left out, the +// index of the old spec was reused, and they stayed open side by side. +func TestWindowAdmission_AGitTargetChangeProvesTheOpenWindowsAgain(t *testing.T) { + f := newAdmissionFixture(t, "admission-generation", nil) + f.writeBy("alice", "x", "x2") + f.writeBy("bob", "z", "z2") + require.Equal(t, []string{"alice", "bob"}, authorsOf(f.loop.windows.all())) + + var target configv1alpha3.GitTarget + require.NoError(t, f.worker.Client.Get(f.worker.ctx, keyOf(echoTarget), &target)) + target.Spec.Placement = &configv1alpha3.GitTargetPlacementSpec{Default: "all.yaml"} + target.Generation++ + require.NoError(t, f.worker.Client.Update(f.worker.ctx, &target)) + f.writeBy("carol", "q", "q3") + + assert.Equal(t, []string{"carol"}, authorsOf(f.loop.windows.all()), + "proven again under the new spec, the open windows are unknown, and close before carol's write") +} + +// TestWindowAdmission_AResetProvesTheOpenWindowsAgain pins the resets in a footprint's revision. A push +// from outside makes carol's held write a no-op and moves z's document in with x's; the replay onto +// it resets the checkout and commits nothing, so only the reset says the tree changed. The next write +// proves alice's and bob's windows again, finds both in one file, and closes them. With the resets left +// out, the index of the tree before the reset was reused, and both stayed open on one file. +func TestWindowAdmission_AResetProvesTheOpenWindowsAgain(t *testing.T) { + f := newAdmissionFixture(t, "admission-reset", nil) + f.loop.pushPending() // the bootstrap write; the cooldown holds every push after it + f.writeBy("alice", "x", "x2") + f.writeBy("bob", "z", "z2") + f.writeBy("carol", "q", "q3") + open := f.loop.windows.all() + require.Equal(t, []string{"alice", "bob", "carol"}, authorsOf(open)) + f.loop.finalizeWindow(open[2], windowFinalizeReasonIdleTimeout) + f.loop.endWake(0) + require.Len(t, f.loop.pendingWrites, 1, "precondition: the cooldown holds carol's commit") + // Proven on the tree carol's commit made, so only the reset below can make these proofs stale. + f.writeBy("alice", "x", "x3") + + gitIn(t, f.clone, "pull", "--quiet", "--ff-only", "origin", "main") + write := func(path, content string) { + require.NoError(t, os.WriteFile(filepath.Join(f.clone, path), []byte(content), 0o600)) + } + write(echoFile("q"), echoDocument("q", "q3")) + write(echoFile("x"), echoDocument("x", "x1")+"---\n"+echoDocument("z", "z1")) + gitIn(t, f.clone, "rm", "--quiet", echoFile("z")) + gitIn(t, f.clone, "add", "-A") + gitIn(t, f.clone, "commit", "--quiet", "-m", "outside: q3, and z in with x") + gitIn(t, f.clone, "push", "--quiet", "origin", "HEAD:main") + resets, commits := f.worker.checkoutResets.Load(), f.worker.checkoutCommits.Load() + f.loop.pushPending() + require.Empty(t, f.loop.pendingWrites) + require.Greater(t, f.worker.checkoutResets.Load(), resets, "precondition: the replay reset the checkout") + require.Equal(t, commits, f.worker.checkoutCommits.Load(), "precondition: and committed nothing") + + f.writeBy("dave", "q", "q4") + assert.Equal(t, []string{"dave"}, authorsOf(f.loop.windows.all()), + "proven again on the tree the reset brought, alice's and bob's windows share a file") +} + +// TestWindowAdmission_AWindowAnAttachClosedTakesNoRequest pins the guard in +// attachWaitingCommitRequests: an attach closes every window that is due, so a window later in the +// pass may be gone, and a request that would have attached to it keeps waiting for the next. The +// state is built by hand: a request waits only while its author has no window to take, so the model +// never meets a second window with a request waiting for it in the pass. +func TestWindowAdmission_AWindowAnAttachClosedTakesNoRequest(t *testing.T) { + f := newAdmissionFixture(t, "admission-attach-closed", nil) + f.writeBy("alice", "x", "x2") + f.writeBy("bob", "z", "z2") + open := f.loop.windows.all() + require.Equal(t, []string{"alice", "bob"}, authorsOf(open)) + open[1].timers.maxAt = f.worker.now().Add(-time.Second) // bob's window is due, its timer not delivered + + waiting := func(author string, seq uint64) *pendingCommitRequest { + return &pendingCommitRequest{ + id: commitRequestID{Namespace: "default", Name: "save-" + author, UID: "uid-" + author}, + author: author, gitTargetName: echoTarget, gitTargetNamespace: "default", seq: seq, + message: "feat: " + author, maxDuration: time.Hour, attachDeadline: f.worker.now().Add(time.Minute), + } + } + alice, bob := waiting("alice", 1), waiting("bob", 2) + f.loop.pendingCRs = map[commitRequestID]*pendingCommitRequest{alice.id: alice, bob.id: bob} + + f.loop.attachWaitingCommitRequests() + + assert.True(t, alice.attached, "alice's request takes her window") + assert.Equal(t, []string{"alice"}, authorsOf(f.loop.windows.all()), "and the attach closed bob's, which was due") + assert.False(t, bob.attached, "bob's request takes no window that is gone: it waits for his next") +} From 8ab64d0dfcf8840d49d080833236db9da744010b Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Mon, 9 Nov 2026 18:19:08 +0000 Subject: [PATCH 4/5] test(e2e): two authors save through a Git outage Co-Authored-By: Claude Opus 5.5 --- .../e2e/concurrent_authors_outage_e2e_test.go | 110 ++++++++++++++++++ 1 file changed, 110 insertions(+) create mode 100644 test/e2e/concurrent_authors_outage_e2e_test.go diff --git a/test/e2e/concurrent_authors_outage_e2e_test.go b/test/e2e/concurrent_authors_outage_e2e_test.go new file mode 100644 index 00000000..250447aa --- /dev/null +++ b/test/e2e/concurrent_authors_outage_e2e_test.go @@ -0,0 +1,110 @@ +// SPDX-License-Identifier: Apache-2.0 + +package e2e + +import ( + "fmt" + "path" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// Concurrent Authors through an outage is the flagship of docs/design/concurrent-author-windows.md +// with Git refusing every push while two people work: alice and bob each work and save with their +// windows open side by side, and the pushes their saves make fail. The branch worker keeps both saves' +// writes, committed locally, and its retry publishes them once Git takes pushes again, with no further +// edit: each save still comes out as one commit of its own, its author, its message and its files. +// +// The outage is the GitProvider's credential going bad: every push and fetch is refused, as when the +// Git host is down, while the Gitea every other spec uses stays up. Putting it back ends the outage. +// +// Not Serial: the spec owns its namespace, its Gitea repository and the credential it breaks. +var _ = Describe("Concurrent Authors through an outage", Label("concurrent-authors-outage", "commit-request", + "audit-consumer"), Ordered, func() { + const ( + basePath = "e2e/concurrent-authors-outage" + commitWindow = "120s" + ) + var p windowedDeploymentPipeline + + BeforeAll(func() { + p = setupWindowedDeploymentPipeline("authors-outage", basePath, commitWindow) + }) + + AfterAll(func() { + cleanupPipeline(p.ns, p.gitProvider, p.gitTarget, p.watchRule) + cleanupNamespace(p.ns) + }) + + It("publishes both saves, one commit each, once Git takes pushes again", func() { + seed := GinkgoRandomSeed() + alice := author{user: "alice", message: "feat: add the storefront"} + bob := author{user: "bob", message: "feat: add the order queue"} + aliceAPI := fmt.Sprintf("storefront-api-%d", seed) + aliceWeb := fmt.Sprintf("storefront-web-%d", seed) + bobQueue := fmt.Sprintf("order-queue-%d", seed) + file := func(name string) string { + return path.Join(basePath, p.ns, "apps/deployments", name+".yaml") + } + + By("seeding the branch, so the folder's bootstrap files are committed") + seedName := fmt.Sprintf("seed-%d", seed) + _, err := kubectlRunWithStdin(p.ns, + commitRequestManifest(p.ns, "save-seed", p.gitTarget, "chore: seed", "8s")+"---\n"+ + deploymentManifest(p.ns, seedName), "apply", "-f", "-") + Expect(err).NotTo(HaveOccurred()) + var seedSHA string + Eventually(func(g Gomega) { + seedSHA = expectCommitRequestCommitted(g, p.ns, "save-seed", + recentCommitDiagnostics(p.repo.CheckoutDir, basePath)) + }, 2*time.Minute, 2*time.Second).Should(Succeed()) + + By("Git starts refusing every push and fetch") + _, err = kubectlRunInNamespace(p.ns, "patch", "secret", p.repo.GitSecretHTTP, "--type=merge", + "-p", `{"stringData":{"password":"refused-during-the-outage"}}`) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { + _, _ = kubectlRunInNamespace(p.ns, "apply", "-f", p.repo.SecretsYAML) + }) + + By("alice and bob each work, interleaved, and save") + alice.create(p.ns, aliceAPI) + bob.create(p.ns, bobQueue) + alice.create(p.ns, aliceWeb) + alice.save(p.ns, "save-storefront", p.gitTarget) + bob.save(p.ns, "save-order-queue", p.gitTarget) + + By("nothing reaches Git while it refuses pushes, and neither save gives up") + Consistently(func(g Gomega) { + pullLatestRepoState(g, p.repo.CheckoutDir) + g.Expect(headOf(g, p.repo.CheckoutDir)).To(Equal(seedSHA), "a push landed during the outage\n%s", + recentCommitDiagnostics(p.repo.CheckoutDir, basePath)) + for _, save := range []string{"save-storefront", "save-order-queue"} { + outcome := readCommitRequestOutcome(g, p.ns, save) + g.Expect(outcome.isTerminal()).To(BeFalse(), "%s ended during the outage: %s", save, outcome) + } + }, 30*time.Second, 3*time.Second).Should(Succeed()) + + By("Git takes pushes again; the retry publishes both saves without another edit") + _, err = kubectlRunInNamespace(p.ns, "apply", "-f", p.repo.SecretsYAML) + Expect(err).NotTo(HaveOccurred()) + var aliceSHA, bobSHA string + Eventually(func(g Gomega) { + diagnostics := recentCommitDiagnostics(p.repo.CheckoutDir, basePath) + aliceSHA = expectCommitRequestCommitted(g, p.ns, "save-storefront", diagnostics) + bobSHA = expectCommitRequestCommitted(g, p.ns, "save-order-queue", diagnostics) + }, 3*time.Minute, 2*time.Second).Should(Succeed()) + + By("each save is one commit: its author, its message, its files and nothing else") + Eventually(func(g Gomega) { + pullLatestRepoState(g, p.repo.CheckoutDir) + g.Expect(commitsSince(g, p.repo.CheckoutDir, seedSHA)).To(ConsistOf(aliceSHA, bobSHA), + "exactly the two saves were committed after the seed\n%s", + recentCommitDiagnostics(p.repo.CheckoutDir, basePath)) + alice.expectCommit(g, p.repo.CheckoutDir, aliceSHA, file(aliceAPI), file(aliceWeb)) + bob.expectCommit(g, p.repo.CheckoutDir, bobSHA, file(bobQueue)) + }, time.Minute, 2*time.Second).Should(Succeed()) + }) +}) From b62616f987087cbc581301046f21333fbbd99400 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Mon, 9 Nov 2026 18:19:08 +0000 Subject: [PATCH 5/5] docs: stage 3 of concurrent commit windows is built Co-Authored-By: Claude Opus 5.5 --- docs/TODO.md | 42 ++++++++++-------------- docs/design/concurrent-author-windows.md | 35 +++++++++++++++++--- 2 files changed, 49 insertions(+), 28 deletions(-) diff --git a/docs/TODO.md b/docs/TODO.md index 781f0e37..59617b24 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -7,30 +7,11 @@ should land before the next tag is cut (the last release is v0.55.0). Checked ag ## Before the next release -- [ ] **Test the failure boundaries of concurrent commit windows** (stage 3 of - [concurrent-author-windows.md](design/concurrent-author-windows.md#stages)). Stage 2 ships with no - flag: a branch worker keeps a window per author whenever their footprints are proven disjoint. - Before it is released, the reference model (`internal/git/window_model_test.go`) must cover both - close orders, conflicting writes, delete and recreate, `attach_next`, zero-duration windows, a - resync, shutdown, failed publication and replay, and buffer pressure, each with several windows - open. The reviews of stages 1 and 2 add: - - **gate:** the model retains work across failed pushes and retries before several windows are - released. Today it publishes at once after every wake (`windowModelRun.wake`) and asserts no - pending writes after it (`drained`), so decided-but-uncommitted writes with windows open - (cooldown, failed push, replay) are never modeled; - - reach floors the model asserts: shutdown, flush and push hint with two or more windows open, - `attach_next` with others open, a reproof overlap, `path_conflict`; a concurrency-biased profile; - CI raised to about 20 seeds × 60 steps; - - model steps: a shutdown mid-run; a timer overdue but not yet delivered (the `!isOpen` guard in - `attachWaitingCommitRequests` survives its removal); a late stale applier write (the model - never withholds); - - the GitTarget-generation and checkout-reset parts of a footprint revision are untested; - - measure the footprint index on realistic repositories and commit rates before widening the - layouts it proves. It is rebuilt whenever its revision changes, every local commit included; - a review measured a rebuild at about 3.5 ms and 3.9 MB allocated for 100 small manifests, and - 33 ms and 37 MB for 1,000 (allocation, not retained heap), against 19 µs for a reuse. The - retained-byte budget does not charge the indexes, the accepted versions or the withheld-field - records, so a full budget does not bound the worker's memory. +- [ ] **See the concurrent-authors outage e2e pass in CI** (`concurrent-authors-outage`, + `test/e2e/concurrent_authors_outage_e2e_test.go`): two authors save while every push is refused, + and the retry publishes both, one commit each, once Git takes pushes again. It is the last part of + stage 3 of [concurrent-author-windows.md](design/concurrent-author-windows.md#stages); the reference + model's part is done. ## Applier claims and bi-directional editing @@ -120,6 +101,19 @@ Their argument for each is in [gitops-api-open-asks.md](future/gitops-api-open-a ## Branch worker and CommitRequests +- [ ] **Measure the footprint index** on realistic repositories and commit rates before widening the + layouts it proves. It is rebuilt whenever its revision changes, every local commit included; a + review measured a rebuild at about 3.5 ms and 3.9 MB allocated for 100 small manifests, and 33 ms + and 37 MB for 1,000 (allocation, not retained heap), against 19 µs for a reuse. The retained-byte + budget does not charge the indexes, the accepted versions or the withheld-field records, so a full + budget does not bound the worker's memory. +- [ ] What the window reference model (`internal/git/window_model_test.go`) does not cover: a resync + waiting for its retry whose fetch finds a push from outside and discards it (the model makes no + push from outside while a resync waits); objects of more than one field, so a withheld apply's + record is only ever one value; an applier write after its apply window's grace (the window runs on + the wall clock, so the model sends each late write inside it). Two guards no input reaches are pinned by unit + tests instead: the outage ending when its writes settle with no push, and an attach that closes a + later window in the same pass. - [ ] `expireWaitingCommitRequests` ranges over a map, so empty commits for saves that expire in the same pass land in random order; sort by `seq` (`internal/git/commit_request_attach_loop.go`). - [ ] `buildRequestRecordWrite` keeps only the username (`UserInfo{Username: pcr.author}`); carry the diff --git a/docs/design/concurrent-author-windows.md b/docs/design/concurrent-author-windows.md index d94f92c6..985bdbb3 100644 --- a/docs/design/concurrent-author-windows.md +++ b/docs/design/concurrent-author-windows.md @@ -295,8 +295,8 @@ window, so they are proven against today's behavior before they are used to judg ## Stages -Stages 1 and 2 are built; stage 3 is in progress, stage 4 was decided against, and stage 5 is -proposed. Stage 3 gates the next release, because stage 2 shipped without a flag. +Stages 1, 2 and 3 are built; stage 4 was decided against, and stage 5 is proposed. Stage 3 gates the +next release, because stage 2 shipped without a flag: its last part is the outage e2e passing in CI. 1. **Built. Extract window management without changing behavior.** Move the window bookkeeping (`openWindow`, its timers, attach and finalize) behind one type that holds a set of windows, with @@ -305,10 +305,10 @@ proposed. Stage 3 gates the next release, because stage 2 shipped without a flag 2. **Built. Allow several windows for proven-independent destinations.** Writes keep whole-object commits and apply behavior stays as it is. Writes of another `GitTarget` keep closing other windows (`identity_change`), and every layout in the footprint table's "no" rows stays serial. -3. **Proposed; gates the next release. Test the failure boundaries.** Both close orders, +3. **Built; gates the next release. Test the failure boundaries.** Both close orders, conflicting writes, delete and recreate, `attach_next`, zero-duration windows, resync, shutdown, failed publication and replay, and buffer pressure, each against the reference model, which - first has to keep work pending across failed pushes and retries. + keeps work pending across failed pushes and retries. See [What stage 3 built](#what-stage-3-built). 4. **Decided against (2026-10-28): no startup flag.** Stage 2 is always on, so stage 3's tests gate the release instead of a default-off rollout. What the flag's measurement period was for is read from the metrics: `git_commit_windows_open` shows windows open side by side, @@ -346,6 +346,33 @@ The rules above, with these choices where the page left room: window is open records a mismatch only when no window of its own author is open. - **The window ID** (`window`) is in the open, close and finalized log lines, with the branch. +### What stage 3 built + +The reference model keeps work across wakes. A decided write is committed at the end of the wake +that decided it, unless it needs a connection while a failed attempt waits for its retry (a resync +during an outage), and published by a push that the cooldown spaces after the last one that landed; +a save and an apply's flush push at once. A failed push keeps every write for the retry, 10s doubling +to 5m. The push cooldown runs on the worker's clock, as the window timers and the retry already did, +and the test driver delivers the push and retry timers too. + +- **Checked after every step:** the pending writes in order (which carry a save, which are a resync, + which are committed), what each is charged, the total retained charge, and all four timers. +- **Inputs added:** a Git outage that starts and ends (every push, fetch and sync fails), a restart + mid-run (a new worker on an empty checkout), a timer overdue while an item is handled first, an + applier's write that lands over a newer edit (withheld, and a Repair asked for), and a push from + outside that moves one author's document into another's file under their open windows. +- **Profiles:** wide, concurrent (authors keep to their own files) and outage (short clock steps, so + windows stay open across the cooldown and the retries). CI runs 20 seeds of 60 steps, about 15 s. +- **Reach floors:** the search must reach, across its seeds, each of shutdown, flush and push hint + with several windows open, `attach_next` beside another window, a reproof that finds windows + overlap, `path_conflict`, a failed push, the cooldown and a retry wait with several windows open, a + commit waiting for its retry, a replay with a window open, buffer pressure with several windows + open, an apply waiting for the retry, a restart, an overdue timer, and a withheld applier write. +- **Found:** no fault of the worker. The model learned four rules it never met while it pushed at + once: a pending resync holds every applier back (the barrier) and ends the flush; a resync's edits + are the newest writes of their objects; a save that changes nothing commits nothing, so it adds no + bootstrap files; a push that lands makes the base trusted again. + ## What it costs - **The loop's state.** The single `openWindow` becomes a set with an opening order, the close timer