mesh-mail: the sender records a send's outcome, retries a not-taken failure and emits mail.send.sent or mail.send.failed, acknowledging only once it is kept (ADR 0436 clauses 10, 11; issue 1198)

2026-10-19 10:11:41 +02:00
parent e7dce9616c
commit 33a97e6d13
5 changed files with 703 additions and 30 deletions
@@ -3,9 +3,10 @@ package main
 // mesh-mail send (novox/hq ADR 0436 §7): the sender, a process with a bus account of its own
 // (mesh-mail-send), never a seat. It alone of the module's parts reads an account's password (issue 1193);
 // the tools never send — they record the operator's Approve and queue it (issue 1194) on the work queue
-// mail-sends, which this process consumes through its one durable consumer. The per-message work (the
-// warrant verified again, the claim, the send itself, its outcome) is issues 1196 to 1198; this is the
-// skeleton that connects, consumes and stops cleanly (issue 1195).
+// mail-sends, which this process consumes through its one durable consumer (issue 1195). For each message
+// it verifies the warrant again and claims the send once (issue 1196), performs the one handover to the
+// server (issue 1197's classification), and records the outcome, retries a not-taken failure and emits the
+// result (issue 1198). The tell to the operator and the overdue health check are issue 1199's.

 import (
 	"context"
@@ -14,6 +15,7 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
+	"time"

 	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/process"
@@ -84,8 +86,22 @@ func send(env Env) int {
 	}
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
+	conn, err := process.Connect(account)
+	if err != nil {
+		logf("[mesh-mail] send: the bus cannot be reached: %v", err)
+		return 1
+	}
+	defer conn.Close()
+	h := &senderHandler{
+		Store:      processSends{state: conn.State(SendsState)},
+		Record:     routerRecordOf(conn),
+		Accounts:   func() ([]Account, error) { return ReadAccounts(env.Accounts) },
+		SecretsDir: env.Secrets,
+		Emit:       emitOf(conn),
+		Logf:       logf,
+	}
 	logf("[mesh-mail] send is consuming %s as %s", MailSendsStream, account.User)
-	if err := runSend(ctx, account, handleQueuedSend); err != nil && ctx.Err() == nil {
+	if err := conn.Consume(ctx, MailSendsStream, h.handle); err != nil && ctx.Err() == nil {
 		logf("[mesh-mail] send stopped: %v", err)
 		return 1
 	}
@@ -105,13 +121,111 @@ func runSend(ctx context.Context, account process.Account, handle func(*process.
 	return conn.Consume(ctx, MailSendsStream, handle)
 }

-// handleQueuedSend is the production handler: the warrant verified again and the claim (issue 1196), the
-// send itself (issue 1197) and its outcome, retry and the operator's tell (issue 1198) are not built yet.
-// Until then it terminates every message — never acknowledged blindly, which would silently drop a send
-// this process cannot yet act on, and never left to be redelivered forever either.
-func handleQueuedSend(msg *process.Message) {
-	logf("[mesh-mail] send received %s, not yet handled (novox/hq issue 1196)", msg.Subject)
-	if err := msg.Term(); err != nil {
-		logf("[mesh-mail] send: terminating %s: %v", msg.Subject, err)
+// emitOf is the real Emit (send.go), wrapping conn's own Emit with the given id as both x-event-id and the
+// bus's own dedup id (novox/hq ADR 0436 clauses 4, 11).
+func emitOf(conn *process.Conn) Emit {
+	return func(eventType, id string, body any) error {
+		return conn.Emit(eventType, body, stdio.EmitOptions{MessageID: id})
 	}
 }
+
+// senderHandler is everything handleQueuedSend needs: the `sends` state and the router's record, both
+// reached under the sender's own account, the accounts setting (to find the server to connect to) and the
+// secrets directory (to read the one password this process holds), and how it sends and emits — real
+// unless a test gives its own (novox/hq ADR 0436 §7, issues 1196 to 1198).
+type senderHandler struct {
+	Store      SendStore
+	Record     RouterRecord
+	Accounts   func() ([]Account, error)
+	SecretsDir string
+	Dial       SMTPDialer
+	Emit       Emit
+	Now        func() time.Time
+	Logf       func(format string, a ...any)
+}
+
+func (h *senderHandler) now() time.Time {
+	if h.Now != nil {
+		return h.Now()
+	}
+	return time.Now()
+}
+
+func (h *senderHandler) logf(format string, a ...any) {
+	if h.Logf != nil {
+		h.Logf(format, a...)
+	}
+}
+
+// handle is the handler mesh-mail send's Consume hands every message to mail-sends (novox/hq ADR 0436
+// clauses 9, 10, 11): the warrant verified again and the send claimed (claimQueuedSend, send.go), the one
+// handover to the server attempted, and the outcome recorded, retried or emitted (recordOutcome, send.go) —
+// then, and only then, the message is acknowledged, asked redelivered, or terminated, exactly as that
+// outcome says.
+func (h *senderHandler) handle(msg *process.Message) {
+	var q QueuedSend
+	if err := json.Unmarshal(msg.Body, &q); err != nil {
+		h.logf("[mesh-mail] send: a queued message on %s could not be read: %v", msg.Subject, err)
+		if err := msg.Term(); err != nil {
+			h.logf("[mesh-mail] send: terminating %s: %v", msg.Subject, err)
+		}
+		return
+	}
+	req, claimed, rev, outcome, err := claimQueuedSend(h.Store, h.Record, q.Send, h.now())
+	if err != nil {
+		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:
+		return // another delivery of the same message already owns it.
+	case claimRefused, claimNotProven:
+		if err := msg.Term(); err != nil {
+			h.logf("[mesh-mail] send: terminating %s: %v", q.Send, err)
+		}
+		return
+	}
+	sendErr := h.attempt(req)
+	_, action, delay := recordOutcome(h.Store, h.Emit, req, claimed, rev, sendErr, h.now())
+	switch action {
+	case actionAck:
+		if err := msg.Ack(); err != nil {
+			h.logf("[mesh-mail] send: acknowledging %s: %v", q.Send, err)
+		}
+	case actionRetry:
+		if err := msg.NakWithDelay(delay); err != nil {
+			h.logf("[mesh-mail] send: asking %s redelivered: %v", q.Send, err)
+		}
+	case actionTerm:
+		if err := msg.Term(); err != nil {
+			h.logf("[mesh-mail] send: terminating %s: %v", q.Send, err)
+		}
+	}
+}
+
+// attempt is the one handover to req.Account's outgoing server (novox/hq ADR 0436 clause 10): it reads the
+// accounts setting and the account's own password fresh, never from the queue's message, and every way it
+// cannot even be attempted (the account vanished, its password not given, the setting unreadable) is a
+// *SendError classified NotTaken, exactly as SendMail's own would be — safe to retry.
+func (h *senderHandler) attempt(req SendRequest) error {
+	accounts, err := h.Accounts()
+	if err != nil {
+		return &SendError{Outcome: NotTaken, Err: fmt.Errorf("the accounts setting cannot be read: %w", err)}
+	}
+	acc, found := accountByName(accounts, req.Account)
+	if !found {
+		return &SendError{Outcome: NotTaken, Err: fmt.Errorf("the account %s no longer exists", req.Account)}
+	}
+	pw, given, err := secret.Read(h.SecretsDir, acc.Secret())
+	if err != nil {
+		return &SendError{Outcome: NotTaken, Err: fmt.Errorf("the account %s's password cannot be read: %w", acc.Name, err)}
+	}
+	if !given {
+		return &SendError{Outcome: NotTaken, Err: fmt.Errorf("the account %s has no password yet; the operator gives it at the desk", acc.Name)}
+	}
+	dial := h.Dial
+	if dial == nil {
+		dial = DialSMTP
+	}
+	return SendMail(dial, acc, pw, req.To, req.Subject, req.Body, h.now(), req.ID)
+}