Hear mesh-mail's added events, recorded before acknowledged, in arrival order (issue 1518)

2026-10-20 23:56:17 +02:00
parent 0d2d19daa0
commit d7b5c87887
8 changed files with 355 additions and 6 deletions
@@ -0,0 +1,33 @@
+package main
+
+// mesh-mail-reader carries a byte-identical copy of mesh-mail's event shape (novox/hq ADR 0459 §3, issue 1518),
+// the way mesh-mail-notifier carries it too (ADR 0378 §9, copies_test.go) and modules/networkmanager's
+// uplink-tools carries ifwatchd's own file (ADR 0241): each builds alone, so each holds its own copy, and this
+// test holds them to one text.
+
+import (
+	"bytes"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+// source is where mesh-mail's own copy of the file lives.
+var source = filepath.Join("..", "..", "..", "mesh-mail", "cmd", "mesh-mail")
+
+// 1518/AC1.
+func TestCarriesTheSameMailEventAsMeshMail(t *testing.T) {
+	for _, f := range []string{"mailevent.go"} {
+		mine, err := os.ReadFile(f)
+		if err != nil {
+			t.Fatal(err)
+		}
+		theirs, err := os.ReadFile(filepath.Join(source, f))
+		if err != nil {
+			t.Fatalf("mesh-mail's copy of %s could not be read: %v", f, err)
+		}
+		if !bytes.Equal(mine, theirs) {
+			t.Errorf("1518/AC1: this copy of %s differs from mesh-mail's: change both together", f)
+		}
+	}
+}
@@ -0,0 +1,64 @@
+package main
+
+// Hearing mesh-mail's added events (novox/hq ADR 0459 §3, §12, issue 1518): mesh-mail-reader consumes
+// mesh-mail.mail.*.message.added under its own runs-as account, exactly as mesh-mail-notifier does (main.go
+// AddedPattern, watchWith), but records each one as a waiting mail (store.go, pgstore.go) BEFORE the handler
+// returns success — so a mail heard is never lost, even across a restart: the controller composes a durable
+// consumer (MaxDeliver 5, AckWait 30s) that redelivers an event five times before it is dead-lettered, and a
+// store failure here must make that redelivery happen, never swallow the event.
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+)
+
+// EventBody is what mesh-mail's added event says of a message (metadata only; mailevent.go names where it is
+// said). Unknown fields are ignored: the reader reads only what it needs, never a body.
+type EventBody struct {
+	Account     string   `json:"account"`
+	Folder      string   `json:"folder"`
+	UID         uint32   `json:"uid"`
+	UIDValidity uint32   `json:"uidvalidity"`
+	MessageID   string   `json:"message-id"`
+	Date        string   `json:"date"`
+	From        []Person `json:"from"`
+	Subject     string   `json:"subject"`
+	Size        int64    `json:"size"`
+	Flags       []string `json:"flags"`
+}
+
+// Hear records one added event as a waiting mail (1518/AC2): the store is written to before this returns
+// nil, so the caller (the bus subscription's handler) only then acknowledges the event. A store failure is
+// returned unchanged, so the event is redelivered rather than lost (1518/AC2). The same event heard twice
+// (1518/AC3) is recorded once: AddWaiting itself is idempotent on the mail's own identity.
+func Hear(ctx context.Context, store Store, raw json.RawMessage, now func() time.Time) error {
+	var body EventBody
+	if err := json.Unmarshal(raw, &body); err != nil {
+		// An event this reader cannot even read is taken and dropped, exactly as mesh-mail-notifier does
+		// (main.go watchWith): a malformed event would otherwise be redelivered forever, never succeeding.
+		return nil
+	}
+	date, err := time.Parse(time.RFC3339, body.Date)
+	if err != nil {
+		date = now()
+	}
+	w := WaitingMail{
+		Account:     body.Account,
+		Folder:      body.Folder,
+		MessageID:   body.MessageID,
+		UIDValidity: body.UIDValidity,
+		UID:         body.UID,
+		Date:        date,
+		From:        body.From,
+		Subject:     body.Subject,
+		Flags:       body.Flags,
+		Size:        body.Size,
+		Heard:       now(),
+	}
+	return store.AddWaiting(ctx, w)
+}
+
+// AddedPattern is the mesh-mail stream's event this module consumes (its own `consumes` entry, novox/hq ADR
+// 0459 §3): a new message's `added` only.
+const AddedPattern = "mesh-mail.mail.*.message.added"
@@ -0,0 +1,113 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+)
+
+func addedEventBody(t *testing.T, account, folder, messageID string, uid uint32) json.RawMessage {
+	t.Helper()
+	raw, err := json.Marshal(map[string]any{
+		"account": account, "folder": folder, "uid": uid, "uidvalidity": uint32(1),
+		"message-id": messageID, "date": "2026-10-20T10:00:00Z",
+		"from":    []Person{{Name: "Alice", Address: "alice@example.org"}},
+		"subject": "hello", "size": int64(1234), "flags": []string{},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return raw
+}
+
+// 1518/AC2: an added event is recorded as a waiting mail before the handler returns success.
+func TestHearRecordsTheEventAsAWaitingMailBeforeSucceeding(t *testing.T) {
+	store := newMemStore()
+	now := func() time.Time { return time.Date(2026, 10, 20, 10, 0, 1, 0, time.UTC) }
+	raw := addedEventBody(t, "bob", "INBOX", "<msg-1@example.org>", 7)
+	if err := Hear(context.Background(), store, raw, now); err != nil {
+		t.Fatal(err)
+	}
+	waiting, err := store.Waiting(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(waiting) != 1 {
+		t.Fatalf("waiting = %v, want one waiting mail", waiting)
+	}
+	w := waiting[0]
+	if w.Account != "bob" || w.Folder != "INBOX" || w.MessageID != "<msg-1@example.org>" || w.UID != 7 ||
+		w.Subject != "hello" || w.Size != 1234 || len(w.From) != 1 || w.From[0].Address != "alice@example.org" {
+		t.Fatalf("waiting mail: %+v", w)
+	}
+}
+
+// 1518/AC2: a store failure is returned unchanged, so the event is redelivered rather than lost.
+func TestHearReturnsAStoreFailureSoTheEventIsRedelivered(t *testing.T) {
+	store := newMemStore()
+	store.failNext = errStoreFake
+	raw := addedEventBody(t, "bob", "INBOX", "<msg-2@example.org>", 8)
+	err := Hear(context.Background(), store, raw, time.Now)
+	if err != errStoreFake {
+		t.Fatalf("Hear with a failing store: %v, want %v", err, errStoreFake)
+	}
+	waiting, _ := store.Waiting(context.Background())
+	if len(waiting) != 0 {
+		t.Fatalf("waiting = %v, want none (the failed add kept nothing)", waiting)
+	}
+}
+
+// 1518/AC3: the same event heard twice gives one waiting mail.
+func TestHearTwiceGivesOneWaitingMail(t *testing.T) {
+	store := newMemStore()
+	raw := addedEventBody(t, "bob", "INBOX", "<msg-3@example.org>", 9)
+	if err := Hear(context.Background(), store, raw, time.Now); err != nil {
+		t.Fatal(err)
+	}
+	if err := Hear(context.Background(), store, raw, time.Now); err != nil {
+		t.Fatal(err)
+	}
+	waiting, _ := store.Waiting(context.Background())
+	if len(waiting) != 1 {
+		t.Fatalf("waiting = %v, want exactly one", waiting)
+	}
+}
+
+// 1518/AC4: waiting mails are handed out in arrival order (the order they were heard), oldest first — not by
+// their own date, and regardless of whether reading was healthy when they arrived (Hear never consults
+// health; a mail heard while unhealthy is recorded exactly the same way).
+func TestHearKeepsArrivalOrder(t *testing.T) {
+	store := newMemStore()
+	// Heard out of date order: the third mail (by date) arrives first.
+	first := addedEventBody(t, "bob", "INBOX", "<a@example.org>", 1)
+	second := addedEventBody(t, "bob", "INBOX", "<b@example.org>", 2)
+	third := addedEventBody(t, "bob", "INBOX", "<c@example.org>", 3)
+	for _, raw := range []json.RawMessage{first, second, third} {
+		if err := Hear(context.Background(), store, raw, time.Now); err != nil {
+			t.Fatal(err)
+		}
+	}
+	waiting, _ := store.Waiting(context.Background())
+	if len(waiting) != 3 {
+		t.Fatalf("waiting = %v, want three", waiting)
+	}
+	wantOrder := []string{"<a@example.org>", "<b@example.org>", "<c@example.org>"}
+	for i, w := range waiting {
+		if w.MessageID != wantOrder[i] {
+			t.Fatalf("waiting[%d].MessageID = %q, want %q (arrival order)", i, w.MessageID, wantOrder[i])
+		}
+	}
+}
+
+// A malformed event is taken and dropped, never redelivered forever (mirrors mesh-mail-notifier's watchWith).
+func TestHearDropsAMalformedEvent(t *testing.T) {
+	store := newMemStore()
+	if err := Hear(context.Background(), store, json.RawMessage(`not json`), time.Now); err != nil {
+		t.Fatalf("Hear on a malformed event: %v, want it taken and dropped", err)
+	}
+	waiting, _ := store.Waiting(context.Background())
+	if len(waiting) != 0 {
+		t.Fatalf("waiting = %v, want none", waiting)
+	}
+}
@@ -0,0 +1,78 @@
+package main
+
+// What the tests stand on: an in-memory double of Store (store.go), for the hearing (1518) and health (1519)
+// tests alike. The postgres-backed PgStore (pgstore.go) is proven only by pgstore_test.go's unit tests and by
+// pgstore_integration_test.go, which this context never runs.
+
+import (
+	"context"
+	"errors"
+	"sync"
+)
+
+type memStore struct {
+	mu       sync.Mutex
+	waiting  []WaitingMail
+	keys     map[string]bool
+	nextSeq  int64
+	failNext error // AddWaiting fails once with this, then clears, so a test can prove a failure is returned
+	health   ReaderHealth
+}
+
+func newMemStore() *memStore {
+	return &memStore{keys: map[string]bool{}, health: ReaderHealth{Healthy: true}}
+}
+
+func (s *memStore) AddWaiting(_ context.Context, w WaitingMail) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.failNext != nil {
+		err := s.failNext
+		s.failNext = nil
+		return err
+	}
+	key := w.Key()
+	if s.keys[key] {
+		return nil // 1518/AC3: the same mail heard twice is recorded once.
+	}
+	s.keys[key] = true
+	s.nextSeq++
+	w.Seq = s.nextSeq
+	s.waiting = append(s.waiting, w)
+	return nil
+}
+
+func (s *memStore) Waiting(context.Context) ([]WaitingMail, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	out := make([]WaitingMail, len(s.waiting))
+	copy(out, s.waiting) // already in arrival order: appended in the order AddWaiting succeeded (1518/AC4).
+	return out, nil
+}
+
+func (s *memStore) TakeWaiting(_ context.Context, seq int64) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for i, w := range s.waiting {
+		if w.Seq == seq {
+			s.waiting = append(s.waiting[:i], s.waiting[i+1:]...)
+			return nil
+		}
+	}
+	return nil
+}
+
+func (s *memStore) ReaderHealth(context.Context) (ReaderHealth, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.health, nil
+}
+
+func (s *memStore) SetReaderHealth(_ context.Context, h ReaderHealth) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.health = h
+	return nil
+}
+
+var errStoreFake = errors.New("the store refused (fake, for a test)")
@@ -0,0 +1,13 @@
+package main
+
+// The shape of a mail-store message event (novox/hq ADR 0378 §2, issue 800 part A), moved here on its own so
+// mesh-mail-notifier (issue 800 part B) carries a byte-identical copy, the way modules/networkmanager's
+// uplink-tools carries ifwatchd's own file (novox/hq ADR 0241, copies_test.go). Nothing here reaches a mailbox or
+// a server; it only says where an account's change is told on the bus and what the three changes are called.
+
+// EventKey is where an account's change is said: mail.<account>.message.<added|changed|removed> (research 049
+// document 02 §5), the account by its name in the setting, never its address.
+func EventKey(account, change string) string { return "mail." + account + ".message." + change }
+
+// Changes are the three things said of a message.
+var Changes = []string{"added", "changed", "removed"}
@@ -1,13 +1,12 @@
 // mesh-mail-reader: the default claimant of mail-reading (novox/hq ADR 0459 §1), built from mesh-mail's own
 // source, like mesh-mail-web and mesh-mail-notifier. One instance, held once for the mesh, on the machine
 // mesh-mail runs on. It hears mesh-mail's added events under its own runs-as account (ADR 0459 §3, issue
-// 1518 — not yet built), judges a batch of them at a time (1427, 1520 — not yet built), and serves the six
-// mail-reading verbs (entries, groups, correct, lessons, undo, status) and its own health tool,
-// mail_reader_health (1519 — not yet built). It reads mail; it moves, flags, deletes and sends nothing
-// (novox/hq ADR 0459 §11).
+// 1518), judges a batch of them at a time (1427, 1520 — not yet built), and serves the six mail-reading
+// verbs (entries, groups, correct, lessons, undo, status) and its own health tool, mail_reader_health
+// (1519 — not yet built). It reads mail; it moves, flags, deletes and sends nothing (novox/hq ADR 0459 §11).
 //
-//	mesh-mail-reader            the bundle the node's runtime launches: the six mail-reading verbs plus
-//	                            mail_reader_health, served over stdio (tools.go)
+//	mesh-mail-reader            the bundle the node's runtime launches: the hearing loop (hear.go) and the
+//	                            six mail-reading verbs plus mail_reader_health, served over stdio (tools.go)
 //	mesh-mail-reader prepare    the preparation step: the index's schema (store.go, pgstore.go) brought to
 //	                            this version's (issue 1517)
 //
@@ -36,13 +35,47 @@ func main() {
 	if len(os.Args) > 1 && os.Args[1] == "prepare" {
 		os.Exit(prepare())
 	}
+	storeFor := func(ctx context.Context) (Store, error) { return PgStoreFromEnv(ctx) }
 	b := &Bundle{Env: envFromOS()}
+	go hearLoop(storeFor, stdio.Subscribe, time.Sleep)
 	if err := stdio.ServeOn("", b.Tools(), os.Stdin, os.Stdout); err != nil {
 		logf("[mesh-mail-reader] %v", err)
 		os.Exit(1)
 	}
 }

+// ---- hearing --------------------------------------------------------------------------------------------------
+
+// hearLoop subscribes to mesh-mail's added events and hears each one (hear.go), exactly as
+// mesh-mail-notifier's watchWith does (issue 1518): a subscribe the runtime refuses is retried, 2 seconds
+// doubling to at most a minute, logging each refusal, never fatal — a refusal here must never take the six
+// mail-reading verbs or mail_reader_health down with it. Started before stdio.ServeOn, so it keeps running
+// whatever stdio.ServeOn does with the stream it is given.
+func hearLoop(storeFor func(context.Context) (Store, error), subscribe func(string, func(stdio.Envelope) error) error, sleep func(time.Duration)) {
+	for wait := 2 * time.Second; ; wait = minDuration(wait*2, time.Minute) {
+		err := subscribe(AddedPattern, func(e stdio.Envelope) error {
+			ctx := context.Background()
+			store, err := storeFor(ctx)
+			if err != nil {
+				return err
+			}
+			return Hear(ctx, store, e.Body, time.Now)
+		})
+		if err == nil {
+			return
+		}
+		logf("[mesh-mail-reader] does not hear %s yet (%v); asking again in %s", AddedPattern, err, wait)
+		sleep(wait)
+	}
+}
+
+func minDuration(a, b time.Duration) time.Duration {
+	if a < b {
+		return a
+	}
+	return b
+}
+
 // ---- prepare ------------------------------------------------------------------------------------------------------

 // prepare is the preparation step (1517/AC2): the schema brought to what this version needs, run once
@@ -23,6 +23,7 @@ type manifest struct {
 	RunsAs      string                    `json:"runs-as"`
 	Claims      []seat                    `json:"claims"`
 	Invokes     []string                  `json:"invokes"`
+	Consumes    []string                  `json:"consumes"`
 	Tools       []string                  `json:"tools"`
 	Requires    []string                  `json:"requires"`
 	Contributes map[string]map[string]any `json:"contributes"`
@@ -115,3 +116,14 @@ func TestTheManifestRequiresItsOwnDatabase(t *testing.T) {
 		t.Fatalf("prepare's DATABASE_URL_FILE is %v, database-url's own path is %v", env["DATABASE_URL_FILE"], databaseURL["path"])
 	}
 }
+
+// 1518/AC5: the manifest consumes mesh-mail.mail.*.message.added and runs-as mesh-mail-reader.
+func TestTheManifestConsumesMeshMailsAddedEvent(t *testing.T) {
+	m, _ := readManifest(t)
+	if !reflect.DeepEqual(m.Consumes, []string{AddedPattern}) {
+		t.Fatalf("consumes %v, want [%s]", m.Consumes, AddedPattern)
+	}
+	if m.RunsAs != Module {
+		t.Fatalf("runs-as %q, want %q", m.RunsAs, Module)
+	}
+}
@@ -9,6 +9,9 @@
      "serves": ["entries", "groups", "correct", "lessons", "undo", "status"]
    }
  ],
+  "consumes": [
+    "mesh-mail.mail.*.message.added"
+  ],
  "invokes": [
    "seat:mail-store.read",
    "seat:mail-store.accounts"