mesh-mail: fix the work-queue stream's real bus name (MAIL_SENDS, not MAIL-SENDS) and build the integration tests' stream, subject and duplicate window exactly as the controller composes them, exercising the send id's dedup (review round 1 finding 7; issue 1198)

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
 }
 

modules/mesh-mail/cmd/mesh-mail/sender.go

@@ -27,8 +27,13 @@ import (
 // its own bus account as the own secret broker-<SendProcessID> (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):

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 {

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