From a8601f5a5ca2df22d260163aea7a57f481b8ad6c Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 01:03:31 +0100 Subject: [PATCH] fix9-5: the indexer, the runner and the teller read the router's record through mesh-sdk's operatorask; no copy of the seat-holder lookup, no messenger_asks constant MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G A-101, A-74, §700 --- .../cmd/mesh-delivery-indexer/registry.go | 45 +++++-------------- .../cmd/mesh-delivery-pipeline/approval.go | 44 ++++-------------- .../cmd/mesh-delivery-teller/decided.go | 37 +++++---------- .../mesh-delivery-teller/integration_test.go | 25 +++++++++-- .../cmd/mesh-delivery-teller/main.go | 16 +++---- 5 files changed, 60 insertions(+), 107 deletions(-) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/registry.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/registry.go index 8778e00b..527c941f 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/registry.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/registry.go @@ -8,12 +8,12 @@ import ( "time" mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" - "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" gitseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/git" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/operatorask" "git.novox.be/novox/mesh-sdk/go/process" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -135,41 +135,16 @@ func newRegistry(js jetstream.JetStream, conn *nats.Conn, cred process.Account, Logf: logf} } -// routerBucket is the bucket of the router's asks: the state `asks` of the seat operator-channel (the contract's StateAsks) in -// the bucket of the seat's holder on the seat list, which the seat mesh-module-assigner's verb assignments answers with seats -// (design 47 G §700, A-74); no module's name is in the program. -func routerBucket(call toolCaller) (string, error) { - raw, err := call("seat:"+mma.Seat+"."+mma.VerbAssignments, mma.AssignmentsInput{Seats: mma.AssignmentsInputSeatsTrue}) - if err != nil { - return "", fmt.Errorf("who holds the seat %s is not asked of the seat %s: %w", oc.Seat, mma.Seat, err) - } - var held struct { - Seats []mma.SeatHolders `json:"seats"` - } - if err := json.Unmarshal(raw, &held); err != nil { - return "", fmt.Errorf("the answer of %s.%s is not its seat's: %w", mma.Seat, mma.VerbAssignments, err) - } - for _, s := range held.Seats { - if s.Seat == oc.Seat && len(s.Holders) > 0 { - return s.Holders[0].Module + "_" + oc.StateAsks, nil - } - } - return "", fmt.Errorf("no module holds the seat %s on the seat list, so the router's record of an ask is not read", oc.Seat) -} - -// routerRecordOf is Registry.Where: the router's own record of an ask, read under the indexer's module's name through -// operatorchannelclient, whose Record is a get of that key (`.` of the seat's state asks) on the process's own bus -// connection (record is process.Conn.Ask), as the tool runner gives it the runtime's mesh/seat.record (design 47 G §700): never -// the decided event's body. A warrant is then left unapplied and said while the record cannot be read. +// routerRecordOf is Registry.Where: the router's own record of an ask, read under the indexer's module's name by mesh-sdk's +// operatorask (design 47 G A-101, §700, A-74): the state asks of the seat operator-channel (the contract's StateAsks) in the +// bucket of the seat's holder on the seat list, which the seat mesh-module-assigner's verb assignments answers with seats; +// record is process.Conn.Ask, as the tool runner gives it the runtime's mesh/seat.record: never the decided event's body. A +// warrant is then left unapplied and said while the record cannot be read. func routerRecordOf(module string, call toolCaller, record func(bucket, id string) (*stdio.StateEntry, error)) func(id string) (string, *oc.Warrant, error) { - client := &operatorchannelclient.Client{Module: module, Record: func(id string) (*stdio.StateEntry, error) { - bucket, err := routerBucket(call) - if err != nil { - return nil, err - } - return record(bucket, id) - }} - return client.Where + asker := &operatorask.Asker[oc.Warrant]{Module: module, Call: operatorask.Caller(call), Record: record, + Contract: operatorask.Contract{Lists: "seat:" + mma.Seat + "." + mma.VerbAssignments, ListBody: mma.AssignmentsInput{Seats: mma.AssignmentsInputSeatsTrue}, + Seat: oc.Seat, StateAsks: oc.StateAsks}} + return asker.Where } // heardDecided is the handler of the decided events for this module: a warrant is applied by the registry; one it could not diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/approval.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/approval.go index ff6cae89..a5419fcb 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/approval.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/approval.go @@ -3,16 +3,15 @@ package main import ( "encoding/json" "errors" - "fmt" "time" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" - "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/operatorask" "git.novox.be/novox/mesh-sdk/go/process" "mesh-delivery-pipeline/internal/pipeline" @@ -21,40 +20,15 @@ import ( // toolCaller is a kept call of a seat's verb on the bus (process.Conn.Call). type toolCaller func(key string, body any) (json.RawMessage, error) -// routerBucket is the bucket of the router's asks: the state `asks` of the seat operator-channel (the contract's StateAsks) in -// the bucket of the seat's holder on the seat list, which the seat mesh-module-assigner's verb assignments answers with seats -// (design 47 G §700, A-74); no module's name is in the program. (The indexer's function, to move into mesh-sdk.) -func routerBucket(call toolCaller) (string, error) { - raw, err := call("seat:"+mma.Seat+"."+mma.VerbAssignments, mma.AssignmentsInput{Seats: mma.AssignmentsInputSeatsTrue}) - if err != nil { - return "", fmt.Errorf("who holds the seat %s is not asked of the seat %s: %w", oc.Seat, mma.Seat, err) - } - var held struct { - Seats []mma.SeatHolders `json:"seats"` - } - if err := json.Unmarshal(raw, &held); err != nil { - return "", fmt.Errorf("the answer of %s.%s is not its seat's: %w", mma.Seat, mma.VerbAssignments, err) - } - for _, s := range held.Seats { - if s.Seat == oc.Seat && len(s.Holders) > 0 { - return s.Holders[0].Module + "_" + oc.StateAsks, nil - } - } - return "", fmt.Errorf("no module holds the seat %s on the seat list, so the router's record of an ask is not read", oc.Seat) -} - -// routerRecordOf is Hearer.Where: the router's own record of an ask, read under the runner's module's name through -// operatorchannelclient, whose Record is a get of that key (`.` of the seat's state asks) on the process's own bus -// connection (design 47 G §700, A-89): never the decided event's body. +// routerRecordOf is Hearer.Where: the router's own record of an ask, read under the runner's module's name by mesh-sdk's +// operatorask (design 47 G A-101, §700, A-89): the state asks of the seat operator-channel in the bucket of the seat's holder +// on the seat list, which the seat mesh-module-assigner's verb assignments answers with seats; record is a get of the key +// `.` on the process's own bus connection: never the decided event's body. func routerRecordOf(module string, call toolCaller, record func(bucket, id string) (*stdio.StateEntry, error)) func(id string) (string, *oc.Warrant, error) { - client := &operatorchannelclient.Client{Module: module, Record: func(id string) (*stdio.StateEntry, error) { - bucket, err := routerBucket(call) - if err != nil { - return nil, err - } - return record(bucket, id) - }} - return client.Where + asker := &operatorask.Asker[oc.Warrant]{Module: module, Call: operatorask.Caller(call), Record: record, + Contract: operatorask.Contract{Lists: "seat:" + mma.Seat + "." + mma.VerbAssignments, ListBody: mma.AssignmentsInput{Seats: mma.AssignmentsInputSeatsTrue}, + Seat: oc.Seat, StateAsks: oc.StateAsks}} + return asker.Where } // busAsker publishes the runner's approval ask on the seat operator-channel's accept subject under the runner's own name, its 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 1b25805e..8270d794 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,15 @@ import ( "context" "encoding/json" "errors" + mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" 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" "github.com/nats-io/nats.go/jetstream" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/operatorask" ) // The teller's three bus adapters for mesh-sdk go/asker, which in a bundle reaches the bus through the runtime: here the @@ -20,9 +20,6 @@ import ( // the seat operator-channel (mesh-control internal/broker seattraffic.go SeatTrafficOf): publish its own ask, cancel and // condition, hear its own decided event, and read the router's record of its own asks. -// callTimeout bounds a read of the router's record. -const callTimeout = 30 * time.Second - // publishVia publishes an accept of the seat, de-duplicated by id: the same id twice is one message. func publishVia(ctx context.Context, js jetstream.JetStream) func(subject string, body any, id string) error { return func(subject string, body any, id string) error { @@ -39,30 +36,20 @@ func publishVia(ctx context.Context, js jetstream.JetStream) func(subject string } } -// messengerAsks is the router's records bucket as the seat operator-channel's manifest declares it (its `records`), the -// one the controller composes this account's direct get on (mesh-sdk buscompose SeatTrafficOf): never `asks`, a bucket -// no grant names (issue 774). The SDK names no constant for it. -const messengerAsks = "messenger_asks" - -// recordVia reads the router's record of one of this module's asks: the key `.` of the router's bucket -// messenger_asks, by a direct get on the subject the controller grants for it, -// `$JS.API.DIRECT.GET.KV_messenger_asks.$KV.messenger_asks..>`. The record is {state, warrant}. An ask the -// router holds no record of is nil. -func recordVia(ctx context.Context, conn *nats.Conn, module string) func(id string) (*stdio.StateEntry, error) { +// recordVia reads the router's record of one of this module's asks under the module's own name: the key `.` of the +// state asks of the seat operator-channel, in the bucket of the seat's holder on the seat list (mesh-sdk's operatorask, design +// 47 G A-101, A-74); no bucket's name is in the program. The record is {state, warrant}; an ask the router holds no record +// of is nil. call asks the seat list's verb (process.Conn.Call) and ask is the get of the record (process.Conn.Ask), which the +// controller grants for a module that uses the seat. +func recordVia(call operatorask.Caller, ask func(bucket, id string) (*stdio.StateEntry, error)) func(id string) (*stdio.StateEntry, error) { + contract := operatorask.Contract{Lists: "seat:" + mma.Seat + "." + mma.VerbAssignments, ListBody: mma.AssignmentsInput{Seats: mma.AssignmentsInputSeatsTrue}, + Seat: oc.Seat, StateAsks: oc.StateAsks} return func(id string) (*stdio.StateEntry, error) { - ctx, cancel := context.WithTimeout(ctx, callTimeout) - defer cancel() - msg, err := conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET.KV_"+messengerAsks+".$KV."+messengerAsks+"."+module+"."+id, nil) - if errors.Is(err, nats.ErrNoResponders) { - return nil, nil - } + bucket, err := operatorask.Bucket(call, contract) if err != nil { return nil, err } - if status := msg.Header.Get("Status"); status == "404" || len(msg.Data) == 0 { - return nil, nil - } - return &stdio.StateEntry{Key: id, Value: msg.Data}, nil + return ask(bucket, id) } } 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 636318ee..2ae5d2ad 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 @@ -8,6 +8,7 @@ import ( "fmt" 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/views/viewstest" "os" "path/filepath" @@ -178,13 +179,16 @@ const ( 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" + // routerAsksBucket is the router's state asks in the bucket of the seat operator-channel's holder on the seat list (here ops-router): + // the holder's module name and the seat's state, never a name in the program (design 47 G A-101, A-74). + routerAsksBucket = "ops-router_asks" ) func fullConf(dir string) string { // ends is the teller's own state (manifest state entry ends), which the controller makes before it starts: a bed // 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.>") + seatAsk, seatCancel, seatCondition, "$JS.API.DIRECT.GET.KV_"+routerAsksBucket+".$KV."+routerAsksBucket+".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+".>") // The finding writer finds the stream of the keeper's findings-asked (design 47 G §384): the bed has none, so it is @@ -261,7 +265,7 @@ func startFull(t *testing.T) *fullBed { } ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) t.Cleanup(cancel) - for name, history := range map[string]uint8{gateAsksKV: gateAsksHistory, tellerAnswers: answersHistory, tellerAsked: 1, ends.Bucket: 1, messengerAsks: 1, assignerAsks: assignerAsksHistory, viewAnswers: viewAnswersHistory, firstViewsBucket: 1, findingsBucket: 1} { + for name, history := range map[string]uint8{gateAsksKV: gateAsksHistory, tellerAnswers: answersHistory, tellerAsked: 1, ends.Bucket: 1, routerAsksBucket: 1, assignerAsks: assignerAsksHistory, viewAnswers: viewAnswersHistory, firstViewsBucket: 1, findingsBucket: 1} { if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history}); err != nil { t.Fatal(err) } @@ -376,7 +380,7 @@ func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2 t.Cleanup(func() { rn.finish(t, 10*time.Second) }) // Issue 774: the router's record is read on the teller's own credential from the bucket the controller grants it. - if mkv, err := b.js.KeyValue(b.ctx, messengerAsks); err != nil { + if mkv, err := b.js.KeyValue(b.ctx, routerAsksBucket); err != nil { t.Fatal(err) } else if _, err := mkv.Put(b.ctx, "mesh-delivery-teller.r1", []byte(`{"state":"open"}`)); err != nil { t.Fatal(err) @@ -386,7 +390,20 @@ func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2 t.Fatal(err) } t.Cleanup(tc.Close) - if rec, err := recordVia(b.ctx, tc, "mesh-delivery-teller")("r1"); err != nil || rec == nil { + seatList := func(key string, _ any) (json.RawMessage, error) { + if key != "seat:mesh-module-assigner.assignments" { + t.Errorf("the router's holder was asked of %s, want the seat mesh-module-assigner's assignments", key) + } + return json.RawMessage(`{"seats":[{"seat":"operator-channel","scope":"mesh","holders":[{"module":"ops-router","node":"n2"}]}]}`), nil + } + getRecord := func(bucket, id string) (*stdio.StateEntry, error) { + msg, err := tc.RequestWithContext(b.ctx, "$JS.API.DIRECT.GET.KV_"+bucket+".$KV."+bucket+".mesh-delivery-teller."+id, nil) + if err != nil { + return nil, err + } + return &stdio.StateEntry{Key: id, Value: msg.Data}, nil + } + if rec, err := recordVia(seatList, getRecord)("r1"); err != nil || rec == nil { t.Fatalf("the router's record read as the teller: %v, %v; want it", rec, err) } 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 3342bd6b..34ef4bc2 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go @@ -76,8 +76,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // started once both are registered. hub := process.NewEventHub(logger.Printf) defer hub.Stop() + // The assigner's asks are its state, read by the first view of its read tool and then its views family, never its + // bucket (ADR 0615 §1, DLV-M3): the picture reads as the bucket did. + calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) + if err != nil { + return fmt.Errorf("the teller's connection for its tool calls: %w", err) + } + defer calls.Close() client := &occ.Client{Module: tellerModule, State: state, Publish: publish, - Record: recordVia(ctx, conn, tellerModule), Subscribe: decidedVia(hub.Handle, logger.Printf)} + Record: recordVia(calls.Call, calls.Ask), 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, tellerModule), Logf: logger.Printf, Publish: publish, // The teller's own events (its manifest's emits), on EVENTS under its module's subject. @@ -92,13 +99,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The view's answers to the assigner's asks (issue 1297 B10): both buckets are the controller's to create, and a bucket not // there is said and leaves the teller without that path rather than without the gate's asks (a wait for it needs a timer). failed := make(chan error, 4) - // The assigner's asks are its state, read by the first view of its read tool and then its views family, never its - // bucket (ADR 0615 §1, DLV-M3): the picture reads as the bucket did. - calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) - if err != nil { - return fmt.Errorf("the teller's connection for its tool calls: %w", err) - } - defer calls.Close() t.Answered = func(key, askID string, answer []byte) error { return calls.Emit(answeredEvent, Answered{Key: key, Ask: askID, Answer: answer}, stdio.EmitOptions{MessageID: answeredID(key, askID)}) }