D11: the builder and the indexer say a manifest-reads call unanswered past five minutes, its machine heard, as call-unanswered in their own findings (issue 1966, ADR 0668 §3; review round 2)

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
This commit is contained in:
2026-11-10 04:38:38 +01:00
parent 00e84bb977
commit 0f52bb80ad
14 changed files with 491 additions and 8 deletions
@@ -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
@@ -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, "<part>.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()
}
}
@@ -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
@@ -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
@@ -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) }()
}
@@ -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 {
@@ -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) {
+2 -1
View File
@@ -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",
@@ -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())
}
@@ -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
}
@@ -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
@@ -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) }()
}
@@ -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
+2 -1
View File
@@ -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",