CAT-D4b, D5: the windows' driver moves the old-name and the direct-write windows on start, delivered and manifest-reads-changed

The windows' logic was built but nothing woke it, so both windows stayed open for ever. The program now reads every
node through node_haves and the node-engine seat's manifest-reads at its start, on each delivered and on each
manifest-reads-changed, steps both windows, says their closing once on distribution.events and raises the seven-day
finding. A build module's have counts as publishing through the seats when it runs a build heard publishing (the
tools bundle records the fingerprint of each publish's proved caller), so a paused build agent running the same
build holds nothing open.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
2026-11-04 12:55:03 +01:00
parent 75986b2960
commit 99512eade4
7 changed files with 552 additions and 4 deletions
@@ -125,6 +125,51 @@ func bootstrapCloser(windows kept[holder.Window]) func() {
})
}
// learned are the proved callers whose build this process recorded as publishing through the seats.
var learned sync.Map
// learnPublishing records the fingerprint the proved caller's node runs of its build module as one that publishes
// through the seats (holder.PublishingKey), once per caller and process: the windows' driver in the program counts a
// node's have running it as publishing, closing the direct-write window when every node's build modules do.
func learnPublishing(windows kept[holder.Window], caller string) {
node, module, ok := strings.Cut(caller, ".")
if !ok || strings.HasPrefix(caller, "person.") {
return
}
if _, done := learned.Load(caller); done {
return
}
raw, err := stdio.Ask("mesh-delivery-view.node_haves", map[string]any{"node": node, "module": module})
if err != nil {
logf("the build of %s that published cannot be read (%v): the direct-write window counts it at its next publish", caller, err)
return
}
var a struct {
Keys map[string]struct {
FP string `json:"fp"`
} `json:"keys"`
}
if json.Unmarshal(raw, &a) != nil || a.Keys[caller].FP == "" {
logf("node_haves names no build of %s: the direct-write window counts it at its next publish", caller)
return
}
pub := kept[holder.Publishing]{state: "windows", key: func(p holder.Publishing) string { return holder.PublishingKey }}
cur, _, err := pub.Get(holder.PublishingKey)
if err != nil {
logf("the publishing builds cannot be read: %v", err)
return
}
cur.Name = holder.PublishingKey
if cur.Learn(module, a.Keys[caller].FP) {
if err := pub.Put(cur); err != nil {
logf("the publishing build of %s is not kept: %v", caller, err)
return
}
logf("%s's build %s publishes through the seats", module, a.Keys[caller].FP)
}
learned.Store(caller, true)
}
func arg[T any](args map[string]any) (T, error) {
var v T
clean := map[string]any{}
@@ -159,6 +204,9 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool {
if err == nil && closeBootstrap != nil {
closeBootstrap()
}
if err == nil {
learnPublishing(windows, stdio.CallerOf(args))
}
return a, err
}},
stdio.Tool{Name: seat + "." + holder.VerbPublishPart, Description: "Send one chunk (at most 1 MiB) of one part of an open transfer, from the caller that opened it.",
+14 -2
View File
@@ -1,8 +1,8 @@
// Command distribution is distribution's program (novox/hq design 51): it serves the artifact store's port in front of
// the registry container, which listens on its node's loopback only. Reads (GET, HEAD) go through to every node's
// container runtime; a write is refused but while the direct-write window is open, never over a tag the seat holds
// published, or with the bootstrap's one-off credential (holder.Front). The window is distribution's state windows,
// written by its tools bundle; the program reads it on each write and holds no timer.
// published, or with the bootstrap's one-off credential (holder.Front). The windows are distribution's state windows:
// the program moves them (windows.go) and follows the direct-write window's record on each change; it holds no timer.
package main
import (
@@ -59,6 +59,18 @@ func run(ctx context.Context, getenv func(string) string) error {
if windows != nil {
go follow(ctx, windows, &open, logger)
go followBootstrap(ctx, windows, logger)
calls, err := process.Connect(cred)
if err != nil {
return err
}
defer calls.Close()
callers, err := holder.ParseCallers(getenv("MESH_CONFIG_CALLERS"))
if err != nil {
return err
}
findings := process.OptionalBucket(ctx, js, "distribution_findings", "a window held for seven days is said in the log only", logger.Printf)
go drive(ctx, js, cred, &holder.Driver{Delivery: busDelivery{calls}, State: kvWindows{windows}, Callers: callers,
Closed: sayClosed(js, cred), Finding: findingsKV(findings, logger.Printf)}, logger)
}
front := &holder.Front{Store: store, Window: &open, Logf: logger.Printf,
Published: func(name, tag string) (bool, error) {
@@ -0,0 +1,239 @@
package main
// The windows' driver (design 51 *The move*): at the program's start, on each `delivered` of the status view and on each
// node-engine's `manifest-reads-changed` (ADR 0618 §4) it reads every node through node_haves and the node-engine
// seat's manifest-reads, and steps the old-name and the direct-write windows in distribution's state windows. The
// program does it, not the tools bundle: it holds distribution's own account, its durable consumer of the events
// stream and its own stream for the closing events.
import (
"context"
"encoding/json"
"errors"
"fmt"
"log"
"strings"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/health"
"git.novox.be/novox/mesh-sdk/go/process"
"distribution/internal/holder"
)
// The addresses the driver asks (manifest invokes).
const (
havesAsk = "mesh-delivery-view.node_haves"
readsAsk = "seat:node-engine.manifest-reads"
)
// caller is a process's call of a tool or a seat verb.
type caller interface {
Call(key string, body any) (json.RawMessage, error)
CallOn(node, key string, body any) (json.RawMessage, error)
}
// busDelivery answers holder.Delivery through the bus.
type busDelivery struct{ c caller }
func (d busDelivery) Haves() ([]holder.NodeHave, error) {
raw, err := d.c.Call(havesAsk, map[string]any{})
if err != nil {
return nil, err
}
var a struct {
Keys map[string]struct {
FP string `json:"fp"`
} `json:"keys"`
Cut bool `json:"cut"`
}
if err := json.Unmarshal(unwrap(raw), &a); err != nil {
return nil, fmt.Errorf("node_haves answered what is not its listing: %w", err)
}
if a.Cut {
return nil, errors.New("node_haves cut its answer, so not every node is read")
}
out := make([]holder.NodeHave, 0, len(a.Keys))
for key, h := range a.Keys {
node, module, ok := strings.Cut(key, ".")
if ok {
out = append(out, holder.NodeHave{Node: node, Module: module, FP: h.FP})
}
}
return out, nil
}
func (d busDelivery) Reads(node string) (bool, []string, error) {
raw, err := d.c.CallOn(node, readsAsk, map[string]any{})
if err != nil {
if notServed(err) {
return false, nil, nil // a node-engine not serving manifest-reads states nothing (design 51)
}
return false, nil, err
}
var a struct {
Reads []string `json:"manifest-reads"`
}
if err := json.Unmarshal(unwrap(raw), &a); err != nil {
return true, nil, fmt.Errorf("%s's manifest-reads answered what is not its statement: %w", node, err)
}
return true, a.Reads, nil
}
// notServed is whether a call's error says nothing serves the address (an away node, an older node-engine).
func notServed(err error) bool {
if errors.Is(err, nats.ErrNoResponders) || errors.Is(err, nats.ErrTimeout) || errors.Is(err, context.DeadlineExceeded) {
return true
}
m := strings.ToLower(err.Error())
for _, w := range []string{"no responders", "serves no verb", "not served", "unknown tool", "timeout"} {
if strings.Contains(m, w) {
return true
}
}
return false
}
// unwrap takes a tool's answer out of the MCP content form the tool runner may answer in.
func unwrap(raw json.RawMessage) json.RawMessage {
var s string
if json.Unmarshal(raw, &s) == nil {
return unwrap(json.RawMessage(s))
}
var c struct {
Content []struct {
Text string `json:"text"`
} `json:"content"`
}
if json.Unmarshal(raw, &c) == nil && len(c.Content) > 0 {
return json.RawMessage(c.Content[0].Text)
}
return raw
}
// kvWindows answers holder.WindowState over the state windows.
type kvWindows struct{ kv jetstream.KeyValue }
func (k kvWindows) get(key string, v any) error {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
e, err := k.kv.Get(ctx, key)
if errors.Is(err, jetstream.ErrKeyNotFound) {
return nil
}
if err != nil {
return err
}
return json.Unmarshal(e.Value(), v)
}
func (k kvWindows) Window(name string) (holder.Window, error) {
var w holder.Window
return w, k.get(name, &w)
}
func (k kvWindows) PutWindow(w holder.Window) error {
b, _ := json.Marshal(w)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
_, err := k.kv.Put(ctx, w.Name, b)
return err
}
func (k kvWindows) Publishing() (holder.Publishing, error) {
var p holder.Publishing
return p, k.get(holder.PublishingKey, &p)
}
// findingsKV writes the driver's findings into distribution's state findings, as the tools bundle writes its own.
func findingsKV(kv jetstream.KeyValue, logf func(string, ...any)) func(kind, subject, words string, raised bool) {
return func(kind, subject, words string, raised bool) {
if kv == nil {
return
}
key := health.FindingKey(health.ScopeSeat, holder.Self, kind+"."+strings.NewReplacer(".", "_", "@", "_", "/", "_").Replace(subject))
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if !raised {
_ = kv.Delete(ctx, key)
return
}
f := health.Finding{V: health.FindingVersion, Key: key, Kind: kind, Ref: key[strings.LastIndex(key, ".")+1:],
Subject: health.Subject{Scope: health.ScopeSeat, ID: holder.Self}, Severity: health.Warning,
Summary: words, Headline: words, Explanation: words + "; registry_old_name or registry_direct_writes drops a retired node by hand",
Resolved: "cleared when the window closes", Source: kind, Observer: holder.Self, At: time.Now().UTC()}
b, _ := json.Marshal(f)
if _, err := kv.Put(ctx, key, b); err != nil {
logf("the finding %s could not be written: %v", key, err)
}
}
}
// sayClosed says a window's closing on distribution's own family distribution.events, once (its id is the window's).
func sayClosed(js jetstream.JetStream, cred process.Account) func(holder.Closed) error {
return func(c holder.Closed) error {
event := holder.OldNameClosedEvent
if c.Window == holder.DirectWritesWindow {
event = holder.DirectWritesClosedEvent
}
b, _ := json.Marshal(c)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
_, err := js.Publish(ctx, cred.StreamSubject(event), b, jetstream.WithMsgID(holder.Self+"."+event))
return err
}
}
// wakes says which events wake the driver, by subject: the status view's delivered, a node-engine's
// manifest-reads-changed (its node the subject's last token).
func wakes(subject string) (string, bool) {
switch {
case strings.HasSuffix(subject, ".mesh-delivery-view.event.delivered"):
return "delivered", true
case strings.Contains(subject, ".node-engine.event.manifest-reads-changed."):
return "manifest-reads-changed on " + subject[strings.LastIndex(subject, ".")+1:], true
}
return "", false
}
// drive wakes the driver at start and on each waking event of the program's durable consumer of the events stream.
// A wake that could not judge is said and its event asked again a minute later.
func drive(ctx context.Context, js jetstream.JetStream, cred process.Account, d *holder.Driver, logger *log.Logger) {
wake := func(why string) error {
said, err := d.Wake(why)
for _, s := range said {
logger.Printf("%s", s)
}
if err != nil {
logger.Printf("the windows were not judged (%s): %v", why, err)
}
return err
}
_ = wake("distribution started")
cons, err := cred.EventConsumer(ctx, js, process.EventsStream)
if err != nil {
logger.Printf("the events stream cannot be heard (%v): the windows move only at the next start", err)
return
}
cc, err := process.HearEvents(cons, func(m jetstream.Msg) {
why, ok := wakes(m.Subject())
if !ok {
_ = m.Ack()
return
}
if wake(why) != nil {
_ = m.NakWithDelay(time.Minute)
return
}
_ = m.Ack()
}, logger.Printf)
if err != nil {
logger.Printf("the events stream cannot be heard (%v): the windows move only at the next start", err)
return
}
<-ctx.Done()
cc.Stop()
}
@@ -0,0 +1,190 @@
package holder
import (
"errors"
"fmt"
"slices"
"sort"
"strings"
"time"
)
// PublishingKey is the key of distribution's state windows that records, per build module, the fingerprints of the
// builds heard publishing through the seats: the tools bundle adds the have's fingerprint of the proved caller of each
// accepted publish. A node's have of a build module running one of them counts as publishing through the seats, so a
// node whose build module runs the same build but builds nothing (a paused build agent) does not hold the window.
const PublishingKey = "publishing"
// Publishing is that record: module -> fingerprints.
type Publishing struct {
Name string `json:"name"`
Modules map[string][]string `json:"modules"`
}
// Learn adds fp for module; it answers whether it was new.
func (p *Publishing) Learn(module, fp string) bool {
if module == "" || fp == "" {
return false
}
if p.Modules == nil {
p.Modules = map[string][]string{}
}
if slices.Contains(p.Modules[module], fp) {
return false
}
p.Modules[module] = append(p.Modules[module], fp)
sort.Strings(p.Modules[module])
return true
}
// NodeHave is one key of the status view's node_haves: a node, a module, the fingerprint it runs.
type NodeHave struct {
Node, Module, FP string
}
// Delivery is what the windows' driver asks of the delivery parts and the node-engine seat, through their tools and
// verbs on the bus (ADR 0596): every node's haves, away ones included, and one node-engine's stated reader
// capabilities. Reads answers served false when the node-engine does not serve `manifest-reads` (it states nothing).
type Delivery interface {
Haves() ([]NodeHave, error)
Reads(node string) (served bool, stated []string, err error)
}
// WindowState is distribution's state windows.
type WindowState interface {
Window(name string) (Window, error)
PutWindow(w Window) error
Publishing() (Publishing, error)
}
// Driver moves the old-name and the direct-write windows: at distribution's start, on each `delivered` and on each
// node-engine's `manifest-reads-changed` (ADR 0618 §4), never on a timer. Each wake reads every node, since a window
// closes only when no node holds it.
type Driver struct {
Delivery Delivery
State WindowState
Callers Callers
// Closed says a window's closing on distribution.events.
Closed func(c Closed) error
Finding func(kind, subject, words string, raised bool)
Now func() time.Time
}
// FindingWindowHeld is the finding of a node holding a window open for seven days.
const FindingWindowHeld = "window-held"
// Read reads every node: its haves, its node-engine's stated capabilities, and whether each build module's have runs a
// build heard publishing through the seats.
func (d *Driver) Read() ([]NodeRead, error) {
haves, err := d.Delivery.Haves()
if err != nil {
return nil, fmt.Errorf("node_haves cannot be read, so no window moves: %w", err)
}
pub, err := d.State.Publishing()
if err != nil {
return nil, fmt.Errorf("the publishing builds cannot be read, so no window moves: %w", err)
}
builds := d.Callers[VerbPublish]
byNode := map[string]*NodeRead{}
var order []string
for _, h := range haves {
if h.Module == "node" || h.Node == "" {
continue
}
n := byNode[h.Node]
if n == nil {
n = &NodeRead{Node: h.Node, NamesArtifact: true}
byNode[h.Node] = n
order = append(order, h.Node)
}
if slices.Contains(builds, h.Module) {
if n.BuildModules == nil {
n.BuildModules = map[string]bool{}
}
n.BuildModules[h.Module] = slices.Contains(pub.Modules[h.Module], h.FP)
}
}
sort.Strings(order)
out := make([]NodeRead, 0, len(order))
for _, node := range order {
n := byNode[node]
served, stated, err := d.Delivery.Reads(node)
if err != nil && served {
return nil, fmt.Errorf("%s's manifest-reads cannot be read, so no window moves: %w", node, err)
}
n.Serves, n.Stated = served, stated
out = append(out, *n)
}
return out, nil
}
// Wake reads every node and steps both windows; it answers a line for each thing it did, and an error when nothing
// could be judged.
func (d *Driver) Wake(why string) ([]string, error) {
nodes, err := d.Read()
if err != nil {
return nil, err
}
now := d.now()
var said, errs []string
for _, c := range []struct {
name, event, seat string
holding func([]NodeRead, []DroppedNode) []string
}{
{OldNameWindow, OldNameClosedEvent, OldSeat, HoldingOldName},
{DirectWritesWindow, DirectWritesClosedEvent, "", HoldingDirectWrites},
} {
cur, err := d.State.Window(c.name)
if err != nil {
errs = append(errs, err.Error())
continue
}
if cur.Name == "" {
cur = Window{Name: c.name, Open: true}
}
next, closed, overdue := cur.Step(c.holding(nodes, cur.Dropped), now)
if !windowsEqual(cur, next) {
if err := d.State.PutWindow(next); err != nil {
errs = append(errs, fmt.Sprintf("the %s window is not kept: %v", c.name, err))
continue
}
}
for _, n := range cur.Holding {
if !slices.Contains(overdue, n) && d.Finding != nil {
d.Finding(FindingWindowHeld, c.name+"."+n, "", false)
}
}
for _, n := range overdue {
if d.Finding != nil {
d.Finding(FindingWindowHeld, c.name+"."+n, fmt.Sprintf("%s has held distribution's %s window open since %s", n, c.name, next.Since[n].Format(time.RFC3339)), true)
}
}
if closed != nil {
closed.Seat = c.seat
if d.Closed != nil {
if err := d.Closed(*closed); err != nil {
errs = append(errs, fmt.Sprintf("%s was not said: %v", c.event, err))
}
}
said = append(said, fmt.Sprintf("the %s window is closed (%s); it was held last by %v", c.name, why, closed.LastHolding))
} else if next.Open {
said = append(said, fmt.Sprintf("the %s window stays open (%s): held by %v", c.name, why, next.Holding))
}
}
if len(errs) > 0 {
return said, errors.New(strings.Join(errs, "; "))
}
return said, nil
}
func (d *Driver) now() time.Time {
if d.Now != nil {
return d.Now()
}
return time.Now().UTC()
}
func windowsEqual(a, b Window) bool {
return a.Open == b.Open && slices.Equal(a.Holding, b.Holding) && a.Closed.Equal(b.Closed) &&
len(a.Since) == len(b.Since) && len(a.Dropped) == len(b.Dropped)
}
@@ -441,6 +441,42 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) {
t.Error("a closed window closed again")
}
})
t.Run("the windows' driver: every node read on each wake, closed once, a paused build agent's same build counted", func(t *testing.T) {
del := &fakeDelivery{haves: []NodeHave{
{"ace", "mesh-build-go", "fp-new"}, {"novox", "mesh-build-go", "fp-new"}, {"g14", "mesh-build-go", "fp-old"},
{"g14", "node", ""}, {"ace", "gitea", "x"}, {"novox", "gitea", "x"},
}, stated: map[string][]string{"ace": {NewNamesCapability}, "novox": {NewNamesCapability}}}
st := &memWindows{w: map[string]Window{}}
st.pub.Learn("mesh-build-go", "fp-new")
var closed []string
found := map[string]bool{}
now := time.Unix(0, 0)
d := &Driver{Delivery: del, State: st, Callers: DefaultCallers(), Now: func() time.Time { return now },
Closed: func(c Closed) error { closed = append(closed, c.Window+" "+c.Seat); return nil },
Finding: func(kind, subject, _ string, raised bool) { found[kind+" "+subject] = raised }}
if _, err := d.Wake("start"); err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(st.w[OldNameWindow].Holding, []string{"g14"}) || !reflect.DeepEqual(st.w[DirectWritesWindow].Holding, []string{"g14"}) {
t.Errorf("windows = %+v, want g14 holding both (away: states nothing; runs the old build)", st.w)
}
now = now.Add(WindowFindingAfter)
_, _ = d.Wake("delivered")
if !found[FindingWindowHeld+" "+OldNameWindow+".g14"] {
t.Errorf("findings %v, want g14 named after seven days", found)
}
del.haves[2].FP, del.stated["g14"] = "fp-new", []string{NewNamesCapability}
if _, err := d.Wake("manifest-reads-changed on g14"); err != nil {
t.Fatal(err)
}
_, _ = d.Wake("delivered")
if want := []string{OldNameWindow + " " + OldSeat, DirectWritesWindow + " "}; !reflect.DeepEqual(closed, want) {
t.Errorf("closed %v, want each window said once, %v", closed, want)
}
if found[FindingWindowHeld+" "+OldNameWindow+".g14"] {
t.Error("the finding stays after the window closed")
}
})
t.Run("the manifest", func(t *testing.T) {
raw, err := os.ReadFile("../../module.json")
if err != nil {
@@ -85,3 +85,24 @@ func (f *fakeRegistry) ServeHTTP(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusNotFound)
}
}
// fakeDelivery answers the driver's reads: haves, and per node whether its node-engine serves manifest-reads.
type fakeDelivery struct {
haves []NodeHave
stated map[string][]string // a node absent does not serve manifest-reads
}
func (f *fakeDelivery) Haves() ([]NodeHave, error) { return f.haves, nil }
func (f *fakeDelivery) Reads(node string) (bool, []string, error) {
s, ok := f.stated[node]
return ok, s, nil
}
type memWindows struct {
w map[string]Window
pub Publishing
}
func (m *memWindows) Window(n string) (Window, error) { return m.w[n], nil }
func (m *memWindows) PutWindow(w Window) error { m.w[w.Name] = w; return nil }
func (m *memWindows) Publishing() (Publishing, error) { return m.pub, nil }
+4 -2
View File
@@ -54,7 +54,8 @@
"seat:node-engine.manifest-reads"
],
"consumes": [
"mesh-delivery-view.delivered"
"mesh-delivery-view.delivered",
"node-engine.manifest-reads-changed"
],
"emits": [
"old-name-closed",
@@ -192,7 +193,8 @@
"env": {
"MESH_BROKER_FILE": "/var/lib/mesh/registry/broker",
"MESH_REGISTRY_LOOPBACK": "http://127.0.0.1:5001",
"MESH_REGISTRY_LISTEN": ":5000"
"MESH_REGISTRY_LISTEN": ":5000",
"MESH_CONFIG_CALLERS": "${config:callers}"
},
"user": "distribution",
"health": {