mesh-mail: emit mail.send.failed when the sender's own claim refuses or finds a send already sending, same as a failed send after it (review round 1 finding 4; issue 1198)

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

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)
 		}

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