From bdb6fe7787876180716dd24000935243db45572d Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 18 Oct 2026 20:58:34 +0200 Subject: [PATCH] The watchman's logic: presence from the bus's events, pulses with a time to live, stuck keys on expiry (issue 1135) --- .../internal/watch/mirror.go | 90 ++++ .../internal/watch/presence.go | 185 ++++++++ .../internal/watch/watch.go | 240 ++++++++++ .../internal/watch/watch_test.go | 442 ++++++++++++++++++ 4 files changed, 957 insertions(+) create mode 100644 modules/mesh-delivery-watchman/internal/watch/mirror.go create mode 100644 modules/mesh-delivery-watchman/internal/watch/presence.go create mode 100644 modules/mesh-delivery-watchman/internal/watch/watch.go create mode 100644 modules/mesh-delivery-watchman/internal/watch/watch_test.go diff --git a/modules/mesh-delivery-watchman/internal/watch/mirror.go b/modules/mesh-delivery-watchman/internal/watch/mirror.go new file mode 100644 index 0000000..d6fa687 --- /dev/null +++ b/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 `.build.` 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 +} diff --git a/modules/mesh-delivery-watchman/internal/watch/presence.go b/modules/mesh-delivery-watchman/internal/watch/presence.go new file mode 100644 index 0000000..b6bcab2 --- /dev/null +++ b/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.` is the +// node-engine of that node, `.` 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 +} diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go new file mode 100644 index 0000000..e2025f7 --- /dev/null +++ b/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 +} diff --git a/modules/mesh-delivery-watchman/internal/watch/watch_test.go b/modules/mesh-delivery-watchman/internal/watch/watch_test.go new file mode 100644 index 0000000..78ed5b6 --- /dev/null +++ b/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) + } +}