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 8e9a3b74..8f0a801b 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go @@ -13,6 +13,7 @@ import ( "context" "encoding/json" "fmt" + "strings" "time" ) @@ -79,7 +80,8 @@ func NewJudge(ask Ask, node string, now func() time.Time) JudgeFunc { deadline := now().Add(JudgeWithin + time.Minute) for { if err != nil { - return nil, OutcomeRefused, err.Error() + outcome, why := classifyJudgeError(err) + return nil, outcome, why } var out struct { Answer json.RawMessage `json:"answer"` @@ -107,6 +109,34 @@ func NewJudge(ask Ask, node string, now func() time.Time) JudgeFunc { } } +// classifyJudgeError sorts an error the judge's own ask (claude_code_judge, claude_code_judged) returned +// into its real outcome (review round 1 finding 4, issue 1519/AC2, 1520/AC4): the real claude-code module +// answers "did not answer within ; the judgement was ended" for a timeout and "answered, and not in +// the shape asked: " for a misshapen answer (modules/claude-code/cmd/claude-code/judge.go), both as +// this ask's own error, not as a JSON body NewJudge's loop above could read a Pending/Answer shape from — +// before this, every one of those landed as OutcomeRefused, indistinguishable from the agent module +// actually refusing the ask ("holds no licence"). Anything else, including that real refusal, is +// OutcomeRefused, as before. +func classifyJudgeError(err error) (HealthOutcome, string) { + msg := err.Error() + switch { + case strings.Contains(msg, "did not answer within"): + return OutcomeTimedOut, msg + case strings.Contains(msg, "not in the shape asked"): + return OutcomeOutOfShape, msg + default: + return OutcomeRefused, msg + } +} + +// isNoMessageError says whether err is mail-store.read's own "the account holds no message in +// " (mesh-mail's cmd/mesh-mail/tools.go readTool): the mail named by account/folder/uid is no +// longer there — moved or deleted since it was heard — the one read failure RunOneBatch treats as the mail +// being simply gone rather than a reading problem (review round 1 finding 3, issue 1520). +func isNoMessageError(err error) bool { + return err != nil && strings.Contains(err.Error(), "holds no message") +} + // 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) @@ -155,16 +185,24 @@ func recordOutcome(ctx context.Context, store Store, outcome HealthOutcome, why // caller's to select, so a test can hand it an exact slice), builds the question (question.go) from them, the // existing mail groups and the applicable reading lessons, judges it, and applies the answer: // +// - a mail whose read answers "holds no message" (moved or deleted since it was heard, review round 1 +// finding 3, issue 1520) is taken out of the waiting queue, with a log line, and never reaches the +// question — the batch goes on with the rest; +// - any other read failure leaves every mail of this batch waiting, in the same order, for the next +// batch slot — the whole batch, not only the one that failed to read; // - on success (1520/AC3): one mail entry per mail, any new mail group found-or-created by its label, and // each mail leaves the waiting queue; // - on a refused, timed-out or out-of-shape judgement (1520/AC4): no entry is written, every given mail // stays waiting, in the same order, and the reader's health (1519) is set unhealthy with the outcome and // why; the next successful batch sets it healthy again. // +// Neither a mail that is simply gone nor an ordinary read failure ever makes the reader unhealthy (ADR 0459 +// §12: health is the judgement's, review round 1 finding 3) — only a judgement's own outcome does. +// // mesh-mail-reader never asks to send, move, flag or delete (ADR 0459 §11): this function asks only readText // (seat:mail-store.read) and judge (claude-code.claude_code_judge, claude_code_judged) — 1520/AC5 is proven by // building both from a recording Ask and checking what it recorded, not by anything this function does itself. -func RunOneBatch(ctx context.Context, store Store, readText ReadTextFunc, judge JudgeFunc, waiting []WaitingMail, now func() time.Time) error { +func RunOneBatch(ctx context.Context, store Store, readText ReadTextFunc, judge JudgeFunc, waiting []WaitingMail, now func() time.Time, logf func(string, ...any)) error { if len(waiting) > MaxMailsPerBatch { waiting = waiting[:MaxMailsPerBatch] } @@ -179,17 +217,31 @@ func RunOneBatch(ctx context.Context, store Store, readText ReadTextFunc, judge if err != nil { return err } - mails := make([]QuestionMail, len(waiting)) - ids := make([]int64, len(waiting)) - for i, w := range waiting { + live := make([]WaitingMail, 0, len(waiting)) + mails := make([]QuestionMail, 0, len(waiting)) + ids := make([]int64, 0, len(waiting)) + for _, w := range waiting { text, err := readText(w.Account, w.Folder, w.UID) if err != nil { - // mail-store.read itself failing stops the batch exactly as a refused judgement would: the mails - // stay waiting, and the reader is made unhealthy with the cause. - return recordOutcome(ctx, store, OutcomeRefused, "reading the mail's text: "+err.Error(), now()) + if isNoMessageError(err) { + if takeErr := store.TakeWaiting(ctx, w.Seq); takeErr != nil { + return takeErr + } + logf("[mesh-mail-reader] %s %s uid %d is gone (moved or deleted since it was heard): taken out of the waiting queue, never judged", w.Account, w.Folder, w.UID) + continue + } + // Any other read failure: the whole batch — this mail and whatever is after it — waits for + // the next batch slot, and the reader stays whatever health it already was. + logf("[mesh-mail-reader] reading %s %s uid %d: %v; this batch waits for the next slot", w.Account, w.Folder, w.UID, err) + return nil } - mails[i] = QuestionMail{ID: w.Seq, Account: w.Account, Folder: w.Folder, Date: w.Date, From: w.From, Subject: w.Subject, Flags: w.Flags, Text: text} - ids[i] = w.Seq + live = append(live, w) + mails = append(mails, QuestionMail{ID: w.Seq, Account: w.Account, Folder: w.Folder, Date: w.Date, From: w.From, Subject: w.Subject, Flags: w.Flags, Text: text}) + ids = append(ids, w.Seq) + } + if len(live) == 0 { + // Every mail given was gone: the queue already reflects it, and there is nothing left to judge. + return nil } question := BuildQuestion(mails, groups, lessons) raw, outcome, why := judge(question.Text, AnswerSchema()) @@ -201,7 +253,7 @@ func RunOneBatch(ctx context.Context, store Store, readText ReadTextFunc, judge return recordOutcome(ctx, store, OutcomeOutOfShape, err.Error(), now()) } judgedAt := now() - for _, w := range waiting { + for _, w := range live { j := answer.Mails[w.Seq] var groupID int64 if j.Group != "" { @@ -262,7 +314,7 @@ func RunBatches(ctx context.Context, store Store, readText ReadTextFunc, judge J waiting = waiting[:MaxMailsPerBatch] } lastStart = now() - if err := RunOneBatch(ctx, store, readText, judge, waiting, now); err != nil { + if err := RunOneBatch(ctx, store, readText, judge, waiting, now, logf); err != nil { logf("[mesh-mail-reader] a batch failed: %v", err) } if err := SendSignal(ctx, store, check, tell); err != nil { 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 06882585..2403babd 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 @@ -5,10 +5,14 @@ import ( "encoding/json" "fmt" "strings" + "sync" "testing" "time" ) +// noopLogf is the logf RunOneBatch and RunBatches want, for a test that does not care what was logged. +func noopLogf(string, ...any) {} + func seedWaiting(t *testing.T, store Store, n int) { t.Helper() for i := 0; i < n; i++ { @@ -58,45 +62,77 @@ func TestNextBatchWait(t *testing.T) { } } -// 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). +// 1520/AC1 (review round 1 finding 5): with 30 mails waiting, RunBatches ITSELF — never a hand-driven copy +// of its own loop — runs three batches of 10, their starts at least MinBatchGap (160s) apart. A test clock: +// now reads a clock this test owns, and sleep (the only thing RunBatches calls to wait) advances that same +// clock instead of actually waiting 320s+ of wall time. The judge stand-in records when each batch was +// judged (synchronous with RunOneBatch's own lastStart, since nothing sleeps between them) and cancels ctx +// once the 3rd has run, so RunBatches returns on its own next turn through the loop — this fails the moment +// RunBatches stops pacing, not only when nextBatchWait's own math would (TestNextBatchWait, above). func TestThirtyWaitingMailsRunAsThreeBatchesAtLeast160sApart(t *testing.T) { - ctx := context.Background() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() store := newMemStore() seedWaiting(t, store, 30) - - now := time.Date(2026, 10, 20, 10, 0, 0, 0, time.UTC) - var lastStart time.Time - var starts []time.Time - batches := 0 - for { - all, err := store.Waiting(ctx) - if err != nil { - t.Fatal(err) - } - if len(all) == 0 { - break - } - if wait := nextBatchWait(lastStart, now, MinBatchGap); wait > 0 { - now = now.Add(wait) - } - batch := all - if len(batch) > MaxMailsPerBatch { - batch = batch[:MaxMailsPerBatch] - } - nowFunc := func() time.Time { return now } - if err := RunOneBatch(ctx, store, noopReadText, judgeAllNeither(batch), batch, nowFunc); err != nil { - t.Fatal(err) - } - starts = append(starts, now) - lastStart = now - batches++ - now = now.Add(time.Second) // the batch itself takes a little time + seeded, err := store.Waiting(context.Background()) + if err != nil { + t.Fatal(err) } - if batches != 3 { - t.Fatalf("ran %d batches, want 3 (30 mails / 10 per batch)", batches) + ids := make([]int64, len(seeded)) + for i, w := range seeded { + ids[i] = w.Seq + } + + var clockMu sync.Mutex + clock := time.Date(2026, 10, 20, 10, 0, 0, 0, time.UTC) + now := func() time.Time { + clockMu.Lock() + defer clockMu.Unlock() + return clock + } + sleep := func(d time.Duration) { + clockMu.Lock() + clock = clock.Add(d) + clockMu.Unlock() + } + + var starts []time.Time + call := 0 + judge := func(string, map[string]any) (json.RawMessage, HealthOutcome, string) { + starts = append(starts, now()) + start := call * MaxMailsPerBatch + end := start + MaxMailsPerBatch + if end > len(ids) { + end = len(ids) + } + batch := ids[start:end] + call++ + mails := make([]map[string]any, len(batch)) + for i, id := range batch { + mails[i] = map[string]any{"id": id, "mark": "neither"} + } + raw, _ := json.Marshal(map[string]any{"mails": mails}) + if len(starts) == 3 { + cancel() + } + return raw, OutcomeOK, "" + } + check := func(string) (bool, error) { return true, nil } + tell := func(string, []string, string) (string, error) { return "", nil } + + done := make(chan struct{}) + go func() { + RunBatches(ctx, store, noopReadText, judge, check, tell, now, sleep, noopLogf) + close(done) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("RunBatches did not stop after its own 3rd batch and this test's cancel") + } + + if len(starts) != 3 { + t.Fatalf("judge was called %d time(s), want 3 (30 mails / %d per batch)", len(starts), MaxMailsPerBatch) } for i := 1; i < len(starts); i++ { if gap := starts[i].Sub(starts[i-1]); gap < MinBatchGap { @@ -183,6 +219,34 @@ func TestNewJudgeRefused(t *testing.T) { } } +// 1519/AC2, 1520/AC4 (review round 1 finding 4): the ask's own error is sorted into its real outcome by the +// exact words the real claude-code module answers with (modules/claude-code/cmd/claude-code/judge.go) — +// never OutcomeRefused for all three, which is what every ask error landed as before. +func TestNewJudgeSortsItsCauseByTheRealJudgesWords(t *testing.T) { + cases := []struct { + name string + err string + want HealthOutcome + }{ + {"timed out", "the agent on novox did not answer within 40s; the judgement was ended", OutcomeTimedOut}, + {"out of shape", `the agent on novox answered, and not in the shape asked: "mails" is not an array`, OutcomeOutOfShape}, + {"refused", "the agent on novox could not judge: holds no licence", OutcomeRefused}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + ask := func(string, map[string]any) (json.RawMessage, error) { return nil, fmt.Errorf("%s", c.err) } + judge := NewJudge(ask, "novox", time.Now) + _, outcome, why := judge("the question", AnswerSchema()) + if outcome != c.want { + t.Fatalf("outcome = %q, want %q", outcome, c.want) + } + if why != c.err { + t.Fatalf("why = %q, want the ask's own error %q, unchanged", why, c.err) + } + }) + } +} + // 1520/AC3: an answered batch writes one mail entry per mail and each new mail group, and the waiting mails // leave the queue. func TestRunOneBatchWritesEntriesAndGroups(t *testing.T) { @@ -199,7 +263,7 @@ func TestRunOneBatchWritesEntriesAndGroups(t *testing.T) { return raw, OutcomeOK, "" } now := time.Date(2026, 10, 20, 11, 0, 0, 0, time.UTC) - if err := RunOneBatch(ctx, store, noopReadText, judge, waiting, func() time.Time { return now }); err != nil { + if err := RunOneBatch(ctx, store, noopReadText, judge, waiting, func() time.Time { return now }, noopLogf); err != nil { t.Fatal(err) } entries, _ := store.Entries(ctx) @@ -245,7 +309,7 @@ func TestRunOneBatchFailureKeepsWaitingAndSetsUnhealthy(t *testing.T) { seedWaiting(t, store, 2) waiting, _ := store.Waiting(ctx) now := time.Date(2026, 10, 20, 9, 0, 0, 0, time.UTC) - if err := RunOneBatch(ctx, store, noopReadText, c.judge, waiting, func() time.Time { return now }); err != nil { + if err := RunOneBatch(ctx, store, noopReadText, c.judge, waiting, func() time.Time { return now }, noopLogf); err != nil { t.Fatal(err) } if entries, _ := store.Entries(ctx); len(entries) != 0 { @@ -262,7 +326,7 @@ func TestRunOneBatchFailureKeepsWaitingAndSetsUnhealthy(t *testing.T) { // The next good batch sets it healthy again. later := now.Add(time.Hour) - if err := RunOneBatch(ctx, store, noopReadText, judgeAllNeither(waiting), waiting, func() time.Time { return later }); err != nil { + if err := RunOneBatch(ctx, store, noopReadText, judgeAllNeither(waiting), waiting, func() time.Time { return later }, noopLogf); err != nil { t.Fatal(err) } h, _ = store.ReaderHealth(ctx) @@ -292,7 +356,7 @@ func TestRunOneBatchNeverAsksToSendMoveFlagOrDelete(t *testing.T) { } readText := NewReadText(recordingAsk) judge := NewJudge(recordingAsk, "novox", time.Now) - if err := RunOneBatch(ctx, store, readText, judge, waiting, time.Now); err != nil { + if err := RunOneBatch(ctx, store, readText, judge, waiting, time.Now, noopLogf); err != nil { t.Fatal(err) } if len(asked) == 0 { @@ -306,3 +370,73 @@ func TestRunOneBatchNeverAsksToSendMoveFlagOrDelete(t *testing.T) { } } } + +// 1520 (review round 1 finding 3): a mail-store.read that answers "holds no message" (the mail was moved +// or deleted since it was heard) takes that one mail out of the waiting queue and the batch goes on with +// the rest — never unhealthy (ADR 0459 §12: health is the judgement's). +func TestRunOneBatchTakesAGoneMailOutAndGoesOnWithTheRest(t *testing.T) { + ctx := context.Background() + store := newMemStore() + seedWaiting(t, store, 2) + waiting, _ := store.Waiting(ctx) + gone, stays := waiting[0], waiting[1] + + readText := func(account, folder string, uid uint32) (string, error) { + if uid == gone.UID { + return "", fmt.Errorf("the account %s holds no message %d in %s", account, uid, folder) + } + return "the mail's text", nil + } + judge := func(string, map[string]any) (json.RawMessage, HealthOutcome, string) { + raw, _ := json.Marshal(map[string]any{"mails": []map[string]any{{"id": stays.Seq, "mark": "neither"}}}) + return raw, OutcomeOK, "" + } + now := time.Date(2026, 10, 20, 11, 0, 0, 0, time.UTC) + if err := RunOneBatch(ctx, store, readText, judge, waiting, func() time.Time { return now }, noopLogf); err != nil { + t.Fatal(err) + } + entries, _ := store.Entries(ctx) + if len(entries) != 1 || entries[0].MessageID != stays.MessageID { + t.Fatalf("entries = %+v, want exactly one, for the mail that was still there", entries) + } + left, _ := store.Waiting(ctx) + if len(left) != 0 { + t.Fatalf("waiting = %v, want none (the gone mail taken out, the other judged)", left) + } + h, _ := store.ReaderHealth(ctx) + if !h.Healthy { + t.Fatalf("health = %+v, want healthy: a gone mail is never the judgement's problem", h) + } +} + +// 1520 (review round 1 finding 3): any OTHER read failure leaves the whole batch's mails waiting, in +// order, for the next batch slot — the judge is never even asked — and never makes the reader unhealthy. +func TestRunOneBatchOtherReadFailureLeavesTheWholeBatchWaiting(t *testing.T) { + ctx := context.Background() + store := newMemStore() + seedWaiting(t, store, 2) + waiting, _ := store.Waiting(ctx) + + readText := func(string, string, uint32) (string, error) { + return "", fmt.Errorf("the mailbox is unreachable") + } + judge := func(string, map[string]any) (json.RawMessage, HealthOutcome, string) { + t.Fatal("the judge must never be asked when a mail could not be read for a reason other than being gone") + return nil, OutcomeRefused, "" + } + now := time.Date(2026, 10, 20, 11, 0, 0, 0, time.UTC) + if err := RunOneBatch(ctx, store, readText, judge, waiting, func() time.Time { return now }, noopLogf); err != nil { + t.Fatal(err) + } + if entries, _ := store.Entries(ctx); len(entries) != 0 { + t.Fatalf("entries = %v, want none", entries) + } + left, _ := store.Waiting(ctx) + if len(left) != 2 || left[0].Seq != waiting[0].Seq || left[1].Seq != waiting[1].Seq { + t.Fatalf("waiting = %v, want the same 2 mails, in the same order", left) + } + h, _ := store.ReaderHealth(ctx) + if !h.Healthy { + t.Fatalf("health = %+v, want healthy: an ordinary read failure is never the judgement's problem", h) + } +} 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 4531ce8f..a45480cb 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 @@ -13,12 +13,17 @@ import ( "encoding/json" "errors" "fmt" + "regexp" "strings" "sync" "testing" "time" ) +// sinceRE matches the "(since HH:MM YYYY-MM-DD)" health.go's own Answer embeds in why (review round 1 +// finding 6): proves since when is actually named, not merely that the cause's own word ("licence") is. +var sinceRE = regexp.MustCompile(`since \d{2}:\d{2} \d{4}-\d{2}-\d{2}`) + // healthAnswer reads mail_reader_health through the player, as a plain map (the tool carries no caller // check, ADR 0459 §12). func healthAnswer(t *testing.T, r *startedReader) map[string]any { @@ -37,10 +42,14 @@ func healthAnswer(t *testing.T, r *startedReader) map[string]any { return out } -// 1524/AC1, AC2: a refused judgement ("holds no licence", as claude-code's own judge.go answers when the -// licence binding is removed) makes the reader unhealthy with that cause and since when, and sends no tell; -// once the judge answers again, both mails (one heard while unhealthy) are read in the order they were -// published, and the reader answers healthy again. +// 1524/AC1, AC2 (review round 1 finding 6): a mattering mail judged and told is answered FIRST, its own +// tell dropped (so it is genuinely untold, the one thing that would make a later batch try the mail +// signal again) — then a refused judgement ("holds no licence", as claude-code's own judge.go answers when +// the licence binding is removed) makes the reader unhealthy with that cause and since when, and sends no +// tell for that still-untold mail; once the judge answers again, both mails (the second heard while +// unhealthy) are read in the order they were published, and the reader answers healthy again. The earlier +// version of this test asserted "no tell while unhealthy" with no mattering mail ever judged at all, so +// SendSignal had nothing to try telling regardless of health — proving nothing. func TestAcceptanceRefusedJudgementIsUnhealthyThenCatchesUpInPublishedOrder(t *testing.T) { r := startReader(t) r.stand.MailAccounts = oneAccountAnswer("bob") @@ -48,12 +57,14 @@ func TestAcceptanceRefusedJudgementIsUnhealthyThenCatchesUpInPublishedOrder(t *t r.stand.MessengerCheck = func(map[string]any) (any, error) { return map[string]any{"allowed": true}, nil } var tellCalls count r.stand.Tell = func(map[string]any) (any, error) { - tellCalls.inc() + if tellCalls.inc() == 1 { + return map[string]any{"sent_on": ""}, nil // dropped: the operator channel's own cap (1522/AC2) + } return map[string]any{"sent_on": time.Now().UTC().Format(time.RFC3339)}, nil } var mu sync.Mutex - refusing := true + refusing := false var judgedOrder recorded[int64] r.stand.Judge = func(args map[string]any) (any, error) { mu.Lock() @@ -71,15 +82,39 @@ func TestAcceptanceRefusedJudgementIsUnhealthyThenCatchesUpInPublishedOrder(t *t } mails := make([]map[string]any, len(ids)) for i, id := range ids { - mails[i] = map[string]any{"id": id, "mark": "neither", "summary": "s"} + mails[i] = map[string]any{"id": id, "mark": "matters", "summary": "s", "why": "pay", "due-date": "2026-11-01"} } raw, _ := json.Marshal(map[string]any{"mails": mails}) return map[string]any{"answer": json.RawMessage(raw)}, nil } - publishAdded(t, r.nc, "bob", "INBOX", 1, []map[string]any{{"name": "Alice", "address": "alice@example.org"}}, "first", time.Now()) + conn := pgConnect(t, r.dbURL) + ctx := context.Background() - // 1524/AC1: unhealthy with cause and since, no tell. + // A mattering mail, judged successfully, its own tell dropped — so it stays genuinely untold. + publishAdded(t, r.nc, "bob", "INBOX", 1, []map[string]any{{"name": "Alice", "address": "alice@example.org"}}, "first", time.Now()) + waitFor(t, 30*time.Second, "the first (dropped) tell attempt", func() bool { return tellCalls.get() >= 1 }) + waitFor(t, 10*time.Second, "the first entry to be written", func() bool { + var n int + _ = conn.QueryRow(ctx, `select count(*) from mail_entry`).Scan(&n) + return n == 1 + }) + var told bool + if err := conn.QueryRow(ctx, `select told from mail_entry limit 1`).Scan(&told); err != nil { + t.Fatal(err) + } + if told { + t.Fatal("the mail was marked told even though its own tell was dropped") + } + + // From here, every judgement is refused (1524/AC1) — set before the second mail is even heard, so its + // own batch, whenever it runs, is refused. + mu.Lock() + refusing = true + mu.Unlock() + publishAdded(t, r.nc, "bob", "INBOX", 2, []map[string]any{{"name": "Bob", "address": "bob2@example.org"}}, "second", time.Now().Add(time.Second)) + + // 1524/AC1: unhealthy with cause and since when, no tell for the mail that is already untold. waitFor(t, 30*time.Second, "mail_reader_health to answer unhealthy", func() bool { h := healthAnswer(t, r) healthy, _ := h["healthy"].(bool) @@ -90,22 +125,20 @@ func TestAcceptanceRefusedJudgementIsUnhealthyThenCatchesUpInPublishedOrder(t *t if !strings.Contains(why, "licence") { t.Fatalf("health = %v, want why naming the licence", h) } - if n := tellCalls.get(); n != 0 { - t.Fatalf("a tell was sent while unhealthy: %d call(s)", n) + if !sinceRE.MatchString(why) { + t.Fatalf("health's why = %q, want it to name since when (HH:MM YYYY-MM-DD, health.go's own Answer)", why) } - - // A second mail, heard while still unhealthy — it must stay waiting, in order, not lost (1518's own - // guarantee, exercised here end to end). - publishAdded(t, r.nc, "bob", "INBOX", 2, []map[string]any{{"name": "Bob", "address": "bob2@example.org"}}, "second", time.Now().Add(time.Second)) + before := tellCalls.get() time.Sleep(2 * time.Second) // give the next failed batch attempt, if any, time to run and fail again + if n := tellCalls.get(); n != before { + t.Fatalf("a tell was sent while unhealthy: %d call(s), want still %d (the untold mail from above)", n, before) + } // The judge answers again (1524/AC2): catch-up in the order the two mails were published. mu.Lock() refusing = false mu.Unlock() - conn := pgConnect(t, r.dbURL) - ctx := context.Background() waitFor(t, BatchGap+60*time.Second, "both entries written", func() bool { var n int if err := conn.QueryRow(ctx, `select count(*) from mail_entry`).Scan(&n); err != nil { 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 c92c0b16..2478127e 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go @@ -19,6 +19,7 @@ import ( "encoding/json" "fmt" "os" + "sync" "time" stdio "git.novox.be/novox/mesh-sdk/go" @@ -33,14 +34,46 @@ func envFromOS() Env { return Env{Callers: os.Getenv("MESH_MAIL_READER_CALLERS")} } +// cachedStore opens the index once, lazily, and hands every later caller that SAME store (review round 1 +// finding 2, issue 1517/1518): the Bundle's own tool calls (one per mesh/ask, tools.go), hearLoop's handler +// (one per event heard) and batchLoop. Before this, each of those called storeFor itself, which opened a +// fresh *pgxpool.Pool (PgStoreFromEnv) every time and never closed it — one connection pool leaked per tool +// call and per heard event, for the life of the process. Mirrors mesh-mail's own main.go Index (serve): +// a failed open is never retried here on its own; it is retried the next time any caller calls get, exactly +// as hearLoop's own backoff or a tool call's own caller already does. +type cachedStore struct { + open func(ctx context.Context) (Store, error) + mu sync.Mutex + s Store +} + +func newCachedStore(open func(ctx context.Context) (Store, error)) *cachedStore { + return &cachedStore{open: open} +} + +// get is this cachedStore's own storeFor: the open store once it has one, or a fresh attempt to open one. +func (c *cachedStore) get(ctx context.Context) (Store, error) { + c.mu.Lock() + defer c.mu.Unlock() + if c.s != nil { + return c.s, nil + } + s, err := c.open(ctx) + if err != nil { + return nil, err + } + c.s = s + return c.s, nil +} + func main() { if len(os.Args) > 1 && os.Args[1] == "prepare" { os.Exit(prepare()) } - storeFor := func(ctx context.Context) (Store, error) { return PgStoreFromEnv(ctx) } - b := &Bundle{Env: envFromOS(), Store: storeFor} - go hearLoop(storeFor, stdio.Subscribe, time.Sleep) - go batchLoop(storeFor) + store := newCachedStore(func(ctx context.Context) (Store, error) { return PgStoreFromEnv(ctx) }) + b := &Bundle{Env: envFromOS(), Store: store.get} + go hearLoop(store.get, stdio.Subscribe, time.Sleep) + go batchLoop(store.get) if err := stdio.ServeOn("", b.Tools(), os.Stdin, os.Stdout); err != nil { logf("[mesh-mail-reader] %v", err) os.Exit(1) diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/main_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/main_test.go new file mode 100644 index 00000000..58125c92 --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/main_test.go @@ -0,0 +1,62 @@ +package main + +import ( + "context" + "errors" + "testing" +) + +// TestCachedStoreOpensOnceAndSharesTheSameStore (review round 1 finding 2, issue 1517/1518): two separate +// callers (as the Bundle's own tool calls and hearLoop's handler are, in the running bundle) each calling +// get get back the exact same Store, and open — the thing that would have opened a fresh *pgxpool.Pool +// every time — is called exactly once. +func TestCachedStoreOpensOnceAndSharesTheSameStore(t *testing.T) { + opens := 0 + want := newMemStore() + c := newCachedStore(func(context.Context) (Store, error) { + opens++ + return want, nil + }) + first, err := c.get(context.Background()) + if err != nil { + t.Fatal(err) + } + second, err := c.get(context.Background()) + if err != nil { + t.Fatal(err) + } + if first != want || second != want { + t.Fatalf("get returned %v and %v, want both to be %v", first, second, want) + } + if opens != 1 { + t.Fatalf("open was called %d time(s), want 1 (opened once, shared everywhere)", opens) + } +} + +// TestCachedStoreRetriesAfterAFailedOpen: a failed open is retried on the NEXT call, never looped on its +// own — the caller already retries (hearLoop's own backoff, a tool call's own caller). +func TestCachedStoreRetriesAfterAFailedOpen(t *testing.T) { + opens := 0 + want := newMemStore() + failWanted := errors.New("the index cannot be reached") + c := newCachedStore(func(context.Context) (Store, error) { + opens++ + if opens == 1 { + return nil, failWanted + } + return want, nil + }) + if _, err := c.get(context.Background()); err != failWanted { + t.Fatalf("first get: %v, want %v", err, failWanted) + } + got, err := c.get(context.Background()) + if err != nil { + t.Fatal(err) + } + if got != want { + t.Fatalf("second get = %v, want %v", got, want) + } + if opens != 2 { + t.Fatalf("open was called %d time(s), want 2 (the failed attempt, then the one that succeeded)", opens) + } +} 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 90d42212..7ac2eec1 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 @@ -55,6 +55,13 @@ func (r *recordedArgs) wasSet() bool { func TestAcceptanceMailCorrectionAndUndoReachTheStandIn(t *testing.T) { mw := startMailWeb(t, acceptanceAccounts(), []string{"alice"}) + // The classical list below the mail-reading section still asks seat:mail-store.accounts on every page + // load (accounts.go's own accountsPage), exactly as milestone 1's own tests set it up (review round 1 + // finding 1, issue 1523): left unset, GET "/" fails this test as an unknown ask, never reaching the + // mail-reading flow this test is actually about. + mw.store.Accounts = func(map[string]any) (any, error) { + return map[string]any{"accounts": []any{map[string]any{"account": "a", "kind": "imap", "address": "a@example.org", "state": "current"}}}, nil + } mw.mailReading.Status = func(map[string]any) (any, error) { return map[string]any{"healthy": true, "weeks": []map[string]any{}}, nil } diff --git a/modules/mesh-mail-web/cmd/mesh-mail-web/runtime_integration_test.go b/modules/mesh-mail-web/cmd/mesh-mail-web/runtime_integration_test.go index 698a8d55..a6e45b05 100644 --- a/modules/mesh-mail-web/cmd/mesh-mail-web/runtime_integration_test.go +++ b/modules/mesh-mail-web/cmd/mesh-mail-web/runtime_integration_test.go @@ -87,12 +87,12 @@ type standInMailStore struct { calls []askRecord } -// answer is this stand-in's own mailStoreFunc. +// answer is this stand-in's own mailStoreFunc. The key is matched BEFORE anything is recorded (review +// round 1 finding 1, issue 1430/1523): a key that is not one of mail-store's own five verbs — such as +// seat:mail-reading., which handleAsk tries here first and falls through on — is never added to +// Calls/Keys at all, so a test's own "only these verbs were asked" check (acceptance_integration_test.go's +// wantKeys) is never tripped by an ask this stand-in was never for. func (s *standInMailStore) answer(key string, args map[string]any) (any, error, bool) { - raw, _ := json.Marshal(args) - s.mu.Lock() - s.calls = append(s.calls, askRecord{Key: key, Body: raw}) - s.mu.Unlock() var f func(args map[string]any) (any, error) switch key { case "seat:mail-store.accounts": @@ -108,6 +108,10 @@ func (s *standInMailStore) answer(key string, args map[string]any) (any, error, default: return nil, nil, false } + raw, _ := json.Marshal(args) + s.mu.Lock() + s.calls = append(s.calls, askRecord{Key: key, Body: raw}) + s.mu.Unlock() if f == nil { return nil, nil, false } @@ -160,12 +164,14 @@ type standInMailReading struct { calls []askRecord } -// answer is this stand-in's own mailReadingFunc. +// answer is this stand-in's own mailReadingFunc. A key matched below but left unset by the test answers an +// error, never "unknown" (review round 1 finding 1, issue 1430/1523): accounts.go's mailReadingSection asks +// mail-reading.status on every page load now (issue 1430), so every test from before mail-reading existed +// that loads "/" and never configured this stand-in at all would otherwise fail as an unknown ask; an error +// here is exactly what mailReadingSection itself reads as "mail-reading does not answer" and takes its own +// "unavailable" path, same as the real seat refusing or timing out would. A key that is not one of +// mail-reading's own six verbs is still unknown, as before (never recorded, below). func (s *standInMailReading) answer(key string, args map[string]any) (any, error, bool) { - raw, _ := json.Marshal(args) - s.mu.Lock() - s.calls = append(s.calls, askRecord{Key: key, Body: raw}) - s.mu.Unlock() var f func(args map[string]any) (any, error) switch key { case "seat:mail-reading.entries": @@ -183,8 +189,12 @@ func (s *standInMailReading) answer(key string, args map[string]any) (any, error default: return nil, nil, false } + raw, _ := json.Marshal(args) + s.mu.Lock() + s.calls = append(s.calls, askRecord{Key: key, Body: raw}) + s.mu.Unlock() if f == nil { - return nil, nil, false + return nil, fmt.Errorf("the test's own stand-in for %s answers nothing", key), true } out, err := f(args) return out, err, true