diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/delivered.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/delivered.go index 97ccbbf1..8e861022 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/delivered.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/delivered.go @@ -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) }) + } + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/delivered_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/delivered_test.go index 1a60b297..75364e8f 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/delivered_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/delivered_test.go @@ -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) + } } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index c3163794..6d410c29 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -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 ``, 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 diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_test.go index 8841044f..8438cdb3 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_test.go @@ -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) + } } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_tools.go index bd913301..e915e82a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_tools.go @@ -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 diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index 313737b9..99e5985d 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -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) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go index 849d3b66..1911af8f 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go @@ -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), diff --git a/modules/mesh-delivery-assigner/internal/buckets/buckets.go b/modules/mesh-delivery-assigner/internal/buckets/buckets.go index 4bccf1e1..2b2674c6 100644 --- a/modules/mesh-delivery-assigner/internal/buckets/buckets.go +++ b/modules/mesh-delivery-assigner/internal/buckets/buckets.go @@ -23,6 +23,10 @@ const ( // Waits holds the waits on an outside cause, one key `..` (design 47 G §706). WaitsState = "waits" Waits = plc.Module + "_" + WaitsState + // DeliveredAgain holds each ask of assigner_delivered_again, one key ``, 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 ``, its value the wait-held said // (design 47 G §376, §498, A-246). DeliveryWaitsState = "delivery-waits" diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 07ae1aea..9beaf86f 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -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 { diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index 4eca5a31..e305a49d 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -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 } ],