A3k finding 2: distribution reads its caller list as mesh-config resolves it, not from an environment nothing renders

The node-engine fills ${config:} only in files, unit names, owners and groups, never in a
process's environment, so MESH_CONFIG_CALLERS was never set and design 51's defaults always held.
The tools bundle now reads `callers` through stdio.Config on each call (holder.ConfiguredCallers,
Verbs.WithCallers per call); the program, which has no runtime resolving config, asks its own
tools bundle's new registry_callers on each wake of the windows (manifest invokes). The
environment variable is gone.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
2026-11-04 13:54:17 +01:00
parent 0b2260d9fb
commit b665156b0f
9 changed files with 164 additions and 21 deletions
@@ -3,6 +3,7 @@ package main
import (
"bytes"
"encoding/json"
"errors"
"reflect"
"strings"
"testing"
@@ -30,7 +31,7 @@ func TestTheVerbsSpeakTheSeatsContract(t *testing.T) {
reg.VerbVersions: reg.VersionsInput{}, reg.VerbRemove: reg.RemoveInput{}, reg.VerbHold: reg.HoldInput{},
reg.VerbRelease: reg.ReleaseInput{}, reg.VerbMirror: reg.MirrorInput{}, reg.VerbTags: reg.TagsInput{}}
seen := 0
for _, tool := range tools(&holder.Verbs{}, kept[holder.Window]{}) {
for _, tool := range tools(&holder.Verbs{}, kept[holder.Window]{}, func() (holder.Callers, error) { return holder.DefaultCallers(), nil }) {
seat, verb, ok := strings.Cut(tool.Name, ".")
in, isVerb := inputs[verb]
if !ok || !isVerb || (seat != holder.ImageSeat && seat != holder.ArtifactSeat) {
@@ -56,6 +57,28 @@ func TestTheVerbsSpeakTheSeatsContract(t *testing.T) {
t.Errorf("%s's body %s is not the contract's: %v", event, b, err)
}
}
// The caller list is read on each call: one that cannot be read answers no verb, and registry_callers says it as read.
unread := func() (holder.Callers, error) { return nil, errors.New("config refused") }
for _, tool := range tools(&holder.Verbs{}, kept[holder.Window]{}, unread) {
if tool.Name == holder.ImageSeat+"."+holder.VerbVersions {
if _, err := tool.Run(map[string]any{"name": "m/a"}); err == nil || !strings.Contains(err.Error(), "caller list") {
t.Errorf("versions with an unread caller list = %v, want refused naming the caller list", err)
}
}
}
answered := false
for _, tool := range tools(&holder.Verbs{}, kept[holder.Window]{}, func() (holder.Callers, error) { return holder.Callers{holder.VerbPublish: {"mesh-build-go"}}, nil }) {
if tool.Name == holder.CallersTool {
answered = true
got, err := tool.Run(map[string]any{})
if b, _ := json.Marshal(got); err != nil || string(b) != `{"callers":{"publish":["mesh-build-go"]}}` {
t.Errorf("%s = %s, %v", holder.CallersTool, b, err)
}
}
}
if !answered {
t.Errorf("no tool %s: the program could not read the caller list", holder.CallersTool)
}
if a := publishAnswer(holder.PublishAnswer{Transfer: "t", Missing: []string{"sha256:bb"}}); !reflect.DeepEqual(a.Lacks, []string{"sha256:bb"}) || a.Held {
t.Errorf("publish answers %+v", a)
}
@@ -238,7 +238,16 @@ func arg[T any](args map[string]any) (T, error) {
func str(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }
func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool {
// tools are distribution's tools over v; each call that checks a caller reads the caller list on that call (callers:
// the config value as mesh-config resolves it now), so an operator's change of it holds from the next call on.
func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.Callers, error)) []stdio.Tool {
now := func() (*holder.Verbs, error) {
c, err := callers()
if err != nil {
return nil, fmt.Errorf("the caller list cannot be read, so no verb is answered: %w", err)
}
return v.WithCallers(c), nil
}
obj := func(props map[string]any, req ...string) map[string]any {
return map[string]any{"type": "object", "required": req, "properties": props}
}
@@ -249,6 +258,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool {
stdio.Tool{Name: seat + "." + holder.VerbPublish, Description: "Open a transfer of a publish (ADR 0611): its name, version, kind, digest and parts; answers the parts the holder lacks.",
Input: obj(map[string]any{"transfer": s("the transfer's id"), "name": s("<module>/<artifact>"), "version": s("the builder's version"), "kind": s("image, bundle, archive or upstream"), "digest": s("sha256:…"), "parts": map[string]any{"type": "array"}}, "transfer", "name", "kind", "digest", "parts"),
Run: func(args map[string]any) (any, error) {
v, err := now()
if err != nil {
return nil, err
}
in, err := arg[reg.PublishInput](args)
if err != nil {
return nil, err
@@ -266,6 +279,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool {
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.",
Input: obj(map[string]any{"transfer": s("its id"), "part": s("the part's digest"), "sequence": map[string]any{"type": "integer"}, "last": map[string]any{"type": "boolean"}, "data": s("base64")}, "transfer", "part", "sequence", "last", "data"),
Run: func(args map[string]any) (any, error) {
v, err := now()
if err != nil {
return nil, err
}
in, err := arg[reg.PublishPartInput](args)
if err != nil {
return nil, err
@@ -279,6 +296,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool {
}},
stdio.Tool{Name: seat + "." + holder.VerbVersions, Description: "The versions the seat holds of a name.", Input: obj(map[string]any{"name": s("<module>/<artifact>")}, "name"),
Run: func(args map[string]any) (any, error) {
v, err := now()
if err != nil {
return nil, err
}
a, err := v.Versions(seat, stdio.CallerOf(args), str(args, "name"))
if err != nil {
return nil, err
@@ -288,6 +309,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool {
stdio.Tool{Name: seat + "." + holder.VerbRemove, Description: "Remove a tagged version (the builder), or for retention an untagged one by its digest (the artifact sweep); says removed.",
Input: obj(map[string]any{"name": s("name"), "version": s("a tagged version"), "tag": s("its tag"), "digest": s("for retention"), "reason": s("why")}, "name", "reason"),
Run: func(args map[string]any) (any, error) {
v, err := now()
if err != nil {
return nil, err
}
in, err := arg[reg.RemoveInput](args)
if err != nil {
return nil, err
@@ -301,6 +326,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool {
stdio.Tool{Name: seat + "." + holder.VerbHold, Description: "Keep one untagged bundle, archive or copied base from retention (the artifact sweep alone); says held.",
Input: obj(map[string]any{"name": s("name"), "digest": s("sha256:…"), "reason": s("which of the sweep's rules keeps it")}, "name", "digest", "reason"),
Run: func(args map[string]any) (any, error) {
v, err := now()
if err != nil {
return nil, err
}
in, err := arg[reg.HoldInput](args)
if err != nil {
return nil, err
@@ -313,6 +342,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool {
stdio.Tool{Name: seat + "." + holder.VerbRelease, Description: "Let a hold go (the artifact sweep, or the operator's session through registry_holds); says released.",
Input: obj(map[string]any{"name": s("name"), "digest": s("sha256:…"), "reason": s("why it is let go")}, "name", "digest", "reason"),
Run: func(args map[string]any) (any, error) {
v, err := now()
if err != nil {
return nil, err
}
in, err := arg[reg.ReleaseInput](args)
if err != nil {
return nil, err
@@ -327,6 +360,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool {
out = append(out,
stdio.Tool{Name: holder.ImageSeat + "." + holder.VerbTags, Description: "The tags of an image.", Input: obj(map[string]any{"name": s("name")}, "name"),
Run: func(args map[string]any) (any, error) {
v, err := now()
if err != nil {
return nil, err
}
a, err := v.Tags(holder.ImageSeat, stdio.CallerOf(args), str(args, "name"))
if err != nil {
return nil, err
@@ -336,6 +373,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool {
stdio.Tool{Name: holder.ImageSeat + "." + holder.VerbMirror, Description: "Copy an outside image, pinned by its digest, into the store as upstream/<host>/<path> (every platform, only the parts it lacks): answers at once with the reference the copy will have, and says published (or refused, with why) on the seat's events when the copy ends; wait on that event (mesh-build-image alone).",
Input: obj(map[string]any{"source": s("<host>/<path>"), "digest": s("sha256:…")}, "source", "digest"),
Run: func(args map[string]any) (any, error) {
v, err := now()
if err != nil {
return nil, err
}
in, err := arg[reg.MirrorInput](args)
if err != nil {
return nil, err
@@ -346,6 +387,15 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool {
}
return reg.MirrorAnswer{Digest: in.Digest, Reference: ref}, nil
}},
stdio.Tool{Name: holder.CallersTool, Description: "Who may call which verb of distribution's seats: the trusted config value callers as mesh-config resolves it on this machine now (design 51's list where unset). The program asks it at each wake of the windows, as it has no config of its own.",
Input: obj(map[string]any{}),
Run: func(map[string]any) (any, error) {
c, err := callers()
if err != nil {
return nil, err
}
return map[string]any{"callers": c}, nil
}},
stdio.Tool{Name: "registry_health", Description: "distribution's health check, run by the runtime each minute: versions of each seat asked through the store's read side (the program's port 5000); fails naming the program when it does not answer.",
Input: obj(map[string]any{}),
Run: func(map[string]any) (any, error) {
@@ -355,6 +405,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool {
stdio.Tool{Name: "registry_incoming", Description: "Open transfers of incoming and its fill; with drop, drops one by hand (a hand act, why required).",
Input: obj(map[string]any{"drop": s("a transfer id to drop"), "why": s("why it is dropped")}),
Run: func(args map[string]any) (any, error) {
v, err := now()
if err != nil {
return nil, err
}
idle, _ := v.DropIdle()
if id := str(args, "drop"); id != "" {
if str(args, "why") == "" {
@@ -371,6 +425,10 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool {
stdio.Tool{Name: "registry_holds", Description: "What is held, since when and for whom; with release, lets one go through the seat's release as the operator's session (a hand act).",
Input: obj(map[string]any{"seat": s("the seat"), "release": s("the name to release"), "digest": s("its digest"), "why": s("why")}),
Run: func(args map[string]any) (any, error) {
v, err := now()
if err != nil {
return nil, err
}
if name := str(args, "release"); name != "" {
if str(args, "why") == "" {
return nil, errors.New("a release by hand is a hand act: say why")
@@ -414,20 +472,15 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool {
}
func main() {
callers, err := holder.ParseCallers(os.Getenv("MESH_CONFIG_CALLERS"))
if err != nil {
logf("%v", err)
os.Exit(1)
}
base, _ := url.Parse("http://127.0.0.1:5001")
store := &holder.OCI{Base: base, Registry: os.Getenv("MESH_REGISTRY_REACH"), Upstream: &holder.Upstream{}}
h := &holder.Holder{Callers: callers, Objects: parts.BusObjects(holder.IncomingObjects), Builder: delivery.BuilderVersions{Ask: stdio.Ask}, Store: store, Say: say, Finding: finding,
h := &holder.Holder{Objects: parts.BusObjects(holder.IncomingObjects), Builder: delivery.BuilderVersions{Ask: stdio.Ask}, Store: store, Say: say, Finding: finding,
Transfers: parts.BusTransfers[holder.Opening](holder.TransfersState)}
v := &holder.Verbs{Holder: h, Catalog: store, Holds: kept[holder.Hold]{state: "holds", key: holder.Hold.Key}, Mirrors: mirrors{}}
go followMirrors(v)
windows := kept[holder.Window]{state: "windows", key: func(w holder.Window) string { return w.Name }}
closeBootstrap = bootstrapCloser(windows)
if err := stdio.Serve("", tools(v, windows)); err != nil {
if err := stdio.Serve("", tools(v, windows, holder.ConfiguredCallers)); err != nil {
logf("%v", err)
os.Exit(1)
}
@@ -64,12 +64,8 @@ func run(ctx context.Context, getenv func(string) string) error {
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,
go drive(ctx, js, cred, &holder.Driver{Delivery: busDelivery{calls}, State: kvWindows{windows}, Callers: busCallers{calls}.Read,
Closed: sayClosed(js, cred), Finding: findingsKV(findings, logger.Printf)}, logger)
}
front := &holder.Front{Store: store, Window: &open, Logf: logger.Printf,
@@ -6,8 +6,9 @@ package main
// run() as the node starts it, its broker account on a nats-server of this test alone, the registry container a fake
// OCI store on the loopback, the status view's node_haves and the node-engine seat's manifest-reads answered on their
// call subjects. An away node holds the old name open; its return (manifest-reads-changed) closes it, said once on
// distribution.events, after which a direct write is refused naming publish; reads stay open; and a waiter on a seat's
// `published` hears none of the windows' events.
// distribution.events, after which a direct write is refused naming publish; reads stay open; a waiter on a seat's
// `published` hears none of the windows' events; and the build modules counted are publish's callers as the tools
// bundle answers the config value, not design 51's defaults.
import (
"context"
@@ -76,10 +77,22 @@ func TestDistributionWhole(t *testing.T) {
answer(m, map[string]any{"keys": map[string]any{
"ace.mesh-build-go": map[string]any{"fp": "fp-new"}, "g14.mesh-build-go": map[string]any{"fp": fp},
"ace.gitea": map[string]any{"fp": "x"}, "g14.gitea": map[string]any{"fp": "x"},
// A build module the operator took off publish's list: it holds no window (the caller list as configured).
"g14.mesh-build-node": map[string]any{"fp": "fp-node-old"},
}})
}); err != nil {
t.Fatal(err)
}
// distribution's own tools bundle answers the config value callers as mesh-config resolved it: publish by
// mesh-build-go alone. The program reads it there, never from an environment variable nothing renders.
if _, err := nc.Subscribe("mesh.mod.distribution.>", func(m *nats.Msg) {
if !strings.Contains(m.Subject, holder.CallersTool) {
return
}
answer(m, map[string]any{"callers": map[string]any{"publish": []string{"mesh-build-go"}}})
}); err != nil {
t.Fatal(err)
}
engine := func(node string) {
for _, subj := range []string{"mesh.seat.node-engine.call.manifest-reads." + node + ".>", "mesh.seat.node-engine.tool.manifest-reads." + node} {
if _, err := nc.Subscribe(subj, func(m *nats.Msg) {
@@ -28,6 +28,8 @@ import (
const (
havesAsk = "mesh-delivery-view.node_haves"
readsAsk = "seat:node-engine.manifest-reads"
// callersAsk is distribution's own tool answering its config value callers as resolved.
callersAsk = holder.Self + "." + holder.CallersTool
)
// caller is a process's call of a tool or a seat verb.
@@ -36,6 +38,25 @@ type caller interface {
CallOn(node, key string, body any) (json.RawMessage, error)
}
// busCallers reads the caller list from distribution's own tools bundle (holder.CallersTool, manifest invokes): a
// program has no runtime that resolves its module's config values, the tools bundle has (stdio.Config), so the program
// asks it on each wake rather than read an environment variable nothing renders.
type busCallers struct{ c caller }
func (b busCallers) Read() (holder.Callers, error) {
raw, err := b.c.Call(callersAsk, map[string]any{})
if err != nil {
return nil, fmt.Errorf("%s: %w", callersAsk, err)
}
var a struct {
Callers json.RawMessage `json:"callers"`
}
if err := json.Unmarshal(unwrap(raw), &a); err != nil || len(a.Callers) == 0 {
return nil, fmt.Errorf("%s answered what is not a caller list: %s", callersAsk, unwrap(raw))
}
return holder.ParseCallers(string(a.Callers))
}
// busDelivery answers holder.Delivery through the bus.
type busDelivery struct{ c caller }
@@ -89,6 +89,37 @@ func ParseCallers(raw string) (Callers, error) {
if err != nil {
return nil, err
}
return selfNotReleasing(c)
}
// ConfiguredCallers is the config value `callers` as mesh-config resolved it for this machine, read through the tools
// runtime's stdio.Config on each call (never an environment variable: nothing renders a config value into a
// program's environment); unset is DefaultCallers, and Self on release's list is refused as ParseCallers refuses it.
func ConfiguredCallers() (Callers, error) {
c, err := stdio.ConfiguredVerbCallers(CallersConfig, VerbsOf(ImageSeat), DefaultCallers())
if err != nil {
return nil, err
}
return selfNotReleasing(c)
}
// CallersConfig is the config value's key, and CallersTool the tools bundle's tool answering it as resolved: the
// program, which has no tools runtime, asks that tool over the bus (manifest invokes).
const (
CallersConfig = "callers"
CallersTool = "registry_callers"
)
// WithCallers is v with the caller list c: a copy for one call, so a list read on that call never changes another's.
func (v *Verbs) WithCallers(c Callers) *Verbs {
h := *v.Holder
h.Callers = c
out := *v
out.Holder = &h
return &out
}
func selfNotReleasing(c Callers) (Callers, error) {
if slices.Contains(c[VerbRelease], Self) {
return nil, fmt.Errorf("callers puts %s on release's list: registry_holds would release in its own name", Self)
}
@@ -63,7 +63,8 @@ type WindowState interface {
type Driver struct {
Delivery Delivery
State WindowState
Callers Callers
// Callers is the caller list as it holds on this wake: publish's callers are the build modules counted.
Callers func() (Callers, error)
// Closed says a window's closing on distribution.events.
Closed func(c Closed) error
Finding func(kind, subject, words string, raised bool)
@@ -84,7 +85,11 @@ func (d *Driver) Read() ([]NodeRead, error) {
if err != nil {
return nil, fmt.Errorf("the publishing builds cannot be read, so no window moves: %w", err)
}
builds := d.Callers[VerbPublish]
callers, err := d.Callers()
if err != nil {
return nil, fmt.Errorf("the caller list cannot be read, so no window moves: %w", err)
}
builds := callers[VerbPublish]
byNode := map[string]*NodeRead{}
var order []string
for _, h := range haves {
@@ -532,7 +532,7 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) {
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 },
d := &Driver{Delivery: del, State: st, Callers: func() (Callers, error) { return DefaultCallers(), nil }, 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 {
+3 -2
View File
@@ -43,6 +43,7 @@
}
],
"tools": [
"registry_callers",
"registry_direct_writes",
"registry_health",
"registry_holds",
@@ -50,6 +51,7 @@
"registry_old_name"
],
"invokes": [
"distribution.registry_callers",
"mesh-delivery-builder.builder_versions",
"mesh-delivery-view.node_haves",
"seat:node-engine.manifest-reads"
@@ -199,8 +201,7 @@
"env": {
"MESH_BROKER_FILE": "/var/lib/mesh/registry/broker",
"MESH_REGISTRY_LOOPBACK": "http://127.0.0.1:5001",
"MESH_REGISTRY_LISTEN": ":5000",
"MESH_CONFIG_CALLERS": "${config:callers}"
"MESH_REGISTRY_LISTEN": ":5000"
},
"user": "distribution",
"health": {