diff --git a/modules/mesh-mail/cmd/mesh-mail/claim_test.go b/modules/mesh-mail/cmd/mesh-mail/claim_test.go index da27507f..dfdde9db 100644 --- a/modules/mesh-mail/cmd/mesh-mail/claim_test.go +++ b/modules/mesh-mail/cmd/mesh-mail/claim_test.go @@ -78,7 +78,8 @@ func warrantFor(req SendRequest, at time.Time) asks.Warrant { func TestClaimQueuedSendClaimsSendingWithTheAttemptBeforeAnyConnection(t *testing.T) { id, router, sends := claimTestSend(t) - req, d, _, outcome, err := claimQueuedSend(sends, router.Record, id, judgedAt.Add(3*time.Minute)) + emit := &memEmit{} + req, d, _, outcome, err := claimQueuedSend(sends, router.Record, emit.emit, id, judgedAt.Add(3*time.Minute)) if err != nil { t.Fatalf("claiming: %v", err) } @@ -95,6 +96,9 @@ func TestClaimQueuedSendClaimsSendingWithTheAttemptBeforeAnyConnection(t *testin if err != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != "sending" { t.Fatalf("not kept as sending: %v %v %+v", err, ok, kept.Decided) } + if len(emit.calls) != 0 { + t.Fatalf("claiming to send emitted something: %+v", emit.calls) + } } // 1196/AC2: given a router record that is not mesh-mail's, not closed by Approve, approved after its @@ -109,29 +113,35 @@ func TestClaimQueuedSendRefusesAMismatchedOrUnclosedWarrant(t *testing.T) { w := warrantFor(req, judgedAt.Add(time.Minute)) w.Asker = "another-module" router.close(w) - _, d, _, outcome, err := claimQueuedSend(sends, router.Record, req.ID, judgedAt.Add(3*time.Minute)) + emit := &memEmit{} + _, d, _, outcome, err := claimQueuedSend(sends, router.Record, emit.emit, req.ID, judgedAt.Add(3*time.Minute)) if err != nil || outcome != claimRefused || !strings.HasPrefix(d.Outcome, "refused: ") { t.Fatalf("outcome %v, decision %+v, err %v", outcome, d, err) } + assertEmittedFailed(t, emit, req.ID, d.Outcome) }) t.Run("not closed by approve", func(t *testing.T) { sends, router := newMemSends(), newSendRouter() req := rawQueuedSend(t, sends) // The router keeps no record of this ask at all: never published, or never answered. - _, d, _, outcome, err := claimQueuedSend(sends, router.Record, req.ID, judgedAt.Add(3*time.Minute)) + emit := &memEmit{} + _, d, _, outcome, err := claimQueuedSend(sends, router.Record, emit.emit, req.ID, judgedAt.Add(3*time.Minute)) if err != nil || outcome != claimRefused || !strings.HasPrefix(d.Outcome, "refused: ") { t.Fatalf("outcome %v, decision %+v, err %v", outcome, d, err) } + assertEmittedFailed(t, emit, req.ID, d.Outcome) }) t.Run("approved after the ask's expiry", func(t *testing.T) { sends, router := newMemSends(), newSendRouter() req := rawQueuedSend(t, sends) w := warrantFor(req, req.Ask().Expires.Add(time.Hour)) router.close(w) - _, d, _, outcome, err := claimQueuedSend(sends, router.Record, req.ID, judgedAt.Add(3*time.Minute)) + emit := &memEmit{} + _, d, _, outcome, err := claimQueuedSend(sends, router.Record, emit.emit, req.ID, judgedAt.Add(3*time.Minute)) if err != nil || outcome != claimRefused || !strings.HasPrefix(d.Outcome, "refused: ") { t.Fatalf("outcome %v, decision %+v, err %v", outcome, d, err) } + assertEmittedFailed(t, emit, req.ID, d.Outcome) }) t.Run("a digest the published ask never bound", func(t *testing.T) { sends, router := newMemSends(), newSendRouter() @@ -139,31 +149,48 @@ func TestClaimQueuedSendRefusesAMismatchedOrUnclosedWarrant(t *testing.T) { w := warrantFor(req, judgedAt.Add(time.Minute)) w.AskDigest = "sha256:" + strings.Repeat("0", 64) router.close(w) - _, d, _, outcome, err := claimQueuedSend(sends, router.Record, req.ID, judgedAt.Add(3*time.Minute)) + emit := &memEmit{} + _, d, _, outcome, err := claimQueuedSend(sends, router.Record, emit.emit, req.ID, judgedAt.Add(3*time.Minute)) if err != nil || outcome != claimRefused || !strings.HasPrefix(d.Outcome, "refused: ") { t.Fatalf("outcome %v, decision %+v, err %v", outcome, d, err) } + assertEmittedFailed(t, emit, req.ID, d.Outcome) }) } +// assertEmittedFailed checks that exactly one mail.send.failed was emitted, naming the send id and the +// decision's own outcome in words (novox/hq review round 1 finding 4). +func assertEmittedFailed(t *testing.T, emit *memEmit, id, outcome string) { + t.Helper() + if len(emit.calls) != 1 { + t.Fatalf("emitted %d time(s), want exactly one mail.send.failed: %+v", len(emit.calls), emit.calls) + } + c := emit.calls[0] + if c.eventType != MailSendFailed || c.id != id || c.body.Send != id || c.body.Outcome != outcome { + t.Fatalf("emitted %+v, want mail.send.failed naming %s and outcome %q", c, id, outcome) + } +} + // 1196/AC3: given a send already sending (the sender stopped mid-send), when its message is delivered // again, then the sender sends nothing, records not proven and terminates. func TestClaimQueuedSendOnARedeliveryOfAnAlreadySendingClaimRecordsNotProven(t *testing.T) { id, router, sends := claimTestSend(t) - req, _, _, outcome, err := claimQueuedSend(sends, router.Record, id, judgedAt.Add(3*time.Minute)) + emit := &memEmit{} + req, _, _, outcome, err := claimQueuedSend(sends, router.Record, emit.emit, id, judgedAt.Add(3*time.Minute)) if err != nil || outcome != claimToSend { t.Fatalf("first claim: %v %v", outcome, err) } _ = req // Redelivered: the first claim's handler may have stopped mid-send. - _, d, _, outcome, err := claimQueuedSend(sends, router.Record, id, judgedAt.Add(4*time.Minute)) + _, d, _, outcome, err := claimQueuedSend(sends, router.Record, emit.emit, id, judgedAt.Add(4*time.Minute)) if err != nil { t.Fatalf("redelivered claim: %v", err) } if outcome != claimNotProven || d.Outcome != "not proven" { t.Fatalf("outcome %v, decision %+v", outcome, d) } + assertEmittedFailed(t, emit, id, d.Outcome) } // 1196/AC4: given a lost compare-and-set (another claim), when taken, then it sends nothing and leaves the @@ -181,7 +208,8 @@ func TestClaimQueuedSendOnALostRaceSendsNothingAndLeavesTheMessageToTheBus(t *te } }} - _, d, _, outcome, err := claimQueuedSend(racy, router.Record, id, judgedAt.Add(3*time.Minute)) + emit := &memEmit{} + _, d, _, outcome, err := claimQueuedSend(racy, router.Record, emit.emit, id, judgedAt.Add(3*time.Minute)) if err != nil { t.Fatalf("claiming: %v", err) } @@ -191,6 +219,9 @@ func TestClaimQueuedSendOnALostRaceSendsNothingAndLeavesTheMessageToTheBus(t *te if d.Outcome != "sending" || d.Attempt != 1 { t.Fatalf("the decision read back is not the racing claim's own: %+v", d) } + if len(emit.calls) != 0 { + t.Fatalf("a lost race emitted something: %+v", emit.calls) + } } // 1196/AC5: given a send still approved (publishQueued published to mail-sends before it recorded queued, @@ -209,7 +240,8 @@ func TestClaimQueuedSendOnASendStillApprovedAsksRedeliveryAndRecordsNothing(t *t } req = current - _, d, _, outcome, err := claimQueuedSend(sends, newSendRouter().Record, req.ID, judgedAt.Add(3*time.Minute)) + emit := &memEmit{} + _, d, _, outcome, err := claimQueuedSend(sends, newSendRouter().Record, emit.emit, req.ID, judgedAt.Add(3*time.Minute)) if err != nil { t.Fatalf("claiming: %v", err) } @@ -227,6 +259,9 @@ func TestClaimQueuedSendOnASendStillApprovedAsksRedeliveryAndRecordsNothing(t *t if err != nil || len(ids) != 0 { t.Fatalf("a tell was queued despite nothing being told: %v %v", err, ids) } + if len(emit.calls) != 0 { + t.Fatalf("a send still approved emitted something: %+v", emit.calls) + } } // review round 1 finding 3: a failed write of the tells-pending index does not stop the refusal itself @@ -237,7 +272,8 @@ func TestClaimQueuedSendStillRefusesWhenTheTellsPendingIndexCannotBeWritten(t *t 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)) + emit := &memEmit{} + _, d, _, outcome, err := claimQueuedSend(sends, router.Record, emit.emit, req.ID, judgedAt.Add(3*time.Minute)) if outcome != claimRefused || !strings.HasPrefix(d.Outcome, "refused: ") { t.Fatalf("outcome %v, decision %+v", outcome, d) } @@ -248,6 +284,7 @@ func TestClaimQueuedSendStillRefusesWhenTheTellsPendingIndexCannotBeWritten(t *t 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) } + assertEmittedFailed(t, emit, req.ID, d.Outcome) } // Same finding, for the redelivery-finds-sending path (not proven). @@ -264,7 +301,8 @@ func TestClaimQueuedSendStillRecordsNotProvenWhenTheTellsPendingIndexCannotBeWri } 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)) + emit := &memEmit{} + _, d, _, outcome, err := claimQueuedSend(sends, router.Record, emit.emit, req.ID, judgedAt.Add(4*time.Minute)) if outcome != claimNotProven || d.Outcome != "not proven" { t.Fatalf("outcome %v, decision %+v", outcome, d) } @@ -275,6 +313,7 @@ func TestClaimQueuedSendStillRecordsNotProvenWhenTheTellsPendingIndexCannotBeWri 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) } + assertEmittedFailed(t, emit, req.ID, d.Outcome) } // Same finding, for publishQueued: a failed sender-pending index write does not stop "queued" from being diff --git a/modules/mesh-mail/cmd/mesh-mail/send.go b/modules/mesh-mail/cmd/mesh-mail/send.go index f8dcc328..61b8815a 100644 --- a/modules/mesh-mail/cmd/mesh-mail/send.go +++ b/modules/mesh-mail/cmd/mesh-mail/send.go @@ -763,7 +763,7 @@ const NotYetQueuedDelay = 5 * time.Second // is the option's binds (Option.Performs) — exactly the tools' own check (Decide, above) mirrored here, // because the tool runner's account, which every agent session on this machine also reaches the bus as, may // publish on the queue and write `sends` (this is the boundary clause 9 describes). -func claimQueuedSend(store SendStore, record RouterRecord, id string, now time.Time) (SendRequest, SendDecision, uint64, claimOutcome, error) { +func claimQueuedSend(store SendStore, record RouterRecord, emit Emit, id string, now time.Time) (SendRequest, SendDecision, uint64, claimOutcome, error) { req, rev, ok, err := getSend(store, id) if err != nil { return SendRequest{}, SendDecision{}, 0, claimLostRace, err @@ -772,18 +772,21 @@ func claimQueuedSend(store SendStore, record RouterRecord, id string, now time.T return SendRequest{}, SendDecision{}, 0, claimLostRace, fmt.Errorf("no send %q is kept", id) } if req.Decided == nil { - // Nothing was ever approved: no tell owed (clause 11 ties the tell to a *failed approved* send). + // Nothing was ever approved: no tell owed (clause 11 ties the tell to a *failed approved* send), + // and nothing emitted either — clause 11's two events are for a send the operator did approve. d, _, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "refused: the send is not yet approved"}) return req, d, 0, claimRefused, err } // 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). + // refusal owes the operator a tell and the same mail.send.failed event as clause 10's refusal does + // (novox/hq ADR 0436 clause 11, issue 1199; review round 1 finding 4). 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}) + _ = emit(MailSendFailed, req.ID, OutcomeEvent{Send: req.ID, Account: req.Account, Outcome: d.Outcome}) if err != nil { return req, d, 0, claimRefused, errOr(indexErr, err) } @@ -799,6 +802,7 @@ func claimQueuedSend(store SendStore, record RouterRecord, id string, now time.T // 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 { return req, d, 0, claimNotProven, errOr(indexErr, err) } diff --git a/modules/mesh-mail/cmd/mesh-mail/sender.go b/modules/mesh-mail/cmd/mesh-mail/sender.go index 75a35316..7f376d67 100644 --- a/modules/mesh-mail/cmd/mesh-mail/sender.go +++ b/modules/mesh-mail/cmd/mesh-mail/sender.go @@ -217,7 +217,7 @@ func (h *senderHandler) handle(msg *process.Message) { } return } - req, claimed, rev, outcome, err := claimQueuedSend(h.Store, h.Record, q.Send, h.now()) + req, claimed, rev, outcome, err := claimQueuedSend(h.Store, h.Record, h.Emit, 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