From 04c2c60f02441310dfb453ceb7427b978a461f49 Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 24 Oct 2026 16:37:37 +0200 Subject: [PATCH] The assigner acts on answers: a release writes the held want again, a backup answer goes into the node's want, a record move waits for its yes, a restore answer sets a data module back (issue 1906 D4) --- .../integration_test.go | 182 ++++++++++++++++++ .../cmd/mesh-delivery-assigner/main.go | 2 + .../cmd/mesh-delivery-assigner/pass.go | 118 +++++++++++- .../cmd/mesh-delivery-assigner/world.go | 12 ++ .../internal/assign/asks.go | 75 +++++++- .../internal/assign/asks_test.go | 40 ++++ .../internal/assign/wants.go | 52 ++++- .../internal/assign/wants_test.go | 127 ++++++++++++ .../mesh-delivery-assigner/manifest_test.go | 2 +- modules/mesh-delivery-assigner/module.json | 1 + 10 files changed, 602 insertions(+), 9 deletions(-) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index 16f7e01..93dc6d1 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -22,6 +22,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/provider" + + "mesh-delivery-assigner/internal/assign" ) // bed is an embedded bus with JetStream, every bucket the assigner reads and keeps, and the stream the mesh's events @@ -515,3 +517,183 @@ func TestAWantWaitsForItsComposition(t *testing.T) { t.Fatalf("the node's own want = %+v, %v", w, err) } } + +// fixture is a module web (or the manifest given) at fp on the node ace, which is online: built, granted and assigned. +func (b *bed) fixture(t *testing.T, name, fp, manifest string) { + t.Helper() + digest := b.manifest(t, name, manifest) + mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/" + name, Commit: strings.Repeat("1", 40), Path: delivery.RootPath}) + b.putJSON(t, delivery.ModulesBucket, name, mod) + built, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, By: "builder.novox.1", Step: "pushed", Manifest: digest}) + bkey, _ := delivery.BuilderBuildKey(name, fp) + b.putJSON(t, delivery.BuilderBuildsBucket, bkey, built) + gkey, _ := delivery.ControllerGrantKey(name, fp) + grant, _ := delivery.EncodeControllerGrant(delivery.ControllerGrant{}) + b.putJSON(t, delivery.ControllerGrantsBucket, gkey, grant) + on, _ := delivery.EncodePresence(delivery.Presence{Online: true}) + b.putJSON(t, delivery.PresenceBucket, "ace", on) + b.putJSON(t, provider.AssignmentsBucket, "ace."+name, []byte(`{"node":"ace","module":"`+name+`","build":"b-1"}`)) +} + +func (b *bed) putWant(t *testing.T, node, module string, w delivery.Want) { + t.Helper() + key, _ := delivery.WantKey(node, module) + v, err := delivery.EncodeWant(w) + if err != nil { + t.Fatal(err) + } + b.putJSON(t, delivery.AssignerWantsBucket, key, v) +} + +// wantNow waits until the want of module on node satisfies ok and answers it. +func (b *bed) wantNow(t *testing.T, node, module, what string, ok func(delivery.Want) bool) delivery.Want { + t.Helper() + key, _ := delivery.WantKey(node, module) + var last delivery.Want + for deadline := time.Now().Add(15 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if e, err := b.buckets[delivery.AssignerWantsBucket].Get(b.ctx, key); err == nil { + if w, err := delivery.DecodeWant(e.Value()); err == nil { + last = w + if ok(w) { + return w + } + } + } + } + t.Fatalf("never saw: %s; the want is %+v", what, last) + return last +} + +// answer says the teller wrote the operator's answer to the ask at key. +func (b *bed) answer(t *testing.T, key, option string) { + t.Helper() + v, err := delivery.EncodeAnswer(delivery.Answer{Option: option, Warrant: "r1@2026-10-24T12:00:00Z", Words: "the operator"}) + if err != nil { + t.Fatal(err) + } + b.putJSON(t, delivery.AnswersBucket, key, v) +} + +func (b *bed) hasKey(bucket, key string) bool { + _, err := b.buckets[bucket].Get(b.ctx, key) + return err == nil +} + +// 1906/D4 (ADR 0539 §19), whole on the bus: the operator's release of a held fingerprint writes its want again, by the rules, +// and clears the condition. +func TestAReleaseAnswerWritesTheWantAgain(t *testing.T) { + b := startBed(t, "") + fNew, fOld := delivery.SourcePrefix2+strings.Repeat("b", 64), delivery.SourcePrefix2+strings.Repeat("a", 64) + b.fixture(t, "web", fNew, `{"module":"web","version":"1"}`) + b.compose(t, "ace", "web", fNew, 'b', delivery.JudgedNeither) + b.compose(t, "ace", "web", fOld, 'a', delivery.JudgedNeither) + healthy, _ := delivery.EncodeHave(delivery.Have{FP: fOld, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}) + b.putJSON(t, delivery.HaveBucket, "ace.web", healthy) // it ran the older fingerprint healthy before + failed, _ := delivery.EncodeHave(delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed, Detail: "connection refused"}) + b.putJSON(t, delivery.HaveBucket, "ace.web", failed) + b.putWant(t, "ace", "web", delivery.Want{FP: fNew, Composed: "sha256:" + strings.Repeat("b", 64)}) + b.start(t) + w := b.wantNow(t, "ace", "web", "set back to the older fingerprint", func(w delivery.Want) bool { return w.FP == fOld }) + if w.Newest != fNew { + t.Fatalf("the set-back want = %+v", w) + } + hold, _ := delivery.HoldAskKey("web", fNew) + b.get(t, delivery.AssignerAsksBucket, hold) + b.answer(t, hold, "release") + w = b.wantNow(t, "ace", "web", "the held fingerprint's want written again", func(w delivery.Want) bool { return w.FP == fNew }) + if w.Composed != "sha256:"+strings.Repeat("b", 64) || w.Rule != 0 { + t.Errorf("the want = %+v", w) + } + for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) { + cleared := false + for _, c := range b.conditionsSaid(t) { + cleared = cleared || (c.Subject.ID == "web" && c.State == asks.StateCleared) + } + if cleared { + return + } + } + t.Error("the condition of the held fingerprint was not cleared") +} + +// 1906/D4 (ADR 0539 §16), whole on the bus: a module held on a node because its backup failed is asked about, and the answer +// is written into the node's want. +func TestABackupAnswerIsWrittenIntoTheNodesWant(t *testing.T) { + b := startBed(t, "") + fNew := delivery.SourcePrefix2 + strings.Repeat("b", 64) + b.fixture(t, "web", fNew, `{"module":"web","version":"1"}`) + b.compose(t, "ace", "web", fNew, 'b', delivery.JudgedNeither) + b.putWant(t, "ace", "web", delivery.Want{FP: fNew, Composed: "sha256:" + strings.Repeat("b", 64)}) + held, _ := delivery.EncodeHave(delivery.Have{FP: fNew, Phase: delivery.PhaseHeld, Health: delivery.HealthWaits, Detail: "no space left"}) + b.putJSON(t, delivery.HaveBucket, "ace.web", held) + b.start(t) + akey, _ := delivery.BackupAskKey("ace", "web", fNew) + a, err := delivery.DecodeAssignerAsk(akey, b.get(t, delivery.AssignerAsksBucket, akey)) + if err != nil || !strings.Contains(a.Ask.Explanation, "no space left") { + t.Fatalf("the backup ask = %+v, %v", a, err) + } + b.answer(t, akey, delivery.BackupSkip) + w := b.wantNow(t, "ace", "web", "the answer in the want", func(w delivery.Want) bool { return w.Backup == delivery.BackupSkip }) + if w.FP != fNew || w.Composed != "sha256:"+strings.Repeat("b", 64) { + t.Errorf("the want = %+v", w) + } + time.Sleep(time.Second) + if w := b.wantNow(t, "ace", "web", "the answer kept", func(delivery.Want) bool { return true }); w.Backup != delivery.BackupSkip { + t.Errorf("a later pass erased the answer: %+v", w) + } +} + +const recorded = `{"module":"web","version":"1","upgrade":{"policy":"record","why":"a person takes each build"}}` + +// 1906/D4 (ADR 0539 §14), whole on the bus: a move of a module of policy record waits for its ask's yes when judged to recreate +// a container, to move data or not to be judgeable, at a new fingerprint and at a new revision of the same alike; one judged +// neither is written at once. +func TestARecordMoveWaitsForTheYesAndANeitherMoveDoesNot(t *testing.T) { + b := startBed(t, "") + fNew, fOld := delivery.SourcePrefix2+strings.Repeat("b", 64), delivery.SourcePrefix2+strings.Repeat("a", 64) + b.fixture(t, "web", fNew, recorded) + old := "sha256:" + strings.Repeat("a", 64) + b.putWant(t, "ace", "web", delivery.Want{FP: fOld, Composed: old}) + nw := b.compose(t, "ace", "web", fNew, 'b', delivery.JudgedRecreates) + b.start(t) + akey, _ := delivery.RecordAskKey("ace", "web", nw) + a, err := delivery.DecodeAssignerAsk(akey, b.get(t, delivery.AssignerAsksBucket, akey)) + if err != nil || !strings.Contains(a.Ask.Explanation, "recreates a container") { + t.Fatalf("the record ask = %+v, %v", a, err) + } + time.Sleep(time.Second) + if w := b.wantNow(t, "ace", "web", "the old want", func(delivery.Want) bool { return true }); w.FP != fOld { + t.Fatalf("the move was written before the yes: %+v", w) + } + b.answer(t, akey, assign.OptionYes) + b.wantNow(t, "ace", "web", "the move after the yes", func(w delivery.Want) bool { return w.FP == fNew && w.Composed == nw }) + // a new revision at the same fingerprint, judged to move data, waits as well + rev := b.compose(t, "ace", "web", fNew, 'c', delivery.JudgedMovesData) + rkey, _ := delivery.RecordAskKey("ace", "web", rev) + b.get(t, delivery.AssignerAsksBucket, rkey) + time.Sleep(time.Second) + if w := b.wantNow(t, "ace", "web", "the want", func(delivery.Want) bool { return true }); w.Composed != nw { + t.Fatalf("the revision was written before the yes: %+v", w) + } + // and a revision judged neither is written at once + free := b.compose(t, "ace", "web", fNew, 'd', delivery.JudgedNeither) + b.wantNow(t, "ace", "web", "a move judged neither written at once", func(w delivery.Want) bool { return w.Composed == free }) +} + +// 1906/D4 (ADR 0539 §7): a revision of the bus module's composition at the same fingerprint that is judged neither (its user +// list only) is written at once, and no ask is made; the bus's new fingerprint is not written. +func TestAUserListRevisionOfTheBusIsWrittenAtOnce(t *testing.T) { + b := startBed(t, "") + fNew, fOld := delivery.SourcePrefix2+strings.Repeat("b", 64), delivery.SourcePrefix2+strings.Repeat("a", 64) + b.fixture(t, "bus", fNew, `{"module":"bus","version":"1","provides":[{"name":"mesh-bus"}]}`) + b.putWant(t, "ace", "bus", delivery.Want{FP: fOld, Composed: "sha256:" + strings.Repeat("a", 64)}) + b.compose(t, "ace", "bus", fNew, 'b', delivery.JudgedNeither) + b.start(t) + time.Sleep(time.Second) + if w := b.wantNow(t, "ace", "bus", "the want", func(delivery.Want) bool { return true }); w.FP != fOld { + t.Fatalf("the assigner moved the bus to a new fingerprint: %+v", w) + } + b.putWant(t, "ace", "bus", delivery.Want{FP: fNew, Composed: "sha256:" + strings.Repeat("b", 64)}) + free := b.compose(t, "ace", "bus", fNew, 'e', delivery.JudgedNeither) + b.wantNow(t, "ace", "bus", "the user-list revision written at once", func(w delivery.Want) bool { return w.Composed == free }) +} 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 980abe2..e4b5910 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -42,6 +42,7 @@ var readBuckets = []string{ provider.AssignmentsBucket, delivery.ControllerGrantsBucket, delivery.ControllerComposedBucket, + delivery.AnswersBucket, delivery.BuilderBuildsBucket, delivery.CandidatesBucket, delivery.ManifestsBucket, @@ -134,6 +135,7 @@ var watched = []string{ provider.AssignmentsBucket, delivery.ControllerGrantsBucket, delivery.ControllerComposedBucket, + delivery.AnswersBucket, delivery.BuilderBuildsBucket, delivery.CandidatesBucket, delivery.ModulesBucket, 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 ceb7223..df54bcd 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -30,6 +30,7 @@ func (p *passer) pass(ctx context.Context) { p.compositions(ctx) p.wants(ctx) p.nodeWants(ctx) + p.backups(ctx) } // nodeWants writes each node's own want, `want..node`, naming the digest of the node's own composition, once the @@ -84,11 +85,24 @@ func (p *passer) wants(ctx context.Context) { fp := trunk[name].FP2 name := name m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Nodes: nodesOf[name], - Composed: func(node, fp string) (delivery.ComposedDigest, bool) { return p.w.composed(node, name, fp) }} + Composed: func(node, fp string) (delivery.ComposedDigest, bool) { return p.w.composed(node, name, fp) }, + Restored: func(node string) bool { + key, err := delivery.RestoreAskKey(node, name, fp) + return err == nil && p.answeredWith(key, assign.OptionRestore) + }, + Allowed: func(node, digest string) bool { + key, err := delivery.RecordAskKey(node, name, digest) + return err == nil && p.answeredWith(key, assign.OptionYes) + }} if fp != "" { held, _ := delivery.HeldKey(name, fp) grant, _ := delivery.ControllerGrantKey(name, fp) - m.Built, m.Held, m.Granted = p.w.built(name, fp), p.w.has(delivery.AssignerWantsBucket, held), p.w.has(delivery.ControllerGrantsBucket, grant) + hold, _ := delivery.HoldAskKey(name, fp) + m.Released = p.answeredWith(hold, "release") + m.Built, m.Held, m.Granted = p.w.built(name, fp), p.w.has(delivery.AssignerWantsBucket, held) && !m.Released, p.w.has(delivery.ControllerGrantsBucket, grant) + if m.Released { + p.clearHeld(name, fp) + } } last := map[string]string{} for _, node := range m.Nodes { @@ -109,6 +123,9 @@ func (p *passer) wants(ctx context.Context) { for _, node := range out.Restore { p.restoreAsk(ctx, name, node, fp) } + for _, need := range out.Record { + p.recordAsk(ctx, name, fp, need) + } for node, want := range out.Wants { key, err := delivery.WantKey(node, name) if err != nil { @@ -227,3 +244,100 @@ func (p *passer) restoreAsk(ctx context.Context, module, node, fp string) { a := assign.RestoreAsk(key, node, module, fp, words, time.Now()) p.put(ctx, delivery.AssignerAsksBucket, key, func() ([]byte, error) { return delivery.EncodeAssignerAsk(key, a) }) } + +// answeredWith says whether the teller wrote an answer to the ask at key and the option chosen is option. The answer is the +// operator's, given in the view or on the phone, and stays under its key (ADR 0539 §14, §16, §19). +func (p *passer) answeredWith(key, option string) bool { + a, ok := p.w.answer(key) + return ok && a.Option == option +} + +// clearHeld clears, once, the condition of a hold the operator released. +func (p *passer) clearHeld(module, fp string) { + c := assign.HeldCondition(module, fp, "", "", false, false) + if p.say == nil || p.said[c.Key+".cleared"] { + return + } + if err := p.say(c); err != nil { + p.logf("the condition of the released %s could not be cleared, and is cleared at the next pass: %v", module, err) + return + } + p.said[c.Key+".cleared"] = true +} + +// recordAsk writes, once, the ask about a move of a module of policy record that waits for the operator's yes. +func (p *passer) recordAsk(ctx context.Context, module, fp string, need assign.RecordNeed) { + key, err := delivery.RecordAskKey(need.Node, module, need.Digest) + if err != nil || p.w.has(delivery.AssignerAsksBucket, key) { + return + } + a := assign.RecordAsk(key, need.Node, module, fp, need, time.Now()) + p.put(ctx, delivery.AssignerAsksBucket, key, func() ([]byte, error) { return delivery.EncodeAssignerAsk(key, a) }) +} + +// backups sees to the modules held on a node because the backup before an apply failed (the have's phase held, ADR 0539 §16): +// it asks, once for each asking, whether to back up again or go on without a backup, raises the condition naming node, module and +// the backup's words, and writes the answer into the node's want for the module, which the node agent reads from it and from +// nothing else. A backup that fails again after the answer again is asked again, as the second asking. +func (p *passer) backups(ctx context.Context) { + wanted := p.w.wantsOf() + for key, value := range p.w.read(delivery.HaveBucket) { + node, module, isNode, err := delivery.ParseHaveKey(key) + if err != nil || isNode { + continue + } + have, err := delivery.DecodeHave(value) + if err != nil || have.Phase != delivery.PhaseHeld { + continue + } + want, ok := wanted[module][node] + if !ok || want.Removed { + continue + } + akey, err := delivery.BackupAskKey(node, module, want.FP) + if err != nil { + continue + } + answer, answered := p.w.answer(akey) + switch { + case answered && (answer.Option == delivery.BackupAgain || answer.Option == delivery.BackupSkip): + if want.Backup != answer.Option { + want.Backup = answer.Option + wkey, err := delivery.WantKey(node, module) + if err == nil { + w := want + p.put(ctx, delivery.AssignerWantsBucket, wkey, func() ([]byte, error) { return delivery.EncodeWant(w) }) + } + continue + } + if answer.Option == delivery.BackupAgain { + // the node tried again and the backup failed again: the second asking, once + p.backupAsk(ctx, akey, node, module, want.FP, have.Detail, 2) + } + case !answered: + p.backupAsk(ctx, akey, node, module, want.FP, have.Detail, 0) + c := assign.BackupCondition(node, module, want.FP, have.Detail, true) + if p.say != nil && !p.said[c.Key] { + if err := p.say(c); err != nil { + p.logf("the condition of the failed backup of %s on %s could not be said: %v", module, node, err) + } else { + p.said[c.Key] = true + } + } + } + } +} + +// backupAsk writes the backup ask at its asking, unless that asking stands already. +func (p *passer) backupAsk(ctx context.Context, key, node, module, fp, words string, asking int) { + if p.w.has(delivery.AssignerAsksBucket, key) { + if asking == 0 { + return + } + if a, err := delivery.DecodeAssignerAsk(key, p.w.read(delivery.AssignerAsksBucket)[key]); err == nil && a.Ask.ID == delivery.AskingID(key, asking) { + return + } + } + a := assign.BackupAsk(key, node, module, fp, words, asking, time.Now()) + p.put(ctx, delivery.AssignerAsksBucket, key, func() ([]byte, error) { return delivery.EncodeAssignerAsk(key, a) }) +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index 48bcbea..04ee003 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -374,3 +374,15 @@ func (w *world) nodeComposed() map[string]delivery.ComposedDigest { } return out } + +// answer is the teller's answer to the ask at key (mesh-delivery-teller_answers, the same key), and whether there is one. +func (w *world) answer(key string) (delivery.Answer, bool) { + w.mu.Lock() + value, ok := w.values[delivery.AnswersBucket][key] + w.mu.Unlock() + if !ok { + return delivery.Answer{}, false + } + a, err := delivery.DecodeAnswer(value) + return a, err == nil +} diff --git a/modules/mesh-delivery-assigner/internal/assign/asks.go b/modules/mesh-delivery-assigner/internal/assign/asks.go index 6be4437..33e48ff 100644 --- a/modules/mesh-delivery-assigner/internal/assign/asks.go +++ b/modules/mesh-delivery-assigner/internal/assign/asks.go @@ -10,17 +10,16 @@ import ( ) // HoldAsk is the ask about a module's fingerprint the assigner holds, at its key (delivery.HoldAskKey), first asking: keep -// it held or release it. Both options acknowledge and bind nothing: in release 5 the wants are only compared, so no answer -// changes what a node runs (ADR 0429 §7); the node agent acts on the hold from release 6. +// it held or release it. Releasing writes the held fingerprint's want again by the rules (ADR 0539 §19). func HoldAsk(key, module, fp string, now time.Time) delivery.AssignerAsk { return delivery.AssignerAsk{Ask: asks.Ask{ ID: delivery.AskingID(key, 0), Headline: fmt.Sprintf("%s is held at %s", module, short(fp)), Explanation: fmt.Sprintf("A node's health check of %s failed at %s, so the assigner holds that version and wants "+ - "no node to move to it. Until release 6 this changes nothing a node runs: the controller still sends.", module, short(fp)), + "no node to move to it. Releasing it lets the wants move to it again.", module, short(fp)), Options: []asks.Option{ {ID: "keep", Label: "Keep it held", Does: "the version stays held", Level: asks.Acknowledge}, - {ID: "release", Label: "Release it", Does: "the version is released when wants are applied, from release 6", Level: asks.Acknowledge}, + {ID: "release", Label: "Release it", Does: "the version is released and the wants move to it by the rules again", Level: asks.Acknowledge}, }, Who: asks.Operator, Expires: now.Add(7 * 24 * time.Hour), @@ -77,3 +76,71 @@ func HeldCondition(module, fp, node, words string, raised, needsYou bool) asks.C } return c } + +// The options of the record ask (ADR 0539 §14) and of the backup ask (§16). +const ( + OptionYes = "yes" + OptionNo = "no" + OptionBackupAgain = delivery.BackupAgain + OptionBackupSkip = delivery.BackupSkip +) + +// RecordAsk is the ask about a move of a module of policy record on node to the composition with the digest, judged by the +// controller to recreate a container, to move data, or not to be judgeable (ADR 0539 §14). The want names the composition only +// after the answer yes. The way the move was judged is in the ask. +func RecordAsk(key, node, module, fp string, need RecordNeed, now time.Time) delivery.AssignerAsk { + what := map[delivery.Judgement]string{ + delivery.JudgedRecreates: "recreates a container", + delivery.JudgedMovesData: "moves the module's data", + delivery.JudgedCannot: "cannot be judged: " + need.Words, + }[need.Judgement] + return delivery.AssignerAsk{Ask: asks.Ask{ + ID: delivery.AskingID(key, 0), + Headline: fmt.Sprintf("%s waits for your yes on %s", module, node), + Explanation: fmt.Sprintf("%s is recorded: a person takes each of its moves. The next one on %s (version %s) %s. "+ + "The node is told to apply it only after you say yes.", module, node, short(fp), what), + Options: []asks.Option{ + {ID: OptionYes, Label: "Yes, move it", Does: "the node is told to apply the new composition of " + module, Level: asks.Acknowledge}, + {ID: OptionNo, Label: "No, not now", Does: "the node keeps what it runs", Level: asks.Acknowledge}, + }, + Who: asks.Operator, + Expires: now.Add(7 * 24 * time.Hour), + OnExpiry: "the node keeps what it runs", + About: key, + }} +} + +// BackupAsk is the ask about a backup of a module on node that failed before an apply (ADR 0539 §16): back up again, or go on +// without a backup. asking is the number of the asking (0 the first, 2 the second: delivery.AskingID), so a backup that fails again after the answer again is asked +// again. words are the backup's own words, from the node's have. +func BackupAsk(key, node, module, fp, words string, asking int, now time.Time) delivery.AssignerAsk { + return delivery.AssignerAsk{Ask: asks.Ask{ + ID: delivery.AskingID(key, asking), + Headline: fmt.Sprintf("The backup of %s failed on %s", module, node), + Explanation: fmt.Sprintf("The backup before applying %s at %s on %s failed: %s. %s keeps running what it has until you answer.", + module, short(fp), node, words, module), + Options: []asks.Option{ + {ID: OptionBackupAgain, Label: "Back up again", Does: "the node tries the backup again before it applies", Level: asks.Acknowledge}, + {ID: OptionBackupSkip, Label: "Go on without a backup", Does: "the node applies the new version without a backup", Level: asks.Acknowledge}, + }, + Who: asks.Operator, + Expires: now.Add(7 * 24 * time.Hour), + OnExpiry: "the module keeps running what it has", + About: key, + }} +} + +// BackupCondition is the condition raised when a backup before an apply fails and the module is held on the node (ADR 0539 §16), +// naming the node, the module and the backup's words; cleared under the same key. +func BackupCondition(node, module, fp, words string, raised bool) asks.Condition { + key := delivery.AssignerModule + ".backup." + node + "." + module + "." + strings.TrimPrefix(fp, delivery.SourcePrefix2) + subject := asks.Scope{Scope: "module", ID: module, Machine: node} + if !raised { + return asks.Condition{Key: key, Subject: subject, State: asks.StateCleared} + } + headline := fmt.Sprintf("The backup of %s failed on %s", module, node) + needs := "answer whether to back up again or go on without a backup" + return asks.Condition{Key: key, Severity: asks.SeverityWarning, Subject: subject, State: asks.StateRaised, + Summary: headline, Headline: headline, Needs: needs, + Explanation: asks.NeedsYou + needs + ". The backup before applying " + module + " at " + short(fp) + " on " + node + " failed: " + words + "."} +} diff --git a/modules/mesh-delivery-assigner/internal/assign/asks_test.go b/modules/mesh-delivery-assigner/internal/assign/asks_test.go index a1bf789..0e2fd77 100644 --- a/modules/mesh-delivery-assigner/internal/assign/asks_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/asks_test.go @@ -73,3 +73,43 @@ func TestTheHeldConditionNamesModuleNodeAndWords(t *testing.T) { t.Errorf("cleared = %+v", got) } } + +// 1906/D4: the record ask and the backup ask are valid asks of the router and of mesh-sdk's layout at their keys, answerable in +// the view, and name the move and the node. +func TestTheRecordAndBackupAsksAreValid(t *testing.T) { + now := time.Now() + rkey, err := delivery.RecordAskKey("ace", "mail", "sha256:"+strings.Repeat("b", 64)) + if err != nil { + t.Fatal(err) + } + r := RecordAsk(rkey, "ace", "mail", fNew, RecordNeed{Node: "ace", Digest: "sha256:" + strings.Repeat("b", 64), Judgement: delivery.JudgedCannot, Words: "the composition it runs is gone"}, now) + bkey, err := delivery.BackupAskKey("ace", "mail", fNew) + if err != nil { + t.Fatal(err) + } + b := BackupAsk(bkey, "ace", "mail", fNew, "no space left", 0, now) + for key, a := range map[string]delivery.AssignerAsk{rkey: r, bkey: b} { + if err := a.Ask.Check(now); err != nil { + t.Errorf("%s: the router refuses the ask: %v", key, err) + } + if _, err := delivery.EncodeAssignerAsk(key, a); err != nil { + t.Errorf("%s: cannot be written: %v", key, err) + } + if _, err := delivery.ViewAnswerKey(key); err != nil { + t.Errorf("%s: the view may not answer it: %v", key, err) + } + } + if !strings.Contains(r.Ask.Explanation, "gone") || r.Ask.Options[0].ID != OptionYes { + t.Errorf("the record ask = %+v", r.Ask) + } + if !strings.Contains(b.Ask.Explanation, "no space left") || b.Ask.Options[0].ID != delivery.BackupAgain || b.Ask.Options[1].ID != delivery.BackupSkip { + t.Errorf("the backup ask = %+v", b.Ask) + } + if again := BackupAsk(bkey, "ace", "mail", fNew, "x", 2, now); again.Ask.ID == b.Ask.ID { + t.Errorf("asked again under the same id") + } + c := BackupCondition("ace", "mail", fNew, "no space left", true) + if _, err := asks.EncodeCondition(c); err != nil { + t.Errorf("the backup condition is refused: %v", err) + } +} diff --git a/modules/mesh-delivery-assigner/internal/assign/wants.go b/modules/mesh-delivery-assigner/internal/assign/wants.go index d10e2a4..6788ec9 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants.go @@ -23,6 +23,14 @@ type Module struct { // Composed answers the controller's composition of the module for a node at a fingerprint (mesh-controller_composed): its // digest and the judgement of the move to it, and whether it exists. Nil: none exists (ADR 0539 §1, §6). Composed func(node, fp string) (delivery.ComposedDigest, bool) + // Allowed says whether the operator answered yes to the record ask about the move of this module on node to the composition + // with this digest (ADR 0539 §14). Nil: none did. + Allowed func(node, digest string) bool + // Released is a hold on FP the operator released (ADR 0539 §19): FP is not held again by a health check that failed. + Released bool + // Restored says whether the operator answered "restore" to the restore ask of this data module on node: it is then set back + // as any module is, losing the writes made since (ADR 0539 §19). Nil: none did. + Restored func(node string) bool } func (m Module) composition(node, fp string) (delivery.ComposedDigest, bool) { @@ -130,6 +138,8 @@ type Outcome struct { Wants map[string]delivery.Want // Held is the hold of m.FP to write, when a node's health check failed on it and it is not held yet. Held *delivery.Held + // Record is the moves a person must allow before the want names them: each is asked once (ADR 0539 §14). + Record []RecordNeed // Restore is the nodes of a module holding data whose held fingerprint failed its health check there and that are not set // back: the operator is asked whether to restore each (ADR 0539 §19). Restore []string @@ -137,6 +147,15 @@ type Outcome struct { Rules map[string]Holding } +// RecordNeed is a move of a module of policy record on a node that waits for the operator's yes: the composition's digest, the +// controller's judgement of the move and its words. +type RecordNeed struct { + Node string + Digest string + Judgement delivery.Judgement + Words string +} + // Holding is a rule holding a node back, and why. type Holding struct { Rule int @@ -151,7 +170,7 @@ type Holding struct { func Wants(m Module, n Nodes, wants map[string]delivery.Want, lastHealthy map[string]string) Outcome { out := Outcome{Wants: map[string]delivery.Want{}, Rules: map[string]Holding{}} held := m.Held - if !held && m.FP != "" { + if !held && !m.Released && m.FP != "" { for _, node := range slices.Sorted(slices.Values(m.Nodes)) { if h, ok := n.Haves[node][m.Name]; ok && h.FP == m.FP && h.Health == delivery.HealthFailed { out.Held = &delivery.Held{Node: node, Words: h.Detail} @@ -172,11 +191,20 @@ func Wants(m Module, n Nodes, wants map[string]delivery.Want, lastHealthy map[st case ok: d, _ := m.composition(node, m.FP) next = delivery.Want{FP: m.FP, Composed: d.Digest} + if had && !before.Removed && before.FP == m.FP && before.Composed == d.Digest { + next.Backup = before.Backup // the answer to a failed backup stays while its target stays (ADR 0539 §16) + } + if need, wait := m.mustAsk(node, before, had, d); wait { + if need != nil { + out.Record = append(out.Record, *need) + } + continue + } case held && had && before.FP == m.FP: out.Rules[node] = Holding{Rule: rule, Words: words} last := lastHealthy[node] switch { - case keepsData(m.Manifest): + case keepsData(m.Manifest) && (m.Restored == nil || !m.Restored(node)): next = before next.Words = fmt.Sprintf("%s holds data, so it is not set back from %s: the operator is asked whether to restore it", m.Name, short(m.FP)) out.Restore = append(out.Restore, node) @@ -227,3 +255,23 @@ func short(fp string) string { } return part[:12] } + +// mustAsk says whether the move to the composition d on node waits: for the operator's yes (need names the ask) or, for the bus, +// for the operator's hand (ADR 0539 §14, §15). A module the node never ran, a move judged neither, a module that is not of +// policy record and a want that stays are written as any want; so is a revision of the bus module's composition at the +// same fingerprint judged neither (a user list only, §7). +func (m Module) mustAsk(node string, before delivery.Want, had bool, d delivery.ComposedDigest) (*RecordNeed, bool) { + if !had || before.Removed || (before.FP == m.FP && before.Composed == d.Digest) { + return nil, false + } + if catalogue.ProvidesBus(m.Manifest) && before.FP != m.FP { + return nil, true // the bus is never moved by the assigner; the operator writes its want (§15) + } + if policy, _, _ := catalogue.DerivedUpgrade(m.Manifest); policy != catalogue.PolicyRecord || d.Judgement == delivery.JudgedNeither { + return nil, false + } + if m.Allowed != nil && m.Allowed(node, d.Digest) { + return nil, false + } + return &RecordNeed{Node: node, Digest: d.Digest, Judgement: d.Judgement, Words: d.Words}, true +} diff --git a/modules/mesh-delivery-assigner/internal/assign/wants_test.go b/modules/mesh-delivery-assigner/internal/assign/wants_test.go index d29ceb5..4b7b303 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants_test.go @@ -241,3 +241,130 @@ func TestASetBackNamesTheOlderCompositionOrStays(t *testing.T) { t.Errorf("set-back want = %+v", w) } } + +// movingTo is a module ready to move, whose composition at F is judged j against the one it runs, and whose digest is dig(F). +func movingTo(j delivery.Judgement) (Module, Nodes, map[string]delivery.Want) { + m, n := ready() + m.Composed = func(node, fp string) (delivery.ComposedDigest, bool) { + d := delivery.ComposedDigest{Digest: dig(fp), Judgement: j} + if j != delivery.JudgedNeither { + d.Against, d.Words = dig(fOld), "judged" + } + return d, true + } + return m, n, wantedOld() +} + +func recording(m *Module) { + m.Manifest.Upgrade = &catalogue.UpgradePolicy{Policy: catalogue.PolicyRecord, Why: "a person takes each build"} +} + +// 1906/D4 (ADR 0539 §14): a module of policy record whose move is judged to recreate a container, to move data, or cannot be +// judged, is not written until its ask is answered yes; the same at a new revision of the same fingerprint. +func TestARecordModuleWaitsForItsAskWhenTheMoveIsNotJudgedNeither(t *testing.T) { + for _, j := range []delivery.Judgement{delivery.JudgedRecreates, delivery.JudgedMovesData, delivery.JudgedCannot} { + m, n, wanted := movingTo(j) + recording(&m) + out := Wants(m, n, wanted, nil) + if w, ok := out.Wants["ace"]; ok { + t.Errorf("%s: ace's want was written without the ask's yes: %+v", j, w) + } + var ace *RecordNeed + for i := range out.Record { + if out.Record[i].Node == "ace" { + ace = &out.Record[i] + } + } + if ace == nil || ace.Digest != dig(fNew) || ace.Judgement != j { + t.Errorf("%s: the asks = %+v, want one for ace naming the digest and the judgement", j, out.Record) + } + // a new revision at the same fingerprint is asked too + same := map[string]delivery.Want{"ace": {FP: fNew, Composed: dig(fOld)}} + if out := Wants(m, n, same, nil); len(out.Record) == 0 || len(out.Wants) != 0 && out.Wants["ace"].Composed == dig(fNew) { + t.Errorf("%s: a new revision at the same fingerprint: %+v", j, out) + } + m.Allowed = func(node, digest string) bool { return node == "ace" && digest == dig(fNew) } + out = Wants(m, n, wanted, nil) + if w := out.Wants["ace"]; w.FP != fNew || w.Composed != dig(fNew) { + t.Errorf("%s: after the yes ace's want = %+v", j, w) + } + } +} + +// 1906/D4: a move judged neither is written at once, whatever the policy; so is a module the node never ran; and a module +// that rolls (not record) is never asked. +func TestAMoveJudgedNeitherOrNeverRunOrNotRecordedIsWrittenAtOnce(t *testing.T) { + m, n, wanted := movingTo(delivery.JudgedNeither) + recording(&m) + if out := Wants(m, n, wanted, nil); out.Wants["ace"].FP != fNew || len(out.Record) != 0 { + t.Errorf("neither: %+v", out) + } + m, n, _ = movingTo(delivery.JudgedRecreates) + recording(&m) + if out := Wants(m, n, nil, nil); out.Wants["ace"].FP != fNew || len(out.Record) != 0 { + t.Errorf("a node that never ran it: %+v", out) + } + m, n, wanted = movingTo(delivery.JudgedRecreates) + if out := Wants(m, n, wanted, nil); out.Wants["ace"].FP != fNew || len(out.Record) != 0 { + t.Errorf("a rolled module: %+v", out) + } +} + +// 1906/D4 (ADR 0539 §7, §15): the bus module's new fingerprint is never written by the assigner, but a revision of its +// composition at the same fingerprint that only changes the user list (judged neither) is written at once. +func TestTheBusIsNotMovedButItsUserListRevisionIsWrittenAtOnce(t *testing.T) { + m, n, _ := movingTo(delivery.JudgedNeither) + m.Manifest.Provides = []catalogue.Offer{{Name: "mesh-bus"}} + if out := Wants(m, n, wantedOld(), nil); out.Wants["ace"].FP == fNew { + t.Errorf("the bus moved to a new fingerprint: %+v", out.Wants["ace"]) + } + running := map[string]delivery.Want{"ace": {FP: fNew, Composed: dig(fOld)}, "g14": {FP: fNew, Composed: dig(fOld)}} + out := Wants(m, n, running, nil) + if w := out.Wants["ace"]; w.FP != fNew || w.Composed != dig(fNew) { + t.Errorf("a user-list revision of the bus: ace's want = %+v", w) + } + if len(out.Record) != 0 { + t.Errorf("a user-list revision was asked: %+v", out.Record) + } +} + +// 1906/D4 (ADR 0539 §16): the answer to a failed backup stays in the want while its target stays. +func TestTheBackupAnswerStaysInTheWantWhileItsTargetStays(t *testing.T) { + m, n := ready() + running := map[string]delivery.Want{"ace": {FP: fNew, Composed: dig(fNew), Backup: delivery.BackupSkip}} + if w, ok := Wants(m, n, running, nil).Wants["ace"]; ok && w.Backup == "" { + t.Errorf("the answer was erased: %+v", w) + } +} + +// 1906/D4 (ADR 0539 §19): a hold released by the operator is not held again by a health check that failed before, and the +// wants move by the rules again. +func TestAReleasedHoldIsNotHeldAgain(t *testing.T) { + m, n := ready() + n.Haves["ace"]["mail"] = delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed, Detail: "x"} + m.Released = true + out := Wants(m, n, map[string]delivery.Want{"ace": {FP: fOld, Composed: dig(fOld), Newest: fNew, Rule: delivery.RuleNotHeld}}, map[string]string{"ace": fOld}) + if out.Held != nil { + t.Errorf("a released fingerprint was held again: %+v", out.Held) + } + if w := out.Wants["ace"]; w.FP != fNew || w.Composed != dig(fNew) || w.Rule != 0 { + t.Errorf("ace's want = %+v, want the held fingerprint written again", w) + } +} + +// 1906/D4: a data module whose restore ask the operator answered "restore" is set back like any other, losing the newer writes; +// answered "leave stopped" (or not answered) it stays. +func TestARestoreAnswerSetsADataModuleBack(t *testing.T) { + m, n := ready() + m.Manifest.Data = &catalogue.Data{Own: []catalogue.DataItem{{ID: "mail", Class: catalogue.ClassIrreplaceable}}} + n.Haves["ace"]["mail"] = delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed} + running := map[string]delivery.Want{"ace": {FP: fNew, Composed: dig(fNew)}} + m.Restored = func(node string) bool { return false } + if w, ok := Wants(m, n, running, map[string]string{"ace": fOld}).Wants["ace"]; ok && w.FP == fOld { + t.Errorf("set back without the operator's restore: %+v", w) + } + m.Restored = func(node string) bool { return node == "ace" } + if w := Wants(m, n, running, map[string]string{"ace": fOld}).Wants["ace"]; w.FP != fOld || w.Composed != dig(fOld) { + t.Errorf("after restore ace's want = %+v", w) + } +} diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 31f2192..e34793a 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -34,7 +34,7 @@ func TestManifest(t *testing.T) { slices.Sort(reads) want := []string{"mesh-controller.assignments", "mesh-controller.composed", "mesh-controller.grants", "mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", - "mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"} + "mesh-delivery-indexer.modules", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "node-engine.have"} if !reflect.DeepEqual(reads, want) { t.Errorf("reads = %v, want exactly %v", reads, want) } diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index 38541cc..5035591 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -10,6 +10,7 @@ "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", + "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "node-engine.have" ],