diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman-tools/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman-tools/main.go index b3a0c2cf..9e3cb313 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman-tools/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman-tools/main.go @@ -10,8 +10,10 @@ import ( "os" "sort" "strings" + "time" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-watchman/internal/socket" "mesh-delivery-watchman/internal/watch" @@ -28,18 +30,22 @@ const ( VerbDeadLetters = "dead-letters" VerbResync = "resync" ToolFindings = "watchman_findings" + // ToolFirstViews and ToolFirstViewAgain are where the watchman's first views of the assigner's wants and the indexer's + // modules stand, and asking one again by hand (novox/hq ADR 0615, DLV-M5). + ToolFirstViews = "watchman_first_views" + ToolFirstViewAgain = "watchman_first_view_again" ) // Verbs are the seat's verbs, in the manifest's order. var Verbs = []string{VerbSignals, VerbStuck, VerbPresence, VerbDeadLetters, VerbResync} // ModuleTools are the module's own tools, in the manifest's order. -var ModuleTools = []string{ToolFindings} +var ModuleTools = []string{ToolFindings, ToolFirstViews, ToolFirstViewAgain} // Reads names, per tool, the states it reads whole; Acts, per tool, the program's act it starts. The clause-12 test holds // that every state the manifest keeps is read by one, and every act the program takes by hand is started by one. var ( - Reads = map[string][]string{VerbSignals: {"pulse"}, VerbStuck: {"stuck"}, VerbPresence: {"presence"}, ToolFindings: {"findings"}} + Reads = map[string][]string{VerbSignals: {"pulse"}, VerbStuck: {"stuck"}, VerbPresence: {"presence"}, ToolFindings: {"findings"}, ToolFirstViews: {views.StatusState}} Acts = map[string][]string{VerbDeadLetters: {socket.AskDeadLetters}, VerbResync: {socket.AskResync}} ) @@ -124,7 +130,7 @@ func Tools(states func(string) state, ask func(socket.Ask) (json.RawMessage, err } resync := func(map[string]any) (any, error) { return ask(socket.Ask{Ask: socket.AskResync}) } verb := func(v string) string { return Seat + "." + v } - return []stdio.Tool{ + return append([]stdio.Tool{ {Name: verb(VerbSignals), Input: keyArgs, Run: signals, Description: "The signals table's rows the watchman holds (S1, S8, S9, S10, S11, S21 and the haves' health), each a pulse key " + "whose time to live is its bound and whose expiry is its finding (novox/hq ADR 0564 §4), and every pulse the watchman " + @@ -157,7 +163,7 @@ func Tools(states func(string) state, ask func(socket.Ask) (json.RawMessage, err Description: "The watchman's findings (novox/hq ADR 0564 §2): each thing it observes wrong now, .., with its " + "severity, subject, evidence and the row that observed it; the condition keeper raises and clears the conditions from " + "them. Give key for one, prefix to narrow. Read-only."}, - } + }, views.StatusTools(watch.Observer, ToolFirstViews, ToolFirstViewAgain, stdio.State(views.StatusState), time.Now)...) } func main() { 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 5f94991b..0077f109 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,6 +17,7 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-delivery/sdk/firstview/firstviewtest" "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" @@ -57,6 +58,14 @@ accounts { } ` +// serveOwners serves the assigner's wants and the indexer's modules as their owners do, by their read tools and views +// families (DLV-M5, ADR 0615): the watchman (n1) follows both, never their buckets. +func serveOwners(t *testing.T, ctx context.Context, js jetstream.JetStream, nc *nats.Conn) { + t.Helper() + firstviewtest.Serve(t, ctx, js, nc, delivery.AssignerModule, wantsState, "n1_"+delivery.WatchmanModule) + firstviewtest.Serve(t, ctx, js, nc, delivery.IndexerModule, modulesState, "n1_"+delivery.WatchmanModule) +} + // TestWatchmanWhole starts the program the way the node does (its entry, run, with its environment and a broker file) // against an embedded bus with the system exports and JetStream, and proves the four things the issue names. func TestWatchmanWhole(t *testing.T) { // 1135/AC8 @@ -109,6 +118,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 {Bucket: delivery.AssignerWantsBucket, History: 20}, {Bucket: delivery.HaveBucket, History: 20}, {Bucket: watch.FindingsBucket, History: 5}, + {Bucket: firstViewsBucket, History: 1}, } { kv, err := js.CreateKeyValue(ctx, c) if err != nil { @@ -116,6 +126,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 } kvs[c.Bucket] = kv } + serveOwners(t, ctx, js, ctl) // The controller's consumer on DEAD_LETTER_NOTICES, the dead letters' streams and a source stream with one message a // consumer will give up on (1952/WM5); no lease bucket at all (issue 1126). @@ -355,6 +366,7 @@ func TestAnAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909 {Bucket: delivery.AssignerWantsBucket, History: 20}, {Bucket: delivery.HaveBucket, History: 20}, {Bucket: watch.FindingsBucket, History: 5}, + {Bucket: firstViewsBucket, History: 1}, } { kv, err := js.CreateKeyValue(ctx, c) if err != nil { @@ -362,6 +374,7 @@ func TestAnAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909 } kvs[c.Bucket] = kv } + serveOwners(t, ctx, js, ctl) const id = "m0000009ffffff" body, err := delivery.EncodeStep(delivery.Step{ID: id, Tool: "build_go_test", Node: "n1", State: delivery.StepRunning, Started: time.Now().UTC()}) if err != nil { @@ -440,6 +453,7 @@ func TestAWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) { {Bucket: delivery.AssignerWantsBucket, History: 20}, {Bucket: delivery.HaveBucket, History: 20}, {Bucket: watch.FindingsBucket, History: 5}, + {Bucket: firstViewsBucket, History: 1}, } { kv, err := js.CreateKeyValue(ctx, c) if err != nil { @@ -447,6 +461,7 @@ func TestAWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) { } kvs[c.Bucket] = kv } + serveOwners(t, ctx, js, ctl) fpNew := delivery.SourcePrefix2 + strings.Repeat("b", 64) fpOld := delivery.SourcePrefix2 + strings.Repeat("a", 64) putWant := func(node, module, fp string) { 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 8ec5142d..183e5fd8 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -18,6 +18,9 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/views" + + "git.novox.be/novox/mesh-delivery/sdk/firstview" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -122,6 +125,26 @@ func settle(ctx context.Context, kv jetstream.KeyValue, each func(jetstream.KeyV } } +// readLate reads an owner's state the watchman follows by its first view (pic), on a goroutine of its own once that is +// answered, so an owner that is down never holds the start: every key it holds now, then the start's end for its bucket +// (watch.ReadLate), then each change until ctx ends. +func readLate(ctx context.Context, w *watch.Watchman, pic *firstview.Picture, bucket string, each func(jetstream.KeyValueEntry)) error { + select { + case <-ctx.Done(): + return nil + case <-pic.Follower().Ready(): + } + sw, err := settle(ctx, pic, each) + if err != nil { + if ctx.Err() != nil { + return nil + } + return err + } + w.ReadLate(bucket) + return follow(ctx, sw, each) +} + // follow hands each change of a watcher to each until ctx ends; a watch that ends is an error, and the process // manager starts the program again. func follow(ctx context.Context, w jetstream.KeyWatcher, each func(jetstream.KeyValueEntry)) error { @@ -147,6 +170,25 @@ func stepTaken(w *watch.Watchman, language string, e jetstream.KeyValueEntry) { w.Step(language, e.Key(), e.Value(), opOf(e)) } +// createdOr is when an entry was written, or at when its picture does not say: the first view of an owner's state names +// no key's time (novox/hq ADR 0615), so a want that stood before the start waits from the start. +func createdOr(e jetstream.KeyValueEntry, at time.Time) time.Time { + if t := e.Created(); !t.IsZero() { + return t + } + return at +} + +// The watchman's own states of its first views (views.StatusState); its findings of a first view unanswered past its +// bound go to its findings bucket with the others (ADR 0615 §6). The owners' states it follows are the assigner's wants +// and the indexer's modules (DLV-M5). +var firstViewsBucket = delivery.WatchmanModule + "_" + views.StatusState + +const ( + wantsState = "wants" + modulesState = "modules" +) + // 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) @@ -168,9 +210,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } defer conn.Close() // Its own presence, pulses and stuck rows are what it keeps: each stops the start, naming its grant. The assigner's wants - // and the node-engine's haves only make the pulses of a move: without one the watchman watches presence and the build - // seats' steps, and says that moves are not watched (issue 1945). - var kvs struct{ presence, pulse, stuck, findings, wants, haves jetstream.KeyValue } + // and the node-engine's haves only make the pulses of a move: without the haves the watchman watches presence and the + // build seats' steps, and says that moves are not watched (issue 1945). + var kvs struct{ presence, pulse, stuck, findings, haves jetstream.KeyValue } for _, b := range []struct { name string into *jetstream.KeyValue @@ -179,14 +221,26 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } } - kvs.wants = process.OptionalBucket(ctx, js, delivery.AssignerWantsBucket, "a want makes no pulse, so a move that stops is not seen, until the watchman starts again", logger.Printf) kvs.haves = process.OptionalBucket(ctx, js, delivery.HaveBucket, "a want makes no pulse, so a move that stops is not seen, until the watchman starts again", logger.Printf) - if kvs.wants == nil || kvs.haves == nil { - kvs.wants, kvs.haves = nil, nil // a want is a pulse only against the have that ends it + // The assigner's wants and the indexer's modules are their owners' states, read by the first view of their read tools and + // then their views families, never their buckets (ADR 0615 §1, DLV-M5). An owner that is down never holds the start: each + // is read on a goroutine once its first view is answered, and until then the unanswered first view is the watchman's + // finding. + calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) + if err != nil { + return fmt.Errorf("the watchman's connection for its tool calls: %w", err) } + defer calls.Close() + reader := &firstview.Reader{Module: delivery.WatchmanModule, Call: calls.Call, Logf: logger.Printf, + Findings: func() jetstream.KeyValue { return kvs.findings }} + wants := reader.Follow(delivery.AssignerModule, wantsState) + modules := reader.Follow(delivery.IndexerModule, modulesState) w := watch.New(&kvOut{ctx: ctx, presence: kvs.presence, pulse: kvs.pulse, stuk: kvs.stuck, findings: kvs.findings}, time.Now) w.Logf(logger.Printf) + // What the wants and the modules hold is ended by their own late read (watch.ReadLate), never by the start's end. + w.Awaits(delivery.AssignerWantsBucket) + w.Awaits(delivery.ModulesBucket) // A test shortens how long a move may take; the mesh never sets it. if v := strings.TrimSpace(getenv("MESH_DELIVERY_PULSE_BOUND")); v != "" { d, err := time.ParseDuration(v) @@ -232,18 +286,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } - // What the node agents run, then the assigner's wants (issue 1906, ADR 0539 §19): a want is a pulse owed by its node's agent - // until the have names its fingerprint, so the haves are read first and a want already applied makes no pulse. - var haveSeen, wantSeen jetstream.KeyWatcher + // What the node agents run (issue 1906, ADR 0539 §19): a want is a pulse owed by its node's agent until the have names its + // fingerprint, so the haves are read at the start and the wants after them, once their first view is answered. + var haveSeen jetstream.KeyWatcher if kvs.haves != nil { if haveSeen, err = settle(ctx, kvs.haves, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) }); err != nil { process.NotWatched(logger.Printf, delivery.HaveBucket, err, "moves are not watched until the watchman starts again") - } else if wantSeen, err = settle(ctx, kvs.wants, func(e jetstream.KeyValueEntry) { w.Want(e.Key(), e.Value(), opOf(e), e.Created()) }); err != nil { - process.NotWatched(logger.Printf, delivery.AssignerWantsBucket, err, "moves are not watched until the watchman starts again") - haveSeen.Stop() haveSeen = nil - } else { - w.ReadWhole(delivery.AssignerWantsBucket) } } // The builder's builds (ADR 0603 §8): a key queued behind a package is a wait, mirrored from the time it was written. @@ -358,17 +407,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro go func() { _ = follow(ctx, hw, healthEach) }() }) } - // The indexer's modules (ADR 0578 §1): a module whose value carries a refusal of its manifest is a finding; read - // before the findings are reconciled, each change heard on the watch. - verdictEach := func(e jetstream.KeyValueEntry) { w.ModuleVerdict(e.Key(), e.Value(), opOf(e)) } - var verdictWatch jetstream.KeyWatcher - if kv, err := js.KeyValue(ctx, delivery.ModulesBucket); err == nil { - if verdictWatch, err = settle(ctx, kv, verdictEach); err != nil { - process.NotWatched(logger.Printf, delivery.ModulesBucket, err, "no refused manifest is said until the watchman starts again") - } - } else { - process.NotWatched(logger.Printf, delivery.ModulesBucket, err, "no refused manifest is said until the watchman starts again") - } dead := &deadKeeper{ctx: ctx, js: js, w: w, logf: logger.Printf} w.SetRecount(dead.count) dead.count() @@ -384,8 +422,20 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if healthWatch != nil { go func() { errs <- follow(ctx, healthWatch, healthEach) }() } - if verdictWatch != nil { - go func() { errs <- follow(ctx, verdictWatch, verdictEach) }() + // The first views of the wants and the modules, asked now and read each on its own goroutine once answered (ADR 0615). + reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "watchman_first_views reads nothing and watchman_first_view_again is not heard", logger.Printf)) + // The indexer's modules (ADR 0578 §1): a module whose value carries a refusal of its manifest is a finding. + verdictEach := func(e jetstream.KeyValueEntry) { w.ModuleVerdict(e.Key(), e.Value(), opOf(e)) } + go func() { errs <- readLate(ctx, w, modules, delivery.ModulesBucket, verdictEach) }() + if haveSeen != nil { + go func() { + errs <- follow(ctx, haveSeen, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) }) + }() + go func() { + errs <- readLate(ctx, w, wants, delivery.AssignerWantsBucket, func(e jetstream.KeyValueEntry) { + w.Want(e.Key(), e.Value(), opOf(e), createdOr(e, time.Now())) + }) + }() } if provWatch != nil { go func() { @@ -402,14 +452,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()) }) }() - if haveSeen != nil && wantSeen != nil { - go func() { - errs <- follow(ctx, haveSeen, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) }) - }() - go func() { - errs <- follow(ctx, wantSeen, func(e jetstream.KeyValueEntry) { w.Want(e.Key(), e.Value(), opOf(e), e.Created()) }) - }() - } if buildSeen != nil { go func() { errs <- follow(ctx, buildSeen, func(e jetstream.KeyValueEntry) { w.Build(e.Key(), e.Value(), opOf(e), e.Created()) }) diff --git a/modules/mesh-delivery-watchman/go.mod b/modules/mesh-delivery-watchman/go.mod index 9a0d188f..e404582a 100644 --- a/modules/mesh-delivery-watchman/go.mod +++ b/modules/mesh-delivery-watchman/go.mod @@ -22,3 +22,7 @@ require ( ) replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun => ../mesh-delivery-builder/buildrun + +require git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000 + +replace git.novox.be/novox/mesh-delivery/sdk => ../../sdk diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index 0e8b7925..35908f75 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch.go @@ -101,6 +101,10 @@ type Watchman struct { standing map[string]map[string]bool readWhole map[string]bool reconciled bool + // awaited is each bucket read only once its owner's first view is answered, after the start's end (Awaits): its seeded + // pulses, stuck keys and findings wait for its own end (ReadLate), so a restart while its owner is down neither drops + // nor raises again what it holds. + awaited map[string]bool } // New is a watchman writing to out, reading the time from now. @@ -111,7 +115,7 @@ func New(out Out, now func() time.Time) *Watchman { haveState: map[string]delivery.Have{}, unhealthy: map[string]int{}, advised: map[string]int{}, advisedSince: map[string]time.Time{}, control: map[string]bool{}, failedSince: map[string]time.Time{}, provFailed: map[string]map[string]ProvReport{}, closures: map[string]map[string]time.Time{}, waiting: map[string]map[string]string{}, health: map[string]delivery.NodeHealth{}, tunnels: map[string]string{}, - standing: map[string]map[string]bool{}, readWhole: map[string]bool{}} + standing: map[string]map[string]bool{}, readWhole: map[string]bool{}, awaited: map[string]bool{}} } // Logf sets where a write that failed is said; it is silent until set. @@ -294,7 +298,7 @@ func (w *Watchman) Reconcile() { w.mu.Lock() defer w.mu.Unlock() for pk, p := range w.pulses { - if p.seeded { + if p.seeded && !w.awaited[p.bucket] { w.deletePulse(pk) } } @@ -319,7 +323,7 @@ func (w *Watchman) ReadWhole(bucket string) { // stands records a mirrored key read standing during the start. func (w *Watchman) stands(bucket, key string) { - if w.reconciled { + if w.reconciled && !w.awaited[bucket] { return } if w.standing[bucket] == nil { @@ -330,15 +334,61 @@ func (w *Watchman) stands(bucket, key string) { func (w *Watchman) closeGone() { for sk, s := range w.stuck { - if s.Moved || !w.readWhole[s.Bucket] || w.standing[s.Bucket][s.Key] { + if s.Moved || w.awaited[s.Bucket] || !w.readWhole[s.Bucket] || w.standing[s.Bucket][s.Key] { continue } - s.Moved = true - w.say("the stuck key %s is closed: its key %s is no longer in %s (it aged out of the bucket or was deleted while nobody heard), so nothing waits for it since %s", - sk, s.Key, s.Bucket, s.Since.UTC().Format(time.RFC3339)) - w.putStuck(sk, s) + w.closeStuck(sk, s) } - w.standing, w.readWhole, w.reconciled = map[string]map[string]bool{}, map[string]bool{}, true + standing := map[string]map[string]bool{} + for b := range w.awaited { + standing[b] = w.standing[b] + } + w.standing, w.readWhole, w.reconciled = standing, map[string]bool{}, true +} + +func (w *Watchman) closeStuck(sk string, s delivery.Stuck) { + s.Moved = true + w.say("the stuck key %s is closed: its key %s is no longer in %s (it aged out of the bucket or was deleted while nobody heard), so nothing waits for it since %s", + sk, s.Key, s.Bucket, s.Since.UTC().Format(time.RFC3339)) + w.putStuck(sk, s) +} + +// Awaits says bucket (an owner's state the watchman reads through its first view: the assigner's wants, the indexer's +// modules; novox/hq ADR 0615) is read only once that first view is answered, which may be after the start's end: Reconcile +// leaves its seeded pulses, its stuck keys and the seeded findings it is the source of to ReadLate. Called before Reconcile. +func (w *Watchman) Awaits(bucket string) { + w.mu.Lock() + defer w.mu.Unlock() + w.awaited[bucket] = true +} + +// ReadLate ends the start for an awaited bucket, called once every key it holds was read standing: as Reconcile does for +// the others, a seeded pulse of it no key made again is deleted, a stuck key of it whose key is gone is closed, and a +// seeded finding it is the source of and nothing observed again is deleted. +func (w *Watchman) ReadLate(bucket string) { + w.mu.Lock() + defer w.mu.Unlock() + if !w.awaited[bucket] { + return + } + for pk, p := range w.pulses { + if p.seeded && p.bucket == bucket { + w.deletePulse(pk) + } + } + for sk, s := range w.stuck { + if !s.Moved && s.Bucket == bucket && !w.standing[bucket][s.Key] { + w.closeStuck(sk, s) + } + } + for key := range w.seededFindings { + if f, err := DecodeFinding(w.findings[key]); err == nil && f.Source == bucket { + delete(w.seededFindings, key) + w.unfind(key) + } + } + delete(w.awaited, bucket) + delete(w.standing, bucket) } // reconcileFindings ends the start for the findings: one read at the start that what the watchman read since did not @@ -350,6 +400,9 @@ func (w *Watchman) reconcileFindings() { kept[due.Finding.Key()] = true } for key := range w.seededFindings { + if f, err := DecodeFinding(w.findings[key]); err == nil && w.awaited[f.Source] { + continue // ended by ReadLate + } delete(w.seededFindings, key) if kept[key] || signalKind(w.findings[key]) { continue diff --git a/modules/mesh-delivery-watchman/internal/watch/watch_test.go b/modules/mesh-delivery-watchman/internal/watch/watch_test.go index ed96fa6a..29c2759a 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch_test.go @@ -668,12 +668,17 @@ func TestAStuckKeyWhoseKeyIsGoneIsClosedAtTheStart(t *testing.T) { standing func(w *Watchman) // what the start reads standing readWhole bool closed bool + late bool // read after the start's end (Awaits, ReadLate) }{ - {"an ask's step aged out of the steps bucket", checksKey, "askold1", func(*Watchman) {}, true, true}, - {"an ask's step still running", checksKey, "askold1", func(w *Watchman) { w.Step("go", "n1.askold1", runningStep("askold1", t0), Put) }, true, false}, - {"the steps bucket could not be read", checksKey, "askold1", func(*Watchman) {}, false, false}, - {"a want deleted while nobody heard", delivery.AssignerWantsBucket, wantKey, func(*Watchman) {}, true, true}, - {"a want still standing", delivery.AssignerWantsBucket, wantKey, func(w *Watchman) { w.Want(wantKey, wantValue(t, fpNew, false), Put, t0) }, true, false}, + {"an ask's step aged out of the steps bucket", checksKey, "askold1", func(*Watchman) {}, true, true, false}, + {"an ask's step still running", checksKey, "askold1", func(w *Watchman) { w.Step("go", "n1.askold1", runningStep("askold1", t0), Put) }, true, false, false}, + {"the steps bucket could not be read", checksKey, "askold1", func(*Watchman) {}, false, false, false}, + {"a want deleted while nobody heard", delivery.AssignerWantsBucket, wantKey, func(*Watchman) {}, true, true, false}, + {"a want still standing", delivery.AssignerWantsBucket, wantKey, func(w *Watchman) { w.Want(wantKey, wantValue(t, fpNew, false), Put, t0) }, true, false, false}, + // DLV-M5 (ADR 0615): the wants are read from the assigner's first view, which may come after the start's end; what + // they hold waits for it, and is then reconciled as at the start. + {"a want deleted while nobody heard, read from a late first view", delivery.AssignerWantsBucket, wantKey, func(*Watchman) {}, true, true, true}, + {"a want still standing, read from a late first view", delivery.AssignerWantsBucket, wantKey, func(w *Watchman) { w.Want(wantKey, wantValue(t, fpNew, false), Put, t0) }, true, false, true}, } { t.Run(row.name, func(t *testing.T) { w, o, c := setup() @@ -684,11 +689,36 @@ func TestAStuckKeyWhoseKeyIsGoneIsClosedAtTheStart(t *testing.T) { if _, ok := o.findings[FindingKey(ScopeDelivery, sk, KindStuck)]; !ok { t.Fatal("the seeded stuck key raised no finding") } - row.standing(w) - if row.readWhole { - w.ReadWhole(row.bucket) + if row.late { + // A pulse of another key of the bucket, and a finding the bucket is the source of, stood at the start. + pk := pulseKeyOf(t, row.bucket, "want.n9.gone") + w.SeedPulse(pk) + fv, err := EncodeFinding(Finding{Scope: ScopeModule, ID: "gone", Kind: KindManifestRefused, Severity: Warning, Source: row.bucket, Summary: "refused", Since: t0}) + if err != nil { + t.Fatal(err) + } + fk := FindingKey(ScopeModule, "gone", KindManifestRefused) + o.findings[fk], _ = DecodeFinding(fv) + w.SeedFinding(fk, fv) + w.Awaits(row.bucket) + w.Reconcile() + _, open := o.findings[FindingKey(ScopeDelivery, sk, KindStuck)] + _, seeded := o.findings[fk] + if o.stuck[sk].Moved || !open || !seeded || slices.Contains(o.pulseDeletes, pk) { + t.Fatalf("the start's end took what an awaited bucket holds: moved %v, stuck open %v, finding %v, pulses deleted %v", o.stuck[sk].Moved, open, seeded, o.pulseDeletes) + } + row.standing(w) + w.ReadLate(row.bucket) + if _, seeded := o.findings[fk]; seeded || !slices.Contains(o.pulseDeletes, pk) { + t.Errorf("the late read kept the finding (%v) or the pulse (deleted %v) nothing made again", seeded, o.pulseDeletes) + } + } else { + row.standing(w) + if row.readWhole { + w.ReadWhole(row.bucket) + } + w.Reconcile() } - w.Reconcile() _, open := o.findings[FindingKey(ScopeDelivery, sk, KindStuck)] if got := o.stuck[sk].Moved; got != row.closed || open == row.closed { t.Errorf("moved %v, finding open %v; want closed %v", got, open, row.closed) diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go index a577361e..5cb3db48 100644 --- a/modules/mesh-delivery-watchman/manifest_test.go +++ b/modules/mesh-delivery-watchman/manifest_test.go @@ -51,6 +51,8 @@ type manifest struct { Version string `json:"version"` Slug string `json:"slug"` Reads []string `json:"reads"` + Invokes []string `json:"invokes"` + Consumes []string `json:"consumes"` State []state `json:"state"` OwnSecrets map[string]string `json:"own-secrets"` SecretsOwner string `json:"secrets-owner"` @@ -84,24 +86,37 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6 if m.Module != "mesh-delivery-watchman" || m.Moved || m.Slug != "watchman" { t.Errorf("module %q moved %v (it has no template, so it is not moved) slug %q", m.Module, m.Moved, m.Slug) } - want := []state{{Name: "presence", History: 20}, {Name: "pulse", KeyTTL: true}, {Name: "stuck", History: 20}, {Name: "findings", History: 5}} + want := []state{{Name: "presence", History: 20}, {Name: "pulse", KeyTTL: true}, {Name: "stuck", History: 20}, {Name: "findings", History: 5}, {Name: "first-views", History: 1}} if !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } reads := slices.Clone(m.Reads) slices.Sort(reads) if !reflect.DeepEqual(reads, []string{"mesh-build-go.steps", "mesh-build-image.steps", "mesh-build-node.steps", - "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-indexer.modules", "mesh-host.health", "mesh-provisioner.provisions", "node-engine.have"}) { + "mesh-delivery-builder.builds", "mesh-host.health", "mesh-provisioner.provisions", "node-engine.have"}) { t.Errorf("reads = %v", reads) } + // DLV-M5 (ADR 0615 §1): the assigner's wants and the indexer's modules are their owners' states, asked by their read + // tools and followed by their views families, never read. + for _, r := range m.Reads { + if strings.HasPrefix(r, "mesh-delivery-indexer.") || strings.HasPrefix(r, "mesh-delivery-assigner.") { + t.Errorf("reads %s: the indexer's and the assigner's state is read by their tools and views families alone", r) + } + } + if want := []string{"mesh-delivery-assigner.assigner_read_wants", "mesh-delivery-indexer.indexer_read_modules"}; !reflect.DeepEqual(m.Invokes, want) { + t.Errorf("invokes = %v, want %v", m.Invokes, want) + } + if want := []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.views.modules"}; !reflect.DeepEqual(m.Consumes, want) { + t.Errorf("consumes = %v, want %v", m.Consumes, want) + } // 1952/WM1: the watchman holds the seat mesh-watchdogs at mesh scope, its first seat (novox/hq ADR 0564 §1, §4). if !reflect.DeepEqual(m.Claims, []claim{{Name: "mesh-watchdogs", Scope: "mesh", Serves: []string{"signals", "stuck", "presence", "dead-letters", "resync"}}}) { t.Errorf("claims = %+v", m.Claims) } - if !reflect.DeepEqual(m.Tools, []string{"watchman_findings"}) { + if !reflect.DeepEqual(m.Tools, []string{"watchman_findings", "watchman_first_views", "watchman_first_view_again"}) { t.Errorf("tools = %v", m.Tools) } - for _, k := range []string{"emits", "context", "items", "consumes", "provides"} { + for _, k := range []string{"emits", "context", "items", "provides"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the watchman has none", k) } diff --git a/modules/mesh-delivery-watchman/module.json b/modules/mesh-delivery-watchman/module.json index 4c85f5d7..5d0dc5ec 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/modules/mesh-delivery-watchman/module.json @@ -19,21 +19,30 @@ "mesh-build-go.steps", "mesh-build-image.steps", "mesh-build-node.steps", - "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", - "mesh-delivery-indexer.modules", "mesh-host.health", "mesh-provisioner.provisions", "node-engine.have" ], + "invokes": [ + "mesh-delivery-assigner.assigner_read_wants", + "mesh-delivery-indexer.indexer_read_modules" + ], + "consumes": [ + "mesh-delivery-assigner.views.wants", + "mesh-delivery-indexer.views.modules" + ], "tools": [ - "watchman_findings" + "watchman_findings", + "watchman_first_views", + "watchman_first_view_again" ], "state": [ { "name": "presence", "history": 20 }, { "name": "pulse", "key-ttl": true }, { "name": "stuck", "history": 20 }, - { "name": "findings", "history": 5 } + { "name": "findings", "history": 5 }, + { "name": "first-views", "history": 1 } ], "own-secrets": { "broker": "${dir:state}/broker"