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 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/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") +} 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) + } + } } 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 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()) + }) +})