The watchman mirrors no builds in release 2; a check's pulse is renewed at every heartbeat of its key (issue 1135, review round 1)

This commit is contained in:
2026-10-18 21:05:50 +02:00
parent 431f3f023b
commit e5fd3fcd7a
7 changed files with 120 additions and 180 deletions
@@ -100,7 +100,6 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
{Bucket: delivery.PresenceBucket, History: 20},
{Bucket: delivery.StuckBucket, History: 20},
{Bucket: delivery.PulseBucket, LimitMarkerTTL: time.Second},
{Bucket: watch.BuildsBucket, History: 5},
{Bucket: watch.ChecksBucket, History: 5},
} {
kv, err := js.CreateKeyValue(ctx, c)
@@ -168,16 +167,19 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
n1.Close()
until("node n1 offline", func() bool { on, ok := online("n1"); return ok && !on })
// A build asked and unanswered gets a pulse; let to expire, it writes a stuck key.
now := time.Now().UTC()
asked, _ := json.Marshal(map[string]any{"id": "b1", "module": "x", "state": "asked", "asked": now})
if _, err := kvs[watch.BuildsBucket].Put(ctx, "x.build.b1", asked); err != nil {
// A check running gets a pulse, as its builder puts the key (mesh-controller CheckRunning); a builder that stops
// heartbeating lets the pulse expire, and that writes a stuck key.
running := func() []byte {
b, _ := json.Marshal(map[string]any{"head": "abc", "trunk": "def", "builder": "n1", "build": "b9", "started": time.Now().UTC()})
return b
}
if _, err := kvs[watch.ChecksBucket].Put(ctx, "o.r.7", running()); err != nil {
t.Fatal(err)
}
until("a pulse for the build asked", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(watch.BuildsBucket, "x.build.b1")); return ok })
until("a pulse for the check running", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(watch.ChecksBucket, "o.r.7")); return ok })
var stuck delivery.Stuck
until("a stuck key for the build never answered", func() bool {
e, ok := has(delivery.StuckBucket, stuckOf(watch.BuildsBucket, "x.build.b1"))
until("a stuck key for the check whose builder stopped", func() bool {
e, ok := has(delivery.StuckBucket, stuckOf(watch.ChecksBucket, "o.r.7"))
if !ok {
return false
}
@@ -185,7 +187,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
stuck, err = delivery.DecodeStuck(e.Value())
return err == nil
})
if stuck.Moved || stuck.Waited != "the build of x to be answered" || stuck.Bucket != watch.BuildsBucket || stuck.Key != "x.build.b1" {
if stuck.Moved || stuck.Waited != "the check of o.r.7 to end" || stuck.Bucket != watch.ChecksBucket || stuck.Key != "o.r.7" {
t.Fatalf("stuck = %+v", stuck)
}
// The node n1, gone offline and never back, is stuck as well.
@@ -202,17 +204,17 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
connect("node.n2")
until("node n2 online again", func() bool { on, ok := online("n2"); return ok && on })
until("the pulse of node n2 deleted", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(delivery.PresenceBucket, "n2")); return !ok })
// A build answered at once is the same: no pulse, no stuck key, however long the bound passes.
answered, _ := json.Marshal(map[string]any{"id": "b2", "module": "x", "state": "asked", "asked": time.Now().UTC()})
if _, err := kvs[watch.BuildsBucket].Put(ctx, "x.build.b2", answered); err != nil {
// A check that ends at once is the same: no stuck key, however long the bound passes.
if _, err := kvs[watch.ChecksBucket].Put(ctx, "o.r.8", running()); err != nil {
t.Fatal(err)
}
done2, _ := json.Marshal(map[string]any{"id": "b2", "module": "x"})
if _, err := kvs[watch.BuildsBucket].Put(ctx, "x.build.b2", done2); err != nil {
until("a pulse for the second check", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(watch.ChecksBucket, "o.r.8")); return ok })
if err := kvs[watch.ChecksBucket].Delete(ctx, "o.r.8"); err != nil {
t.Fatal(err)
}
until("the pulse of the second check deleted", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(watch.ChecksBucket, "o.r.8")); return !ok })
time.Sleep(6 * time.Second) // twice the bound and the marker time
for _, k := range []string{stuckOf(delivery.PresenceBucket, "n2"), stuckOf(watch.BuildsBucket, "x.build.b2")} {
for _, k := range []string{stuckOf(delivery.PresenceBucket, "n2"), stuckOf(watch.ChecksBucket, "o.r.8")} {
if _, ok := has(delivery.StuckBucket, k); ok {
t.Errorf("a stuck key %s was written for a move that happened before the bound", k)
}
@@ -1,6 +1,6 @@
// Command mesh-delivery-watchman is delivery's watchman (novox/hq ADR 0429 §1, ADR 0431, ADR 0433, issue 1135): it
// hears the bus's connect and disconnect events and writes presence, mirrors every move a machine owes (a build asked
// and not answered, a check running, a node or part gone offline) into the pulse bucket with a time to live, and writes
// hears the bus's connect and disconnect events and writes presence, mirrors every move a machine owes (a check running,
// a node or part gone offline) into the pulse bucket with a time to live, and writes
// a stuck key when the server expires one. It holds no timer: every reaction is an event, a watch or a request at
// start or at a reconnect, and the server's expiry is the only clock.
package main
@@ -223,12 +223,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
return kv, nil
}
var kvs struct{ presence, pulse, stuck, builds, checks jetstream.KeyValue }
var kvs struct{ presence, pulse, stuck, checks jetstream.KeyValue }
for _, b := range []struct {
name string
into *jetstream.KeyValue
}{{delivery.PresenceBucket, &kvs.presence}, {delivery.PulseBucket, &kvs.pulse}, {delivery.StuckBucket, &kvs.stuck},
{watch.BuildsBucket, &kvs.builds}, {watch.ChecksBucket, &kvs.checks}} {
{watch.ChecksBucket, &kvs.checks}} {
if *b.into, err = bucket(b.name); err != nil {
return err
}
@@ -245,7 +245,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
w.SetBound(d)
}
// What stands now: presence and stuck as written, pulses as the server holds them, then the builds and checks.
// What stands now: presence and stuck as written, pulses as the server holds them, then the checks.
for _, s := range []struct {
kv jetstream.KeyValue
each func(jetstream.KeyValueEntry)
@@ -275,12 +275,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
if err != nil {
return err
}
builds, err := settle(ctx, kvs.builds, func(e jetstream.KeyValueEntry) {
w.Build(e.Key(), e.Value(), e.Created(), opOf(e))
})
if err != nil {
return err
}
checks, err := settle(ctx, kvs.checks, func(e jetstream.KeyValueEntry) {
w.Check(e.Key(), e.Value(), e.Created(), opOf(e))
})
@@ -327,9 +321,6 @@ 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, builds, func(e jetstream.KeyValueEntry) { w.Build(e.Key(), e.Value(), e.Created(), opOf(e)) })
}()
go func() {
errs <- follow(ctx, checks, func(e jetstream.KeyValueEntry) { w.Check(e.Key(), e.Value(), e.Created(), opOf(e)) })
}()
@@ -2,62 +2,24 @@ 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"}
// ChecksBucket is the bucket of the checks the builders are running (mesh-controller internal/broker
// CheckRunningBucket).
const ChecksBucket = "mesh-controller_checks-running"
// 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).
// checkRecord is the part of a running check's key the watchman reads, as the controller's builder writes it
// (mesh-controller internal/link/checks_nats.go CheckRunning): the head being checked.
type checkRecord struct {
Head string `json:"head"`
Started time.Time `json:"started"`
Head string `json:"head"`
}
// 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.
// Check is a change in the checks-running bucket. The builder puts a check's key when the check starts and puts it
// again at every heartbeat while the check lives, and deletes it when the check ends. Each new revision of the key is
// the check moving: the pulse is deleted and written again with the full bound from that revision's time, so only a
// check whose builder stops heartbeating for the whole bound is stuck; the key deleted 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()
@@ -70,21 +32,11 @@ func (w *Watchman) Check(key string, value []byte, at time.Time, op Op) {
}
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
}
if at.IsZero() {
at = w.now()
}
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
}
if _, stalled := w.ttl(w.now().Sub(at)); !stalled {
w.stuckMoved(ChecksBucket, key)
}
return nil
w.mirror(ChecksBucket, key, describe("the check of %s to end", key), rec.Head, at, 0)
}
@@ -23,12 +23,6 @@ const (
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
@@ -131,6 +125,11 @@ func (w *Watchman) moved(bucket, key string) {
if _, ok := w.pulses[pk]; ok {
w.deletePulse(pk)
}
w.stuckMoved(bucket, key)
}
// stuckMoved writes a stuck key of the key as moved, if it stands and is not yet.
func (w *Watchman) stuckMoved(bucket, key string) {
sk, err := delivery.StuckKey(bucket, key)
if err != nil {
return
@@ -4,7 +4,6 @@ import (
"encoding/json"
"errors"
"slices"
"strings"
"testing"
"time"
@@ -67,25 +66,14 @@ func setup() (*Watchman, *fakeOut, *clock) {
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
}
// checkValue is a check's key as the controller's builder writes it (mesh-controller internal/link/checks_nats.go
// CheckRunning, kept by keepRunning on every heartbeat).
func checkValue(started time.Time, head string) []byte {
b, _ := json.Marshal(map[string]any{"head": head, "started": started})
b, _ := json.Marshal(map[string]any{"head": head, "trunk": "def456", "builder": "n1", "build": "b9", "started": started})
return b
}
const (
buildsKey = "mesh-controller_builds"
checksKey = "mesh-controller_checks-running"
presenceNS = delivery.PresenceBucket
)
@@ -219,14 +207,13 @@ func TestRebuildTakesThePresenceOfWhoIsGone(t *testing.T) { // 1135/AC2
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.Check("o.r.7", checkValue(t0, "abc"), t0.Add(10*time.Second), 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")} {
for _, k := range []string{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)
}
@@ -235,11 +222,11 @@ func TestRebuildPulsesFromCurrentKeys(t *testing.T) { // 1135/AC2
func TestReconcileDropsPulsesNothingMirrorsAnyMore(t *testing.T) { // 1135/AC2
w, o, _ := setup()
stale := pulseKeyOf(t, buildsKey, "x.build.gone")
live := pulseKeyOf(t, buildsKey, "x.build.b1")
stale := pulseKeyOf(t, checksKey, "o.r.gone")
live := pulseKeyOf(t, checksKey, "o.r.7")
w.SeedPulse(stale)
w.SeedPulse(live)
w.Build("x.build.b1", buildValue("x", "asked", t0), t0, Put)
w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
w.Reconcile()
if !slices.Equal(o.pulseDeletes, []string{stale}) {
t.Fatalf("deleted = %v, want only %s", o.pulseDeletes, stale)
@@ -252,19 +239,14 @@ func TestReconcileDropsPulsesNothingMirrorsAnyMore(t *testing.T) { // 1135/AC2
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")
w.Check("o.r.7", checkValue(t0, "abc"), t0, Put) // the key found at start, last renewed 30s ago
k := pulseKeyOf(t, checksKey, "o.r.7")
if o.ttls[k] != 90*time.Second {
t.Fatalf("ttl = %v, want 90s (asked 30s ago)", o.ttls[k])
t.Fatalf("ttl = %v, want 90s (last renewed 30s ago)", o.ttls[k])
}
p := o.pulses[k]
if p.Bucket != buildsKey || p.Key != "x.build.b1" {
if p := o.pulses[k]; p.Bucket != checksKey || p.Key != "o.r.7" {
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 {
@@ -275,15 +257,15 @@ func TestPulseTimeToLiveIsTheBoundMinusTheAge(t *testing.T) { // 1135/AC3
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)
w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
if len(o.pulses) != 0 {
t.Fatalf("a pulse was written for a move 120s old: %v", o.pulses)
t.Fatalf("a pulse was written for a heartbeat 120s old: %v", o.pulses)
}
s, ok := o.stuck[stuckKeyOf(t, buildsKey, "x.build.b1")]
s, ok := o.stuck[stuckKeyOf(t, checksKey, "o.r.7")]
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" {
if s.Moved || !s.Since.Equal(t0) || s.Waited != "the check of o.r.7 to end" {
t.Fatalf("stuck = %+v", s)
}
}
@@ -308,69 +290,96 @@ func TestWaitedInWords(t *testing.T) { // 1135/AC3
}
}
func TestEveryHeartbeatRenewsThePulse(t *testing.T) { // 1135/AC3
w, o, c := setup()
k := pulseKeyOf(t, checksKey, "o.r.7")
w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
if o.ttls[k] != 120*time.Second {
t.Fatalf("ttl = %v, want the full 120s", o.ttls[k])
}
o.ttls[k] = 0
c.t = t0.Add(100 * time.Second)
w.Check("o.r.7", checkValue(t0, "abc"), c.t, Put) // the builder re-puts the key: same check, started long ago
if o.ttls[k] != 120*time.Second {
t.Fatalf("a heartbeat left the pulse as it was: ttl = %v, want the full 120s again", o.ttls[k])
}
if len(o.stuck) != 0 {
t.Fatalf("a check that heartbeats is stuck: %v", o.stuck)
}
// The same revision heard twice changes nothing.
o.ttls[k] = 0
w.Check("o.r.7", checkValue(t0, "abc"), c.t, Put)
if o.ttls[k] != 0 {
t.Fatal("one revision rewrote the pulse twice")
}
}
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.Check("o.r.7", nil, c.t, Delete) // the check ended
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.pulseDeletes) != 2 {
t.Fatalf("deletes = %v, want 2", 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
func TestACheckWhoseBuilderStopsIsStuckFromItsLastHeartbeat(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])
c.t = t0.Add(50 * time.Second)
w.Check("o.r.7", checkValue(t0, "abc"), c.t, Put) // the last heartbeat
c.t = t0.Add(170 * time.Second)
w.Pulse(pulseKeyOf(t, checksKey, "o.r.7"), Expired, c.t)
s := o.stuck[stuckKeyOf(t, checksKey, "o.r.7")]
if !s.Since.Equal(t0.Add(50 * time.Second)) {
t.Fatalf("stuck = %+v, want since the last heartbeat", s)
}
}
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")
w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
pk := pulseKeyOf(t, checksKey, "o.r.7")
c.t = t0.Add(120 * time.Second)
w.Pulse(pk, Expired, c.t)
sk := stuckKeyOf(t, buildsKey, "x.build.b1")
sk := stuckKeyOf(t, checksKey, "o.r.7")
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)
w.Check("o.r.7", nil, c.t, Delete)
s := o.stuck[sk]
if !s.Moved || !s.Since.Equal(t0) || s.Waited != "the build of x to be answered" {
if !s.Moved || !s.Since.Equal(t0) || s.Waited != "the check of o.r.7 to end" {
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)
w.Check("o.r.7", nil, c.t, Delete)
if o.stuckWrites != n {
t.Fatal("a stuck key already moved was written again")
}
}
func TestAHeartbeatAfterStuckMovesIt(t *testing.T) { // 1135/AC4
w, o, c := setup()
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)
w.Check("o.r.7", checkValue(t0, "abc"), c.t, Put) // the builder is back
if !o.stuck[stuckKeyOf(t, checksKey, "o.r.7")].Moved {
t.Fatal("a check heartbeating again is still stuck")
}
}
func TestPresenceStuckThenBackOnline(t *testing.T) { // 1135/AC4
w, o, c := setup()
w.Connect("node.ace")
@@ -389,7 +398,6 @@ func TestPresenceStuckThenBackOnline(t *testing.T) { // 1135/AC4
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)
@@ -404,9 +412,9 @@ func TestExpiryMarkerWritesStuck(t *testing.T) { // 1135/AC5
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.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
pk := pulseKeyOf(t, checksKey, "o.r.7")
w.Check("o.r.7", nil, c.t, Delete)
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)
@@ -415,28 +423,17 @@ func TestOwnDeleteOfAPulseWritesNothing(t *testing.T) { // 1135/AC5
}
}
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 {
w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
if got := o.ttls[pulseKeyOf(t, checksKey, "o.r.7")]; 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)
w.Check("o.r.8", checkValue(t0, "abc"), t0, Put)
if _, ok := o.stuck[stuckKeyOf(t, checksKey, "o.r.8")]; !ok {
t.Fatalf("a heartbeat older than the bound is stuck at once; have %v", o.stuck)
}
}
@@ -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-controller.builds", "mesh-controller.checks-running"}) {
if !reflect.DeepEqual(reads, []string{"mesh-controller.checks-running"}) {
t.Errorf("reads = %v", reads)
}
for _, k := range []string{"emits", "context", "items", "consumes", "provides", "claims"} {
@@ -3,7 +3,6 @@
"version": "1",
"slug": "watchman",
"reads": [
"mesh-controller.builds",
"mesh-controller.checks-running"
],
"state": [