AS6: assigner_delivered_again asks a version said delivered again, heard at once in delivered-again

The tool records its hand act, writes the ask {module, version, by, why} into the state
delivered-again and answers that it is asked; the process's watch of that state wakes a pass at once,
which says version-delivered again when every present machine placing the module runs that version
healthy and takes the ask away, else keeps it with why, which assigner_waits lists.

Builds: design 47 G §376, A-250

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-12 09:29:57 +01:00
parent fea9de98eb
commit 86af4c29c4
10 changed files with 185 additions and 13 deletions
@@ -378,3 +378,74 @@ func (p *passer) endDeliveryWait(ctx context.Context, key string, w deliveryWait
}
p.remove(ctx, buckets.DeliveryWaits, key)
}
// againAsk is one ask of assigner_delivered_again, kept in the state delivered-again under its module (A-250): the version to say
// delivered again, who asked and why, and, while it does not hold, why not.
type againAsk struct {
Module string `json:"module"`
Version string `json:"version"`
By string `json:"by"`
Why string `json:"why"`
Waits string `json:"waits,omitempty"`
}
// releaseOf is the release heard of module's version.
func (r *releasedVersions) releaseOf(module, version string) (releaseHeard, bool) {
r.mu.Lock()
defer r.mu.Unlock()
if h, ok := r.newest[module]; ok && h.Version == version {
return h, true
}
for key, k := range r.seen {
if strings.HasPrefix(key, module+"@") && k.Version == version {
return releaseHeard{Version: version, Parts: k.Parts}, true
}
}
return releaseHeard{}, false
}
// deliveredAgain takes each ask of assigner_delivered_again, heard at once through the watch of the state delivered-again: it says
// version-delivered again when every present machine placing the module runs that version healthy and removes the ask, else keeps
// it with why it does not hold, which assigner_waits reads.
func (p *passer) deliveredAgain(ctx context.Context) {
if p.kvs[buckets.DeliveredAgain] == nil {
return
}
for key, value := range p.w.read(buckets.DeliveredAgain) {
var a againAsk
if err := json.Unmarshal(value, &a); err != nil || a.Version == "" {
p.logf("%s %s holds no ask, left: %v", buckets.DeliveredAgain, key, err)
continue
}
module := a.Module
if module == "" {
module = key
}
waits := ""
rel, ok := p.w.released.releaseOf(module, a.Version)
notRunning, away, placing := p.deliveryStanding(module, a.Version)
switch {
case !ok:
waits = fmt.Sprintf("no version-released of %s %s is heard", module, a.Version)
case placing == 0:
waits = fmt.Sprintf("no machine places %s", module)
case len(away) == placing:
waits = fmt.Sprintf("every machine placing %s is away: %s", module, strings.Join(away, ", "))
case len(notRunning) > 0:
waits = fmt.Sprintf("%s %s does not run healthy on %s", module, a.Version, strings.Join(notRunning, ", "))
}
if waits == "" {
if err := p.sayDelivered(module, rel, away); err != nil {
p.logf("version-delivered of %s %s asked again is not said, and is asked at the next pass: %v", module, a.Version, err)
continue
}
p.logf("version-delivered of %s %s said again, asked by %s: %s", module, a.Version, a.By, a.Why)
p.remove(ctx, buckets.DeliveredAgain, key)
continue
}
if a.Waits != waits {
a.Waits = waits
p.put(ctx, buckets.DeliveredAgain, key, func() ([]byte, error) { return json.Marshal(a) })
}
}
}
@@ -25,7 +25,8 @@ func TestADependentWaitsForItsDependencysDelivery(t *testing.T) {
ctx := context.Background()
at := time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC)
waits := &memKV{values: map[string][]byte{}}
p := newPasser(kvSet{buckets.Wants: &memKV{values: map[string][]byte{}}, buckets.DeliveryWaits: waits}, at)
again := &memKV{values: map[string][]byte{}}
p := newPasser(kvSet{buckets.Wants: &memKV{values: map[string][]byte{}}, buckets.DeliveryWaits: waits, buckets.DeliveredAgain: again}, at)
p.logf = t.Logf
var events []said
p.publishWants = func(event string, body any, id string) error {
@@ -51,8 +52,10 @@ func TestADependentWaitsForItsDependencysDelivery(t *testing.T) {
t.Fatal(err)
}
}
source := int64(3)
release := func(module, version string) {
raw, _ := json.Marshal(meshcatalog.VersionReleasedEvent{V: 1, Module: module, Version: version, Source: 4, At: &at,
source++
raw, _ := json.Marshal(meshcatalog.VersionReleasedEvent{V: 1, Module: module, Version: version, Source: source, At: &at,
Artifacts: []meshcatalog.VersionArtifact{{Artifact: "program", Digest: "sha256:d"}}})
if err := p.w.released.put(raw); err != nil {
t.Fatal(err)
@@ -163,4 +166,23 @@ func TestADependentWaitsForItsDependencysDelivery(t *testing.T) {
if p.w.awayNow()["ace"] || !p.w.online()["ace"] {
t.Error("an older machine-away read after a newer machine-back leaves ace away")
}
// A-250: an ask of assigner_delivered_again of 1.1.0, which ace (present) runs and g14 (away) does not, is said again and taken
// away; one of 1.2.0, which no machine runs, is kept with why.
ask := func(version string) {
raw, _ := json.Marshal(againAsk{Module: "db", Version: version, By: "op", Why: "the catalog missed it"})
p.w.put(buckets.DeliveredAgain, "db", raw)
}
before := len(saidOf(meshwants.EventVersionDelivered))
ask("1.1.0")
p.deliveredAgain(ctx)
if len(saidOf(meshwants.EventVersionDelivered)) != before+1 || p.w.has(buckets.DeliveredAgain, "db") {
t.Errorf("the ask of 1.1.0 said %d version-delivered and is kept %v", len(saidOf(meshwants.EventVersionDelivered))-before, p.w.has(buckets.DeliveredAgain, "db"))
}
ask("1.2.0")
p.deliveredAgain(ctx)
var a againAsk
if json.Unmarshal(p.w.value(buckets.DeliveredAgain, "db"), &a) != nil || a.Waits != "db 1.2.0 does not run healthy on ace" ||
len(saidOf(meshwants.EventVersionDelivered)) != before+1 {
t.Errorf("the ask of 1.2.0 = %+v", a)
}
}
@@ -76,7 +76,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.PutWaits, buckets.DeliveryWaits, buckets.Inputs, pinsBucket}
engineReadsBucket, retiredBucket, copiesBucket, writtenBucket, buckets.Waits, buckets.PutWaits, buckets.DeliveryWaits, buckets.DeliveredAgain, 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.
@@ -117,10 +117,11 @@ var keptStates = []announce.State{
{Name: copiesState, Readable: true},
{Name: writtenState, Readable: true},
{Name: buckets.WaitsState, Readable: true},
{Name: buckets.PutWaitsState, Readable: true}, // the put waits said (putwaits.go, design 47 G A-211)
{Name: buckets.DeliveryWaitsState, Readable: true}, // the delivery waits said (delivered.go, design 47 G A-246)
{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.DeliveryWaitsState, Readable: true}, // the delivery waits said (delivered.go, design 47 G A-246)
{Name: buckets.DeliveredAgainState, Readable: true}, // the asks of assigner_delivered_again (delivered.go, design 47 G A-250)
{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.
@@ -189,6 +190,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.DeliveredAgain && err != nil {
// New with A-250: without it assigner_delivered_again is not taken until the assigner starts again.
logf("the bucket %s cannot be opened (%v): no version is said delivered again until the assigner starts again", name, err)
continue
}
if name == buckets.DeliveryWaits && err != nil {
// New with A-246: without it no dependent waits for its dependency's delivery until the assigner starts again.
logf("the bucket %s cannot be opened (%v): no delivery wait is kept until the assigner starts again", name, err)
@@ -223,7 +229,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, engineReadsBucket, retiredBucket, writtenBucket, buckets.Waits, buckets.PutWaits, buckets.DeliveryWaits, buckets.Inputs, pinsBucket:
buckets.Puts, engineReadsBucket, retiredBucket, writtenBucket, buckets.Waits, buckets.PutWaits, buckets.DeliveryWaits, buckets.DeliveredAgain, buckets.Inputs, pinsBucket:
return true
}
return slices.Contains(seatBuckets, name) || name == buckets.Grants
@@ -353,6 +359,7 @@ var watched = []string{
buckets.Waits,
buckets.PutWaits,
buckets.DeliveryWaits,
buckets.DeliveredAgain, // its watch wakes a pass at once on an ask of assigner_delivered_again (A-250)
}
// follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after
@@ -259,7 +259,7 @@ func TestTheToolsReadAndReleaseAnOutsideWait(t *testing.T) {
r.cond("machine.hub.unreachable", "ace")
r.review(false, nil)
var acts []string
s := newPutTools(memStore{waitsState: &memKV{values: map[string][]byte{}}, buckets.DeliveryWaitsState: &memKV{values: map[string][]byte{}}}, func(key string, body any) (json.RawMessage, error) {
s := newPutTools(memStore{waitsState: &memKV{values: map[string][]byte{}}, buckets.DeliveryWaitsState: &memKV{values: map[string][]byte{}}, buckets.DeliveredAgainState: &memKV{values: map[string][]byte{}}}, func(key string, body any) (json.RawMessage, error) {
acts = append(acts, key)
return json.RawMessage(`{"ok":true,"log":"logged"}`), nil
})
@@ -318,4 +318,16 @@ func TestTheToolsReadAndReleaseAnOutsideWait(t *testing.T) {
if json.Unmarshal(raw, &released) != nil || !released.Released || !strings.Contains(released.By, "db is fine") || len(acts) != before+1 {
t.Errorf("the delivery wait after its release = %+v, hand acts %v", released, acts[before:])
}
// A-250: assigner_delivered_again records its hand act, writes the ask and answers asked; assigner_waits lists it.
if _, err := s.doDeliveredAgain(map[string]any{"module": "db", "version": "1.1.0"}, "op"); err == nil {
t.Error("an ask with no why is taken")
}
ans, err := s.doDeliveredAgain(map[string]any{"module": "db", "version": "1.1.0", "why": "the catalog missed it"}, "op")
if err != nil || ans.(map[string]any)["asked"] != "db" || len(acts) != before+2 {
t.Fatalf("assigner_delivered_again = %+v %v", ans, err)
}
got, _ = s.doWaits(map[string]any{})
if a := got.(map[string]any)["delivered-again"].([]againAsk); len(a) != 1 || a[0].Version != "1.1.0" || a[0].By != "op" {
t.Errorf("the asks listed = %+v", a)
}
}
@@ -80,7 +80,59 @@ func (s *putTools) doWaits(args map[string]any) (any, error) {
}
delivery = append(delivery, w)
}
return map[string]any{"waits": out, "delivery": delivery}, nil
// And each ask of assigner_delivered_again not yet holding, with why (A-250).
again := []againAsk{}
akeys, err := s.st.keys(buckets.DeliveredAgainState)
if err != nil {
return nil, fmt.Errorf("the asks to say a version delivered again cannot be listed: %w", err)
}
sort.Strings(akeys)
for _, key := range akeys {
raw, _, found, err := s.st.get(buckets.DeliveredAgainState, key)
var a againAsk
if err != nil || !found || json.Unmarshal(raw, &a) != nil || (arg(args, "module") != "" && a.Module != arg(args, "module")) {
continue
}
again = append(again, a)
}
return map[string]any{"waits": out, "delivery": delivery, "delivered-again": again}, nil
}
// toolDeliveredAgain is the tool that asks version-delivered said again for one version (design 47 G §376, A-250).
const toolDeliveredAgain = "assigner_delivered_again"
// doDeliveredAgain is assigner_delivered_again: the hand act is recorded first, then the ask is written into the state
// delivered-again, whose watch the assigner's process hears at once; it answers that it is asked.
func (s *putTools) doDeliveredAgain(args map[string]any, by string) (any, error) {
module, version, why := arg(args, "module"), arg(args, "version"), arg(args, "why")
if module == "" || version == "" || why == "" {
return nil, errors.New("assigner_delivered_again names the module, its version and why: it is kept in the hand-act log")
}
rec := meshhandactsclient.Recorder{Ask: s.ask}
taken, err := rec.Record(meshhandactsclient.Request{What: fmt.Sprintf("assigner_delivered_again %s %s", module, version), Why: why,
Cause: "version-delivered", Verb: "mesh-wants.assigner_delivered_again"})
if err != nil {
return nil, fmt.Errorf("the hand act could not be recorded, so nothing was asked: %w", err)
}
data, err := json.Marshal(againAsk{Module: module, Version: version, By: by, Why: why})
if err != nil {
return nil, err
}
if err := s.st.put(buckets.DeliveredAgainState, keys.EscapeName(module), data); err != nil {
return nil, fmt.Errorf("the ask of %s %s is not written: %w", module, version, err)
}
return map[string]any{"asked": module, "version": version, "hand-act": taken.Log,
"follow": "the assigner says version-delivered again once every present machine placing it runs it healthy; assigner_waits says why it does not yet"}, nil
}
// deliveredAgainTool is assigner_delivered_again as the runtime serves it.
func (s *putTools) deliveredAgainTool() stdio.Tool {
return stdio.Tool{Name: toolDeliveredAgain, Description: "Say version-delivered again for one version of a module, for the operator and the healers " +
"(design 47 G §376): written in the hand-act log first, then asked of the assigner, which says it once every present machine placing " +
"the module runs that version healthy; until then assigner_waits lists the ask with why it does not hold. Answers that it is asked.",
Input: schema(map[string]string{"module": "the module", "version": "its MAJOR.MINOR.PATCH", "why": "why, kept in the hand-act log"},
[]string{"module", "version", "why"}),
Run: func(a map[string]any) (any, error) { return s.doDeliveredAgain(a, stdio.CallerOf(a)) }}
}
// doReleaseDelivery is assigner_release with step delivery: the hand act is recorded first, then the release is written on the
@@ -206,6 +206,7 @@ func (p *passer) pass(ctx context.Context) {
p.assignmentsSeat(ctx)
p.keepGraph(ctx)
p.judgeDelivered()
p.deliveredAgain(ctx)
p.wants(ctx)
p.puts(ctx)
p.nodeWants(ctx)
@@ -353,6 +353,7 @@ func (s *putTools) tools() []stdio.Tool {
[]string{"module", "why"}),
Run: func(a map[string]any) (any, error) { return s.doRelease(a, stdio.CallerOf(a)) }},
s.waitsTool(),
s.deliveredAgainTool(),
{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),
@@ -23,6 +23,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
// DeliveredAgain holds each ask of assigner_delivered_again, one key `<module>`, its value the version, who asked and why, and
// why it does not hold yet (design 47 G §376, A-250).
DeliveredAgainState = "delivered-again"
DeliveredAgain = plc.Module + "_" + DeliveredAgainState
// DeliveryWaits holds each dependent's delivery wait the assigner said, one key `<module>`, its value the wait-held said
// (design 47 G §376, §498, A-246).
DeliveryWaitsState = "delivery-waits"
@@ -70,7 +70,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, "put-waits": 1, "delivery-waits": 1, "inputs": 1, "pins": 5}; !reflect.DeepEqual(states, want) {
"engine-reads": 1, "written": 5, "waits": 5, "put-waits": 1, "delivery-waits": 1, "delivered-again": 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 {
@@ -105,7 +105,7 @@ func TestManifest(t *testing.T) {
// which lists and asks again what each node's engine reads.
if !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts", "assigner_waits", "assigner_inputs", "assignments", "seat-holders", "wants",
"assigner_read_assignments", "assigner_read_seat_holders", "assigner_read_wants", "assigner_read_findings", "assigner_read_asks", "assigner_read_differences", "assigner_read_copies",
"assigner_first_views", "assigner_first_view_again", "assigner_engine_reads", "assigner_seat_list", "assigner_seat_list_again", "assigner_ask_again", "assigner_written", "assigner_pins", "assigner_unpin"}) {
"assigner_first_views", "assigner_first_view_again", "assigner_engine_reads", "assigner_seat_list", "assigner_seat_list_again", "assigner_ask_again", "assigner_written", "assigner_pins", "assigner_unpin", "assigner_delivered_again"}) {
t.Errorf("the assigner serves tools %v, want its three put tools, three first views, six read tools, assigner_engine_reads, the seat list's two (ADR 0636 §7) and assigner_ask_again (ADR 0668 §5)", m.Tools)
}
if m.RunsAs != m.Module || m.SecretsOwner != m.RunsAs {
+3 -1
View File
@@ -124,7 +124,8 @@
"assigner_ask_again",
"assigner_written",
"assigner_pins",
"assigner_unpin"
"assigner_unpin",
"assigner_delivered_again"
],
"state": [
{ "name": "assignments", "history": 64 },
@@ -147,6 +148,7 @@
{ "name": "waits", "history": 5 },
{ "name": "put-waits", "history": 1 },
{ "name": "delivery-waits", "history": 1 },
{ "name": "delivered-again", "history": 1 },
{ "name": "inputs", "history": 1 },
{ "name": "pins", "history": 5, "max-mib": 4 }
],