A-204: the assigner keeps the put waits it said in its state put-waits

A put that ends while the assigner is stopped now gets its wait-ended at the
next start: the said waits are read from the bus, not seeded from the puts
that wait now.

Builds: design 47 G §498, A-204, A-211

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-12 06:03:58 +01:00
parent f86ccc378c
commit d3e555e2f4
7 changed files with 89 additions and 67 deletions
@@ -80,7 +80,7 @@ var readFamilies = []string{buckets.ComposedRead}
var ownBuckets = []string{buckets.Asks, buckets.Differences,
buckets.Puts, buckets.Wants, plc.AssignmentsBucket, plc.ProposalsBucket,
plc.CandidatesBucket, plc.PulsesBucket, assign.FindingsBucket, assign.PoliciesBucket, plc.SeatHoldersBucket,
engineReadsBucket, retiredBucket, copiesBucket, writtenBucket, buckets.Waits, buckets.Inputs, pinsBucket}
engineReadsBucket, retiredBucket, copiesBucket, writtenBucket, buckets.Waits, buckets.PutWaits, buckets.Inputs, pinsBucket}
// engineReadsState is the assigner's state of what each node's engine states it reads (ADR 0618 §3), one key `<node>`, as
// the verb manifest-reads last answered it or the fallback read it: what assigner_engine_reads lists and asks again.
@@ -121,8 +121,9 @@ var keptStates = []announce.State{
{Name: copiesState, Readable: true},
{Name: writtenState, Readable: true},
{Name: buckets.WaitsState, Readable: true},
{Name: buckets.InputsState, Readable: true}, // read by the tool process alone: the verb input and assigner_inputs (design 47 G §354)
{Name: pinsState, Readable: true}, // the records pinning each assignment graph revision (design 47 G §579)
{Name: buckets.PutWaitsState, Readable: true}, // the put waits said (putwaits.go, design 47 G A-211)
{Name: buckets.InputsState, Readable: true}, // read by the tool process alone: the verb input and assigner_inputs (design 47 G §354)
{Name: pinsState, Readable: true}, // the records pinning each assignment graph revision (design 47 G §579)
}
// kvSet is every bucket the assigner opened, by name.
@@ -197,6 +198,11 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any
logf("the bucket %s cannot be opened (%v): no outside wait is kept until the assigner starts again", name, err)
continue
}
if name == buckets.PutWaits && err != nil {
// New with A-211: without it no put's wait is said until the assigner starts again.
logf("the bucket %s cannot be opened (%v): no put's wait is said until the assigner starts again", name, err)
continue
}
if name == retiredBucket && err != nil {
// The retired state is read only for the copy into `assignments` (migrate.go): a mesh made after the switch has
// none, and nothing is copied.
@@ -221,7 +227,7 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any
func startsWithout(name string) bool {
switch name {
case plc.SeatHoldersBucket, buckets.Provisions, assign.PoliciesBucket, assign.FindingsBucket,
buckets.Puts, buckets.Source, buckets.Merges, engineReadsBucket, retiredBucket, writtenBucket, buckets.Waits, buckets.Inputs, pinsBucket:
buckets.Puts, buckets.Source, buckets.Merges, engineReadsBucket, retiredBucket, writtenBucket, buckets.Waits, buckets.PutWaits, buckets.Inputs, pinsBucket:
return true
}
return slices.Contains(seatBuckets, name) || name == buckets.Grants
@@ -351,6 +357,7 @@ var watched = []string{
engineReadsBucket,
writtenBucket,
buckets.Waits,
buckets.PutWaits,
}
// follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after
@@ -63,8 +63,6 @@ type putActs struct {
rec meshhandactsclient.Recorder
warrantNo map[string]string
said map[string]string
// waiting is each put without hold whose wait-held was said, by key (putwaits.go); nil until the first pass seeds it.
waiting map[string]putWaiting
// snapshot takes the bus's snapshot on node; a variable so a test takes none.
snapshot func(ctx context.Context, node, module string) (string, error)
}
@@ -109,8 +107,7 @@ func (p *passer) puts(ctx context.Context) {
}
p.placementPuts(ctx)
all := p.putsByModule()
p.acts.seedPutWaits(all)
defer func() { p.putWaits(p.putsByModule()) }()
defer func() { p.putWaits(ctx, p.putsByModule()) }()
if len(all) == 0 {
return
}
@@ -1,12 +1,15 @@
package main
import (
"context"
"encoding/json"
"fmt"
"time"
meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
"mesh-delivery-assigner/internal/assignment"
"mesh-delivery-assigner/internal/buckets"
"mesh-delivery-assigner/internal/puts"
)
@@ -38,47 +41,42 @@ func putWaitBody(node, module string, p puts.Put) meshwants.Wait {
Words: fmt.Sprintf("the put of %s at %s on %s waits for %s to run it", module, shortFP(p.FP), node, node)}
}
// putWaiting is a put whose wait-held was said: its identity and the put as it was then.
// putWaiting is a put whose wait-held was said, as the state put-waits keeps it under the put's key (design 47 G A-204,
// A-211): its identity and the put as it was then.
type putWaiting struct {
id string
put puts.Put
ID string `json:"id"`
Put puts.Put `json:"put"`
}
// seedPutWaits takes, at the process's first pass, every put that waits as said already: it was said by the process before.
func (a *putActs) seedPutWaits(all map[string]map[string]puts.Put) {
a.mu.Lock()
defer a.mu.Unlock()
if a.waiting != nil {
// saidPutWaits is every put wait the state put-waits holds, by key; a value that does not decode is said and left.
func (p *passer) saidPutWaits() map[string]putWaiting {
out := map[string]putWaiting{}
for key, value := range p.w.read(buckets.PutWaits) {
var w putWaiting
if err := json.Unmarshal(value, &w); err != nil || w.ID == "" {
p.logf("%s %s holds no put wait, left: %v", buckets.PutWaits, key, err)
continue
}
out[key] = w
}
return out
}
// putWaits says wait-ended of each put whose said wait the state put-waits keeps and that no longer waits, whether it ended
// while this process ran or while the assigner was stopped (A-204), and wait-held of each put that waits and was not said;
// one not said, or not kept, is said at the next pass. Without the state no put's wait is said.
func (p *passer) putWaits(ctx context.Context, all map[string]map[string]puts.Put) {
if p.kvs[buckets.PutWaits] == nil {
return
}
a.waiting = map[string]putWaiting{}
for module, byNode := range all {
for node, p := range byNode {
if key, err := puts.Key(node, module); err == nil && putWaits(p) {
a.waiting[key] = putWaiting{id: identity(key, p), put: p}
}
}
}
}
// putWaits says wait-ended of each put that waited and no longer does, and wait-held of each put that waits and was not said;
// one not said is said at the next pass.
func (p *passer) putWaits(all map[string]map[string]puts.Put) {
a := p.acts
a.seedPutWaits(all)
a.mu.Lock()
waiting := make(map[string]putWaiting, len(a.waiting))
for k, v := range a.waiting {
waiting[k] = v
}
a.mu.Unlock()
waiting := p.saidPutWaits()
for key, was := range waiting {
node, module, err := puts.ParseKey(key)
if err != nil {
continue
}
cur, there := all[module][node]
same := there && identity(key, cur) == was.id
same := there && identity(key, cur) == was.ID
if same && putWaits(cur) {
continue
}
@@ -88,15 +86,14 @@ func (p *passer) putWaits(all map[string]map[string]puts.Put) {
} else if same && cur.Step == puts.Released {
how = putWaitReleased
}
body := putWaitBody(node, module, was.put)
body := putWaitBody(node, module, was.Put)
body.How = how
if err := p.wantsSay(meshwants.EventWaitEnded, body, was.id+".put."+how); err != nil {
if err := p.wantsSay(meshwants.EventWaitEnded, body, was.ID+".put."+how); err != nil {
p.logf("the end of the put %s's wait is not said, and is said at the next pass: %v", key, err)
continue
}
a.mu.Lock()
delete(a.waiting, key)
a.mu.Unlock()
p.remove(ctx, buckets.PutWaits, key)
delete(waiting, key)
}
for module, byNode := range all {
for node, put := range byNode {
@@ -105,19 +102,15 @@ func (p *passer) putWaits(all map[string]map[string]puts.Put) {
continue
}
id := identity(key, put)
a.mu.Lock()
said := a.waiting[key].id == id
a.mu.Unlock()
if said {
if w, said := waiting[key]; said && w.ID == id {
continue
}
if err := p.wantsSay(meshwants.EventWaitHeld, putWaitBody(node, module, put), id+".put"); err != nil {
p.logf("the put %s's wait is not said, and is said at the next pass: %v", key, err)
continue
}
a.mu.Lock()
a.waiting[key] = putWaiting{id: id, put: put}
a.mu.Unlock()
kept := putWaiting{ID: id, Put: put}
p.put(ctx, buckets.PutWaits, key, func() ([]byte, error) { return json.Marshal(kept) })
}
}
}
@@ -1,47 +1,67 @@
package main
import (
"context"
"testing"
"time"
meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
"mesh-delivery-assigner/internal/buckets"
"mesh-delivery-assigner/internal/puts"
)
// Design 47 G §498 and the main file's *A put without hold that stands*: a put without hold whose want is written is said as
// wait-held, step put, with its node, module, fingerprint, since and a 30-minute bound, and wait-ended when it runs, is
// released or is put again; a held put is said as no wait; a put that waits at the process's start was said before and is
// not said again; assigner_puts answers since and bound.
// released or is put again; a held put is said as no wait; assigner_puts answers since and bound. A-204, A-211: the said
// waits are kept in the state put-waits, so a put that waits at the start was said before and is not said again, and one
// that ran while the assigner was stopped gets its wait-ended at the next start.
func TestAPutWithoutHoldIsSaidAsAWait(t *testing.T) {
at := time.Date(2026, 11, 12, 8, 0, 0, 0, time.UTC)
p := newPasser(kvSet{}, at)
p.acts = newPutActs(&fakeBus{calls: make(chan string, 8)}, func() {}, func(string, ...any) {})
kept := &memKV{values: map[string][]byte{}}
var events []string
var bodies []meshwants.Wait
p.publishWants = func(event string, body any, id string) error {
events, bodies = append(events, event), append(bodies, body.(meshwants.Wait))
return nil
start := func() *passer { // a process's start: a fresh world that hands what the state holds
p := newPasser(kvSet{buckets.PutWaits: kept}, at)
for k, v := range kept.values {
p.w.put(buckets.PutWaits, k, v)
}
p.acts = newPutActs(&fakeBus{calls: make(chan string, 8)}, func() {}, func(string, ...any) {})
p.publishWants = func(event string, body any, id string) error {
events, bodies = append(events, event), append(bodies, body.(meshwants.Wait))
return nil
}
return p
}
ctx := context.Background()
asked := puts.Put{FP: fpTest, Why: "a try", At: at}.Said(at, puts.Asked, "asked")
wanted := asked.Said(at.Add(time.Minute), puts.Wanted, "the want is written").Said(at.Add(2*time.Minute), puts.Wanted, "its check failed")
held := wanted
held.Hold = true
// At the start a put that waits was said before: nothing is said of it.
p.putWaits(map[string]map[string]puts.Put{"nats": {"ace": wanted}})
p := start()
p.putWaits(ctx, map[string]map[string]puts.Put{"nats": {"ace": wanted}})
if len(events) != 1 || events[0] != meshwants.EventWaitHeld || kept.values["ace.nats"] == nil {
t.Fatalf("a put that waits said %v, kept %q", events, kept.values)
}
events, bodies = nil, nil
// At the next start the put still waits: it was said before, nothing is said of it.
p = start()
p.putWaits(ctx, map[string]map[string]puts.Put{"nats": {"ace": wanted}})
if len(events) != 0 {
t.Fatalf("a put waiting at the start was said again: %v", events)
}
// It runs: wait-ended, ran.
p.putWaits(map[string]map[string]puts.Put{"nats": {"ace": wanted.Said(at.Add(3*time.Minute), puts.Done, "runs")}})
if len(events) != 1 || events[0] != meshwants.EventWaitEnded || bodies[0].How != putWaitRan || bodies[0].Step != meshwants.WaitStepPut {
t.Fatalf("a put that ran said %v %+v", events, bodies)
// It ran while the assigner was stopped: at the next start, wait-ended, ran, and the state forgets it.
p = start()
p.putWaits(ctx, map[string]map[string]puts.Put{"nats": {"ace": wanted.Said(at.Add(3*time.Minute), puts.Done, "runs")}})
if len(events) != 1 || events[0] != meshwants.EventWaitEnded || bodies[0].How != putWaitRan || bodies[0].Step != meshwants.WaitStepPut ||
kept.values["ace.nats"] != nil {
t.Fatalf("a put that ran while stopped said %v %+v, kept %q", events, bodies, kept.values)
}
events, bodies = nil, nil
// A new put's want is written: wait-held, since the want was written, bound 30m; a held put says nothing.
again := wanted
again.At = at.Add(time.Hour)
p.putWaits(map[string]map[string]puts.Put{"nats": {"ace": again}, "web": {"ace": held}})
p.putWaits(ctx, map[string]map[string]puts.Put{"nats": {"ace": again}, "web": {"ace": held}})
if len(events) != 1 || events[0] != meshwants.EventWaitHeld {
t.Fatalf("a put's written want said %v", events)
}
@@ -51,14 +71,14 @@ func TestAPutWithoutHoldIsSaidAsAWait(t *testing.T) {
t.Errorf("the put's wait is %+v", b)
}
// Said once.
p.putWaits(map[string]map[string]puts.Put{"nats": {"ace": again}, "web": {"ace": held}})
p.putWaits(ctx, map[string]map[string]puts.Put{"nats": {"ace": again}, "web": {"ace": held}})
if len(events) != 1 {
t.Errorf("a put's wait was said again: %v", events)
}
// Put again under its key: the first wait ends superseded, the new one is held.
newer := asked
newer.At = at.Add(2 * time.Hour)
p.putWaits(map[string]map[string]puts.Put{"nats": {"ace": newer}})
p.putWaits(ctx, map[string]map[string]puts.Put{"nats": {"ace": newer}})
if len(events) != 2 || events[1] != meshwants.EventWaitEnded || bodies[1].How != putWaitSuperseded || bodies[1].Fingerprint != fpTest {
t.Errorf("a put asked again said %v %+v", events, bodies)
}
@@ -25,6 +25,10 @@ const (
// Waits holds the waits on an outside cause, one key `<node>.<module>.<fingerprint hex>` (design 47 G §706).
WaitsState = "waits"
Waits = plc.Module + "_" + WaitsState
// PutWaits holds each put wait the assigner said, one key `<node>.<module>`, its value the wait's id and the put as said
// (design 47 G A-204, A-211), so a put that ends while the assigner is stopped gets its wait-ended at the next start.
PutWaitsState = "put-waits"
PutWaits = plc.Module + "_" + PutWaitsState
// Inputs holds the inputs of the composition steps, `<time>.<digest>`, which the seat mesh-wants' verb input answers (design 47 G §354).
InputsState = "inputs"
Inputs = plc.Module + "_" + InputsState
@@ -67,7 +67,7 @@ func TestManifest(t *testing.T) {
}
if want := map[string]int{"asks": 5, "differences": 64, "puts": 20, "wants": 20,
"assignments": 64, "placement": 64, "copies": 1, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5, views.MarksState: 1, views.StatusState: 1,
"engine-reads": 1, "written": 5, "waits": 5, "inputs": 1, "pins": 5}; !reflect.DeepEqual(states, want) {
"engine-reads": 1, "written": 5, "waits": 5, "put-waits": 1, "inputs": 1, "pins": 5}; !reflect.DeepEqual(states, want) {
t.Errorf("state = %v, want %v", states, want)
}
if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 {
@@ -143,6 +143,7 @@
{ "name": "engine-reads", "history": 1 },
{ "name": "written", "history": 5 },
{ "name": "waits", "history": 5 },
{ "name": "put-waits", "history": 1 },
{ "name": "inputs", "history": 1 },
{ "name": "pins", "history": 5, "max-mib": 4 }
],