novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

D13: the builder and the indexer ask the watchdogs seat's presence when a manifest-reads call's bound ends, through the node-engine seat's client's ReadsCalls, and read no watchman state (issue 1966, ADR 0596 §2, ADR 0668 §3, ADR 0634 §1; review round 3, findings 1 and 2)

Browse Source 
Both manifests drop the read mesh-delivery-watchman.presence and invoke seat:mesh-watchdogs.presence; buildrun's ReadsCalls, its presence bucket and both watchPresence copies go. The machine is asked once when the bound key expires; a node that returns is asked again on manifest-reads-changed, and the finding clears on its answer. The indexer's finding is stamped with the wait's clock.

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-10 05:01:15 +01:00
parent 8e426b236d
commit 98ffc1fb80
15 changed files with 179 additions and 309 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-builder/buildrun/go.mod
-1
View File
@@ -6,7 +6,6 @@ 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
modules/mesh-delivery-builder/buildrun/readscalls.go
-135
View File
@@ -1,135 +0,0 @@
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()
}
}
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go
+6 -7
View File
@@ -22,6 +22,7 @@ import (
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
sdkfindings "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/findings"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds/forms"
@@ -294,11 +295,9 @@ 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)
// A kept manifest-reads call unanswered past five minutes, its node's machine heard (the watchdogs seat's presence,
// asked when the bound ends), is call-unanswered in the builder's findings (issue 1966, ADR 0668 §3, ADR 0596 §2).
reads.callsOf(found, pubBoundsKV.KV, heardBy(calls.CallKept), 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)
@@ -412,9 +411,9 @@ 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 node, ok := enginereads.BoundOf(e.Key()); ok {
if e.Operation() == jetstream.KeyValuePurge { // its five minutes ended
reads.stepCalls()
reads.expiredCall(node)
}
return
}
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reads.go
+18 -9
View File
@@ -35,7 +35,7 @@ type engineReads struct {
window *oldForm
ctx context.Context
// calls is the wait of the kept calls (callsOf); nil weighs none.
calls *buildrun.ReadsCalls
calls *enginereads.ReadsCalls
}
// newEngineReads asks through call (process.Conn.CallKeptOn, a kept call; ADR 0668) and keeps each answer in the state once it is bound.
@@ -62,17 +62,26 @@ 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.
// stepCalls weighs the kept manifest-reads calls (enginereads.ReadsCalls): after a statement changed and the assignments's
// change.
func (r *engineReads) stepCalls() {
if r.calls == nil {
return
if r.calls != nil {
r.calls.Step(r.context())
}
ctx := r.ctx
if ctx == nil {
ctx = context.Background()
}
// expiredCall weighs node's kept manifest-reads call when its bound ended (the key's expiry in the state bounds).
func (r *engineReads) expiredCall(node string) {
if r.calls != nil {
r.calls.Expired(r.context(), node)
}
r.calls.Step(ctx)
}
func (r *engineReads) context() context.Context {
if r.ctx == nil {
return context.Background()
}
return r.ctx
}
// named are the nodes the assignments names, none dropped: what the window's step judges a read by.
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/readscall.go
+42 -33
View File
@@ -2,27 +2,54 @@ package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
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-contracts/health-and-repair/go/seats/meshwatchdogs"
"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)
// presenceInvoke is the watchdogs seat's verb presence as the manifest's invokes names it: whether a node's machine is
// heard, asked of the seat, never read from its holder's state (novox/hq ADR 0596 §2).
const presenceInvoke = "seat:" + meshwatchdogs.Seat + "." + meshwatchdogs.VerbPresence
// heardBy asks the watchdogs seat's presence of one machine through call (process.Conn.CallKept, a read) and reads its
// declared answer: whether the machine is heard now.
func heardBy(call func(key string, body any) (json.RawMessage, error)) func(context.Context, string) (bool, error) {
return func(_ context.Context, machine string) (bool, error) {
raw, err := call(presenceInvoke, meshwatchdogs.PresenceInput{Machine: machine})
if err != nil {
return false, err
}
var a meshwatchdogs.PresenceAnswer
if err := json.Unmarshal(raw, &a); err != nil {
return false, fmt.Errorf("%s does not read: %w", presenceInvoke, err)
}
return a.Online, nil
}
}
// callsOf is the wait of the builder's kept manifest-reads calls (enginereads.ReadsCalls): a node whose engine has not
// answered for five minutes is, when its bound ends and the watchdogs seat says its machine is heard, 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, heard func(context.Context, string) (bool, error), logf func(string, ...any)) *enginereads.ReadsCalls {
c := &enginereads.ReadsCalls{Keeper: r.keeper, Nodes: r.named, Heard: heard, Bound: mcc.CallUnansweredAfter, Logf: logf,
Raise: func(ctx context.Context, u enginereads.UnansweredCall) error {
o := mcc.CallUnanswered(enginereads.CallID(u.Node), enginereads.Invoke, u.Node, u.Words, "builder."+mcc.KindCallUnanswered, u.Since)
o.Needs = "builder_engine_reads with ask_again asks the node again now and shows its statement."
f.raiseObservation(ctx, o)
return nil
},
Clear: func(ctx context.Context, node string) error {
f.clear(ctx, mcc.KindCallUnanswered, mcc.ScopeCall, enginereads.CallID(node))
return nil
},
Arm: func(ctx context.Context, node string, ends time.Time) error {
@@ -31,33 +58,15 @@ func (r *engineReads) callsOf(f *findings, bounds func() jetstream.KeyValue, pre
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))
_, err := kv.Create(ctx, enginereads.BoundPrefix+node, versions.EncodeBound("the manifest-reads call to "+node, ends), jetstream.KeyTTL(left))
if errors.Is(err, jetstream.ErrKeyExists) {
err = nil
}
return err
}}
if f != nil && f.now != nil {
c.Now = f.now
}
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) }()
}
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go
+22 -33
View File
@@ -139,8 +139,9 @@ func theCountedNodesEngineReads(t *testing.T) {
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.
// n2 answers nothing: a bound is kept at five minutes and the watchdogs seat's presence is asked only when it ends
// (ADR 0596 §2), once; away, no finding; heard, the finding call-unanswered in the builder's findings at the
// builder's clock; answered, it is cleared.
place("n1", "n2")
now := time.Unix(1000, 0)
clock := func() time.Time { return now }
@@ -154,53 +155,41 @@ func theCountedNodesEngineReads(t *testing.T) {
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
online, asked := false, []any{}
heard := heardBy(func(key string, body any) (json.RawMessage, error) {
asked = append(asked, key, body)
return json.Marshal(meshwatchdogs.PresenceAnswer{Machine: "n2", Online: online})
})
calls := r2.callsOf(&findings{store: st, logf: t.Logf, now: clock}, func() jetstream.KeyValue { return nil }, heard, t.Logf)
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)
if len(st.vals) != 0 || len(asked) != 0 || !armed["n2"].Equal(time.Unix(1000, 0).Add(meshconditionsclient.CallUnansweredAfter)) {
t.Fatalf("before five minutes: findings %v, presence asked %v, bounds %v", st.vals, asked, 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 = now.Add(meshconditionsclient.CallUnansweredAfter)
r2.expiredCall("n2")
if len(st.vals) != 0 || len(asked) != 2 || asked[0] != presenceInvoke || asked[1] != (meshwatchdogs.PresenceInput{Machine: "n2"}) {
t.Fatalf("its machine away: findings %v, presence asked %v", st.vals, asked)
}
now = time.Unix(1000, 0).Add(meshconditionsclient.CallUnansweredAfter)
r2.stepCalls()
online = true
r2.expiredCall("n2")
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)))
if err != nil || f.Observer != builds.BuilderModule || f.Kind != meshconditionsclient.KindCallUnanswered || len(asked) != 4 {
t.Fatalf("heard five minutes on: finding %+v, %v (keys %v), presence asked %v", f, err, slices.Collect(maps.Keys(st.vals)), asked)
}
presence.Put("n2", presenceOnline(t, false), false)
r2.stepCalls()
if len(st.vals) != 0 {
t.Errorf("its machine away again leaves %v", st.vals)
if len(st.vals) != 1 || len(asked) != 4 {
t.Errorf("a step after it: findings %v, presence asked %v", st.vals, asked)
}
presence.Put("n2", presenceOnline(t, true), false)
r2.stepCalls()
down = false
r2.keeper.AskNode("n2", "answered now")
r2.keeper.AskNode("n2", "answered now, its manifest-reads-changed heard")
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 {
t.Helper()
raw, err := enginereads.Encode(s)
modules/mesh-delivery-builder/manifest_test.go
+6 -5
View File
@@ -100,11 +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 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).
// 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). Whether a
// node's machine is heard it asks the watchdogs seat's presence (issue 1966, ADR 0596 §2, 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", "mesh-delivery-watchman.presence"}; !reflect.DeepEqual(sorted(m.Reads), want) {
"mesh-delivery-assigner.assignments", "mesh-delivery-gate.source"}; !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) {
@@ -124,9 +124,10 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
}
}
// Design 51 *The floor*: the floor's three sources, the registry seats' versions, bridge_tags and indexer_requirements.
// The watchdogs seat's presence says a node's machine is heard (issue 1966, ADR 0596 §2, ADR 0668 §3).
if want := []string{"mesh-delivery-assigner.assigner_read_assignments", "mesh-delivery-forge-bridge.bridge_tags", "mesh-delivery-indexer.indexer_read_candidates", "mesh-delivery-indexer.indexer_read_manifests",
"mesh-delivery-indexer.indexer_read_modules", "mesh-delivery-indexer.indexer_read_packages", "mesh-delivery-indexer.indexer_requirements", "seat:container-image-registry.remove", "seat:container-image-registry.versions",
"seat:go-package-registry.remove", "seat:go-package-registry.versions", "seat:mesh-artifact-sweep.held", "seat:mesh-hand-acts.hand-act", "seat:node-build-go.drop", "seat:node-build-image.drop", "seat:node-build-node.drop", "seat:node-engine.manifest-reads", "seat:npm-package-registry.remove",
"seat:go-package-registry.remove", "seat:go-package-registry.versions", "seat:mesh-artifact-sweep.held", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence", "seat:node-build-go.drop", "seat:node-build-image.drop", "seat:node-build-node.drop", "seat:node-engine.manifest-reads", "seat:npm-package-registry.remove",
"seat:npm-package-registry.versions", "seat:oci-artifact-registry.remove", "seat:oci-artifact-registry.versions"}; !reflect.DeepEqual(sorted(m.Invokes), want) || len(m.Config) != 0 {
t.Errorf("invokes = %v, config = %v, want %v and no config", m.Invokes, m.Config, want)
}
modules/mesh-delivery-builder/module.json
+2 -2
View File
@@ -67,6 +67,7 @@
"seat:go-package-registry.versions",
"seat:mesh-artifact-sweep.held",
"seat:mesh-hand-acts.hand-act",
"seat:mesh-watchdogs.presence",
"seat:node-build-go.drop",
"seat:node-build-image.drop",
"seat:node-build-node.drop",
@@ -107,8 +108,7 @@
"mesh-build-node.steps",
"mesh-build-node.toolchains",
"mesh-delivery-assigner.assignments",
"mesh-delivery-gate.source",
"mesh-delivery-watchman.presence"
"mesh-delivery-gate.source"
],
"uses": [
"node-build-go",
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go
+5 -7
View File
@@ -506,16 +506,14 @@ 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)
// A kept manifest-reads call unanswered past five minutes, its node's machine heard (the watchdogs seat's presence,
// asked when the bound ends), is call-unanswered in the indexer's findings (issue 1966, ADR 0668 §3, ADR 0596 §2).
reads.callsOf(ctx, findings.KV, ix.HoldBounds, heardBy(calls.CallKept), logger.Printf)
holdBounds.Ready(func(kv jetstream.KeyValue) {
each := func(e jetstream.KeyValueEntry) {
if _, ok := buildrun.ReadsCallOf(e.Key()); ok {
if node, ok := enginereads.BoundOf(e.Key()); ok {
if e.Operation() == jetstream.KeyValuePurge { // a manifest-reads call's five minutes ended
reads.stepCalls()
reads.expiredCall(node)
}
return
}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go
+20 -31
View File
@@ -18,7 +18,6 @@ import (
"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"
@@ -99,8 +98,8 @@ func TestAMessageIsIndexedAndAcked(t *testing.T) { // 1238/AC2 (unit part)
}
})
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.
// A bound is kept at five minutes; when it ends the watchdogs seat's presence is asked, once (ADR 0596 §2): away,
// no finding; heard, the finding in the indexer's findings, stamped with the wait's clock; answered, it is cleared.
now := time.Unix(1000, 0)
clock := func() time.Time { return now }
down := true
@@ -114,43 +113,33 @@ func TestAMessageIsIndexedAndAcked(t *testing.T) { // 1238/AC2 (unit part)
}}
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)
online, asked := false, []any{}
heard := heardBy(func(key string, body any) (json.RawMessage, error) {
asked = append(asked, key, body)
return json.Marshal(meshwatchdogs.PresenceAnswer{Machine: "ace", Online: online})
})
calls := reads.callsOf(context.Background(), func() jetstream.KeyValue { return held }, nil, heard, 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)
if len(held.vals) != 0 || len(asked) != 0 || !armed["ace"].Equal(time.Unix(1000, 0).Add(mcc.CallUnansweredAfter)) {
t.Fatalf("before five minutes: findings %v, presence asked %v, bounds %v", held.vals, asked, 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 + time.Second)
reads.expiredCall("ace")
if len(held.vals) != 0 || len(asked) != 2 || asked[0] != presenceInvoke || asked[1] != (meshwatchdogs.PresenceInput{Machine: "ace"}) {
t.Fatalf("its machine away: findings %v, presence asked %v", held.vals, asked)
}
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 = true
reads.expiredCall("ace")
if f, err := mcc.DecodeFinding(held.vals[key]); err != nil || f.Observer != "mesh-delivery-indexer" || f.Kind != mcc.KindCallUnanswered || !f.At.Equal(now) {
t.Fatalf("heard five minutes on: finding %+v, %v; want it at the wait's clock %v", f, err, now)
}
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)
if len(held.vals) != 0 || !reads.keeper.Statements()["ace"].Answered || len(asked) != 4 {
t.Errorf("answered leaves %v, presence asked %v", held.vals, asked)
}
})
t.Run("not a forge message: terminated and logged", func(t *testing.T) { // 1238 (never redelivered)
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/reads.go
+10 -3
View File
@@ -29,7 +29,7 @@ type engineReads struct {
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
calls *enginereads.ReadsCalls
ctx context.Context
}
@@ -45,14 +45,21 @@ func newEngineReads(call enginereads.Caller, gate *buildrun.ToolchainGate, chang
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.
// stepCalls weighs the kept manifest-reads calls (enginereads.ReadsCalls): after a statement changed and the assignments's
// change.
func (r *engineReads) stepCalls() {
if r.calls != nil {
r.calls.Step(r.ctx)
}
}
// expiredCall weighs node's kept manifest-reads call when its bound ended (the key's expiry in the state hold-bounds).
func (r *engineReads) expiredCall(node string) {
if r.calls != nil {
r.calls.Expired(r.ctx, node)
}
}
// bind keeps the statements in kv and hears it: every node's last statement at the start, a person's ask again and drop
// later. Every counted node is asked again once bound, so what was asked before is kept in it too.
func (r *engineReads) bind(ctx context.Context, kv jetstream.KeyValue, errs chan<- error, logf func(string, ...any)) {
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/readscall.go
+42 -36
View File
@@ -2,43 +2,68 @@ package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"slices"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
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-contracts/health-and-repair/go/seats/meshwatchdogs"
"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 {
// presenceInvoke is the watchdogs seat's verb presence as the manifest's invokes names it: whether a node's machine is
// heard, asked of the seat, never read from its holder's state (novox/hq ADR 0596 §2).
const presenceInvoke = "seat:" + meshwatchdogs.Seat + "." + meshwatchdogs.VerbPresence
// heardBy asks the watchdogs seat's presence of one machine through call (process.Conn.CallKept, a read) and reads its
// declared answer: whether the machine is heard now.
func heardBy(call func(key string, body any) (json.RawMessage, error)) func(context.Context, string) (bool, error) {
return func(_ context.Context, machine string) (bool, error) {
raw, err := call(presenceInvoke, meshwatchdogs.PresenceInput{Machine: machine})
if err != nil {
return false, err
}
var a meshwatchdogs.PresenceAnswer
if err := json.Unmarshal(raw, &a); err != nil {
return false, fmt.Errorf("%s does not read: %w", presenceInvoke, err)
}
return a.Online, nil
}
}
// callsOf is the wait of the indexer's kept manifest-reads calls (enginereads.ReadsCalls): a node whose engine has not
// answered for five minutes is, when its bound ends and the watchdogs seat says its machine is heard, the finding
// call-unanswered in the indexer's findings, stamped with the wait's clock (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, heard func(context.Context, string) (bool, error), logf func(string, ...any)) *enginereads.ReadsCalls {
c := &enginereads.ReadsCalls{Keeper: r.keeper, Heard: heard, Bound: mcc.CallUnansweredAfter, Logf: logf,
Nodes: func() []string { r.mu.Lock(); defer r.mu.Unlock(); return slices.Clone(r.nodes) },
Raise: func(ctx context.Context, u enginereads.UnansweredCall) error {
kv := findings()
if kv == nil {
return process.ErrNotYet
}
value, err := mcc.EncodeFinding(o.Finding(index.Module, "", time.Now().UTC()))
o := mcc.CallUnanswered(enginereads.CallID(u.Node), enginereads.Invoke, u.Node, u.Words, "indexer."+mcc.KindCallUnanswered, u.Since)
o.Needs = index.EngineReadsTool + " with ask_again asks the node again now and shows its statement."
value, err := mcc.EncodeFinding(o.Finding(index.Module, "", u.At))
if err == nil {
_, err = kv.Put(ctx, o.Key(), value)
}
return err
},
Clear: func(ctx context.Context, id string) error {
Clear: func(ctx context.Context, node string) error {
kv := findings()
if kv == nil {
return nil
}
key := mcc.Key(mcc.ScopeCall, id, mcc.KindCallUnanswered)
key := mcc.Key(mcc.ScopeCall, enginereads.CallID(node), mcc.KindCallUnanswered)
if _, err := kv.Get(ctx, key); err != nil {
if errors.Is(err, jetstream.ErrKeyNotFound) {
return nil
@@ -46,31 +71,12 @@ func (r *engineReads) callsOf(ctx context.Context, findings func() jetstream.Key
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)
}}
if bounds != nil {
c.Arm = func(ctx context.Context, node string, ends time.Time) error {
return bounds.Arm(ctx, enginereads.BoundPrefix+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) }()
}
modules/mesh-delivery-indexer/go.mod
+1 -1
View File
@@ -68,7 +68,7 @@ require (
require (
git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhandacts v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/clients/containerimageregistryclient v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-delivery/contracts/go/seats/containerimageregistry v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-delivery/contracts/go/seats/ociartifactregistry v0.0.0-00010101000000-000000000000 // indirect
modules/mesh-delivery-indexer/manifest_test.go
+2 -3
View File
@@ -137,8 +137,7 @@ 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.
// 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) {
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) {
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
@@ -158,7 +157,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
// Three invokes: the file of a registered module's manifest and lock files (registered.go), the forge's trees (the named paths' objects, since the indexer is the one reader of a manifest's paths:
// ADR 0499 clauses 2 and 3, ADR 0495 clause 1; and the module directories of the trunk, issue 1900) and whether a
// commit is in another (issue 1900).
if want := []string{"mesh-delivery-assigner.assignments", "gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file", "gitea.gitea_open_pulls", "mesh-delivery-checker.checker_tries", "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition", "seat:node-engine.manifest-reads", "seat:mesh-hand-acts.hand-act"}; !reflect.DeepEqual(m.Invokes, want) {
if want := []string{"mesh-delivery-assigner.assignments", "gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file", "gitea.gitea_open_pulls", "mesh-delivery-checker.checker_tries", "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition", "seat:node-engine.manifest-reads", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence"}; !reflect.DeepEqual(m.Invokes, want) {
t.Errorf("invokes = %v, want %v", m.Invokes, want)
}
// 1945: indexer_take_in, served by its tools bundle through the takeins state; the process serves none. The same bundle
modules/mesh-delivery-indexer/module.json
+3 -3
View File
@@ -119,8 +119,7 @@
"mesh-build-node.toolchains",
"mesh-delivery-builder.build-log",
"mesh-delivery-checker.previews",
"mesh-delivery-gate.source",
"mesh-delivery-watchman.presence"
"mesh-delivery-gate.source"
],
"invokes": [
"mesh-delivery-assigner.assignments",
@@ -132,7 +131,8 @@
"mesh-delivery-forge-bridge.bridge_merged_message",
"seat:node-engine.composition",
"seat:node-engine.manifest-reads",
"seat:mesh-hand-acts.hand-act"
"seat:mesh-hand-acts.hand-act",
"seat:mesh-watchdogs.presence"
],
"own-secrets": {
"broker": "${dir:state}/broker"