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)

2026-10-21 01:24:57 +02:00
parent bb594652fa
commit cfea9de530
7 changed files with 190 additions and 36 deletions
@@ -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
 		}
@@ -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)
+	}
+}
@@ -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 {
@@ -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)
+	}
+}
@@ -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)
@@ -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)
 	}
@@ -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, &params)
+		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)