diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go index ee7780b5..18c2b3d8 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go +++ b/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,15 +66,32 @@ 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. -func decidedVia(conn *nats.Conn, module string, logf func(string, ...any)) func(pattern string, handle func(stdio.Envelope) error) error { +// publishes (the grant to subscribe it names the module). They are pulled through the module's durable consumer on EVENTS +// (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) { - if err := handle(stdio.Envelope{Key: pattern, Body: m.Data}); err != nil { + return hear(func(m jetstream.Msg) { + 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) + } }) - return 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 } } diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided_test.go new file mode 100644 index 00000000..ee1c86db --- /dev/null +++ b/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") + } +} diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go index 5754987a..f7e015f0 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go @@ -165,15 +165,16 @@ func user(name string, pub, sub []string) string { } const ( - tellerAnswers = "mesh-delivery-teller_answers" - tellerAsked = "mesh-delivery-teller_asked" - gateAsksKV = "mesh-delivery-gate_asks" - assignerAsks = "mesh-delivery-assigner_asks" - viewAnswers = "mesh-delivery-view_answers" - 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" - seatDecided = "mesh.seat.operator-channel.event.decided.mesh-delivery-teller" + tellerAnswers = "mesh-delivery-teller_answers" + tellerAsked = "mesh-delivery-teller_asked" + 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" + seatDecided = "mesh.seat.operator-channel.event.decided.mesh-delivery-teller" ) func fullConf(dir string) string { @@ -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") } } } diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go index ad4a99fe..8557c036 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go +++ b/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. diff --git a/modules/mesh-delivery-teller/manifest_test.go b/modules/mesh-delivery-teller/manifest_test.go index 2111e273..74420f55 100644 --- a/modules/mesh-delivery-teller/manifest_test.go +++ b/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 - // §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" + // The teller hears its own `decided` event through its durable consumer, pulled (decided.go): it holds no subscription. 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) }