The watchman's logic: presence from the bus's events, pulses with a time to live, stuck keys on expiry (issue 1135)

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

@@ -0,0 +1,90 @@
+package watch
+
+import (
+	"encoding/json"
+	"slices"
+	"strings"
+	"time"
+)
+
+// reservedBuildModules are the first parts of the builds bucket's keys that are not a module's.
+var reservedBuildModules = []string{"manifest", "req", "id", "layout"}
+
+// buildRecord is the part of the controller's build record (mesh-controller internal/chain Build) the watchman reads:
+// a record in state asked is a request that no outcome has answered.
+type buildRecord struct {
+	ID     string    `json:"id"`
+	State  string    `json:"state"`
+	Module string    `json:"module"`
+	Asked  time.Time `json:"asked"`
+	At     time.Time `json:"at"`
+}
+
+// Build is a change in the builds bucket. A record `<module>.build.<id>` in state asked, with no outcome, is mirrored;
+// when its outcome is written it has moved. No other key of the bucket is mirrored.
+func (w *Watchman) Build(key string, value []byte, at time.Time, op Op) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	parts := strings.Split(key, ".")
+	if len(parts) != 3 || parts[1] != "build" || slices.Contains(reservedBuildModules, parts[0]) || op == Expired {
+		return
+	}
+	var rec buildRecord
+	if op == Delete || json.Unmarshal(value, &rec) != nil || rec.State != "asked" {
+		w.moved(BuildsBucket, key)
+		return
+	}
+	module := rec.Module
+	if module == "" {
+		module = parts[0]
+	}
+	since := rec.Asked
+	if since.IsZero() {
+		since = rec.At
+	}
+	if since.IsZero() {
+		since = at
+	}
+	w.mirror(BuildsBucket, key, describe("the build of %s to be answered", module), rec.ID, since, 0)
+}
+
+// checkRecord is the part of a running check's key the watchman reads: the head being checked and when the check
+// started (mesh-controller issue 1155).
+type checkRecord struct {
+	Head    string    `json:"head"`
+	Started time.Time `json:"started"`
+}
+
+// Check is a change in the checks-running bucket. A key is a check running, mirrored from its start; a builder's
+// renewal of the key changes nothing, a new head or start is a new check, and a deleted key is the check ended. An
+// aged-out key is the bucket's own expiry of a builder that died: the pulse is left to run out.
+func (w *Watchman) Check(key string, value []byte, at time.Time, op Op) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	switch op {
+	case Expired:
+		return
+	case Delete:
+		w.moved(ChecksBucket, key)
+		return
+	}
+	var rec checkRecord
+	_ = json.Unmarshal(value, &rec)
+	since := rec.Started
+	if since.IsZero() {
+		since = at
+		if cur := w.pulseOf(ChecksBucket, key); cur != nil && cur.head == rec.Head && !cur.seeded {
+			since = cur.since
+		}
+	}
+	w.mirror(ChecksBucket, key, describe("the check of %s to end", key), rec.Head, since, 0)
+}
+
+func (w *Watchman) pulseOf(bucket, key string) *pulse {
+	for _, p := range w.pulses {
+		if p.bucket == bucket && p.key == key {
+			return p
+		}
+	}
+	return nil
+}

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

@@ -0,0 +1,185 @@
+package watch
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// partPrefix is what the watchman keeps presence of among the module users: the parts of delivery.
+const partPrefix = "mesh-delivery-"
+
+// presenceOf is the presence key of a bus user, and what a wait for its return is called: `node.<node>` is the
+// node-engine of that node, `<node>.<module>` a part's instance for a module of delivery, and any other user has none.
+func presenceOf(user string) (key, waited string, ok bool) {
+	first, rest, found := strings.Cut(user, ".")
+	if !found || first == "" || rest == "" {
+		return "", "", false
+	}
+	if first == "node" {
+		k, err := delivery.PresenceNodeKey(rest)
+		if err != nil {
+			return "", "", false
+		}
+		return k, rest + " to come back online", true
+	}
+	if !strings.HasPrefix(rest, partPrefix) {
+		return "", "", false
+	}
+	k, err := delivery.PresencePartKey(rest, first)
+	if err != nil {
+		return "", "", false
+	}
+	return k, rest + " on " + first + " to come back online", true
+}
+
+// waitedOf is the words a presence key's return is waited in.
+func waitedOf(key string) string {
+	node, part, instance, err := delivery.ParsePresenceKey(key)
+	switch {
+	case err != nil:
+		return key + " to come back online"
+	case part != "":
+		return part + " on " + instance + " to come back online"
+	default:
+		return node + " to come back online"
+	}
+}
+
+// Connect is a connection of the user opened.
+func (w *Watchman) Connect(user string) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if key, _, ok := presenceOf(user); ok {
+		w.counts[key]++
+		w.settle(key)
+	}
+}
+
+// Disconnect is a connection of the user closed; the user is offline when none is left.
+func (w *Watchman) Disconnect(user string) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if key, _, ok := presenceOf(user); ok {
+		if w.counts[key] > 0 {
+			w.counts[key]--
+		}
+		w.settle(key)
+	}
+}
+
+// settle makes the presence of key what its connections say, writing only when that differs from what it says now.
+func (w *Watchman) settle(key string) {
+	online := w.counts[key] > 0
+	cur, known := w.presence[key]
+	if known && cur.online == online {
+		if !online {
+			w.ensureOffline(key)
+		}
+		return
+	}
+	if err := w.out.PutPresence(key, online); err != nil {
+		w.say("cannot write presence %s: %v", key, err)
+		return
+	}
+	w.presence[key] = presence{online: online, since: w.now()}
+	if online {
+		w.moved(delivery.PresenceBucket, key)
+		return
+	}
+	w.ensureOffline(key)
+}
+
+// ensureOffline mirrors the return of a node or part that is offline.
+func (w *Watchman) ensureOffline(key string) {
+	p := w.presence[key]
+	w.mirror(delivery.PresenceBucket, key, waitedOf(key), "", p.since, 0)
+}
+
+// SeedPresence tells the watchman a presence key as it stands in its bucket at start, with the time it was written.
+func (w *Watchman) SeedPresence(key string, online bool, at time.Time) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.presence[key] = presence{online: online, since: at}
+}
+
+// Connections replaces what the watchman knows of the open connections with the users listed: every user listed is
+// online, every presence it knew online and that is not listed is offline now.
+func (w *Watchman) Connections(users []string) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	counts := map[string]int{}
+	for _, u := range users {
+		if key, _, ok := presenceOf(u); ok {
+			counts[key]++
+		}
+	}
+	w.counts = counts
+	keys := map[string]bool{}
+	for k := range counts {
+		keys[k] = true
+	}
+	for k := range w.presence {
+		keys[k] = true
+	}
+	for k := range keys {
+		w.settle(k)
+	}
+}
+
+// ParseEvent is the user of a connect or disconnect event of the bus (the server's ConnectEventMsg and
+// DisconnectEventMsg: client.user). An event that names no user gives an empty one.
+func ParseEvent(data []byte) (user string, err error) {
+	var ev struct {
+		Client struct {
+			User string `json:"user"`
+		} `json:"client"`
+	}
+	if err := json.Unmarshal(data, &ev); err != nil {
+		return "", fmt.Errorf("the bus event is not JSON: %w", err)
+	}
+	return ev.Client.User, nil
+}
+
+// ConnzRequest is the body of the one request of the open connections: without auth the server leaves the user names
+// out (nats-server monitor.go ConnzOptions.Username).
+func ConnzRequest() []byte { return []byte(`{"auth":true}`) }
+
+// Rebuild asks for the open connections once, with ask, and takes presence from the answer. The caller makes it at
+// start and at each reconnect.
+func (w *Watchman) Rebuild(ask func(body []byte) ([]byte, error)) error {
+	reply, err := ask(ConnzRequest())
+	if err != nil {
+		return fmt.Errorf("the open connections cannot be asked: %w", err)
+	}
+	var answer struct {
+		Data struct {
+			Connections []struct {
+				User string `json:"authorized_user"`
+			} `json:"connections"`
+			Total int `json:"total"`
+		} `json:"data"`
+		Error *struct {
+			Description string `json:"description"`
+		} `json:"error"`
+	}
+	if err := json.Unmarshal(reply, &answer); err != nil {
+		return fmt.Errorf("the answer about the open connections is not JSON: %w", err)
+	}
+	if answer.Error != nil {
+		return errors.New("the bus refused the open connections: " + answer.Error.Description)
+	}
+	users := make([]string, 0, len(answer.Data.Connections))
+	for _, c := range answer.Data.Connections {
+		users = append(users, c.User)
+	}
+	if answer.Data.Total > len(users) {
+		w.say("the bus lists %d of %d connections", len(users), answer.Data.Total)
+	}
+	w.Connections(users)
+	return nil
+}

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

@@ -0,0 +1,240 @@
+// Package watch is the watchman's logic (novox/hq ADR 0433, ADR 0431, issue 1135): it turns the bus's connect and
+// disconnect events, the connections request, the builds bucket and the checks-running bucket into presence, pulse and
+// stuck keys. It holds no timer: every reaction is an event, a watch entry or a request the caller makes.
+package watch
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// Op is what happened to a watched key.
+type Op int
+
+const (
+	// Put is a key written.
+	Put Op = iota
+	// Delete is a key deleted by a writer (the watchman's own delete, or a builder ending a check).
+	Delete
+	// Expired is the server's marker for a key whose time to live ran out.
+	Expired
+)
+
+// The buckets the watchman mirrors.
+const (
+	BuildsBucket = "mesh-controller_builds"
+	ChecksBucket = "mesh-controller_checks-running"
+)
+
+// Out is where the watchman writes. Every call is a write the caller makes on the bus.
+type Out interface {
+	PutPresence(key string, online bool) error
+	// PutPulse writes the pulse key with a time to live, replacing a live one.
+	PutPulse(key string, value []byte, ttl time.Duration) error
+	// DeletePulse deletes the pulse key (a delete, not an expiry).
+	DeletePulse(key string) error
+	PutStuck(key string, value []byte) error
+}
+
+type presence struct {
+	online bool
+	since  time.Time
+}
+
+type pulse struct {
+	bucket, key, waited string
+	since               time.Time
+	head                string
+	token               string
+	seeded              bool
+}
+
+// Watchman is the state the watchman keeps in memory; it is rebuilt from the buckets at every start. Its methods are
+// safe to call from several watches at once.
+type Watchman struct {
+	mu       sync.Mutex
+	out      Out
+	now      func() time.Time
+	logf     func(format string, args ...any)
+	counts   map[string]int // open connections per presence key
+	presence map[string]presence
+	pulses   map[string]*pulse // live pulses, by pulse key
+	stuck    map[string]delivery.Stuck
+	bound    time.Duration
+}
+
+// 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}
+}
+
+// Logf sets where a write that failed is said; it is silent until set.
+func (w *Watchman) Logf(f func(format string, args ...any)) { w.logf = f }
+
+func (w *Watchman) say(format string, args ...any) {
+	if w.logf != nil {
+		w.logf(format, args...)
+	}
+}
+
+// mirror makes the pulse of the move (bucket, key) that began at since, unless it stands already for the same move.
+// A move that is already as old as the bound is stuck at once and has no pulse.
+func (w *Watchman) mirror(bucket, key, waited, head string, since time.Time, revision uint64) {
+	pk, err := delivery.PulseKey(bucket, key)
+	if err != nil {
+		w.say("no pulse for %s %s: %v", bucket, key, err)
+		return
+	}
+	token := head + "|" + since.UTC().Format(time.RFC3339Nano)
+	cur, tracked := w.pulses[pk]
+	if tracked && !cur.seeded && cur.token == token {
+		return
+	}
+	ttl, stalled := w.ttl(w.now().Sub(since))
+	if stalled {
+		if tracked {
+			w.deletePulse(pk)
+		}
+		w.writeStuck(bucket, key, waited, since)
+		return
+	}
+	value, err := delivery.EncodePulse(delivery.Pulse{Bucket: bucket, Key: key, Revision: revision, Owes: waited})
+	if err != nil {
+		w.say("no pulse for %s %s: %v", bucket, key, err)
+		return
+	}
+	if err := w.out.PutPulse(pk, value, ttl); err != nil {
+		w.say("cannot write the pulse %s: %v", pk, err)
+		return
+	}
+	w.pulses[pk] = &pulse{bucket: bucket, key: key, waited: waited, since: since, head: head, token: token}
+}
+
+func (w *Watchman) deletePulse(pk string) {
+	delete(w.pulses, pk)
+	if err := w.out.DeletePulse(pk); err != nil {
+		w.say("cannot delete the pulse %s: %v", pk, err)
+	}
+}
+
+// moved is the mirrored key moving: its pulse is deleted, so it never expires, and a stuck key it left is written as
+// moved (never deleted).
+func (w *Watchman) moved(bucket, key string) {
+	pk, err := delivery.PulseKey(bucket, key)
+	if err != nil {
+		return
+	}
+	if _, ok := w.pulses[pk]; ok {
+		w.deletePulse(pk)
+	}
+	sk, err := delivery.StuckKey(bucket, key)
+	if err != nil {
+		return
+	}
+	s, ok := w.stuck[sk]
+	if !ok || s.Moved {
+		return
+	}
+	s.Moved = true
+	w.putStuck(sk, s)
+}
+
+func (w *Watchman) writeStuck(bucket, key, waited string, since time.Time) {
+	sk, err := delivery.StuckKey(bucket, key)
+	if err != nil {
+		w.say("no stuck key for %s %s: %v", bucket, key, err)
+		return
+	}
+	if cur, ok := w.stuck[sk]; ok && cur.Since.Equal(since) {
+		return
+	}
+	w.putStuck(sk, delivery.Stuck{Bucket: bucket, Key: key, Waited: waited, Since: since})
+}
+
+func (w *Watchman) putStuck(sk string, s delivery.Stuck) {
+	value, err := delivery.EncodeStuck(s)
+	if err != nil {
+		w.say("cannot encode the stuck key %s: %v", sk, err)
+		return
+	}
+	if err := w.out.PutStuck(sk, value); err != nil {
+		w.say("cannot write the stuck key %s: %v", sk, err)
+		return
+	}
+	s.V = delivery.Version
+	w.stuck[sk] = s
+}
+
+// Pulse is a change heard on the pulse bucket. Only the server's expiry marker of a pulse the watchman made writes a
+// stuck key: the watchman's own delete, and a key it never made, write nothing.
+func (w *Watchman) Pulse(key string, op Op, at time.Time) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if op != Expired {
+		return
+	}
+	p, ok := w.pulses[key]
+	if !ok || p.seeded {
+		return
+	}
+	delete(w.pulses, key)
+	w.writeStuck(p.bucket, p.key, p.waited, p.since)
+}
+
+// SeedStuck tells the watchman a stuck key that stands in its bucket.
+func (w *Watchman) SeedStuck(key string, value []byte) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if s, err := delivery.DecodeStuck(value); err == nil {
+		w.stuck[key] = s
+	}
+}
+
+// SeedPulse tells the watchman a pulse key that stands in its bucket at its start; Reconcile deletes it unless a
+// mirrored key made it again since.
+func (w *Watchman) SeedPulse(key string) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	bucket, k, err := delivery.ParsePulseKey(key)
+	if err != nil {
+		return
+	}
+	w.pulses[key] = &pulse{bucket: bucket, key: k, seeded: true}
+}
+
+// Reconcile ends the start: a pulse that stood and that no mirrored key has made again is a move that happened while
+// the watchman was away, so it is deleted and writes no stuck key.
+func (w *Watchman) Reconcile() {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	for pk, p := range w.pulses {
+		if p.seeded {
+			w.deletePulse(pk)
+		}
+	}
+}
+
+func describe(format string, args ...any) string { return fmt.Sprintf(format, args...) }
+
+// SetBound sets how long a move may take; it is delivery.PulseMoveBound unless a test shortens it.
+func (w *Watchman) SetBound(d time.Duration) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.bound = d
+}
+
+// ttl is the time to live of a pulse for a move of the given age, and whether the move is stalled already
+// (delivery.PulseTTL, with the bound this watchman keeps).
+func (w *Watchman) ttl(age time.Duration) (time.Duration, bool) {
+	if w.bound == delivery.PulseMoveBound {
+		return delivery.PulseTTL(age)
+	}
+	if age >= w.bound {
+		return 0, true
+	}
+	return w.bound - age, false
+}

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

@@ -0,0 +1,442 @@
+package watch
+
+import (
+	"encoding/json"
+	"errors"
+	"slices"
+	"strings"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+type fakeOut struct {
+	presence      map[string]bool
+	presenceWrite int
+	pulses        map[string]delivery.Pulse
+	ttls          map[string]time.Duration
+	pulseDeletes  []string
+	stuck         map[string]delivery.Stuck
+	stuckWrites   int
+}
+
+func newOut() *fakeOut {
+	return &fakeOut{presence: map[string]bool{}, pulses: map[string]delivery.Pulse{}, ttls: map[string]time.Duration{}, stuck: map[string]delivery.Stuck{}}
+}
+
+func (o *fakeOut) PutPresence(key string, online bool) error {
+	o.presence[key] = online
+	o.presenceWrite++
+	return nil
+}
+
+func (o *fakeOut) PutPulse(key string, value []byte, ttl time.Duration) error {
+	p, err := delivery.DecodePulse(value)
+	if err != nil {
+		return err
+	}
+	o.pulses[key], o.ttls[key] = p, ttl
+	return nil
+}
+
+func (o *fakeOut) DeletePulse(key string) error {
+	delete(o.pulses, key)
+	o.pulseDeletes = append(o.pulseDeletes, key)
+	return nil
+}
+
+func (o *fakeOut) PutStuck(key string, value []byte) error {
+	s, err := delivery.DecodeStuck(value)
+	if err != nil {
+		return err
+	}
+	o.stuck[key] = s
+	o.stuckWrites++
+	return nil
+}
+
+var t0 = time.Date(2026, 10, 18, 12, 0, 0, 0, time.UTC)
+
+type clock struct{ t time.Time }
+
+func (c *clock) now() time.Time { return c.t }
+
+func setup() (*Watchman, *fakeOut, *clock) {
+	o, c := newOut(), &clock{t: t0}
+	return New(o, c.now), o, c
+}
+
+func buildValue(module, state string, asked time.Time) []byte {
+	m := map[string]any{"id": "b1", "module": module, "repository": "r"}
+	if state != "" {
+		m["state"] = state
+	}
+	if !asked.IsZero() {
+		m["asked"] = asked
+	}
+	b, _ := json.Marshal(m)
+	return b
+}
+
+func checkValue(started time.Time, head string) []byte {
+	b, _ := json.Marshal(map[string]any{"head": head, "started": started})
+	return b
+}
+
+const (
+	buildsKey  = "mesh-controller_builds"
+	checksKey  = "mesh-controller_checks-running"
+	presenceNS = delivery.PresenceBucket
+)
+
+func pulseKeyOf(t *testing.T, bucket, key string) string {
+	t.Helper()
+	k, err := delivery.PulseKey(bucket, key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return k
+}
+
+func stuckKeyOf(t *testing.T, bucket, key string) string {
+	t.Helper()
+	k, err := delivery.StuckKey(bucket, key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return k
+}
+
+func TestPresenceFromEvents(t *testing.T) { // 1135/AC1
+	w, o, _ := setup()
+	w.Connect("node.ace")
+	w.Connect("ace.mesh-delivery-view")
+	w.Connect("ace.gitea")
+	want := map[string]bool{"ace": true, "part.mesh-delivery-view.ace": true}
+	if len(o.presence) != 2 || o.presence["ace"] != true || o.presence["part.mesh-delivery-view.ace"] != true {
+		t.Fatalf("presence = %v, want %v and nothing for another module's user", o.presence, want)
+	}
+	w.Disconnect("node.ace")
+	w.Disconnect("ace.gitea")
+	if online, ok := o.presence["ace"]; !ok || online {
+		t.Fatalf("presence = %v, want ace offline", o.presence)
+	}
+	if len(o.presence) != 2 {
+		t.Fatalf("another module's user wrote presence: %v", o.presence)
+	}
+}
+
+func TestPresenceWrittenOnlyWhenItDiffers(t *testing.T) { // 1135/AC1
+	w, o, _ := setup()
+	w.Connect("node.ace")
+	w.Connect("node.ace")
+	if o.presenceWrite != 1 {
+		t.Fatalf("writes = %d, want 1 (the second connection changes nothing)", o.presenceWrite)
+	}
+	w.Disconnect("node.ace")
+	if !o.presence["ace"] || o.presenceWrite != 1 {
+		t.Fatalf("one of two connections closed: ace must stay online without a write, got %v, %d writes", o.presence, o.presenceWrite)
+	}
+	w.Disconnect("node.ace")
+	if o.presence["ace"] || o.presenceWrite != 2 {
+		t.Fatalf("both closed: ace offline, got %v, %d writes", o.presence, o.presenceWrite)
+	}
+	w.Disconnect("node.ace") // a stray event
+	if o.presenceWrite != 2 {
+		t.Fatalf("a disconnect of a user already offline wrote %d times", o.presenceWrite)
+	}
+}
+
+func TestParseEvent(t *testing.T) { // 1135/AC1
+	user, err := ParseEvent([]byte(`{"type":"io.nats.server.advisory.v1.client_connect","client":{"acc":"MESH","user":"node.ace","id":4}}`))
+	if err != nil || user != "node.ace" {
+		t.Fatalf("user = %q, %v", user, err)
+	}
+	if _, err := ParseEvent([]byte(`not json`)); err == nil {
+		t.Fatal("an event that is not JSON must be refused")
+	}
+	if user, err := ParseEvent([]byte(`{"client":{}}`)); err != nil || user != "" {
+		t.Fatalf("an event with no user = %q, %v; want empty and no error", user, err)
+	}
+}
+
+func connzAnswer(users ...string) []byte {
+	var conns []map[string]string
+	for _, u := range users {
+		conns = append(conns, map[string]string{"authorized_user": u})
+	}
+	b, _ := json.Marshal(map[string]any{"data": map[string]any{"connections": conns, "total": len(conns)}})
+	return b
+}
+
+func TestRebuildAsksConnectionsWithAuth(t *testing.T) { // 1135/AC2
+	w, o, _ := setup()
+	var body []byte
+	err := w.Rebuild(func(b []byte) ([]byte, error) {
+		body = b
+		return connzAnswer("node.ace", "ace.mesh-delivery-view", "ace.mesh-delivery-view", "ace.gitea"), nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(body) != `{"auth":true}` || string(ConnzRequest()) != `{"auth":true}` {
+		t.Fatalf("request body = %s, want {\"auth\":true}", body)
+	}
+	if !o.presence["ace"] || !o.presence["part.mesh-delivery-view.ace"] || len(o.presence) != 2 {
+		t.Fatalf("presence = %v", o.presence)
+	}
+	// Two connections of the part: one closing leaves it online.
+	w.Disconnect("ace.mesh-delivery-view")
+	if !o.presence["part.mesh-delivery-view.ace"] {
+		t.Fatal("the part went offline with one connection still open")
+	}
+}
+
+func TestRebuildTakesThePresenceOfWhoIsGone(t *testing.T) { // 1135/AC2
+	w, o, c := setup()
+	w.SeedPresence("ace", true, t0.Add(-time.Hour))
+	w.SeedPresence("g14", true, t0.Add(-time.Hour))
+	if err := w.Rebuild(func([]byte) ([]byte, error) { return connzAnswer("node.g14"), nil }); err != nil {
+		t.Fatal(err)
+	}
+	if online, ok := o.presence["ace"]; !ok || online {
+		t.Fatalf("ace was online in the bucket and is not among the connections: %v", o.presence)
+	}
+	if _, ok := o.presence["g14"]; ok {
+		t.Fatalf("g14 is online and was online: nothing to write, got %v", o.presence)
+	}
+	k := pulseKeyOf(t, presenceNS, "ace")
+	if o.ttls[k] != 120*time.Second {
+		t.Fatalf("ttl of the pulse of ace gone = %v, want 120s", o.ttls[k])
+	}
+	_ = c
+	if err := w.Rebuild(func([]byte) ([]byte, error) { return nil, errors.New("no answer") }); err == nil {
+		t.Fatal("a connections request that fails must be an error")
+	}
+}
+
+func TestRebuildPulsesFromCurrentKeys(t *testing.T) { // 1135/AC2
+	w, o, c := setup()
+	c.t = t0.Add(30 * time.Second)
+	w.Build("x.build.b1", buildValue("x", "asked", t0), t0, Put)
+	w.Check("o.r.7", checkValue(t0.Add(10*time.Second), "abc"), t0, Put)
+	w.SeedPresence("ace", false, t0)
+	w.Reconcile()
+	if err := w.Rebuild(func([]byte) ([]byte, error) { return connzAnswer(), nil }); err != nil {
+		t.Fatal(err)
+	}
+	for _, k := range []string{pulseKeyOf(t, buildsKey, "x.build.b1"), pulseKeyOf(t, checksKey, "o.r.7"), pulseKeyOf(t, presenceNS, "ace")} {
+		if _, ok := o.pulses[k]; !ok {
+			t.Errorf("no pulse %s; have %v", k, o.pulses)
+		}
+	}
+}
+
+func TestReconcileDropsPulsesNothingMirrorsAnyMore(t *testing.T) { // 1135/AC2
+	w, o, _ := setup()
+	stale := pulseKeyOf(t, buildsKey, "x.build.gone")
+	live := pulseKeyOf(t, buildsKey, "x.build.b1")
+	w.SeedPulse(stale)
+	w.SeedPulse(live)
+	w.Build("x.build.b1", buildValue("x", "asked", t0), t0, Put)
+	w.Reconcile()
+	if !slices.Equal(o.pulseDeletes, []string{stale}) {
+		t.Fatalf("deleted = %v, want only %s", o.pulseDeletes, stale)
+	}
+	if len(o.stuck) != 0 {
+		t.Fatalf("a stale pulse must not become stuck: %v", o.stuck)
+	}
+}
+
+func TestPulseTimeToLiveIsTheBoundMinusTheAge(t *testing.T) { // 1135/AC3
+	w, o, c := setup()
+	c.t = t0.Add(30 * time.Second)
+	w.Build("x.build.b1", buildValue("x", "asked", t0), t0.Add(time.Second), Put)
+	k := pulseKeyOf(t, buildsKey, "x.build.b1")
+	if o.ttls[k] != 90*time.Second {
+		t.Fatalf("ttl = %v, want 90s (asked 30s ago)", o.ttls[k])
+	}
+	p := o.pulses[k]
+	if p.Bucket != buildsKey || p.Key != "x.build.b1" {
+		t.Fatalf("pulse = %+v", p)
+	}
+	w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
+	if got := o.ttls[pulseKeyOf(t, checksKey, "o.r.7")]; got != 90*time.Second {
+		t.Fatalf("a check started 30s ago: ttl = %v, want 90s", got)
+	}
+	w.Connect("node.ace")
+	w.Disconnect("node.ace")
+	if got := o.ttls[pulseKeyOf(t, presenceNS, "ace")]; got != 120*time.Second {
+		t.Fatalf("presence just gone offline: ttl = %v, want 120s", got)
+	}
+}
+
+func TestAMoveAtTheBoundIsStuckAtOnce(t *testing.T) { // 1135/AC3
+	w, o, c := setup()
+	c.t = t0.Add(120 * time.Second)
+	w.Build("x.build.b1", buildValue("x", "asked", t0), t0, Put)
+	if len(o.pulses) != 0 {
+		t.Fatalf("a pulse was written for a move 120s old: %v", o.pulses)
+	}
+	s, ok := o.stuck[stuckKeyOf(t, buildsKey, "x.build.b1")]
+	if !ok {
+		t.Fatalf("no stuck key; have %v", o.stuck)
+	}
+	if s.Moved || !s.Since.Equal(t0) || s.Waited != "the build of x to be answered" {
+		t.Fatalf("stuck = %+v", s)
+	}
+}
+
+func TestWaitedInWords(t *testing.T) { // 1135/AC3
+	w, o, c := setup()
+	c.t = t0.Add(200 * time.Second)
+	w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
+	w.SeedPresence("ace", false, t0)
+	w.SeedPresence("part.mesh-delivery-view.ace", false, t0)
+	w.Reconcile()
+	_ = w.Rebuild(func([]byte) ([]byte, error) { return connzAnswer(), nil })
+	want := map[string]string{
+		stuckKeyOf(t, checksKey, "o.r.7"):                        "the check of o.r.7 to end",
+		stuckKeyOf(t, presenceNS, "ace"):                         "ace to come back online",
+		stuckKeyOf(t, presenceNS, "part.mesh-delivery-view.ace"): "mesh-delivery-view on ace to come back online",
+	}
+	for k, waited := range want {
+		if o.stuck[k].Waited != waited {
+			t.Errorf("%s waited = %q, want %q", k, o.stuck[k].Waited, waited)
+		}
+	}
+}
+
+func TestMirroredMovesDeleteThePulse(t *testing.T) { // 1135/AC4
+	w, o, c := setup()
+	c.t = t0.Add(10 * time.Second)
+	w.Build("x.build.b1", buildValue("x", "asked", t0), t0, Put)
+	w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
+	w.Connect("node.ace")
+	w.Disconnect("node.ace")
+	w.Build("x.build.b1", buildValue("x", "", t0), t0, Put) // answered: an outcome has no state
+	w.Check("o.r.7", nil, t0, Delete)
+	w.Connect("node.ace")
+	if len(o.pulses) != 0 {
+		t.Fatalf("pulses left: %v", o.pulses)
+	}
+	if len(o.pulseDeletes) != 3 {
+		t.Fatalf("deletes = %v, want 3", o.pulseDeletes)
+	}
+	if len(o.stuck) != 0 {
+		t.Fatalf("a move before expiry wrote stuck: %v", o.stuck)
+	}
+}
+
+func TestACheckWithANewHeadStartsItsPulseAgain(t *testing.T) { // 1135/AC4
+	w, o, c := setup()
+	c.t = t0.Add(60 * time.Second)
+	w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
+	k := pulseKeyOf(t, checksKey, "o.r.7")
+	if o.ttls[k] != 60*time.Second {
+		t.Fatalf("ttl = %v", o.ttls[k])
+	}
+	c.t = t0.Add(70 * time.Second)
+	w.Check("o.r.7", checkValue(t0, "abc"), t0, Put) // a builder's heartbeat: same check
+	if o.ttls[k] != 60*time.Second {
+		t.Fatalf("a heartbeat rewrote the pulse: ttl = %v", o.ttls[k])
+	}
+	w.Check("o.r.7", checkValue(t0.Add(65*time.Second), "def"), t0, Put) // a new head
+	if o.ttls[k] != 115*time.Second {
+		t.Fatalf("the new head's pulse ttl = %v, want 115s", o.ttls[k])
+	}
+}
+
+func TestAStuckKeyIsWrittenMovedNeverDeleted(t *testing.T) { // 1135/AC4
+	w, o, c := setup()
+	w.Build("x.build.b1", buildValue("x", "asked", t0), t0, Put)
+	pk := pulseKeyOf(t, buildsKey, "x.build.b1")
+	c.t = t0.Add(120 * time.Second)
+	w.Pulse(pk, Expired, c.t)
+	sk := stuckKeyOf(t, buildsKey, "x.build.b1")
+	if o.stuck[sk].Moved {
+		t.Fatal("stuck written as moved at expiry")
+	}
+	c.t = t0.Add(300 * time.Second)
+	w.Build("x.build.b1", buildValue("x", "", t0), t0, Put)
+	s := o.stuck[sk]
+	if !s.Moved || !s.Since.Equal(t0) || s.Waited != "the build of x to be answered" {
+		t.Fatalf("stuck after the move = %+v, want the same key moved", s)
+	}
+	n := o.stuckWrites
+	w.Build("x.build.b1", buildValue("x", "", t0), t0, Put)
+	if o.stuckWrites != n {
+		t.Fatal("a stuck key already moved was written again")
+	}
+}
+
+func TestPresenceStuckThenBackOnline(t *testing.T) { // 1135/AC4
+	w, o, c := setup()
+	w.Connect("node.ace")
+	w.Disconnect("node.ace")
+	c.t = t0.Add(120 * time.Second)
+	w.Pulse(pulseKeyOf(t, presenceNS, "ace"), Expired, c.t)
+	sk := stuckKeyOf(t, presenceNS, "ace")
+	if s := o.stuck[sk]; s.Moved || !s.Since.Equal(t0) {
+		t.Fatalf("stuck = %+v", s)
+	}
+	w.Connect("node.ace")
+	if !o.stuck[sk].Moved {
+		t.Fatalf("stuck = %+v, want moved once ace is back", o.stuck[sk])
+	}
+}
+
+func TestExpiryMarkerWritesStuck(t *testing.T) { // 1135/AC5
+	w, o, c := setup()
+	c.t = t0.Add(10 * time.Second)
+	w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
+	c.t = t0.Add(120 * time.Second)
+	w.Pulse(pulseKeyOf(t, checksKey, "o.r.7"), Expired, c.t)
+	s, ok := o.stuck[stuckKeyOf(t, checksKey, "o.r.7")]
+	if !ok {
+		t.Fatalf("no stuck key; have %v", o.stuck)
+	}
+	if s.Bucket != checksKey || s.Key != "o.r.7" || s.Waited != "the check of o.r.7 to end" || !s.Since.Equal(t0) || s.Moved {
+		t.Fatalf("stuck = %+v", s)
+	}
+}
+
+func TestOwnDeleteOfAPulseWritesNothing(t *testing.T) { // 1135/AC5
+	w, o, c := setup()
+	w.Build("x.build.b1", buildValue("x", "asked", t0), t0, Put)
+	pk := pulseKeyOf(t, buildsKey, "x.build.b1")
+	w.Build("x.build.b1", buildValue("x", "", t0), t0, Put)
+	w.Pulse(pk, Delete, c.t) // the delete the watchman made, heard on its own watch
+	w.Pulse(pk, Expired, c.t)
+	w.Pulse("mesh-delivery-watchman_pulse.unknown.key", Expired, c.t)
+	if len(o.stuck) != 0 {
+		t.Fatalf("stuck written for a pulse the watchman deleted or never made: %v", o.stuck)
+	}
+}
+
+func TestOtherBuildKeysAreNotMirrored(t *testing.T) { // 1135/AC3
+	w, o, _ := setup()
+	w.Build("x.newest", []byte("b1"), t0, Put)
+	w.Build("id.b1", []byte("x"), t0, Put)
+	w.Build("manifest.build.b1", buildValue("x", "asked", t0), t0, Put)
+	w.Build("x.build.b2", buildValue("x", "", t0), t0, Put)
+	if len(o.pulses) != 0 || strings.Join(o.pulseDeletes, "") != "" {
+		t.Fatalf("pulses %v, deletes %v", o.pulses, o.pulseDeletes)
+	}
+}
+
+func TestTheBoundCanBeShortenedForATest(t *testing.T) { // 1135/AC8
+	w, o, c := setup()
+	w.SetBound(2 * time.Second)
+	c.t = t0.Add(500 * time.Millisecond)
+	w.Build("x.build.b1", buildValue("x", "asked", t0), t0, Put)
+	if got := o.ttls[pulseKeyOf(t, buildsKey, "x.build.b1")]; got != 1500*time.Millisecond {
+		t.Fatalf("ttl = %v, want 1.5s", got)
+	}
+	c.t = t0.Add(3 * time.Second)
+	w.Build("x.build.b2", buildValue("x", "asked", t0), t0, Put)
+	if _, ok := o.stuck[stuckKeyOf(t, buildsKey, "x.build.b2")]; !ok {
+		t.Fatalf("a move older than the bound is stuck at once; have %v", o.stuck)
+	}
+}