diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go index dd9bfa22..ba794d32 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go @@ -12,8 +12,8 @@ import ( "sync" "time" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" stdio "git.novox.be/novox/mesh-sdk/go" - "git.novox.be/novox/mesh-sdk/go/handacts" "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-teller/internal/ends" @@ -107,7 +107,7 @@ func main() { // teller_first_views and teller_first_view_again: where the teller's first view of the assigner's asks stands, and // asking it again by hand (ADR 0615, design 25 ยง 12.2). tools := append(Tools(askEnd(stdio.State(ends.State), 25*time.Second), time.Now), - views.StatusTools("mesh-delivery-teller", "teller_first_views", "teller_first_view_again", stdio.State(views.StatusState), handacts.Recorder{Ask: stdio.Ask}, time.Now)...) + views.StatusTools("mesh-delivery-teller", "teller_first_views", "teller_first_view_again", stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...) if err := stdio.Serve("", tools); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-teller-tools] %v\n", err) os.Exit(1) diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go index 950cc248..60f55915 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go @@ -5,8 +5,8 @@ import ( "fmt" "slices" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + 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" stdio "git.novox.be/novox/mesh-sdk/go" diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go index e1017c9c..1e8a143d 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go @@ -2,8 +2,8 @@ package main import ( "encoding/json" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + 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" "testing" 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 c46c60b5..6f850e5c 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go @@ -4,15 +4,14 @@ import ( "context" "encoding/json" "errors" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + 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" "time" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" stdio "git.novox.be/novox/mesh-sdk/go" - "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 @@ -84,16 +83,24 @@ func decidedVia(hear func(func(jetstream.Msg)) error, logf func(string, ...any)) } } -// 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 +// 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. 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 cda389ea..0e8c6027 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 @@ -6,8 +6,8 @@ import ( "context" "encoding/json" "fmt" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + 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" "git.novox.be/novox/mesh-sdk/go/views/viewstest" "os" "path/filepath" @@ -189,7 +189,7 @@ func fullConf(dir string) string { teller = append(teller, readOf(viewAnswers)...) // issue 1297 B10: the manifest's reads // The assigner's asks by its read tool and its views family (ADR 0615, DLV-M3), and the teller's own first-views and // findings; the bucket itself is no longer the teller's to read. - teller = append(append(append(teller, viewstest.Grants(delivery.AssignerModule, delivery.AssignerModule+".assigner_read_asks", "assigner-views", tellerConsumer)...), keepsOf(firstViewsBucket)...), keepsOf(findingsBucket)...) + teller = append(append(append(teller, viewstest.Grants(delivery.AssignerModule, delivery.AssignerModule+".assigner_read_asks", tellerConsumer)...), keepsOf(firstViewsBucket)...), keepsOf(findingsBucket)...) view := append(append(append(readOf(tellerAnswers), readOf(gateAsksKV)...), readOf(assignerAsks)...), keepsOf(viewAnswers)...) assigner := keepsOf(assignerAsks) tools := append(keepsOf(gateAsksKV), readOf(tellerAnswers)...) @@ -265,15 +265,15 @@ 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 { + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.seat.operator-channel.event.>", "mesh.mod.*.event.>"}}); err != nil { t.Fatal(err) } - if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: tellerConsumer, FilterSubject: seatDecided, + if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: tellerConsumer, AckPolicy: jetstream.AckExplicitPolicy}); err != nil { t.Fatal(err) } // The assigner as the owner of its asks: its read tool and its views family (ADR 0615). - viewstest.Serve(t, ctx, js, ctl, delivery.AssignerModule, "asks", delivery.AssignerModule+".assigner_read_asks", "assigner-views", tellerConsumer) + viewstest.Serve(t, ctx, js, ctl, delivery.AssignerModule, "asks", delivery.AssignerModule+".assigner_read_asks", tellerConsumer) return &fullBed{srv: srv, ctl: ctl, js: js, ctx: ctx} } 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 623f3724..b306cbc0 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go @@ -11,7 +11,8 @@ import ( "context" "errors" "fmt" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + sdkfindings "git.novox.be/novox/mesh-delivery/sdk/findings" "io" "log" "os" @@ -69,13 +70,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 { + if _, err := cred.EventConsumer(ctx, js, process.EventsStream); 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) } + warrants := &decidedHandler{} client := &occ.Client{Module: delivery.TellerModule, State: state, Publish: publish, - Record: recordVia(ctx, conn, delivery.TellerModule), Subscribe: decidedVia(pullDecided(ctx, cons, logger.Printf), logger.Printf)} + Record: recordVia(ctx, conn, delivery.TellerModule), Subscribe: decidedVia(warrants.hear, 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. @@ -92,10 +93,10 @@ 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} + reader := &views.ProcessReader{Module: delivery.TellerModule, Call: calls.Call, Logf: logger.Printf, Other: warrants.other} findings := process.AwaitBucket(ctx, js, conn, findingsBucket, "a first view unanswered past its bound is said in the log alone", logger.Printf) - reader.Findings = findings.KV - asks, err := reader.Follow(delivery.AssignerModule, assignerAsksState, delivery.AssignerModule+".assigner_read_asks", "assigner-views") + reader.Findings = func() views.FindingWriter { return sdkfindings.Writer(ctx, findings.KV) } + asks, err := reader.Follow(delivery.AssignerModule, assignerAsksState, delivery.AssignerModule+".assigner_read_asks") if err != nil { return err } diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/room_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/room_test.go index 2884b08c..cdf1275c 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/room_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/room_test.go @@ -2,8 +2,8 @@ package main import ( "encoding/json" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + 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" "testing" diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go index 380f28ab..1ee44804 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go @@ -10,8 +10,8 @@ import ( "sync" "time" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + 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" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/delivery" diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go index 2dea32e7..65b87038 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go @@ -5,8 +5,8 @@ import ( "encoding/json" "errors" "fmt" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + 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" "testing" "time" diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers.go index 42a865dd..293055e7 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers.go @@ -1,7 +1,7 @@ package main import ( - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" "strings" "time" diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go index da35692f..726ff215 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go @@ -1,8 +1,8 @@ package main import ( - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + 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" "testing" "time" diff --git a/modules/mesh-delivery-teller/go.mod b/modules/mesh-delivery-teller/go.mod index d2606693..3df9f4fc 100644 --- a/modules/mesh-delivery-teller/go.mod +++ b/modules/mesh-delivery-teller/go.mod @@ -3,6 +3,8 @@ module mesh-delivery-teller go 1.26.0 require ( + git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104174359-05b3e2e1b94f github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 @@ -23,8 +25,8 @@ require ( ) require ( - git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient v0.0.0-20261104174344-ae5266e77639 - git.novox.be/novox/mesh-contracts/go/seats/operatorchannel v0.0.0-20261104174344-ae5266e77639 + git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient v0.0.0-20261104174344-ae5266e77639 + git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel v0.0.0-20261104174344-ae5266e77639 ) replace git.novox.be/novox/mesh-delivery/sdk => ../../sdk