mesh-delivery-teller: pull the durable consumer for the router's warrants, so an answer sent while it is down is not lost (issue 1126)

Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go

@@ -11,6 +11,7 @@ import (
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/asks"
+	"git.novox.be/novox/mesh-sdk/go/process"
 )
 
 // The teller's three bus adapters for mesh-sdk go/asker, which in a bundle reaches the bus through the runtime: here the
@@ -65,16 +66,33 @@ func recordVia(ctx context.Context, conn *nats.Conn, module string) func(id stri
 }
 
 // decidedVia hears the warrants the router says to this module alone: its own `decided` event, a subject only the router
-// publishes (the grant to subscribe it names the module). The handler is the asker's: pattern is the key it expects.
+// publishes (the grant to subscribe it names the module). They are pulled through the module's durable consumer on EVENTS
-func decidedVia(conn *nats.Conn, module string, logf func(string, ...any)) func(pattern string, handle func(stdio.Envelope) error) error {
+// (made with its grant for the seat it uses), so an answer the router published while the teller was down is delivered
+// when it is back, never lost (issue 1126). hear starts the pull, handing each message to its handler; the message is
+// acknowledged once handled. The handler is the asker's: pattern is the key it expects.
+func decidedVia(hear func(func(jetstream.Msg)) error, logf func(string, ...any)) func(pattern string, handle func(stdio.Envelope) error) error {
 	return func(pattern string, handle func(stdio.Envelope) error) error {
-		_, err := conn.Subscribe(asks.DecidedSubject(module), func(m *nats.Msg) {
+		return hear(func(m jetstream.Msg) {
-			if err := handle(stdio.Envelope{Key: pattern, Body: m.Data}); err != nil {
+			if err := handle(stdio.Envelope{Key: pattern, Body: m.Data()}); err != nil {
 				logf("a warrant could not be handled and is read again from the router's record at the next start: %v", err)
 			}
+			if err := m.Ack(); err != nil {
+				logf("a warrant was handled but could not be acknowledged: %v", err)
+			}
 		})
+	}
+}
+
+// pullDecided starts the pull of the module's durable consumer (process.HearEvents) and stops it when ctx ends.
+func pullDecided(ctx context.Context, cons jetstream.Consumer, logf func(string, ...any)) func(func(jetstream.Msg)) error {
+	return func(handle func(jetstream.Msg)) error {
+		cc, err := process.HearEvents(cons, handle, logf)
+		if err != nil {
 			return err
 		}
+		context.AfterFunc(ctx, cc.Stop)
+		return nil
+	}
 }
 
 // sayVia publishes a condition on the seat's one subject for conditions, under the module's own name.

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided_test.go

@@ -0,0 +1,54 @@
+package main
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+)
+
+// pulled is a message of the durable consumer: only what decidedVia reads and does.
+type pulled struct {
+	jetstream.Msg
+	data  []byte
+	acked *int
+}
+
+func (m pulled) Data() []byte { return m.data }
+func (m pulled) Ack() error   { *m.acked++; return nil }
+
+// An answer the router published while the teller was down is pulled from its durable consumer when it starts again:
+// the handler hears the warrant under the asker's key, and the message is acknowledged, handled or not (issue 1126).
+func TestDecidedViaHearsThePulledWarrantAndAcknowledgesIt(t *testing.T) {
+	var deliver func(jetstream.Msg)
+	var logs []string
+	sub := decidedVia(func(h func(jetstream.Msg)) error { deliver = h; return nil },
+		func(f string, _ ...any) { logs = append(logs, f) })
+	var got []stdio.Envelope
+	fail := false
+	if err := sub("k", func(e stdio.Envelope) error {
+		got = append(got, e)
+		if fail {
+			return errors.New("no")
+		}
+		return nil
+	}); err != nil || deliver == nil {
+		t.Fatalf("subscribe: %v, delivering %v", err, deliver != nil)
+	}
+	acked := 0
+	deliver(pulled{data: []byte(`{"ask":"a1"}`), acked: &acked})
+	fail = true
+	deliver(pulled{data: []byte(`{"ask":"a2"}`), acked: &acked})
+	if len(got) != 2 || got[0].Key != "k" || string(got[0].Body) != `{"ask":"a1"}` || string(got[1].Body) != `{"ask":"a2"}` {
+		t.Errorf("heard %+v", got)
+	}
+	if acked != 2 || len(logs) != 1 {
+		t.Errorf("acknowledged %d of 2, logged %d lines, want 1 (the failed handling)", acked, len(logs))
+	}
+	failing := decidedVia(func(func(jetstream.Msg)) error { return errors.New("no consumer") }, nil)
+	if err := failing("k", nil); err == nil {
+		t.Error("a pull that could not start is no error")
+	}
+}

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go

@@ -170,6 +170,7 @@ const (
 	gateAsksKV     = "mesh-delivery-gate_asks"
 	assignerAsks   = "mesh-delivery-assigner_asks"
 	viewAnswers    = "mesh-delivery-view_answers"
+	tellerConsumer = "n1_mesh-delivery-teller"
 	seatAsk        = "mesh.seat.operator-channel.accept.ask.mesh-delivery-teller"
 	seatCancel     = "mesh.seat.operator-channel.accept.cancel.mesh-delivery-teller"
 	seatCondition  = "mesh.seat.operator-channel.accept.condition.mesh-delivery-teller"
@@ -181,6 +182,8 @@ func fullConf(dir string) string {
 	// without it held the start ten seconds on a read the bus never answered, and the tests raced that stall.
 	teller := append(append(append(append(keepsOf(tellerAnswers), keepsOf(tellerAsked)...), keepsOf(ends.Bucket)...), readOf(gateAsksKV)...),
 		seatAsk, seatCancel, seatCondition, "$JS.API.DIRECT.GET.KV_"+messengerAsks+".$KV."+messengerAsks+".mesh-delivery-teller.>")
+	// The durable consumer on EVENTS the controller makes for the seat it uses, pulled and acknowledged (issue 1126).
+	teller = append(teller, "$JS.API.CONSUMER.INFO.EVENTS."+tellerConsumer, "$JS.API.CONSUMER.MSG.NEXT.EVENTS."+tellerConsumer, "$JS.ACK.EVENTS."+tellerConsumer+".>")
 	teller = append(append(teller, readOf(assignerAsks)...), readOf(viewAnswers)...) // issue 1297 B10: the manifest's reads
 	view := append(append(append(readOf(tellerAnswers), readOf(gateAsksKV)...), readOf(assignerAsks)...), keepsOf(viewAnswers)...)
 	assigner := keepsOf(assignerAsks)
@@ -197,7 +200,7 @@ accounts {
   }
 }
 `, filepath.Join(dir, "js"),
-		user("n1.mesh-delivery-teller", teller, []string{seatDecided}),
+		user("n1.mesh-delivery-teller", teller, nil),
 		user("n1.mesh-delivery-view", view, nil),
 		user("n1.mesh-delivery-assigner", assigner, nil),
 		user("n1.node-tools", tools, nil))
@@ -257,20 +260,30 @@ func startFull(t *testing.T) *fullBed {
 		Retention: jetstream.WorkQueuePolicy}); err != nil {
 		t.Fatal(err)
 	}
+	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.seat.operator-channel.event.>"}}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: tellerConsumer, FilterSubject: seatDecided,
+		AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
+		t.Fatal(err)
+	}
 	return &fullBed{srv: srv, ctl: ctl, js: js, ctx: ctx}
 }
 
-// listening waits until the program hears the router's warrants (its subscription on the decided subject is on the bus),
+// listening waits until the program hears the router's warrants (its pull of the durable consumer is waiting on the bus),
 // which it does before it watches any bucket: an event, not a pause a loaded machine can outlast.
 func (b *fullBed) listening(t *testing.T) {
 	t.Helper()
-	acc, err := b.srv.LookupAccount("MESH")
+	cons, err := b.js.Consumer(b.ctx, "EVENTS", tellerConsumer)
 	if err != nil {
 		t.Fatal(err)
 	}
-	for deadline := time.Now().Add(20 * time.Second); !acc.SubscriptionInterest(seatDecided); time.Sleep(10 * time.Millisecond) {
+	for deadline := time.Now().Add(20 * time.Second); ; time.Sleep(10 * time.Millisecond) {
+		if info, err := cons.Info(b.ctx); err == nil && info.NumWaiting > 0 {
+			return
+		}
 		if time.Now().After(deadline) {
-			t.Fatal("the program did not subscribe to the router's warrants within 20 s")
+			t.Fatal("the program did not pull the router's warrants within 20 s")
 		}
 	}
 }

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go

@@ -67,8 +67,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 	state := kvStore{ctx: ctx, kv: kvs[1]}
 	publish := publishVia(ctx, js)
+	cons, err := cred.EventConsumer(ctx, js, process.EventsStream)
+	if err != nil {
+		// Essential: the router's warrants are how an answer reaches the teller.
+		return fmt.Errorf("the consumer %s on %s cannot be read (%v): the teller hears no answer without it, so it stops; where the bus refused it or did not answer, this module's consumer (made with its grant for the seat it uses) has not reached the bus yet", cred.EventConsumerName(), process.EventsStream, err)
+	}
 	client := &asker.Client{Module: delivery.TellerModule, State: state, Publish: publish,
-		Record: recordVia(ctx, conn, delivery.TellerModule), Subscribe: decidedVia(conn, delivery.TellerModule, logger.Printf)}
+		Record: recordVia(ctx, conn, delivery.TellerModule), Subscribe: decidedVia(pullDecided(ctx, cons, logger.Printf), logger.Printf)}
 	t := &Teller{Asker: client, Store: state, Answers: kvAnswers{ctx: ctx, kv: kvs[0]}, Gate: kvGate{ctx: ctx, kv: gate},
 		Say: sayVia(ctx, js, delivery.TellerModule), Logf: logger.Printf, Publish: publish}
 	// Hear the warrants first, so none that comes while the asks are read is missed.

modules/mesh-delivery-teller/manifest_test.go

@@ -149,9 +149,7 @@ func TestManifestProcess(t *testing.T) { // 1240/AC1
 
 func TestServesNoRequestSubjectAndHoldsNoTicker(t *testing.T) { // 1240/AC3 (ADR 0437 §8)
 	forbidden := []string{"Subscribe(", "QueueSubscribe", "ChanSubscribe", "Respond(", "RespondMsg(", "time.NewTicker", "time.Tick", "time.NewTimer", "time.AfterFunc", "time.After(", "time.Sleep"}
-	// The one subscription the teller makes is to its own `decided` event, the router's word on its asks (ADR 0259, ADR 0437
+	// The teller hears its own `decided` event through its durable consumer, pulled (decided.go): it holds no subscription.
-	// §6): an event it hears, not a request it answers. It is allowed in decided.go and there only on that subject.
-	const decidedFile = "decided.go"
 	var files int
 	for _, root := range []string{"cmd", "internal"} {
 		err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
@@ -167,13 +165,6 @@ func TestServesNoRequestSubjectAndHoldsNoTicker(t *testing.T) { // 1240/AC3 (ADR
 				return err
 			}
 			for _, f := range forbidden {
-				if f == "Subscribe(" && filepath.Base(path) == decidedFile {
-					if n := strings.Count(string(src), "conn.Subscribe("); n != 1 || !strings.Contains(string(src), "conn.Subscribe(asks.DecidedSubject(module),") ||
-						strings.Count(string(src), f) != n {
-						t.Errorf("%s holds %d subscriptions, and not just one to asks.DecidedSubject(module)", path, strings.Count(string(src), f))
-					}
-					continue
-				}
 				if strings.Contains(string(src), f) {
 					t.Errorf("%s holds %s: the teller serves no request subject and holds no timer", path, f)
 				}