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)

This commit is contained in:
jochen
2026-10-21 01:11:48 +02:00
parent d61fe4e2da
commit 794de09c4d
7 changed files with 186 additions and 57 deletions
@@ -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 {
@@ -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
@@ -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
}
@@ -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)
}
@@ -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()
}
})
@@ -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
}
@@ -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