From 794de09c4dde3bb71f67f513c5ad8cf94d5aa139 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 01:11:48 +0200 Subject: [PATCH] Fix the try: guard shared test state with mutexes, pgxpool over one conn, a short batch gap for integration builds, unique+forced database drops (issues 1431, 1524, 1517) --- .../acceptance_integration_test.go | 45 ++++++------ .../cmd/mesh-mail-reader/batch.go | 9 ++- .../mesh-mail-reader/batch_gap_integration.go | 20 ++++++ .../catchup_integration_test.go | 20 +++--- .../pgstore_integration_test.go | 23 +++++-- .../runtime_player_integration_test.go | 68 +++++++++++++++++-- .../mailwrite_integration_test.go | 58 +++++++++++++--- 7 files changed, 186 insertions(+), 57 deletions(-) create mode 100644 modules/mesh-mail-reader/cmd/mesh-mail-reader/batch_gap_integration.go diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/acceptance_integration_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/acceptance_integration_test.go index 0773ccba..ca013fff 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/acceptance_integration_test.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/acceptance_integration_test.go @@ -149,34 +149,35 @@ func TestAcceptanceOneMatteringMailSendsExactlyOneTellNamingIt(t *testing.T) { r.stand.MailRead = func(map[string]any) (any, error) { return map[string]any{"text": "pay by friday"}, nil } r.stand.Judge = judgeEveryMailAs("matters", "", "due friday", "2026-11-01") - var checked []string + var checked recorded[string] r.stand.MessengerCheck = func(args map[string]any) (any, error) { text, _ := args["text"].(string) - checked = append(checked, text) + checked.add(text) // The content rule's own shape (ADR 0461 §4): refuse words carrying a URL, an address or a host name. allowed := !strings.Contains(text, "://") && !strings.Contains(text, "example.com") return map[string]any{"allowed": allowed}, nil } - var tells []map[string]any + var tells recorded[map[string]any] r.stand.Tell = func(args map[string]any) (any, error) { - tells = append(tells, args) + tells.add(args) return map[string]any{"sent_on": time.Now().UTC().Format(time.RFC3339)}, nil } publishAdded(t, r.nc, "bob", "INBOX", 1, []map[string]any{{"name": "Alice", "address": "alice@example.org"}}, "invoice", time.Now()) - waitFor(t, 30*time.Second, "exactly one tell", func() bool { return len(tells) >= 1 }) + waitFor(t, 30*time.Second, "exactly one tell", func() bool { return tells.len() >= 1 }) time.Sleep(2 * time.Second) // give a stray second tell a chance to prove this wrong, if it would happen - if len(tells) != 1 { - t.Fatalf("tells = %d, want exactly 1", len(tells)) + sent := tells.snapshot() + if len(sent) != 1 { + t.Fatalf("tells = %d, want exactly 1", len(sent)) } - if tells[0]["headline"] != "1 mail matters" { - t.Fatalf("headline = %v, want '1 mail matters'", tells[0]["headline"]) + if sent[0]["headline"] != "1 mail matters" { + t.Fatalf("headline = %v, want '1 mail matters'", sent[0]["headline"]) } - if tells[0]["subject"] != "mail" { - t.Fatalf("subject = %v, want mail", tells[0]["subject"]) + if sent[0]["subject"] != "mail" { + t.Fatalf("subject = %v, want mail", sent[0]["subject"]) } - lines, _ := tells[0]["lines"].([]any) + lines, _ := sent[0]["lines"].([]any) named := false for _, l := range lines { if s, ok := l.(string); ok && strings.Contains(s, "Alice") { @@ -186,7 +187,7 @@ func TestAcceptanceOneMatteringMailSendsExactlyOneTellNamingIt(t *testing.T) { if !named { t.Fatalf("lines %v do not name Alice", lines) } - for _, text := range checked { + for _, text := range checked.snapshot() { if strings.Contains(text, "://") || strings.Contains(text, "example.com") { t.Fatalf("messenger_check was asked a word carrying a URL or host name: %q", text) } @@ -204,16 +205,17 @@ func TestAcceptanceSenderNameWithAHostNameGivesTheSafeForm(t *testing.T) { text, _ := args["text"].(string) return map[string]any{"allowed": !strings.Contains(text, "evil.example.com")}, nil } - var tells []map[string]any + var tells recorded[map[string]any] r.stand.Tell = func(args map[string]any) (any, error) { - tells = append(tells, args) + tells.add(args) return map[string]any{"sent_on": time.Now().UTC().Format(time.RFC3339)}, nil } publishAdded(t, r.nc, "bob", "INBOX", 1, []map[string]any{{"name": "mail.evil.example.com", "address": "x@example.org"}}, "hi", time.Now()) - waitFor(t, 30*time.Second, "one tell", func() bool { return len(tells) >= 1 }) - lines, _ := tells[0]["lines"].([]any) + waitFor(t, 30*time.Second, "one tell", func() bool { return tells.len() >= 1 }) + sent := tells.snapshot() + lines, _ := sent[0]["lines"].([]any) if len(lines) != 1 || lines[0] != "open the mail page" { t.Fatalf("lines = %v, want the safe form (just 'open the mail page')", lines) } @@ -228,17 +230,16 @@ func TestAcceptanceDroppedTellLeavesMailUntoldAndCountedNextTime(t *testing.T) { r.stand.MailRead = func(map[string]any) (any, error) { return map[string]any{"text": "hi"}, nil } r.stand.Judge = judgeEveryMailAs("matters", "", "", "") r.stand.MessengerCheck = func(map[string]any) (any, error) { return map[string]any{"allowed": true}, nil } - var tellCalls int + var tellCalls count r.stand.Tell = func(map[string]any) (any, error) { - tellCalls++ - if tellCalls == 1 { + if tellCalls.inc() == 1 { return map[string]any{"sent_on": ""}, nil // dropped: the operator channel's own cap } return map[string]any{"sent_on": time.Now().UTC().Format(time.RFC3339)}, nil } publishAdded(t, r.nc, "bob", "INBOX", 1, []map[string]any{{"name": "Alice", "address": "alice@example.org"}}, "hi", time.Now()) - waitFor(t, 30*time.Second, "the first (dropped) tell attempt", func() bool { return tellCalls >= 1 }) + waitFor(t, 30*time.Second, "the first (dropped) tell attempt", func() bool { return tellCalls.get() >= 1 }) conn := pgConnect(t, r.dbURL) ctx := context.Background() @@ -258,7 +259,7 @@ func TestAcceptanceDroppedTellLeavesMailUntoldAndCountedNextTime(t *testing.T) { // A second mail, and its own batch (at least BatchGap after the first started), triggers the next mail // signal — which counts the still-untold first mail too (1429/AC4, 1522/AC2). publishAdded(t, r.nc, "bob", "INBOX", 2, []map[string]any{{"name": "Bob", "address": "bob2@example.org"}}, "second", time.Now()) - waitFor(t, BatchGap+60*time.Second, "a second tell attempt", func() bool { return tellCalls >= 2 }) + waitFor(t, BatchGap+60*time.Second, "a second tell attempt", func() bool { return tellCalls.get() >= 2 }) waitFor(t, 10*time.Second, "the first entry to become told", func() bool { var firstTold bool if err := conn.QueryRow(ctx, `select told from mail_entry order by id limit 1`).Scan(&firstTold); err != nil { 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 30ba8bcc..6f6ecca3 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go @@ -16,8 +16,13 @@ import ( "time" ) -// BatchGap is how long must pass between the start of one batch and the next (ADR 0459 §15, 1520/AC1). -const BatchGap = 160 * time.Second +// 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 // JudgeWithin is the bound every judgement asks: never above 40 seconds (ADR 0459 §14, 1520/AC2). const JudgeWithin = 40 * time.Second diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch_gap_integration.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch_gap_integration.go new file mode 100644 index 00000000..96a281a4 --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch_gap_integration.go @@ -0,0 +1,20 @@ +//go:build integration + +package main + +import "time" + +// batch_gap_integration.go shortens BatchGap (batch.go) for a binary built with `go build -tags +// integration` alone — never a production build, which never carries that tag, and never a setting an +// operator or the manifest could change (novox/hq ADR 0459 §15 still holds there: at least 160s). This is +// what lets an integration test (acceptance_integration_test.go, catchup_integration_test.go) exercise the +// reader's real pacing loop without actually waiting out 160s for each batch; a unit test on a test clock +// (batch_test.go's TestNextBatchWait, TestThirtyWaitingMailsRunAsThreeBatchesAtLeast160sApart) still proves +// the real 160s value, unaffected by this file, which they are never built with. +// +// issue 1431 (named by the lead's merge-check feedback, which also named 1524 and 1517): the earlier +// integration tests waited out the real pacing (one took 161.62s), which the merge check's own time bound +// for the integration level refuses. +func init() { + BatchGap = 3 * time.Second +} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/catchup_integration_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/catchup_integration_test.go index d6a31bb6..4531ce8f 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/catchup_integration_test.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/catchup_integration_test.go @@ -46,15 +46,15 @@ func TestAcceptanceRefusedJudgementIsUnhealthyThenCatchesUpInPublishedOrder(t *t r.stand.MailAccounts = oneAccountAnswer("bob") r.stand.MailRead = func(map[string]any) (any, error) { return map[string]any{"text": "hi"}, nil } r.stand.MessengerCheck = func(map[string]any) (any, error) { return map[string]any{"allowed": true}, nil } - var tellCalls int + var tellCalls count r.stand.Tell = func(map[string]any) (any, error) { - tellCalls++ + tellCalls.inc() return map[string]any{"sent_on": time.Now().UTC().Format(time.RFC3339)}, nil } var mu sync.Mutex refusing := true - var judgedOrder []int64 + var judgedOrder recorded[int64] r.stand.Judge = func(args map[string]any) (any, error) { mu.Lock() stillRefusing := refusing @@ -66,9 +66,9 @@ func TestAcceptanceRefusedJudgementIsUnhealthyThenCatchesUpInPublishedOrder(t *t } question, _ := args["question"].(string) ids := extractMailIDs(question) - mu.Lock() - judgedOrder = append(judgedOrder, ids...) - mu.Unlock() + for _, id := range ids { + judgedOrder.add(id) + } mails := make([]map[string]any, len(ids)) for i, id := range ids { mails[i] = map[string]any{"id": id, "mark": "neither", "summary": "s"} @@ -90,8 +90,8 @@ func TestAcceptanceRefusedJudgementIsUnhealthyThenCatchesUpInPublishedOrder(t *t if !strings.Contains(why, "licence") { t.Fatalf("health = %v, want why naming the licence", h) } - if tellCalls != 0 { - t.Fatalf("a tell was sent while unhealthy: %d call(s)", tellCalls) + if n := tellCalls.get(); n != 0 { + t.Fatalf("a tell was sent while unhealthy: %d call(s)", n) } // A second mail, heard while still unhealthy — it must stay waiting, in order, not lost (1518's own @@ -114,9 +114,7 @@ func TestAcceptanceRefusedJudgementIsUnhealthyThenCatchesUpInPublishedOrder(t *t return n == 2 }) - mu.Lock() - order := append([]int64(nil), judgedOrder...) - mu.Unlock() + order := judgedOrder.snapshot() if len(order) < 2 { t.Fatalf("judged ids = %v, want at least 2 (both mails, in one batch or more)", order) } 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 a85cd8a6..f997b1e7 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 @@ -10,10 +10,12 @@ package main import ( "context" + "fmt" "net/url" "os" "reflect" "strings" + "sync/atomic" "testing" "github.com/jackc/pgx/v5/pgxpool" @@ -28,8 +30,17 @@ func testDatabaseURL(t *testing.T) string { return url } -// freshIntegrationDatabase makes a new, empty database on the server MESH_TEST_POSTGRES names, and returns its -// own URL. +// freshDatabaseCounter makes freshIntegrationDatabase's own name unique across every call, even two calls +// within the SAME test (t.Name() alone is not: TestPrepareOnADatabaseAtMigration0001AppliesWhatFollows +// calls it twice, for "fresh" and "stepped", and two databases of the same name collided — the first +// call's own pool was still open on it when the second tried to drop and recreate it, "database ... is +// being accessed by other users", SQLSTATE 55006; the coordinator's own fix for issue 1517). +var freshDatabaseCounter atomic.Int64 + +// freshIntegrationDatabase makes a new, empty database on the server MESH_TEST_POSTGRES names, and returns +// its own URL. WITH (FORCE) on both drops is a second, independent safety (the coordinator's own +// suggestion): even a database this helper's own counter kept unique may still carry a connection this +// test has not yet closed (a *PgStore's pool, say) at cleanup time. func freshIntegrationDatabase(t *testing.T) string { t.Helper() ctx := context.Background() @@ -39,9 +50,9 @@ func freshIntegrationDatabase(t *testing.T) string { t.Fatal(err) } defer admin.Close() - name := "mesh_mail_reader_test_" + strings.ReplaceAll(t.Name(), "/", "_") - name = strings.ToLower(name) - if _, err := admin.Exec(ctx, "drop database if exists "+name); err != nil { + n := freshDatabaseCounter.Add(1) + name := fmt.Sprintf("mesh_mail_reader_test_%s_%d", strings.ToLower(strings.ReplaceAll(t.Name(), "/", "_")), n) + if _, err := admin.Exec(ctx, "drop database if exists "+name+" with (force)"); err != nil { t.Fatal(err) } if _, err := admin.Exec(ctx, "create database "+name); err != nil { @@ -50,7 +61,7 @@ func freshIntegrationDatabase(t *testing.T) string { t.Cleanup(func() { cleanup, err := pgxpool.New(context.Background(), base) if err == nil { - _, _ = cleanup.Exec(context.Background(), "drop database if exists "+name) + _, _ = cleanup.Exec(context.Background(), "drop database if exists "+name+" with (force)") cleanup.Close() } }) 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 4c58abd8..c9805802 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 @@ -33,7 +33,7 @@ import ( "testing" "time" - "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" ) @@ -172,7 +172,10 @@ func buildReaderBinary(t *testing.T) string { } t.Cleanup(func() { _ = os.RemoveAll(dir) }) bin := filepath.Join(dir, "mesh-mail-reader") - cmd := exec.Command("go", "build", "-o", bin, ".") + // -tags integration: batch_gap_integration.go shortens BatchGap for this binary alone, never a + // production build (the coordinator's own fix for issue 1431: the first try waited out the real 160s + // pacing and the merge check's own time bound refused it). + cmd := exec.Command("go", "build", "-tags", "integration", "-o", bin, ".") cmd.Dir, err = os.Getwd() if err != nil { t.Fatal(err) @@ -507,6 +510,55 @@ func publishAdded(t *testing.T, nc *nats.Conn, account, folder string, uid uint3 } } +// recorded is a mutex-guarded slice: a stand-in's own closure (called from the player's readLoop goroutine, +// runtime_player_integration_test.go's handleBundleRequest) records what it saw into one of these, and a +// test (its own, main goroutine) reads it back safely. The coordinator's own fix for issue 1431: the merge +// check's `go test -race` flagged every bare slice or counter a stand-in closure wrote and a test read with +// no synchronization between the two goroutines — a real race, not a false one, since the two communicate +// only over a stdio pipe and a nats connection, neither of which the race detector can see as +// synchronizing ordinary Go memory. +type recorded[T any] struct { + mu sync.Mutex + vals []T +} + +func (r *recorded[T]) add(v T) { + r.mu.Lock() + r.vals = append(r.vals, v) + r.mu.Unlock() +} + +func (r *recorded[T]) len() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.vals) +} + +func (r *recorded[T]) snapshot() []T { + r.mu.Lock() + defer r.mu.Unlock() + return append([]T(nil), r.vals...) +} + +// count is a mutex-guarded tally, recorded's own shape for a plain counter. +type count struct { + mu sync.Mutex + n int +} + +func (c *count) inc() int { + c.mu.Lock() + defer c.mu.Unlock() + c.n++ + return c.n +} + +func (c *count) get() int { + c.mu.Lock() + defer c.mu.Unlock() + return c.n +} + // waitFor polls cond every 200ms until it answers true or timeout passes, failing the test otherwise (every // test that waits on the reader's own asynchronous hearing and batching does so through this, never a bare // sleep). @@ -526,12 +578,16 @@ func waitFor(t *testing.T, timeout time.Duration, why string, cond func() bool) // pgConnect is a plain connection to the reader's own database, for a test to read its tables directly // (1431/AC1: "a search of every text column... finds nothing"). -func pgConnect(t *testing.T, url string) *pgx.Conn { +// pgConnect is a pgxpool.Pool, never a single pgx.Conn, to the reader's own database: a test that nests a +// query inside another's own rows.Next() loop (1431's own fix for "conn busy", acceptance_integration_test.go) +// then has each its own connection checked out of the pool, instead of racing one connection's single +// in-flight query. +func pgConnect(t *testing.T, url string) *pgxpool.Pool { t.Helper() - conn, err := pgx.Connect(context.Background(), url) + pool, err := pgxpool.New(context.Background(), url) if err != nil { t.Fatal(err) } - t.Cleanup(func() { _ = conn.Close(context.Background()) }) - return conn + t.Cleanup(pool.Close) + return pool } diff --git a/modules/mesh-mail-web/cmd/mesh-mail-web/mailwrite_integration_test.go b/modules/mesh-mail-web/cmd/mesh-mail-web/mailwrite_integration_test.go index 924e3aa9..90d42212 100644 --- a/modules/mesh-mail-web/cmd/mesh-mail-web/mailwrite_integration_test.go +++ b/modules/mesh-mail-web/cmd/mesh-mail-web/mailwrite_integration_test.go @@ -12,9 +12,47 @@ import ( "io" "net/http" "net/url" + "sync" "testing" ) +// recordedArgs is a mutex-guarded holder: a stand-in's own closure (called from the player's readLoop +// goroutine, runtime_integration_test.go's handleBundleRequest) records what it was asked here, and the +// test (its own, main goroutine) reads it back safely afterward. The coordinator's own fix for issue 1523: +// the merge check's `go test -race` found this exact unsynchronized pattern — a bare variable written by +// one goroutine and read by another, with no memory synchronization the race detector can see (the two +// communicate only over a stdio pipe and an HTTP round trip, neither of which it tracks as ordinary Go +// synchronization). +type recordedArgs struct { + mu sync.Mutex + args map[string]any + set bool +} + +func (r *recordedArgs) record(args map[string]any) { + r.mu.Lock() + r.args, r.set = args, true + r.mu.Unlock() +} + +func (r *recordedArgs) clear() { + r.mu.Lock() + r.args, r.set = nil, false + r.mu.Unlock() +} + +func (r *recordedArgs) get() map[string]any { + r.mu.Lock() + defer r.mu.Unlock() + return r.args +} + +func (r *recordedArgs) wasSet() bool { + r.mu.Lock() + defer r.mu.Unlock() + return r.set +} + func TestAcceptanceMailCorrectionAndUndoReachTheStandIn(t *testing.T) { mw := startMailWeb(t, acceptanceAccounts(), []string{"alice"}) mw.mailReading.Status = func(map[string]any) (any, error) { @@ -33,13 +71,13 @@ func TestAcceptanceMailCorrectionAndUndoReachTheStandIn(t *testing.T) { {"id": 9, "sender-name": "Bob", "subject": "fyi", "change": "mark: neither -> matters", "made": "2026-10-20T10:00:00Z"}, }}, nil } - var correctArgs, undoArgs map[string]any + var correctArgs, undoArgs recordedArgs mw.mailReading.Correct = func(args map[string]any) (any, error) { - correctArgs = args + correctArgs.record(args) return map[string]any{"id": args["id"]}, nil } mw.mailReading.Undo = func(args map[string]any) (any, error) { - undoArgs = args + undoArgs.record(args) return map[string]any{"undone": args["id"]}, nil } @@ -65,19 +103,19 @@ func TestAcceptanceMailCorrectionAndUndoReachTheStandIn(t *testing.T) { if correctResp.StatusCode != http.StatusOK { t.Fatalf("the correction: status %d: %s", correctResp.StatusCode, correctBody) } - if correctArgs["id"] != float64(1) || correctArgs["mark"] != "matters" || correctArgs["group"] != "bills" { - t.Fatalf("mail-reading.correct was asked %v, want id 1, mark matters, group bills", correctArgs) + if got := correctArgs.get(); got["id"] != float64(1) || got["mark"] != "matters" || got["group"] != "bills" { + t.Fatalf("mail-reading.correct was asked %v, want id 1, mark matters, group bills", got) } // Without the token, the correction never reaches the stand-in (1523/AC1). - correctArgs = nil + correctArgs.clear() noTokenResp := doForm(t, mw, "/mail-correct", url.Values{"id": {"1"}, "mark": {"neither"}}, cookie) _ = noTokenResp.Body.Close() if noTokenResp.StatusCode != http.StatusForbidden { t.Fatalf("a correction with no token: status %d, want 403", noTokenResp.StatusCode) } - if correctArgs != nil { - t.Fatalf("mail-reading.correct was asked %v with no token", correctArgs) + if correctArgs.wasSet() { + t.Fatalf("mail-reading.correct was asked %v with no token", correctArgs.get()) } // 1523/AC2: an undo posted with the session's own token reaches the stand-in's Undo. @@ -87,8 +125,8 @@ func TestAcceptanceMailCorrectionAndUndoReachTheStandIn(t *testing.T) { if undoResp.StatusCode != http.StatusOK { t.Fatalf("the undo: status %d: %s", undoResp.StatusCode, undoBody) } - if undoArgs["id"] != float64(9) { - t.Fatalf("mail-reading.undo was asked %v, want id 9", undoArgs) + if got := undoArgs.get(); got["id"] != float64(9) { + t.Fatalf("mail-reading.undo was asked %v, want id 9", got) } // 1523/AC4: nothing of mail is read but through mail-store — this flow never asks mail-store.search or