A-89: the runner asks its approval and reads the warrant from the router's record through the seat holder

Builds: design 47 G §37, §700, A-74, A-89 (routerRecordOf copied from the indexer's b0d2841f, to move into mesh-sdk)

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-11 23:26:22 +01:00
parent 7ee394b7d9
commit 56a6c64586
6 changed files with 161 additions and 3 deletions
@@ -0,0 +1,107 @@
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/process"
"mesh-delivery-pipeline/internal/pipeline"
)
// 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 (`<module>.<id>` 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.
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
}
// busAsker publishes the runner's approval ask on the seat operator-channel's accept subject under the runner's own name, its
// duplicate id the ask's own, so an ask published twice is kept once (§37).
type busAsker struct {
conn *nats.Conn
cred process.Account
}
func (b busAsker) Ask(a oc.Ask) error {
data, err := json.Marshal(a)
if err != nil {
return err
}
msg := nats.NewMsg(oc.AcceptSubjectAsk(b.cred.Module))
msg.Header.Set("x-event-id", a.ID)
msg.Header.Set("x-source", b.cred.Module)
msg.Header.Set("x-node", b.cred.Node)
msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
msg.Header.Set("content-type", "application/json")
msg.Header.Set(nats.MsgIdHdr, a.ID)
msg.Data = data
return b.conn.PublishMsg(msg)
}
var _ pipeline.Asker = busAsker{}
// hearDecided wakes the runner on the operator-channel's decided for its own asks (§37, A-89): the warrant is read from the
// router's record, so an event the record does not back yet is heard again.
func hearDecided(hub *process.EventHub, h *pipeline.Hearer, module string, logf func(string, ...any)) error {
return hub.Handle(oc.SubjectDecided(module), func(m jetstream.Msg) {
var ev oc.Warrant
if err := json.Unmarshal(m.Data(), &ev); err != nil {
logf("%s holds what is no decided event, dropped: %v", m.Subject(), err)
_ = m.Term()
return
}
_, err := h.OnDecided(ev)
switch {
case errors.Is(err, pipeline.ErrRefused), errors.Is(err, pipeline.ErrNoAsk):
logf("the decided of ask %s takes nothing: %v", ev.Ask, err)
_ = m.Term()
case err != nil:
logf("the decided of ask %s is not taken yet, heard again: %v", ev.Ask, err)
_ = m.Nak()
default:
_ = m.Ack()
}
})
}
@@ -74,7 +74,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return err
}
starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf, Say: say}
if err := hearStepEnded(hub, newHearer(ctx, js, cred, pipelines, kept, say, logger.Printf), logger.Printf); err != nil {
hearer := newHearer(ctx, js, cred, pipelines, kept, say, logger.Printf)
hearer.Asker = busAsker{conn: conn, cred: cred}
hearer.Where = routerRecordOf(cred.Module, calls.Call, calls.Ask)
if err := hearStepEnded(hub, hearer, logger.Printf); err != nil {
return err
}
if err := hearDecided(hub, hearer, cred.Module, logger.Printf); err != nil {
return err
}
if err := hearPublished(ctx, hub, starter, calls, rv.Catalog, logger.Printf); err != nil {
@@ -0,0 +1,37 @@
package main
import (
"encoding/json"
"strings"
"testing"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// Design 47 G §700, A-74: the router's record of an ask is read under the indexer's module's name from the state asks of the
// seat operator-channel, in the bucket of the seat's holder on the seat list, and carries the ask's warrant.
func TestTheRoutersRecordIsReadFromTheBucketOfTheSeatsHolder(t *testing.T) {
var asked []string
call := func(key string, body any) (json.RawMessage, error) {
asked = append(asked, key)
return json.RawMessage(`{"seats":[{"seat":"node-packet-filter","scope":"node","node":"n1","holders":[{"module":"pf","node":"n1"}]},` +
`{"seat":"operator-channel","scope":"mesh","holders":[{"module":"ops-router","node":"n2"}]}],"holds":2}`), nil
}
var bucket, id string
record := func(b, i string) (*stdio.StateEntry, error) {
bucket, id = b, i
return &stdio.StateEntry{Key: "mesh-delivery-pipeline." + i, Value: json.RawMessage(`{"state":"chosen","warrant":{"ask":"a1","asker":"mesh-delivery-pipeline","choice":"approve"}}`)}, nil
}
state, w, err := routerRecordOf("mesh-delivery-pipeline", call, record)("a1")
if err != nil || state != "chosen" || w == nil || w.Ask != "a1" {
t.Fatalf("Where = %q, %+v, %v; want the state chosen and the warrant of a1", state, w, err)
}
if bucket != "ops-router_asks" || id != "a1" || len(asked) != 1 || asked[0] != "seat:mesh-module-assigner.assignments" {
t.Errorf("read %q key %q after asking %v; want the holder's bucket ops-router_asks, the id, one question of the seat mesh-module-assigner", bucket, id, asked)
}
// No holder on the seat list: nothing is read, and the warrant stays unapplied with the reason said.
none := func(string, any) (json.RawMessage, error) { return json.RawMessage(`{"seats":[],"holds":0}`), nil }
if _, _, err := routerRecordOf("mesh-delivery-pipeline", none, record)("a1"); err == nil || !strings.Contains(err.Error(), "no module holds the seat operator-channel") {
t.Errorf("with no holder Where answered %v; want the reason", err)
}
}
+1
View File
@@ -9,6 +9,7 @@ require (
)
require (
git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000
@@ -68,7 +68,10 @@ func TestManifest(t *testing.T) {
// asked again by hand (a hand act).
{"invokes", equalSets(m.Invokes, []string{"seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-hand-acts.hand-act",
// RN3 (§53, §334): a publication's dependents behind.
"seat:mesh-catalog.outdated"})},
"seat:mesh-catalog.outdated",
// A-74, A-89: who holds the seat operator-channel, whose state asks holds the router's record of the approval ask.
"seat:mesh-module-assigner.assignments"})},
{"uses operator-channel", equalSets(m.Uses, []string{"operator-channel"})},
{"first-view tools", equalSets(m.Tools, []string{"pipeline_first_views", "pipeline_first_view_again"})},
// RN1/AC5 (§95, §130): the new assigner is delivered before the runner.
{"program needs the assigner's program", slices.Equal(program.Needs, []string{"mesh-delivery-assigner:program"})},
+5 -1
View File
@@ -43,7 +43,11 @@
"seat:mesh-catalog.registrations",
"seat:mesh-catalog.revision",
"seat:mesh-hand-acts.hand-act",
"seat:mesh-catalog.outdated"
"seat:mesh-catalog.outdated",
"seat:mesh-module-assigner.assignments"
],
"uses": [
"operator-channel"
],
"state": [
{