diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go index f2758c6..6906ffd 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go @@ -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) } diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go index 6eab4f9..ba95465 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -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)) }) }() diff --git a/modules/mesh-delivery-watchman/internal/watch/mirror.go b/modules/mesh-delivery-watchman/internal/watch/mirror.go index d6fa687..9e6a767 100644 --- a/modules/mesh-delivery-watchman/internal/watch/mirror.go +++ b/modules/mesh-delivery-watchman/internal/watch/mirror.go @@ -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 `.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). +// 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) } diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index e2025f7..052dbf2 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch.go @@ -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 diff --git a/modules/mesh-delivery-watchman/internal/watch/watch_test.go b/modules/mesh-delivery-watchman/internal/watch/watch_test.go index 78ed5b6..4dd5d3f 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch_test.go @@ -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) } } diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go index 05cd1a8..4c2968d 100644 --- a/modules/mesh-delivery-watchman/manifest_test.go +++ b/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-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"} { diff --git a/modules/mesh-delivery-watchman/module.json b/modules/mesh-delivery-watchman/module.json index 7c6ca3b..2b723d2 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/modules/mesh-delivery-watchman/module.json @@ -3,7 +3,6 @@ "version": "1", "slug": "watchman", "reads": [ - "mesh-controller.builds", "mesh-controller.checks-running" ], "state": [