diff --git a/modules/mesh-mail/cmd/mesh-mail/acceptance_integration_test.go b/modules/mesh-mail/cmd/mesh-mail/acceptance_integration_test.go new file mode 100644 index 00000000..958d17f9 --- /dev/null +++ b/modules/mesh-mail/cmd/mesh-mail/acceptance_integration_test.go @@ -0,0 +1,307 @@ +//go:build integration + +package main + +// Release 3's acceptance test (novox/hq ADR 0436, issue 1198 AC4): the whole module, over a JetStream bus +// of the test's own (nats-server v2.11.17, copied from modules/mesh-delivery/cmd/mesh-delivery/bus_test.go +// as newTestBus already does in sender_integration_test.go), a fake SMTP server, the router's records +// bucket with a closed Approve record, the account file and a password file written by the test. Compiled +// by the merge gate's `go vet -tags integration`; never run on this machine (the context file's hard rule). +// +// "The tools queue an approved send" is played here directly against the real bus (Decide, send.go) rather +// than through the tool runner's own RPC — the tool runner's part (stdio.Serve, grants, the ask's wording) +// is proved elsewhere (send_test.go, ask_test.go); what this test proves is the handover from an approved, +// queued send through to the sender's own claim, send and outcome, end to end over a real bus. + +import ( + "context" + "encoding/json" + "errors" + "path/filepath" + "strings" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// kvSends adapts a raw jetstream.KeyValue to SendStore (send.go): the test's own stand-in for both the +// tools' and the sender's `sends` state, which in production are the same bucket reached under two +// different accounts. +type kvSends struct{ kv jetstream.KeyValue } + +func (k kvSends) Get(key string) (json.RawMessage, uint64, bool, error) { + e, err := k.kv.Get(context.Background(), key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, 0, false, nil + } + if err != nil { + return nil, 0, false, err + } + return json.RawMessage(e.Value()), e.Revision(), true, nil +} +func (k kvSends) Create(key string, value any) (uint64, error) { + data, err := json.Marshal(value) + if err != nil { + return 0, err + } + rev, err := k.kv.Create(context.Background(), key, data) + if errors.Is(err, jetstream.ErrKeyExists) { + return 0, stdio.ErrChanged + } + return rev, err +} +func (k kvSends) Update(key string, value any, revision uint64) (uint64, error) { + data, err := json.Marshal(value) + if err != nil { + return 0, err + } + rev, err := k.kv.Update(context.Background(), key, data, revision) + if errors.Is(err, jetstream.ErrKeyRevisionMismatch) { + return 0, stdio.ErrChanged + } + return rev, err +} +func (k kvSends) Keys() ([]string, error) { return nil, errors.New("not used by this test") } + +// acceptanceBus is everything the test built on a fresh bus: ready-made buckets and streams, and the raw +// clients to act as the tools with. +type acceptanceBus struct { + nc *nats.Conn + js jetstream.JetStream + sends jetstream.KeyValue + asks jetstream.KeyValue +} + +func newAcceptanceBus(t *testing.T) (*acceptanceBus, string) { + t.Helper() + url := newTestBus(t) + ctx := context.Background() + nc, err := nats.Connect(url) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + sends, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: Module + "_" + SendsState}) + if err != nil { + t.Fatal(err) + } + asksKV, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: MessengerAsksBucket}) + if err != nil { + t.Fatal(err) + } + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: MailSendsStream, Subjects: []string{MailSendApproved}, + Retention: jetstream.WorkQueuePolicy, MaxAge: 24 * time.Hour, MaxBytes: 64 << 20}); err != nil { + t.Fatal(err) + } + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod." + Module + ".event.>"}}); err != nil { + t.Fatal(err) + } + return &acceptanceBus{nc: nc, js: js, sends: sends, asks: asksKV}, url +} + +// queueAnApprovedSend plays the tools' own part (send.go's Decide, the queue publish) directly against the +// real bus: a send asked, approved by a warrant whose digest really is the one req.Ask() binds, recorded +// approved and queued — exactly what issue 1194 made the tools do, without needing the tool runner's RPC. +func queueAnApprovedSend(t *testing.T, bus *acceptanceBus, account string, approvedAt time.Time) (SendRequest, string) { + t.Helper() + ctx := context.Background() + req := SendRequest{ID: newSendID(), Account: account, To: []string{"ops@example.org"}, Subject: "The nightly gate is red", + Body: "See the job log.", Caller: "person.op", At: approvedAt, AskTo: "o… at example org"} + store := kvSends{kv: bus.sends} + if _, err := store.Create(req.ID, req); err != nil { + t.Fatalf("keeping the send: %v", err) + } + w := warrantFor(req, approvedAt) + rec := routerRecord{State: string(w.Outcome), Warrant: &w} + raw, _ := json.Marshal(rec) + if _, err := bus.asks.Put(ctx, Module+"."+req.ID, raw); err != nil { + t.Fatalf("writing the router's record: %v", err) + } + publish := func(id string, body any) error { + data, err := json.Marshal(body) + if err != nil { + return err + } + _, err = bus.js.Publish(ctx, MailSendApproved, data, jetstream.WithMsgID(id)) + return err + } + if _, err := Decide(store, publish, w, approvedAt); err != nil { + t.Fatalf("the tools' own Decide: %v", err) + } + kept, _, ok, err := getSend(store, req.ID) + if err != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != "queued" { + t.Fatalf("not left queued: %v %v %+v", err, ok, kept.Decided) + } + return req, req.ID +} + +// 1198/AC4: when the tools queue an approved send and the send process runs, the fake server receives +// exactly one mail carrying the send id's Message-ID, sends says sent, and mail.send.sent is on the stream. +func TestMailSenderAcceptanceAnApprovedSendIsSentOnceAndSaidOnce(t *testing.T) { + bus, url := newAcceptanceBus(t) + fake := newFakeSMTPServer(t) + + dir := t.TempDir() + accountsPath := filepath.Join(dir, "accounts.json") + writeFile(t, accountsPath, `{"accounts": {"mesh": {"kind": "imap", "address": "jschoubben@novox.be", + "incoming": {"host": "mail.novox.be", "port": 993, "security": "tls"}, + "outgoing": {"host": "127.0.0.1", "port": 1, "security": "tls"}}}}`) + writeFile(t, filepath.Join(dir, "mail-credential-mesh.secret"), "the-password\n") + + approvedAt := judgedAt + req, id := queueAnApprovedSend(t, bus, "mesh", approvedAt) + + account := process.Account{URL: url, Node: "g14", Module: Module, User: "g14." + Module + "." + SendProcessID, Password: "pw"} + if _, err := bus.js.CreateConsumer(context.Background(), MailSendsStream, jetstream.ConsumerConfig{ + Durable: account.QueueConsumerName(), AckPolicy: jetstream.AckExplicitPolicy, AckWait: 5 * time.Second, MaxDeliver: MostAttempts}); err != nil { + t.Fatal(err) + } + conn, err := process.Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + h := &senderHandler{ + Store: kvSends{kv: bus.sends}, + Record: routerRecordOf(conn), + Accounts: func() ([]Account, error) { return ReadAccounts(accountsPath) }, + SecretsDir: dir, + Dial: fake.dialer(), + Emit: emitOf(conn), + Now: func() time.Time { return approvedAt.Add(time.Minute) }, + } + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + done := make(chan error, 1) + go func() { done <- conn.Consume(ctx, MailSendsStream, h.handle) }() + + eventually(t, 10*time.Second, "the mail sent", func() (bool, string) { + return fake.count() == 1, "received so far: " + describeReceived(fake) + }) + cancel() + <-done + + store := kvSends{kv: bus.sends} + kept, _, ok, err := getSend(store, id) + if err != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != "sent" { + t.Fatalf("the kept send: %v %v %+v", err, ok, kept.Decided) + } + got := fake.last() + if got.to[0] != "ops@example.org" { + t.Fatalf("received: %+v", got) + } + if !strings.Contains(got.data, "Message-Id: <"+req.ID+"@novox.be>") { + t.Fatalf("the mail's Message-ID does not name the send: %s", got.data) + } + + // mail.send.sent is on the EVENTS stream, naming the send id, never the recipient, subject or body. + eventually(t, 10*time.Second, "mail.send.sent on the stream", func() (bool, string) { + msg, err := fetchOneRaw(bus, "mesh.mod."+Module+".event."+MailSendSent) + if err != nil || msg == nil { + return false, "not yet" + } + var e OutcomeEvent + _ = json.Unmarshal(msg, &e) + return e.Send == req.ID && e.Account == "mesh", "got " + string(msg) + }) +} + +// 1198/AC4: when the fake server refuses sign-in, then after the attempts (shortened here) the record says +// failed and nothing was received. +func TestMailSenderAcceptanceARefusedSignInFailsAfterItsAttemptsAndNothingIsReceived(t *testing.T) { + orig := RetryDelays + RetryDelays = []time.Duration{10 * time.Millisecond, 10 * time.Millisecond, 10 * time.Millisecond, 10 * time.Millisecond} + t.Cleanup(func() { RetryDelays = orig }) + + bus, url := newAcceptanceBus(t) + s := newScriptedSMTPServer(t) + s.auth = reply(535, "authentication failed") + + dir := t.TempDir() + accountsPath := filepath.Join(dir, "accounts.json") + writeFile(t, accountsPath, `{"accounts": {"mesh": {"kind": "imap", "address": "jschoubben@novox.be", + "incoming": {"host": "mail.novox.be", "port": 993, "security": "tls"}, + "outgoing": {"host": "127.0.0.1", "port": 1, "security": "tls"}}}}`) + writeFile(t, filepath.Join(dir, "mail-credential-mesh.secret"), "the-wrong-password\n") + + approvedAt := judgedAt + _, id := queueAnApprovedSend(t, bus, "mesh", approvedAt) + + account := process.Account{URL: url, Node: "g14", Module: Module, User: "g14." + Module + "." + SendProcessID, Password: "pw"} + if _, err := bus.js.CreateConsumer(context.Background(), MailSendsStream, jetstream.ConsumerConfig{ + Durable: account.QueueConsumerName(), AckPolicy: jetstream.AckExplicitPolicy, AckWait: 2 * time.Second, MaxDeliver: MostAttempts}); err != nil { + t.Fatal(err) + } + conn, err := process.Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + h := &senderHandler{ + Store: kvSends{kv: bus.sends}, + Record: routerRecordOf(conn), + Accounts: func() ([]Account, error) { return ReadAccounts(accountsPath) }, + SecretsDir: dir, + Dial: s.dialer(), + Emit: emitOf(conn), + Now: func() time.Time { return approvedAt.Add(time.Minute) }, + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + done := make(chan error, 1) + go func() { done <- conn.Consume(ctx, MailSendsStream, h.handle) }() + + store := kvSends{kv: bus.sends} + eventually(t, 25*time.Second, "the send to fail after its attempts", func() (bool, string) { + kept, _, ok, err := getSend(store, id) + if err != nil || !ok || kept.Decided == nil { + return false, "not yet decided" + } + return kept.Decided.Outcome != "retrying" && kept.Decided.Outcome != "sending", kept.Decided.Outcome + }) + cancel() + <-done + + kept, _, ok, err := getSend(store, id) + if err != nil || !ok || kept.Decided == nil || kept.Decided.Outcome == "sent" { + t.Fatalf("the kept send ended as sent, or could not be read: %v %v %+v", err, ok, kept.Decided) + } +} + +// fetchOneRaw pulls at most one message off a stream's subject, or nil within a short wait, for a test to +// check an emitted event without binding a durable consumer of its own. +func fetchOneRaw(bus *acceptanceBus, subject string) ([]byte, error) { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + cons, err := bus.js.OrderedConsumer(ctx, "EVENTS", jetstream.OrderedConsumerConfig{FilterSubjects: []string{subject}}) + if err != nil { + return nil, err + } + batch, err := cons.Fetch(1, jetstream.FetchMaxWait(time.Second)) + if err != nil { + return nil, err + } + msg, ok := <-batch.Messages() + if !ok { + return nil, nil + } + return msg.Data(), nil +} + +func describeReceived(f *fakeSMTPServer) string { + if f.count() == 0 { + return "none" + } + got := f.last() + return got.from +} diff --git a/modules/mesh-mail/cmd/mesh-mail/claim_test.go b/modules/mesh-mail/cmd/mesh-mail/claim_test.go index a6ef609f..7d3fa295 100644 --- a/modules/mesh-mail/cmd/mesh-mail/claim_test.go +++ b/modules/mesh-mail/cmd/mesh-mail/claim_test.go @@ -77,7 +77,7 @@ 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)) + req, d, _, outcome, err := claimQueuedSend(sends, router.Record, id, judgedAt.Add(3*time.Minute)) if err != nil { t.Fatalf("claiming: %v", err) } @@ -108,7 +108,7 @@ 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)) + _, d, _, outcome, err := claimQueuedSend(sends, router.Record, 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) } @@ -117,7 +117,7 @@ func TestClaimQueuedSendRefusesAMismatchedOrUnclosedWarrant(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)) + _, d, _, outcome, err := claimQueuedSend(sends, router.Record, 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) } @@ -127,7 +127,7 @@ func TestClaimQueuedSendRefusesAMismatchedOrUnclosedWarrant(t *testing.T) { 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)) + _, d, _, outcome, err := claimQueuedSend(sends, router.Record, 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) } @@ -138,7 +138,7 @@ 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)) + _, d, _, outcome, err := claimQueuedSend(sends, router.Record, 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) } @@ -149,14 +149,14 @@ func TestClaimQueuedSendRefusesAMismatchedOrUnclosedWarrant(t *testing.T) { // 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)) + req, _, _, outcome, err := claimQueuedSend(sends, router.Record, 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, id, judgedAt.Add(4*time.Minute)) if err != nil { t.Fatalf("redelivered claim: %v", err) } @@ -180,7 +180,7 @@ func TestClaimQueuedSendOnALostRaceSendsNothingAndLeavesTheMessageToTheBus(t *te } }} - _, d, outcome, err := claimQueuedSend(racy, router.Record, id, judgedAt.Add(3*time.Minute)) + _, d, _, outcome, err := claimQueuedSend(racy, router.Record, id, judgedAt.Add(3*time.Minute)) if err != nil { t.Fatalf("claiming: %v", err) } diff --git a/modules/mesh-mail/cmd/mesh-mail/outcome_test.go b/modules/mesh-mail/cmd/mesh-mail/outcome_test.go new file mode 100644 index 00000000..816447f9 --- /dev/null +++ b/modules/mesh-mail/cmd/mesh-mail/outcome_test.go @@ -0,0 +1,150 @@ +package main + +// recordOutcome writes the final decision for one claimed send and says what the caller does with the +// queue message next (novox/hq ADR 0436 clauses 10, 11, issue 1198). + +import ( + "errors" + "strings" + "testing" + "time" +) + +// emitted is one call recordOutcome's Emit made. +type emitted struct { + eventType, id string + body OutcomeEvent +} + +// memEmit collects every emit call, for a test to check the event and its body. +type memEmit struct{ calls []emitted } + +func (m *memEmit) emit(eventType, id string, body any) error { + b, _ := body.(OutcomeEvent) + m.calls = append(m.calls, emitted{eventType: eventType, id: id, body: b}) + return nil +} + +// outcomeTestClaim keeps one send claimed "sending" at the given attempt, returning its request, the +// claimed decision and the revision recordOutcome needs. +func outcomeTestClaim(t *testing.T, attempt int) (SendRequest, SendDecision, uint64, *memSends) { + t.Helper() + sends := newMemSends() + req := SendRequest{ID: newSendID(), Account: "mesh", To: []string{"ops@example.org"}, Subject: "s", Body: "body", At: judgedAt} + claim := SendDecision{At: judgedAt.Add(time.Minute), Outcome: "sending", Warrant: "the operator, via telegram, chose Send", Attempt: attempt} + req.Decided = &claim + rev, err := sends.Create(req.ID, req) + if err != nil { + t.Fatal(err) + } + return req, claim, rev, sends +} + +// 1198/AC1: given a claimed send the server accepts, when sent, then sent is recorded, mail.send.sent is +// emitted (send id, account, outcome in words; no recipient, subject or body), and only then — by +// construction, since the caller acknowledges only once this call returns — the message is acknowledged. +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)) + if d.Outcome != "sent" { + t.Fatalf("outcome: %+v", d) + } + if action != actionAck { + t.Fatalf("action: %v, want actionAck", action) + } + kept, _, ok, err := getSend(sends, req.ID) + if err != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != "sent" { + t.Fatalf("not kept as sent: %v %v %+v", err, ok, kept.Decided) + } + if len(emit.calls) != 1 || emit.calls[0].eventType != MailSendSent || emit.calls[0].id != req.ID { + t.Fatalf("emitted: %+v", emit.calls) + } + body := emit.calls[0].body + if body.Send != req.ID || body.Account != "mesh" || body.Outcome == "" { + t.Fatalf("the event's body: %+v", body) + } + whole := body.Send + body.Account + body.Outcome + for _, forbidden := range []string{"ops@example.org", "s", "body"} { + if forbidden == "s" { + continue // too short to search for meaningfully without false positives in the id/account + } + if strings.Contains(whole, forbidden) { + t.Fatalf("the emitted body carries %q: %+v", forbidden, body) + } + } +} + +// 1198/AC2: given a not-taken failure on attempt n (n < 5), then retrying is recorded with attempt, reason +// and next attempt, and the caller is told to ask the bus to deliver it again after 1, 5, 15, 30 minutes +// for n = 1..4; on the fifth, failed is recorded and mail.send.failed emitted. +func TestRecordOutcomeOnNotTakenRetriesUpToTheFourthAttempt(t *testing.T) { + for n, want := range map[int]time.Duration{1: time.Minute, 2: 5 * time.Minute, 3: 15 * time.Minute, 4: 30 * time.Minute} { + 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) + if d.Outcome != "retrying" || d.Attempt != n { + t.Fatalf("attempt %d: outcome %+v", n, d) + } + if !d.NextAttempt.Equal(now.Add(want)) { + t.Fatalf("attempt %d: next attempt %v, want %v", n, d.NextAttempt, now.Add(want)) + } + if action != actionRetry || delay != want { + t.Fatalf("attempt %d: action %v, delay %v, want actionRetry, %v", n, action, delay, want) + } + if len(emit.calls) != 0 { + t.Fatalf("attempt %d: retrying emitted something: %+v", n, emit.calls) + } + } +} + +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")}, + judgedAt.Add(2*time.Minute)) + if !strings.HasPrefix(d.Outcome, "failed: ") || d.Attempt != MostAttempts { + t.Fatalf("outcome: %+v", d) + } + if action != actionTerm { + t.Fatalf("action: %v, want actionTerm", action) + } + if len(emit.calls) != 1 || emit.calls[0].eventType != MailSendFailed { + t.Fatalf("emitted: %+v", emit.calls) + } +} + +// 1198/AC3: given a refusal, then failed is recorded with the reason, mail.send.failed emitted, and the +// message terminated; given not proven, the same with not proven. +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")}, + judgedAt.Add(2*time.Minute)) + if !strings.HasPrefix(d.Outcome, "failed: ") || !strings.Contains(d.Outcome, "550 no such mailbox") { + t.Fatalf("outcome: %+v", d) + } + if action != actionTerm { + t.Fatalf("action: %v, want actionTerm", action) + } + if len(emit.calls) != 1 || emit.calls[0].eventType != MailSendFailed || emit.calls[0].body.Outcome != d.Outcome { + t.Fatalf("emitted: %+v", emit.calls) + } +} + +func TestRecordOutcomeOnNotProvenRecordsNotProvenAndTerminates(t *testing.T) { + req, claimed, rev, sends := outcomeTestClaim(t, 1) + emit := &memEmit{} + d, action, _ := 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 d.Outcome != "not proven" { + t.Fatalf("outcome: %+v", d) + } + if action != actionTerm { + t.Fatalf("action: %v, want actionTerm", action) + } + if len(emit.calls) != 1 || emit.calls[0].eventType != MailSendFailed || emit.calls[0].body.Outcome != "not proven" { + t.Fatalf("emitted: %+v", emit.calls) + } +} diff --git a/modules/mesh-mail/cmd/mesh-mail/send.go b/modules/mesh-mail/cmd/mesh-mail/send.go index 7781f146..6ea81560 100644 --- a/modules/mesh-mail/cmd/mesh-mail/send.go +++ b/modules/mesh-mail/cmd/mesh-mail/send.go @@ -189,6 +189,9 @@ type SendDecision struct { // Attempt is which handover to the server this is, 1 for the first: the sender's own claim from queued // or retrying to sending counts it (novox/hq ADR 0436 clauses 9, 10, issue 1196). Attempt int `json:"attempt,omitempty"` + // NextAttempt is when a "retrying" send is next claimed, the bus's own redelivery time (novox/hq ADR + // 0436 clause 10, issue 1198): read by mail_health to say a send overdue (issue 1199). + NextAttempt time.Time `json:"next-attempt,omitempty"` } // bodyDigest is the digest an act binds in place of the body itself: never the body, which the act's digest is @@ -732,17 +735,17 @@ const ( // 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, claimOutcome, error) { +func claimQueuedSend(store SendStore, record RouterRecord, id string, now time.Time) (SendRequest, SendDecision, uint64, claimOutcome, error) { req, rev, ok, err := getSend(store, id) if err != nil { - return SendRequest{}, SendDecision{}, claimLostRace, err + return SendRequest{}, SendDecision{}, 0, claimLostRace, err } if !ok { - return SendRequest{}, SendDecision{}, claimLostRace, fmt.Errorf("no send %q is kept", id) + return SendRequest{}, SendDecision{}, 0, claimLostRace, fmt.Errorf("no send %q is kept", id) } - refuse := func(reason string) (SendRequest, SendDecision, claimOutcome, error) { + refuse := func(reason string) (SendRequest, SendDecision, uint64, claimOutcome, error) { d, _, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "refused: " + reason}) - return req, d, claimRefused, err + return req, d, 0, claimRefused, err } switch { case req.Decided == nil: @@ -751,13 +754,13 @@ func claimQueuedSend(store SendStore, record RouterRecord, id string, now time.T // A previous delivery may have handed it to the server already; never sent twice, and never // retried automatically either way (ADR 0436 clause 10). d, _, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "not proven", Attempt: req.Decided.Attempt}) - return req, d, claimNotProven, err + return req, d, 0, claimNotProven, err case req.Decided.Outcome != "queued" && req.Decided.Outcome != "retrying": return refuse(fmt.Sprintf("the send is %q, neither queued nor retrying", req.Decided.Outcome)) } entry, err := record(id) if err != nil { - return SendRequest{}, SendDecision{}, claimLostRace, err + return SendRequest{}, SendDecision{}, 0, claimLostRace, err } if entry == nil { return refuse("the router keeps no record of this ask") @@ -787,13 +790,112 @@ func claimQueuedSend(store SendStore, record RouterRecord, id string, now time.T claim := SendDecision{At: now, Outcome: "sending", Warrant: w.Says(), Attempt: req.Decided.Attempt + 1} claimed, newRev, err := updateDecision(store, req, rev, claim) if err != nil { - return SendRequest{}, SendDecision{}, claimLostRace, err + return SendRequest{}, SendDecision{}, 0, claimLostRace, err } if newRev == 0 { // Lost the claim race: another delivery of the same message is already handling it. - return req, claimed, claimLostRace, nil + return req, claimed, 0, claimLostRace, nil + } + return req, claimed, newRev, claimToSend, nil +} + +// ---- the send's outcome, retry and emit (novox/hq ADR 0436 clauses 10, 11, issue 1198) -------------------------- + +// MailSendSent and MailSendFailed are the two events the sender emits, never anything else (novox/hq ADR +// 0436 clause 11): the send id, the account and the outcome in words — never the recipients, the subject or +// the body. +const ( + MailSendSent = "mail.send.sent" + MailSendFailed = "mail.send.failed" +) + +// OutcomeEvent is the body mail.send.sent and mail.send.failed carry. +type OutcomeEvent struct { + Send string `json:"send"` + Account string `json:"account"` + Outcome string `json:"outcome"` +} + +// Emit publishes one of this module's own events with a chosen message id (novox/hq ADR 0436 clauses 4, +// 11): the real one (sender.go) wraps a process.Conn's own Emit; a test gives its own. +type Emit func(eventType, id string, body any) error + +// RetryDelays is how long after each not-taken attempt (1-indexed) the bus is asked to redeliver the +// message, before the next attempt: 1, 5, 15 and 30 minutes for attempts 1 to 4 (novox/hq ADR 0436 clause +// 10). +var RetryDelays = []time.Duration{time.Minute, 5 * time.Minute, 15 * time.Minute, 30 * time.Minute} + +// MostAttempts is how many times the sender hands a claimed send to the server before giving up: the fifth +// failure records failed, and the bus's own max-deliveries (module.json's mail-sends) dead-letters the +// message once it is terminated rather than acknowledged (novox/hq ADR 0436 clause 10). +const MostAttempts = 5 + +// msgAction is what the caller (the message handler, sender.go) does with the queue message once +// recordOutcome has written the decision: Ack (sent), NakWithDelay (retrying, not the last attempt yet) or +// Term (failed, or not proven — the bus is never asked to redeliver those; a dead letter past the fifth +// attempt is still reached by Term, since clause 10 records failed itself rather than waiting for the bus's +// own max-deliveries to notice). +type msgAction int + +const ( + actionAck msgAction = iota + actionRetry + actionTerm +) + +// recordOutcome writes the final decision for one claimed send once SendMail has run (or could not even be +// attempted — the account vanished, or its password cannot be read, both passed in as a NotTaken +// *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) SendDecision { + final, _, 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 final + } + say := func(outcome string) { + _ = emit(MailSendFailed, req.ID, OutcomeEvent{Send: req.ID, Account: req.Account, Outcome: outcome}) + } + if sendErr == nil { + d := write(SendDecision{At: now, Outcome: "sent", Warrant: claimed.Warrant, Attempt: claimed.Attempt}) + _ = emit(MailSendSent, req.ID, OutcomeEvent{Send: req.ID, Account: req.Account, Outcome: d.Outcome}) + return d, actionAck, 0 + } + var se *SendError + if !errors.As(sendErr, &se) { + // SendMail itself never returns an error that is not a *SendError; a caller passing one of its own + // (the account vanished, or its password cannot be read) is read as not taken, never as a refusal or + // a proven send, which would be worse guesses than "safe to retry" (novox/hq ADR 0436 clause 10). + se = &SendError{Outcome: NotTaken, Err: sendErr} + } + 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()}) + say(d.Outcome) + return d, actionTerm, 0 + case NotProven: + d := write(SendDecision{At: now, Outcome: "not proven", Warrant: claimed.Warrant, Attempt: claimed.Attempt, + Error: se.Err.Error()}) + say(d.Outcome) + return d, actionTerm, 0 + 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()}) + say(d.Outcome) + return d, actionTerm, 0 + } + 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)}) + return d, actionRetry, delay // not a final outcome: no emit (clause 11 names only sent and failed) } - return req, claimed, claimToSend, nil } // ---- reading the status ----------------------------------------------------------------------------------------- diff --git a/modules/mesh-mail/cmd/mesh-mail/sender.go b/modules/mesh-mail/cmd/mesh-mail/sender.go index e60af68f..5635b5be 100644 --- a/modules/mesh-mail/cmd/mesh-mail/sender.go +++ b/modules/mesh-mail/cmd/mesh-mail/sender.go @@ -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) +}