From 204a59dab417f3cfd61c8eea85060c0cb507a7f5 Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 24 Oct 2026 16:26:20 +0200 Subject: [PATCH] The assigner asks whether to restore a data module whose held fingerprint failed its health check, and raises the condition for a held fingerprint (issue 1906 D3) --- .../integration_test.go | 80 +++++++++++++++++++ .../cmd/mesh-delivery-assigner/main.go | 24 +++++- .../cmd/mesh-delivery-assigner/pass.go | 48 +++++++++++ .../internal/assign/asks.go | 50 ++++++++++++ .../internal/assign/asks_test.go | 47 +++++++++++ .../internal/assign/wants.go | 8 +- .../internal/assign/wants_test.go | 15 ++++ .../mesh-delivery-assigner/manifest_test.go | 3 + modules/mesh-delivery-assigner/module.json | 3 + 9 files changed, 273 insertions(+), 5 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 57cc144..ecd2d44 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 @@ -17,6 +17,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/announce" + "git.novox.be/novox/mesh-sdk/go/asks" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/provider" @@ -73,6 +74,10 @@ func startBed(t *testing.T, leaveOut string) *bed { if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil { t.Fatal(err) } + // The operator-channel seat's stream of conditions, which the assigner says its conditions on. + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "CONDITIONS", Subjects: []string{"mesh.seat.operator-channel.accept.condition.>"}}); err != nil { + t.Fatal(err) + } return b } @@ -294,3 +299,78 @@ func TestAHeldFingerprintIsAskedAbout(t *testing.T) { t.Fatalf("the ask = %+v, %v", a, err) } } + +// conditionsSaid reads what the assigner said on the operator channel's subject for conditions, by condition key. +func (b *bed) conditionsSaid(t *testing.T) map[string]asks.Condition { + t.Helper() + stream, err := b.js.Stream(b.ctx, "CONDITIONS") + if err != nil { + t.Fatal(err) + } + out := map[string]asks.Condition{} + cons, err := stream.CreateOrUpdateConsumer(b.ctx, jetstream.ConsumerConfig{AckPolicy: jetstream.AckNonePolicy}) + if err != nil { + t.Fatal(err) + } + for { + batch, err := cons.FetchNoWait(100) + if err != nil { + t.Fatal(err) + } + n := 0 + for m := range batch.Messages() { + n++ + c, err := asks.DecodeCondition(m.Data()) + if err != nil { + t.Fatal(err) + } + out[c.Key] = c + } + if n == 0 { + return out + } + } +} + +// 1906/D3 (ADR 0539 §19), whole on the bus: a module holding data whose new fingerprint failed its health check on the +// canary is held, not set back, and asked about (restore it, or leave it stopped); the condition names the module, the node +// and the check's words. +func TestAFailedDataModuleIsAskedToRestore(t *testing.T) { + b := startBed(t, "") + fNew := delivery.SourcePrefix2 + strings.Repeat("d", 64) + digest := b.manifest(t, "mail", `{"module":"mail","version":"1","data":{"own":[{"id":"mail","class":"irreplaceable"}]}}`) + mod, _ := delivery.EncodeModule(delivery.Module{FP2: fNew, Manifest: digest, Repository: "novox/mail", Commit: strings.Repeat("1", 40), Path: delivery.RootPath}) + b.putJSON(t, delivery.ModulesBucket, "mail", mod) + on, _ := delivery.EncodePresence(delivery.Presence{Online: true}) + b.putJSON(t, delivery.PresenceBucket, "ace", on) + failed, _ := delivery.EncodeHave(delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed, Detail: "connection refused"}) + b.putJSON(t, delivery.HaveBucket, "ace.mail", failed) + wantKey, _ := delivery.WantKey("ace", "mail") + want, _ := delivery.EncodeWant(delivery.Want{FP: fNew}) + b.putJSON(t, delivery.AssignerWantsBucket, wantKey, want) + b.putJSON(t, provider.AssignmentsBucket, "ace.mail", []byte(`{"node":"ace","module":"mail","build":"b-1"}`)) + b.start(t) + + akey, _ := delivery.RestoreAskKey("ace", "mail", fNew) + a, err := delivery.DecodeAssignerAsk(akey, b.get(t, delivery.AssignerAsksBucket, akey)) + if err != nil || len(a.Ask.Options) != 2 || a.Ask.Options[0].ID != "restore" || a.Ask.Options[1].ID != "leave-stopped" { + t.Fatalf("the restore ask = %+v, %v", a, err) + } + if w, err := delivery.DecodeWant(b.get(t, delivery.AssignerWantsBucket, wantKey)); err != nil || w.FP != fNew { + t.Errorf("the data module was set back: want = %+v, %v", w, err) + } + key := "" + for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline) && key == ""; time.Sleep(100 * time.Millisecond) { + for k, c := range b.conditionsSaid(t) { + if c.State == asks.StateRaised && c.Subject.ID == "mail" { + if c.Subject.Machine != "ace" || !strings.Contains(c.Explanation, "connection refused") { + t.Fatalf("the condition = %+v, want it to name ace and the check's words", c) + } + key = k + } + } + } + if key == "" { + t.Fatal("no condition for the held fingerprint was said") + } +} 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 eabfebc..7fdea8b 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -21,6 +21,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/announce" + "git.novox.be/novox/mesh-sdk/go/asks" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/provider" @@ -123,7 +124,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro logger.Printf("what the assigner keeps could not be said on the bus, so its readers are not granted until it is: %v", err) } logger.Printf("started on the account %s, reading %d buckets", account.User, len(kvs)) - return follow(ctx, kvs, logger) + return follow(ctx, kvs, sayVia(ctx, js), logger) } // watched are the buckets whose every value a pass reads; the manifests are read by digest when a pass needs one, and @@ -144,7 +145,7 @@ var watched = []string{ // follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after // each change; changes that come during a pass make one more. It returns when ctx ends or a watch ends. -func follow(ctx context.Context, kvs buckets, logger *log.Logger) error { +func follow(ctx context.Context, kvs buckets, say func(asks.Condition) error, logger *log.Logger) error { w := newWorld() dirty := make(chan struct{}, 1) wake := func() { @@ -208,7 +209,7 @@ func follow(ctx context.Context, kvs buckets, logger *log.Logger) error { case <-settled: } } - p := &passer{kvs: kvs, w: w, logf: logger.Printf} + p := &passer{kvs: kvs, w: w, logf: logger.Printf, say: say, said: map[string]bool{}} wake() for { select { @@ -222,3 +223,20 @@ func follow(ctx context.Context, kvs buckets, logger *log.Logger) error { } } } + +// sayVia publishes a condition on the operator-channel seat's one subject for conditions, under the assigner's own name +// (ADR 0437 clause 9). +func sayVia(ctx context.Context, js jetstream.JetStream) func(asks.Condition) error { + return func(c asks.Condition) error { + subject, err := asks.ConditionSubject(delivery.AssignerModule) + if err != nil { + return err + } + data, err := asks.EncodeCondition(c) + if err != nil { + return err + } + _, err = js.Publish(ctx, subject, data) + return err + } +} 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 4c7eeee..ea1a1b4 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -5,6 +5,7 @@ import ( "errors" "time" + "git.novox.be/novox/mesh-sdk/go/asks" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/sdkversion" @@ -19,6 +20,10 @@ type passer struct { w *world logf func(string, ...any) sdk string + // say publishes a condition to the operator; said is, by condition key and state, what this process has said, so a + // pass says a condition once and a restart says it again (the router keeps one per key). + say func(asks.Condition) error + said map[string]bool } func (p *passer) pass(ctx context.Context) { @@ -79,6 +84,10 @@ func (p *passer) wants(ctx context.Context) { } if out.Held != nil || m.Held { p.holdAsk(ctx, name, fp) + p.heldCondition(ctx, name, fp, out.Held, len(out.Restore) > 0) + } + for _, node := range out.Restore { + p.restoreAsk(ctx, name, node, fp) } for node, want := range out.Wants { key, err := delivery.WantKey(node, name) @@ -159,3 +168,42 @@ func (p *passer) holdAsk(ctx context.Context, module, fp string) { a := assign.HoldAsk(key, module, fp, time.Now()) p.put(ctx, delivery.AssignerAsksBucket, key, func() ([]byte, error) { return delivery.EncodeAssignerAsk(key, a) }) } + +// heldHold is the hold of fp as written: the node whose health check failed and its words. +func (p *passer) heldHold(module, fp string, fresh *delivery.Held) delivery.Held { + if fresh != nil { + return *fresh + } + key, err := delivery.HeldKey(module, fp) + if err != nil { + return delivery.Held{} + } + h, _ := delivery.DecodeHeld(p.w.read(delivery.AssignerWantsBucket)[key]) + return h +} + +// heldCondition raises, once, the condition for a fingerprint the assigner holds (ADR 0429 §6, ADR 0539 §19): the module, the +// node whose health check failed and the check's words. A module holding data needs the operator's answer to its restore ask. +func (p *passer) heldCondition(ctx context.Context, module, fp string, fresh *delivery.Held, data bool) { + h := p.heldHold(module, fp, fresh) + c := assign.HeldCondition(module, fp, h.Node, h.Words, true, data) + if p.say == nil || p.said[c.Key+".raised"] { + return + } + if err := p.say(c); err != nil { + p.logf("the condition of the held %s could not be said, and is said at the next pass: %v", module, err) + return + } + p.said[c.Key+".raised"] = true +} + +// restoreAsk writes, once, the ask about a module holding data whose held fingerprint failed its health check on node. +func (p *passer) restoreAsk(ctx context.Context, module, node, fp string) { + key, err := delivery.RestoreAskKey(node, module, fp) + if err != nil || p.w.has(delivery.AssignerAsksBucket, key) { + return + } + words := p.heldHold(module, fp, nil).Words + a := assign.RestoreAsk(key, node, module, fp, words, time.Now()) + p.put(ctx, delivery.AssignerAsksBucket, key, func() ([]byte, error) { return delivery.EncodeAssignerAsk(key, a) }) +} diff --git a/modules/mesh-delivery-assigner/internal/assign/asks.go b/modules/mesh-delivery-assigner/internal/assign/asks.go index 52d7cb7..6be4437 100644 --- a/modules/mesh-delivery-assigner/internal/assign/asks.go +++ b/modules/mesh-delivery-assigner/internal/assign/asks.go @@ -2,6 +2,7 @@ package assign import ( "fmt" + "strings" "time" "git.novox.be/novox/mesh-sdk/go/asks" @@ -27,3 +28,52 @@ func HoldAsk(key, module, fp string, now time.Time) delivery.AssignerAsk { About: key, }} } + +// The options of the restore ask (ADR 0537 §3, ADR 0539 §19), and what each does on the node. +const ( + OptionRestore = "restore" + OptionLeaveStopped = "leave-stopped" +) + +// RestoreAsk is the ask about a module holding data whose fingerprint failed its health check on a node (ADR 0429 §6): it is +// not set back by itself, since a set-back would lose the writes made since. words is the check's words. Restore sets the +// node back to the last healthy fingerprint and loses the newer writes; leaving it stopped keeps the data as it is. +func RestoreAsk(key, node, module, fp, words string, now time.Time) delivery.AssignerAsk { + return delivery.AssignerAsk{Ask: asks.Ask{ + ID: delivery.AskingID(key, 0), + Headline: fmt.Sprintf("%s failed its health check on %s", module, node), + Explanation: fmt.Sprintf("%s holds data, and its health check failed on %s at %s: %s. Setting it back to the version that ran healthy "+ + "would lose the writes made since; leaving it stopped keeps the data as it is.", module, node, short(fp), words), + Options: []asks.Option{ + {ID: OptionRestore, Label: "Restore it", Does: "the node is set back to the last healthy version and the newer writes are lost", Level: asks.Acknowledge}, + {ID: OptionLeaveStopped, Label: "Leave it stopped", Does: "the module stays as it is on the node and the data is kept", Level: asks.Acknowledge}, + }, + Who: asks.Operator, + Expires: now.Add(7 * 24 * time.Hour), + OnExpiry: "the module stays as it is on the node", + About: key, + }} +} + +// HeldCondition is the condition the assigner raises when it holds a fingerprint of a module (ADR 0429 §6, ADR 0539 §19), or +// clears (raised false) when the hold is released. It names the module, the node whose health check failed and the check's +// words, under the assigner's own name (ADR 0437 clause 9), keyed by the held fingerprint. needsYou is true for a module holding +// data, whose restore ask waits for the operator. +func HeldCondition(module, fp, node, words string, raised, needsYou bool) asks.Condition { + key := delivery.AssignerModule + ".held." + 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} + } + c := asks.Condition{Key: key, Severity: asks.SeverityWarning, Subject: subject, State: asks.StateRaised, + Summary: fmt.Sprintf("%s is held at %s", module, short(fp)), + Headline: fmt.Sprintf("%s is held at %s", module, short(fp))} + what := fmt.Sprintf("The health check of %s failed on %s at %s: %s. The assigner holds that version and wants no node to move to it.", module, node, short(fp), words) + if needsYou { + c.Needs = "answer whether to restore " + module + " on " + node + c.Explanation = asks.NeedsYou + c.Needs + ". " + what + } else { + c.Explanation = asks.NothingToDo + " " + what + " Nodes that wanted it are set back to the version that ran healthy." + } + return c +} diff --git a/modules/mesh-delivery-assigner/internal/assign/asks_test.go b/modules/mesh-delivery-assigner/internal/assign/asks_test.go index 7de1348..a1bf789 100644 --- a/modules/mesh-delivery-assigner/internal/assign/asks_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/asks_test.go @@ -1,9 +1,11 @@ package assign import ( + "strings" "testing" "time" + "git.novox.be/novox/mesh-sdk/go/asks" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -26,3 +28,48 @@ func TestTheHoldAskIsValid(t *testing.T) { t.Errorf("the view may not answer it: %v", err) } } + +// 1906/D3 (ADR 0539 §19): the ask about a data module held on a node is a valid ask of the router and of mesh-sdk's layout at +// its restore key, of a kind the view may answer; it offers restoring (losing the newer writes) and leaving it stopped. +func TestTheRestoreAskIsValid(t *testing.T) { + key, err := delivery.RestoreAskKey("ace", "mail", fNew) + if err != nil { + t.Fatal(err) + } + now := time.Now() + a := RestoreAsk(key, "ace", "mail", fNew, "connection refused", now) + if err := a.Ask.Check(now); err != nil { + t.Errorf("the router refuses the ask: %v", err) + } + if _, err := delivery.EncodeAssignerAsk(key, a); err != nil { + t.Errorf("the ask cannot be written: %v", err) + } + if _, err := delivery.ViewAnswerKey(key); err != nil { + t.Errorf("the view may not answer it: %v", err) + } + var ids []string + for _, o := range a.Ask.Options { + ids = append(ids, o.ID) + } + if len(ids) != 2 || ids[0] != OptionRestore || ids[1] != OptionLeaveStopped { + t.Errorf("options = %v", ids) + } + if !strings.Contains(a.Ask.Explanation, "connection refused") || !strings.Contains(a.Ask.Explanation, "ace") { + t.Errorf("the ask does not name the node and the check's words: %q", a.Ask.Explanation) + } +} + +// 1906/D3: the condition for a held fingerprint names the module, the node and the check's words, under the assigner's own +// name (ADR 0437 clause 9), and is cleared under the same key. +func TestTheHeldConditionNamesModuleNodeAndWords(t *testing.T) { + c := HeldCondition("mail", fNew, "ace", "connection refused", true, true) + if _, err := asks.EncodeCondition(c); err != nil { + t.Fatalf("the condition is refused: %v", err) + } + if c.State != asks.StateRaised || c.Subject.ID != "mail" || c.Subject.Machine != "ace" || !strings.Contains(c.Explanation, "connection refused") || !strings.Contains(c.Key, "mail") { + t.Errorf("condition = %+v", c) + } + if got := HeldCondition("mail", fNew, "", "", false, false); got.State != asks.StateCleared || got.Key != c.Key { + t.Errorf("cleared = %+v", got) + } +} diff --git a/modules/mesh-delivery-assigner/internal/assign/wants.go b/modules/mesh-delivery-assigner/internal/assign/wants.go index 96e83b4..8615a3a 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants.go @@ -117,6 +117,9 @@ 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 + // 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 // Rules is, by node, the rule holding the node back from m.FP and why. Rules map[string]Holding } @@ -130,7 +133,7 @@ type Holding struct { // Wants is the pass of one module m: for each of its nodes, the want by the six rules, starting from the want written // before (wants, by node); lastHealthy is, by node, the newest fingerprint other than m.FP the node ran healthy. A failed // health check of m.FP on a node holds it: every node wanting m.FP is set back to its last healthy fingerprint, but a -// module holding data is not set back (in release 5 it is asked about from release 6 on: wants are only compared). A +// module holding data is not set back but asked about (Restore). A // node that wants m but is no longer among its nodes gets its want written removed, never deleted (ADR 0408 §4). 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{}} @@ -161,7 +164,8 @@ func Wants(m Module, n Nodes, wants map[string]delivery.Want, lastHealthy map[st switch { case keepsData(m.Manifest): next = before - next.Words = fmt.Sprintf("%s holds data, so it is not set back from %s: the operator is asked from release 6", m.Name, short(m.FP)) + 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) case last == "": next = before next.Words = fmt.Sprintf("%s is held at %s and %s ran no other fingerprint healthy to set it back to", m.Name, short(m.FP), node) diff --git a/modules/mesh-delivery-assigner/internal/assign/wants_test.go b/modules/mesh-delivery-assigner/internal/assign/wants_test.go index c0ff7df..cdae3cd 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants_test.go @@ -1,6 +1,7 @@ package assign import ( + "slices" "strings" "testing" @@ -122,6 +123,20 @@ func TestADataModuleIsNotSetBack(t *testing.T) { if w := out.Wants["ace"]; w.FP != fNew || !strings.Contains(w.Words, "holds data") { t.Errorf("ace's want = %+v, want it left at F, saying it holds data", w) } + // 1906/D3: and the node is to be asked whether to restore it + if !slices.Equal(out.Restore, []string{"ace"}) { + t.Errorf("restore asks = %v, want ace", out.Restore) + } +} + +// 1906/D3: a module without data is set back and asked about nothing. +func TestAModuleWithoutDataIsNotAskedToRestore(t *testing.T) { + m, n := ready() + n.Haves["ace"]["mail"] = delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed} + out := Wants(m, n, map[string]delivery.Want{"ace": {FP: fNew}}, map[string]string{"ace": fOld}) + if len(out.Restore) != 0 { + t.Errorf("restore asks = %v", out.Restore) + } } // 1297/B4, ADR 0408 §4: a node that leaves the module's nodes gets its want written removed, never deleted; a want that diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index f0385fa..13508e5 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -51,6 +51,9 @@ func TestManifest(t *testing.T) { if m.SecretsOwner != "mesh-delivery-assigner" { t.Errorf("secrets-owner = %q", m.SecretsOwner) } + if !reflect.DeepEqual(m.Uses, []string{"operator-channel"}) { + t.Errorf("uses = %v, want the operator channel, which it says its conditions on (1906/D3)", m.Uses) + } if len(m.Claims) != 0 || len(m.Tools) != 0 || len(m.Invokes) != 0 { t.Errorf("the assigner claims, serves or calls something: claims %v, tools %v, invokes %v", m.Claims, m.Tools, m.Invokes) } diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index 0d42866..2d8b303 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -12,6 +12,9 @@ "mesh-delivery-watchman.presence", "node-engine.have" ], + "uses": [ + "operator-channel" + ], "state": [ { "name": "asks", "history": 5 }, { "name": "compositions", "history": 5 },