diff --git a/modules/mesh-mail/cmd/mesh-mail/claim_test.go b/modules/mesh-mail/cmd/mesh-mail/claim_test.go index 1675331e..cb3ee606 100644 --- a/modules/mesh-mail/cmd/mesh-mail/claim_test.go +++ b/modules/mesh-mail/cmd/mesh-mail/claim_test.go @@ -344,6 +344,40 @@ func TestDecideSendRecordsQueuedDirectlyAndPublishes(t *testing.T) { } } +// 1196/AC2 (review round 2): a store that changes the revision between the claim and the refusal's own +// write — another writer's decision landing in between — never emits mail.send.failed for a write that did +// not actually land (refuse() does not retry: that is recordOutcome's own job, send.go, review round 2). +func TestClaimQueuedSendRefuseEmitsNothingWhenTheWriteDidNotLand(t *testing.T) { + sends, router := newMemSends(), newSendRouter() + req := rawQueuedSend(t, sends) + w := warrantFor(req, judgedAt.Add(time.Minute)) + w.AskDigest = "sha256:" + strings.Repeat("0", 64) // mismatched: claimQueuedSend refuses it + router.close(w) + racy := &raceOnFirstUpdate{SendStore: sends, race: func() { + r, rv, ok, err := getSend(sends, req.ID) + if err != nil || !ok { + t.Fatalf("racing write could not read the send: %v %v", err, ok) + } + if _, _, err := settleSend(sends, r, rv, SendDecision{At: judgedAt, Outcome: "not proven", Attempt: 1}); err != nil { + t.Fatalf("racing write: %v", err) + } + }} + emit := &memEmit{} + + _, d, _, outcome, _ := claimQueuedSend(racy, router.Record, emit.emit, req.ID, judgedAt.Add(3*time.Minute)) + if outcome != claimRefused { + t.Fatalf("outcome: %v, want claimRefused", outcome) + } + if len(emit.calls) != 0 { + t.Fatalf("emitted mail.send.failed despite the refusal's own write not landing: %+v", emit.calls) + } + // The kept record is the racing write's own decision, not this call's: refuse() never retries. + kept, _, ok, kerr := getSend(sends, req.ID) + if kerr != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != "not proven" { + t.Fatalf("kept: %v %v %+v (returned %+v)", kerr, ok, kept.Decided, d) + } +} + // 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). diff --git a/modules/mesh-mail/cmd/mesh-mail/outcome_test.go b/modules/mesh-mail/cmd/mesh-mail/outcome_test.go index f58168d5..f4358bec 100644 --- a/modules/mesh-mail/cmd/mesh-mail/outcome_test.go +++ b/modules/mesh-mail/cmd/mesh-mail/outcome_test.go @@ -231,3 +231,78 @@ func TestRecordOutcomeOnRefusedDoesNotTerminateOrEmitWhenTheWriteCannotBeKept(t t.Fatalf("emitted mail.send.failed despite the write never being kept: %+v", emit.calls) } } + +// 1198/AC1 (review round 2): a store that changes the revision between the claim and the outcome — another +// writer's decision landing in between — is retried once with the fresh revision; recordOutcome's own +// decision, once it does land, is what is kept, acked and emitted, never the other writer's state read +// back with a nil error mistaken for success. +func TestRecordOutcomeRetriesOnceWhenTheRevisionChangedBetweenClaimAndOutcome(t *testing.T) { + req, claimed, rev, sends := outcomeTestClaim(t, 1) + racy := &raceOnFirstUpdate{SendStore: sends, race: func() { + // A different delivery's own claim lands first, bumping the revision without ever finishing this + // call's own write — settleSend (via updateDecision) answers that with revision 0 and a nil error. + r, rv, ok, err := getSend(sends, req.ID) + if err != nil || !ok { + t.Fatalf("racing write could not read the send: %v %v", err, ok) + } + if _, _, err := settleSend(sends, r, rv, SendDecision{At: judgedAt, Outcome: "sending", Attempt: 2}); err != nil { + t.Fatalf("racing write: %v", err) + } + }} + emit := &memEmit{} + + d, action, _, err := recordOutcome(racy, emit.emit, req, claimed, rev, nil, judgedAt.Add(2*time.Minute)) + if err != nil { + t.Fatalf("recordOutcome: %v", err) + } + if d.Outcome != "sent" || action != actionAck { + t.Fatalf("outcome %+v, action %v, want sent/actionAck once retried with the fresh revision", d, action) + } + if len(emit.calls) != 1 || emit.calls[0].eventType != MailSendSent || emit.calls[0].body.Outcome != "sent" { + t.Fatalf("emitted: %+v", emit.calls) + } + kept, _, ok, kerr := getSend(sends, req.ID) + if kerr != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != "sent" { + t.Fatalf("the kept record is not this call's own decision: %v %v %+v", kerr, ok, kept.Decided) + } +} + +// Same finding: when the race repeats on the retry too (a second writer keeps landing first), recordOutcome +// gives up rather than looping or mistaking the other writer's state for its own — actionLeave, nothing +// emitted, the error reported. +func TestRecordOutcomeGivesUpWhenTheRevisionKeepsChanging(t *testing.T) { + req, claimed, rev, sends := outcomeTestClaim(t, 1) + always := &raceEveryUpdate{SendStore: sends, race: func() { + r, rv, ok, err := getSend(sends, req.ID) + if err != nil || !ok { + t.Fatalf("racing write could not read the send: %v %v", err, ok) + } + if _, _, err := settleSend(sends, r, rv, SendDecision{At: judgedAt, Outcome: "sending", Attempt: r.Decided.Attempt + 1}); err != nil { + t.Fatalf("racing write: %v", err) + } + }} + emit := &memEmit{} + + _, action, _, err := recordOutcome(always, emit.emit, req, claimed, rev, nil, judgedAt.Add(2*time.Minute)) + if err == nil { + t.Fatalf("a persistently changing revision was not reported") + } + if action != actionLeave { + t.Fatalf("action: %v, want actionLeave", action) + } + if len(emit.calls) != 0 { + t.Fatalf("emitted despite never keeping its own write: %+v", emit.calls) + } +} + +// raceEveryUpdate performs the competing write before every Update call, not only the first — simulating a +// revision that keeps changing out from under every retry (novox/hq review round 2, issue 1198). +type raceEveryUpdate struct { + SendStore + race func() +} + +func (r *raceEveryUpdate) Update(key string, value any, revision uint64) (uint64, error) { + r.race() + return r.SendStore.Update(key, value, revision) +} diff --git a/modules/mesh-mail/cmd/mesh-mail/send.go b/modules/mesh-mail/cmd/mesh-mail/send.go index 04fab74c..42319036 100644 --- a/modules/mesh-mail/cmd/mesh-mail/send.go +++ b/modules/mesh-mail/cmd/mesh-mail/send.go @@ -759,11 +759,13 @@ func claimQueuedSend(store SendStore, record RouterRecord, emit Emit, id string, // 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}) - _ = emit(MailSendFailed, req.ID, OutcomeEvent{Send: req.ID, Account: req.Account, Outcome: d.Outcome}) - if err != nil { + d, newRev, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "refused: " + reason}) + if newRev == 0 { + // The write did not land (another writer's decision already stands, or it failed outright): + // nothing of this call's own making to emit (novox/hq review round 2, issue 1198). return req, d, 0, claimRefused, errOr(indexErr, err) } + _ = emit(MailSendFailed, req.ID, OutcomeEvent{Send: req.ID, Account: req.Account, Outcome: d.Outcome}) // 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)) @@ -775,11 +777,13 @@ func claimQueuedSend(store SendStore, record RouterRecord, emit Emit, id string, // A previous delivery may have handed it to the server already; never sent twice, and never // 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}) - _ = emit(MailSendFailed, req.ID, OutcomeEvent{Send: req.ID, Account: req.Account, Outcome: d.Outcome}) - if err != nil { + d, newRev, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "not proven", Attempt: req.Decided.Attempt}) + if newRev == 0 { + // The write did not land: nothing of this call's own making to emit (novox/hq review round 2, + // issue 1198). return req, d, 0, claimNotProven, errOr(indexErr, err) } + _ = emit(MailSendFailed, req.ID, OutcomeEvent{Send: req.ID, Account: req.Account, Outcome: d.Outcome}) if rmErr := removeSenderPending(store, req.ID); rmErr != nil { indexErr = errOr(indexErr, fmt.Errorf("dropping %s from the sender-pending index: %w", req.ID, rmErr)) } @@ -887,7 +891,12 @@ func recordOutcome(store SendStore, emit Emit, req SendRequest, claimed SendDeci // fresh one) before giving up: kept is false only when the record's own write could not be kept even // after that retry, in which case the caller emits nothing and leaves the message to the bus entirely — // never announcing an outcome (sent, failed, retrying) that was not actually recorded (novox/hq review - // round 1 finding 5). + // round 1 finding 5). "Kept" is decided from settleSend's own returned revision, never from its error + // alone: on a changed revision whose record already carries another writer's decision, settleSend + // (through updateDecision) returns that other decision with revision 0 and a nil error — a nil error + // is not success here, and without checking the revision the retry never ran at all, so the caller + // could emit an outcome (and ack) while the record actually held someone else's state (novox/hq review + // round 2, issue 1198). write := func(d SendDecision, final, tellWorthy bool) (result SendDecision, kept bool, outErr error) { var indexErr error if tellWorthy { @@ -897,11 +906,18 @@ func recordOutcome(store SendStore, emit Emit, req SendRequest, claimed SendDeci indexErr = addTellPending(store, req.ID) } curReq, curRev := req, rev + var newRev uint64 var writeErr error for attempt := 0; attempt < 2; attempt++ { - result, _, writeErr = settleSend(store, curReq, curRev, d) - if writeErr == nil || !errors.Is(writeErr, stdio.ErrChanged) { - break + result, newRev, writeErr = settleSend(store, curReq, curRev, d) + if newRev != 0 { + break // this write landed + } + if attempt == 1 { + break // retried once already; give up + } + if writeErr != nil && !errors.Is(writeErr, stdio.ErrChanged) { + break // a hard failure, not a revision race: retrying blindly would not help } fresh, freshRev, ok, rerr := getSend(store, req.ID) if rerr != nil || !ok { @@ -909,8 +925,12 @@ func recordOutcome(store SendStore, emit Emit, req SendRequest, claimed SendDeci } curReq, curRev = fresh, freshRev } - if writeErr != nil { - return d, false, errOr(indexErr, fmt.Errorf("the send %s's outcome could not be kept, even retried on a changed revision (%w)", req.ID, writeErr)) + if newRev == 0 { + reportErr := writeErr + if reportErr == nil { + reportErr = fmt.Errorf("the send %s's outcome could not be kept: another writer's decision already stands, even retried on its fresh revision", req.ID) + } + return d, false, errOr(indexErr, reportErr) } if final { // Only once the outcome is durably recorded is the sender-pending index's entry dropped: from