mesh-mail: a send process that connects on its own bus account and consumes the mail-sends queue, stopping cleanly on a signal (ADR 0436 clause 7; issue 1195)

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

@@ -9,12 +9,16 @@
 //	mesh-mail keep       acts: watches every account (IDLE on the inbox, a look at the other folders every five
 //	                     minutes), keeps the index and the attachment store, and writes each change's event beside it
 //	mesh-mail watch      judges: reads every minute how each account is, and leaves its verdict for mail_health
+//	mesh-mail send       the sender (novox/hq ADR 0436 §7): its own bus account, consuming the work queue
+//	                     mail-sends; it alone reads a password, and performs the one send an operator's
+//	                     Approve authorises, once the tools have recorded it approved and queued it
 //
 // The two long-running processes are the act and the judge, as mesh-knowledge's are: keep is judged by its unit, watch
-// by the tool mail_health (ADR 0359 §6, ADR 0240 §3 and §7). Only the bundle reaches the bus, through the runtime that
-// launched it, so the keeper hands its events over in the index, written in the same step as the change they say,
-// and the bundle says them and removes each once the bus took it. stdout is the bundle's MCP channel; what this module
-// says, it says on stderr.
+// by the tool mail_health (ADR 0359 §6, ADR 0240 §3 and §7); send is judged by its unit too (it holds a bus account
+// of its own, never a seat, so no tool answers for it). Only the bundle reaches the bus through the runtime that
+// launched it; send reaches it on its own account instead (mesh-sdk go/process), so the keeper hands its events over
+// in the index, written in the same step as the change they say, and the bundle says them and removes each once the
+// bus took it. stdout is the bundle's MCP channel; what this module says, it says on stderr.
 package main
 
 import (
@@ -64,6 +68,8 @@ func main() {
 			os.Exit(keep(envFromOS()))
 		case "watch":
 			os.Exit(watch(envFromOS()))
+		case "send":
+			os.Exit(send(envFromOS()))
 		}
 	}
 	serve(envFromOS())

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

@@ -40,11 +40,12 @@ func TestTheToolsBundleNeverImportsThePasswordPackage(t *testing.T) {
 	}
 }
 
-// 1193/AC2: keep (imap.go) reads the password through the moved package (the sender, issue 1195's sender.go,
-// joins this list once it exists); SecretGiven (stat only) stays in config.go, available to the tools, and
-// never reads the password itself.
+// 1193/AC2: keep (imap.go) and the sender (sender.go, issue 1195) read the password through the moved
+// package; SecretGiven (stat only) stays in config.go, available to the tools, and never reads the password
+// itself. (sender.go's own account file is read through secret.Path too, though its own account is not a
+// password; the real per-message password read is issue 1196's.)
 func TestKeepAndTheSenderReadThePasswordThroughTheMovedPackage(t *testing.T) {
-	for _, file := range []string{"imap.go"} {
+	for _, file := range []string{"imap.go", "sender.go"} {
 		fset := token.NewFileSet()
 		f, err := parser.ParseFile(fset, file, nil, parser.ImportsOnly)
 		if err != nil {

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

@@ -0,0 +1,79 @@
+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).
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"git.novox.be/novox/mesh-sdk/go/process"
+
+	"mesh-mail/cmd/mesh-mail/internal/secret"
+)
+
+// SendProcessID is this process's id, as module.json's resource "send" names it: the controller delivers
+// 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"
+
+// send is `mesh-mail send`: it reads its own bus account and the accounts setting, connects, and consumes
+// mail-sends until stopped. It refuses to start naming what is missing when the account or the accounts
+// setting cannot be read — never a password, which no error here or anywhere in this process logs
+// (novox/hq ADR 0436 §7, §12).
+func send(env Env) int {
+	if env.Secrets == "" {
+		logf("[mesh-mail] send was started without its secrets directory (MESH_MAIL_SECRETS)")
+		return 2
+	}
+	if _, err := ReadAccounts(env.Accounts); err != nil {
+		logf("[mesh-mail] send: %v", err)
+		return 1
+	}
+	account, err := process.ReadAccount(secret.Path(env.Secrets, "broker-"+SendProcessID))
+	if err != nil {
+		logf("[mesh-mail] send: %v", err)
+		return 1
+	}
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+	logf("[mesh-mail] send is consuming %s as %s", MailSendsStream, account.User)
+	if err := runSend(ctx, account, handleQueuedSend); err != nil && ctx.Err() == nil {
+		logf("[mesh-mail] send stopped: %v", err)
+		return 1
+	}
+	return 0
+}
+
+// runSend connects as the sender's own account and consumes mail-sends until ctx ends, handing each message
+// to handle. Apart from send so a test can drive it over a bus of its own, with a handler of its own
+// (novox/hq issue 1195, AC2): on ctx ending (a signal), Consume stops pulling and returns without
+// acknowledging a message still in handle's hands — only handle decides a message's fate (process/queue.go).
+func runSend(ctx context.Context, account process.Account, handle func(*process.Message)) error {
+	conn, err := process.Connect(account)
+	if err != nil {
+		return fmt.Errorf("the bus cannot be reached: %w", err)
+	}
+	defer conn.Close()
+	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)
+	}
+}

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

@@ -0,0 +1,120 @@
+//go:build integration
+
+package main
+
+// mesh-mail send over a real bus (novox/hq ADR 0436 §7, issue 1195 AC2): a JetStream server of the test's
+// own, nats-server v2.11.17, copied from modules/mesh-delivery/cmd/mesh-delivery/bus_test.go:52-58
+// (newTestBus) — the SDK's own testbus is internal, so this module keeps its own copy. Compiled by the
+// merge gate's `go vet -tags integration`; never run on this machine (the context file's hard rule).
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/process"
+	"github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// 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 {
+	t.Helper()
+	opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(),
+		NoLog: true, NoSigs: true}
+	s, err := server.NewServer(opts)
+	if err != nil {
+		t.Fatalf("a bus for this test could not be made: %v", err)
+	}
+	go s.Start()
+	if !s.ReadyForConnections(30 * time.Second) {
+		s.Shutdown()
+		t.Fatal("a bus for this test did not come up within 30s")
+	}
+	t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() })
+	return s.ClientURL()
+}
+
+// 1195/AC2: given the mail-sends work-queue stream and its consumer <machine>_mesh-mail_send, when the
+// process runs with a handler and a message is published, then the handler receives it; when the process
+// gets SIGTERM (here, its context cancelled, the same signal main.go's send() reacts to) it stops without
+// acknowledging a message in hand.
+func TestSendConsumesMailSendsAndStopsCleanlyOnSignal(t *testing.T) {
+	url := newTestBus(t)
+	ctx := context.Background()
+
+	nc, err := nats.Connect(url)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer nc.Close()
+	js, err := jetstream.New(nc)
+	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)
+	}
+
+	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 {
+		t.Fatal(err)
+	}
+
+	received := make(chan *process.Message, 1)
+	runCtx, cancel := context.WithCancel(context.Background())
+	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 {
+		t.Fatal(err)
+	}
+
+	select {
+	case msg := <-received:
+		if msg.Subject != MailSendApproved {
+			t.Fatalf("received on %q, want %q", msg.Subject, MailSendApproved)
+		}
+		// 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")
+	}
+
+	cancel() // the signal send() itself reacts to (SIGTERM, or ^C)
+	select {
+	case err := <-done:
+		if err == nil {
+			t.Fatal("runSend returned nil, want ctx.Err() once its context was cancelled")
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("runSend did not stop once its context was cancelled")
+	}
+
+	// The message was never acknowledged: it is still pending (or ack-pending) on the same durable consumer,
+	// read fresh over a new connection.
+	nc2, err := nats.Connect(url)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer nc2.Close()
+	js2, err := jetstream.New(nc2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	cons, err := js2.Consumer(ctx, MailSendsStream, account.QueueConsumerName())
+	if err != nil {
+		t.Fatal(err)
+	}
+	info, err := cons.Info(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if info.NumPending == 0 && info.NumAckPending == 0 {
+		t.Fatal("the message was acknowledged despite the process stopping mid-handle")
+	}
+}

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

@@ -0,0 +1,79 @@
+package main
+
+// mesh-mail send (novox/hq ADR 0436 §7, issue 1195): refuses to start naming what is missing, never
+// attempting to reach a bus when it cannot even read its own account or the accounts setting.
+
+import (
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+// 1195/AC1: given `mesh-mail send` with MESH_MAIL_SECRETS holding broker-send.secret in the controller's
+// form, when it starts, then it reads the account and the accounts file, and refuses to start naming what
+// is missing when either is absent.
+func TestSendRefusesToStartNamingWhatIsMissing(t *testing.T) {
+	validAccount := `{"url": "nats://127.0.0.1:4222", "node": "g14", "module": "mesh-mail", ` +
+		`"user": "g14.mesh-mail.send", "password": "pw"}`
+
+	t.Run("no secrets directory", func(t *testing.T) {
+		dir := t.TempDir()
+		env := Env{Accounts: filepath.Join(dir, "accounts.json")}
+		writeFile(t, env.Accounts, stageOne)
+		if code := send(env); code == 0 {
+			t.Fatal("send started with no secrets directory at all")
+		}
+	})
+
+	t.Run("no accounts file", func(t *testing.T) {
+		dir := t.TempDir()
+		env := Env{Accounts: filepath.Join(dir, "accounts.json"), Secrets: dir}
+		writeFile(t, filepath.Join(dir, "broker-send.secret"), validAccount)
+		if code := send(env); code == 0 {
+			t.Fatal("send started with no accounts setting")
+		}
+	})
+
+	t.Run("accounts file is not the module's", func(t *testing.T) {
+		dir := t.TempDir()
+		env := Env{Accounts: filepath.Join(dir, "accounts.json"), Secrets: dir}
+		writeFile(t, env.Accounts, `not json`)
+		writeFile(t, filepath.Join(dir, "broker-send.secret"), validAccount)
+		if code := send(env); code == 0 {
+			t.Fatal("send started with an accounts setting it cannot read")
+		}
+	})
+
+	t.Run("no own account delivered yet", func(t *testing.T) {
+		dir := t.TempDir()
+		env := Env{Accounts: filepath.Join(dir, "accounts.json"), Secrets: dir}
+		writeFile(t, env.Accounts, stageOne)
+		// broker-send.secret is not written at all.
+		if code := send(env); code == 0 {
+			t.Fatal("send started with no own account delivered")
+		}
+	})
+
+	t.Run("own account missing a field", func(t *testing.T) {
+		dir := t.TempDir()
+		env := Env{Accounts: filepath.Join(dir, "accounts.json"), Secrets: dir}
+		writeFile(t, env.Accounts, stageOne)
+		writeFile(t, filepath.Join(dir, "broker-send.secret"), `{"url": "nats://127.0.0.1:4222", "node": "g14", "module": "mesh-mail"}`)
+		code := send(env)
+		if code == 0 {
+			t.Fatal("send started with an own account missing its user and password")
+		}
+	})
+}
+
+// The account file's path is exactly what secret.Path gives for the member "broker-<process id>", inside
+// the directory MESH_MAIL_SECRETS already names — no secret of its own needs declaring in the manifest for
+// it (novox/hq ADR 0436 §2: the controller delivers it automatically for every process with a bus).
+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)
+	}
+}

modules/mesh-mail/go.mod

@@ -7,19 +7,25 @@ require (
 	github.com/emersion/go-imap/v2 v2.0.0-beta.8
 	github.com/emersion/go-message v0.18.2
 	github.com/jackc/pgx/v5 v5.11.0
+	github.com/nats-io/nats-server/v2 v2.11.17
+	github.com/nats-io/nats.go v1.54.0
 )
 
 require (
+	github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect
 	github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6 // indirect
+	github.com/google/go-tpm v0.9.8 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
 	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/klauspost/compress v1.20.0 // indirect
-	github.com/nats-io/nats.go v1.54.0 // indirect
+	github.com/minio/highwayhash v1.0.4 // indirect
+	github.com/nats-io/jwt/v2 v2.8.1 // indirect
 	github.com/nats-io/nkeys v0.4.16 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	golang.org/x/crypto v0.57.0 // indirect
 	golang.org/x/sync v0.23.0 // indirect
 	golang.org/x/sys v0.48.0 // indirect
 	golang.org/x/text v0.42.0 // indirect
+	golang.org/x/time v0.15.0 // indirect
 )

modules/mesh-mail/go.sum

@@ -64,6 +64,7 @@ golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
 golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=