Serve the put as the assigner's own tools from the compositions and grants the controller already writes, with no seat and no compose verb: the controller takes no change (ADR 0546)

This commit is contained in:
2026-10-25 21:36:38 +01:00
parent 2f742fd3fe
commit 45fae9da8a
13 changed files with 186 additions and 207 deletions
@@ -30,9 +30,9 @@ import (
func main() {
if os.Getenv("MESH_BROKER_FILE") == "" {
// Started by the tool runner, with no credential of its own: the seat's verbs (ADR 0546), served as every seat a module
// holds is, since a process never serves (ADR 0436). The process the manifest runs names its credential.
if err := serveSeat(); err != nil {
// Started by the tool runner, with no credential of its own: the put's tools (ADR 0546), served through the runtime,
// since a process never serves (ADR 0436). The process the manifest runs names its credential.
if err := serveTools(); err != nil {
fmt.Fprintln(os.Stderr, "mesh-delivery-assigner:", err)
os.Exit(1)
}
@@ -130,7 +130,7 @@ func (m *memKV) Update(_ context.Context, key string, value []byte, revision uin
return uint64(len(value)), nil
}
// memStore is the runtime's states for the seat's verbs, over the same memKVs a pass writes.
// memStore is the runtime's states for the put's tools, over the same memKVs a pass writes.
type memStore map[string]*memKV
func (s memStore) get(state, key string) ([]byte, uint64, bool, error) {
@@ -167,8 +167,8 @@ func (b *fakeBus) Emit(eventType string, body any) error {
return nil
}
// ADR 0546, the first put (2026-10-25, want.novox.nats): the verb writes the put; a pass asks the controller's compose verb;
// once the composition and grant are there it asks the operator with a yes that approves; only on that warrant does it have
// ADR 0546, the first put (2026-10-25, want.novox.nats): the tool writes the put; a pass waits for the composition and grant
// the controller's take-in writes; once they are there it asks the operator with a yes that approves; only on that warrant does it have
// the streams snapshotted, then writes the want held and records the warrant; the have running it healthy holds it; a release
// gives the want back to the rules. Each step is said as the event put.
func putOfTheBus(t *testing.T) {
@@ -213,7 +213,7 @@ func putOfTheBus(t *testing.T) {
builds.values[bkey] = built
modules := mem()
modules.values["nats"] = mod
s := newSeat(memStore{putsState: puts, delivery.BuilderModule + ".builds": builds, delivery.IndexerModule + ".modules": modules})
s := newPutTools(memStore{putsState: puts, delivery.BuilderModule + ".builds": builds, delivery.IndexerModule + ".modules": modules})
s.now = func() time.Time { return now }
was := answerWithin
answerWithin = 0
@@ -245,13 +245,14 @@ func putOfTheBus(t *testing.T) {
}
return got
}
if got := step(); got.Step != delivery.PutCompose {
if got := step(); got.Step != delivery.PutCompose || !strings.Contains(got.Words, "no composition") {
t.Fatalf("the first pass left the put at %s: %s", got.Step, got.Words)
}
if call := <-bus.calls; !strings.HasPrefix(call, "seat:mesh-controller.compose") || !strings.Contains(call, fp) {
t.Errorf("the controller was asked %q", call)
select {
case call := <-bus.calls:
t.Errorf("a put asked %q: it reads the compositions and asks nobody for one", call)
default:
}
<-woke
ckey, _ := delivery.ComposedKey("novox", "nats", fp)
cd, _ := delivery.EncodeComposedDigest(delivery.ComposedDigest{Digest: digest, Judgement: delivery.JudgedCannot, Words: "the have names no composition"})
world(delivery.ControllerComposedBucket, ckey, cd)
@@ -16,12 +16,12 @@ import (
"mesh-delivery-assigner/internal/assign"
)
// The puts (novox/hq ADR 0546): the seat's verb `put` writes the key `<node>.<module>` of the assigner's puts, and each pass
// takes every put that has not ended one step further (assign.StepPut), doing the acts the step asks for: the controller's
// compose verb, the put's ask, the bus's snapshot, the warranted record. Each act's outcome is heard back as a change of the
// world (the composition, the grant, the answer) or, for the calls, as a wake once they answer; nothing polls but the
// snapshot's read-back of the backup holder, as the controller's bus step reads it (ADR 0235), bounded. Every step is said as
// the event mesh-delivery-assigner.put, so a caller waits on it.
// The puts (novox/hq ADR 0546): the tool `assigner_put` writes the key `<node>.<module>` of the assigner's puts, and each pass
// takes every put that has not ended one step further (assign.StepPut), doing the acts the step asks for: the put's ask, the
// bus's snapshot, the warranted record. Each act's outcome is heard back as a change of the world (the composition, the grant,
// the answer) or, for the calls, as a wake once they answer; nothing polls but the snapshot's read-back of the backup holder,
// as the controller's bus step reads it (ADR 0235), bounded. Every step is said as the event mesh-delivery-assigner.put, so a
// caller waits on it.
// PutEvent is the event a put's every step is said as: mesh.mod.mesh-delivery-assigner.event.put.
const PutEvent = "put"
@@ -35,14 +35,12 @@ type caller interface {
}
// putActs keeps, by put (its key and when it was asked), what the acts of a put came to while this process runs: a restart
// asks the controller and the snapshot again, which compose and back up again without harm.
// takes the snapshot again, without harm.
type putActs struct {
mu sync.Mutex
bus caller
wake func()
logf func(string, ...any)
composing map[string]time.Time
composeNo map[string]string
snapping map[string]bool
snapped map[string]string
snapNo map[string]string
@@ -53,8 +51,7 @@ type putActs struct {
}
func newPutActs(bus caller, wake func(), logf func(string, ...any)) *putActs {
a := &putActs{bus: bus, wake: wake, logf: logf, composing: map[string]time.Time{}, composeNo: map[string]string{},
snapping: map[string]bool{}, snapped: map[string]string{}, snapNo: map[string]string{}, warranted: map[string]bool{},
a := &putActs{bus: bus, wake: wake, logf: logf, snapping: map[string]bool{}, snapped: map[string]string{}, snapNo: map[string]string{}, warranted: map[string]bool{},
said: map[string]string{}}
a.snapshot = a.snapshotTheBus
return a
@@ -118,12 +115,12 @@ func (p *passer) puts(ctx context.Context) {
key, _ := delivery.PutKey(node, module)
id := identity(key, put)
world := p.putWorld(node, module, put, manifest, assigned[node])
world.ComposeAsked, world.ComposeFailed, world.Snapshot, world.Snapshotting, world.SnapshotFailed = p.acts.state(id)
world.Snapshot, world.Snapshotting, world.SnapshotFailed = p.acts.state(id)
next := assign.StepPut(world)
// A held bus whose user list the rules moved on (ADR 0539 §7): the put follows the composition its want names.
if w, ok := wanted[module][node]; ok && w.Put && w.FP == next.Put.FP && w.Composed != "" && w.Composed != next.Put.Composed &&
(next.Put.Step == delivery.PutWanted || next.Put.Step == delivery.PutHolding) {
if d, there := p.w.composed(node, module, w.FP); there && d.UserListOnly && d.Digest == w.Composed {
if d, _ := p.composedFor(node, module, w.FP); d != nil && d.UserListOnly && d.Digest == w.Composed {
next.Put = next.Put.Said(world.Now, next.Put.Step, fmt.Sprintf("the bus's user list changed: the want names composition %s", shortHex(w.Composed)))
next.Put.Composed = w.Composed
}
@@ -167,8 +164,8 @@ func (p *passer) putWorld(node, module string, put delivery.Put, manifest catalo
}
}
}
if d, ok := p.w.composed(node, module, put.FP); ok {
w.Composed = &d
w.Composed, w.Granted = p.composedFor(node, module, put.FP)
if d := w.Composed; d != nil {
policy := d.Policy
if policy == "" {
policy, _, _ = catalogue.DerivedUpgrade(manifest)
@@ -183,13 +180,6 @@ func (p *passer) putWorld(node, module string, put delivery.Put, manifest catalo
}
}
}
if gkey, err := delivery.ControllerGrantKey(module, put.FP); err == nil {
if raw := p.w.value(delivery.ControllerGrantsBucket, gkey); raw != nil {
if g, err := delivery.DecodeControllerGrant(raw); err == nil {
w.Granted = g.NamesNode(node, module)
}
}
}
if hkey, err := delivery.HaveKey(node, module); err == nil {
if raw := p.w.value(delivery.HaveBucket, hkey); raw != nil {
if h, err := delivery.DecodeHave(raw); err == nil {
@@ -200,11 +190,28 @@ func (p *passer) putWorld(node, module string, put delivery.Put, manifest catalo
return w
}
// composedFor is the composition of module for node at fp and whether the grant at fp names the node's accounts: the one
// place a put reads them (ADR 0546 clause 4). Today their writer is the controller (mesh-controller_composed and
// mesh-controller_grants); when composing and grants move out of the controller into delivery's own parts, this reads them
// from their new owner and nothing else changes.
func (p *passer) composedFor(node, module, fp string) (*delivery.ComposedDigest, bool) {
var out *delivery.ComposedDigest
if d, ok := p.w.composed(node, module, fp); ok {
out = &d
}
granted := false
if gkey, err := delivery.ControllerGrantKey(module, fp); err == nil {
if raw := p.w.value(delivery.ControllerGrantsBucket, gkey); raw != nil {
if g, err := delivery.DecodeControllerGrant(raw); err == nil {
granted = g.NamesNode(node, module)
}
}
}
return out, granted
}
// actOn does the acts a step asks for.
func (p *passer) actOn(ctx context.Context, node, module, id string, w assign.PutWorld, next assign.PutNext) {
if next.Compose {
p.acts.compose(ctx, id, node, module, next.Put.FP)
}
if next.Ask && w.Composed != nil {
akey := next.Put.Ask
if !p.w.has(delivery.AssignerAsksBucket, akey) {
@@ -229,7 +236,7 @@ func samePut(a, b delivery.Put) bool {
return a.Step == b.Step && a.Words == b.Words && a.Composed == b.Composed && a.Ask == b.Ask && a.Snapshot == b.Snapshot
}
// writePut writes a put's next value over the one the pass read, by compare-and-set, unless the verb has put again or
// writePut writes a put's next value over the one the pass read, by compare-and-set, unless the tool has put again or
// released it since; and says it as the event put.
func (p *passer) writePut(ctx context.Context, key string, read, next delivery.Put) {
kv := p.kvs[delivery.AssignerPutsBucket]
@@ -262,55 +269,10 @@ func (p *passer) writePut(ctx context.Context, key string, read, next delivery.P
}
// state is what this process knows of a put's acts.
func (a *putActs) state(id string) (asked time.Time, composeNo, snapshot string, snapping bool, snapNo string) {
func (a *putActs) state(id string) (snapshot string, snapping bool, snapNo string) {
a.mu.Lock()
defer a.mu.Unlock()
return a.composing[id], a.composeNo[id], a.snapped[id], a.snapping[id], a.snapNo[id]
}
// composeAnswer is what the controller's compose verb answers inside its seat's answer ({output, ok, answer}).
type composeAnswer struct {
OK bool `json:"ok"`
Output string `json:"output"`
Answer *struct {
Digest string `json:"digest"`
Granted bool `json:"granted"`
Grant string `json:"grant"`
} `json:"answer"`
}
// compose asks the controller's compose verb once for a put. Its answer wakes a pass; the composition and the grant are
// read from their buckets, and a refusal is kept as the put's words.
func (a *putActs) compose(ctx context.Context, id, node, module, fp string) {
a.mu.Lock()
if _, asked := a.composing[id]; asked {
a.mu.Unlock()
return
}
a.composing[id] = time.Now()
a.mu.Unlock()
go func() {
defer a.wake()
raw, err := a.bus.Call("seat:"+catalogue.ControllerSeatName+".compose", map[string]any{"node": node, "module": module, "fp": fp})
why := ""
var ans composeAnswer
switch {
case err != nil:
why = err.Error()
case json.Unmarshal(raw, &ans) != nil:
// An answer of another shape: the controller still running it after its answer window. The buckets say the rest.
case !ans.OK && strings.TrimSpace(ans.Output) != "":
why = strings.TrimSpace(ans.Output)
case ans.Answer != nil && ans.Answer.Digest != "" && !ans.Answer.Granted:
why = ans.Answer.Grant
}
if why != "" {
a.mu.Lock()
a.composeNo[id] = why
a.mu.Unlock()
a.logf("the controller's compose of %s on %s at %s: %s", module, node, fp, why)
}
}()
return a.snapped[id], a.snapping[id], a.snapNo[id]
}
// snap has the bus's snapshot taken once for a put, in the background.
@@ -6,35 +6,36 @@ import (
"fmt"
"os"
"regexp"
"slices"
"sort"
"strings"
"sync"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// The assigner's seat (novox/hq ADR 0546), served through the tool runner as every seat a module holds is: a process never
// serves (ADR 0436, mesh-sdk process). The verbs write and read the puts; the assigner's process takes each put further
// (puts.go) and says every step as the event put.
// The assigner's tools (novox/hq ADR 0546), served through the tool runner from its bundle: a process never serves (ADR
// 0436, mesh-sdk process). They write and read the puts; the assigner's process takes each put further (puts.go) and says
// every step as the event put. They are the module's own tools, not a seat's verbs: a new seat needs a change of the
// controller, which takes none; they become the verbs of a seat once composing leaves the controller.
// putsState is the assigner's state of puts, as the runtime names a module's own state.
const putsState = "puts"
// answerWithin is how long `put` waits for the assigner's process to take the put up before it answers what stands.
// answerWithin is how long assigner_put waits for the assigner's process to take the put up before it answers what stands.
var answerWithin = 20 * time.Second
// store is what the verbs need of the states: the runtime's (stdio.State), or a test's.
// store is what the tools need of the states: the runtime's (stdio.State), or a test's.
type store interface {
get(state, key string) ([]byte, uint64, bool, error)
put(state, key string, value []byte) error
keys(state string) ([]string, error)
}
// seat serves the verbs over a store, and keeps the newest value of every put it heard, to wait on.
type seat struct {
// putTools serves the tools over a store, and keeps the newest value of every put it heard, to wait on.
type putTools struct {
st store
now func() time.Time
mu sync.Mutex
@@ -43,12 +44,12 @@ type seat struct {
changed chan struct{}
}
func newSeat(st store) *seat {
return &seat{st: st, now: time.Now, heard: map[string][]byte{}, changed: make(chan struct{})}
func newPutTools(st store) *putTools {
return &putTools{st: st, now: time.Now, heard: map[string][]byte{}, changed: make(chan struct{})}
}
// hear takes a put's change in from the watch.
func (s *seat) hear(key string, value []byte) {
func (s *putTools) hear(key string, value []byte) {
s.mu.Lock()
s.heard[key] = value
close(s.changed)
@@ -64,7 +65,7 @@ var (
// version reads the version a put names: newest (the indexer's fingerprint at the trunk), a whole fingerprint, or the start
// of one the builder built, which must name one build alone.
func (s *seat) version(module, version string) (string, error) {
func (s *putTools) version(module, version string) (string, error) {
v := strings.ToLower(strings.TrimSpace(version))
switch {
case v == "newest":
@@ -110,7 +111,7 @@ func (s *seat) version(module, version string) (string, error) {
return "", fmt.Errorf("%q is no version: newest, a fingerprint (src2: and 64 hex digits) or at least its first 12 hex digits", version)
}
// putAnswer is what `put` and `puts` answer of one put.
// putAnswer is what assigner_put and assigner_puts answer of one put.
type putAnswer struct {
Node string `json:"node"`
Module string `json:"module"`
@@ -118,8 +119,8 @@ type putAnswer struct {
Follow string `json:"follow,omitempty"`
}
// doPut is the verb put.
func (s *seat) doPut(args map[string]any, by string) (any, error) {
// doPut is the tool assigner_put.
func (s *putTools) doPut(args map[string]any, by string) (any, error) {
node, module, why := arg(args, "node"), arg(args, "module"), arg(args, "why")
key, err := delivery.PutKey(node, module)
if err != nil {
@@ -165,7 +166,7 @@ func (s *seat) doPut(args map[string]any, by string) (any, error) {
}
// await is the put at key once the assigner's process took it up past its asking, or as it stands after answerWithin.
func (s *seat) await(key string, p delivery.Put) delivery.Put {
func (s *putTools) await(key string, p delivery.Put) delivery.Put {
deadline := time.After(answerWithin)
for {
s.mu.Lock()
@@ -188,11 +189,11 @@ func (s *seat) await(key string, p delivery.Put) delivery.Put {
// followPut says how a caller follows a put after the answer.
func followPut(node, module string) string {
return fmt.Sprintf("the put goes on: mesh_wait on mesh-delivery-assigner.put with filter {\"node\": %q, \"module\": %q}, then "+
"mesh-delivery-assigner.puts for where it stands", node, module)
"mesh-delivery-assigner.assigner_puts for where it stands", node, module)
}
// doRelease is the verb release.
func (s *seat) doRelease(args map[string]any, by string) (any, error) {
// doRelease is the tool assigner_release.
func (s *putTools) doRelease(args map[string]any, by string) (any, error) {
node, module, why := arg(args, "node"), arg(args, "module"), arg(args, "why")
key, err := delivery.PutKey(node, module)
if err != nil {
@@ -227,8 +228,8 @@ func (s *seat) doRelease(args map[string]any, by string) (any, error) {
return putAnswer{Node: node, Module: module, Put: p}, nil
}
// doPuts is the verb puts.
func (s *seat) doPuts(args map[string]any) (any, error) {
// doPuts is the tool assigner_puts.
func (s *putTools) doPuts(args map[string]any) (any, error) {
keys, err := s.st.keys(putsState)
if err != nil {
return nil, fmt.Errorf("the puts cannot be listed: %w", err)
@@ -263,21 +264,55 @@ func arg(args map[string]any, name string) string {
return ""
}
// tools are the seat's verbs as the runtime serves them.
func (s *seat) tools() []stdio.Tool {
verbs := map[string]catalogue.Verb{}
if seat, ok := catalogue.SeatNamed(catalogue.AssignerSeat); ok {
for _, v := range seat.Serves {
verbs[v.Name] = v
// The tools' names, as the manifest lists them.
const (
toolPut = "assigner_put"
toolRelease = "assigner_release"
toolPuts = "assigner_puts"
)
// schema is a tool's arguments, each a string: required ones first named, and switches true or false.
func schema(properties map[string]string, required []string, switches ...string) map[string]any {
props := map[string]any{}
for name, description := range properties {
p := map[string]any{"type": "string", "description": description}
if slices.Contains(switches, name) {
p["enum"] = []string{"true", "false"}
}
props[name] = p
}
tool := func(name string, run func(map[string]any) (any, error)) stdio.Tool {
return stdio.Tool{Name: catalogue.AssignerSeat + "." + name, Description: verbs[name].Description, Input: verbs[name].Input, Run: run}
out := map[string]any{"type": "object", "properties": props}
if len(required) > 0 {
out["required"] = required
}
return out
}
// tools are the assigner's tools as the runtime serves them (ADR 0546 clause 1).
func (s *putTools) tools() []stdio.Tool {
return []stdio.Tool{
tool("put", func(a map[string]any) (any, error) { return s.doPut(a, stdio.CallerOf(a)) }),
tool("release", func(a map[string]any) (any, error) { return s.doRelease(a, stdio.CallerOf(a)) }),
tool("puts", s.doPuts),
{Name: toolPut, Description: "Put one version of a module on one machine (novox/hq ADR 0546): the build at that fingerprint, " +
"the composition and grant the controller wrote for that machine, then the machine's want naming it, held there until " +
"released when hold is true. A module whose moves a person takes (the bus, a module of policy record) is asked of the " +
"operator first, and the bus has its streams snapshotted after the yes. Answers what was done or what it waits on; the " +
"put goes on after the answer, each step said as the event mesh-delivery-assigner.put, and assigner_puts says where it stands.",
Input: schema(map[string]string{
"node": "the machine, where the module is assigned",
"module": "the module",
"version": "newest, a fingerprint (src2: and 64 hex digits), or at least its first 12 hex digits",
"hold": "\"true\": the want stays at this version until released; \"false\" (the default): the rules move it again once it runs",
"why": "why, kept with the put and said in any ask",
}, []string{"node", "module", "version", "why"}, "hold"),
Run: func(a map[string]any) (any, error) { return s.doPut(a, stdio.CallerOf(a)) }},
{Name: toolRelease, Description: "End a put on one machine (novox/hq ADR 0546): its hold is taken away and the assigner's " +
"rules write the machine's want again from their next pass. Nothing changes on the machine until they move it.",
Input: schema(map[string]string{"node": "the machine", "module": "the module", "why": "why, kept with the put"},
[]string{"node", "module", "why"}),
Run: func(a map[string]any) (any, error) { return s.doRelease(a, stdio.CallerOf(a)) }},
{Name: toolPuts, Description: "Every put and where it stands (novox/hq ADR 0546): its version, hold, why, each step with " +
"its time and words. Narrowed by machine or module. Read-only.",
Input: schema(map[string]string{"node": "one machine", "module": "one module"}, nil),
Run: s.doPuts},
}
}
@@ -299,9 +334,9 @@ func (runtimeStore) put(state, key string, value []byte) error {
func (runtimeStore) keys(state string) ([]string, error) { return stdio.State(state).Keys() }
// serveSeat serves the seat's verbs to the runtime until it stops, hearing every put to answer `put` with its first steps.
func serveSeat() error {
s := newSeat(runtimeStore{})
// serveTools serves the tools to the runtime until it stops, hearing every put to answer assigner_put with its first steps.
func serveTools() error {
s := newPutTools(runtimeStore{})
go func() {
<-stdio.Served()
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {