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

Teller: the hand acts through the seat's client, its findings through the conditions client's writer, the assigner's asks followed from the shared events stream, the warrants handed over by the views reader's one consumer; contract import paths follow the domain folders (D-12..D-22 teller part; issue 1966)

Browse Source 
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
jschoubben
2026-11-05 14:47:28 +01:00
parent 145cc70ceb
commit b9b589a02c
12 changed files with 52 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go
+2 -2
View File
@@ -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)
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go
+2 -2
View File
@@ -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"
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go
+2 -2
View File
@@ -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"
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go
+19 -12
View File
@@ -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.
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go
+6 -6
View File
@@ -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}
}
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go
+8 -7
View File
@@ -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
}
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/room_test.go
+2 -2
View File
@@ -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"
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go
+2 -2
View File
@@ -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"
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go
+2 -2
View File
@@ -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"
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers.go
+1 -1
View File
@@ -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"
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go
+2 -2
View File
@@ -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"
modules/mesh-delivery-teller/go.mod
+4 -2
View File
@@ -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