From dd3ed68483dc25020518f651829e167e4e839f57 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 09:49:24 +0200 Subject: [PATCH] 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 | 14 +- .../cmd/mesh-mail/secret_boundary_test.go | 9 +- modules/mesh-mail/cmd/mesh-mail/sender.go | 79 ++++++++++++ .../cmd/mesh-mail/sender_integration_test.go | 120 ++++++++++++++++++ .../mesh-mail/cmd/mesh-mail/sender_test.go | 79 ++++++++++++ modules/mesh-mail/go.mod | 8 +- modules/mesh-mail/go.sum | 1 + 7 files changed, 301 insertions(+), 9 deletions(-) create mode 100644 modules/mesh-mail/cmd/mesh-mail/sender.go create mode 100644 modules/mesh-mail/cmd/mesh-mail/sender_integration_test.go create mode 100644 modules/mesh-mail/cmd/mesh-mail/sender_test.go diff --git a/modules/mesh-mail/cmd/mesh-mail/main.go b/modules/mesh-mail/cmd/mesh-mail/main.go index 02d4fffa..683d3016 100644 --- a/modules/mesh-mail/cmd/mesh-mail/main.go +++ b/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()) diff --git a/modules/mesh-mail/cmd/mesh-mail/secret_boundary_test.go b/modules/mesh-mail/cmd/mesh-mail/secret_boundary_test.go index e819899d..6dd6d1b5 100644 --- a/modules/mesh-mail/cmd/mesh-mail/secret_boundary_test.go +++ b/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 { diff --git a/modules/mesh-mail/cmd/mesh-mail/sender.go b/modules/mesh-mail/cmd/mesh-mail/sender.go new file mode 100644 index 00000000..88ff5880 --- /dev/null +++ b/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- (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) + } +} diff --git a/modules/mesh-mail/cmd/mesh-mail/sender_integration_test.go b/modules/mesh-mail/cmd/mesh-mail/sender_integration_test.go new file mode 100644 index 00000000..d36ee676 --- /dev/null +++ b/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 _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") + } +} diff --git a/modules/mesh-mail/cmd/mesh-mail/sender_test.go b/modules/mesh-mail/cmd/mesh-mail/sender_test.go new file mode 100644 index 00000000..57fb55c0 --- /dev/null +++ b/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-", 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) + } +} diff --git a/modules/mesh-mail/go.mod b/modules/mesh-mail/go.mod index 9f3bf724..b5826dcc 100644 --- a/modules/mesh-mail/go.mod +++ b/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 ) diff --git a/modules/mesh-mail/go.sum b/modules/mesh-mail/go.sum index bea2f6aa..8aa47987 100644 --- a/modules/mesh-mail/go.sum +++ b/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=