Watchman: say needs-operator from mesh-provisioner's provisions in waits, so the row the haves could not carry has its facts on the bus (issue 1954 D2)

Pins mesh-sdk 4a96793.
2026-10-29 22:37:39 +01:00
parent 3aec7f9b02
commit 78f03b7422
8 changed files with 107 additions and 10 deletions
@@ -24,6 +24,7 @@ import (

 	"git.novox.be/novox/mesh-sdk/go/buscompose"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/provisions"

 	"mesh-delivery-watchman/internal/socket"
 	"mesh-delivery-watchman/internal/watch"
@@ -297,6 +298,29 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err := w.Rebuild(ask); err != nil {
 		return err
 	}
+	// The seat mesh-provisioner's provisions (ADR 0570 §7): a provision in waits is needs-operator. Read before the findings are
+	// reconciled when the bucket is there; bound once it appears when it is not (mesh-provisioner comes in its own delivery).
+	provisionsWatch := func(kv jetstream.KeyValue) (jetstream.KeyWatcher, error) {
+		return settle(ctx, kv, func(e jetstream.KeyValueEntry) { w.Provision(e.Key(), e.Value(), opOf(e)) })
+	}
+	var provWatch jetstream.KeyWatcher
+	if kv, err := js.KeyValue(ctx, provisions.ProvisionsBucket); err == nil {
+		if provWatch, err = provisionsWatch(kv); err != nil {
+			process.NotWatched(logger.Printf, provisions.ProvisionsBucket, err, "no provision waiting for the operator is said until the watchman starts again")
+		}
+	} else {
+		process.AwaitBucket(ctx, js, conn, provisions.ProvisionsBucket, "no provision waiting for the operator is said",
+			logger.Printf).Ready(func(kv jetstream.KeyValue) {
+			pw, err := provisionsWatch(kv)
+			if err != nil {
+				process.NotWatched(logger.Printf, provisions.ProvisionsBucket, err, "no provision waiting for the operator is said until the watchman starts again")
+				return
+			}
+			go func() {
+				_ = follow(ctx, pw, func(e jetstream.KeyValueEntry) { w.Provision(e.Key(), e.Value(), opOf(e)) })
+			}()
+		})
+	}
 	dead := &deadKeeper{ctx: ctx, js: js, w: w, logf: logger.Printf}
 	w.SetRecount(dead.count)
 	dead.count()
@@ -309,6 +333,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}

 	errs := make(chan error, 8)
+	if provWatch != nil {
+		go func() {
+			errs <- follow(ctx, provWatch, func(e jetstream.KeyValueEntry) { w.Provision(e.Key(), e.Value(), opOf(e)) })
+		}()
+	}
 	if kvs.lease != nil {
 		lw, err := kvs.lease.Watch(ctx, buscompose.ControllerLeaseKey)
 		if err != nil {
@@ -3,7 +3,7 @@ module mesh-delivery-watchman
 go 1.26.0

 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029205750-b6ef01d6c6a7
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029213215-4a967933b368
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )
@@ -1,7 +1,5 @@
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba h1:uXwkcQWgbu/UthizGHFaKPiA4NBVJgqL9q2jJnnGR1o=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029205750-b6ef01d6c6a7 h1:/XEAhD7/8QoMjqI7XiX9HM4HkpKubWheQosqgfe2WrU=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029205750-b6ef01d6c6a7/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029213215-4a967933b368 h1:LaBxsyU3HeTIPKYmvpJuw+Wm2plpf6O7qHO91O+mAi0=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029213215-4a967933b368/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
 github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op h1:1BOWQJweNyvZMlpAHXGLiZQn9S+QXGcz3xh94lC0w6E=
 github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
 github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
@@ -6,6 +6,7 @@ import (
 	"time"

 	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/provisions"
 )

 func haveOf(t *testing.T, phase delivery.Phase, health delivery.Health, detail string) []byte {
@@ -154,6 +155,24 @@ func TestTheWatchmansFindings(t *testing.T) {
 			expire(w, c, signalKey("unhealthy", "web.ace"), UnhealthyUrgentAfter)
 			w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthWaits, "a part waits for the secret token"), Put) // needs-operator is not observed here
 		}, map[string]string{"module.web.ace.unhealthy": Urgent}},
+		{"1954/D2: a provision in waits is its consumer's needs-operator; provided, it ends", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
+			enc := func(state, why string) []byte {
+				b, err := provisions.Encode(provisions.Provision{Provision: "oidc", Consumer: provisions.End{Module: "photos", Node: "ace"},
+					Provider: provisions.End{Module: "keycloak", Node: "novox"}, State: state, Why: why, At: t0})
+				if err != nil {
+					t.Fatal(err)
+				}
+				return b
+			}
+			w.Provision("ace.photos.oidc", enc(provisions.Waits, "the operator gives the client's redirect"), Put)
+			w.Provision("ace.photos.db", enc(provisions.Provided, ""), Put)
+			if f := o.findings["module.photos.ace.needs-operator"]; !strings.Contains(f.Summary, "redirect") || f.Machine != "ace" {
+				t.Fatalf("findings = %v", o.findings)
+			}
+			w.Provision("ace.photos.oidc", enc(provisions.Provided, ""), Put)
+			w.Provision("ace.web.oidc", enc(provisions.Waits, "x"), Put)
+			w.Provision("ace.web.oidc", nil, Delete)
+		}, map[string]string{}},
 		{"WM3: a have removed ends its findings", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
 			w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthFailed, "x"), Put)
 			w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthFailed, "x"), Put)
@@ -9,6 +9,7 @@ import (
 	"time"

 	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/provisions"
 )

 // The rows of the signals table the watchman holds (novox/hq ADR 0564 §4, Limits 3 and 11). Each is a key of the pulse
@@ -458,9 +459,8 @@ func ToolsNode(subject string) string {
 // unhealthy (two writes in a row, urgent after four hours), and where the controller runs (S1's control node). A have
 // removed or deleted ends its part of them.
 //
-// needs-operator is not observed here: the controller excuses a wait only after checking it against the module's manifest
-// and against whether a person gave the secret on that machine (its operator_wait.go), and neither each resource's wait nor
-// what a person gave is on the bus (the have carries the node's words only). The row waits for those facts' home.
+// needs-operator is not observed from the haves: a module's part that waits for the operator is a provision in the state
+// `waits` of the seat mesh-provisioner (Provision, ADR 0570 §7).
 func (w *Watchman) haveFindings(key string, node, module string, h *delivery.Have) {
 	unhealthyKey := FindingKey(ScopeModule, module+"."+node, "unhealthy")
 	if module == ControllerModule {
@@ -511,6 +511,53 @@ func (w *Watchman) haveFindings(key string, node, module string, h *delivery.Hav
 // ControllerModule is the module whose have names a control node.
 const ControllerModule = "mesh-controller"

+// Provision is a change of a key of mesh-provisioner's `provisions` (ADR 0570 §7): a provision in `waits` is a part of its
+// consumer waiting for the operator, said as `module.<consumer>.<node>.needs-operator` naming each waiting provision and
+// why; the finding ends when no provision of that consumer on that node waits. A key deleted, or one that cannot be read,
+// waits no more.
+func (w *Watchman) Provision(key string, value []byte, op Op) {
+	node, consumer, _, _, err := provisions.ParseKey(key)
+	if err != nil {
+		return
+	}
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	id := consumer + "." + node
+	if w.waiting[id] == nil {
+		w.waiting[id] = map[string]string{}
+	}
+	delete(w.waiting[id], key)
+	if op == Put {
+		if p, err := provisions.Decode(value); err != nil {
+			w.say("the provision %s cannot be read, and is taken as not waiting: %v", key, err)
+		} else if p.State == provisions.Waits {
+			w.waiting[id][key] = fmt.Sprintf("%s from %s on %s: %s", p.Provision, p.Provider.Module, p.Provider.Node, p.Why)
+		}
+	}
+	fkey := FindingKey(ScopeModule, id, KindNeedsOperator)
+	if len(w.waiting[id]) == 0 {
+		delete(w.waiting, id)
+		w.unfind(fkey)
+		return
+	}
+	var said []string
+	for _, k := range sortedKeys(w.waiting[id]) {
+		said = append(said, w.waiting[id][k])
+	}
+	w.find(Finding{Scope: ScopeModule, ID: id, Kind: KindNeedsOperator, Machine: node, Severity: Warning, Source: "provisions",
+		Summary: fmt.Sprintf("%s on %s waits for the operator: %s", consumer, node, strings.Join(said, "; ")),
+		Said:    strings.Join(said, "; ")})
+}
+
+func sortedKeys(m map[string]string) []string {
+	out := make([]string, 0, len(m))
+	for k := range m {
+		out = append(out, k)
+	}
+	sort.Strings(out)
+	return out
+}
+
 // ProvReport is a machine's provisioner client's word on one module's files (`prov.applied.<node>.<module>`, ADR 0362 §3).
 type ProvReport struct {
 	Node   string    `json:"node"`
@@ -786,6 +833,7 @@ var SignalRows = []struct{ Row, Signal, Bound, Kind string }{
 	{"S11", "a machine's tool runner heartbeat", "3 × its interval (" + (3 * HeartbeatEvery).String() + " when it says none)", KindToolsSilent},
 	{"S21", "a machine's have failed or its provisioner client's files not placed", ApplyEvery.String(), KindApplyFailing},
 	{"health", "a have unhealthy in two writes in a row; urgent after " + UnhealthyUrgentAfter.String(), "two writes", KindUnhealthy},
+	{"provisions", "a provision of mesh-provisioner in waits (ADR 0570 §7)", "at once", KindNeedsOperator},
 }

 // Signal is one signal's pulse as `signals` shows it.
@@ -84,6 +84,8 @@ type Watchman struct {
 	provFailed  map[string]map[string]ProvReport
 	closures    map[string]map[string]time.Time
 	recount     func()
+	// waiting is, by `<consumer>.<node>`, each of its provisions in waits, with its words.
+	waiting map[string]map[string]string
 	// stepBound is how long a build step may run once a node took the ask; a test shortens it.
 	stepBound time.Duration
 }
@@ -94,7 +96,7 @@ func New(out Out, now func() time.Time) *Watchman {
 		pulses: map[string]*pulse{}, stuck: map[string]delivery.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, bound: delivery.PulseMoveBound, stepBound: StepBound,
 		findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, away: map[string]bool{},
 		haveState: map[string]delivery.Have{}, unhealthy: map[string]int{}, advised: map[string]int{}, advisedSince: map[string]time.Time{},
-		control: map[string]bool{}, failedSince: map[string]time.Time{}, provFailed: map[string]map[string]ProvReport{}, closures: map[string]map[string]time.Time{}}
+		control: map[string]bool{}, failedSince: map[string]time.Time{}, provFailed: map[string]map[string]ProvReport{}, closures: map[string]map[string]time.Time{}, waiting: map[string]map[string]string{}}
 }

 // Logf sets where a write that failed is said; it is silent until set.
@@ -91,7 +91,7 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
 	reads := slices.Clone(m.Reads)
 	slices.Sort(reads)
 	if !reflect.DeepEqual(reads, []string{"mesh-build-go.steps", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-controller.lease",
-		"mesh-delivery-assigner.wants", "node-engine.have"}) {
+		"mesh-delivery-assigner.wants", "mesh-provisioner.provisions", "node-engine.have"}) {
 		t.Errorf("reads = %v", reads)
 	}
 	// 1952/WM1: the watchman holds the seat mesh-watchdogs at mesh scope, its first seat (novox/hq ADR 0564 §1, §4).
@@ -21,6 +21,7 @@
    "mesh-build-node.steps",
    "mesh-controller.lease",
    "mesh-delivery-assigner.wants",
+    "mesh-provisioner.provisions",
    "node-engine.have"
  ],
  "tools": [