novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

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)

Browse Source 
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
jschoubben
2026-11-05 15:33:37 +01:00
parent 25a704a1f5
commit 8c80e1b097
3 changed files with 29 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go
+10 -27
View File
@@ -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.<seat>.event.<event>.<module>`) 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, `<seat>.<event>.<module>`, 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 {
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided_test.go
+9 -5
View File
@@ -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")
}
}
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go
+10 -3
View File
@@ -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 {