The watchman mirrors each want of the assigner as a pulse owed by the node's agent until the have names its fingerprint (issue 1906 D6)

modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go

@@ -9,6 +9,7 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
 
@@ -99,6 +100,8 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
 		{Bucket: delivery.StuckBucket, History: 20},
 		{Bucket: delivery.PulseBucket, LimitMarkerTTL: time.Second},
 		{Bucket: delivery.StepsBucket("mesh-build-go"), History: 5},
+		{Bucket: delivery.AssignerWantsBucket, History: 20},
+		{Bucket: delivery.HaveBucket, History: 20},
 	} {
 		kv, err := js.CreateKeyValue(ctx, c)
 		if err != nil {
@@ -265,6 +268,8 @@ func TestAnAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909
 		{Bucket: delivery.StuckBucket, History: 20},
 		{Bucket: delivery.PulseBucket, LimitMarkerTTL: time.Second},
 		{Bucket: delivery.StepsBucket("mesh-build-go"), History: 5},
+		{Bucket: delivery.AssignerWantsBucket, History: 20},
+		{Bucket: delivery.HaveBucket, History: 20},
 	} {
 		kv, err := js.CreateKeyValue(ctx, c)
 		if err != nil {
@@ -302,3 +307,134 @@ func TestAnAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909
 	}
 	t.Fatalf("no pulse for the ask running before the start (%s)", pk)
 }
+
+// TestAWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt (1906/D6, ADR 0539 §19): started whole, the watchman mirrors a want of
+// the assigner as a pulse owed by the node's agent, one that stood before its start as well, and the node-engine's have that
+// names the want's fingerprint deletes the pulse and writes no stuck key.
+func TestAWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
+	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "w")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { os.RemoveAll(dir) })
+	conf := filepath.Join(dir, "bus.conf")
+	if err := os.WriteFile(conf, []byte(fmt.Sprintf(busConf, filepath.Join(dir, "js"))), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	opts, err := server.ProcessConfigFile(conf)
+	if err != nil {
+		t.Fatal(err)
+	}
+	opts.NoLog, opts.NoSigs = true, true
+	srv, err := server.NewServer(opts)
+	if err != nil {
+		t.Fatal(err)
+	}
+	go srv.Start()
+	if !srv.ReadyForConnections(10 * time.Second) {
+		t.Fatal("the bus did not come up")
+	}
+	t.Cleanup(srv.Shutdown)
+	ctl, err := nats.Connect(srv.ClientURL(), nats.UserInfo("ctl", "pw"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(ctl.Close)
+	js, err := jetstream.New(ctl)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+	kvs := map[string]jetstream.KeyValue{}
+	for _, c := range []jetstream.KeyValueConfig{
+		{Bucket: delivery.PresenceBucket, History: 20},
+		{Bucket: delivery.StuckBucket, History: 20},
+		{Bucket: delivery.PulseBucket, LimitMarkerTTL: time.Second},
+		{Bucket: delivery.StepsBucket("mesh-build-go"), History: 5},
+		{Bucket: delivery.AssignerWantsBucket, History: 20},
+		{Bucket: delivery.HaveBucket, History: 20},
+	} {
+		kv, err := js.CreateKeyValue(ctx, c)
+		if err != nil {
+			t.Fatal(err)
+		}
+		kvs[c.Bucket] = kv
+	}
+	fpNew := delivery.SourcePrefix2 + strings.Repeat("b", 64)
+	fpOld := delivery.SourcePrefix2 + strings.Repeat("a", 64)
+	putWant := func(node, module, fp string) {
+		t.Helper()
+		body, err := delivery.EncodeWant(delivery.Want{FP: fp})
+		if err != nil {
+			t.Fatal(err)
+		}
+		key, _ := delivery.WantKey(node, module)
+		if _, err := kvs[delivery.AssignerWantsBucket].Put(ctx, key, body); err != nil {
+			t.Fatal(err)
+		}
+	}
+	putHave := func(node, module, fp string) {
+		t.Helper()
+		body, err := delivery.EncodeHave(delivery.Have{FP: fp, Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
+		if err != nil {
+			t.Fatal(err)
+		}
+		key, _ := delivery.HaveKey(node, module)
+		if _, err := kvs[delivery.HaveBucket].Put(ctx, key, body); err != nil {
+			t.Fatal(err)
+		}
+	}
+	putHave("n2", "web", fpOld)
+	putWant("n2", "web", fpNew) // stood before the start
+	broker := filepath.Join(dir, "broker")
+	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-watchman", User: "n1.mesh-delivery-watchman", Password: "pw"})
+	if err := os.WriteFile(broker, cred, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	env := map[string]string{"MESH_BROKER_FILE": broker}
+	runCtx, stop := context.WithCancel(ctx)
+	done := make(chan error, 1)
+	go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
+	t.Cleanup(func() { stop(); <-done })
+	pulse := func(node, module string) string {
+		wk, _ := delivery.WantKey(node, module)
+		pk, err := delivery.PulseKey(delivery.AssignerWantsBucket, wk)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return pk
+	}
+	until := func(what string, ok func() bool) {
+		t.Helper()
+		for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+			if ok() {
+				return
+			}
+		}
+		t.Fatalf("never saw: %s", what)
+	}
+	has := func(key string) bool { _, err := kvs[delivery.PulseBucket].Get(ctx, key); return err == nil }
+	until("a pulse for the want that stood before the start", func() bool { return has(pulse("n2", "web")) })
+	putWant("n2", "mail", fpNew)
+	until("a pulse for a want written after the start", func() bool { return has(pulse("n2", "mail")) })
+	e, err := kvs[delivery.PulseBucket].Get(ctx, pulse("n2", "web"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if p, err := delivery.DecodePulse(e.Value()); err != nil || !strings.Contains(p.Owes, "n2") {
+		t.Errorf("the pulse = %+v, %v; want it owed by the agent of n2", p, err)
+	}
+	putHave("n2", "web", fpNew)
+	until("the pulse of the want its have names deleted", func() bool { return !has(pulse("n2", "web")) })
+	if !has(pulse("n2", "mail")) {
+		t.Errorf("the pulse of another want ended with web's have")
+	}
+	for _, m := range []string{"web"} {
+		wk, _ := delivery.WantKey("n2", m)
+		sk, _ := delivery.StuckKey(delivery.AssignerWantsBucket, wk)
+		if _, err := kvs[delivery.StuckBucket].Get(ctx, sk); err == nil {
+			t.Errorf("a stuck key was written for the want of %s that was applied", m)
+		}
+	}
+}

modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go

@@ -229,11 +229,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		}
 		return kv, nil
 	}
-	var kvs struct{ presence, pulse, stuck jetstream.KeyValue }
+	var kvs struct{ presence, pulse, stuck, wants, haves jetstream.KeyValue }
 	for _, b := range []struct {
 		name string
 		into *jetstream.KeyValue
-	}{{delivery.PresenceBucket, &kvs.presence}, {delivery.PulseBucket, &kvs.pulse}, {delivery.StuckBucket, &kvs.stuck}} {
+	}{{delivery.PresenceBucket, &kvs.presence}, {delivery.PulseBucket, &kvs.pulse}, {delivery.StuckBucket, &kvs.stuck},
+		{delivery.AssignerWantsBucket, &kvs.wants}, {delivery.HaveBucket, &kvs.haves}} {
 		if *b.into, err = bucket(b.name); err != nil {
 			return err
 		}
@@ -280,6 +281,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err != nil {
 		return err
 	}
+	// What the node agents run, then the assigner's wants (issue 1906, ADR 0539 §19): a want is a pulse owed by its node's agent
+	// until the have names its fingerprint, so the haves are read first and a want already applied makes no pulse.
+	haveSeen, err := settle(ctx, kvs.haves, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) })
+	if err != nil {
+		return err
+	}
+	wantSeen, err := settle(ctx, kvs.wants, func(e jetstream.KeyValueEntry) { w.Want(e.Key(), e.Value(), opOf(e), e.Created()) })
+	if err != nil {
+		return err
+	}
 	// The build seats' steps: a step key `<node>.<ask id>` written is the ask taken (issue 1887 A8b). A seat whose build module
 	// keeps no steps bucket (yet) is said and left; the asks of its stream are mirrored all the same.
 	taken := map[string]jetstream.KeyWatcher{}
@@ -337,6 +348,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	go func() {
 		errs <- follow(ctx, pulses, func(e jetstream.KeyValueEntry) { w.Pulse(e.Key(), opOf(e), e.Created()) })
 	}()
+	go func() {
+		errs <- follow(ctx, haveSeen, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) })
+	}()
+	go func() {
+		errs <- follow(ctx, wantSeen, func(e jetstream.KeyValueEntry) { w.Want(e.Key(), e.Value(), opOf(e), e.Created()) })
+	}()
 	for lang, sw := range taken {
 		lang, sw := lang, sw
 		go func() { errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) { stepTaken(w, lang, e) }) }()

modules/mesh-delivery-watchman/internal/watch/mirror.go

@@ -1,6 +1,9 @@
 package watch
 
 import (
+	"strings"
+	"time"
+
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 )
 
@@ -52,3 +55,96 @@ func (w *Watchman) Step(language, key string, value []byte, op Op) {
 	}
 	w.mirrorWithin(w.stepBound, bucket, id, describe("the ask %s to end on %s", id, node), node, since, 0)
 }
+
+// A want is a move too (novox/hq ADR 0539 §19, issue 1906): the assigner writes `want.<node>.<module>` in its wants bucket,
+// and the node's agent owes the node-engine's have naming the want's fingerprint. The pulse is the pulse of that bucket under
+// the want's key, owed by the node's agent, and it begins when the assigner wrote the want. The have naming the fingerprint
+// ends it; a want that is removed, or a key deleted, ends it as well. Nothing renews it: the bound is how long a node may
+// take to apply a want, the same as a build step's (StepBound). Keys of that bucket that are not a want of a module (a hold,
+// and the node's own key) are left.
+
+// Want is a change of a key of the assigner's wants bucket, read from the watch (at the start the keys that stand, then each
+// change), at the time the entry was written.
+func (w *Watchman) Want(key string, value []byte, op Op, at time.Time) {
+	node, module, err := delivery.ParseWantKey(key)
+	if err != nil {
+		return
+	}
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	switch op {
+	case Expired:
+		return
+	case Delete:
+		delete(w.wants, key)
+		w.moved(delivery.AssignerWantsBucket, key)
+		return
+	}
+	want, err := delivery.DecodeWant(value)
+	if err != nil {
+		w.say("the want %s cannot be read, and is left: %v", key, err)
+		return
+	}
+	if want.Removed {
+		delete(w.wants, key)
+		w.moved(delivery.AssignerWantsBucket, key)
+		return
+	}
+	w.wants[key] = want.FP
+	if w.have(node, module) == want.FP {
+		w.moved(delivery.AssignerWantsBucket, key)
+		return
+	}
+	if at.IsZero() {
+		at = w.now()
+	}
+	w.mirrorWithin(w.stepBound, delivery.AssignerWantsBucket, key,
+		describe("the node agent of %s to run %s at %s", node, module, shortFP(want.FP)), want.FP, at, 0)
+}
+
+// Have is a change of a key of the node-engine's have bucket: the fingerprint a node runs of a module. One that names the
+// fingerprint of the node's want for the module ends the want's pulse. A node's own key and a key deleted are left.
+func (w *Watchman) Have(key string, value []byte, op Op) {
+	node, module, isNode, err := delivery.ParseHaveKey(key)
+	if err != nil || isNode {
+		return
+	}
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	switch op {
+	case Expired:
+		return
+	case Delete:
+		delete(w.haves, key)
+		return
+	}
+	have, err := delivery.DecodeHave(value)
+	if err != nil {
+		w.say("the have %s cannot be read, and is left: %v", key, err)
+		return
+	}
+	w.haves[key] = have.FP
+	wk, err := delivery.WantKey(node, module)
+	if err != nil {
+		return
+	}
+	if fp, ok := w.wants[wk]; ok && fp == have.FP {
+		w.moved(delivery.AssignerWantsBucket, wk)
+	}
+}
+
+func (w *Watchman) have(node, module string) string {
+	key, err := delivery.HaveKey(node, module)
+	if err != nil {
+		return ""
+	}
+	return w.haves[key]
+}
+
+func shortFP(fp string) string {
+	hex := strings.TrimPrefix(fp, delivery.SourcePrefix2)
+	if len(hex) > 12 {
+		hex = hex[:12]
+	}
+	return hex
+}

modules/mesh-delivery-watchman/internal/watch/watch.go

@@ -57,6 +57,8 @@ type Watchman struct {
 	presence map[string]presence
 	pulses   map[string]*pulse // live pulses, by pulse key
 	stuck    map[string]delivery.Stuck
+	wants    map[string]string // the fingerprint of each want that stands, by want key
+	haves    map[string]string // the fingerprint each node runs of a module, by have key
 	bound    time.Duration
 	// stepBound is how long a build step may run once a node took the ask; a test shortens it.
 	stepBound time.Duration
@@ -65,7 +67,7 @@ type Watchman struct {
 // New is a watchman writing to out, reading the time from now.
 func New(out Out, now func() time.Time) *Watchman {
 	return &Watchman{out: out, now: now, counts: map[string]int{}, presence: map[string]presence{},
-		pulses: map[string]*pulse{}, stuck: map[string]delivery.Stuck{}, bound: delivery.PulseMoveBound, stepBound: StepBound}
+		pulses: map[string]*pulse{}, stuck: map[string]delivery.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, bound: delivery.PulseMoveBound, stepBound: StepBound}
 }
 
 // Logf sets where a write that failed is said; it is silent until set.

modules/mesh-delivery-watchman/internal/watch/watch_test.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"errors"
 	"slices"
+	"strings"
 	"testing"
 	"time"
 
@@ -531,3 +532,88 @@ func TestARunningStepPastItsBoundIsStuck(t *testing.T) { // 1908
 		t.Errorf("stuck %v pulses %v", o2.stuck, o2.pulses)
 	}
 }
+
+// 1906/D6 (ADR 0539 §19): a want is a pulse owed by the node's agent from when the assigner wrote it, until the have names
+// its fingerprint.
+func wantValue(t *testing.T, fp string, removed bool) []byte {
+	t.Helper()
+	w := delivery.Want{FP: fp, Removed: removed}
+	b, err := delivery.EncodeWant(w)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return b
+}
+
+func haveValue(t *testing.T, fp string) []byte {
+	t.Helper()
+	b, err := delivery.EncodeHave(delivery.Have{FP: fp, Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return b
+}
+
+var (
+	fpOld = delivery.SourcePrefix2 + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+	fpNew = delivery.SourcePrefix2 + "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+)
+
+func TestAWantIsAPulseOwedByTheNodeAndTheHaveNamingItEndsIt(t *testing.T) {
+	w, o, c := setup()
+	c.t = t0.Add(10 * time.Second)
+	w.Want("want.ace.web", wantValue(t, fpNew, false), Put, t0)
+	pk := pulseKeyOf(t, delivery.AssignerWantsBucket, "want.ace.web")
+	p, ok := o.pulses[pk]
+	if !ok || p.Key != "want.ace.web" || !strings.Contains(p.Owes, "ace") || !strings.Contains(p.Owes, "web") {
+		t.Fatalf("pulses = %v", o.pulses)
+	}
+	w.Have("ace.web", haveValue(t, fpOld), Put) // the node still runs the old one
+	if _, ok := o.pulses[pk]; !ok {
+		t.Fatalf("a have of another fingerprint ended the pulse")
+	}
+	w.Have("ace.web", haveValue(t, fpNew), Put)
+	if _, ok := o.pulses[pk]; ok {
+		t.Errorf("the have naming the want's fingerprint left the pulse standing: %v", o.pulses)
+	}
+	if len(o.stuck) != 0 {
+		t.Errorf("stuck = %v", o.stuck)
+	}
+}
+
+func TestAHaveThatCameBeforeTheWantEndsItAtOnceAndARemovedWantHasNoPulse(t *testing.T) {
+	w, o, c := setup()
+	c.t = t0.Add(time.Second)
+	w.Have("ace.web", haveValue(t, fpNew), Put)
+	w.Want("want.ace.web", wantValue(t, fpNew, false), Put, t0)
+	w.Want("want.ace.mail", wantValue(t, "", true), Put, t0)
+	if len(o.pulses) != 0 {
+		t.Errorf("pulses = %v", o.pulses)
+	}
+	// the want moving on to a new fingerprint makes a new pulse; the key `want.ace.node` and a held key are left
+	w.Want("want.ace.web", wantValue(t, fpOld, false), Put, t0.Add(time.Second))
+	w.Want("want.ace.node", []byte(`{}`), Put, t0)
+	w.Want("held.web.bbbb", []byte(`{}`), Put, t0)
+	if len(o.pulses) != 1 {
+		t.Errorf("pulses = %v, want the web want's only", o.pulses)
+	}
+}
+
+func TestAWantNotAppliedByItsBoundIsStuckAndAKeyDeletedEndsIt(t *testing.T) {
+	o, c := newOut(), &clock{t: t0}
+	w := New(o, c.now)
+	w.Want("want.ace.web", wantValue(t, fpNew, false), Put, t0)
+	pk := pulseKeyOf(t, delivery.AssignerWantsBucket, "want.ace.web")
+	c.t = t0.Add(31 * time.Minute)
+	w.Pulse(pk, Expired, c.t)
+	if s, ok := o.stuck[stuckKeyOf(t, delivery.AssignerWantsBucket, "want.ace.web")]; !ok || s.Moved || !s.Since.Equal(t0) {
+		t.Fatalf("stuck = %v", o.stuck)
+	}
+	w2, o2, c2 := setup()
+	c2.t = t0.Add(time.Second)
+	w2.Want("want.ace.web", wantValue(t, fpNew, false), Put, t0)
+	w2.Want("want.ace.web", nil, Delete, t0)
+	if len(o2.pulses) != 0 {
+		t.Errorf("a deleted want left its pulse: %v", o2.pulses)
+	}
+}

modules/mesh-delivery-watchman/manifest_test.go

@@ -81,7 +81,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"}) {
+	if !reflect.DeepEqual(reads, []string{"mesh-build-go.steps", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-delivery-assigner.wants", "node-engine.have"}) {
 		t.Errorf("reads = %v", reads)
 	}
 	for _, k := range []string{"emits", "context", "items", "consumes", "provides", "claims"} {

modules/mesh-delivery-watchman/module.json

@@ -5,7 +5,9 @@
   "reads": [
     "mesh-build-go.steps",
     "mesh-build-image.steps",
-    "mesh-build-node.steps"
+    "mesh-build-node.steps",
+    "mesh-delivery-assigner.wants",
+    "node-engine.have"
   ],
   "state": [
     { "name": "presence", "history": 20 },