diff --git a/tests/chaos/child.go b/tests/chaos/child.go index 64c621fce..7ad5514ec 100644 --- a/tests/chaos/child.go +++ b/tests/chaos/child.go @@ -48,6 +48,7 @@ const ( envPrune = "CONDUIT_CHAOS_PRUNE" envPaceMS = "CONDUIT_CHAOS_PACE_MS" envPersistDelayMS = "CONDUIT_CHAOS_PERSIST_DELAY_MS" + envHoldAt = "CONDUIT_CHAOS_HOLD_AT" envTotal = "CONDUIT_CHAOS_TOTAL" envSnapshotK = "CONDUIT_CHAOS_SNAPSHOT_K" envSnapshotPaceMS = "CONDUIT_CHAOS_SNAPSHOT_PACE_MS" @@ -62,6 +63,7 @@ const ( markerFatal = "FATAL" markerCorruptPo = "CORRUPT_POSITION" markerHandoff = "HANDOFF" // Property 1/2: producer crossed the snapshot->stream boundary, see upstream.go/produceLoop + markerHeld = "HELD" // production ceiling reached, see chaosPlugin.holdAt (upstream.go) markerAckOrder = "ACK_ORDER" // Property 3: per-key ack delivery ledger, see upstream.go/ackLoop // markerSigtermDone is the SIGTERM/invariant-7 case's completion marker // (sigterm_test.go, runChildSigterm) - deliberately distinct from @@ -129,7 +131,21 @@ type childEnv struct { // persistDelayMS overrides the persister debounce; 0 = default. See // childConfig.persistDelayMS in harness.go for why this is configurable. persistDelayMS int - total uint64 + + // holdAt caps how far this process's source may ever produce, making a + // mid-run SIGKILL's landing point bounded BY CONSTRUCTION instead of by + // the parent winning a race. Once position holdAt has been sent, the + // producer prints markerHeld and stops for good (it never reaches total), + // so no matter how long the parent is descheduled between deciding to + // kill and the signal actually landing, the durable committed watermark + // this process can leave behind is <= holdAt. See childConfig.holdAt in + // harness.go and chaosPlugin.holdAt in upstream.go for the full story. + // + // 0 (the default, and the value every scenario that does not need a + // bound leaves it at) disables the cap entirely - the producer behaves + // exactly as it did before this seam existed. + holdAt uint64 + total uint64 // snapshotK/snapshotPaceMS: Property 1/2's two-phase producer knobs. // snapshotK == 0 means "no distinct snapshot phase" (DBZ-1's original @@ -181,6 +197,13 @@ func parseChildEnv() childEnv { cfg.persistDelayMS = d } + holdAt, err := strconv.ParseUint(os.Getenv(envHoldAt), 10, 64) + if err != nil { + fmt.Fprintf(os.Stderr, "%s: invalid %s: %v\n", markerFatal, envHoldAt, err) + os.Exit(exitBadArgs) + } + cfg.holdAt = holdAt + total, err := strconv.ParseUint(os.Getenv(envTotal), 10, 64) if err != nil { fmt.Fprintf(os.Stderr, "%s: invalid %s: %v\n", markerFatal, envTotal, err) @@ -227,6 +250,19 @@ func parseChildEnv() childEnv { fmt.Fprintf(os.Stderr, "%s: %s and %s are required\n", markerFatal, envDBDir, envUpstreamDir) os.Exit(exitBadArgs) } + + // A cap at or above total bounds nothing - the producer stops at total on + // its own, so the cap would never be reached. The only legitimate use of + // a cap is strictly below total, where the child is crashable-only by + // design: it never reaches total, so the read loop never terminates and + // the process must be SIGKILLed (the same contract #2835's recovery + // child gives its holdAt seam). Any other combination could only surface + // as an opaque timeout or a silently pointless knob - fail loudly and + // immediately instead, like every other misconfiguration here. + if cfg.holdAt > 0 && cfg.total > 0 && cfg.holdAt >= cfg.total { + fmt.Fprintf(os.Stderr, "%s: %s (%d) must be below %s (%d) to bound anything\n", markerFatal, envHoldAt, cfg.holdAt, envTotal, cfg.total) + os.Exit(exitBadArgs) + } return cfg } @@ -375,6 +411,7 @@ func buildChild(ctx context.Context, cfg childEnv) (*childBuilt, error) { snapshotPaceMS: cfg.snapshotPaceMS, numKeys: cfg.numKeys, driftAt: cfg.driftAt, + holdAt: cfg.holdAt, } fetcher := staticFetcher{instance.Plugin: staticDispenser{source: plugin}} diff --git a/tests/chaos/harness.go b/tests/chaos/harness.go index 22ffb9d1d..baa2e5262 100644 --- a/tests/chaos/harness.go +++ b/tests/chaos/harness.go @@ -56,6 +56,26 @@ type childConfig struct { // unchanged and just as strict. See sigkillCases. persistDelayMS int + // holdAt caps how far a child's producer may ever go, making a mid-run + // SIGKILL's landing point bounded BY CONSTRUCTION instead of by the + // parent winning a race (the #2835 pattern, applied to this harness's + // chaosPlugin - see chaosPlugin.holdAt in upstream.go). Once position + // holdAt has been sent, the producer prints "HELD " (markerHeld) + // and stops for good: nothing past it is ever produced, so nothing past + // it can ever be acked or committed, however long the parent is + // descheduled between deciding to kill and the signal landing. A capped + // child also never reaches total, so its read loop blocks forever and it + // stays alive until the parent SIGKILLs it - the kill can never miss an + // exited process. + // + // 0 (the default, and the value every scenario that does not need a + // bound leaves it at) disables the cap entirely. + // + // The cap is a property of the FIRST (killed) child only: a RESUMED + // child must run to total unencumbered, so scenarios that set this also + // spawn their second child with it zeroed. + holdAt uint64 + // snapshotK/snapshotPaceMS: DBZ-2 Property 1/2's two-phase producer // knobs (see chaosPlugin's type doc, upstream.go). Zero values preserve // DBZ-1's original single-phase behavior. @@ -98,6 +118,7 @@ func (c childConfig) env() []string { envDriftAt + "=" + strconv.FormatUint(c.driftAt, 10), envSigtermMode + "=" + strconv.FormatBool(c.sigtermMode), envPersistDelayMS + "=" + strconv.Itoa(c.persistDelayMS), + envHoldAt + "=" + strconv.FormatUint(c.holdAt, 10), } } diff --git a/tests/chaos/property2_test.go b/tests/chaos/property2_test.go index bf4849cc4..3617e28fd 100644 --- a/tests/chaos/property2_test.go +++ b/tests/chaos/property2_test.go @@ -33,9 +33,11 @@ const ( // resumeEmpty: RESUME_POSITION must be the empty/nil position - no // state was ever durably persisted before the kill. resumeEmpty resumeShape = iota - // resumeValidNonZero: RESUME_POSITION must be a valid, non-zero, stale - // (behind the kill point) position - at least one flush landed before - // the kill. + // resumeValidNonZero: RESUME_POSITION must be a valid, non-zero + // position - at least one flush landed before the kill. Not "stale": + // the watermark-gated kill lands exactly on a flush boundary, so the + // checkpoint equals the kill point rather than lagging it (see the + // mid-position-write case comment). resumeValidNonZero ) @@ -58,12 +60,35 @@ type property2Case struct { paceMS int total uint64 - // Kill-timing gate: exactly one of these is set per case, both - // READ-progress-gated (never ACK-gated, never a fixed sleep - see - // waitForReadCount's and waitForMarker's doc comments for why that - // matters under Approach A's deferred acks). - killAfterReads int // mid-snapshot, mid-position-write: gate on N observed READ lines - killOnHandoff bool // mid-handoff: gate on the HANDOFF marker itself + // persistDelayMS overrides the persister debounce for the FIRST (killed) + // child only; 0 = default. The two resumeEmpty cases set it far beyond + // any plausible scheduling delay so NO flush can ever land before the + // kill - the "nothing durably persisted at kill time" precondition + // becomes structural instead of a race against the ~1s automatic flush + // (the #2534 pattern; see childConfig.persistDelayMS). The RESUMED child + // always runs with the real default. + persistDelayMS int + + // holdAt caps the FIRST (killed) child's production (the #2835 pattern; + // see childConfig.holdAt and chaosPlugin.holdAt). Two structural roles: + // - the child can never produce - and therefore never ack or commit - + // anything past holdAt, bounding the kill's landing point above no + // matter how long the parent is descheduled; and + // - the child never reaches total, so its read loop blocks and it + // stays alive until SIGKILLed - the kill can never miss an exited + // process (an uncapped child that finishes its run and exits on its + // own would fail the kill instead of the test). + // 0 = no cap. The RESUMED child always runs with it zeroed. + holdAt uint64 + + // Kill-timing gate: exactly one of these is set per case. All three are + // LOWER-bound observations - the kill can never fire before the gate + // returns - while the UPPER bound is structural in every case, which is + // what makes the kill's exact landing point irrelevant to the verdict + // (see assertProperty2Case's doc). + killAfterReads int // mid-snapshot: gate on N observed READ lines + killOnHandoff bool // mid-handoff: gate on the HANDOFF marker itself + killAfterCommitted uint64 // mid-position-write: gate on the durable upstream commit watermark (waitForUpstreamCommittedAtLeast) // The two-state resume discriminator this case's kill timing is chosen // to land in - see resumeShape's doc. @@ -74,50 +99,107 @@ var property2Cases = []property2Case{ { // Mid-snapshot: identical timing to DBZ-1's own "mid-snapshot" case // (sigkill_test.go) - an initial, fast, unpaced-ish burst (1ms/read). - // killAfterReads=30 is ~30ms in, far short of the persister's FIRST - // automatic flush (~1000ms), so Conduit has durably persisted - // NOTHING at all when the kill lands. This is the highest-stakes - // edge case named in the design doc: a crash before the snapshot - // watermark is durably recorded at all - the engine-side reflection - // of the conduit-connector-mysql #182 bug class. + // The precondition is that Conduit has durably persisted NOTHING at + // all when the kill lands - the highest-stakes edge case named in + // the design doc: a crash before the snapshot watermark is durably + // recorded at all - the engine-side reflection of the + // conduit-connector-mysql #182 bug class. + // + // That used to be inferred from arithmetic ("30 reads x 1ms = ~30ms, + // far short of the persister's FIRST automatic flush (~1000ms)"). On + // a loaded CI box the parent can be descheduled for longer than that + // window: the flush fires, a position IS persisted, and the case + // fails (or, worse, the child finishes its whole run and the resumed + // child starves - #2836's 41.6s hang). Both preconditions are now + // STRUCTURAL: persistDelayMS (600s) means no flush can ever fire + // before the kill, and holdAt caps production below total so the + // child can never finish its run and exit before the kill lands. + // The kill itself is still gated on observed READ progress as the + // LOWER bound - the child has genuinely produced - but the verdict + // no longer depends on how long the parent stalls around it. name: "mid-snapshot", paceMS: 1, killAfterReads: 30, total: 500, + persistDelayMS: 600_000, + holdAt: 80, expectResume: resumeEmpty, }, { // Mid-handoff (producer-pacing variant, NOT a distinct engine // state - see the design doc's Property 2 section and resumeShape's // doc above). snapshotK=40 at 1ms/read means the HANDOFF marker - // fires at ~40ms elapsed - still far short of the first ~1000ms - // flush, so this lands in the SAME "empty" persisted state as - // mid-snapshot above, not a fictional third "boundary" shape. - // Killing on the marker itself (waitForMarker), rather than a read - // count chosen to merely be close to it, is what makes this - // genuinely "just after HANDOFF" rather than a guess. + // fires at ~40ms elapsed - killing on the marker itself + // (waitForMarker), rather than a read count chosen to merely be + // close to it, is what makes this genuinely "just after HANDOFF" + // rather than a guess. The precondition is the SAME "empty" + // persisted state as mid-snapshot above - still far short of the + // first ~1000ms flush, and not a fictional third "boundary" shape - + // and it is structural the same way: a 600s persister debounce + // means no flush can ever land before the kill, and holdAt caps + // production so the child stays alive until the kill lands. name: "mid-handoff", snapshotK: 40, snapshotPaceMS: 1, - paceMS: 15, // stream-phase pace; never reached before the kill + paceMS: 15, // stream-phase pace; only ever reached up to holdAt total: 300, killOnHandoff: true, + persistDelayMS: 600_000, + holdAt: 80, expectResume: resumeEmpty, }, { // Mid-position-write: identical timing to DBZ-1's own "mid-stream" - // case (sigkill_test.go) - steady-state 15ms/read pacing. - // killAfterReads=95 is ~1.4s in: by then one automatic flush has - // already happened (~1s, around read ~66) and a second debounce - // window has already started (on the next ack after that flush) - // but not yet fired (its own 1s timer would land around read - // ~133). So Conduit's persisted position is a valid but STALE - // checkpoint - the other edge case named in the design doc. - name: "mid-position-write", - paceMS: 15, - killAfterReads: 95, - total: 400, - expectResume: resumeValidNonZero, + // case (sigkill_test.go) - steady-state 15ms/read pacing. The + // precondition is a valid, non-zero checkpoint landed exactly at a + // flush boundary: the kill is gated on the child's DURABLE upstream + // commit watermark (killAfterCommitted=70, + // waitForUpstreamCommittedAtLeast), and the watermark only moves + // when a persister flush lands, so the kill can only ever land on a + // flushed state. At the kill instant produced == acked == committed + // (exactly the ceiling, 100, on an unloaded machine; somewhere in + // [70, 100] under load): the checkpoint equals the watermark equals + // the kill point, and the in-flight window is empty by construction + // - its size no longer depends on how long the parent was + // descheduled, only on the ack plumbing at the flush instant. That + // is the no-gap equality the shared assertions check: the case pins + // the "resume at a flush boundary with a full checkpoint" shape. + // + // Losing the mid-flush in-flight window (records produced but not + // yet durable at the kill) is the DELIBERATE cost of bounding the + // kill on the watermark - it is what makes the landing point + // structural instead of a race with the parent's scheduler. The + // crash-with-records-in-flight scenario's intended home is DBZ-1's + // own mid-stream SIGKILL case (sigkill_test.go), which still kills + // a free-running child. + // + // holdAt=100 is the upper bound: nothing past it is ever produced, + // so nothing past it can ever be acked or committed, and the resume + // position can never catch up past the ceiling however long the + // parent is descheduled - the old "resumePos < killAfterReads" + // guard raced exactly this (a stalled parent let the child blow + // through it, #2836). + // + // 70/100 is a bracket, not a tuning: the gate cannot fire before + // the watermark reaches 70 (its own guarantee, restated as a + // precondition below), and the ceiling means the watermark can + // never exceed 100 - so the kill always lands with the watermark in + // [70, 100], every value of which yields the identical verdict. On + // an unloaded machine the gate fires at the second flush (at 15ms + // pace a flush covers ~66 records, so 66 < 70 <= ~132), but under + // the very descheduling this PR targets, production slows and the + // gate can fire at a later flush with coverage as low as just over + // 70 - so the case-level resumePos <= holdAt assertion is only a + // SOFT cap-removal detector. The DETERMINISTIC detector is + // TestChild_HoldAt_CapsProductionBelowTotal: without the ceiling + // the HELD marker never prints, and that test fails outright on any + // machine. + name: "mid-position-write", + paceMS: 15, + killAfterCommitted: 70, + holdAt: 100, + total: 400, + expectResume: resumeValidNonZero, }, } @@ -159,6 +241,42 @@ func TestSIGKILL_Property2_DurableUpstream(t *testing.T) { // in, so a case silently landing in the wrong window fails loudly // instead of passing having tested nothing (see the design doc's // "timing flakiness" failure mode). +// +// # Why the kill point is bounded on both sides, and neither bound is a race +// +// Every case's kill timing used to be a parent-side observation of a +// FREE-RUNNING child: waitForReadCount/waitForMarker watched stdout, and the +// child kept producing while the parent was descheduled, so "I last saw N +// READs" said nothing about where the child was when SIGKILL actually +// landed. With an injected stall, mid-handoff and mid-position-write failed +// and mid-snapshot hung for ~41.6s (issue #2836) - the starvation +// sigkillCase's own doc comment describes. Each case now brackets its kill +// point with one observed LOWER bound and one STRUCTURAL upper bound, and +// every value inside the bracket yields the identical verdict: +// +// - mid-snapshot, mid-handoff (resumeEmpty): the lower bound is the READ / +// HANDOFF observation (the child has genuinely produced, and the +// mid-handoff kill is genuinely post-boundary); the upper bound is +// structural TWICE - a 600s persister debounce (persistDelayMS) means no +// flush can ever land before the kill, so nothing is ever durably +// persisted however long the parent stalls, and holdAt caps production so +// the child can never finish its run and exit before the kill lands. +// +// - mid-position-write (resumeValidNonZero): the kill is gated on the +// child's DURABLE upstream commit watermark reaching killAfterCommitted, +// read off the same on-disk marker the assertions read back after the +// kill (waitForUpstreamCommittedAtLeast). The watermark only moves when +// a persister flush lands, so a watermark at or past killAfterCommitted +// is positive proof at least one flush has fired - the resume position +// is provably non-zero. The upper bound is the holdAt ceiling: nothing +// past it is ever produced, so nothing past it can ever be acked or +// committed, and the resume position can never catch up past it however +// long the parent is descheduled. +// +// The tunables (killAfterReads, killAfterCommitted, holdAt) therefore affect +// where inside the bracket the kill lands - how much production happened +// before it, never whether the test passes - which is exactly the difference +// between bounding a test and tuning one. func assertProperty2Case(t *testing.T, tc property2Case, prune bool) { t.Helper() is := is.New(t) @@ -173,10 +291,25 @@ func assertProperty2Case(t *testing.T, tc property2Case, prune bool) { total: tc.total, } - first := spawnChild(t, cfg) - if tc.killOnHandoff { + // The structural knobs apply ONLY to the first (killed) child. The + // RESUMED child must run with the real default debounce and no cap, or + // it would never flush or run to completion either - which is not the + // scenario under test, and would make the resume assertions vacuous. + firstCfg := cfg + firstCfg.persistDelayMS = tc.persistDelayMS + firstCfg.holdAt = tc.holdAt + + first := spawnChild(t, firstCfg) + switch { + case tc.killOnHandoff: first.waitForMarker(t, markerHandoff, 30*time.Second) - } else { + case tc.killAfterCommitted > 0: + // Gate the kill on durable, flushed progress rather than a lagging + // parent-side read count. This wait can never overshoot: firstCfg.holdAt + // caps what this child can ever produce or commit, so the watermark is + // provably in [killAfterCommitted, holdAt] when the kill lands. + waitForUpstreamCommittedAtLeast(t, first, cfg.upstreamDir, tc.killAfterCommitted, 30*time.Second) + default: first.waitForReadCount(t, tc.killAfterReads, 30*time.Second) } first.sigkill(t) @@ -186,6 +319,43 @@ func assertProperty2Case(t *testing.T, tc property2Case, prune bool) { watermarkAtKill, err := committedAtKill.Committed() is.NoErr(err) + // Make each case's PRECONDITION explicit and self-diagnosing (the + // sigkill_test.go precedent, #2534): the structural bounds above should + // make these unreachable, but assert them rather than trusting them, so + // a future violation identifies itself instead of surfacing as a + // confusing downstream mismatch. A run that lands outside its case's + // precondition tests nothing - failing loudly here is the alternative + // to passing vacuously. + switch tc.expectResume { + case resumeEmpty: + if watermarkAtKill != 0 { + t.Fatalf( + "precondition violated: %s (prune=%v) requires NO persisted position at kill time, "+ + "but the upstream watermark was already %d. A persister flush beat the kill "+ + "despite a %dms debounce, so this run did not test the crash-before-first-"+ + "checkpoint scenario at all.\n%s", + tc.name, prune, watermarkAtKill, tc.persistDelayMS, first.diagnostics(), + ) + } + case resumeValidNonZero: + if watermarkAtKill < tc.killAfterCommitted { + t.Fatalf( + "precondition violated: %s (prune=%v) requires the kill to land after at least one "+ + "durable flush (watermark >= %d), but the watermark at kill time was %d - the "+ + "kill gate did not do what it claims.\n%s", + tc.name, prune, tc.killAfterCommitted, watermarkAtKill, first.diagnostics(), + ) + } + if watermarkAtKill > tc.holdAt { + t.Fatalf( + "precondition violated: %s (prune=%v): the upstream watermark at kill time (%d) "+ + "exceeded the production ceiling holdAt (%d) - the child produced past its cap, "+ + "so the upper bound this case rests on is not in effect.\n%s", + tc.name, prune, watermarkAtKill, tc.holdAt, first.diagnostics(), + ) + } + } + second := spawnChild(t, cfg) second.waitExit(t, parentWaitExit) @@ -226,10 +396,10 @@ func assertProperty2Case(t *testing.T, tc property2Case, prune bool) { if resumePos != 0 { t.Fatalf( "Property 2 (%s, prune=%v): expected RESUME_POSITION to be empty/fresh "+ - "(kill landed before the first debounce flush), got %d - either the kill "+ - "timing no longer lands where this case's comment claims, or the persister's "+ - "debounce threshold changed underneath this test\n%s", - tc.name, prune, resumePos, second.diagnostics(), + "(no flush could have landed before the kill: %dms debounce, cap %d), got %d - "+ + "either the persister's debounce threshold changed underneath this test, or "+ + "the structural precondition no longer holds\n%s", + tc.name, prune, tc.persistDelayMS, tc.holdAt, resumePos, second.diagnostics(), ) } case resumeValidNonZero: @@ -242,13 +412,92 @@ func assertProperty2Case(t *testing.T, tc property2Case, prune bool) { tc.name, prune, second.diagnostics(), ) } - if resumePos >= uint64(tc.killAfterReads) { + // The structural form of the old stale guard. The old check - + // resumePos strictly behind killAfterReads - raced a free-running + // child: a stalled parent let the child blow through it (#2836). + // Nothing past the ceiling is ever produced, so nothing past it can + // ever be persisted: resumePos <= holdAt holds by construction. It + // is a SOFT cap-removal detector, though - under load the gate can + // fire with uncapped coverage as low as ~70 (see the case comment) + // - so the deterministic detector of a missing ceiling is the + // regression test's HELD-marker wait, not this assertion. + if resumePos > tc.holdAt { t.Fatalf( - "Property 2 (%s, prune=%v): expected RESUME_POSITION (%d) to be STALE - strictly "+ - "behind the kill point (read #%d) - a valid checkpoint that already caught up to "+ - "the kill point would mean this case no longer exercises a mid-flush window\n%s", - tc.name, prune, resumePos, tc.killAfterReads, second.diagnostics(), + "Property 2 (%s, prune=%v): expected RESUME_POSITION (%d) to be at or behind the "+ + "production ceiling (holdAt %d) - a checkpoint that caught up past the ceiling "+ + "means the upper bound this case rests on is not in effect\n%s", + tc.name, prune, resumePos, tc.holdAt, second.diagnostics(), ) } } } + +// property2HoldStall is how long TestChild_HoldAt_CapsProductionBelowTotal +// deliberately does nothing after the producer reports it has hit its +// ceiling. +// +// This is not a "wait long enough and hope" sleep - it is the opposite, and +// the distinction matters because this package forbids the former. An +// unsound kill gate is one whose precondition decays as the parent is +// descheduled for longer; this sleep IS that descheduling, injected on +// purpose, and the assertions after it must hold no matter how large it +// gets - making it larger can only make an unsound cap fail harder. It is +// sized at 400ms because that is empirically enough for the uncapped +// producer this test guards against (paceMS 1) to run far past the 80 +// ceiling the cap pins - i.e. enough for #2836's original failure mode to +// manifest. The sleep is adversarial, not load-bearing: an uncapped child +// fails the HELD-marker wait outright, and a child capped too high fails +// the position assertions. +const property2HoldStall = 400 * time.Millisecond + +// TestChild_HoldAt_CapsProductionBelowTotal is #2836's regression test: it +// pins the production ceiling (chaosPlugin.holdAt) that +// TestSIGKILL_Property2_*'s resume-shape guards now rest on. +// +// Before the ceiling existed, those tests raced their own children: they +// observed read progress and then SIGKILLed, and the child kept producing +// throughout the window in between - a stalled parent could let the child +// blow through every kill-point guard (#2836). This test reproduces that +// window directly and adversarially: it waits for the producer to report +// its ceiling, then stalls for property2HoldStall (long enough that a +// ceiling-less child would have run far past the ceiling - exactly how the +// original flake was reproduced) and asserts the child has not moved. +// +// Run against a child without the cap, the marker wait below times out +// (the producer free-runs past the ceiling, printing READ lines instead), +// and the position assertions fail outright. +func TestChild_HoldAt_CapsProductionBelowTotal(t *testing.T) { + is := is.New(t) + dir := t.TempDir() + cfg := childConfig{ + dbDir: dir + "/db", + upstreamDir: dir + "/upstream", + paceMS: 1, + total: 500, + holdAt: 80, + } + // The ceiling only bounds anything if it is genuinely below total; the + // child enforces this too (parseChildEnv), but state it here so the + // scenario's own numbers can't drift into vacuity unnoticed. + is.True(cfg.holdAt < cfg.total) + + child := spawnChild(t, cfg) + child.waitForMarker(t, markerHeld+" ", 10*time.Second) + is.Equal(maxProgressPosition(child, markerHeld), cfg.holdAt) // the marker reports the ceiling itself + + time.Sleep(property2HoldStall) // see property2HoldStall: adversarial, not load-bearing + + // Nothing beyond the ceiling was ever produced, however long we looked + // away... + is.True(maxProgressPosition(child, "READ") <= cfg.holdAt) + + // No durable-side check here: at ~480ms elapsed the default 1s + // persister debounce has not fired, so the committed watermark is 0 by + // construction and "committed <= holdAt" would assert nothing. The + // committed side of the ceiling is structurally implied (nothing past + // it is ever produced, so nothing past it can ever be committed) and is + // asserted where a flush has provably landed: the mid-position-write + // case's precondition and resume-shape checks. + + child.sigkill(t) // crashable variant: it would otherwise block forever +} diff --git a/tests/chaos/upstream.go b/tests/chaos/upstream.go index 40ed23455..b29628efe 100644 --- a/tests/chaos/upstream.go +++ b/tests/chaos/upstream.go @@ -191,6 +191,26 @@ type chaosPlugin struct { // every existing scenario. driftAt uint64 + // holdAt caps how far this process's producer may ever go, making a + // mid-run SIGKILL's landing point bounded BY CONSTRUCTION instead of by + // the parent winning a race (the #2835 pattern, applied to this harness). + // Once position holdAt has been sent, produceLoop prints markerHeld and + // returns, so this process can never produce - and therefore never ack or + // commit - anything past holdAt, no matter how long the parent is + // descheduled between deciding to kill and the signal actually landing. + // + // Because the read loop keeps blocking on a stream that stays open, a + // capped child also stays alive indefinitely instead of running to total + // and exiting on its own - which is what makes "SIGKILL at any time" + // reliable (an uncapped child that finishes its run before the parent + // kills it would fail the kill, not the test). + // + // 0 (the default, every scenario that does not need a bound) disables the + // cap entirely - the producer behaves exactly as it did before this seam + // existed. See childEnv.holdAt (child.go) for the env plumbing and its + // validation. + holdAt uint64 + // sourceTag is the N-source shared-destination collision scenario's // record-level salt (docs/design-documents/20260801-archv2-multiconnector- // nsource.md's H2 section; nsource_child.go, nsource_sigkill_test.go). @@ -327,6 +347,13 @@ func (p *chaosPlugin) Run(ctx context.Context, stream pconnector.SourceRunStream // that problem: production is paced deterministically, so harness.go's // kill-timing waits on READ (and, for mid-handoff, HANDOFF) lines, never ACK // lines. +// +// If p.holdAt is nonzero, production stops for good once position holdAt has +// been sent: the loop prints markerHeld and returns, exactly as it does on +// reaching p.total, leaving the stream open and the ack path running but no +// further records ever produced by this process. That is a hard ceiling on +// how far this process can get, which is what lets a parent test SIGKILL it +// mid-run without racing it — see chaosPlugin.holdAt's field doc. func (p *chaosPlugin) produceLoop(server pconnector.SourceRunStreamServer, start uint64) { if p.startGate != nil { // H2 injection scenario only (see the field doc) - block here, @@ -365,6 +392,14 @@ func (p *chaosPlugin) produceLoop(server pconnector.SourceRunStreamServer, start fmt.Printf("%s %d\n", markerHandoff, pos) } + // The production ceiling. Checked after the send (so holdAt is the + // last position actually produced, inclusive) and before the pace + // sleep, so the marker is printed the instant the ceiling is hit. + if p.holdAt > 0 && pos >= p.holdAt { + printProgress(markerHeld, pos) + return + } + pace := p.paceMS if p.snapshotK > 0 && pos < p.snapshotK { pace = p.snapshotPaceMS