diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go index 0b279ac..336b33a 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go @@ -602,3 +602,97 @@ func TestProcessWholeWritesTheAnswerToAnAskNeedingNoProof(t *testing.T) { // 129 } wait(func(s view.Snapshot) bool { return len(s.Asks) == 1 && s.Asks[0].Key == mkey }, "the answered ask never left the list") } + +// TestProcessWholeShowsTheBacklogOfBuildAsks starts the program whole against a bus holding the go seat's work-queue stream +// with its three shared workers, three check asks waiting and one held by a node: the socket's backlog row names three +// waiting, one held, and the oldest's age (1887/A8a AC; compiled in the unit run, run by the merge check). +func TestProcessWholeShowsTheBacklogOfBuildAsks(t *testing.T) { // 1887/A8a AC + srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true}) + if err != nil { + t.Fatal(err) + } + go srv.Start() + if !srv.ReadyForConnections(10 * time.Second) { + t.Fatal("the bus did not come up") + } + t.Cleanup(srv.Shutdown) + nc, err := nats.Connect(srv.ClientURL()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + stream, _ := delivery.AskStream(delivery.AskGo) + st, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{"mesh.seat.node-build-go.accept.>"}, Retention: jetstream.WorkQueuePolicy}) + if err != nil { + t.Fatal(err) + } + var checkWorker jetstream.Consumer + for _, kind := range delivery.AskKinds { + name, _ := delivery.AskWorker(delivery.AskGo, kind) + subject, _ := delivery.AskSubject(delivery.AskGo, kind) + c, err := st.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{Durable: name, FilterSubject: subject, AckPolicy: jetstream.AckExplicitPolicy}) + if err != nil { + t.Fatal(err) + } + if kind == delivery.AskCheck { + checkWorker = c + } + } + subject, _ := delivery.AskSubject(delivery.AskGo, delivery.AskCheck) + oldest := time.Now().UTC().Add(-2 * time.Minute) + for i, id := range []string{"m0000001aaaaaa", "m0000002bbbbbb", "m0000003cccccc", "m0000004dddddd"} { + body, err := delivery.EncodeStepAsk(delivery.StepAsk{ID: id, Kind: delivery.AskCheck, Language: delivery.AskGo, Asker: "mesh-delivery-checker", At: oldest.Add(time.Duration(i) * time.Second), + Check: &delivery.CheckJob{Fetch: map[string]any{"repository": "novox/x", "commit": "0123456789abcdef0123456789abcdef01234567"}, Levels: []string{delivery.LevelUnit}}}) + if err != nil { + t.Fatal(err) + } + if _, err := js.Publish(ctx, subject, body); err != nil { + t.Fatal(err) + } + } + // A node pulls one and holds it unacknowledged. + held, err := checkWorker.Fetch(1) + if err != nil { + t.Fatal(err) + } + for range held.Messages() { + } + + dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(dir) }) + broker := filepath.Join(dir, "broker") + cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) + if err := os.WriteFile(broker, cred, 0o600); err != nil { + t.Fatal(err) + } + socket := filepath.Join(dir, "run", "cli.sock") + env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"} + runCtx, stop := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() + t.Cleanup(func() { stop(); <-done }) + + var last view.Snapshot + for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) { + s, err := view.Ask(socket) + if err != nil { + continue + } + last = s + for _, b := range s.Backlog { + if b.Language == "go" && b.Kind == "check" && b.Waiting == 3 && b.Held == 1 && b.OldestSecs >= 118 { + return + } + } + } + t.Fatalf("the backlog never showed three waiting asks with the oldest's age; last: %+v", last.Backlog) +} diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go index 5863ce1..8b190eb 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go @@ -138,6 +138,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } model.SetWriter(kvWriter{ctx: ctx, js: js}) + model.SetBacklog(view.BacklogOver(js, 5*time.Second)) for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket, delivery.CandidatesBucket, delivery.ModulesBucket, delivery.AsksBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, delivery.AssignerDifferencesBucket, delivery.AssignerAsksBucket, delivery.ViewAnswersBucket} { go follow(ctx, js, bucket, model, logger) } diff --git a/modules/mesh-delivery-view/internal/view/backlog.go b/modules/mesh-delivery-view/internal/view/backlog.go new file mode 100644 index 0000000..14e5598 --- /dev/null +++ b/modules/mesh-delivery-view/internal/view/backlog.go @@ -0,0 +1,116 @@ +package view + +import ( + "context" + "fmt" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// The backlog of the build seats (novox/hq ADR 0525 §4, issue 1887 A8a): for each language and kind of ask, how many asks wait +// on the seat's stream for a node to pull them, how many a node holds without having finished them, and how old the oldest +// waiting ask is. It is read from the stream's shared pull worker (delivery.AskWorker) when the view is asked: NumPending +// waits, NumAckPending is held by a node, and the oldest waiting ask is the stream's message at the worker's next sequence. + +// BacklogRow is one language's kind of ask: Waiting asks not yet pulled, Held asks pulled by a node and not yet ended, and the +// oldest waiting ask's time and age in seconds. Error says why the row could not be read, never dropping it silently. +type BacklogRow struct { + Language string `json:"language"` + Kind string `json:"kind"` + Waiting uint64 `json:"waiting"` + Held uint64 `json:"held"` + Oldest time.Time `json:"oldest,omitzero"` + OldestSecs int64 `json:"oldest-secs,omitempty"` + Error string `json:"error,omitempty"` +} + +// Backlog reads the seats' workers. Info is the worker's counts and the stream sequence it delivers next; Message is the body +// of the first message of the stream at or after seq on the subject. +type Backlog struct { + Info func(stream, worker string) (waiting, held, next uint64, err error) + Message func(stream, subject string, seq uint64) ([]byte, error) + Now func() time.Time +} + +// Rows are the rows of every language and kind that has an ask waiting or held, or that could not be read. +func (b *Backlog) Rows() []BacklogRow { + now := time.Now + if b.Now != nil { + now = b.Now + } + out := []BacklogRow{} + for _, lang := range delivery.AskLanguages { + stream, _ := delivery.AskStream(lang) + for _, kind := range delivery.AskKinds { + worker, _ := delivery.AskWorker(lang, kind) + subject, _ := delivery.AskSubject(lang, kind) + row := BacklogRow{Language: lang, Kind: kind} + waiting, held, next, err := b.Info(stream, worker) + if err != nil { + row.Error = err.Error() + out = append(out, row) + continue + } + row.Waiting, row.Held = waiting, held + if waiting > 0 { + body, err := b.Message(stream, subject, next) + if err == nil { + var a delivery.StepAsk + a, err = delivery.DecodeStepAsk(body) + if err == nil { + row.Oldest, row.OldestSecs = a.At, int64(now().Sub(a.At).Seconds()) + } + } + if err != nil { + row.Error = fmt.Sprintf("the oldest waiting ask cannot be read: %v", err) + } + } + if row.Waiting > 0 || row.Held > 0 || row.Error != "" { + out = append(out, row) + } + } + } + return out +} + +// BacklogOver reads the backlog from the bus: the worker's consumer info, and the stream's message by sequence. +func BacklogOver(js jetstream.JetStream, timeout time.Duration) *Backlog { + ctx := func() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), timeout) + } + return &Backlog{ + Info: func(stream, worker string) (uint64, uint64, uint64, error) { + c, cancel := ctx() + defer cancel() + s, err := js.Stream(c, stream) + if err != nil { + return 0, 0, 0, fmt.Errorf("the stream %s: %w", stream, err) + } + cons, err := s.Consumer(c, worker) + if err != nil { + return 0, 0, 0, fmt.Errorf("the worker %s: %w", worker, err) + } + info := cons.CachedInfo() + if i, err := cons.Info(c); err == nil { + info = i + } + return uint64(info.NumPending), uint64(info.NumAckPending), info.Delivered.Stream + 1, nil + }, + Message: func(stream, subject string, seq uint64) ([]byte, error) { + c, cancel := ctx() + defer cancel() + s, err := js.Stream(c, stream) + if err != nil { + return nil, err + } + m, err := s.GetMsg(c, seq, jetstream.WithGetMsgSubject(subject)) + if err != nil { + return nil, err + } + return m.Data, nil + }, + } +} diff --git a/modules/mesh-delivery-view/internal/view/backlog_test.go b/modules/mesh-delivery-view/internal/view/backlog_test.go new file mode 100644 index 0000000..2ec5bc4 --- /dev/null +++ b/modules/mesh-delivery-view/internal/view/backlog_test.go @@ -0,0 +1,87 @@ +package view + +import ( + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +func stepAsk(t *testing.T, id, kind string, at time.Time) []byte { + t.Helper() + b, err := delivery.EncodeStepAsk(delivery.StepAsk{ID: id, Kind: kind, Language: delivery.AskGo, Asker: "mesh-delivery-checker", At: at, + Check: &delivery.CheckJob{Fetch: map[string]any{"repository": "novox/x", "commit": "0123456789abcdef0123456789abcdef01234567"}, Levels: []string{delivery.LevelUnit}}}) + if err != nil { + t.Fatal(err) + } + return b +} + +// fakeSeats is the check worker of the go seat with three asks waiting and one held, the oldest at t0. +func fakeSeats(t *testing.T, now time.Time) *Backlog { + worker, _ := delivery.AskWorker(delivery.AskGo, delivery.AskCheck) + subject, _ := delivery.AskSubject(delivery.AskGo, delivery.AskCheck) + oldest := stepAsk(t, "m0000001aaaaaa", delivery.AskCheck, now.Add(-90*time.Second)) + return &Backlog{Now: func() time.Time { return now }, + Info: func(stream, w string) (uint64, uint64, uint64, error) { + switch w { + case worker: + return 3, 1, 42, nil + case "SEAT_NODE_BUILD_IMAGE_DELIVERY_worker": + return 0, 0, 1, fmt.Errorf("the worker is not there") + } + return 0, 0, 1, nil + }, + Message: func(stream, s string, seq uint64) ([]byte, error) { + if s != subject || seq != 42 { + return nil, fmt.Errorf("asked %s at %d", s, seq) + } + return oldest, nil + }} +} + +func TestThreeWaitingAsksAreShownWithTheOldestsAge(t *testing.T) { // 1887/A8a AC + now := time.Date(2026, 10, 23, 12, 0, 0, 0, time.UTC) + m := New() + m.SetBacklog(fakeSeats(t, now)) + rows := m.Snapshot().Backlog + var check, image *BacklogRow + for i := range rows { + switch { + case rows[i].Language == "go" && rows[i].Kind == "check": + check = &rows[i] + case rows[i].Language == "image" && rows[i].Kind == "delivery": + image = &rows[i] + default: + t.Errorf("a quiet seat is listed: %+v", rows[i]) + } + } + if check == nil || check.Waiting != 3 || check.Held != 1 || check.OldestSecs != 90 || !check.Oldest.Equal(now.Add(-90*time.Second)) || check.Error != "" { + t.Fatalf("check row = %+v", check) + } + if image == nil || image.Error == "" { + t.Errorf("a worker that cannot be read is dropped: %+v", rows) + } + var out strings.Builder + Table(&out, m.Snapshot()) + for _, w := range []string{"BACKLOG OF BUILD ASKS", "go", "check", "3", "1m30s", "unreadable"} { + if !strings.Contains(out.String(), w) { + t.Errorf("table lacks %q:\n%s", w, out.String()) + } + } + rec := httptest.NewRecorder() + Web(m).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/", nil)) + if !strings.Contains(rec.Body.String(), "BACKLOG OF BUILD ASKS") || !strings.Contains(rec.Body.String(), "1m30s") { + t.Errorf("page lacks the backlog:\n%s", rec.Body.String()) + } +} + +func TestNoBacklogReaderGivesNoRows(t *testing.T) { + if got := New().Snapshot().Backlog; got == nil || len(got) != 0 { + t.Errorf("backlog = %#v", got) + } +} diff --git a/modules/mesh-delivery-view/internal/view/serve.go b/modules/mesh-delivery-view/internal/view/serve.go index 613552d..f8f6685 100644 --- a/modules/mesh-delivery-view/internal/view/serve.go +++ b/modules/mesh-delivery-view/internal/view/serve.go @@ -236,6 +236,24 @@ func differenceText(s Snapshot) [][]string { return out } +// backlogText is one line per language and kind: its waiting asks, the asks held by a node, the oldest waiting ask's age. +func backlogText(s Snapshot) [][]string { + var out [][]string + for _, b := range s.Backlog { + age := "" + if !b.Oldest.IsZero() { + age = (time.Duration(b.OldestSecs) * time.Second).String() + " (since " + since(b.Oldest) + ")" + } + if b.Error != "" { + age = "unreadable: " + b.Error + } + out = append(out, []string{b.Language, b.Kind, fmt.Sprint(b.Waiting), fmt.Sprint(b.Held), age}) + } + return out +} + +var backlogHead = []string{"LANGUAGE", "KIND", "WAITING", "HELD", "OLDEST WAITING"} + var differenceHead = []string{"NODE", "MODULE", "WANT", "SENT", "RULE", "WORDS"} var head = []string{"NODE", "MODULE", "ASSIGNED", "HAVE", "PHASE", "HEALTH", "SINCE", "DETAIL", "WAITS"} @@ -277,6 +295,14 @@ func Table(w io.Writer, s Snapshot) { row(cells) } } + if b := backlogText(s); len(b) > 0 { + fmt.Fprintln(tw) + row([]string{"BACKLOG OF BUILD ASKS"}) + row(backlogHead) + for _, cells := range b { + row(cells) + } + } if d := differenceText(s); len(d) > 0 { fmt.Fprintln(tw) row([]string{"WANTS AGAINST SENDS"}) @@ -329,6 +355,7 @@ var page = template.Must(template.New("p").Parse(`{{range .}}{{.}}{{end}}{{end}} {{if .Groups}}

CANDIDATES

{{range .Groups}}{{range .}}{{end}}{{end}}
GROUPPULL REQUESTHEADMODULES
{{.}}
{{end}} {{if .Asks}}

ASKS

{{range .Asks}}{{range .Cells}}{{end}}{{end}}
ASKHEADLINEWORDSPULL REQUESTSANSWER
{{.}}{{if .Options}}
{{range .Options}} {{end}}
{{end}}
{{end}} +{{if .Backlog}}

BACKLOG OF BUILD ASKS

{{range .BacklogHead}}{{end}}{{range .Backlog}}{{range .}}{{end}}{{end}}
{{.}}
{{.}}
{{end}} {{if .Differences}}

WANTS AGAINST SENDS

{{range .DifferenceHead}}{{end}}{{range .Differences}}{{range .}}{{end}}{{end}}
{{.}}
{{.}}
{{end}} {{if .Fingerprints}}

FINGERPRINTS

{{range .Fingerprints}}{{range .}}{{end}}{{end}}
MODULESTATEINDEXER AGAINST BUILD
{{.}}
{{end}} {{if .Offline}}

OFFLINE

{{range .Offline}}{{range .}}{{end}}{{end}}
{{.}}
{{end}} @@ -383,7 +410,7 @@ func Web(m *Model) http.Handler { for _, u := range snap.Unreadable { bad = append(bad, []string{u.Bucket, u.Key, u.Error}) } - page.Execute(w, map[string]any{"Head": head, "Rows": rows, "Groups": groupText(snap), "Asks": askPage(snap), "Fingerprints": fingerprintText(snap), "Differences": differenceText(snap), "DifferenceHead": differenceHead, "Offline": off, "Stuck": stuck, "Unreadable": bad}) + page.Execute(w, map[string]any{"Head": head, "Rows": rows, "Groups": groupText(snap), "Asks": askPage(snap), "Fingerprints": fingerprintText(snap), "Differences": differenceText(snap), "Backlog": backlogText(snap), "BacklogHead": backlogHead, "DifferenceHead": differenceHead, "Offline": off, "Stuck": stuck, "Unreadable": bad}) }) } diff --git a/modules/mesh-delivery-view/internal/view/view.go b/modules/mesh-delivery-view/internal/view/view.go index 2f82e00..dfc0db2 100644 --- a/modules/mesh-delivery-view/internal/view/view.go +++ b/modules/mesh-delivery-view/internal/view/view.go @@ -69,6 +69,9 @@ type Snapshot struct { // Differences is every difference between what the assigner wants a node to run and what the controller sends, with the rule that // explains it or unexplained (novox/hq ADR 0528 clause 4, issue 1297 B8). Differences []DifferenceRow `json:"differences"` + // Backlog is, for each language and kind of build ask with one waiting or held, the asks waiting for a node, the asks a node + // holds and the oldest waiting ask's age (issue 1887 A8a). + Backlog []BacklogRow `json:"backlog"` } // DifferenceRow is a module on a node whose want and send differ: the fingerprint wanted (empty when the node has no want, @@ -186,6 +189,7 @@ type Model struct { aasks map[string]assignerAskRecord // the assigner's asks (hold, restore), by key chosen map[string]delivery.ViewAnswer // the answers the view wrote, by ask key writer Writer + backlog *Backlog answers map[string]string // by key: the router id of the asking the teller's answer is for ("" when unreadable) newest map[string]string // module -> id haves map[string]haveEntry @@ -452,6 +456,25 @@ func errText(err error) string { // Snapshot derives one row per module and node with an open assignment, a row marked left behind for each have with // none, and a row per node's own key. Rows are ordered by node and module. func (m *Model) Snapshot() Snapshot { + s := m.snapshot() + m.mu.Lock() + b := m.backlog + m.mu.Unlock() + s.Backlog = []BacklogRow{} + if b != nil { + s.Backlog = b.Rows() // read from the bus outside the model's lock + } + return s +} + +// SetBacklog gives the model the seats' workers to read the backlog of asks from. +func (m *Model) SetBacklog(b *Backlog) { + m.mu.Lock() + defer m.mu.Unlock() + m.backlog = b +} + +func (m *Model) snapshot() Snapshot { m.mu.Lock() defer m.mu.Unlock() rows := []Row{} diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go index 2f25bfe..de423f5 100644 --- a/modules/mesh-delivery-view/manifest_test.go +++ b/modules/mesh-delivery-view/manifest_test.go @@ -48,6 +48,7 @@ type manifest struct { Version string `json:"version"` Slug string `json:"slug"` Reads []string `json:"reads"` + Uses []string `json:"uses"` Tools []string `json:"tools"` OwnSecrets map[string]string `json:"own-secrets"` SecretsOwner string `json:"secrets-owner"` @@ -93,6 +94,10 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 t.Errorf("the manifest declares %q: the view writes only its answers", k) } } + // The view reads the build seats' streams for the backlog of asks (issue 1887 A8a): the three seats it uses. + if want := []string{"node-build-go", "node-build-image", "node-build-node"}; !reflect.DeepEqual(m.Uses, want) { + t.Errorf("uses = %v, want %v", m.Uses, want) + } // The one state the view writes: the operator's answers to the asks that need no proof (ADR 0528 clause 6, 1297/B9), which the teller reads. if want := []state{{Name: "answers", History: 5}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json index 61ed274..0d5ed50 100644 --- a/modules/mesh-delivery-view/module.json +++ b/modules/mesh-delivery-view/module.json @@ -28,6 +28,11 @@ "mesh-delivery-watchman.stuck", "node-engine.have" ], + "uses": [ + "node-build-go", + "node-build-image", + "node-build-node" + ], "own-secrets": { "broker": "${dir:state}/broker" }, 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 6906ffd..b4a4207 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 @@ -17,8 +17,6 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" - - "mesh-delivery-watchman/internal/watch" ) // busConf is the composed accounts file's shape (mesh-controller internal/broker): the account MESH imports the system @@ -100,7 +98,7 @@ 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.ChecksBucket, History: 5}, + {Bucket: delivery.StepsBucket("mesh-build-go"), History: 5}, } { kv, err := js.CreateKeyValue(ctx, c) if err != nil { @@ -167,19 +165,26 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 n1.Close() until("node n1 offline", func() bool { on, ok := online("n1"); return ok && !on }) - // 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 + // An ask put on a build seat gets a pulse; one no node takes lets the pulse expire, and that writes a stuck key. + askStream, _ := delivery.AskStream(delivery.AskGo) + askSubject, _ := delivery.AskSubject(delivery.AskGo, delivery.AskCheck) + putAsk := func(id string) { + t.Helper() + body, err := delivery.EncodeStepAsk(delivery.StepAsk{ID: id, Kind: delivery.AskCheck, Language: delivery.AskGo, Asker: "mesh-delivery-checker", At: time.Now().UTC(), + Check: &delivery.CheckJob{Fetch: map[string]any{"repository": "novox/x", "commit": "0123456789abcdef0123456789abcdef01234567"}, Levels: []string{delivery.LevelUnit}}}) + if err != nil { + t.Fatal(err) + } + if err := ctl.Publish(askSubject, body); err != nil { + t.Fatal(err) + } } - if _, err := kvs[watch.ChecksBucket].Put(ctx, "o.r.7", running()); err != nil { - t.Fatal(err) - } - until("a pulse for the check running", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(watch.ChecksBucket, "o.r.7")); return ok }) + const ask1, ask2 = "m0000001aaaaaa", "m0000002bbbbbb" + putAsk(ask1) + until("a pulse for the ask waiting", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask1)); return ok }) var stuck delivery.Stuck - until("a stuck key for the check whose builder stopped", func() bool { - e, ok := has(delivery.StuckBucket, stuckOf(watch.ChecksBucket, "o.r.7")) + until("a stuck key for the ask no node took", func() bool { + e, ok := has(delivery.StuckBucket, stuckOf(askStream, ask1)) if !ok { return false } @@ -187,7 +192,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 stuck, err = delivery.DecodeStuck(e.Value()) return err == nil }) - if stuck.Moved || stuck.Waited != "the check of o.r.7 to end" || stuck.Bucket != watch.ChecksBucket || stuck.Key != "o.r.7" { + if stuck.Moved || stuck.Bucket != askStream || stuck.Key != ask1 { t.Fatalf("stuck = %+v", stuck) } // The node n1, gone offline and never back, is stuck as well. @@ -204,17 +209,15 @@ 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 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 { + // An ask a node takes at once is the same: no stuck key, however long the bound passes. + putAsk(ask2) + until("a pulse for the second ask", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask2)); return ok }) + if _, err := kvs[delivery.StepsBucket("mesh-build-go")].Put(ctx, "n1."+ask2, []byte(`{}`)); err != nil { t.Fatal(err) } - 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 }) + until("the pulse of the second ask deleted", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask2)); return !ok }) time.Sleep(6 * time.Second) // twice the bound and the marker time - for _, k := range []string{stuckOf(delivery.PresenceBucket, "n2"), stuckOf(watch.ChecksBucket, "o.r.8")} { + for _, k := range []string{stuckOf(delivery.PresenceBucket, "n2"), stuckOf(askStream, ask2)} { 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 ba95465..ac5ec78 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -1,5 +1,5 @@ // 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 check running, +// hears the bus's connect and disconnect events and writes presence, mirrors every move a machine owes (a build ask waiting for a node, // 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. @@ -179,6 +179,17 @@ func follow(ctx context.Context, w jetstream.KeyWatcher, each func(jetstream.Key } } +// stepTaken is a change of a build module's steps: a step key written for an ask is that ask taken by a node. The node's +// `paused.` key and a key that is no step key are left. +func stepTaken(w *watch.Watchman, language string, e jetstream.KeyValueEntry) { + if opOf(e) != watch.Put { + return + } + if _, id, err := delivery.ParseStepKey(e.Key()); err == nil { + w.Taken(language, id) + } +} + // run is the program: it returns when ctx ends or the bus cannot be used. func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error { logger := log.New(logOut, "mesh-delivery-watchman: ", log.LstdFlags) @@ -223,12 +234,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } return kv, nil } - var kvs struct{ presence, pulse, stuck, checks jetstream.KeyValue } + var kvs struct{ presence, pulse, stuck jetstream.KeyValue } for _, b := range []struct { name string into *jetstream.KeyValue - }{{delivery.PresenceBucket, &kvs.presence}, {delivery.PulseBucket, &kvs.pulse}, {delivery.StuckBucket, &kvs.stuck}, - {watch.ChecksBucket, &kvs.checks}} { + }{{delivery.PresenceBucket, &kvs.presence}, {delivery.PulseBucket, &kvs.pulse}, {delivery.StuckBucket, &kvs.stuck}} { if *b.into, err = bucket(b.name); err != nil { return err } @@ -275,11 +285,22 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro 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)) - }) - if err != nil { - return err + // The build seats' steps: a step key `.` written is the ask taken (issue 1887 A8b). A seat whose build module + // keeps no steps bucket (yet) is said and left; the asks of its stream are mirrored all the same. + taken := map[string]jetstream.KeyWatcher{} + for _, lang := range delivery.AskLanguages { + module, _ := delivery.BuildModuleOf(lang) + kv, err := js.KeyValue(ctx, delivery.StepsBucket(module)) + if err != nil { + logger.Printf("the steps of %s cannot be read (%v): its asks are mirrored and never seen taken", module, err) + continue + } + lang := lang + sw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { stepTaken(w, lang, e) }) + if err != nil { + return err + } + taken[lang] = sw } // The events first, then the one request of the open connections, so no connection falls between them. @@ -297,6 +318,23 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } } } + // The asks put on the build seats' streams: a plain subscription hears each as it is published, as the stream captures it. + for _, lang := range delivery.AskLanguages { + for _, kind := range delivery.AskKinds { + subject, _ := delivery.AskSubject(lang, kind) + lang := lang + if _, err := conn.Subscribe(subject, func(m *nats.Msg) { + a, err := delivery.DecodeStepAsk(m.Data) + if err != nil { + logger.Printf("an ask on %s is not readable: %v", m.Subject, err) + return + } + w.Asked(lang, a.ID, a.At) + }); err != nil { + return err + } + } + } for subject, open := range map[string]bool{connectSubject: true, disconnectSubject: false} { if _, err := conn.Subscribe(subject, onEvent(open)); err != nil { return err @@ -317,13 +355,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } w.Reconcile() - errs := make(chan error, 4) + errs := make(chan error, 8) go func() { errs <- follow(ctx, pulses, func(e jetstream.KeyValueEntry) { w.Pulse(e.Key(), opOf(e), e.Created()) }) }() - go func() { - errs <- follow(ctx, checks, func(e jetstream.KeyValueEntry) { w.Check(e.Key(), e.Value(), e.Created(), opOf(e)) }) - }() + for lang, sw := range taken { + lang, sw := lang, sw + go func() { errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) { stepTaken(w, lang, e) }) }() + } go func() { for { select { diff --git a/modules/mesh-delivery-watchman/internal/watch/mirror.go b/modules/mesh-delivery-watchman/internal/watch/mirror.go index 9e6a767..8297755 100644 --- a/modules/mesh-delivery-watchman/internal/watch/mirror.go +++ b/modules/mesh-delivery-watchman/internal/watch/mirror.go @@ -1,42 +1,42 @@ package watch import ( - "encoding/json" + "git.novox.be/novox/mesh-sdk/go/delivery" "time" ) -// ChecksBucket is the bucket of the checks the builders are running (mesh-controller internal/broker -// CheckRunningBucket). -const ChecksBucket = "mesh-controller_checks-running" +// A waiting ask is a move (novox/hq ADR 0525 §4, issue 1887 A8b): a message on a build seat's stream (mesh-sdk go/delivery +// AskStream) that a node of the seat must pull. Its pulse is the pulse of the bucket named as the stream, under the ask's id; +// the ask put on the seat is the move beginning at its time, and the ask taken (its step key `.` written in +// the build module's steps, delivery.StepsBucket) is the move done. An ask no node takes within the bound is stuck, as any +// move. The watchman reads no bucket of the controller for it. -// 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"` -} +// AskedKey names a waiting ask for the pulse and stuck keys: the seat's stream and the ask's id. +func askedBucket(language string) (string, error) { return delivery.AskStream(language) } -// 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() - switch op { - case Expired: - return - case Delete: - w.moved(ChecksBucket, key) +// Asked is an ask of a build seat of language put on its stream at time at (the ask's own time). +func (w *Watchman) Asked(language, id string, at time.Time) { + bucket, err := askedBucket(language) + if err != nil { + w.say("no pulse for the ask %s: %v", id, err) return } - var rec checkRecord - _ = json.Unmarshal(value, &rec) + w.mu.Lock() + defer w.mu.Unlock() if at.IsZero() { at = w.now() } - if _, stalled := w.ttl(w.now().Sub(at)); !stalled { - w.stuckMoved(ChecksBucket, key) + w.mirror(bucket, id, describe("the ask %s to be taken by a node", id), "", at, 0) +} + +// Taken is the ask taken by a node: its step key written. The pulse is deleted, so it never expires, and a stuck key it left is +// written as moved. +func (w *Watchman) Taken(language, id string) { + bucket, err := askedBucket(language) + if err != nil { + return } - w.mirror(ChecksBucket, key, describe("the check of %s to end", key), rec.Head, at, 0) + w.mu.Lock() + defer w.mu.Unlock() + w.moved(bucket, id) } diff --git a/modules/mesh-delivery-watchman/internal/watch/watch_test.go b/modules/mesh-delivery-watchman/internal/watch/watch_test.go index 4dd5d3f..1c8a6be 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch_test.go @@ -74,7 +74,7 @@ func checkValue(started time.Time, head string) []byte { } const ( - checksKey = "mesh-controller_checks-running" + checksKey = "SEAT_NODE_BUILD_GO" presenceNS = delivery.PresenceBucket ) @@ -207,13 +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.Check("o.r.7", checkValue(t0, "abc"), t0.Add(10*time.Second), Put) + w.Asked("go", "askid1", t0.Add(10*time.Second)) 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, checksKey, "o.r.7"), pulseKeyOf(t, presenceNS, "ace")} { + for _, k := range []string{pulseKeyOf(t, checksKey, "askid1"), pulseKeyOf(t, presenceNS, "ace")} { if _, ok := o.pulses[k]; !ok { t.Errorf("no pulse %s; have %v", k, o.pulses) } @@ -222,11 +222,11 @@ func TestRebuildPulsesFromCurrentKeys(t *testing.T) { // 1135/AC2 func TestReconcileDropsPulsesNothingMirrorsAnyMore(t *testing.T) { // 1135/AC2 w, o, _ := setup() - stale := pulseKeyOf(t, checksKey, "o.r.gone") - live := pulseKeyOf(t, checksKey, "o.r.7") + stale := pulseKeyOf(t, checksKey, "gone") + live := pulseKeyOf(t, checksKey, "askid1") w.SeedPulse(stale) w.SeedPulse(live) - w.Check("o.r.7", checkValue(t0, "abc"), t0, Put) + w.Asked("go", "askid1", t0) w.Reconcile() if !slices.Equal(o.pulseDeletes, []string{stale}) { t.Fatalf("deleted = %v, want only %s", o.pulseDeletes, stale) @@ -239,12 +239,12 @@ 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.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") + w.Asked("go", "askid1", t0) // the key found at start, last renewed 30s ago + k := pulseKeyOf(t, checksKey, "askid1") if o.ttls[k] != 90*time.Second { t.Fatalf("ttl = %v, want 90s (last renewed 30s ago)", o.ttls[k]) } - if p := o.pulses[k]; p.Bucket != checksKey || p.Key != "o.r.7" { + if p := o.pulses[k]; p.Bucket != checksKey || p.Key != "askid1" { t.Fatalf("pulse = %+v", p) } w.Connect("node.ace") @@ -257,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.Check("o.r.7", checkValue(t0, "abc"), t0, Put) + w.Asked("go", "askid1", t0) if len(o.pulses) != 0 { t.Fatalf("a pulse was written for a heartbeat 120s old: %v", o.pulses) } - s, ok := o.stuck[stuckKeyOf(t, checksKey, "o.r.7")] + s, ok := o.stuck[stuckKeyOf(t, checksKey, "askid1")] if !ok { t.Fatalf("no stuck key; have %v", o.stuck) } - if s.Moved || !s.Since.Equal(t0) || s.Waited != "the check of o.r.7 to end" { + if s.Moved || !s.Since.Equal(t0) || s.Waited != "the ask askid1 to be taken by a node" { t.Fatalf("stuck = %+v", s) } } @@ -273,13 +273,13 @@ func TestAMoveAtTheBoundIsStuckAtOnce(t *testing.T) { // 1135/AC3 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.Asked("go", "askid1", t0) 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, checksKey, "askid1"): "the ask askid1 to be taken by a node", 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", } @@ -292,14 +292,14 @@ 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) + k := pulseKeyOf(t, checksKey, "askid1") + w.Asked("go", "askid1", t0) 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 + w.Asked("go", "askid1", c.t) // 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]) } @@ -308,7 +308,7 @@ func TestEveryHeartbeatRenewsThePulse(t *testing.T) { // 1135/AC3 } // The same revision heard twice changes nothing. o.ttls[k] = 0 - w.Check("o.r.7", checkValue(t0, "abc"), c.t, Put) + w.Asked("go", "askid1", c.t) if o.ttls[k] != 0 { t.Fatal("one revision rewrote the pulse twice") } @@ -317,10 +317,10 @@ func TestEveryHeartbeatRenewsThePulse(t *testing.T) { // 1135/AC3 func TestMirroredMovesDeleteThePulse(t *testing.T) { // 1135/AC4 w, o, c := setup() c.t = t0.Add(10 * time.Second) - w.Check("o.r.7", checkValue(t0, "abc"), t0, Put) + w.Asked("go", "askid1", t0) w.Connect("node.ace") w.Disconnect("node.ace") - w.Check("o.r.7", nil, c.t, Delete) // the check ended + w.Taken("go", "askid1") // the check ended w.Connect("node.ace") if len(o.pulses) != 0 { t.Fatalf("pulses left: %v", o.pulses) @@ -335,12 +335,12 @@ func TestMirroredMovesDeleteThePulse(t *testing.T) { // 1135/AC4 func TestACheckWhoseBuilderStopsIsStuckFromItsLastHeartbeat(t *testing.T) { // 1135/AC4 w, o, c := setup() - w.Check("o.r.7", checkValue(t0, "abc"), t0, Put) + w.Asked("go", "askid1", t0) c.t = t0.Add(50 * time.Second) - w.Check("o.r.7", checkValue(t0, "abc"), c.t, Put) // the last heartbeat + w.Asked("go", "askid1", c.t) // 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")] + w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t) + s := o.stuck[stuckKeyOf(t, checksKey, "askid1")] if !s.Since.Equal(t0.Add(50 * time.Second)) { t.Fatalf("stuck = %+v, want since the last heartbeat", s) } @@ -348,35 +348,35 @@ func TestACheckWhoseBuilderStopsIsStuckFromItsLastHeartbeat(t *testing.T) { // 1 func TestAStuckKeyIsWrittenMovedNeverDeleted(t *testing.T) { // 1135/AC4 w, o, c := setup() - w.Check("o.r.7", checkValue(t0, "abc"), t0, Put) - pk := pulseKeyOf(t, checksKey, "o.r.7") + w.Asked("go", "askid1", t0) + pk := pulseKeyOf(t, checksKey, "askid1") c.t = t0.Add(120 * time.Second) w.Pulse(pk, Expired, c.t) - sk := stuckKeyOf(t, checksKey, "o.r.7") + sk := stuckKeyOf(t, checksKey, "askid1") if o.stuck[sk].Moved { t.Fatal("stuck written as moved at expiry") } c.t = t0.Add(300 * time.Second) - w.Check("o.r.7", nil, c.t, Delete) + w.Taken("go", "askid1") s := o.stuck[sk] - if !s.Moved || !s.Since.Equal(t0) || s.Waited != "the check of o.r.7 to end" { + if !s.Moved || !s.Since.Equal(t0) || s.Waited != "the ask askid1 to be taken by a node" { t.Fatalf("stuck after the move = %+v, want the same key moved", s) } n := o.stuckWrites - w.Check("o.r.7", nil, c.t, Delete) + w.Taken("go", "askid1") if o.stuckWrites != n { t.Fatal("a stuck key already moved was written again") } } -func TestAHeartbeatAfterStuckMovesIt(t *testing.T) { // 1135/AC4 +func TestAnAskTakenAfterStuckMovesIt(t *testing.T) { // 1135/AC4 w, o, c := setup() - w.Check("o.r.7", checkValue(t0, "abc"), t0, Put) + w.Asked("go", "askid1", t0) 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") + w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t) + w.Taken("go", "askid1") // a node took it at last + if !o.stuck[stuckKeyOf(t, checksKey, "askid1")].Moved { + t.Fatal("an ask taken after it was stuck is still stuck") } } @@ -398,23 +398,23 @@ func TestPresenceStuckThenBackOnline(t *testing.T) { // 1135/AC4 func TestExpiryMarkerWritesStuck(t *testing.T) { // 1135/AC5 w, o, c := setup() - w.Check("o.r.7", checkValue(t0, "abc"), t0, Put) + w.Asked("go", "askid1", t0) 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")] + w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t) + s, ok := o.stuck[stuckKeyOf(t, checksKey, "askid1")] 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 { + if s.Bucket != checksKey || s.Key != "askid1" || s.Waited != "the ask askid1 to be taken by a node" || !s.Since.Equal(t0) || s.Moved { t.Fatalf("stuck = %+v", s) } } func TestOwnDeleteOfAPulseWritesNothing(t *testing.T) { // 1135/AC5 w, o, c := setup() - 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.Asked("go", "askid1", t0) + pk := pulseKeyOf(t, checksKey, "askid1") + w.Taken("go", "askid1") 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) @@ -427,13 +427,13 @@ func TestTheBoundCanBeShortenedForATest(t *testing.T) { // 1135/AC8 w, o, c := setup() w.SetBound(2 * time.Second) c.t = t0.Add(500 * 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 { + w.Asked("go", "askid1", t0) + if got := o.ttls[pulseKeyOf(t, checksKey, "askid1")]; got != 1500*time.Millisecond { t.Fatalf("ttl = %v, want 1.5s", got) } c.t = t0.Add(3 * time.Second) - w.Check("o.r.8", checkValue(t0, "abc"), t0, Put) - if _, ok := o.stuck[stuckKeyOf(t, checksKey, "o.r.8")]; !ok { + w.Asked("go", "askid2", t0) + if _, ok := o.stuck[stuckKeyOf(t, checksKey, "askid2")]; !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 4c2968d..d0344b9 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.checks-running"}) { + if !reflect.DeepEqual(reads, []string{"mesh-build-go.steps", "mesh-build-image.steps", "mesh-build-node.steps"}) { 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 2b723d2..8c4c713 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/modules/mesh-delivery-watchman/module.json @@ -3,7 +3,9 @@ "version": "1", "slug": "watchman", "reads": [ - "mesh-controller.checks-running" + "mesh-build-go.steps", + "mesh-build-image.steps", + "mesh-build-node.steps" ], "state": [ { "name": "presence", "history": 20 },