From 0f52bb80ad6ebf82409f970318102d87327891b3 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 04:38:38 +0100 Subject: [PATCH] =?UTF-8?q?D11:=20the=20builder=20and=20the=20indexer=20sa?= =?UTF-8?q?y=20a=20manifest-reads=20call=20unanswered=20past=20five=20minu?= =?UTF-8?q?tes,=20its=20machine=20heard,=20as=20call-unanswered=20in=20the?= =?UTF-8?q?ir=20own=20findings=20(issue=201966,=20ADR=200668=20=C2=A73;=20?= =?UTF-8?q?review=20round=202)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both write it through buildrun.ReadsCalls over enginereads Keeper.Unanswered as the assigner does, read the watchman's presence for a machine heard, arm a bound key that expires at the five minutes (no timer) and clear the finding when the node answers or its machine is away. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- modules/mesh-delivery-builder/buildrun/go.mod | 1 + .../buildrun/readscalls.go | 135 ++++++++++++++++++ .../cmd/mesh-delivery-builder/main.go | 11 ++ .../cmd/mesh-delivery-builder/reads.go | 21 ++- .../cmd/mesh-delivery-builder/readscall.go | 63 ++++++++ .../cmd/mesh-delivery-builder/refused_test.go | 62 ++++++++ .../mesh-delivery-builder/manifest_test.go | 7 +- modules/mesh-delivery-builder/module.json | 3 +- .../cmd/mesh-delivery-indexer/main.go | 11 ++ .../cmd/mesh-delivery-indexer/main_test.go | 82 +++++++++++ .../cmd/mesh-delivery-indexer/reads.go | 21 ++- .../cmd/mesh-delivery-indexer/readscall.go | 76 ++++++++++ .../mesh-delivery-indexer/manifest_test.go | 3 +- modules/mesh-delivery-indexer/module.json | 3 +- 14 files changed, 491 insertions(+), 8 deletions(-) create mode 100644 modules/mesh-delivery-builder/buildrun/readscalls.go create mode 100644 modules/mesh-delivery-builder/cmd/mesh-delivery-builder/readscall.go create mode 100644 modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/readscall.go diff --git a/modules/mesh-delivery-builder/buildrun/go.mod b/modules/mesh-delivery-builder/buildrun/go.mod index 712cc087..f344fa0a 100644 --- a/modules/mesh-delivery-builder/buildrun/go.mod +++ b/modules/mesh-delivery-builder/buildrun/go.mod @@ -6,6 +6,7 @@ require ( git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 diff --git a/modules/mesh-delivery-builder/buildrun/readscalls.go b/modules/mesh-delivery-builder/buildrun/readscalls.go new file mode 100644 index 00000000..9af1275b --- /dev/null +++ b/modules/mesh-delivery-builder/buildrun/readscalls.go @@ -0,0 +1,135 @@ +package buildrun + +import ( + "context" + "maps" + "slices" + "strings" + "sync" + "time" + + "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads" + mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" +) + +// ReadsCalls says a node's manifest-reads left unanswered by its engine while its machine is heard (online), the way the +// assigner does (novox/hq ADR 0668 §3), for a part that keeps the call through enginereads.Keeper: past +// meshconditionsclient.CallUnansweredAfter it is the finding call-unanswered in the part's own findings, cleared once the +// node answers or its machine is away, when none is written. Every step is an event (a statement changed, a node's +// presence changed, the counted nodes changed, a bound's expiry); the bound is a key of the part's own state that expires +// at its end (Arm), never a timer. +type ReadsCalls struct { + Keeper *enginereads.Keeper + Presence *PresenceSet + // Nodes are the counted nodes. + Nodes func() []string + // Source is the finding's source, ".call-unanswered"; Needs says the part's tool that steers the call. + Source, Needs string + // Raise writes the finding o in the part's findings; Clear removes the call-unanswered finding of the call id. + Raise func(ctx context.Context, o mcc.Observation) error + Clear func(ctx context.Context, id string) error + // Arm keeps a bound for node ending at ends, whose expiry steps again; nil arms none. + Arm func(ctx context.Context, node string, ends time.Time) error + Now func() time.Time + Logf func(string, ...any) + + mu sync.Mutex + said map[string]time.Time // node -> the since its finding was written for +} + +// PresenceBucket is the watchman's presence state, which the part reads to know a machine is heard. +const PresenceBucket = "mesh-delivery-watchman_presence" + +// ReadsCallBoundPrefix starts the key of a node's bound in the part's bounds state. +const ReadsCallBoundPrefix = "reads-call." + +// ReadsCallOf is the node a bound key is of. +func ReadsCallOf(key string) (string, bool) { return strings.CutPrefix(key, ReadsCallBoundPrefix) } + +func (c *ReadsCalls) now() time.Time { + if c.Now != nil { + return c.Now() + } + return time.Now() +} + +func (c *ReadsCalls) logf(f string, a ...any) { + if c.Logf != nil { + c.Logf(f, a...) + } +} + +func readsCallID(node string) string { return "manifest-reads." + node } + +// Step weighs every counted node: a node whose engine has not answered while its machine is heard is asked again once +// its last ask ended; past the bound it is the finding, before it a bound is armed; one answered or away has none. +func (c *ReadsCalls) Step(ctx context.Context) { + if c == nil || c.Keeper == nil || c.Presence == nil || c.Raise == nil || c.Clear == nil { + return + } + nodes := c.Nodes() + statements := c.Keeper.Statements() + for _, node := range nodes { + if s, ok := statements[node]; ok && !s.Answered && c.Presence.Online(node) && !c.Keeper.Asking(node) { + c.Keeper.AskNode(node, "unanswered since "+s.UnansweredSince.UTC().Format(time.RFC3339)+", its machine heard") + } + } + now := c.now() + raised := map[string]bool{} + for node, since := range c.Keeper.Unanswered(nodes, c.Presence.Online, mcc.CallUnansweredAfter, now) { + o := mcc.CallUnanswered(readsCallID(node), enginereads.Invoke, node, statements[node].Words, c.Source, since) + o.Needs = c.Needs + key := o.Key() + raised[key] = true + c.mu.Lock() + was, said := c.said[node] + c.mu.Unlock() + if said && was.Equal(since) { + continue + } + if err := c.Raise(ctx, o); err != nil { + c.logf("the finding %s is not written: %v", key, err) + continue + } + c.logf("%s on %s is unanswered since %s and its machine is heard: the finding %s", enginereads.Verb, node, since.UTC().Format(time.RFC3339), key) + c.mu.Lock() + if c.said == nil { + c.said = map[string]time.Time{} + } + c.said[node] = since + c.mu.Unlock() + } + // A node unanswered, heard, and not yet past the bound: its bound's expiry is the next step. + if c.Arm != nil { + for _, node := range nodes { + s, ok := statements[node] + if ok && !s.Answered && !s.UnansweredSince.IsZero() && c.Presence.Online(node) && !raised[mcc.Key(mcc.ScopeCall, readsCallID(node), mcc.KindCallUnanswered)] { + if err := c.Arm(ctx, node, s.UnansweredSince.Add(mcc.CallUnansweredAfter)); err != nil { + c.logf("the bound of %s's manifest-reads is not armed: %v", node, err) + } + } + } + } + // Every counted node's finding no longer raised, and every one written before, is cleared. + nodeSet := map[string]bool{} + for _, node := range nodes { + nodeSet[node] = true + } + c.mu.Lock() + for node := range c.said { + nodeSet[node] = true + } + c.mu.Unlock() + for _, node := range slices.Sorted(maps.Keys(nodeSet)) { + if raised[mcc.Key(mcc.ScopeCall, readsCallID(node), mcc.KindCallUnanswered)] { + continue + } + if err := c.Clear(ctx, readsCallID(node)); err != nil { + c.logf("the finding of %s's manifest-reads is not cleared: %v", node, err) + continue + } + c.mu.Lock() + delete(c.said, node) + c.mu.Unlock() + } +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index 71e15570..4b27a7a2 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -294,6 +294,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro reads.stepWindow() }) gate.OnPlaced = reads.placed + // A kept manifest-reads call unanswered past five minutes, its node's machine heard (the watchman's presence), is + // call-unanswered in the builder's findings (issue 1966, ADR 0668 §3). + presence := &buildrun.PresenceSet{} + reads.callsOf(found, pubBoundsKV.KV, presence, logger.Printf) + watchPresence(ctx, js, presence, reads.stepCalls, errs, logger.Printf) b.Forms = reads.built // Until the assignments's first view is answered the gate passes every report on, as before the assignments is settled. assignmentsErrs := make(chan error, 1) @@ -407,6 +412,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro unput.expired(ctx, e) return } + if _, ok := buildrun.ReadsCallOf(e.Key()); ok { + if e.Operation() == jetstream.KeyValuePurge { // its five minutes ended + reads.stepCalls() + } + return + } if versions.IsOldFormBound(e.Key()) { if e.Operation() == jetstream.KeyValuePurge { reads.stepWindow() // a node's seven days ended: the step raises old-form-held now diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reads.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reads.go index b014f154..f9770a2f 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reads.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reads.go @@ -34,12 +34,17 @@ type engineReads struct { // window is the old reference form's window over these statements (DLV-B9); nil keeps none. window *oldForm ctx context.Context + // calls is the wait of the kept calls (callsOf); nil weighs none. + calls *buildrun.ReadsCalls } // newEngineReads asks through call (process.Conn.CallKeptOn, a kept call; ADR 0668) and keeps each answer in the state once it is bound. func newEngineReads(call enginereads.Caller, gate *buildrun.ToolchainGate, logf func(string, ...any)) *engineReads { r := &engineReads{gate: gate, keeper: &enginereads.Keeper{Call: call, Logf: logf}} - r.keeper.Changed = r.stepWindow + r.keeper.Changed = func() { + r.stepWindow() + r.stepCalls() + } return r } @@ -57,6 +62,19 @@ func (r *engineReads) stepWindow() { r.window.step(ctx, r.named(), counted) } +// stepCalls weighs the kept manifest-reads calls (buildrun.ReadsCalls): after a statement changed, a node's presence, the +// assignments's change and the end of a call's bound. +func (r *engineReads) stepCalls() { + if r.calls == nil { + return + } + ctx := r.ctx + if ctx == nil { + ctx = context.Background() + } + r.calls.Step(ctx) +} + // named are the nodes the assignments names, none dropped: what the window's step judges a read by. func (r *engineReads) named() []string { r.mu.Lock() @@ -106,6 +124,7 @@ func (r *engineReads) placed() { } } r.stepWindow() + r.stepCalls() } // atBind asks every counted node again when the builder's consumer is bound (the hub's OnBound): a manifest-reads-changed diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/readscall.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/readscall.go new file mode 100644 index 00000000..399b152c --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/readscall.go @@ -0,0 +1,63 @@ +package main + +import ( + "context" + "errors" + "time" + + "github.com/nats-io/nats.go/jetstream" + + mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + "git.novox.be/novox/mesh-sdk/go/process" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" + "mesh-delivery-builder/internal/versions" +) + +// callsOf is the wait of the builder's kept manifest-reads calls (buildrun.ReadsCalls): a node whose engine has not +// answered while its machine is heard is, past five minutes, the finding call-unanswered in the builder's findings +// (issue 1966, ADR 0668 §3); its bound is a key of the state bounds with its own time to live. +func (r *engineReads) callsOf(f *findings, bounds func() jetstream.KeyValue, presence *buildrun.PresenceSet, logf func(string, ...any)) *buildrun.ReadsCalls { + c := &buildrun.ReadsCalls{Keeper: r.keeper, Presence: presence, Nodes: r.named, Logf: logf, + Source: "builder." + mcc.KindCallUnanswered, Needs: "builder_engine_reads with ask_again asks the node again now and shows its statement.", + Raise: func(ctx context.Context, o mcc.Observation) error { f.raiseObservation(ctx, o); return nil }, + Clear: func(ctx context.Context, id string) error { + f.clear(ctx, mcc.KindCallUnanswered, mcc.ScopeCall, id) + return nil + }, + Arm: func(ctx context.Context, node string, ends time.Time) error { + kv := bounds() + if kv == nil { + return errors.New(versions.BoundsBucket + " is not on the bus yet") + } + left := max(time.Until(ends), time.Second) + _, err := kv.Create(ctx, buildrun.ReadsCallBoundPrefix+node, versions.EncodeBound("the manifest-reads call to "+node, ends), jetstream.KeyTTL(left)) + if errors.Is(err, jetstream.ErrKeyExists) { + err = nil + } + return err + }} + r.calls = c + return c +} + +// watchPresence follows the watchman's presence, which says a machine is heard: a node coming online or going weighs the +// calls again. +func watchPresence(ctx context.Context, js jetstream.JetStream, presence *buildrun.PresenceSet, step func(), errs chan<- error, logf func(string, ...any)) { + const without = "no node counts as heard, so a manifest-reads call left unanswered is not raised until the builder starts again" + kv := process.OptionalBucket(ctx, js, buildrun.PresenceBucket, without, logf) + if kv == nil { + return + } + each := func(e jetstream.KeyValueEntry) { + if presence.Put(e.Key(), e.Value(), !opOf(e)) { + step() + } + } + w, err := settle(ctx, kv, each) + if err != nil { + process.NotWatched(logf, buildrun.PresenceBucket, err, without) + return + } + go func() { errs <- follow(ctx, w, each) }() +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go index 9d1b8050..c76687e2 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go @@ -24,6 +24,7 @@ import ( "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads" "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs" sweep "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshartifactsweep" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" @@ -137,6 +138,67 @@ func theCountedNodesEngineReads(t *testing.T) { if _, err := tool.Run(map[string]any{"node": "n1", "drop": "true", "why": "a test"}); err != nil || st.v["n1"] != nil { t.Errorf("drop: %v, the key %s", err, st.v["n1"]) } + t.Run("ADR 0668 §3: a manifest-reads call unanswered past five minutes, its machine heard, is call-unanswered", func(t *testing.T) { + // n2 answers nothing: while its machine is away no finding; heard and still unanswered five minutes on, the finding + // call-unanswered in the builder's findings and a bound armed before; answered, or away again, it is cleared. + place("n1", "n2") + now := time.Unix(1000, 0) + clock := func() time.Time { return now } + down := true + r2 := newEngineReads(func(node, _ string, _ any) (json.RawMessage, error) { + if down { + return nil, nats.ErrNoResponders + } + return json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: node, ManifestReads: []string{"form:new"}, StatedAt: now.Format(time.RFC3339Nano)}) + }, gate, t.Logf) + r2.keeper.Sync, r2.keeper.Now = true, clock + st := &memFindings{vals: map[string][]byte{}} + armed := map[string]time.Time{} + presence := &buildrun.PresenceSet{} + calls := r2.callsOf(&findings{store: st, logf: t.Logf, now: clock}, func() jetstream.KeyValue { return nil }, presence, t.Logf) + calls.Now = clock + calls.Arm = func(_ context.Context, node string, ends time.Time) error { armed[node] = ends; return nil } + key := meshconditionsclient.Key(meshconditionsclient.ScopeCall, "manifest-reads.n2", meshconditionsclient.KindCallUnanswered) + r2.placed() // n2 asked, unanswered at 1000 + now = now.Add(6 * time.Minute) + r2.stepCalls() + if len(st.vals) != 0 || len(armed) != 0 { + t.Fatalf("n2's machine is away: findings %v, bounds %v", st.vals, armed) + } + presence.Put("n2", presenceOnline(t, true), false) + now = time.Unix(1000, 0).Add(2 * time.Minute) + r2.stepCalls() + if len(st.vals) != 0 || !armed["n2"].Equal(time.Unix(1000, 0).Add(meshconditionsclient.CallUnansweredAfter)) { + t.Fatalf("heard before five minutes: findings %v, bounds %v", st.vals, armed) + } + now = time.Unix(1000, 0).Add(meshconditionsclient.CallUnansweredAfter) + r2.stepCalls() + f, err := meshconditionsclient.DecodeFinding(st.vals[key]) + if err != nil || f.Observer != builds.BuilderModule || f.Kind != meshconditionsclient.KindCallUnanswered { + t.Fatalf("heard five minutes on: finding %+v, %v (keys %v)", f, err, slices.Collect(maps.Keys(st.vals))) + } + presence.Put("n2", presenceOnline(t, false), false) + r2.stepCalls() + if len(st.vals) != 0 { + t.Errorf("its machine away again leaves %v", st.vals) + } + presence.Put("n2", presenceOnline(t, true), false) + r2.stepCalls() + down = false + r2.keeper.AskNode("n2", "answered now") + if len(st.vals) != 0 { + t.Errorf("answered leaves %v", st.vals) + } + }) +} + +func presenceOnline(t *testing.T, on bool) []byte { + t.Helper() + b, err := json.Marshal(meshwatchdogs.PresenceByNodeValue{Online: on}) + if err != nil { + t.Fatal(err) + } + return b } func mustStatement(t *testing.T, s enginereads.Statement) []byte { diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index 6e883afc..6af6a26e 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -100,10 +100,11 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 if want := []state{{Name: "builds", History: 5}, {Name: "engine-reads", History: 1}, {Name: "findings", History: 5}, {Name: "floor-waits", History: 1}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "old-form", History: 1}, {Name: "bounds", History: 1, KeyTTL: true}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}, {Name: "tag-ask-bounds", History: 1, KeyTTL: true}, {Name: "tag-asks", History: 1}, {Name: "versions", History: 1}, {Name: "first-views", History: 1}}; !reflect.DeepEqual(m.State, want) { // builds.BuilderBuildsHistory t.Errorf("state = %+v, want %+v", m.State, want) } - // 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and - // holds no capacity; it puts asks on the three build seats and reads their steps (ADR 0525 §1, §4, §5). + // 1887/A4c: the builder chooses no node, so it reads no have, invokes no build module's tool and holds no capacity; it + // puts asks on the three build seats and reads their steps (ADR 0525 §1, §4, §5). It reads the watchman's presence + // only to know a node's machine is heard, for a manifest-reads call left unanswered (issue 1966, ADR 0668 §3). if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains", - "mesh-delivery-assigner.assignments", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { + "mesh-delivery-assigner.assignments", "mesh-delivery-gate.source", "mesh-delivery-watchman.presence"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } if want := []string{"node-build-go", "node-build-image", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) { diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index 2f4140d0..a5fc4f0c 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -107,7 +107,8 @@ "mesh-build-node.steps", "mesh-build-node.toolchains", "mesh-delivery-assigner.assignments", - "mesh-delivery-gate.source" + "mesh-delivery-gate.source", + "mesh-delivery-watchman.presence" ], "uses": [ "node-build-go", diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index d973cc3f..409492b9 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -506,8 +506,19 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // weighing only, said. holdBounds := process.AwaitBucket(ctx, js, conn, index.HoldBoundsBucket, "a hold standing past its bound is raised at its next weighing only", logger.Printf) ix.HoldBounds = expiringKeys{holdBounds} + // A kept manifest-reads call unanswered past five minutes, its node's machine heard (the watchman's presence), is + // call-unanswered in the indexer's findings (issue 1966, ADR 0668 §3). + presence := &buildrun.PresenceSet{} + reads.callsOf(ctx, findings.KV, ix.HoldBounds, presence, logger.Printf) + watchPresence(ctx, js, presence, reads.stepCalls, errs, logger.Printf) holdBounds.Ready(func(kv jetstream.KeyValue) { each := func(e jetstream.KeyValueEntry) { + if _, ok := buildrun.ReadsCallOf(e.Key()); ok { + if e.Operation() == jetstream.KeyValuePurge { // a manifest-reads call's five minutes ended + reads.stepCalls() + } + return + } if e.Operation() == jetstream.KeyValuePurge { // its time to live ended ix.HoldBoundEnded(ctx, e.Key()) } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go index 1bb57db3..a4b612c0 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go @@ -13,8 +13,12 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads" + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" "git.novox.be/novox/mesh-sdk/go/process" @@ -94,6 +98,61 @@ func TestAMessageIsIndexedAndAcked(t *testing.T) { // 1238/AC2 (unit part) t.Errorf("at the bind called %v, want ace and g14 once each", called) } }) + t.Run("a manifest-reads call unanswered past five minutes, its machine heard, is call-unanswered (ADR 0668 §3)", func(t *testing.T) { + // While the node's machine is away no finding is written; heard, a bound is armed and, five minutes on, the finding + // is written in the indexer's findings; away again or answered, it is cleared. + now := time.Unix(1000, 0) + clock := func() time.Time { return now } + down := true + held := &findingsKV{vals: map[string][]byte{}} + reads := &engineReads{known: true, nodes: []string{"ace"}} + reads.keeper = &enginereads.Keeper{Sync: true, Now: clock, Call: func(node, _ string, _ any) (json.RawMessage, error) { + if down { + return nil, errors.New("no responders") + } + return json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: node, ManifestReads: []string{"form:new"}, StatedAt: now.Format(time.RFC3339Nano)}) + }} + reads.keeper.Changed = reads.stepCalls + armed := map[string]time.Time{} + presence := &buildrun.PresenceSet{} + calls := reads.callsOf(context.Background(), func() jetstream.KeyValue { return held }, nil, presence, t.Logf) + calls.Now = clock + calls.Arm = func(_ context.Context, node string, ends time.Time) error { armed[node] = ends; return nil } + online := func(on bool) { + b, _ := json.Marshal(meshwatchdogs.PresenceByNodeValue{Online: on}) + presence.Put("ace", b, false) + } + key := mcc.Key(mcc.ScopeCall, "manifest-reads.ace", mcc.KindCallUnanswered) + reads.keeper.AskNode("ace", "counted") // unanswered at 1000 + now = now.Add(6 * time.Minute) + reads.stepCalls() + if len(held.vals) != 0 || len(armed) != 0 { + t.Fatalf("its machine is away: findings %v, bounds %v", held.vals, armed) + } + online(true) + now = time.Unix(1000, 0).Add(time.Minute) + reads.stepCalls() + if len(held.vals) != 0 || !armed["ace"].Equal(time.Unix(1000, 0).Add(mcc.CallUnansweredAfter)) { + t.Fatalf("heard before five minutes: findings %v, bounds %v", held.vals, armed) + } + now = time.Unix(1000, 0).Add(mcc.CallUnansweredAfter) + reads.stepCalls() + if f, err := mcc.DecodeFinding(held.vals[key]); err != nil || f.Observer != "mesh-delivery-indexer" || f.Kind != mcc.KindCallUnanswered { + t.Fatalf("heard five minutes on: finding %+v, %v", f, err) + } + online(false) + reads.stepCalls() + if len(held.vals) != 0 { + t.Errorf("its machine away again leaves %v", held.vals) + } + online(true) + reads.stepCalls() + down = false + reads.keeper.AskNode("ace", "answered") + if len(held.vals) != 0 || !reads.keeper.Statements()["ace"].Answered { + t.Errorf("answered leaves %v", held.vals) + } + }) t.Run("not a forge message: terminated and logged", func(t *testing.T) { // 1238 (never redelivered) var logs bytes.Buffer st := &memStore{vals: map[string][]byte{}} @@ -211,3 +270,26 @@ func TestTheSourceMainKeyIsTakenAtItsRevision(t *testing.T) { // 1344/AC1 (the w } func ptr[T any](v T) *T { return &v } + +// findingsKV is a findings bucket of the two calls the indexer's call finding makes. +type findingsKV struct { + jetstream.KeyValue + vals map[string][]byte +} + +func (f *findingsKV) Put(_ context.Context, key string, value []byte) (uint64, error) { + f.vals[key] = value + return uint64(len(f.vals)), nil +} + +func (f *findingsKV) Get(_ context.Context, key string) (jetstream.KeyValueEntry, error) { + if _, ok := f.vals[key]; !ok { + return nil, jetstream.ErrKeyNotFound + } + return nil, nil +} + +func (f *findingsKV) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error { + delete(f.vals, key) + return nil +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/reads.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/reads.go index 890a8f10..89b7c249 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/reads.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/reads.go @@ -28,11 +28,29 @@ type engineReads struct { mu sync.Mutex known bool // the assignments is read nodes []string + // calls is the wait of the kept calls (callsOf); nil weighs none. ctx is the indexer's. + calls *buildrun.ReadsCalls + ctx context.Context } // newEngineReads asks through call (process.Conn.CallKeptOn, a kept call; ADR 0668) and tells changed after each statement changed. func newEngineReads(call enginereads.Caller, gate *buildrun.ToolchainGate, changed func(), logf func(string, ...any)) *engineReads { - return &engineReads{gate: gate, keeper: &enginereads.Keeper{Call: call, Logf: logf, Changed: changed}} + r := &engineReads{gate: gate, ctx: context.Background()} + r.keeper = &enginereads.Keeper{Call: call, Logf: logf, Changed: func() { + if changed != nil { + changed() + } + r.stepCalls() + }} + return r +} + +// stepCalls weighs the kept manifest-reads calls (buildrun.ReadsCalls): after a statement changed, a node's presence, the +// assignments's change and the end of a call's bound. +func (r *engineReads) stepCalls() { + if r.calls != nil { + r.calls.Step(r.ctx) + } } // bind keeps the statements in kv and hears it: every node's last statement at the start, a person's ask again and drop @@ -76,6 +94,7 @@ func (r *engineReads) placed() { if r.keeper.Changed != nil { r.keeper.Changed() } + r.stepCalls() } // atBind asks every counted node again when the indexer's consumer is bound (the hub's OnBound): a manifest-reads-changed diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/readscall.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/readscall.go new file mode 100644 index 00000000..2246fcf6 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/readscall.go @@ -0,0 +1,76 @@ +package main + +import ( + "context" + "errors" + "slices" + "time" + + "github.com/nats-io/nats.go/jetstream" + + mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" + "git.novox.be/novox/mesh-sdk/go/process" + + "mesh-delivery-indexer/internal/index" +) + +// callsOf is the wait of the indexer's kept manifest-reads calls (buildrun.ReadsCalls): a node whose engine has not +// answered while its machine is heard is, past five minutes, the finding call-unanswered in the indexer's findings +// (issue 1966, ADR 0668 §3); its bound is a key of the state hold-bounds, which expires at its end. +func (r *engineReads) callsOf(ctx context.Context, findings func() jetstream.KeyValue, bounds index.Bounds, presence *buildrun.PresenceSet, logf func(string, ...any)) *buildrun.ReadsCalls { + c := &buildrun.ReadsCalls{Keeper: r.keeper, Presence: presence, Logf: logf, + Nodes: func() []string { r.mu.Lock(); defer r.mu.Unlock(); return slices.Clone(r.nodes) }, + Source: "indexer." + mcc.KindCallUnanswered, Needs: index.EngineReadsTool + " with ask_again asks the node again now and shows its statement.", + Raise: func(ctx context.Context, o mcc.Observation) error { + kv := findings() + if kv == nil { + return process.ErrNotYet + } + value, err := mcc.EncodeFinding(o.Finding(index.Module, "", time.Now().UTC())) + if err == nil { + _, err = kv.Put(ctx, o.Key(), value) + } + return err + }, + Clear: func(ctx context.Context, id string) error { + kv := findings() + if kv == nil { + return nil + } + key := mcc.Key(mcc.ScopeCall, id, mcc.KindCallUnanswered) + if _, err := kv.Get(ctx, key); err != nil { + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil + } + return err + } + return kv.Delete(ctx, key) + }, + Arm: func(ctx context.Context, node string, ends time.Time) error { + return bounds.Arm(ctx, buildrun.ReadsCallBoundPrefix+node, ends) + }} + r.ctx, r.calls = ctx, c + return c +} + +// watchPresence follows the watchman's presence, which says a machine is heard: a node coming online or going weighs the +// calls again. +func watchPresence(ctx context.Context, js jetstream.JetStream, presence *buildrun.PresenceSet, step func(), errs chan<- error, logf func(string, ...any)) { + const without = "no node counts as heard, so a manifest-reads call left unanswered is not raised until the indexer starts again" + kv := process.OptionalBucket(ctx, js, buildrun.PresenceBucket, without, logf) + if kv == nil { + return + } + each := func(e jetstream.KeyValueEntry) { + if presence.Put(e.Key(), e.Value(), !opOf(e)) { + step() + } + } + w, err := settle(ctx, kv, each) + if err != nil { + process.NotWatched(logf, buildrun.PresenceBucket, err, without) + return + } + go func() { errs <- follow(ctx, w, each) }() +} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index c5fa5e79..86c933e3 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -137,7 +137,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build } } // DLV-F1r (ADR 0615 §1): the assigner's assignments is read through its first view and views family, never its state. - if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { + // The watchman's presence says a node's machine is heard, for a manifest-reads call left unanswered (issue 1966, ADR 0668 §3). + if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews", "mesh-delivery-gate.source", "mesh-delivery-watchman.presence"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } // "Reads FORGE" is a consumes of the indexer's stream: the broker grants only buckets by reads (mesh-control diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 41198549..9344bd46 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -119,7 +119,8 @@ "mesh-build-node.toolchains", "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews", - "mesh-delivery-gate.source" + "mesh-delivery-gate.source", + "mesh-delivery-watchman.presence" ], "invokes": [ "mesh-delivery-assigner.assignments",