From 8c80e1b0976e9e9c147af6737eaeb1a6ca4b6def Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 15:33:37 +0100 Subject: [PATCH] Teller: the router's warrants are a handler of the module's event hub beside the views reader, on the seat's own subject (B7c finding 1; issue 1966) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../cmd/mesh-delivery-teller/decided.go | 37 +++++-------------- .../cmd/mesh-delivery-teller/decided_test.go | 14 ++++--- .../cmd/mesh-delivery-teller/main.go | 13 +++++-- 3 files changed, 29 insertions(+), 35 deletions(-) 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 6f850e5c..1b25805e 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go @@ -6,6 +6,7 @@ import ( "errors" occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" + "strings" "time" "github.com/nats-io/nats.go" @@ -65,14 +66,16 @@ 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). 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 { +// decidedVia hears the warrants the router says to this module alone: its own `decided` event, a seat event on the seat's +// subject (`mesh.seat..event..`) that only the router publishes (the grant to subscribe it names the +// module). They are handled on the module's one hub over its 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 registers the handler of the subject on the hub; the message is acknowledged once handled. The handler is the +// asker's: pattern, `..`, is the key it expects. +func decidedVia(hear func(subject string, handle 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 { - return hear(func(m jetstream.Msg) { + seat, event, _ := strings.Cut(pattern, ".") + return hear("mesh.seat."+seat+".event."+event, 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) } @@ -83,26 +86,6 @@ func decidedVia(hear func(func(jetstream.Msg)) error, logf func(string, ...any)) } } -// decidedHandler is where the router's warrants are handed: the module has one durable consumer of the shared events stream, -// pulled by the views reader (views.ProcessReader.Start), which hands every message that is not a views event to its Other. -type decidedHandler struct{ handle func(jetstream.Msg) } - -// hear registers the handler of the warrants; it pulls nothing. -func (d *decidedHandler) hear(handle func(jetstream.Msg)) error { - d.handle = handle - return nil -} - -// other is the views reader's Other: a message of the consumer that is no views event is a warrant (or nothing the teller -// hears: it is left for the bus to deliver again). -func (d *decidedHandler) other(m jetstream.Msg) { - if d.handle == nil { - _ = m.Nak() - return - } - d.handle(m) -} - // sayVia publishes a condition on the seat's one subject for conditions, under the module's own name. func sayVia(ctx context.Context, js jetstream.JetStream, module string) func(c oc.Condition) error { return func(c oc.Condition) error { 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 index ee1c86db..2266ffc3 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided_test.go @@ -23,12 +23,13 @@ func (m pulled) Ack() error { *m.acked++; return nil } // 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 subject string var logs []string - sub := decidedVia(func(h func(jetstream.Msg)) error { deliver = h; return nil }, + sub := decidedVia(func(s string, h func(jetstream.Msg)) error { subject, deliver = s, 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 { + if err := sub("operator-channel.decided.teller", func(e stdio.Envelope) error { got = append(got, e) if fail { return errors.New("no") @@ -37,18 +38,21 @@ func TestDecidedViaHearsThePulledWarrantAndAcknowledgesIt(t *testing.T) { }); err != nil || deliver == nil { t.Fatalf("subscribe: %v, delivering %v", err, deliver != nil) } + if subject != "mesh.seat.operator-channel.event.decided.teller" { + t.Errorf("the warrants are heard on %q", subject) + } 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"}` { + if len(got) != 2 || got[0].Key != "operator-channel.decided.teller" || 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 { + failing := decidedVia(func(string, func(jetstream.Msg)) error { return errors.New("no consumer") }, nil) + if err := failing("operator-channel.decided.teller", 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/main.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go index b306cbc0..f4e9df8c 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go @@ -74,9 +74,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // 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) } - warrants := &decidedHandler{} + // The module has one durable consumer of EVENTS: one hub pulls it for the router's warrants and the views reader, + // started once both are registered. + hub := process.NewEventHub(logger.Printf) + defer hub.Stop() client := &occ.Client{Module: delivery.TellerModule, State: state, Publish: publish, - Record: recordVia(ctx, conn, delivery.TellerModule), Subscribe: decidedVia(warrants.hear, logger.Printf)} + Record: recordVia(ctx, conn, delivery.TellerModule), Subscribe: decidedVia(hub.Handle, 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. @@ -93,7 +96,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return fmt.Errorf("the teller's connection for its tool calls: %w", err) } defer calls.Close() - reader := &views.ProcessReader{Module: delivery.TellerModule, Call: calls.Call, Logf: logger.Printf, Other: warrants.other} + reader := &views.ProcessReader{Module: delivery.TellerModule, Call: calls.Call, Logf: logger.Printf, Hub: hub} findings := process.AwaitBucket(ctx, js, conn, findingsBucket, "a first view unanswered past its bound is said in the log alone", logger.Printf) reader.Findings = func() views.FindingWriter { return sdkfindings.Writer(ctx, findings.KV) } asks, err := reader.Follow(delivery.AssignerModule, assignerAsksState, delivery.AssignerModule+".assigner_read_asks") @@ -102,6 +105,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } assignerKV := jetstream.KeyValue(asks) reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "teller_first_views reads nothing and teller_first_view_again is not heard", logger.Printf)) + // Every part has registered: the hub pulls the module's durable. + if err := hub.Start(ctx, js, cred); err != nil { + return fmt.Errorf("the consumer %s on %s cannot be read (%v): the teller hears no answer without it, so it stops", cred.EventConsumerName(), process.EventsStream, err) + } viewKV := process.OptionalBucket(ctx, js, delivery.ViewAnswersBucket, teller, logger.Printf) // The assigner's asks are raised on the operator channel whether or not the view's answers can be read (issue 1906). if viewKV != nil {