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)

2026-10-24 16:26:20 +02:00
parent f048e2b443
commit 204a59dab4
9 changed files with 273 additions and 5 deletions
@@ -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")
+	}
+}
@@ -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
+	}
+}
@@ -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) })
+}
@@ -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
+}
@@ -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)
+	}
+}
@@ -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)
@@ -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
@@ -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)
 	}
@@ -12,6 +12,9 @@
    "mesh-delivery-watchman.presence",
    "node-engine.have"
  ],
+  "uses": [
+    "operator-channel"
+  ],
  "state": [
    { "name": "asks", "history": 5 },
    { "name": "compositions", "history": 5 },