diff --git a/modules/mesh-mail/cmd/mesh-mail/acceptance_integration_test.go b/modules/mesh-mail/cmd/mesh-mail/acceptance_integration_test.go index a499ed6d..11523fad 100644 --- a/modules/mesh-mail/cmd/mesh-mail/acceptance_integration_test.go +++ b/modules/mesh-mail/cmd/mesh-mail/acceptance_integration_test.go @@ -97,8 +97,7 @@ func newAcceptanceBus(t *testing.T) (*acceptanceBus, string) { 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 { + if _, err := js.CreateStream(ctx, mailSendsStreamConfig()); err != nil { t.Fatal(err) } if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod." + Module + ".event.>"}}); err != nil { @@ -125,12 +124,16 @@ func queueAnApprovedSend(t *testing.T, bus *acceptanceBus, account string, appro if _, err := bus.asks.Put(ctx, Module+"."+req.ID, raw); err != nil { t.Fatalf("writing the router's record: %v", err) } + // The tools' own publish path (send.go's QueuePublish, wired for real to stdio.Emit/sdkQueuePublish): + // the real wire subject the controller composes for mail-sends' one event (mailSendApprovedSubject), + // the send id as the message id, so the stream's own duplicate window is the one actually exercised + // (novox/hq review round 1, finding 7). 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)) + _, err = bus.js.Publish(ctx, mailSendApprovedSubject, data, jetstream.WithMsgID(id)) return err } if _, err := Decide(store, publish, w, approvedAt); err != nil { @@ -140,6 +143,13 @@ func queueAnApprovedSend(t *testing.T, bus *acceptanceBus, account string, appro if err != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != "queued" { t.Fatalf("not left queued: %v %v %+v", err, ok, kept.Decided) } + // Published a second time under the same message id, exactly as the sweep would for a record it finds + // still queued past its grace (Sweep, send.go): the stream's 24-hour duplicate window keeps it one + // message (novox/hq review round 1, finding 7) — the sender's own claim (clause 9) is what then keeps + // it one send, proved by this test's own "sent once" assertion below. + if err := publish(req.ID, QueuedSend{Send: req.ID, Account: req.Account, Ask: req.ID}); err != nil { + t.Fatalf("the duplicate publish: %v", err) + } return req, req.ID } diff --git a/modules/mesh-mail/cmd/mesh-mail/sender.go b/modules/mesh-mail/cmd/mesh-mail/sender.go index f92abfae..a424be1c 100644 --- a/modules/mesh-mail/cmd/mesh-mail/sender.go +++ b/modules/mesh-mail/cmd/mesh-mail/sender.go @@ -27,8 +27,13 @@ import ( // its own bus account as the own secret broker- (novox/hq ADR 0436 §2). const SendProcessID = "send" -// MailSendsStream is the work-queue stream mail-sends carries on the bus (novox/hq ADR 0436 §8). -const MailSendsStream = "MAIL-SENDS" +// MailSendsStream is the work-queue stream mail-sends carries on the bus (novox/hq ADR 0436 §8): the +// manifest's local name "mail-sends", capitalised with its hyphen an underscore, exactly as the controller +// composes a stream's own bus name (mesh-controller internal/catalogue/streams.go StreamBusName, +// strings.ToUpper(strings.ReplaceAll(local, "-", "_")); confirmed against its own fixture, mesh-controller +// cmd/mesh-controller/process_live_test.go's mailSends, which names it "MAIL_SENDS" — review round 1, +// finding 7). +const MailSendsStream = "MAIL_SENDS" // MessengerAsksBucket is the operator channel's own records bucket, as the controller's own fixtures name // it (mesh-controller cmd/mesh-controller/process_live_test.go, internal/broker/process_grants_test.go): diff --git a/modules/mesh-mail/cmd/mesh-mail/sender_integration_test.go b/modules/mesh-mail/cmd/mesh-mail/sender_integration_test.go index d36ee676..e4f9a3ea 100644 --- a/modules/mesh-mail/cmd/mesh-mail/sender_integration_test.go +++ b/modules/mesh-mail/cmd/mesh-mail/sender_integration_test.go @@ -18,6 +18,38 @@ import ( "github.com/nats-io/nats.go/jetstream" ) +// mailSendApprovedSubject is the real wire subject the controller composes for mesh-mail's own stream +// mail-sends' one event, mail.send.approved: "mesh.stream." + module + ".event." + event (mesh-controller +// internal/broker/modstreams.go ModuleStreamSubject, moduleStreamPrefix). The sender's own Consume binds +// by stream and consumer name alone (mesh-sdk go/process queue.go), so this subject matters only to how a +// test builds the stream and publishes to it — it must match what the controller actually composes, or an +// integration test proves nothing about the real path (novox/hq review round 1, finding 7). +const mailSendApprovedSubject = "mesh.stream." + Module + ".event." + MailSendApproved + +// mailSendsStreamConfig is the work-queue stream mail-sends, composed exactly as the controller would +// (mesh-controller internal/catalogue/streams.go StreamDeclaration.Stream and +// internal/broker/modstreams.go ModuleStream.Stream): retention work-queue, discard new, module.json's +// max-age-days 1 and max-mib 64, and its "duplicates": 24 — in hours (novox/hq ADR 0436 clause 6) — which +// the controller turns into DuplicatesSeconds = Duplicates * 60 * 60, 24 hours here. +func mailSendsStreamConfig() jetstream.StreamConfig { + return jetstream.StreamConfig{ + Name: MailSendsStream, + Subjects: []string{mailSendApprovedSubject}, + Retention: jetstream.WorkQueuePolicy, + Discard: jetstream.DiscardNew, + MaxAge: 24 * time.Hour, + MaxBytes: 64 << 20, + Duplicates: 24 * time.Hour, + } +} + +// mailSendsConsumerConfig is the one durable consumer the controller composes for a process's bus.queues +// (mesh-controller internal/broker/modstreams.go QueueConsumer): AckWaitSeconds 30, MaxDeliver from the +// stream's own max-deliveries (module.json: 5, MostAttempts here). +func mailSendsConsumerConfig(durable string) jetstream.ConsumerConfig { + return jetstream.ConsumerConfig{Durable: durable, AckPolicy: jetstream.AckExplicitPolicy, AckWait: 30 * time.Second, MaxDeliver: MostAttempts} +} + // newTestBus starts a JetStream-enabled nats-server for the test alone, on a random port with its own temp // store (modules/mesh-delivery/cmd/mesh-delivery/bus_test.go:52-58). func newTestBus(t *testing.T) string { @@ -54,14 +86,12 @@ func TestSendConsumesMailSendsAndStopsCleanlyOnSignal(t *testing.T) { 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 { + if _, err := js.CreateStream(ctx, mailSendsStreamConfig()); err != nil { t.Fatal(err) } account := process.Account{URL: url, Node: "g14", Module: Module, User: "g14." + Module + "." + SendProcessID, Password: "pw"} - if _, err := js.CreateConsumer(ctx, MailSendsStream, jetstream.ConsumerConfig{ - Durable: account.QueueConsumerName(), AckPolicy: jetstream.AckExplicitPolicy, AckWait: 30 * time.Second, MaxDeliver: 5}); err != nil { + if _, err := js.CreateConsumer(ctx, MailSendsStream, mailSendsConsumerConfig(account.QueueConsumerName())); err != nil { t.Fatal(err) } @@ -70,20 +100,33 @@ func TestSendConsumesMailSendsAndStopsCleanlyOnSignal(t *testing.T) { done := make(chan error, 1) go func() { done <- runSend(runCtx, account, func(m *process.Message) { received <- m }) }() - if _, err := js.Publish(ctx, MailSendApproved, []byte(`{"send":"send-1","account":"mesh","ask":"send-1"}`)); err != nil { + body := []byte(`{"send":"send-1","account":"mesh","ask":"send-1"}`) + // Published twice under the same message id (the send id, as the tools' own publish path always uses, + // novox/hq ADR 0436 clause 8): the stream's own 24-hour duplicate window keeps it one message, proving + // the dedup this test's stream config actually carries (novox/hq review round 1, finding 7). + if _, err := js.Publish(ctx, mailSendApprovedSubject, body, jetstream.WithMsgID("send-1")); err != nil { + t.Fatal(err) + } + if _, err := js.Publish(ctx, mailSendApprovedSubject, body, jetstream.WithMsgID("send-1")); err != nil { t.Fatal(err) } select { case msg := <-received: - if msg.Subject != MailSendApproved { - t.Fatalf("received on %q, want %q", msg.Subject, MailSendApproved) + if msg.Subject != mailSendApprovedSubject { + t.Fatalf("received on %q, want %q", msg.Subject, mailSendApprovedSubject) } // Left exactly here, never acknowledged: the handler's own fate-deciding (Ack, NakWithDelay, Term) is // issues 1196-1198's, not this skeleton's. case <-time.After(10 * time.Second): t.Fatal("the handler never received the published message") } + select { + case <-received: + t.Fatal("the duplicate publish under the same message id was delivered a second time") + case <-time.After(500 * time.Millisecond): + // The duplicate was dropped by the stream's own window, as it should be. + } cancel() // the signal send() itself reacts to (SIGTERM, or ^C) select { diff --git a/modules/mesh-mail/cmd/mesh-mail/sender_test.go b/modules/mesh-mail/cmd/mesh-mail/sender_test.go index 57fb55c0..5df2f9cb 100644 --- a/modules/mesh-mail/cmd/mesh-mail/sender_test.go +++ b/modules/mesh-mail/cmd/mesh-mail/sender_test.go @@ -5,7 +5,6 @@ package main import ( "path/filepath" - "strings" "testing" ) @@ -73,7 +72,9 @@ func TestSendProcessIDNamesItsOwnAccountFile(t *testing.T) { if SendProcessID != "send" { t.Fatalf("SendProcessID: %q; module.json's resource id is \"send\"", SendProcessID) } - if !strings.HasSuffix(MailSendsStream, "MAIL-SENDS") { - t.Fatalf("MailSendsStream: %q", MailSendsStream) + // mesh-controller's own StreamBusName: "mail-sends" capitalised, its hyphen an underscore (review round + // 1, finding 7 — confirmed against cmd/mesh-controller/process_live_test.go's own fixture). + if MailSendsStream != "MAIL_SENDS" { + t.Fatalf("MailSendsStream: %q, want MAIL_SENDS", MailSendsStream) } }