Apply only the node's own compositions, never the controller's
mesh/delivery merged: source.main moved to novox/mesh-host@84737f4cb26d
mesh/check being carried to 38d666f22390 on 84737f4cb26d | from 14a12cc7b6f9

The controller ended by its unassignment, so a node that falls back to the
compositions in its bucket follows what nobody writes (issue 1126, A7). The
node agent now always runs its own composer: without a placement key it
composes nothing and keeps what runs, a placement it cannot watch stops the
agent and is said, and a candidate taking its placement away is refused
instead of answered as a return to the controller. The start-time placement
read, the end-when-placed watchers and the returning answer go, with their
tests and the three integration tests of the controller-composed path.

Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
This commit is contained in:
2026-11-02 10:40:50 +01:00
parent 5cfc4ff4a7
commit 38d666f223
5 changed files with 61 additions and 621 deletions
+2 -70
View File
@@ -798,8 +798,8 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
r := reader{c: c}
c.mu.Unlock()
if !placed {
// The catalogue no longer places it: the agent returns to the controller's compositions (endWhenUnplaced), and
// this is no refusal of what it runs (issue 1126, the seed's revert).
// The catalogue does not place it: it composes nothing, applies no composition made elsewhere, and this is no
// refusal of what it runs (issue 1126).
c.sayOnce("this node's key in the seat's placement state is gone: it composes nothing of its own, and keeps running what it runs")
c.paid() // nothing of its own to compose, so no pass is owed and no timer armed again
return nil
@@ -1496,74 +1496,6 @@ func (r reader) Credentials(ctx context.Context, node string, asConsumer, asProv
return creds, issued, nil
}
// Placed reads whether the catalogue places node now, for the node agent choosing whose compositions it applies at
// each start: true when its placement key holds a value, false only when a read succeeds and finds no key or a delete
// marker. A read that fails chooses neither (review of the shadow, round 1: a timeout or a bus not ready at a restart
// flipped a self-composing node to the controller's compositions for a whole run): it is tried again after wait(try),
// each distinct failure said once, until ctx ends, which is then its error.
//
// switched says the node composed from its own placement key before (its have says it composes its own,
// delivery.ComposesOwn, and that is kept on the bus, so it survives a restart). Only a node that never switched takes a
// read that fails (a timeout, or a state its account may not read yet, as the seat's before its grant arrives) as "not
// placed": it stays on the controller's compositions, says why once, and switches on a positive read of its key (the
// watch endWhenPlaced keeps). A node that switched never goes back on a failed read, however long it lasts.
func Placed(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string,
wait func(try int) time.Duration, say func(string), switched bool) (bool, error) {
key, err := delivery.PlacementNodeKey(node)
if err != nil {
return false, err
}
legacy, _ := delivery.LegacyPlacementKey(node)
said := ""
for try := 1; ; try++ {
// Its node key, or, before the seat's holder converted it, its legacy key (research 072 row 37).
raw, err := get(ctx, delivery.PlacementBucket, key)
if err == nil && len(raw) == 0 {
raw, err = get(ctx, delivery.PlacementBucket, legacy)
}
if errors.Is(err, nats.ErrBucketNotFound) {
return false, nil // the seat has written no placement state yet: nothing places it, read and known
}
if err == nil {
return len(raw) > 0, nil
}
if ctx.Err() != nil {
return false, ctx.Err()
}
if !switched {
say(fmt.Sprintf("this node's key in the seat's placement state cannot be read (%s): it has never composed its own, so it applies the controller's compositions and switches when its key is read: %v", key, err))
return false, nil
}
if why := err.Error(); why != said {
say(fmt.Sprintf("this node's key in the seat's placement state cannot be read (%s): the node agent waits, choosing neither its own compositions nor the controller's: %v", key, err))
said = why
}
select {
case <-ctx.Done():
return false, ctx.Err()
case <-time.After(wait(try)):
}
}
}
// ComposedItsOwn says whether the node's own have, kept on the bus, says it composed from its placement key at its last
// start. A have that cannot be read says yes: the safe side, a node waits and does not fall back.
func ComposedItsOwn(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string) bool {
key, err := delivery.HaveNodeKey(node)
if err != nil {
return true
}
raw, err := get(ctx, delivery.HaveBucket, key)
if err != nil {
return true
}
if len(raw) == 0 {
return false
}
n, err := delivery.DecodeHaveNode(raw)
return err != nil || n.Composes == delivery.ComposesOwn
}
// composeAgain are the waits before a pass that met a bus not answering is tried again.
var composeAgain = []time.Duration{time.Second, 2 * time.Second, 4 * time.Second}
+17 -94
View File
@@ -1162,9 +1162,11 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
return a.Verdict == delivery.CandidateSame && a.Returns == "" && a.Revision == 8 && len(a.Removed) == 0 &&
fmt.Sprint(a.Same) == "[lemurs node]"
},
// Issue 1126, A7: a candidate deleting the node's placement is refused, never answered as a return to the
// controller's compositions: no composition made elsewhere is applied, so the node would compose nothing.
"deletes": func(a delivery.CandidateAnswer) bool {
return a.Verdict == delivery.CandidateSame && a.Returns == delivery.ReturnsToController && a.Revision == 7 &&
len(a.Removed) == 0 && fmt.Sprint(a.Same) == "[lemurs node]"
return a.Verdict == delivery.CandidateRefused && a.Returns == "" && a.Revision == 7 &&
strings.Contains(a.Refused, "composes nothing")
},
} {
if !want(answers[group]) {
@@ -1286,61 +1288,21 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
return !ok && len(a) == 4
})
// Whose compositions a node applies at a start (review of the shadow, round 1): only a read that succeeds chooses; a
// failing read is tried again, said once, and chooses neither, even when the link ends.
for name, row := range map[string]struct {
answers []error // per try; nil then the value
value []byte
end bool
placed bool
tries int
quiet bool // read and known: nothing to say
// switched: the node's have says it composed its own before (it never falls back on a failed read).
switched bool
}{
"switched: placed after two failures": {answers: []error{fmt.Errorf("nats: timeout"), fmt.Errorf("nats: timeout")}, value: []byte(`{}`), placed: true, tries: 3, switched: true},
"never switched: a timeout is not placed, said once": {answers: []error{fmt.Errorf("nats: timeout")}, placed: false, tries: 1},
"no key": {placed: false, tries: 2}, // its node key, then its legacy key (research 072 row 37)
"no placement state yet": {answers: []error{fmt.Errorf("the bucket %s: %w", delivery.PlacementBucket, nats.ErrBucketNotFound)}, placed: false, tries: 1, quiet: true},
"switched: failing until the link ends": {answers: []error{fmt.Errorf("jetstream not ready"), fmt.Errorf("jetstream not ready"),
fmt.Errorf("jetstream not ready")}, end: true, tries: 3, switched: true},
} {
lctx, lcancel := context.WithCancel(ctx)
tries := 0
var said []string
get := func(context.Context, string, string) ([]byte, error) {
tries++
if tries <= len(row.answers) {
if row.end && tries == len(row.answers) {
lcancel()
}
return nil, row.answers[tries-1]
// Issue 1126, A7 AC1: a placement that cannot be read composes nothing and says why; the node never applies a
// composition made elsewhere instead. (A placement read without the node's key composes nothing either: above.)
{
refused := newFakeBus()
env := refused.env()
watch := env.Watch
env.Watch = func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error) {
if bucket == delivery.PlacementBucket {
return nil, fmt.Errorf("nats: Permissions Violation for Subscription to %s", bucket)
}
return row.value, nil
return watch(ctx, bucket, pattern)
}
placed, err := Placed(lctx, get, "n1", func(int) time.Duration { return time.Millisecond }, func(s string) { said = append(said, s) }, row.switched)
lcancel()
if placed != row.placed || tries != row.tries || (err != nil) != row.end || (len(row.answers) > 0 && !row.quiet) != (len(said) == 1) {
t.Errorf("%s: placed %v, err %v after %d reads, said %q", name, placed, err, tries, said)
}
}
// Whether the node composed its own before is read from its own have: own says yes, the controller's or no have says
// no, and a have that cannot be read says yes (the safe side: a node waits and does not fall back).
for name, row := range map[string]struct {
raw []byte
err error
want bool
}{
"own": {raw: []byte(`{"v":1,"composes":"own"}`), want: true},
"the controller's": {raw: []byte(`{"v":1,"composes":"controller"}`), want: false},
"no have yet": {want: false},
"a have that cannot be read": {err: fmt.Errorf("nats: timeout"), want: true},
"a have that is no have": {raw: []byte(`{`), want: true},
} {
get := func(context.Context, string, string) ([]byte, error) { return row.raw, row.err }
if got := ComposedItsOwn(ctx, get, "n1"); got != row.want {
t.Errorf("composed its own, %s: %v, want %v", name, got, row.want)
out, err := New(env).Watch(ctx, "")
if out != nil || err == nil || !strings.Contains(err.Error(), delivery.PlacementBucket) || !strings.Contains(err.Error(), "cannot be watched") {
t.Errorf("an unreadable placement: composed %v, said %v", out != nil, err)
}
}
@@ -1450,45 +1412,6 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
}
}
// The way back, row by row (review of the shadow, round 1): the controller's compositions at n1's wants compared
// with what runs; one differing is answered so; one the controller holds none of refuses, without returning.
var runsList []delivery.Composed
for _, c := range running {
runsList = append(runsList, c)
}
lemursKey, _ = delivery.ComposedKey("n1", "lemurs", fp)
changed := running[lemursKey]
changed.Resources = []json.RawMessage{json.RawMessage(`{"id":"lemurs.service","type":"service","module":"lemurs","unit":"lemurs.service","state":"stopped","boot":"disabled"}`)}
changedRaw, _ := delivery.EncodeComposed(changed)
for name, row := range map[string]struct {
lemurs []byte
check func(delivery.CandidateAnswer) bool
}{
"the controller's differs": {changedRaw, func(a delivery.CandidateAnswer) bool {
return a.Verdict == delivery.CandidateDiffers && a.Returns == delivery.ReturnsToController &&
len(a.Differs) == 1 && a.Differs[0].Module == "lemurs" && fmt.Sprint(a.Same) == "[node]"
}},
"the controller holds none": {nil, func(a delivery.CandidateAnswer) bool {
return a.Verdict == delivery.CandidateRefused && a.Returns == "" && strings.Contains(a.Refused, "lemurs: the controller holds no composition")
}},
} {
get := func(_ context.Context, bucket, key string) ([]byte, error) {
if key == lemursKey {
return row.lemurs, nil
}
readOnly.mu.Lock()
defer readOnly.mu.Unlock()
return readOnly.kv[bucket+"/"+key], nil
}
a := returning(ctx, get, "n1", map[string]delivery.Want{"lemurs": {FP: fp}, agent.NodeModule: {}}, runsList, "pro", 1)
if !row.check(a) {
t.Errorf("%s: %+v", name, a)
}
if _, err := delivery.EncodeHaveNode(delivery.HaveNode{Candidates: map[string]delivery.CandidateAnswer{"pro": a}}); err != nil {
t.Errorf("%s: the answer is not one a have holds: %v", name, err)
}
}
// Issue 1126, the seed's review: a module n1 runs inside its own composition (nats, at a build of the old form) is
// composed there again, not at the trunk's build as one added; lemurs at the build it runs, keeper (added) at the
// trunk's.
+9 -73
View File
@@ -23,8 +23,7 @@ import (
//
// **Event-driven, like the Composer.** It watches every candidate key, the trunk's placement, every node's record, the
// indexer's modules, this node's config values, and this node's module haves (what it runs moved); a burst of changes
// answers once. It runs whether or not the node is placed yet: a node that applies the controller's compositions answers
// against what it runs from them. An answer is handed only when it changes, and a proposal's answer is dropped when its
// answers once. It runs whether or not the node is placed yet, against what it runs. An answer is handed only when it changes, and a proposal's answer is dropped when its
// keys go.
type Shadow struct {
c *Composer
@@ -33,8 +32,7 @@ type Shadow struct {
mu sync.Mutex
candidates map[string]map[string]candidateKey // by proposal, by node
answers map[string]delivery.CandidateAnswer
// wants are the node's wants by module (agent.NodeModule for its own): what a proposal deleting its placement returns
// it to, the controller's compositions at them.
// wants are the node's wants by module (agent.NodeModule for its own); a change of one answers again.
wants map[string]delivery.Want
// ran is, by module, the digest the module's have says runs: what runs, where the agent's list holds only the copy
// from the declaration kept before the switch (issue 1126, probe #532 on shanks).
@@ -161,7 +159,7 @@ func (s *Shadow) Run(ctx context.Context) error {
{ResolvedBucket, node + ".>", (*Composer).takeValue, nil},
// What this node runs moved: an apply ends in its module's have.
{delivery.HaveBucket, node + ".>", nil, nil},
// Its wants: a proposal deleting its placement returns it to the controller's compositions at them.
// Its wants.
{delivery.AssignerWantsBucket, "want." + node + ".>", nil, nil},
}
type tagged struct {
@@ -530,16 +528,12 @@ func (s *Shadow) answerProposal(ctx context.Context, proposal string) (delivery.
running = c.env.Running()
}
if p, ok := given[node]; ok && p == nil && !own.c.Reached {
// The proposal deletes this node's placement key: the node then returns to the controller's compositions, as one the
// catalogue does not place (endWhenUnplaced). Those at its wants are compared with what runs (issue 1126, the
// seed's revert, whose check read "would stop running node").
s.mu.Lock()
wants := make(map[string]delivery.Want, len(s.wants))
for m, w := range s.wants {
wants[m] = w
}
s.mu.Unlock()
return returning(ctx, c.get, node, wants, running, proposal, rev), true
// The proposal deletes this node's placement key: the node would compose nothing and keep running what it runs,
// since no composition made elsewhere is applied (issue 1126: the controller ended by its unassignment). Refused,
// so the catalogue does not take a node's placement away by a merge.
return delivery.CandidateAnswer{Proposal: proposal, Revision: rev, Verdict: delivery.CandidateRefused,
Refused: "the proposal takes this node's placement away: it then composes nothing, and no composition made " +
"elsewhere is applied, so what runs here would be followed by nothing"}, true
}
placement := given[node]
@@ -718,64 +712,6 @@ func (s *Shadow) takeWant(u agent.Update) bool {
return !had || old.FP != w.FP
}
// returning is the node's answer to a proposal deleting its placement: it returns to the controller's compositions, each
// read at the fingerprint its want names in the node's composition bucket and compared with what runs
// (delivery.CompareComposed). A want the controller holds no composition for, or one that cannot be read, refuses the
// answer, naming each: returning then would leave that module with nothing to apply.
func returning(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string,
wants map[string]delivery.Want, running []delivery.Composed, proposal string, rev uint64) delivery.CandidateAnswer {
a := delivery.CandidateAnswer{Proposal: proposal, Revision: rev}
bucket, err := delivery.ComposedBucketFor(node)
if err != nil {
a.Refused, a.Verdict = err.Error(), delivery.CandidateRefused
return a
}
modules := make([]string, 0, len(wants))
for m := range wants {
modules = append(modules, m)
}
sort.Strings(modules)
comps := map[string]delivery.Composed{}
var missing []string
for _, m := range modules {
var key string
if m == agent.NodeModule {
key, err = delivery.ComposedNodeKey(node)
} else {
key, err = delivery.ComposedKey(node, m, wants[m].FP)
}
if err != nil {
missing = append(missing, m+": "+err.Error())
continue
}
raw, err := get(ctx, bucket, key)
switch {
case err != nil:
missing = append(missing, fmt.Sprintf("%s: the controller's composition %s cannot be read: %v", m, key, err))
continue
case raw == nil:
missing = append(missing, fmt.Sprintf("%s: the controller holds no composition %s, which its want names", m, key))
continue
}
comp, err := delivery.DecodeComposed(raw)
if err != nil {
missing = append(missing, fmt.Sprintf("%s: the controller's composition %s: %v", m, key, err))
continue
}
comps[key] = comp
}
if len(missing) > 0 {
a.Refused = "returning to the controller's compositions, " + strings.Join(missing, "; ")
a.Verdict = delivery.CandidateRefused
return a
}
cmp := delivery.CompareComposed(comps, running)
a.Returns = delivery.ReturnsToController
a.Same, a.Differs, a.Added, a.Removed = cmp.Same, cmp.Differs, cmp.Added, cmp.Removed
a.Verdict = delivery.VerdictOf(a)
return a
}
// sameValue says whether two sealed values open, with this node's key, to the same value; unknown when either does not
// open (or the node holds no opener). Only the comparison leaves: never a value (issue 1126, the shadow shows its work).
func (c *Composer) sameValue(a, b string) (same, known bool) {