B2 (watchman): the wants are followed at the holder of the seat mesh-wants, the indexer's tools at the holder of mesh-catalog

FollowSeat(mesh-wants, wants) replaces the follow by the assigner's module; the pulse and stuck keys name the holder's real
bucket taken at each use (WantsAt, AwaitsWants), a start never waiting for it and the seeded keys of any `<holder>_wants`
kept until the late read (A-321). indexer_pending and indexer_verdicts are asked of the holder of mesh-catalog on each
call. The one catch-all of the seats' events is one handler per seat (A-322), source-moved taken from the branches
follower (AdoptEvent). The manifest consumes the seat's views and the seat-holders view.

Builds: design 47 G A-210, A-318, A-321, A-322, §428

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-12 23:47:59 +01:00
parent 8e06b6af65
commit ec7e672c1f
11 changed files with 228 additions and 60 deletions
@@ -10,6 +10,8 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildimage"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildnode"
"slices"
"strings"
"sync"
@@ -34,8 +36,11 @@ import (
// Consumed event names, as the consumes name them (the emitter is read from the subject's emitter token).
const (
eventAccepted = "manifest-accepted"
eventCreated = "created"
eventRefused = "refused"
// eventManifestRefused is the seat mesh-catalog's manifest-refused; meshGrantsSeat the seat of the store whose filling, full, room and waits it hears.
eventManifestRefused = "manifest-refused"
meshGrantsSeat = "mesh-grants"
eventCreated = "created"
eventRefused = "refused"
)
// consumed hands one consumed event to the watchman, by its subject: manifest-accepted, and the seat mesh-broker's created
@@ -85,15 +90,31 @@ func hearConsumed(ctx context.Context, js jetstream.JetStream, cred process.Acco
// one taker; and the binding asks show again at each bind of the hub's consumer, as its Register does.
brokerEvents := []string{seatdef.EventSubject(meshbroker.Seat, "recreated"), seatdef.EventSubject(meshbroker.Seat, "dropped")}
hub.OnBound(func(time.Time) { binding.Start() })
if err := hub.Handle("mesh.seat.*.event.>", func(m jetstream.Msg) {
handleSeat := func(m jetstream.Msg) {
if slices.Contains(brokerEvents, m.Subject()) {
binding.Hear(m)
return
}
consumed(w, m.Subject(), m.Data(), logf)
_ = m.Ack()
}); err != nil {
w.NotHeard("EVENTS", "the seats' events cannot be handled", err)
}
// One handler per seat it consumes, since a message has one taker (design 47 G A-322): the events the branches follower of the
// catalog half adopts (the seat mesh-catalog's registration-changed, the seat mesh-delivery's source-moved) are the follower's, and
// the watchman takes source-moved from it (AdoptEvent, main.go), never from a second handler.
var patterns []string
for _, seat := range []string{meshwants.Seat, meshGrantsSeat, meshpipelinerunner.Seat, nodeengine.Seat, meshbroker.Seat, nodebuildgo.Seat, nodebuildimage.Seat, nodebuildnode.Seat} {
patterns = append(patterns, "mesh.seat."+seat+".event.>")
}
for _, e := range []string{meshdelivery.EventRecordOpened, meshdelivery.EventRecordEnded, meshdelivery.EventFilling, meshdelivery.EventFull, meshdelivery.EventRoom} {
patterns = append(patterns, seatdef.EventSubject(meshdelivery.Seat, e))
}
for _, e := range []string{eventAccepted, eventManifestRefused, meshdelivery.EventFilling, meshdelivery.EventFull, meshdelivery.EventRoom} {
patterns = append(patterns, seatdef.EventSubject(meshcatalog.Seat, e))
}
for _, pattern := range patterns {
if err := hub.Handle(pattern, handleSeat); err != nil {
w.NotHeard("EVENTS", "the seats' events cannot be handled", err)
}
}
var mu sync.Mutex
opened := map[string]bool{}
@@ -21,6 +21,8 @@ import (
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/process"
@@ -67,7 +69,16 @@ accounts {
// families (DLV-M5, ADR 0615): the watchman (n1) follows both, never their buckets.
func serveOwners(t *testing.T, ctx context.Context, js jetstream.JetStream, nc *nats.Conn) {
t.Helper()
viewstest.Serve(t, ctx, js, nc, watch.AssignerModule, wantsState, watch.AssignerModule+".assigner_read_wants", "n1_"+watchmanModule)
// The wants are followed in the seat form, at the module the seat-holders view of the catalog half's claimant names for the seat
// mesh-wants (design 47 G §717, A-321): the stand-in holder's real bucket is `<holder>_wants`, which the watchman names its keys by.
viewstest.Serve(t, ctx, js, nc, testAssigner, meshwants.StateWants, testAssigner+"."+meshwants.StateWants, "n1_"+watchmanModule)
holders := viewstest.Serve(t, ctx, js, nc, testAssigner, "seat-holders", testAssigner+".seat-holders", "n1_"+watchmanModule)
for _, seat := range []string{"mesh-module-assigner", meshwants.Seat} {
if _, err := holders.Put(ctx, seat, []byte(`{"seat":"`+seat+`","holders":[{"module":"`+testAssigner+`","node":"n1"}]}`)); err != nil {
t.Fatal(err)
}
}
serveCatalogHalf(t, nc)
}
@@ -171,7 +182,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
{Bucket: own.Presence, History: 20},
{Bucket: own.Stuck, History: 20},
{Bucket: own.Pulse, LimitMarkerTTL: time.Second},
{Bucket: watch.AssignerWantsBucket, History: 20},
{Bucket: testWantsBucket, History: 20},
{Bucket: (watchmanModule + "_findings"), History: 5},
{Bucket: firstViewsBucket, History: 1},
} {
@@ -419,7 +430,7 @@ func anAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909
{Bucket: own.Presence, History: 20},
{Bucket: own.Stuck, History: 20},
{Bucket: own.Pulse, LimitMarkerTTL: time.Second},
{Bucket: watch.AssignerWantsBucket, History: 20},
{Bucket: testWantsBucket, History: 20},
{Bucket: (watchmanModule + "_findings"), History: 5},
{Bucket: firstViewsBucket, History: 1},
} {
@@ -500,7 +511,7 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
{Bucket: own.Presence, History: 20},
{Bucket: own.Stuck, History: 20},
{Bucket: own.Pulse, LimitMarkerTTL: time.Second},
{Bucket: watch.AssignerWantsBucket, History: 20},
{Bucket: testWantsBucket, History: 20},
{Bucket: (watchmanModule + "_findings"), History: 5},
{Bucket: firstViewsBucket, History: 1},
} {
@@ -522,7 +533,7 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
t.Fatal(err)
}
key, _ := meshwantsclient.WantKey(node, module)
if _, err := kvs[watch.AssignerWantsBucket].Put(ctx, key, body); err != nil {
if _, err := kvs[testWantsBucket].Put(ctx, key, body); err != nil {
t.Fatal(err)
}
}
@@ -548,7 +559,7 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
t.Cleanup(func() { stop(); <-done })
pulse := func(node, module string) string {
wk, _ := meshwantsclient.WantKey(node, module)
pk, err := meshwatchdogsclient.PulseKey(watch.AssignerWantsBucket, wk)
pk, err := meshwatchdogsclient.PulseKey(testWantsBucket, wk)
if err != nil {
t.Fatal(err)
}
@@ -581,9 +592,39 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
}
for _, m := range []string{"web"} {
wk, _ := meshwantsclient.WantKey("n2", m)
sk, _ := meshwatchdogsclient.StuckKey(watch.AssignerWantsBucket, wk)
sk, _ := meshwatchdogsclient.StuckKey(testWantsBucket, wk)
if _, err := kvs[own.Stuck].Get(ctx, sk); err == nil {
t.Errorf("a stuck key was written for the want of %s that was applied", m)
}
}
}
// testAssigner is the stand-in holder of the seats mesh-wants and mesh-module-assigner; testWantsBucket the wants' bucket at it.
const (
testAssigner = "acme-assigner"
testWantsBucket = testAssigner + "_" + meshwants.StateWants
)
// serveCatalogHalf answers the catalog half's first views (design 47 G §614): the gate's one source branch, the stand-in
// registered on it, and its trunk revision whose entry claims mesh-module-assigner.
func serveCatalogHalf(t *testing.T, nc *nats.Conn) {
t.Helper()
respond := func(a any) nats.MsgHandler {
raw, _ := json.Marshal(map[string]any{"result": a, "node": "n0"})
return func(m *nats.Msg) { _ = m.Respond(raw) }
}
revision := meshcatalog.RevisionAnswer{Revision: &meshcatalog.CatalogRevision{Revision: 1, Kind: meshcatalogclient.KindTrunk},
Entries: []meshcatalog.CatalogEntry{{Module: testAssigner, ManifestDigest: "d1", Manifest: map[string]any{"module": testAssigner,
"claims": []any{map[string]any{"name": "mesh-module-assigner", "scope": "mesh"}}}}}}
registrations := meshcatalog.RegistrationsAnswer{Registrations: []meshcatalog.Registration{{Module: testAssigner, Version: 1, Branches: []string{"release"}}}}
for subject, handle := range map[string]nats.MsgHandler{"mesh.seat.mesh-catalog.call.revision": respond(revision),
"mesh.seat.mesh-catalog.call.registrations": respond(registrations), "mesh.seat.mesh-delivery.call.source": respond(map[string]any{"branches": []string{"release"}})} {
for _, s := range []string{subject, subject + ".>"} {
sub, err := nc.Subscribe(s, handle)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = sub.Unsubscribe() })
}
}
}
@@ -10,6 +10,10 @@ import (
"encoding/json"
"errors"
"fmt"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
"io"
"log"
"os"
@@ -143,7 +147,7 @@ func settle(ctx context.Context, kv jetstream.KeyValue, each func(jetstream.KeyV
// readLate reads an owner's state the watchman follows by its first view (pic), on a goroutine of its own once that is
// answered, so an owner that is down never holds the start: every key it holds now, then the start's end for its bucket
// (watch.ReadLate), then each change until ctx ends.
func readLate(ctx context.Context, w *watch.Watchman, pic *views.Picture, bucket string, each func(jetstream.KeyValueEntry)) error {
func readLate(ctx context.Context, w *watch.Watchman, pic *views.Picture, bucket func() string, each func(jetstream.KeyValueEntry)) error {
select {
case <-ctx.Done():
return nil
@@ -156,7 +160,7 @@ func readLate(ctx context.Context, w *watch.Watchman, pic *views.Picture, bucket
}
return err
}
w.ReadLate(bucket)
w.ReadLate(bucket())
return follow(ctx, sw, each)
}
@@ -189,11 +193,6 @@ func createdOr(e jetstream.KeyValueEntry, at time.Time) time.Time {
return at
}
const (
wantsState = "wants"
// The owners' views families, as their manifests' `streams` name them.
)
// run is the program: it returns when ctx ends or the bus cannot be used.
func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
logger := log.New(logOut, "mesh-delivery-watchman: ", log.LstdFlags)
@@ -257,15 +256,31 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
defer held.w.Close()
reader := &views.ProcessReader{Module: cred.Module, Call: calls.Call, Logf: logger.Printf, Hub: hub,
Findings: func() views.WireWriter { return held.w }}
wants, err := reader.Follow(watch.AssignerModule, wantsState, watch.AssignerModule+".assigner_read_wants")
// The catalog half's claimant says each seat's holder (design 47 G §717, A-298): the indexer's tools are asked of the holder of
// the seat mesh-catalog when asked.
branches, err := meshcatalogclient.FollowBranchesProcess(reader, calls.Call)
if err != nil {
return err
}
branches.Logf = logger.Printf
holders := branches.SeatHolders(ctx)
// The assigner's wants are followed in the seat form, at the module holding the seat mesh-wants (design 47 G §717, A-318, A-321):
// the bucket a pulse or a stuck key names is that holder's real bucket, taken at each use, and a start never waits for it.
wants, err := reader.FollowSeat(meshwants.Seat, meshwants.StateWants, holders)
if err != nil {
return err
}
w := watch.New(cred.Module, &kvOut{ctx: ctx, presence: kvs.presence, pulse: kvs.pulse, stuk: kvs.stuck, findings: held, js: js, cred: cred}, time.Now)
w.Logf(logger.Printf)
// The seat mesh-delivery's source-moved is the branches follower's event (one taker, design 47 G A-322): the watchman takes it from the
// follower's subscription, beside the follower's handler.
if err := reader.AdoptEvent(meshdelivery.Seat, meshdelivery.EventSourceMoved, func(_ context.Context, data []byte) error { w.SourceMoved(data); return nil }); err != nil {
return err
}
// What the wants and the modules hold is ended by their own late read (watch.ReadLate), never by the start's end.
w.Awaits(watch.AssignerWantsBucket)
w.WantsAt(func() string { return reader.HolderBucket(meshwants.Seat, meshwants.StateWants) })
w.AwaitsWants()
// A test shortens how long a move may take; the mesh never sets it.
if v := strings.TrimSpace(getenv("MESH_DELIVERY_PULSE_BOUND")); v != "" {
d, err := time.ParseDuration(v)
@@ -382,8 +397,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
if calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls")); err != nil {
logger.Printf("take-in: no connection to ask %s (%v), so a wait the indexer held before this start is said only once a take-in event of it is heard", pendingTool, err)
} else {
seedTakeIn(w, calls.Call, logger.Printf)
seedVerdicts(w, calls.Call, logger.Printf)
atCatalog := atHolder(calls.Call, holders, meshcatalog.Seat) // the indexer holds the seat mesh-catalog
seedTakeIn(w, atCatalog, logger.Printf)
seedVerdicts(w, atCatalog, logger.Printf)
calls.Close()
}
// The take-in events are the indexer's ordinary events (its emits), heard on the shared events stream through this
@@ -418,7 +434,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
hearConsumed(ctx, js, cred, hub, w, binding, logger.Printf)
defer hub.Stop()
go func() {
errs <- readLate(ctx, w, wants, watch.AssignerWantsBucket, func(e jetstream.KeyValueEntry) {
errs <- readLate(ctx, w, wants, func() string { return reader.HolderBucket(meshwants.Seat, meshwants.StateWants) }, func(e jetstream.KeyValueEntry) {
w.Want(e.Key(), e.Value(), opOf(e), createdOr(e, time.Now()))
})
}()
@@ -462,7 +478,19 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
const eventsStream = "EVENTS"
// pendingTool is the indexer's tool that answers the waits it holds (ADR 0633 §5), as the view asks it.
const pendingTool = watch.IndexerModule + ".indexer_pending"
const pendingTool = "indexer_pending"
// atHolder asks a tool of the first module that holds the seat when it is asked (design 47 G A-210, §428), as the view's tools do: the
// holder comes from the seat-holders view of the catalog half's claimant, found for each call, and none is an error that says so.
func atHolder(call func(key string, body any) (json.RawMessage, error), holders func(seat string) []string, seat string) func(key string, body any) (json.RawMessage, error) {
return func(tool string, body any) (json.RawMessage, error) {
h := holders(seat)
if len(h) == 0 {
return nil, fmt.Errorf("the holder of %s is not found, so %s is not asked", seat, tool)
}
return call(h[0]+"."+tool, body)
}
}
// seedTakeIn asks the indexer the waits it holds now and starts each (Watchman.SeedTakeIn). An indexer that does not
// answer is said: the waits are then known only from the events heard after.
@@ -5,6 +5,8 @@ import (
"encoding/json"
"errors"
"fmt"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
"io"
"strings"
"testing"
@@ -137,6 +139,12 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6
}
w.TakeIn(takein.TakeInPendingEvent, body)
seedTakeIn(w, call, func(string, ...any) {})
var asked string
seedTakeIn(w, atHolder(func(key string, _ any) (json.RawMessage, error) { asked = key; return nil, errors.New("no answer") },
func(seat string) []string { return []string{"acme-indexer"} }, meshcatalog.Seat), func(string, ...any) {})
if asked != "acme-indexer."+pendingTool { // the tool is asked of the holder of the seat mesh-catalog, found when asked
t.Errorf("%s: asked %q, want the holder's tool", name, asked)
}
if !out.findings[old.Key()] || out.pulseDeletes != 0 {
t.Errorf("%s: finding stands %v, signals ended %d: want the previous run's finding and the held wait kept", name, out.findings[old.Key()], out.pulseDeletes)
}
@@ -147,10 +155,10 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6
// the time the owner's bus wrote the key, never the watchman's start.
written := time.Date(2026, 11, 4, 9, 0, 0, 0, time.UTC)
r := &views.ProcessReader{Module: watchmanModule, Call: func(string, any) (json.RawMessage, error) {
return json.Marshal(views.Answer{V: views.Version, Owner: watch.AssignerModule, State: wantsState, Revision: 3, Created: written,
return json.Marshal(views.Answer{V: views.Version, Owner: "acme-assigner", State: meshwants.StateWants, Revision: 3, Created: written,
At: written.Add(time.Hour), Values: map[string]json.RawMessage{"ace.mail": json.RawMessage(`{}`)}, Written: map[string]time.Time{"ace.mail": written}})
}}
p, err := r.Follow(watch.AssignerModule, wantsState, watch.AssignerModule+".assigner_read_wants")
p, err := r.Follow("acme-assigner", meshwants.StateWants, "acme-assigner.wants")
if err != nil {
t.Fatal(err)
}
@@ -112,16 +112,19 @@ func (w *Watchman) Want(key string, value []byte, op Op, at time.Time) {
}
w.mu.Lock()
defer w.mu.Unlock()
if w.wantsBucket() == "" {
return // no holder of the seat mesh-wants is followed: the key names no bucket (A-321)
}
switch op {
case Expired:
return
case Delete:
delete(w.wants, key)
delete(w.held, key)
w.moved(AssignerWantsBucket, key)
w.moved(w.wantsBucket(), key)
return
}
w.stands(AssignerWantsBucket, key)
w.stands(w.wantsBucket(), key)
want, err := meshwantsclient.DecodeWant(value)
if err != nil {
w.say("the want %s cannot be read, and is left: %v", key, err)
@@ -130,7 +133,7 @@ func (w *Watchman) Want(key string, value []byte, op Op, at time.Time) {
if want.Removed {
delete(w.wants, key)
delete(w.held, key)
w.moved(AssignerWantsBucket, key)
w.moved(w.wantsBucket(), key)
return
}
w.wants[key] = want.Fp
@@ -141,26 +144,26 @@ func (w *Watchman) Want(key string, value []byte, op Op, at time.Time) {
at = w.now()
}
w.held[key] = true
w.mirrorKept(AssignerWantsBucket, key,
w.mirrorKept(w.wantsBucket(), key,
describe("the hold on %s on %s at %s (rule %d) to end, bound %s: %s", module, node, shortFP(want.Fp), want.Rule, w.stepBound, want.Words),
"held|"+want.Fp+"|"+want.Newest, at)
return
}
if w.held[key] {
delete(w.held, key)
if pk, err := meshwatchdogsclient.PulseKey(AssignerWantsBucket, key); err == nil {
if pk, err := meshwatchdogsclient.PulseKey(w.wantsBucket(), key); err == nil {
delete(w.starts, pk)
}
w.stuckMoved(AssignerWantsBucket, key) // the hold ended: its finding is cleared
w.stuckMoved(w.wantsBucket(), key) // the hold ended: its finding is cleared
}
if w.have(node, module) == want.Fp {
w.moved(AssignerWantsBucket, key)
w.moved(w.wantsBucket(), key)
return
}
if at.IsZero() {
at = w.now()
}
w.mirrorWithin(w.stepBound, AssignerWantsBucket, key,
w.mirrorWithin(w.stepBound, w.wantsBucket(), key,
describe("the node agent of %s to run %s at %s", node, module, shortFP(want.Fp)), want.Fp, at, 0)
}
@@ -188,7 +191,7 @@ func (w *Watchman) EngineRunning(node, module string, body []byte) {
return
}
if fp, ok := w.wants[wk]; ok && fp == r.Fingerprint {
w.moved(AssignerWantsBucket, wk)
w.moved(w.wantsBucket(), wk)
}
}
@@ -2,20 +2,21 @@ package watch
import (
meshwatchdogs "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
)
// The holders and buckets the watchman names, taken from its own manifest (module.json: its name, its `reads`, its
// states and the tools it invokes) and the seats' state names; a unit row checks them against it (TestManifestStateAndReads).
// No contract names a holder (novox/hq ADR 0625 §4, ADR 0634 §1).
const (
AssignerModule = "mesh-delivery-assigner"
// ProvisionsBucket is the seat mesh-provisioner's state provisions (reads: mesh-provisioner.provisions).
ProvisionsBucket = "mesh-provisioner_provisions"
// IndexerModule is the holder of the seat mesh-catalog, named from this module's manifest (consumes and invokes of mesh-delivery-indexer).
IndexerModule = "mesh-delivery-indexer"
// AssignerWantsBucket is the assigner's state wants, followed by its views family (consumes: mesh-delivery-assigner.views.wants).
AssignerWantsBucket = AssignerModule + "_wants"
// WantsState is the state wants of the seat mesh-wants, followed in the seat form (consumes: seat:mesh-wants.views.wants); the
// bucket a pulse or stuck key of it names is the holder's real bucket `<holder>_wants`, taken at each use (WantsAt, A-321).
WantsState = meshwants.StateWants
)
// OwnBuckets are the watchman's own states of the seat mesh-watchdogs, `<module>_<state>`.
@@ -23,8 +23,9 @@ const KindManifestOutgrown = FindingManifestOutgrown
// `manifest-outgrown` or, with none, clears it and any refusal. The watchman seeds both from the indexer's verdicts at its start
// (VerdictsTool). `manifest-waits` is no longer written: every revision is judged by a validator built from its own heads (§16).
// VerdictsTool is the indexer's tool that lists its verdicts, which seeds the findings at the start.
const VerdictsTool = IndexerModule + ".indexer_verdicts"
// VerdictsTool is the indexer's tool that lists its verdicts, which seeds the findings at the start; it is asked of the holder of
// the seat mesh-catalog when it is asked (design 47 G A-210, §428), never of a module named here.
const VerdictsTool = "indexer_verdicts"
// VerdictRow is one verdict as the tool lists it (the fields the watchman reads).
type VerdictRow struct {
@@ -5,6 +5,7 @@ package watch
import (
"fmt"
"strings"
"sync"
"time"
@@ -106,6 +107,10 @@ type Watchman struct {
// pulses, stuck keys and findings wait for its own end (ReadLate), so a restart while its owner is down neither drops
// nor raises again what it holds.
awaited map[string]bool
// wantsAt names the bucket of the assigner's wants as the holder of the seat mesh-wants keeps it, "" while none is followed
// (WantsAt, design 47 G A-321); awaitsWants keeps the seeded entries of any `<holder>_wants` bucket until the late read.
wantsAt func() string
awaitsWants bool
// heads is, by watched branch, the head still waiting for its pipeline (pipelines.go).
heads map[string]head
// pendingBranch is, by key, the watched branch of each open take-in wait that names one (takein.go, design 47 G §497).
@@ -313,7 +318,7 @@ func (w *Watchman) Reconcile() {
w.mu.Lock()
defer w.mu.Unlock()
for pk, p := range w.pulses {
if p.seeded && !w.awaited[p.bucket] {
if p.seeded && !w.awaits(p.bucket) {
w.deletePulse(pk)
}
}
@@ -338,7 +343,7 @@ func (w *Watchman) ReadWhole(bucket string) {
// stands records a mirrored key read standing during the start.
func (w *Watchman) stands(bucket, key string) {
if w.reconciled && !w.awaited[bucket] {
if w.reconciled && !w.awaits(bucket) {
return
}
if w.standing[bucket] == nil {
@@ -349,7 +354,7 @@ func (w *Watchman) stands(bucket, key string) {
func (w *Watchman) closeGone() {
for sk, s := range w.stuck {
if s.Moved || w.awaited[s.Bucket] || !w.readWhole[s.Bucket] || w.standing[s.Bucket][s.Key] {
if s.Moved || w.awaits(s.Bucket) || !w.readWhole[s.Bucket] || w.standing[s.Bucket][s.Key] {
continue
}
w.closeStuck(sk, s)
@@ -358,6 +363,9 @@ func (w *Watchman) closeGone() {
for b := range w.awaited {
standing[b] = w.standing[b]
}
if b := w.wantsBucket(); w.awaitsWants && b != "" {
standing[b] = w.standing[b]
}
w.standing, w.readWhole, w.reconciled = standing, map[string]bool{}, true
}
@@ -383,27 +391,67 @@ func (w *Watchman) Awaits(bucket string) {
func (w *Watchman) ReadLate(bucket string) {
w.mu.Lock()
defer w.mu.Unlock()
if !w.awaited[bucket] {
if !w.awaits(bucket) {
return
}
// The wants' late read ends the start for every `<holder>_wants` bucket a previous run named (A-321): the keys of a holder that
// is gone are not standing in the one followed now.
of := func(b string) bool {
return b == bucket || (w.awaitsWants && isWantsBucket(bucket) && isWantsBucket(b))
}
for pk, p := range w.pulses {
if p.seeded && p.bucket == bucket {
if p.seeded && of(p.bucket) {
w.deletePulse(pk)
}
}
for sk, s := range w.stuck {
if !s.Moved && s.Bucket == bucket && !w.standing[bucket][s.Key] {
if !s.Moved && of(s.Bucket) && !w.standing[bucket][s.Key] {
w.closeStuck(sk, s)
}
}
for key := range w.seededFindings {
if f, err := DecodeFinding(w.findings[key]); err == nil && f.Source == bucket {
if f, err := DecodeFinding(w.findings[key]); err == nil && of(f.Source) {
delete(w.seededFindings, key)
w.unfind(key)
}
}
delete(w.awaited, bucket)
delete(w.standing, bucket)
if w.awaitsWants && isWantsBucket(bucket) {
w.awaitsWants = false
}
}
// WantsAt says how the bucket of the assigner's wants is named, the holder's real bucket `<holder>_wants` at each use ("" while the
// seat mesh-wants has no followed holder; design 47 G A-321, A-303).
func (w *Watchman) WantsAt(bucket func() string) {
w.mu.Lock()
defer w.mu.Unlock()
w.wantsAt = bucket
}
// AwaitsWants awaits the wants as Awaits does a bucket (design 47 G A-321): a start never waits for the holder, and the seeded
// pulses, stuck keys and findings of any `<holder>_wants` bucket are kept until the late read of the resolved one. Called before
// Reconcile.
func (w *Watchman) AwaitsWants() {
w.mu.Lock()
defer w.mu.Unlock()
w.awaitsWants = true
}
func isWantsBucket(b string) bool { return strings.HasSuffix(b, "_"+WantsState) }
// awaits says bucket is read only after the start's end (Awaits, WantsAt).
func (w *Watchman) awaits(bucket string) bool {
return w.awaited[bucket] || (w.awaitsWants && isWantsBucket(bucket))
}
// wantsBucket is the bucket of the wants now, "" while the seat mesh-wants has no followed holder.
func (w *Watchman) wantsBucket() string {
if w.wantsAt == nil {
return ""
}
return w.wantsAt()
}
// reconcileFindings ends the start for the findings: one read at the start that what the watchman read since did not
@@ -415,7 +463,7 @@ func (w *Watchman) reconcileFindings() {
kept[due.Finding.Key()] = true
}
for key := range w.seededFindings {
if f, err := DecodeFinding(w.findings[key]); err == nil && w.awaited[f.Source] {
if f, err := DecodeFinding(w.findings[key]); err == nil && w.awaits(f.Source) {
continue // ended by ReadLate
}
delete(w.seededFindings, key)
@@ -96,6 +96,7 @@ func (c *clock) now() time.Time { return c.t }
func setup() (*Watchman, *fakeOut, *clock) {
o, c := newOut(), &clock{t: t0}
w := New(testModule, o, c.now)
w.WantsAt(testWants)
w.SetStepBound(meshwatchdogsclient.PulseMoveBound) // the older tests move a step as they move any key: stuck at the pulse's bound
return w, o, c
}
@@ -551,6 +552,7 @@ func TestAnEventThatDoesNotReadOrIsNoBuildSeatsIsLeft(t *testing.T) { // 1908
func TestARunningStepThatSaysItsProgressIsNeverStuckAndOneWhoseHolderStopsIs(t *testing.T) { // design 47 G §496, §602
o, c := newOut(), &clock{t: t0}
w := New(testModule, o, c.now)
w.WantsAt(testWants)
askTaken(w, "askid1", t0)
k := pulseKeyOf(t, checksKey, "askid1")
if got := o.ttls[k]; got != ProgressBound {
@@ -583,6 +585,7 @@ func TestARunningStepThatSaysItsProgressIsNeverStuckAndOneWhoseHolderStopsIs(t *
func TestARunningStepPastItsBoundIsStuck(t *testing.T) { // 1908
o, c := newOut(), &clock{t: t0}
w := New(testModule, o, c.now)
w.WantsAt(testWants)
askProgress(w, "askid1", t0)
c.t = t0.Add(6 * time.Minute)
w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t)
@@ -592,7 +595,7 @@ func TestARunningStepPastItsBoundIsStuck(t *testing.T) { // 1908
}
// and one found past its bound at the watchman's start is stuck at once
o2, c2 := newOut(), &clock{t: t0.Add(time.Hour)}
askProgress(New(testModule, o2, c2.now), "askid1", t0)
askProgress(newWith(New(testModule, o2, c2.now)), "askid1", t0)
if _, ok := o2.stuck[stuckKeyOf(t, checksKey, "askid1")]; !ok || len(o2.pulses) != 0 {
t.Errorf("stuck %v pulses %v", o2.stuck, o2.pulses)
}
@@ -664,6 +667,7 @@ func TestAHaveThatCameBeforeTheWantEndsItAtOnceAndARemovedWantHasNoPulse(t *test
func TestAWantNotAppliedByItsBoundIsStuckAndAKeyDeletedEndsIt(t *testing.T) {
o, c := newOut(), &clock{t: t0}
w := New(testModule, o, c.now)
w.WantsAt(testWants)
w.Want("want.ace.web", wantValue(t, fpNew, false), Put, t0)
pk := pulseKeyOf(t, AssignerWantsBucket, "want.ace.web")
c.t = t0.Add(31 * time.Minute)
@@ -691,7 +695,7 @@ func TestAStuckKeyWhoseKeyIsGoneIsClosedAtTheStart(t *testing.T) {
standing func(w *Watchman) // what the start reads standing
readWhole bool
closed bool
late bool // read after the start's end (Awaits, ReadLate)
late bool // read after the start's end (AwaitsWants, ReadLate)
}{
{"a want deleted while nobody heard", AssignerWantsBucket, wantKey, func(*Watchman) {}, true, true, false},
{"a want still standing", AssignerWantsBucket, wantKey, func(w *Watchman) { w.Want(wantKey, wantValue(t, fpNew, false), Put, t0) }, true, false, false},
@@ -699,6 +703,9 @@ func TestAStuckKeyWhoseKeyIsGoneIsClosedAtTheStart(t *testing.T) {
// they hold waits for it, and is then reconciled as at the start.
{"a want deleted while nobody heard, read from a late first view", AssignerWantsBucket, wantKey, func(*Watchman) {}, true, true, true},
{"a want still standing, read from a late first view", AssignerWantsBucket, wantKey, func(w *Watchman) { w.Want(wantKey, wantValue(t, fpNew, false), Put, t0) }, true, false, true},
// A-321: the holder is not followed at the start, so an earlier run's `<holder>_wants` keys wait for the late read of the
// bucket resolved after it, and a holder that is gone left keys that no longer stand.
{"a want of an earlier holder, kept at the start and closed by the late read", "old-assigner_" + WantsState, wantKey, func(*Watchman) {}, true, true, true},
} {
t.Run(row.name, func(t *testing.T) {
w, o, c := setup()
@@ -720,7 +727,7 @@ func TestAStuckKeyWhoseKeyIsGoneIsClosedAtTheStart(t *testing.T) {
fk := FindingKey(ScopeModule, "gone", KindManifestRefused)
o.findings[fk], _ = DecodeFinding(fv)
w.SeedFinding(fk, fv)
w.Awaits(row.bucket)
w.AwaitsWants()
w.Reconcile()
_, open := o.findings[FindingKey(ScopeDelivery, sk, KindStuck)]
_, seeded := o.findings[fk]
@@ -728,7 +735,7 @@ func TestAStuckKeyWhoseKeyIsGoneIsClosedAtTheStart(t *testing.T) {
t.Fatalf("the start's end took what an awaited bucket holds: moved %v, stuck open %v, finding %v, pulses deleted %v", o.stuck[sk].Moved, open, seeded, o.pulseDeletes)
}
row.standing(w)
w.ReadLate(row.bucket)
w.ReadLate(AssignerWantsBucket) // the bucket the holder followed now keeps
if _, seeded := o.findings[fk]; seeded || !slices.Contains(o.pulseDeletes, pk) {
t.Errorf("the late read kept the finding (%v) or the pulse (deleted %v) nothing made again", seeded, o.pulseDeletes)
}
@@ -756,3 +763,11 @@ func TestABusUserIsSaidByItsModuleWithoutAModulesNameInTheCode(t *testing.T) {
}
}
}
// AssignerWantsBucket is the bucket of the wants as the tests' holder of the seat mesh-wants keeps it: `<holder>_wants`, taken at each
// use through WantsAt, never a constant of the code (design 47 G A-321).
const AssignerWantsBucket = "acme-assigner_" + WantsState
func testWants() string { return AssignerWantsBucket }
func newWith(w *Watchman) *Watchman { w.WantsAt(testWants); return w }
@@ -3,6 +3,7 @@ package manifest
import (
"bytes"
"encoding/json"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
"io/fs"
"os"
"path/filepath"
@@ -126,11 +127,11 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
}
// The join: its first view asked again by hand is a hand act, recorded through the seat (views.StatusTools). The waits the
// indexer holds at the watchman's start are asked of its tool (ADR 0633 §5), a verb edge as the view's.
if want := []string{"mesh-delivery-assigner.assigner_read_wants", "mesh-delivery-indexer.indexer_pending", "mesh-delivery-indexer.indexer_verdicts", "seat:mesh-hand-acts.hand-act", "seat:mesh-conditions.conditions", "seat:mesh-broker.show"}; !reflect.DeepEqual(m.Invokes, want) {
if want := []string{"mesh-delivery-indexer.indexer_pending", "mesh-delivery-indexer.indexer_verdicts", "seat:mesh-hand-acts.hand-act", "seat:mesh-conditions.conditions", "seat:mesh-broker.show"}; !reflect.DeepEqual(m.Invokes, want) {
t.Errorf("invokes = %v, want %v", m.Invokes, want)
}
// DLV-M5 its first views, 1585/T20 the indexer's take-in events, and DLV-W1 (ADR 0595 §4): the accepted verdicts and the bus module's created and refused, durably.
if want := []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken", "mesh-catalog.manifest-accepted", "mesh-catalog.manifest-refused", "mesh-delivery.source-moved", "mesh-delivery.filling", "mesh-delivery.full", "mesh-delivery.room", "mesh-delivery.record-opened", "mesh-delivery.record-ended", "mesh-catalog.filling", "mesh-catalog.full", "mesh-catalog.room", "mesh-wants.filling", "mesh-wants.full", "mesh-wants.room", "mesh-grants.filling", "mesh-grants.full", "mesh-grants.room", "node-build-go.ask-taken", "node-build-go.step-progress", "node-build-go.step-ended", "node-build-image.ask-taken", "node-build-image.step-progress", "node-build-image.step-ended", "node-build-node.ask-taken", "node-build-node.step-progress", "node-build-node.step-ended", "mesh-pipeline-runner.pipeline-started", "mesh-pipeline-runner.step-started", "mesh-pipeline-runner.step-ended", "mesh-pipeline-runner.wait-held", "mesh-pipeline-runner.wait-ended", "mesh-wants.wait-held", "mesh-wants.wait-ended", "mesh-grants.wait-held", "mesh-grants.wait-ended", "mesh-delivery-builder.wait-held", "mesh-delivery-builder.wait-ended", "node-engine.wait-held", "node-engine.wait-ended", "node-engine.running", "mesh-broker.created", "mesh-broker.refused", "mesh-broker.recreated", "mesh-broker.dropped", "mesh-conditions.findings-asked"}; !reflect.DeepEqual(m.Consumes, want) {
if want := []string{"seat:mesh-wants.views.wants", "seat:mesh-module-assigner.views.seat-holders", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision", "mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken", "mesh-catalog.manifest-accepted", "mesh-catalog.manifest-refused", "mesh-delivery.source-moved", "mesh-delivery.filling", "mesh-delivery.full", "mesh-delivery.room", "mesh-delivery.record-opened", "mesh-delivery.record-ended", "mesh-catalog.filling", "mesh-catalog.full", "mesh-catalog.room", "mesh-wants.filling", "mesh-wants.full", "mesh-wants.room", "mesh-grants.filling", "mesh-grants.full", "mesh-grants.room", "node-build-go.ask-taken", "node-build-go.step-progress", "node-build-go.step-ended", "node-build-image.ask-taken", "node-build-image.step-progress", "node-build-image.step-ended", "node-build-node.ask-taken", "node-build-node.step-progress", "node-build-node.step-ended", "mesh-pipeline-runner.pipeline-started", "mesh-pipeline-runner.step-started", "mesh-pipeline-runner.step-ended", "mesh-pipeline-runner.wait-held", "mesh-pipeline-runner.wait-ended", "mesh-wants.wait-held", "mesh-wants.wait-ended", "mesh-grants.wait-held", "mesh-grants.wait-ended", "mesh-delivery-builder.wait-held", "mesh-delivery-builder.wait-ended", "node-engine.wait-held", "node-engine.wait-ended", "node-engine.running", "mesh-broker.created", "mesh-broker.refused", "mesh-broker.recreated", "mesh-broker.dropped", "mesh-conditions.findings-asked"}; !reflect.DeepEqual(m.Consumes, want) {
t.Errorf("consumes = %v, want %v", m.Consumes, want)
}
// 1952/WM1: the watchman holds the seat mesh-watchdogs at mesh scope, its first seat (novox/hq ADR 0564 §1, §4).
@@ -173,10 +174,9 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
"presence": {watch.BucketsOf("a-watchman").Presence, "a-watchman_" + m.State[0].Name},
"pulse": {watch.BucketsOf("a-watchman").Pulse, "a-watchman_" + m.State[1].Name},
"stuck": {watch.BucketsOf("a-watchman").Stuck, "a-watchman_" + m.State[2].Name},
"wants": {watch.AssignerWantsBucket, bucket(m.Consumes, "mesh-delivery-assigner.views.wants")},
"wants": {meshwants.Seat + "_" + watch.WantsState, strings.TrimPrefix(bucket(m.Consumes, "seat:mesh-wants.views.wants"), "seat:")},
"provisions": {watch.ProvisionsBucket, bucket(m.Reads, "mesh-provisioner.provisions")},
"assigner": {watch.AssignerModule + ".assigner_read_wants", entry(m.Invokes, "mesh-delivery-assigner.assigner_read_wants")},
"indexer": {watch.IndexerModule + ".indexer_pending", entry(m.Invokes, "mesh-delivery-indexer.indexer_pending")},
"indexer": {"mesh-delivery-indexer.indexer_pending", entry(m.Invokes, "mesh-delivery-indexer.indexer_pending")},
} {
if row.got != row.want {
t.Errorf("the name for %s is %q, its manifest's is %q", name, row.got, row.want)
+4 -2
View File
@@ -19,7 +19,6 @@
"mesh-provisioner.provisions"
],
"invokes": [
"mesh-delivery-assigner.assigner_read_wants",
"mesh-delivery-indexer.indexer_pending",
"mesh-delivery-indexer.indexer_verdicts",
"seat:mesh-hand-acts.hand-act",
@@ -27,7 +26,10 @@
"seat:mesh-broker.show"
],
"consumes": [
"mesh-delivery-assigner.views.wants",
"seat:mesh-wants.views.wants",
"seat:mesh-module-assigner.views.seat-holders",
"mesh-catalog.registration-changed",
"mesh-catalog.catalog-revision",
"mesh-delivery-indexer.take-in.pending",
"mesh-delivery-indexer.take-in.taken",
"mesh-catalog.manifest-accepted",