mesh-mail: index a send for a tell or mail_health's overdue check before its record carries the outcome, drop the old index entry only after, and report an index write that fails instead of dropping it silently (review round 1 finding 3; issues 1194, 1199)

This commit is contained in:
jochen
2026-10-19 10:38:07 +02:00
parent 0a7669b48f
commit acb82992ee
5 changed files with 246 additions and 54 deletions
@@ -8,6 +8,7 @@ package main
import (
"encoding/json"
"errors"
"strings"
"testing"
"time"
@@ -228,6 +229,80 @@ func TestClaimQueuedSendOnASendStillApprovedAsksRedeliveryAndRecordsNothing(t *t
}
}
// review round 1 finding 3: a failed write of the tells-pending index does not stop the refusal itself
// from being recorded — the record write and the message's own fate are never undone by an index that
// could not be kept — but the index failure is still reported, never silently dropped.
func TestClaimQueuedSendStillRefusesWhenTheTellsPendingIndexCannotBeWritten(t *testing.T) {
sends, router := newMemSends(), newSendRouter()
req := rawQueuedSend(t, sends)
sends.failKey, sends.failErr = tellsPendingKey, errors.New("the store is down")
_, d, _, outcome, err := claimQueuedSend(sends, router.Record, req.ID, judgedAt.Add(3*time.Minute))
if outcome != claimRefused || !strings.HasPrefix(d.Outcome, "refused: ") {
t.Fatalf("outcome %v, decision %+v", outcome, d)
}
if err == nil {
t.Fatalf("the index failure was not reported")
}
kept, _, ok, kerr := getSend(sends, req.ID)
if kerr != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != d.Outcome {
t.Fatalf("the refusal itself was not kept despite the index failing: %v %v %+v", kerr, ok, kept.Decided)
}
}
// Same finding, for the redelivery-finds-sending path (not proven).
func TestClaimQueuedSendStillRecordsNotProvenWhenTheTellsPendingIndexCannotBeWritten(t *testing.T) {
sends, router := newMemSends(), newSendRouter()
req := rawQueuedSend(t, sends)
current, rev, ok, err := getSend(sends, req.ID)
if err != nil || !ok {
t.Fatalf("reading it back: %v %v", err, ok)
}
current.Decided.Outcome = "sending"
if _, err := sends.Update(req.ID, current, rev); err != nil {
t.Fatalf("leaving it sending: %v", err)
}
sends.failKey, sends.failErr = tellsPendingKey, errors.New("the store is down")
_, d, _, outcome, err := claimQueuedSend(sends, router.Record, req.ID, judgedAt.Add(4*time.Minute))
if outcome != claimNotProven || d.Outcome != "not proven" {
t.Fatalf("outcome %v, decision %+v", outcome, d)
}
if err == nil {
t.Fatalf("the index failure was not reported")
}
kept, _, ok2, kerr := getSend(sends, req.ID)
if kerr != nil || !ok2 || kept.Decided == nil || kept.Decided.Outcome != "not proven" {
t.Fatalf("not proven was not kept despite the index failing: %v %v %+v", kerr, ok2, kept.Decided)
}
}
// Same finding, for publishQueued: a failed sender-pending index write does not stop "queued" from being
// recorded, since the bus already took the message once.
func TestPublishQueuedStillRecordsQueuedWhenTheSenderPendingIndexCannotBeWritten(t *testing.T) {
sends := newMemSends()
req := SendRequest{ID: newSendID(), Account: "mesh", To: []string{"ops@example.org"}, Subject: "s", Body: "body", At: judgedAt}
claim := SendDecision{At: judgedAt, Outcome: "approved"}
req.Decided = &claim
rev, err := sends.Create(req.ID, req)
if err != nil {
t.Fatal(err)
}
sends.failKey, sends.failErr = senderPendingKey, errors.New("the store is down")
queue := newMemQueue()
d, err := publishQueued(sends, req, rev, claim, queue.publish, judgedAt.Add(time.Minute))
if d.Outcome != "queued" {
t.Fatalf("decision: %+v, want queued despite the index failing", d)
}
if err == nil {
t.Fatalf("the index failure was not reported")
}
if n := queue.count(); n != 1 {
t.Fatalf("published %d time(s)", n)
}
}
// raceOnFirstUpdate performs one competing write the first time Update is called — simulating another
// delivery's claim landing first — so the caller's own compare-and-set loses with stdio.ErrChanged
// (novox/hq issue 1196, AC4).
@@ -46,7 +46,10 @@ func outcomeTestClaim(t *testing.T, attempt int) (SendRequest, SendDecision, uin
func TestRecordOutcomeOnSuccessRecordsSentAndEmitsIt(t *testing.T) {
req, claimed, rev, sends := outcomeTestClaim(t, 1)
emit := &memEmit{}
d, action, _ := recordOutcome(sends, emit.emit, req, claimed, rev, nil, judgedAt.Add(2*time.Minute))
d, action, _, err := recordOutcome(sends, emit.emit, req, claimed, rev, nil, judgedAt.Add(2*time.Minute))
if err != nil {
t.Fatalf("recordOutcome: %v", err)
}
if d.Outcome != "sent" {
t.Fatalf("outcome: %+v", d)
}
@@ -83,7 +86,10 @@ func TestRecordOutcomeOnNotTakenRetriesUpToTheFourthAttempt(t *testing.T) {
req, claimed, rev, sends := outcomeTestClaim(t, n)
emit := &memEmit{}
now := judgedAt.Add(2 * time.Minute)
d, action, delay := recordOutcome(sends, emit.emit, req, claimed, rev, &SendError{Outcome: NotTaken, Err: errors.New("no connection")}, now)
d, action, delay, err := recordOutcome(sends, emit.emit, req, claimed, rev, &SendError{Outcome: NotTaken, Err: errors.New("no connection")}, now)
if err != nil {
t.Fatalf("attempt %d: recordOutcome: %v", n, err)
}
if d.Outcome != "retrying" || d.Attempt != n {
t.Fatalf("attempt %d: outcome %+v", n, d)
}
@@ -102,8 +108,11 @@ func TestRecordOutcomeOnNotTakenRetriesUpToTheFourthAttempt(t *testing.T) {
func TestRecordOutcomeOnNotTakenFailsOnTheFifthAttempt(t *testing.T) {
req, claimed, rev, sends := outcomeTestClaim(t, MostAttempts)
emit := &memEmit{}
d, action, _ := recordOutcome(sends, emit.emit, req, claimed, rev, &SendError{Outcome: NotTaken, Err: errors.New("no connection")},
d, action, _, err := recordOutcome(sends, emit.emit, req, claimed, rev, &SendError{Outcome: NotTaken, Err: errors.New("no connection")},
judgedAt.Add(2*time.Minute))
if err != nil {
t.Fatalf("recordOutcome: %v", err)
}
if !strings.HasPrefix(d.Outcome, "failed: ") || d.Attempt != MostAttempts {
t.Fatalf("outcome: %+v", d)
}
@@ -120,8 +129,11 @@ func TestRecordOutcomeOnNotTakenFailsOnTheFifthAttempt(t *testing.T) {
func TestRecordOutcomeOnRefusedRecordsFailedAndTerminates(t *testing.T) {
req, claimed, rev, sends := outcomeTestClaim(t, 1)
emit := &memEmit{}
d, action, _ := recordOutcome(sends, emit.emit, req, claimed, rev, &SendError{Outcome: Refused, Err: errors.New("550 no such mailbox")},
d, action, _, err := recordOutcome(sends, emit.emit, req, claimed, rev, &SendError{Outcome: Refused, Err: errors.New("550 no such mailbox")},
judgedAt.Add(2*time.Minute))
if err != nil {
t.Fatalf("recordOutcome: %v", err)
}
if !strings.HasPrefix(d.Outcome, "failed: ") || !strings.Contains(d.Outcome, "550 no such mailbox") {
t.Fatalf("outcome: %+v", d)
}
@@ -136,8 +148,11 @@ func TestRecordOutcomeOnRefusedRecordsFailedAndTerminates(t *testing.T) {
func TestRecordOutcomeOnNotProvenRecordsNotProvenAndTerminates(t *testing.T) {
req, claimed, rev, sends := outcomeTestClaim(t, 1)
emit := &memEmit{}
d, action, _ := recordOutcome(sends, emit.emit, req, claimed, rev,
d, action, _, err := recordOutcome(sends, emit.emit, req, claimed, rev,
&SendError{Outcome: NotProven, Err: errors.New("connection lost after the data was written")}, judgedAt.Add(2*time.Minute))
if err != nil {
t.Fatalf("recordOutcome: %v", err)
}
if d.Outcome != "not proven" {
t.Fatalf("outcome: %+v", d)
}
@@ -148,3 +163,26 @@ func TestRecordOutcomeOnNotProvenRecordsNotProvenAndTerminates(t *testing.T) {
t.Fatalf("emitted: %+v", emit.calls)
}
}
// review round 1 finding 3: a failed write of the tells-pending index does not stop the failed decision
// itself from being recorded and emitted — only the index write is reported as failing.
func TestRecordOutcomeStillRecordsFailedWhenTheTellsPendingIndexCannotBeWritten(t *testing.T) {
req, claimed, rev, sends := outcomeTestClaim(t, 1)
sends.failKey, sends.failErr = tellsPendingKey, errors.New("the store is down")
emit := &memEmit{}
d, action, _, err := recordOutcome(sends, emit.emit, req, claimed, rev, &SendError{Outcome: Refused, Err: errors.New("550 no such mailbox")},
judgedAt.Add(2*time.Minute))
if err == nil {
t.Fatalf("the index failure was not reported")
}
if !strings.HasPrefix(d.Outcome, "failed: ") || action != actionTerm {
t.Fatalf("outcome %+v, action %v", d, action)
}
kept, _, ok, kerr := getSend(sends, req.ID)
if kerr != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != d.Outcome {
t.Fatalf("the decision itself was not kept despite the index failing: %v %v %+v", kerr, ok, kept.Decided)
}
if len(emit.calls) != 1 {
t.Fatalf("emitted %d time(s) despite the index failing", len(emit.calls))
}
}
+92 -47
View File
@@ -695,21 +695,28 @@ func publishQueued(store SendStore, req SendRequest, rev uint64, approved SendDe
if err := publish(req.ID, QueuedSend{Send: req.ID, Account: req.Account, Ask: req.ID}); err != nil {
return approved, nil
}
// mail_health's own overdue check (Overdue, issue 1199) needs to find every send the sender now owns,
// the same way the sweep needs the undecided index: a reserved key, since neither side's own state over
// mesh-sdk go/process lists its keys whole. Indexed BEFORE the record carries "queued" (novox/hq review
// round 1 finding 3): a crash between this and the write below must still leave it findable once
// "queued" does land — Overdue itself only ever matches "queued" or "retrying", so an entry that still
// reads "approved" in the meantime is simply not yet counted overdue.
indexErr := addSenderPending(store, req.ID)
queued := SendDecision{At: now, Outcome: "queued", Warrant: approved.Warrant}
final, newRev, err := updateDecision(store, req, rev, queued)
if err != nil {
// The publish stands either way: the bus already took the message once. The caller logs this; it is
// not retried (a retry here would publish a second message under a new id).
return approved, fmt.Errorf("the send %s was queued, and that could not be kept (%w)", req.ID, err)
return approved, errOr(indexErr, fmt.Errorf("the send %s was queued, and that could not be kept (%w)", req.ID, err))
}
if newRev != 0 {
_ = removeUndecided(store, req.ID) // from here the sender owns it; the tools' sweep is done with it
// mail_health's own overdue check (issue 1199) needs to find every send the sender now owns, the
// same way the sweep needed the undecided index: a reserved key, since neither side's own state
// over mesh-sdk go/process lists its keys whole.
_ = addSenderPending(store, req.ID)
// Only once "queued" is durably recorded is the undecided index's entry dropped: removing it first,
// then crashing before the write, would have left the send unreachable by the sweep too.
if rmErr := removeUndecided(store, req.ID); rmErr != nil {
indexErr = errOr(indexErr, fmt.Errorf("dropping %s from the undecided index: %w", req.ID, rmErr))
}
}
return final, nil
return final, indexErr
}
// ---- the sender's own claim (novox/hq ADR 0436 clauses 9, 10, issue 1196) --------------------------------------
@@ -772,19 +779,33 @@ func claimQueuedSend(store SendStore, record RouterRecord, id string, now time.T
// Every other refusal below follows a real Approve (the send reached at least queued): clause 9's own
// refusal owes the operator a tell, same as clause 10's (novox/hq ADR 0436 clause 11, issue 1199).
refuse := func(reason string) (SendRequest, SendDecision, uint64, claimOutcome, error) {
// Indexed for a tell BEFORE the record carries "refused" (novox/hq review round 1 finding 3): a
// crash between this and the write below must still leave it findable once refused does land;
// tellSend itself tolerates finding it still sending, queued or retrying in the meantime.
indexErr := addTellPending(store, req.ID)
d, _, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "refused: " + reason})
_ = removeSenderPending(store, req.ID)
_ = addTellPending(store, req.ID)
return req, d, 0, claimRefused, err
if err != nil {
return req, d, 0, claimRefused, errOr(indexErr, err)
}
// Only once refused is durably recorded is the sender-pending index's entry dropped.
if rmErr := removeSenderPending(store, req.ID); rmErr != nil {
indexErr = errOr(indexErr, fmt.Errorf("dropping %s from the sender-pending index: %w", req.ID, rmErr))
}
return req, d, 0, claimRefused, indexErr
}
switch {
case req.Decided.Outcome == "sending":
// A previous delivery may have handed it to the server already; never sent twice, and never
// retried automatically either way (ADR 0436 clause 10).
// retried automatically either way (ADR 0436 clause 10). Same ordering as refuse() above.
indexErr := addTellPending(store, req.ID)
d, _, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "not proven", Attempt: req.Decided.Attempt})
_ = removeSenderPending(store, req.ID)
_ = addTellPending(store, req.ID)
return req, d, 0, claimNotProven, err
if err != nil {
return req, d, 0, claimNotProven, errOr(indexErr, err)
}
if rmErr := removeSenderPending(store, req.ID); rmErr != nil {
indexErr = errOr(indexErr, fmt.Errorf("dropping %s from the sender-pending index: %w", req.ID, rmErr))
}
return req, d, 0, claimNotProven, indexErr
case req.Decided.Outcome == "approved":
// Not yet visible as queued (publishQueued publishes before it records queued): the tools did
// approve it, so this is not a refusal (novox/hq review round 1 finding 2).
@@ -882,31 +903,34 @@ const (
// *SendError, exactly as a real SendMail failure would be: safe to retry), and says what the caller does
// with the message next. Attempt (req's own, from the claim) decides retrying from failed on a not-taken
// failure: the fifth attempt always fails rather than retries again (novox/hq ADR 0436 clauses 10, 11).
func recordOutcome(store SendStore, emit Emit, req SendRequest, claimed SendDecision, rev uint64, sendErr error, now time.Time) (SendDecision, msgAction, time.Duration) {
write := func(d SendDecision, final bool) SendDecision {
result, _, err := settleSend(store, req, rev, d)
if final {
// From here the sender is done with it, one way or another (novox/hq issue 1199): mail_health's
// overdue check no longer needs to find it.
_ = removeSenderPending(store, req.ID)
func recordOutcome(store SendStore, emit Emit, req SendRequest, claimed SendDecision, rev uint64, sendErr error, now time.Time) (SendDecision, msgAction, time.Duration, error) {
write := func(d SendDecision, final, tellWorthy bool) (SendDecision, error) {
var indexErr error
if tellWorthy {
// Indexed for a tell BEFORE the record carries the outcome that owes it (novox/hq review round
// 1 finding 3): a crash between this and the write below must still leave it findable once the
// outcome does land; tellSend itself tolerates finding it still sending, queued or retrying.
indexErr = addTellPending(store, req.ID)
}
result, _, err := settleSend(store, req, rev, d)
if err != nil {
// The attempt (and its SMTP outcome) stand either way; the caller logs this. A redelivery (or the
// sweep, for a send stuck mid-write) finds whatever the real kept state is and reconciles.
return d
return d, errOr(indexErr, err)
}
return result
}
tell := func(outcome string) {
// Owed a tell (failed, refused or not proven, novox/hq ADR 0436 clause 11): the index the sender's
// own look reads, kept to once by tell: telling then told (issue 1199).
_ = addTellPending(store, req.ID)
_ = emit(MailSendFailed, req.ID, OutcomeEvent{Send: req.ID, Account: req.Account, Outcome: outcome})
if final {
// Only once the outcome is durably recorded is the sender-pending index's entry dropped: from
// here the sender is done with it, one way or another (novox/hq issue 1199).
if rmErr := removeSenderPending(store, req.ID); rmErr != nil {
indexErr = errOr(indexErr, fmt.Errorf("dropping %s from the sender-pending index: %w", req.ID, rmErr))
}
}
return result, indexErr
}
if sendErr == nil {
d := write(SendDecision{At: now, Outcome: "sent", Warrant: claimed.Warrant, Attempt: claimed.Attempt}, true)
d, err := write(SendDecision{At: now, Outcome: "sent", Warrant: claimed.Warrant, Attempt: claimed.Attempt}, true, false)
_ = emit(MailSendSent, req.ID, OutcomeEvent{Send: req.ID, Account: req.Account, Outcome: d.Outcome})
return d, actionAck, 0
return d, actionAck, 0, err
}
var se *SendError
if !errors.As(sendErr, &se) {
@@ -917,26 +941,26 @@ func recordOutcome(store SendStore, emit Emit, req SendRequest, claimed SendDeci
}
switch se.Outcome {
case Refused:
d := write(SendDecision{At: now, Outcome: "failed: " + se.Err.Error(), Warrant: claimed.Warrant,
Attempt: claimed.Attempt, Error: se.Err.Error()}, true)
tell(d.Outcome)
return d, actionTerm, 0
d, err := write(SendDecision{At: now, Outcome: "failed: " + se.Err.Error(), Warrant: claimed.Warrant,
Attempt: claimed.Attempt, Error: se.Err.Error()}, true, true)
_ = emit(MailSendFailed, req.ID, OutcomeEvent{Send: req.ID, Account: req.Account, Outcome: d.Outcome})
return d, actionTerm, 0, err
case NotProven:
d := write(SendDecision{At: now, Outcome: "not proven", Warrant: claimed.Warrant, Attempt: claimed.Attempt,
Error: se.Err.Error()}, true)
tell(d.Outcome)
return d, actionTerm, 0
d, err := write(SendDecision{At: now, Outcome: "not proven", Warrant: claimed.Warrant, Attempt: claimed.Attempt,
Error: se.Err.Error()}, true, true)
_ = emit(MailSendFailed, req.ID, OutcomeEvent{Send: req.ID, Account: req.Account, Outcome: d.Outcome})
return d, actionTerm, 0, err
default: // NotTaken
if claimed.Attempt >= MostAttempts {
d := write(SendDecision{At: now, Outcome: "failed: " + se.Err.Error(), Warrant: claimed.Warrant,
Attempt: claimed.Attempt, Error: se.Err.Error()}, true)
tell(d.Outcome)
return d, actionTerm, 0
d, err := write(SendDecision{At: now, Outcome: "failed: " + se.Err.Error(), Warrant: claimed.Warrant,
Attempt: claimed.Attempt, Error: se.Err.Error()}, true, true)
_ = emit(MailSendFailed, req.ID, OutcomeEvent{Send: req.ID, Account: req.Account, Outcome: d.Outcome})
return d, actionTerm, 0, err
}
delay := RetryDelays[claimed.Attempt-1]
d := write(SendDecision{At: now, Outcome: "retrying", Warrant: claimed.Warrant, Attempt: claimed.Attempt,
Error: se.Err.Error(), NextAttempt: now.Add(delay)}, false)
return d, actionRetry, delay // not a final outcome: no emit (clause 11 names only sent and failed)
d, err := write(SendDecision{At: now, Outcome: "retrying", Warrant: claimed.Warrant, Attempt: claimed.Attempt,
Error: se.Err.Error(), NextAttempt: now.Add(delay)}, false, false)
return d, actionRetry, delay, err // not a final outcome: no emit (clause 11 names only sent and failed)
}
}
@@ -947,6 +971,16 @@ func recordOutcome(store SendStore, emit Emit, req SendRequest, claimed SendDeci
// send ever sent, not with the few still live): a reserved key stands in for a list, exactly as the
// undecided index already does.
// errOr is a, or b when a is nil: the first of two errors a caller that can only report one keeps (novox/hq
// review round 1 finding 3), so an index-bookkeeping failure is never silently dropped once there is also a
// record-write failure to report, and vice versa.
func errOr(a, b error) error {
if a != nil {
return a
}
return b
}
// idIndex is the shape a send-id index keeps: generic across all three (undecided, sender-pending,
// tells-pending).
type idIndex struct {
@@ -1135,7 +1169,18 @@ func tellSend(store SendStore, tell TellOperator, id string, now time.Time) erro
if err != nil {
return err
}
if !ok || req.Decided == nil || !needsTell(req.Decided.Outcome) || req.Decided.Tell == "told" {
if !ok || req.Decided == nil {
_ = removeTellPending(store, id)
return nil
}
switch req.Decided.Outcome {
case "sending", "queued", "retrying":
// Indexed for a tell before the outcome that owes it had actually landed (addTellPending now runs
// before the write, novox/hq review round 1 finding 3): not decided yet whether a tell is owed.
// Left in the index; the next look finds it once the write lands.
return nil
}
if !needsTell(req.Decided.Outcome) || req.Decided.Tell == "told" {
_ = removeTellPending(store, id)
return nil
}
+8 -2
View File
@@ -219,8 +219,11 @@ func (h *senderHandler) handle(msg *process.Message) {
}
req, claimed, rev, outcome, err := claimQueuedSend(h.Store, h.Record, q.Send, h.now())
if err != nil {
// Logged, never returned on alone: an index-bookkeeping error (novox/hq review round 1 finding 3)
// can stand beside a real decision (claimRefused, claimNotProven) that the switch below still
// must act on; a fatal read or claim failure, with no decision made, always reads claimLostRace,
// which the switch already leaves to the bus.
h.logf("[mesh-mail] send: claiming %s: %v", q.Send, err)
return // left to the bus: a transient read or write failure, never a verified refusal.
}
switch outcome {
case claimLostRace:
@@ -239,7 +242,10 @@ func (h *senderHandler) handle(msg *process.Message) {
return
}
sendErr := h.attempt(req)
_, action, delay := recordOutcome(h.Store, h.Emit, req, claimed, rev, sendErr, h.now())
_, action, delay, err := recordOutcome(h.Store, h.Emit, req, claimed, rev, sendErr, h.now())
if err != nil {
h.logf("[mesh-mail] send: recording the outcome of %s: %v", q.Send, err)
}
switch action {
case actionAck:
if err := msg.Ack(); err != nil {
@@ -125,6 +125,34 @@ func TestTellSendNeverTellsForASentSend(t *testing.T) {
}
}
// review round 1 finding 3: a send indexed for a tell before its own outcome write has landed (recordOutcome
// and claimQueuedSend now index first) must not be dropped by a look that finds it still sending, queued or
// retrying — it is left in the index for the next look, once the real outcome lands.
func TestTellSendKeepsAnEntryWhoseRecordIsStillInFlight(t *testing.T) {
for _, outcome := range []string{"sending", "queued", "retrying"} {
sends := newMemSends()
req := SendRequest{ID: newSendID(), Account: "mesh", At: judgedAt}
req.Decided = &SendDecision{At: judgedAt, Outcome: outcome}
if _, err := sends.Create(req.ID, req); err != nil {
t.Fatal(err)
}
if err := addTellPending(sends, req.ID); err != nil {
t.Fatal(err)
}
tell := &memTell{}
if err := tellSend(sends, tell.tell, req.ID, judgedAt.Add(time.Minute)); err != nil {
t.Fatalf("%s: telling: %v", outcome, err)
}
if len(tell.headlines) != 0 {
t.Fatalf("%s: told before its outcome landed: %v", outcome, tell.headlines)
}
ids, err := tellPendingIDs(sends)
if err != nil || len(ids) != 1 {
t.Fatalf("%s: dropped from the index while still in flight: %v %v", outcome, err, ids)
}
}
}
// 1199/AC2: given a send queued more than 10 minutes ago, or retrying 10 minutes past its next attempt,
// when mail_health runs (Overdue), then it names the send; a send waiting for its next attempt is never
// overdue.