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 a2a9ccb1..d671bb71 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -29,6 +29,7 @@ import ( mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" occ "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" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-sdk/go/announce" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/seatlist" @@ -122,7 +123,7 @@ var keptStates = []announce.State{ {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: pinsState, Readable: true}, // the records pinning each assignment graph revision (design 47 G §579) } // kvSet is every bucket the assigner opened, by name. @@ -555,6 +556,9 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, if iv != nil && iv.h != nil { objects := newBusObjects(js, process.Account{Node: thisNode, Module: assignment.Module}, assigngraph.RevisionsObjects) p.graph = &graphKeeper{objects: objects, halves: iv.h.views} + // A held revert's running entry and its want's, asked of the seat mesh-catalog's verb entry (design 47 G §710, A-217). + p.entries = meshcatalogclient.NewEntries(iv.h.call) + p.entries.Changed = func(meshcatalogclient.EntryRow) { wake() } // The pins of the revisions (pins.go, design 47 G §579): the gate's pinned, freed, heads-replaced and pins-stated // on the hub, registered before it starts. if kv := kvs[pinsBucket]; kv != nil { 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 aad8b9c1..9eb26f44 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -14,6 +14,7 @@ import ( "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" mwc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate" @@ -103,6 +104,9 @@ type passer struct { publishWants publisher // graph keeps the assignment graph revision the wants follow (graphkept.go, design 47 G §292, A-67); nil: none is kept. graph *graphKeeper + // entries are the catalog entries by fingerprint asked of the seat mesh-catalog's verb `entry` (design 47 G §614, A-217): a + // held revert's running entry and its want's (revert.go); nil: no revert is asked about. + entries *meshcatalogclient.Entries } // wantsSay says one event of the seat mesh-wants; with no bus to say it on, nothing is said. @@ -210,7 +214,7 @@ func (p *passer) pass(ctx context.Context) { } p.puts(ctx) p.nodeWants(ctx) - p.backups(ctx) + p.backups(ctx, p.reverts(ctx)) } // nodeWants writes each placed node's own want, `want..node`: the node composes its own (ADR 0547 §2, ADR 0571 §6). @@ -743,7 +747,7 @@ func (p *passer) lift(ctx context.Context, module, fp string) { // staleAsk says why the hold, restore or record ask at key (read as the SDK reads it) is about something the trunk has left // behind, or "" when it is not: a module no longer on the trunk, a fingerprint that is no longer the module's trunk fingerprint, -// or, for a record ask, a composition the module's node no longer waits on. A backup ask is stale when the module's want on its +// or, for a record ask, a composition the module's node no longer waits on. A backup ask, and a revert ask (design 47 G §710), is stale when the module's want on its // node names another fingerprint or is removed (the move it is about was superseded; issue 1966 note 165); a want not written // yet, or at the same fingerprint in another composition, leaves it (backups asks that again under the next number). An ask of // another kind is never stale here. A module of the trunk whose fingerprint is not read yet, and a have not read yet, are not stale: nothing known is @@ -754,7 +758,7 @@ func (p *passer) staleAsk(key string, trunk map[string]meshcatalog.ModulesByModu return "" } switch k.Kind { - case asks.AskKindBackup: + case asks.AskKindBackup, asks.AskKindRevert: wkey, err := mwc.WantKey(k.Node, k.Module) if err != nil { return "" @@ -875,6 +879,10 @@ func (p *passer) staleAsks(ctx context.Context, trunk map[string]meshcatalog.Mod } delete(p.said, assign.BackupCondition(k.Node, k.Module, k.FP, "", false).Key) } + // A revert ask's finding goes with it: a newer want replaced the held revert (design 47 G §710, §363). + if kerr == nil && k.Kind == asks.AskKindRevert { + p.revertFinding(assign.RevertHeldFinding(k.Node, k.Module, "", "", ""), false) + } p.logf("the ask %s is about a fingerprint or composition the trunk has left behind and is taken away: %s", key, why) p.remove(ctx, buckets.Asks, key) // The held record stays: it is the memory that this fingerprint failed (ADR 0408 §4), and a revert returns the trunk to @@ -994,7 +1002,9 @@ func (p *passer) recordAsk(ctx context.Context, module, fp string, need assign.R // before the `@`). Each answer is another want value (the asking's number is in the want's words), so a backup that fails again // after "back up again" and is answered "again" again changes the key and the node retries. That the backup failed again is seen // in a new have value after the answer was written. -func (p *passer) backups(ctx context.Context) { +// +// A have that holds a revert (reverts, which answers its `.` in reverting) is left to the revert ask. +func (p *passer) backups(ctx context.Context, reverting map[string]bool) { wanted := p.w.wantsOf() for key, value := range p.w.read(buckets.Have) { node, module, isNode, err := nodeengineclient.ParseHaveKey(key) @@ -1002,7 +1012,7 @@ func (p *passer) backups(ctx context.Context) { continue } have, err := nodeengineclient.DecodeHave(value) - if err != nil || have.Phase != nodeengineclient.PhaseHeld { + if err != nil || have.Phase != nodeengineclient.PhaseHeld || reverting[node+"."+module] { continue } want, ok := wanted[module][node] diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/revert.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/revert.go new file mode 100644 index 00000000..80640eb7 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/revert.go @@ -0,0 +1,160 @@ +package main + +import ( + "context" + "strconv" + "strings" + + "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks" + mcc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" + mwc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient" + "git.novox.be/novox/mesh-sdk/go/views" + + "mesh-delivery-assigner/internal/assign" + "mesh-delivery-assigner/internal/assigngraph" + "mesh-delivery-assigner/internal/buckets" +) + +// revertReasons are the reasons a node's have gives when it holds a revert (design 47 G §300, §567). +var revertReasons = []string{"restore point missing", "restore point damaged"} + +// revertReason is the reason of a have's words when they say a held revert, or "". +func revertReason(words string) string { + for _, r := range revertReasons { + if strings.HasPrefix(words, r) { + return r + } + } + return "" +} + +// reverts sees to each revert a node holds with no restore point (design 47 G §300, §710): the have's phase held with the +// reason `restore point missing` or `restore point damaged`, for a want the assignment graph tells a revert (assigngraph.Reverted, +// A-217) across a kept data's form (assigngraph.FormChanges, the running entry and the want's read through the catalog seat's verb +// `entry`). It asks the operator hold or live, raises the finding revert-held, and on live writes the want again with `revert` +// live; hold writes nothing. The finding is cleared once the reverted build runs. It answers the `.` of each have +// that holds a revert, which backups leaves to it. +func (p *passer) reverts(ctx context.Context) map[string]bool { + taken := map[string]bool{} + wanted := p.w.wantsOf() + for key, value := range p.w.read(buckets.Have) { + node, module, isNode, err := nodeengineclient.ParseHaveKey(key) + if err != nil || isNode { + continue + } + have, err := nodeengineclient.DecodeHave(value) + if err != nil { + continue + } + want, ok := wanted[module][node] + if !ok || want.Removed { + continue + } + akey, err := asks.RevertAskKey(node, module, want.Fp) + if err != nil { + continue + } + reason := revertReason(have.Detail) + if have.Phase != nodeengineclient.PhaseHeld || reason == "" { + if have.Fp == want.Fp && p.w.has(buckets.Asks, akey) { + p.revertFinding(assign.RevertHeldFinding(node, module, "", "", ""), false) // the reverted build runs + } + continue + } + taken[node+"."+module] = true + kept, form, ok := p.revertCrosses(ctx, node, module, want.Fp) + if !ok { + continue + } + current, currentAsk, there := assign.DecodeAsk(akey, p.w.read(buckets.Asks)[akey]) + if there != nil { + now := p.clock() + a := assign.RevertAsk(akey, node, module, want.Fp, kept, form, have.Detail, 0, now) + p.put(ctx, buckets.Asks, akey, func() ([]byte, error) { return assign.EncodeAsk(akey, a) }) + p.revertFinding(assign.RevertHeldFinding(node, module, kept, form, reason), true) + continue + } + p.revertFinding(assign.RevertHeldFinding(node, module, kept, form, reason), true) + answer, answered := p.w.answer(akey) + if router, _, _ := strings.Cut(answer.Warrant, "@"); !answered || router != currentAsk.ID || answer.Option != assign.OptionRevertLive { + continue // not answered, or answered hold: the hold stands and nothing is written + } + words := assign.RevertAnswerWords(answer.Option, current.Asking) + if want.Revert != mwc.RevertLive || want.Words != words { + want.Revert, want.Words = mwc.RevertLive, words + p.putWant(ctx, node, module, want) + } + } + return taken +} + +// revertCrosses reads whether the want of module on node at wantFp is a revert across a kept data's form: the kept data's names +// and the want's forms of them, said as the ask and the finding name them. ok is false while it is not known (the entries not +// answered yet, a pass with the assignment graph not kept) or when the want is no revert across a form. +func (p *passer) revertCrosses(ctx context.Context, node, module, wantFp string) (kept, form string, ok bool) { + if p.graph == nil || p.entries == nil { + return "", "", false + } + run, heard := p.w.runningOf(node, module) + if !heard || run.Stopped || run.Fingerprint == "" { + return "", "", false + } + reverted, err := assigngraph.Reverted(ctx, p.graph.objects, node, module, run.Fingerprint, wantFp) + if err != nil { + p.logf("whether the want of %s on %s at %s is a revert is not read: %v", module, node, wantFp, err) + return "", "", false + } + if !reverted { + return "", "", false + } + running, rs := p.entries.Need(ctx, run.Fingerprint) + applied, as := p.entries.Need(ctx, wantFp) + if rs != mcc.EntryAnswered || as != mcc.EntryAnswered { + return "", "", false // asked: its answer wakes a pass + } + changes, err := assigngraph.FormChanges(running, applied) + if err != nil { + p.logf("the forms of %s's kept data between %s and %s are not read: %v", module, run.Fingerprint, wantFp, err) + return "", "", false + } + var names, forms []string + for _, c := range changes { + f := 0 + if c.Applied != nil { + f = c.Applied.Form + } + names, forms = append(names, c.Name), append(forms, strconv.Itoa(f)) + } + if len(names) == 0 { + return "", "", false + } + return strings.Join(names, ", "), strings.Join(forms, ", "), true +} + +// revertFinding writes the finding revert-held, or clears it, once per process each way. +func (p *passer) revertFinding(o assign.Observed, raised bool) { + key := views.FindingKey(o.Scope, assign.KeyID(o.ID), o.Kind) + said := key + ".raised" + if !raised { + said = key + ".cleared" + } + if p.said[said] { + return + } + var err error + switch { + case raised && p.findings.Put != nil: + err = p.findings.Put(o.Finding(p.clock())) + case !raised && p.findings.Delete != nil: + err = p.findings.Delete(key) + default: + return + } + if err != nil { + p.logf("the finding %s is not written: %v", key, err) + return + } + p.said[said] = true + delete(p.said, map[bool]string{true: key + ".cleared", false: key + ".raised"}[raised]) +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/revert_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/revert_test.go new file mode 100644 index 00000000..87660323 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/revert_test.go @@ -0,0 +1,136 @@ +package main + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" + "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" + mdc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient" + mwc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + + "mesh-delivery-assigner/internal/assign" + "mesh-delivery-assigner/internal/assigngraph" + "mesh-delivery-assigner/internal/buckets" +) + +// Design 47 G §300, §710, A-217: a node holding a revert with no restore point gets the revert ask, keyed by the reverted build, +// and the finding revert-held, never a backup ask; hold writes no want; live writes the want with `revert` live and its words; +// the finding clears once the reverted build runs; a newer want takes the ask away and clears the finding. +func TestAHeldRevertIsAskedAndALiveAnswerIsWrittenIntoTheWant(t *testing.T) { + ctx := context.Background() + at := time.Date(2026, 11, 12, 10, 0, 0, 0, time.UTC) + fpOld := meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64) + asksKV, wantsKV := &memKV{values: map[string][]byte{}}, &memKV{values: map[string][]byte{}} + p := newPasser(kvSet{buckets.Asks: asksKV, buckets.Wants: wantsKV}, at) + p.logf = t.Logf + findings := map[string][]byte{} + p.findings = assign.Findings{Put: putInto(findings), Delete: func(k string) error { delete(findings, k); return nil }} + + // Two assignment graph revisions: web on ace at fpOld, then at fpTest, which it runs. + objects := memObjects{} + for i, fp := range []string{fpOld, fpTest} { + half := meshcatalogclient.Half{Revision: meshcatalog.CatalogRevision{Branch: trunkBranch, Kind: "trunk", Revision: int64(i + 1), Source: int64(i + 1), + Entries: []meshcatalog.EntryRef{{Module: "web", Digest: "sha256:" + strings.Repeat(string(rune('c'+i)), 64)}}}, + Entries: map[string]meshcatalog.CatalogEntry{"web": {Module: "web", Fp: fp}}} + in := assigngraph.Inputs{Halves: map[string]meshcatalogclient.Half{trunkBranch: half}, FirstBranch: func(string) string { return trunkBranch }, + Placements: map[string]meshmoduleassigner.Assignments{"ace": {Node: "ace", Modules: []meshmoduleassigner.PlacedModule{{Module: "web"}}}}} + if _, _, err := assigngraph.Keep(ctx, objects, assigngraph.Compute(in), "test", at); err != nil { + t.Fatal(err) + } + } + p.graph = &graphKeeper{objects: objects} + // The entries by fingerprint, as the catalog seat's verb entry answers them: the kept data sessions at form 1, then form 2. + entry := func(fp string, form int) meshcatalog.CatalogEntry { + return meshcatalog.CatalogEntry{Module: "web", Fp: fp, Manifest: map[string]any{"module": "web", + "state": []any{map[string]any{"name": "sessions", "form": form}}}} + } + answered := make(chan struct{}, 4) + p.entries = meshcatalogclient.NewEntries(func(_ string, body any) (json.RawMessage, error) { + in := body.(meshcatalog.EntryInput) + return json.Marshal(meshcatalog.EntryAnswer{Entry: map[string]meshcatalog.CatalogEntry{fpOld: entry(fpOld, 1), fpTest: entry(fpTest, 2)}[in.Fingerprint]}) + }) + p.entries.Changed = func(meshcatalogclient.EntryRow) { answered <- struct{}{} } + + p.w.running["ace.web"] = nodeengine.Running{Node: "ace", Module: "web", Fingerprint: fpTest} + wkey, _ := mwc.WantKey("ace", "web") + want := mwc.Want{Fp: fpOld, Self: true} + raw, _ := mwc.EncodeWant(want) + p.w.put(buckets.Wants, wkey, raw) + hkey, _ := nodeengineclient.HaveKey("ace", "web") + have := func(fp, phase, detail string) { + raw, err := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Phase: phase, Detail: detail, Health: nodeengineclient.HealthOK}) + if err != nil { + t.Fatal(err) + } + p.w.put(buckets.Have, hkey, raw) + } + have(fpTest, nodeengineclient.PhaseHeld, "restore point missing: web's sessions has none of form 1 taken before "+fpOld) + pass := func() { p.backups(ctx, p.reverts(ctx)) } + + pass() // the entries are asked; nothing is asked of the operator until they answer + for range 2 { + select { + case <-answered: + case <-time.After(10 * time.Second): // a bound on a failure only: the answers come at once + t.Fatalf("the entries were not asked: asks %v", asksKV.values) + } + } + pass() + rkey, _ := asks.RevertAskKey("ace", "web", fpOld) + bkey, _ := asks.BackupAskKey("ace", "web", fpOld) + if !p.w.has(buckets.Asks, rkey) || p.w.has(buckets.Asks, bkey) { + t.Fatalf("asks %v: want the revert ask %s and no backup ask", asksKV.values, rkey) + } + _, ask, err := assign.DecodeAsk(rkey, p.w.read(buckets.Asks)[rkey]) + if err != nil || !strings.Contains(ask.Explanation, "sessions") || !strings.Contains(ask.Explanation, "form 1") { + t.Fatalf("the revert ask = %+v, %v", ask, err) + } + fkey := "delivery.ace.web.revert-held" + if _, raised := findings[fkey]; !raised { + t.Fatalf("findings %v: want %s", findings, fkey) + } + + answer := func(option string) { + v, _ := mdc.EncodeAnswer(mdc.Answer{Option: option, Warrant: ask.ID + "@2026-11-12T10:05:00Z", Words: "the operator"}) + p.w.put(buckets.Answers, rkey, v) + } + puts := wantsKV.puts + answer(assign.OptionRevertHold) + pass() + if wantsKV.puts != puts { + t.Fatal("an answer hold wrote a want") + } + answer(assign.OptionRevertLive) + pass() + got, err := mwc.DecodeWant(p.w.read(buckets.Wants)[wkey]) + if err != nil || got.Fp != fpOld || got.Revert != mwc.RevertLive || got.Words != `the operator answered "live" to the revert ask (asking 1)` { + t.Fatalf("the want after live = %+v, %v", got, err) + } + + have(fpOld, nodeengineclient.PhaseRunning, "") // the reverted build runs: the finding clears + pass() + if _, raised := findings[fkey]; raised { + t.Fatal("the finding stands while the reverted build runs") + } + + // A newer want replaces the held revert: the ask goes and the finding with it. + have(fpTest, nodeengineclient.PhaseHeld, "restore point damaged: the point does not read") + pass() + if _, raised := findings[fkey]; !raised { + t.Fatal("the finding is not raised again for a hold again") + } + raw, _ = mwc.EncodeWant(mwc.Want{Fp: fpTest, Self: true}) + p.w.put(buckets.Wants, wkey, raw) + p.staleAsks(ctx, map[string]meshcatalog.ModulesByModuleValue{"web": {Fp2: fpTest}}) + if _, raised := findings[fkey]; raised || p.w.has(buckets.Asks, rkey) { + t.Fatalf("a newer want left the ask (%v) or the finding (%v)", p.w.has(buckets.Asks, rkey), raised) + } +} diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 5b287c86..f453d867 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -134,7 +134,7 @@ func TestManifest(t *testing.T) { t.Errorf("consumes = %v, want the catalog half's revisions (design 47 G §195), the teller's ask-ended, the engines' reads changed and what they run", m.Consumes) } if want := []string{"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", - "seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list + "seat:mesh-catalog.entry", "seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list "seat:mesh-conditions.conditions", // design 47 G §706: the conditions read at each judging of an outside wait "seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", // ADR 0565 §9, §11: the seat alone "seat:node-backup.backed-up", "seat:node-backup.now", diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index d4bc86ac..cc25c1c1 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -67,6 +67,7 @@ "invokes": [ "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", + "seat:mesh-catalog.entry", "seat:mesh-catalog.registrations", {"verb": "seat:mesh-catalog.revision", "fields": ["none-yet"]}, "seat:mesh-catalog.seat-definitions",