From cfea9de53086d370dd45dc27fee68847b2c48f9a Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 01:24:57 +0200 Subject: [PATCH] Fix the try on bb594652: a const batch gap unit tests assert directly, wait for the bundle's own subscribe before publishing, and a migration-0001 database built through the real path (issues 1520, 1518, 1431, 1517) --- .../cmd/mesh-mail-reader/batch.go | 35 ++++++++----- .../cmd/mesh-mail-reader/batch_gap_test.go | 17 +++++++ .../cmd/mesh-mail-reader/batch_test.go | 29 ++++++----- .../cmd/mesh-mail-reader/hear_test.go | 40 +++++++++++++++ .../cmd/mesh-mail-reader/main.go | 14 +++-- .../pgstore_integration_test.go | 40 +++++++++++++-- .../runtime_player_integration_test.go | 51 ++++++++++++++++++- 7 files changed, 190 insertions(+), 36 deletions(-) create mode 100644 modules/mesh-mail-reader/cmd/mesh-mail-reader/batch_gap_test.go diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go index 6f6ecca3..8e9a3b74 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go @@ -16,13 +16,20 @@ import ( "time" ) -// BatchGap is how long must pass between the start of one batch and the next (ADR 0459 §15, 1520/AC1): a -// var, never a setting, so an integration test can shorten it in its own build alone (issue 1431's own -// fix: batch_gap_integration.go, `//go:build integration`, never part of a production build) without -// production pacing ever becoming configurable. Every normal build (including every unit test) keeps this -// at 160s; TestNextBatchWait and TestThirtyWaitingMailsRunAsThreeBatchesAtLeast160sApart still prove that -// value, on a test clock, in batch_test.go. -var BatchGap = 160 * time.Second +// MinBatchGap is the pace no production path may ever lower (ADR 0459 §15, 1520/AC1): at least 160s +// between the start of one batch and the next. It is a const, never read by anything an integration +// build could shorten; the unit tests in batch_test.go assert against this literal, not against the var +// below, so they keep proving 160s even when compiled into a binary whose BatchGap has been shortened +// (issue 1520, fixed per the lead's try feedback on bb594652). +const MinBatchGap = 160 * time.Second + +// BatchGap is the gap the running binary actually paces its batches by: a var, never a setting, so an +// integration test can shorten it for its own build alone (issue 1431's own fix: batch_gap_integration.go, +// `//go:build integration`, never part of a production build) without production pacing ever becoming +// configurable. Every normal (non-integration) build starts it at MinBatchGap and never changes it; +// TestRunningBatchGapIsMinBatchGap (batch_gap_test.go, `//go:build !integration`) proves that for the +// binary actually shipped, not just for a literal. +var BatchGap = MinBatchGap // JudgeWithin is the bound every judgement asks: never above 40 seconds (ADR 0459 §14, 1520/AC2). const JudgeWithin = 40 * time.Second @@ -100,17 +107,19 @@ func NewJudge(ask Ask, node string, now func() time.Time) JudgeFunc { } } -// nextBatchWait is how long to wait before the next batch may start (1520/AC1): zero once BatchGap has -// passed since the last start (or no batch has started yet), else the remainder. -func nextBatchWait(lastStart, now time.Time) time.Duration { +// nextBatchWait is how long to wait before the next batch may start (1520/AC1): zero once gap has +// passed since the last start (or no batch has started yet), else the remainder. gap is given explicitly +// by the caller (RunBatches passes BatchGap; the unit tests in batch_test.go pass MinBatchGap directly) +// so this pure function never itself reaches into a var an integration build might have shortened. +func nextBatchWait(lastStart, now time.Time, gap time.Duration) time.Duration { if lastStart.IsZero() { return 0 } elapsed := now.Sub(lastStart) - if elapsed >= BatchGap { + if elapsed >= gap { return 0 } - return BatchGap - elapsed + return gap - elapsed } func senderOf(from []Person) (name, address string) { @@ -235,7 +244,7 @@ func RunBatches(ctx context.Context, store Store, readText ReadTextFunc, judge J if ctx.Err() != nil { return } - if wait := nextBatchWait(lastStart, now()); wait > 0 { + if wait := nextBatchWait(lastStart, now(), BatchGap); wait > 0 { sleep(wait) continue } diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch_gap_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch_gap_test.go new file mode 100644 index 00000000..f6ed8a4e --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch_gap_test.go @@ -0,0 +1,17 @@ +//go:build !integration + +package main + +import "testing" + +// TestRunningBatchGapIsMinBatchGap (1520/AC1, ADR 0459 §15, issue 1520 — fixed per the lead's try feedback +// on bb594652): the gap the running binary actually paces its batches by (BatchGap, the var RunBatches +// reads) is MinBatchGap, exactly, in any build this test is itself part of. This file's own `!integration` +// tag keeps it out of a `-tags integration` build entirely — batch_gap_integration.go's init, which +// shortens BatchGap for that build alone, is never even compiled alongside this assertion — so no +// production path is ever caught lowering the pace. +func TestRunningBatchGapIsMinBatchGap(t *testing.T) { + if BatchGap != MinBatchGap { + t.Fatalf("BatchGap = %s, want MinBatchGap (%s): no production path may lower the pace", BatchGap, MinBatchGap) + } +} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch_test.go index 74b2ce32..06882585 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch_test.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch_test.go @@ -39,26 +39,29 @@ func judgeAllNeither(waiting []WaitingMail) JudgeFunc { } } -// 1520/AC1 (the pure pacing math): zero once BatchGap has passed since the last start, or no batch has -// started yet; the remainder otherwise. +// 1520/AC1 (the pure pacing math): zero once MinBatchGap has passed since the last start, or no batch has +// started yet; the remainder otherwise. Passes MinBatchGap explicitly, never the var BatchGap, so this +// still proves 160s even when compiled into a binary whose BatchGap has been shortened (batch_gap_integration.go). func TestNextBatchWait(t *testing.T) { start := time.Date(2026, 10, 20, 10, 0, 0, 0, time.UTC) - if w := nextBatchWait(time.Time{}, start); w != 0 { + if w := nextBatchWait(time.Time{}, start, MinBatchGap); w != 0 { t.Fatalf("no batch yet: wait = %s, want 0", w) } - if w := nextBatchWait(start, start.Add(100*time.Second)); w != 60*time.Second { + if w := nextBatchWait(start, start.Add(100*time.Second), MinBatchGap); w != 60*time.Second { t.Fatalf("100s after: wait = %s, want 60s", w) } - if w := nextBatchWait(start, start.Add(BatchGap)); w != 0 { - t.Fatalf("exactly BatchGap after: wait = %s, want 0 (one waiting mail starts a batch at once)", w) + if w := nextBatchWait(start, start.Add(MinBatchGap), MinBatchGap); w != 0 { + t.Fatalf("exactly MinBatchGap after: wait = %s, want 0 (one waiting mail starts a batch at once)", w) } - if w := nextBatchWait(start, start.Add(BatchGap+time.Second)); w != 0 { - t.Fatalf("past BatchGap: wait = %s, want 0", w) + if w := nextBatchWait(start, start.Add(MinBatchGap+time.Second), MinBatchGap); w != 0 { + t.Fatalf("past MinBatchGap: wait = %s, want 0", w) } } -// 1520/AC1: with 30 mails waiting, three batches of 10 start at least 160s apart (a test clock: the batch -// loop here is driven by hand, jumping straight to each batch's own allowed start, as nextBatchWait says to). +// 1520/AC1: with 30 mails waiting, three batches of 10 start at least MinBatchGap (160s) apart (a test +// clock: the batch loop here is driven by hand, jumping straight to each batch's own allowed start, as +// nextBatchWait says to, passing MinBatchGap explicitly as production's RunBatches would with an +// unshortened BatchGap). func TestThirtyWaitingMailsRunAsThreeBatchesAtLeast160sApart(t *testing.T) { ctx := context.Background() store := newMemStore() @@ -76,7 +79,7 @@ func TestThirtyWaitingMailsRunAsThreeBatchesAtLeast160sApart(t *testing.T) { if len(all) == 0 { break } - if wait := nextBatchWait(lastStart, now); wait > 0 { + if wait := nextBatchWait(lastStart, now, MinBatchGap); wait > 0 { now = now.Add(wait) } batch := all @@ -96,8 +99,8 @@ func TestThirtyWaitingMailsRunAsThreeBatchesAtLeast160sApart(t *testing.T) { t.Fatalf("ran %d batches, want 3 (30 mails / 10 per batch)", batches) } for i := 1; i < len(starts); i++ { - if gap := starts[i].Sub(starts[i-1]); gap < BatchGap { - t.Fatalf("batch %d started %s after the previous, want at least %s", i, gap, BatchGap) + if gap := starts[i].Sub(starts[i-1]); gap < MinBatchGap { + t.Fatalf("batch %d started %s after the previous, want at least %s", i, gap, MinBatchGap) } } if left, _ := store.Waiting(ctx); len(left) != 0 { diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/hear_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/hear_test.go index 806b8e72..0634a553 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/hear_test.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/hear_test.go @@ -3,8 +3,11 @@ package main import ( "context" "encoding/json" + "errors" "testing" "time" + + stdio "git.novox.be/novox/mesh-sdk/go" ) func addedEventBody(t *testing.T, account, folder, messageID string, uid uint32) json.RawMessage { @@ -111,3 +114,40 @@ func TestHearDropsAMalformedEvent(t *testing.T) { t.Fatalf("waiting = %v, want none", waiting) } } + +// TestHearLoopRetriesStartingAt250msDoublingToAMinute (issue 1518, fixed per the lead's try feedback on +// bb594652): a subscribe the runtime refuses is retried starting at hearFirstRetryWait (250ms), doubling +// each time, capped at a minute — never mesh-mail-notifier's own 2s start, so a bundle hears within about a +// second of stdio.ServeOn beginning to serve. storeFor is never called: every refusal is the subscribe's own. +func TestHearLoopRetriesStartingAt250msDoublingToAMinute(t *testing.T) { + var waits []time.Duration + recordSleep := func(d time.Duration) { waits = append(waits, d) } + + calls := 0 + subscribe := func(string, func(stdio.Envelope) error) error { + calls++ + if calls <= 4 { + return errors.New("the runtime refused this subscribe") + } + return nil + } + storeFor := func(context.Context) (Store, error) { + t.Fatal("hearLoop must not call storeFor merely to retry a subscribe refusal") + return nil, nil + } + + hearLoop(storeFor, subscribe, recordSleep) + + want := []time.Duration{250 * time.Millisecond, 500 * time.Millisecond, time.Second, 2 * time.Second} + if len(waits) != len(want) { + t.Fatalf("waited %v, want %v", waits, want) + } + for i, w := range want { + if waits[i] != w { + t.Fatalf("wait[%d] = %s, want %s", i, waits[i], w) + } + } + if calls != 5 { + t.Fatalf("subscribe was called %d time(s), want 5 (four refusals, then the one that succeeded)", calls) + } +} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go index 038cb470..c92c0b16 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go @@ -72,13 +72,21 @@ func batchLoop(storeFor func(context.Context) (Store, error)) { // ---- hearing -------------------------------------------------------------------------------------------------- -// hearLoop subscribes to mesh-mail's added events and hears each one (hear.go), exactly as -// mesh-mail-notifier's watchWith does (issue 1518): a subscribe the runtime refuses is retried, 2 seconds +// hearFirstRetryWait is the hearing loop's first retry wait (issue 1518, fixed per the lead's try feedback +// on bb594652): 250ms, doubling to at most a minute, so a bundle hears within about a second of stdio.ServeOn +// starting to serve rather than mesh-mail-notifier's own 2s start — a window during which a plain NATS +// subject (no durable consumer) loses an event published before the subscribe lands; the mesh's own +// controller-made durable consumer keeps it regardless, so this only shortens an unnecessary wait, it does +// not paper over a production loss. +const hearFirstRetryWait = 250 * time.Millisecond + +// hearLoop subscribes to mesh-mail's added events and hears each one (hear.go), mirroring mesh-mail-notifier's +// watchWith (issue 1518): a subscribe the runtime refuses is retried, starting at hearFirstRetryWait and // doubling to at most a minute, logging each refusal, never fatal — a refusal here must never take the six // mail-reading verbs or mail_reader_health down with it. Started before stdio.ServeOn, so it keeps running // whatever stdio.ServeOn does with the stream it is given. func hearLoop(storeFor func(context.Context) (Store, error), subscribe func(string, func(stdio.Envelope) error) error, sleep func(time.Duration)) { - for wait := 2 * time.Second; ; wait = minDuration(wait*2, time.Minute) { + for wait := hearFirstRetryWait; ; wait = minDuration(wait*2, time.Minute) { err := subscribe(AddedPattern, func(e stdio.Envelope) error { ctx := context.Background() store, err := storeFor(ctx) diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore_integration_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore_integration_test.go index f997b1e7..94ee4e46 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore_integration_test.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore_integration_test.go @@ -93,7 +93,38 @@ func TestPrepareOnAFreshDatabaseAppliesTheConsolidatedSchema(t *testing.T) { } } -// 1517/AC4: a database already at migration 0001 (the consolidated schema's own migration) applies whatever +// stepDatabaseToMigration0001 brings an empty database to exactly migration 0001 (the consolidated +// schema's own migration) the way Prepare itself would: the ledger table, the migration's SQL and its own +// row in `migration`, in one transaction — never a raw exec of the migration's SQL with no ledger entry, +// which is not "at migration 0001" at all but a database Prepare cannot account for (its own refusal, +// 1517/AC4, below the database holds tables and no record of any migration: it is unknown, so nothing is +// changed). This is issue 1517's own fix, from the lead's try feedback on bb594652: the earlier version of +// this test built such an unaccountable database by running every migration's SQL directly, which is what +// TestPrepareOnADatabaseAtMigration0001AppliesWhatFollows's own Prepare call then (correctly) refused. +func stepDatabaseToMigration0001(t *testing.T, ctx context.Context, pool *pgxpool.Pool, m Migration) { + t.Helper() + if _, err := pool.Exec(ctx, ledger); err != nil { + t.Fatal(err) + } + tx, err := pool.Begin(ctx) + if err != nil { + t.Fatal(err) + } + defer func() { _ = tx.Rollback(context.Background()) }() + if _, err := tx.Exec(ctx, m.SQL); err != nil { + t.Fatalf("migration %04d: %v", m.Number, err) + } + if _, err := tx.Exec(ctx, `insert into migration (number, name, checksum) values ($1, $2, $3)`, m.Number, m.Name, m.Checksum); err != nil { + t.Fatal(err) + } + if err := tx.Commit(ctx); err != nil { + t.Fatal(err) + } +} + +// 1517/AC4: a database already at migration 0001 (the consolidated schema's own migration), brought there +// the way Prepare itself brings a database to any migration (stepDatabaseToMigration0001, above — through +// the real migration path, recorded in the ledger, never a raw exec with no record of it), applies whatever // comes after it, and ends at the same tables as a fresh database given the consolidated schema outright. func TestPrepareOnADatabaseAtMigration0001AppliesWhatFollows(t *testing.T) { ctx := context.Background() @@ -115,11 +146,10 @@ func TestPrepareOnADatabaseAtMigration0001AppliesWhatFollows(t *testing.T) { if err != nil { t.Fatal(err) } - for _, m := range migrations { - if _, err := stepped.pool.Exec(ctx, m.SQL); err != nil { - t.Fatalf("migration %04d: %v", m.Number, err) - } + if len(migrations) == 0 { + t.Fatal("no migrations: nothing to bring the database to migration 0001 with") } + stepDatabaseToMigration0001(t, ctx, stepped.pool, migrations[0]) if _, err := stepped.Prepare(ctx); err != nil { t.Fatal(err) } diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/runtime_player_integration_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/runtime_player_integration_test.go index c9805802..f75322f5 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/runtime_player_integration_test.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/runtime_player_integration_test.go @@ -223,6 +223,12 @@ type readerPlayer struct { stand *standIns nc *nats.Conn + + // subscribed carries the pattern of every mesh/subscribe the bundle has asked, in order (issue + // 1431/1518, fixed per the lead's try feedback on bb594652): buffered generously, since the bundle's + // hearLoop asks exactly once on success and a test waits on this before publishing anything, never the + // other way around. + subscribed chan string } // startReaderPlayer builds and starts mesh-mail-reader's own binary with env, plays the handshake, and begins @@ -249,7 +255,7 @@ func startReaderPlayer(t *testing.T, binary string, env []string, stand *standIn if err := cmd.Start(); err != nil { t.Fatal(err) } - p := &readerPlayer{fail: failer, cmd: cmd, in: stdin, waiting: map[string]chan wireMessage{}, stand: stand, nc: nc} + p := &readerPlayer{fail: failer, cmd: cmd, in: stdin, waiting: map[string]chan wireMessage{}, stand: stand, nc: nc, subscribed: make(chan string, 8)} go p.readLoop(stdout) t.Cleanup(func() { _ = cmd.Process.Kill() @@ -282,10 +288,37 @@ func startReaderPlayer(t *testing.T, binary string, env []string, stand *standIn if err != nil { t.Fatal(err) } + // Wait for the bundle's own hearLoop to have asked mesh/subscribe for AddedPattern before this + // function returns, so a test publishing right after startReader can never lose that publish the + // way bb594652's try did (acceptance_integration_test.go's stderr: "does not hear ... yet (before + // Serve ...)"): on the real bus the plain subject this harness bridges from carries no durable + // consumer, so a publish made before the bundle's own listener is registered is gone for good, + // unlike on the mesh itself, where the controller's durable consumer keeps it. Bounded and failing + // loudly, never silently proceeding to race the publish against a subscribe that may never land. + p.waitSubscribed(t, AddedPattern, 10*time.Second) } return p } +// waitSubscribed blocks until the bundle has asked mesh/subscribe for pattern (issue 1431/1518), or fails +// the test loudly once bound has passed — never lets a caller publish against a subscribe that might not +// have landed yet. +func (p *readerPlayer) waitSubscribed(t *testing.T, pattern string, bound time.Duration) { + t.Helper() + deadline := time.After(bound) + for { + select { + case got := <-p.subscribed: + if got == pattern { + return + } + // Some other pattern (none expected today): keep waiting for the one asked. + case <-deadline: + t.Fatalf("mesh-mail-reader never asked mesh/subscribe for %q within %s", pattern, bound) + } + } +} + type prefixedLog struct { t *testing.T prefix string @@ -408,7 +441,21 @@ func (p *readerPlayer) handleBundleRequest(m wireMessage) { p.handleAsk(m, reply) case "mesh/subscribe": // Acked at once: this harness needs no durable consumer semantics of its own (1518's own catch-up is - // the reader's own store, proven in kit_test.go's unit tests, not this bus bridge). + // the reader's own store, proven in kit_test.go's unit tests, not this bus bridge). Recorded on + // p.subscribed BEFORE the ack is sent, and the SDK itself (stdio.go's Subscribe) registers the + // bundle's own local listener before it ever sends this request — so a test that has read this + // pattern off p.subscribed (waitSubscribed) knows the bundle's dispatch table already has it, and a + // publish made only after that can never race the bundle's own subscribe the way bb594652's try did + // (issue 1431/1518). + var params struct { + Pattern string `json:"pattern"` + } + _ = json.Unmarshal(m.Params, ¶ms) + select { + case p.subscribed <- params.Pattern: + default: + p.fail.Errorf("mesh/subscribe for %q: p.subscribed is full (more subscribes than any test expects)", params.Pattern) + } reply(map[string]any{}, "") default: p.fail.Errorf("the bundle asked %s, which the helper does not know (issue 1431)", m.Method)