novox/mesh-host
18
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge pull request 'Node agent: compose from the seat mesh-placement and answer proposals (issue 1126, delivery 2 of 2)' (#167) from feat/1126-placement-seat into main

Browse Source
This commit was merged in pull request #167.
This commit is contained in:
mesh-tools
2026-10-28 18:33:12 +00:00
parent 58b4f8b6bc 45f5e27526
commit 4f6b389a37
13 changed files with 371 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mesh-host/nodeagent.go
+58 -15
View File
@@ -39,6 +39,20 @@ func (w bucketWatch) Watch(ctx context.Context, pattern string) (<-chan agent.Up
if err != nil { if err != nil {
return nil, err return nil, err
} }
return w.updates(ctx, in), nil
}
// WatchAwait is Watch for a state that its holder makes when it starts (the placement seat's): a bucket not on the bus
// yet is waited for on the bus's word that it was created (link.Session.KVWatchAwait), and waiting is told why.
func (w bucketWatch) WatchAwait(ctx context.Context, pattern string, waiting func(error)) (<-chan agent.Update, error) {
in, err := w.session.KVWatchAwait(ctx, w.bucket, pattern, waiting)
if err != nil {
return nil, err
}
return w.updates(ctx, in), nil
}
func (w bucketWatch) updates(ctx context.Context, in <-chan link.KVUpdate) <-chan agent.Update {
out := make(chan agent.Update, 64) out := make(chan agent.Update, 64)
go func() { go func() {
defer close(out) defer close(out)
@@ -50,7 +64,7 @@ func (w bucketWatch) Watch(ctx context.Context, pattern string) (<-chan agent.Up
} }
} }
}() }()
return out, nil return out
} }
// sessionHaves writes the haves the agent says, through the session. // sessionHaves writes the haves the agent says, through the session.
@@ -594,7 +608,9 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
session.NoteComposes(linked, delivery.ComposesController, why) session.NoteComposes(linked, delivery.ComposesController, why)
// And once the catalogue places it, the agent begins again composing its own: the placement key appearing ends // And once the catalogue places it, the agent begins again composing its own: the placement key appearing ends
// this run, so a node does not wait for a restart to follow the seed of its placement (ADR 0549 §4). // this run, so a node does not wait for a restart to follow the seed of its placement (ADR 0549 §4).
go endWhenPlaced(run, session, node, stopRun, say) go endWhenPlaced(run, session, node, stopRun, say, func(reason string) {
session.NoteComposes(linked, delivery.ComposesController, why+"; "+reason)
})
} }
ag = agent.New(env) ag = agent.New(env)
own.set(ag) own.set(ag)
@@ -896,16 +912,25 @@ func selfComposer(ctx context.Context, session *link.Session, node string, say f
return nil, err.Error() return nil, err.Error()
} }
env := composerEnv(session, node, say, hooks) env := composerEnv(session, node, say, hooks)
placed, err := selfcompose.Placed(ctx, env.Get, node, link.Backoff, say) unreadable := ""
placed, err := selfcompose.Placed(ctx, env.Get, node, link.Backoff, func(m string) {
if strings.Contains(m, "cannot be read") {
unreadable = m
}
say(m)
}, selfcompose.ComposedItsOwn(ctx, env.Get, node))
if err != nil { if err != nil {
return nil, "its placement could not be read before the link ended: " + err.Error() return nil, "its placement could not be read before the link ended: " + err.Error()
} }
if !placed { if !placed {
why := "the catalogue does not place it" why := "the seat's placement state holds no key for it"
if unreadable != "" {
why = "its placement cannot be read, and it never composed its own: " + unreadable
}
say("this node applies the controller's compositions: " + why) say("this node applies the controller's compositions: " + why)
return nil, why return nil, why
} }
why := "the catalogue places it (" + key + ")" why := "the seat's placement state holds its key (" + key + ")"
say("this node composes its own modules: " + why) say("this node composes its own modules: " + why)
return selfcompose.New(env), why return selfcompose.New(env), why
} }
@@ -915,6 +940,13 @@ func selfComposer(ctx context.Context, session *link.Session, node string, say f
func composerEnv(session *link.Session, node string, say func(string), hooks composerHooks) selfcompose.Env { func composerEnv(session *link.Session, node string, say func(string), hooks composerHooks) selfcompose.Env {
return selfcompose.Env{Node: node, return selfcompose.Env{Node: node,
Watch: func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error) { Watch: func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error) {
if bucket == delivery.PlacementBucket || bucket == delivery.PlacementCandidatesBucket {
// The seat's states are made when their holder starts: a node-engine that starts first waits for the
// bus's word that the bucket was created, and the shadow and the composer say they wait.
return bucketWatch{session, bucket}.WatchAwait(ctx, pattern, func(err error) {
say("the seat's state " + bucket + " is not on the bus yet (" + err.Error() + "): it is watched when the bus says it was created")
})
}
return bucketWatch{session, bucket}.Watch(ctx, pattern) return bucketWatch{session, bucket}.Watch(ctx, pattern)
}, },
Get: func(ctx context.Context, bucket, key string) ([]byte, error) { Get: func(ctx context.Context, bucket, key string) ([]byte, error) {
@@ -937,7 +969,7 @@ func composerEnv(session *link.Session, node string, say func(string), hooks com
Running: hooks.running, Unseal: hooks.unseal, Refused: hooks.refused, Held: hooks.held} Running: hooks.running, Unseal: hooks.unseal, Refused: hooks.refused, Held: hooks.held}
} }
// endWhenUnplaced calls stop once this node's placement key is deleted from the indexer's placement (after the watch // endWhenUnplaced calls stop once this node's placement key is deleted from the seat's placement (after the watch
// caught up, or gone already), so the agent begins again and applies the controller's compositions, as a node the // caught up, or gone already), so the agent begins again and applies the controller's compositions, as a node the
// catalogue does not place does from its start: the way back (issue 1126, the seed's revert). A watch that cannot be // catalogue does not place does from its start: the way back (issue 1126, the seed's revert). A watch that cannot be
// made is said; the node then returns at its next start. // made is said; the node then returns at its next start.
@@ -948,7 +980,7 @@ func endWhenUnplaced(ctx context.Context, session *link.Session, node string, st
} }
updates, err := bucketWatch{session, delivery.PlacementBucket}.Watch(ctx, key) updates, err := bucketWatch{session, delivery.PlacementBucket}.Watch(ctx, key)
if err != nil { if err != nil {
say("whether the catalogue still places this node is not followed (" + err.Error() + "): it returns to the controller's compositions at its next start") say("whether this node's key in the seat's placement state still exists is not followed (" + err.Error() + "): it returns to the controller's compositions at its next start")
return return
} }
placed := false placed := false
@@ -957,7 +989,7 @@ func endWhenUnplaced(ctx context.Context, session *link.Session, node string, st
case u.Caught && placed: case u.Caught && placed:
continue continue
case u.Caught || u.Deleted || len(u.Value) == 0: case u.Caught || u.Deleted || len(u.Value) == 0:
say("the catalogue no longer places this node (" + key + "): the node agent begins again, applying the controller's compositions") say("this node's key in the seat's placement state is gone (" + key + "): the node agent begins again, applying the controller's compositions")
stop() stop()
return return
default: default:
@@ -965,31 +997,42 @@ func endWhenUnplaced(ctx context.Context, session *link.Session, node string, st
} }
} }
if ctx.Err() == nil { if ctx.Err() == nil {
say("the watch of " + key + " ended while the link stays: whether the catalogue still places this node is followed again at the node agent's next start") say("the watch of " + key + " ended while the link stays: whether this node's key in the seat's placement state still exists is followed again at the node agent's next start")
} }
} }
// endWhenPlaced calls stop once this node's placement key appears in the indexer's placement (a value after the watch // endWhenPlaced calls stop once this node's placement key appears in the seat's placement (a value after the watch
// caught up, or one already there), so the agent begins again and composes its own. A watch that cannot be made is said; // caught up, or one already there), so the agent begins again and composes its own. A watch that cannot be made is said;
// the node then switches at its next start. // the node then switches at its next start.
func endWhenPlaced(ctx context.Context, session *link.Session, node string, stop func(), say link.Announce) { func endWhenPlaced(ctx context.Context, session *link.Session, node string, stop func(), say link.Announce, note func(string)) {
key, err := delivery.PlacementKey(node) key, err := delivery.PlacementKey(node)
if err != nil { if err != nil {
return return
} }
updates, err := bucketWatch{session, delivery.PlacementBucket}.Watch(ctx, key) // The seat's placement state is made when its holder starts; the delivery brings the grants part and the holder first,
// so it is normally there. A node-engine that starts before it waits for the bus's word that the bucket was created
// (no timer), says so once in the journal and in the node's own have, where the watchman reads it, and stays on the
// controller's compositions until its key is written. A watch the account may not make is said the same way, and the
// node stays until its next start.
updates, err := bucketWatch{session, delivery.PlacementBucket}.WatchAwait(ctx, key, func(err error) {
why := "this node's key in the seat's placement state is waited for, the state is not on the bus yet (" + err.Error() + "): it stays on the controller's compositions and is watched when the bus says the state was created"
say(why)
note(why)
})
if err != nil { if err != nil {
say("whether the catalogue places this node is not followed (" + err.Error() + "): it composes its own from its next start") why := "this node's key in the seat's placement state cannot be followed (" + err.Error() + "): it stays on the controller's compositions until its next start"
say(why)
note(why)
return return
} }
for u := range updates { for u := range updates {
if !u.Caught && !u.Deleted && len(u.Value) > 0 { if !u.Caught && !u.Deleted && len(u.Value) > 0 {
say("the catalogue places this node now (" + key + "): the node agent begins again, composing its own modules") say("this node's key in the seat's placement state is written now (" + key + "): the node agent begins again, composing its own modules")
stop() stop()
return return
} }
} }
if ctx.Err() == nil { if ctx.Err() == nil {
say("the watch of " + key + " ended while the link stays: whether the catalogue places this node is followed again at the node agent's next start") say("the watch of " + key + " ended while the link stays: whether this node's key in the seat's placement state is written is followed again at the node agent's next start")
} }
} }
cmd/mesh-host/nodeagent_integration_test.go
+14 -17
View File
@@ -419,7 +419,7 @@ func h8once(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, place []
defer b.mu.Unlock() defer b.mu.Unlock()
n := 0 n := 0
for _, l := range b.said { for _, l := range b.said {
if strings.HasPrefix(l, "answering placement candidate feat/x ") { if strings.HasPrefix(l, "answering placement candidate feat-x ") {
n++ n++
} }
} }
@@ -428,7 +428,7 @@ func h8once(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, place []
time.Sleep(time.Second) time.Sleep(time.Second)
settled := answering() settled := answering()
if settled == 0 { if settled == 0 {
t.Fatal("feat/x was never answered") t.Fatal("feat-x was never answered")
} }
hkey, _ := delivery.HaveKey(itNode, "web") hkey, _ := delivery.HaveKey(itNode, "web")
for _, detail := range []string{"its tool web_health: no link to the bus is open", "its tool web_health: no answer within 10s"} { for _, detail := range []string{"its tool web_health: no link to the bus is open", "its tool web_health: no answer within 10s"} {
@@ -447,15 +447,14 @@ func h8once(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, place []
} }
time.Sleep(1500 * time.Millisecond) time.Sleep(1500 * time.Millisecond)
if got := answering(); got != settled { if got := answering(); got != settled {
t.Errorf("feat/x answered %d more time(s) at the same revision, with nothing it is composed from changed", got-settled) t.Errorf("feat-x answered %d more time(s) at the same revision, with nothing it is composed from changed", got-settled)
} }
same, _ := delivery.DecodePlacement(place) same, _ := delivery.DecodePlacement(place)
craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "feat/x", craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "feat-x", Placement: &same})
Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h1b"}, Placement: &same})
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
ckey, _ := delivery.PlacementCandidateKey("feat/x", itNode) ckey, _ := delivery.PlacementCandidateKey("feat-x", itNode)
if _, err := candidates.Put(ckey, craw); err != nil { if _, err := candidates.Put(ckey, craw); err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -482,20 +481,19 @@ func h8back(t *testing.T, b *itBed, placement, resolved, candidates nats.KeyValu
if n := node(); n.Composes != delivery.ComposesOwn { if n := node(); n.Composes != delivery.ComposesOwn {
t.Errorf("a node composing its own says it composes %q", n.Composes) t.Errorf("a node composing its own says it composes %q", n.Composes)
} }
craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "revert/x", craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "revert-x"})
Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 2, Head: "h2"}})
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
ckey, _ := delivery.PlacementCandidateKey("revert/x", itNode) ckey, _ := delivery.PlacementCandidateKey("revert-x", itNode)
if _, err := candidates.Put(ckey, craw); err != nil { if _, err := candidates.Put(ckey, craw); err != nil {
t.Fatal(err) t.Fatal(err)
} }
// The controller holds no composition at web's want (the node composed its own): returning would leave web with // The controller holds no composition at web's want (the node composed its own): returning would leave web with
// nothing to apply, and the answer says so rather than "same" (review of the shadow, round 1). // nothing to apply, and the answer says so rather than "same" (review of the shadow, round 1).
within(t, 10*time.Second, "the node did not refuse returning to compositions the controller does not hold", func() bool { within(t, 10*time.Second, "the node did not refuse returning to compositions the controller does not hold", func() bool {
a := node().Candidates["revert/x"] a := node().Candidates["revert-x"]
return a.Head == "h2" && a.Returns == "" && a.Verdict == delivery.CandidateRefused && return a.Proposal == "revert-x" && a.Returns == "" && a.Verdict == delivery.CandidateRefused &&
strings.Contains(a.Refused, "web: the controller holds no composition") strings.Contains(a.Refused, "web: the controller holds no composition")
}) })
ran := b.have("web").Composed ran := b.have("web").Composed
@@ -531,7 +529,7 @@ func h8back(t *testing.T, b *itBed, placement, resolved, candidates nats.KeyValu
b.mu.Lock() b.mu.Lock()
said := strings.Join(b.said, "\n") said := strings.Join(b.said, "\n")
b.mu.Unlock() b.mu.Unlock()
return strings.Contains(said, "the catalogue no longer places this node") && node().Composes == delivery.ComposesController return strings.Contains(said, "key in the seat's placement state is gone") && node().Composes == delivery.ComposesController
}) })
time.Sleep(3 * time.Second) // what a pass in flight would mark lands within it time.Sleep(3 * time.Second) // what a pass in flight would mark lands within it
select { select {
@@ -683,12 +681,11 @@ func h8rest(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, hkey str
map[string]any{"name": "service-manager", "present": true}}}}) map[string]any{"name": "service-manager", "present": true}}}})
second := b.have("web").Composed second := b.have("web").Composed
same, _ := delivery.DecodePlacement(place) same, _ := delivery.DecodePlacement(place)
craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "feat/x", craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "feat-x", Placement: &same})
Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h1"}, Placement: &same})
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
ckey, _ := delivery.PlacementCandidateKey("feat/x", itNode) ckey, _ := delivery.PlacementCandidateKey("feat-x", itNode)
crev, err := candidates.Put(ckey, craw) crev, err := candidates.Put(ckey, craw)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
@@ -700,8 +697,8 @@ func h8rest(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, hkey str
return false return false
} }
n, err := delivery.DecodeHaveNode(e.Value()) n, err := delivery.DecodeHaveNode(e.Value())
answer = n.Candidates["feat/x"] answer = n.Candidates["feat-x"]
return err == nil && answer.Head == "h1" && n.Record != nil return err == nil && answer.Proposal == "feat-x" && n.Record != nil
}) })
if answer.Verdict != delivery.CandidateSame || answer.Revision != crev || len(answer.Same) != 1 || answer.Same[0] != "web" { if answer.Verdict != delivery.CandidateSame || answer.Revision != crev || len(answer.Same) != 1 || answer.Same[0] != "web" {
t.Errorf("the answer to a candidate equal to the placement: %+v", answer) t.Errorf("the answer to a candidate equal to the placement: %+v", answer)
go.mod
+1 -1
View File
@@ -3,7 +3,7 @@ module github.com/novox/mesh-host
go 1.26.0 go 1.26.0
require ( require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028162235-0de30064156a git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388
github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0 github.com/nats-io/nats.go v1.54.0
golang.org/x/crypto v0.57.0 golang.org/x/crypto v0.57.0
go.sum
+2
View File
@@ -1,5 +1,7 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028162235-0de30064156a h1:BYJrXqPfeNSF0eypX3UnczLq/geXNkfJbiUK9JHlHNA= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028162235-0de30064156a h1:BYJrXqPfeNSF0eypX3UnczLq/geXNkfJbiUK9JHlHNA=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028162235-0de30064156a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028162235-0de30064156a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388 h1:ycu/sNieTdLpD3bzhZOCXdzcoqXUOxPRmqWV019REpo=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
internal/link/have.go
+3 -3
View File
@@ -50,7 +50,7 @@ type haveState struct {
// nodeSealed is, on a node composing its own, each value its own composition carries sealed to its key (ADR 0547 §3, // nodeSealed is, on a node composing its own, each value its own composition carries sealed to its key (ADR 0547 §3,
// issue 1126): kept in its key so a restart reuses them and applies nothing again. // issue 1126): kept in its key so a restart reuses them and applies nothing again.
nodeSealed delivery.Seals nodeSealed delivery.Seals
// nodeCandidates is the node's answer to each placement candidate group (novox/hq ADR 0559 §4), written into its key. // nodeCandidates is the node's answer to each placement candidate proposal (novox/hq ADR 0559 §4), written into its key.
nodeCandidates map[string]delivery.CandidateAnswer nodeCandidates map[string]delivery.CandidateAnswer
// nodeComposes and nodeComposesWhy are whose compositions the node applies and why (HaveNode.Composes), written into // nodeComposes and nodeComposesWhy are whose compositions the node applies and why (HaveNode.Composes), written into
// its key. // its key.
@@ -733,8 +733,8 @@ func (q *Session) NoteNode(ctx context.Context, composed string, sealed delivery
q.writeHaves(ctx, bus) q.writeHaves(ctx, bus)
} }
// NoteCandidates takes the node's answers to the placement candidates, by group (its shadow composer's, novox/hq ADR // NoteCandidates takes the node's answers to the placement candidates, by proposal (its shadow composer's, novox/hq ADR
// 0559 §4), and writes the node's key where it differs; a group left out is no longer answered. // 0559 §4), and writes the node's key where it differs; a proposal left out is no longer answered.
func (q *Session) NoteCandidates(ctx context.Context, answers map[string]delivery.CandidateAnswer) { func (q *Session) NoteCandidates(ctx context.Context, answers map[string]delivery.CandidateAnswer) {
q.init() q.init()
s := &q.haves s := &q.haves
internal/link/have_test.go
+2 -2
View File
@@ -590,9 +590,9 @@ func TestAnAppliedModulesHaveNamesTheFingerprintAndTheDigestAndFollowsItsHealth(
t.Fatalf("the node key read back: %+v %v", n, ok) t.Fatalf("the node key read back: %+v %v", n, ok)
} }
// The node's answers to placement candidates ride in the same key (ADR 0559 §4), and go when no group is answered. // The node's answers to placement candidates ride in the same key (ADR 0559 §4), and go when no group is answered.
answers := map[string]delivery.CandidateAnswer{"feat/x": {Head: "h", Verdict: delivery.CandidateSame, Same: []string{"web"}}} answers := map[string]delivery.CandidateAnswer{"p-1": {Proposal: "p-1", Revision: 1, Verdict: delivery.CandidateSame, Same: []string{"web"}}}
q.NoteCandidates(context.Background(), answers) q.NoteCandidates(context.Background(), answers)
if n, err := delivery.DecodeHaveNode(l.kv["n1.node"].value); err != nil || n.Candidates["feat/x"].Head != "h" || n.Composed != dig { if n, err := delivery.DecodeHaveNode(l.kv["n1.node"].value); err != nil || n.Candidates["p-1"].Proposal != "p-1" || n.Composed != dig {
t.Fatalf("the node key with its answers: %+v %v", n, err) t.Fatalf("the node key with its answers: %+v %v", n, err)
} }
q.NoteCandidates(context.Background(), nil) q.NoteCandidates(context.Background(), nil)
internal/link/history_integration_test.go
+60 -2
View File
@@ -4,6 +4,7 @@ package link
import ( import (
"context" "context"
"sync/atomic"
"testing" "testing"
"time" "time"
@@ -14,7 +15,10 @@ import (
// The revisions a bucket keeps of a key, read over a real bus: oldest first, a delete and a purge said as deleted, and // The revisions a bucket keeps of a key, read over a real bus: oldest first, a delete and a purge said as deleted, and
// none for a key that never was (issue 1945: the node agent takes the composition its want names from them). // none for a key that never was (issue 1945: the node agent takes the composition its want names from them).
func TestNatsAKeysRevisionsAreReadBack(t *testing.T) { func TestNatsAKeysRevisionsAreReadBack(t *testing.T) {
srv, err := natsserver.NewServer(&natsserver.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true}) // A second account reads as a node's does: refused the one bucket "denied-it" (the seat's state before its grant arrives).
srv, err := natsserver.NewServer(&natsserver.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true,
Users: []*natsserver.User{{Username: "admin", Password: "a"}, {Username: "node", Password: "n", Permissions: &natsserver.Permissions{
Publish: &natsserver.SubjectPermission{Allow: []string{">"}, Deny: []string{"$JS.API.*.*.KV_denied-it", "$JS.API.*.*.*.KV_denied-it", "$JS.API.*.*.*.KV_denied-it.>", "$KV.denied-it.>"}}}}}})
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -23,7 +27,7 @@ func TestNatsAKeysRevisionsAreReadBack(t *testing.T) {
t.Fatal("the bus did not come up") t.Fatal("the bus did not come up")
} }
t.Cleanup(srv.Shutdown) t.Cleanup(srv.Shutdown)
conn, err := nats.Connect(srv.ClientURL()) conn, err := nats.Connect(srv.ClientURL(), nats.UserInfo("admin", "a"))
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -65,4 +69,58 @@ func TestNatsAKeysRevisionsAreReadBack(t *testing.T) {
if got, err := bus.KVHistory(ctx, "composed-it", "it.never.x"); err != nil || len(got) != 0 { if got, err := bus.KVHistory(ctx, "composed-it", "it.never.x"); err != nil || len(got) != 0 {
t.Fatalf("a key that never was: %+v (%v)", got, err) t.Fatalf("a key that never was: %+v (%v)", got, err)
} }
// A bucket that is not there, and one the account may not read, are errors at once, never a wait: the node agent
// follows its placement key by one watch at its start (endWhenPlaced) and, refused, stays on the controller's compositions and says why.
if _, err := bus.KVWatch(ctx, "never-it", ">"); err == nil {
t.Error("a watch of a bucket that is not there gave no error")
}
if _, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "denied-it", History: 1}); err != nil {
t.Fatal(err)
}
nodeConn, err := nats.Connect(srv.ClientURL(), nats.UserInfo("node", "n"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(nodeConn.Close)
started := time.Now()
wctx, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()
if _, err := (OverNATS{Conn: nodeConn}).KVWatch(wctx, "denied-it", ">"); err == nil || wctx.Err() != nil {
t.Errorf("a watch the account may not make: error %v after %s, context %v", err, time.Since(started), wctx.Err())
} else {
t.Logf("a refused watch is an error after %s: %v", time.Since(started).Round(time.Millisecond), err)
}
// A bucket made after the watch was asked is then watched, on the bus's word that it was created and with no timer;
// a bucket the account may not read is no waiting but the error (endWhenPlaced, the shadow's candidates).
q := &Session{Membership: Membership{Node: "n1"}}
q.Attach(ctx, OverNATS{Conn: conn, JS: js})
var waited atomic.Int32
actx, acancel := context.WithTimeout(ctx, 20*time.Second)
defer acancel()
ch, err := q.KVWatchAwait(actx, "later-it", ">", func(error) { waited.Add(1) })
if err != nil || waited.Load() != 1 {
t.Fatalf("a bucket not there yet: watch %v, waiting told %d time(s)", err, waited.Load())
}
select {
case u := <-ch:
t.Fatalf("an update before the bucket exists: %+v", u)
case <-time.After(200 * time.Millisecond):
}
later, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "later-it", History: 1})
if err != nil {
t.Fatal(err)
}
if _, err := later.PutString("placement.n1", "v"); err != nil {
t.Fatal(err)
}
for seen := false; !seen; {
select {
case u := <-ch:
seen = u.Key == "placement.n1" && string(u.Value) == "v"
case <-actx.Done():
t.Fatal("the bucket made after the watch was never watched")
}
}
} }
internal/link/watch.go
+57
View File
@@ -239,3 +239,60 @@ func (q *Session) Hear(subject string) (<-chan []byte, func(), error) {
} }
return h.Hear(subject) return h.Hear(subject)
} }
// CreatedAdvisory is the subject the bus says a bucket's stream was created on (the server's own advisory; the same
// subject mesh-sdk's process.AwaitBucket binds on, and the grants part gives a principal for each bucket it may read).
func CreatedAdvisory(bucket string) string {
return "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_" + bucket
}
// KVWatchAwait is KVWatch for a bucket that may not be on the bus yet (the placement seat's states, made when their
// holder starts): the channel is returned at once, and while the bucket is missing it yields nothing and waiting is told
// why; when the bus says the bucket's stream was created the watch is made and its updates follow. No timer: the
// advisory is heard before the first try, so a creation between the try and the wait is not missed. Any other error of
// the first try is the caller's, as for KVWatch: a bucket the account may not read is not waited for.
func (q *Session) KVWatchAwait(ctx context.Context, bucket, pattern string, waiting func(error)) (<-chan KVUpdate, error) {
created, stop, err := q.Hear(CreatedAdvisory(bucket))
if err != nil {
return nil, err
}
in, err := q.KVWatch(ctx, bucket, pattern)
if err == nil {
stop()
return in, nil
}
if !errors.Is(err, nats.ErrBucketNotFound) {
stop()
return nil, err
}
waiting(err)
out := make(chan KVUpdate, 64)
go func() {
defer close(out)
defer stop()
for in == nil {
select {
case <-ctx.Done():
return
case _, ok := <-created:
if !ok {
return
}
w, err := q.KVWatch(ctx, bucket, pattern)
if err != nil {
waiting(err)
continue
}
in = w
}
}
for u := range in {
select {
case out <- u:
case <-ctx.Done():
return
}
}
}()
return out, nil
}
internal/ownresync/ownresync.go
+4 -4
View File
@@ -89,12 +89,12 @@ func New(env Env) *Resync {
func (r *Resync) Take(ctx context.Context, answers map[string]delivery.CandidateAnswer) []Outcome { func (r *Resync) Take(ctx context.Context, answers map[string]delivery.CandidateAnswer) []Outcome {
seen := map[delivery.OwnValueDiffers]bool{} seen := map[delivery.OwnValueDiffers]bool{}
var drift []delivery.OwnValueDiffers var drift []delivery.OwnValueDiffers
groups := make([]string, 0, len(answers)) proposals := make([]string, 0, len(answers))
for g := range answers { for g := range answers {
groups = append(groups, g) proposals = append(proposals, g)
} }
sort.Strings(groups) sort.Strings(proposals)
for _, g := range groups { for _, g := range proposals {
for _, o := range delivery.OwnValuesDiffer(answers[g]) { for _, o := range delivery.OwnValuesDiffer(answers[g]) {
if !seen[o] { if !seen[o] {
seen[o] = true seen[o] = true
internal/ownresync/ownresync_test.go
+1 -1
View File
@@ -20,7 +20,7 @@ func TestTheNodeHandsTheVaultWhatRunsOfADriftedOwnSecret(t *testing.T) {
digest := "sha256:" + strings.Repeat("a", 64) digest := "sha256:" + strings.Repeat("a", 64)
fp := "src2:" + strings.Repeat("b", 64) fp := "src2:" + strings.Repeat("b", 64)
sum := sha256.Sum256([]byte(running)) sum := sha256.Sum256([]byte(running))
answers := map[string]delivery.CandidateAnswer{"g": {Head: "h", Verdict: delivery.CandidateDiffers, answers := map[string]delivery.CandidateAnswer{"g": {Proposal: "p-1", Revision: 1, Verdict: delivery.CandidateDiffers,
Differs: []delivery.ModuleDiffers{{Module: "only-office", Why: "the resource differs", Bodies: []delivery.ResourceBodies{ Differs: []delivery.ModuleDiffers{{Module: "only-office", Why: "the resource differs", Bodies: []delivery.ResourceBodies{
{ID: "only-office.needs-jwt", Composed: `{"sealed":"` + delivery.MaskedAnotherValue + `"}`, Running: `{"sealed":"` + delivery.MaskedAnotherValue + `"}`}, {ID: "only-office.needs-jwt", Composed: `{"sealed":"` + delivery.MaskedAnotherValue + `"}`, Running: `{"sealed":"` + delivery.MaskedAnotherValue + `"}`},
{ID: "only-office.needs-salt", Composed: `{"sealed":"<masked: another seal of the same value>"}`}}}}}} {ID: "only-office.needs-salt", Composed: `{"sealed":"<masked: another seal of the same value>"}`}}}}}}
internal/selfcompose/selfcompose.go
+32 -4
View File
@@ -483,7 +483,7 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
if !placed { if !placed {
// The catalogue no longer places it: the agent returns to the controller's compositions (endWhenUnplaced), and // 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). // this is no refusal of what it runs (issue 1126, the seed's revert).
c.sayOnce("the catalogue no longer places this node: it composes nothing of its own, and keeps running what it runs") 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.owed.Store(false) // nothing of its own to compose, so no pass is owed and no timer armed again c.owed.Store(false) // nothing of its own to compose, so no pass is owed and no timer armed again
return nil return nil
} }
@@ -993,8 +993,14 @@ func (r reader) Credentials(ctx context.Context, node string, asConsumer, asProv
// marker. A read that fails chooses neither (review of the shadow, round 1: a timeout or a bus not ready at a restart // 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), // 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. // 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, 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)) (bool, error) { wait func(try int) time.Duration, say func(string), switched bool) (bool, error) {
key, err := delivery.PlacementKey(node) key, err := delivery.PlacementKey(node)
if err != nil { if err != nil {
return false, err return false, err
@@ -1003,7 +1009,7 @@ func Placed(ctx context.Context, get func(ctx context.Context, bucket, key strin
for try := 1; ; try++ { for try := 1; ; try++ {
raw, err := get(ctx, delivery.PlacementBucket, key) raw, err := get(ctx, delivery.PlacementBucket, key)
if errors.Is(err, nats.ErrBucketNotFound) { if errors.Is(err, nats.ErrBucketNotFound) {
return false, nil // the indexer made no placement state yet: nothing places it, read and known return false, nil // the seat has written no placement state yet: nothing places it, read and known
} }
if err == nil { if err == nil {
return len(raw) > 0, nil return len(raw) > 0, nil
@@ -1011,8 +1017,12 @@ func Placed(ctx context.Context, get func(ctx context.Context, bucket, key strin
if ctx.Err() != nil { if ctx.Err() != nil {
return false, ctx.Err() 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 { if why := err.Error(); why != said {
say(fmt.Sprintf("whether the catalogue places this node cannot be read (%s): the node agent waits, choosing neither its own compositions nor the controller's: %v", key, err)) 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 said = why
} }
select { select {
@@ -1023,6 +1033,24 @@ func Placed(ctx context.Context, get func(ctx context.Context, bucket, key strin
} }
} }
// 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. // 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} var composeAgain = []time.Duration{time.Second, 2 * time.Second, 4 * time.Second}
internal/selfcompose/selfcompose_test.go
+54 -23
View File
@@ -762,8 +762,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
for _, m := range modules { for _, m := range modules {
p.Modules = append(p.Modules, delivery.PlacedModule{Module: m}) p.Modules = append(p.Modules, delivery.PlacedModule{Module: m})
} }
raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: group, raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: group, Placement: &p})
Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h-" + group}, Placement: &p})
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -776,13 +775,21 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
candidate("ghost", 6, "lemurs", "ghost") candidate("ghost", 6, "lemurs", "ghost")
// A head deleting n1's placement file (the seed's revert, mesh-catalog #526): n1 returns to the controller's // A head deleting n1's placement file (the seed's revert, mesh-catalog #526): n1 returns to the controller's
// compositions, answered as such, not as every module removed. // compositions, answered as such, not as every module removed.
if raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "deletes", if raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "deletes"}); err != nil {
Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h-deletes"}}); err != nil {
t.Fatal(err) t.Fatal(err)
} else { } else {
k, _ := delivery.PlacementCandidateKey("deletes", "n1") k, _ := delivery.PlacementCandidateKey("deletes", "n1")
sb.put(delivery.PlacementCandidatesBucket, k, raw, 7) sb.put(delivery.PlacementCandidatesBucket, k, raw, 7)
} }
// A proposal that changes another node's key and reaches n1's composition (ADR 0567 §3): n1's candidate holds no
// placement of its own change (Reached, nil), and n1 answers against its current placement, never as one whose key
// the proposal takes away.
if raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "reach", Reached: true}); err != nil {
t.Fatal(err)
} else {
k, _ := delivery.PlacementCandidateKey("reach", "n1")
sb.put(delivery.PlacementCandidatesBucket, k, raw, 8)
}
sb.caught() sb.caught()
var answers map[string]delivery.CandidateAnswer var answers map[string]delivery.CandidateAnswer
waitFor := func(why string, ok func(map[string]delivery.CandidateAnswer) bool) { waitFor := func(why string, ok func(map[string]delivery.CandidateAnswer) bool) {
@@ -796,7 +803,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
} }
} }
} }
waitFor("every group answered", func(a map[string]delivery.CandidateAnswer) bool { return len(a) == 5 }) waitFor("every group answered", func(a map[string]delivery.CandidateAnswer) bool { return len(a) == 6 })
// Every watch and get the shadow makes is one the node's grant allows (issue 1126, 2026-10-27 09:21). // Every watch and get the shadow makes is one the node's grant allows (issue 1126, 2026-10-27 09:21).
sb.mu.Lock() sb.mu.Lock()
for _, w := range sb.watches { for _, w := range sb.watches {
@@ -814,7 +821,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
readOnly.mu.Unlock() readOnly.mu.Unlock()
for group, want := range map[string]func(delivery.CandidateAnswer) bool{ for group, want := range map[string]func(delivery.CandidateAnswer) bool{
"same": func(a delivery.CandidateAnswer) bool { "same": func(a delivery.CandidateAnswer) bool {
return a.Verdict == delivery.CandidateSame && a.Head == "h-same" && a.Revision == 3 && len(a.Same) == 2 return a.Verdict == delivery.CandidateSame && a.Proposal == "same" && a.Revision == 3 && len(a.Same) == 2
}, },
"adds": func(a delivery.CandidateAnswer) bool { "adds": func(a delivery.CandidateAnswer) bool {
return a.Verdict == delivery.CandidateDiffers && len(a.Added) == 1 && a.Added[0].Module == "keeper" && return a.Verdict == delivery.CandidateDiffers && len(a.Added) == 1 && a.Added[0].Module == "keeper" &&
@@ -826,6 +833,10 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
"ghost": func(a delivery.CandidateAnswer) bool { "ghost": func(a delivery.CandidateAnswer) bool {
return a.Verdict == delivery.CandidateRefused && a.Refused != "" return a.Verdict == delivery.CandidateRefused && a.Refused != ""
}, },
"reach": func(a delivery.CandidateAnswer) bool {
return a.Verdict == delivery.CandidateSame && a.Returns == "" && a.Revision == 8 && len(a.Removed) == 0 &&
fmt.Sprint(a.Same) == "[lemurs node]"
},
"deletes": func(a delivery.CandidateAnswer) bool { "deletes": func(a delivery.CandidateAnswer) bool {
return a.Verdict == delivery.CandidateSame && a.Returns == delivery.ReturnsToController && a.Revision == 7 && return a.Verdict == delivery.CandidateSame && a.Returns == delivery.ReturnsToController && a.Revision == 7 &&
len(a.Removed) == 0 && fmt.Sprint(a.Same) == "[lemurs node]" len(a.Removed) == 0 && fmt.Sprint(a.Same) == "[lemurs node]"
@@ -935,7 +946,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
} }
k, _ := delivery.PlacementCandidateKey("same", "n1") k, _ := delivery.PlacementCandidateKey("same", "n1")
sb.del(delivery.PlacementCandidatesBucket, k) sb.del(delivery.PlacementCandidatesBucket, k)
waitFor("the deleted group's answer stays", func(a map[string]delivery.CandidateAnswer) bool { _, ok := a["same"]; return !ok && len(a) == 4 }) waitFor("the deleted group's answer stays", func(a map[string]delivery.CandidateAnswer) bool { _, ok := a["same"]; return !ok && len(a) == 5 })
// Review of L2b, round 2: a watch ends while the link stays; a group's key is deleted before the shadow watches // Review of L2b, round 2: a watch ends while the link stays; a group's key is deleted before the shadow watches
// again. The new run reads afresh, so that group is no longer answered. // again. The new run reads afresh, so that group is no longer answered.
@@ -947,7 +958,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
sb.missing(delivery.PlacementCandidatesBucket, false) sb.missing(delivery.PlacementCandidatesBucket, false)
waitFor("a group deleted while the shadow was down is still answered", func(a map[string]delivery.CandidateAnswer) bool { waitFor("a group deleted while the shadow was down is still answered", func(a map[string]delivery.CandidateAnswer) bool {
_, ok := a["drops"] _, ok := a["drops"]
return !ok && len(a) == 3 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 // Whose compositions a node applies at a start (review of the shadow, round 1): only a read that succeeds chooses; a
@@ -959,12 +970,15 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
placed bool placed bool
tries int tries int
quiet bool // read and known: nothing to say 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
}{ }{
"placed after two failures": {answers: []error{fmt.Errorf("nats: timeout"), fmt.Errorf("nats: timeout")}, value: []byte(`{}`), placed: true, tries: 3}, "switched: placed after two failures": {answers: []error{fmt.Errorf("nats: timeout"), fmt.Errorf("nats: timeout")}, value: []byte(`{}`), placed: true, tries: 3, switched: true},
"no key": {placed: false, tries: 1}, "never switched: a timeout is not placed, said once": {answers: []error{fmt.Errorf("nats: timeout")}, placed: false, tries: 1},
"no placement state yet": {answers: []error{fmt.Errorf("the bucket %s: %w", delivery.PlacementBucket, nats.ErrBucketNotFound)}, placed: false, tries: 1, quiet: true}, "no key": {placed: false, tries: 1},
"failing until the link ends": {answers: []error{fmt.Errorf("jetstream not ready"), fmt.Errorf("jetstream not ready"), "no placement state yet": {answers: []error{fmt.Errorf("the bucket %s: %w", delivery.PlacementBucket, nats.ErrBucketNotFound)}, placed: false, tries: 1, quiet: true},
fmt.Errorf("jetstream not ready")}, end: true, tries: 3}, "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) lctx, lcancel := context.WithCancel(ctx)
tries := 0 tries := 0
@@ -979,13 +993,32 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
} }
return row.value, nil return row.value, nil
} }
placed, err := Placed(lctx, get, "n1", func(int) time.Duration { return time.Millisecond }, func(s string) { said = append(said, s) }) placed, err := Placed(lctx, get, "n1", func(int) time.Duration { return time.Millisecond }, func(s string) { said = append(said, s) }, row.switched)
lcancel() lcancel()
if placed != row.placed || tries != row.tries || (err != nil) != row.end || (len(row.answers) > 0 && !row.quiet) != (len(said) == 1) { 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) 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)
}
}
// Issue 1126, the shadow shows its work: one input not read after the wait is said, naming it, in the journal and in // Issue 1126, the shadow shows its work: one input not read after the wait is said, naming it, in the journal and in
// a refused answer to each group known; once read, the group is answered for real. // a refused answer to each group known; once read, the group is answered for real.
setFor(t, &notReadyAfter, 30*time.Millisecond) setFor(t, &notReadyAfter, 30*time.Millisecond)
@@ -1008,13 +1041,11 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
raw, _ := delivery.EncodeModule(m) raw, _ := delivery.EncodeModule(m)
nb.put(delivery.ModulesBucket, name, raw, 1) nb.put(delivery.ModulesBucket, name, raw, 1)
} }
sameRaw, _ := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "same", sameRaw, _ := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "same", Placement: &trunk})
Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h-same"}, Placement: &trunk})
sameKey, _ := delivery.PlacementCandidateKey("same", "n1") sameKey, _ := delivery.PlacementCandidateKey("same", "n1")
nb.put(delivery.PlacementCandidatesBucket, sameKey, sameRaw, 3) nb.put(delivery.PlacementCandidatesBucket, sameKey, sameRaw, 3)
// Issue 1126, T3: an older group (a closed pull request's candidate, rev 2) is answered after the newest. // Issue 1126, T3: an older group (a closed pull request's candidate, rev 2) is answered after the newest.
oldRaw, _ := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "older", oldRaw, _ := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "older", Placement: &trunk})
Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 2, Head: "h-old"}, Placement: &trunk})
oldKey, _ := delivery.PlacementCandidateKey("older", "n1") oldKey, _ := delivery.PlacementCandidateKey("older", "n1")
nb.put(delivery.PlacementCandidatesBucket, oldKey, oldRaw, 2) nb.put(delivery.PlacementCandidatesBucket, oldKey, oldRaw, 2)
for _, bucket := range []string{delivery.PlacementCandidatesBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket} { for _, bucket := range []string{delivery.PlacementCandidatesBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket} {
@@ -1022,7 +1053,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
} }
select { select {
case a := <-nanswered: case a := <-nanswered:
if g := a["same"]; g.Verdict != delivery.CandidateRefused || !strings.Contains(g.Refused, delivery.AssignerWantsBucket) || g.Head != "h-same" { if g := a["same"]; g.Verdict != delivery.CandidateRefused || !strings.Contains(g.Refused, delivery.AssignerWantsBucket) || g.Proposal != "same" {
t.Errorf("a shadow waiting for its wants answered %+v", g) t.Errorf("a shadow waiting for its wants answered %+v", g)
} }
case <-time.After(5 * time.Second): case <-time.After(5 * time.Second):
@@ -1072,7 +1103,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
// Issue 1126, probe #532 on shanks: what runs is the composition the have names, never the copy of the declaration // Issue 1126, probe #532 on shanks: what runs is the composition the have names, never the copy of the declaration
// kept before the switch (no fp, no digest): a module composed at the digest its have says runs is the same. // kept before the switch (no fp, no digest): a module composed at the digest its have says runs is the same.
{ {
probe := delivery.CandidateAnswer{Head: "h", Revision: 68, Verdict: delivery.CandidateDiffers, Same: []string{"adwaita"}, probe := delivery.CandidateAnswer{Proposal: "pro", Revision: 68, Verdict: delivery.CandidateDiffers, Same: []string{"adwaita"},
Differs: []delivery.ModuleDiffers{ Differs: []delivery.ModuleDiffers{
{Module: "xterm", Composed: "sha256:3c0a31df64283c0678b1cdbeff8ac030609cab65cc7e47faaf714da743d9340e", Why: "the fingerprint differs"}, {Module: "xterm", Composed: "sha256:3c0a31df64283c0678b1cdbeff8ac030609cab65cc7e47faaf714da743d9340e", Why: "the fingerprint differs"},
{Module: "zsh", Composed: "sha256:6093f31056d39eaf604410b7289ec6b5fe602b19c8547283562fcfc078c4844f", Resource: "zsh.bundle-tools", Why: "the resource differs"}, {Module: "zsh", Composed: "sha256:6093f31056d39eaf604410b7289ec6b5fe602b19c8547283562fcfc078c4844f", Resource: "zsh.bundle-tools", Why: "the resource differs"},
@@ -1085,7 +1116,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
if fmt.Sprint(got.Same) != "[adwaita xterm zsh]" || len(got.Differs) != 1 || got.Differs[0].Module != "brother" || got.Verdict != delivery.CandidateDiffers { if fmt.Sprint(got.Same) != "[adwaita xterm zsh]" || len(got.Differs) != 1 || got.Differs[0].Module != "brother" || got.Verdict != delivery.CandidateDiffers {
t.Errorf("shanks' probe answered %+v", got) t.Errorf("shanks' probe answered %+v", got)
} }
if one := asHavesSay(delivery.CandidateAnswer{Head: "h", Verdict: delivery.CandidateDiffers, Differs: probe.Differs[:1]}, if one := asHavesSay(delivery.CandidateAnswer{Proposal: "pro", Revision: 1, Verdict: delivery.CandidateDiffers, Differs: probe.Differs[:1]},
map[string]string{"xterm": probe.Differs[0].Composed}); one.Verdict != delivery.CandidateSame { map[string]string{"xterm": probe.Differs[0].Composed}); one.Verdict != delivery.CandidateSame {
t.Errorf("a module the same as its have says is answered %+v", one) t.Errorf("a module the same as its have says is answered %+v", one)
} }
@@ -1121,11 +1152,11 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
defer readOnly.mu.Unlock() defer readOnly.mu.Unlock()
return readOnly.kv[bucket+"/"+key], nil return readOnly.kv[bucket+"/"+key], nil
} }
a := returning(ctx, get, "n1", map[string]delivery.Want{"lemurs": {FP: fp}, agent.NodeModule: {}}, runsList, "h", 1) a := returning(ctx, get, "n1", map[string]delivery.Want{"lemurs": {FP: fp}, agent.NodeModule: {}}, runsList, "pro", 1)
if !row.check(a) { if !row.check(a) {
t.Errorf("%s: %+v", name, a) t.Errorf("%s: %+v", name, a)
} }
if _, err := delivery.EncodeHaveNode(delivery.HaveNode{Candidates: map[string]delivery.CandidateAnswer{"g": a}}); err != nil { 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) t.Errorf("%s: the answer is not one a have holds: %v", name, err)
} }
} }
internal/selfcompose/shadow.go
+83 -86
View File
@@ -16,8 +16,8 @@ import (
"github.com/novox/mesh-host/internal/agent" "github.com/novox/mesh-host/internal/agent"
) )
// Shadow is the node's answer to placement candidates (novox/hq ADR 0559 §4, issue 1126): for every group with a key in // Shadow is the node's answer to placement candidates (novox/hq ADR 0559 §4, issue 1126): for every proposal with a key in
// the indexer's state `placement-candidates`, it composes this node in shadow from the trunk's placement with the group's // the placement seat's state `candidates`, it composes this node in shadow from the trunk's placement with the proposal's
// keys laid over it and its own inputs (the sealed values and boxes it holds among them), compares the result with what // keys laid over it and its own inputs (the sealed values and boxes it holds among them), compares the result with what
// the node runs, and hands the answers to be written into the node's own have key, `<node>.node`, under `candidates`. // the node runs, and hands the answers to be written into the node's own have key, `<node>.node`, under `candidates`.
// It applies nothing and hands no composition: what runs is still composed from `placement` alone, by the Composer. // It applies nothing and hands no composition: what runs is still composed from `placement` alone, by the Composer.
@@ -25,37 +25,37 @@ import (
// **Event-driven, like the Composer.** It watches every candidate key, the trunk's placement, every node's record, the // **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 // 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 // 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 group's answer is dropped when its // against what it runs from them. An answer is handed only when it changes, and a proposal's answer is dropped when its
// keys go. // keys go.
type Shadow struct { type Shadow struct {
c *Composer c *Composer
answer func(ctx context.Context, answers map[string]delivery.CandidateAnswer) answer func(ctx context.Context, answers map[string]delivery.CandidateAnswer)
mu sync.Mutex mu sync.Mutex
candidates map[string]map[string]candidateKey // by group, by node candidates map[string]map[string]candidateKey // by proposal, by node
answers map[string]delivery.CandidateAnswer answers map[string]delivery.CandidateAnswer
// wants are the node's wants by module (agent.NodeModule for its own): what a head deleting its placement returns // 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. // it to, the controller's compositions at them.
wants map[string]delivery.Want 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 // 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). // from the declaration kept before the switch (issue 1126, probe #532 on shanks).
ran map[string]string ran map[string]string
// answered is, by group, the candidate revisions and the input changes (moved) each answer was composed at: a group // answered is, by proposal, the candidate revisions and the input changes (moved) each answer was composed at: a proposal
// is answered again only when either moved (issue 1126, 2026-10-28: shanks answered probe/1126-seed-7 at revision 84 // is answered again only when either moved (issue 1126, 2026-10-28: shanks answered probe/1126-seed-7 at revision 84
// every 13-17 s, woken by its own modules' haves whose health words changed at every look). // every 13-17 s, woken by its own modules' haves whose health words changed at every look).
answered map[string]answeredAt answered map[string]answeredAt
// moved counts, by input (inputWords), the changes of each of the node's own inputs, and retry names the groups // moved counts, by input (inputWords), the changes of each of the node's own inputs, and retry names the proposals
// answered again after a read the bus did not answer: what woke each answer is said with it (issue 1126, // answered again after a read the bus did not answer: what woke each answer is said with it (issue 1126,
// 2026-10-28, seed-9's probe: a safety check could not tell a loop from inputs moving). // 2026-10-28, seed-9's probe: a safety check could not tell a loop from inputs moving).
moved [len(inputWords)]uint64 moved [len(inputWords)]uint64
retry map[string]bool retry map[string]bool
// owed says an answer met a read the bus did not answer: it is no answer to keep, and its group is answered again // owed says an answer met a read the bus did not answer: it is no answer to keep, and its proposal is answered again
// after a bound (laterAgain), whether or not an input moves (issue 1126, 2026-10-28: ace kept "the vault's own: not // after a bound (laterAgain), whether or not an input moves (issue 1126, 2026-10-28: ace kept "the vault's own: not
// yet ... no responders" in its answer to seed-7). // yet ... no responders" in its answer to seed-7).
owed bool owed bool
} }
// answeredAt is what one group's answer was composed at: its keys' revisions, by node, and the changes of each input. // answeredAt is what one proposal's answer was composed at: its keys' revisions, by node, and the changes of each input.
type answeredAt struct { type answeredAt struct {
revs string revs string
moved [len(inputWords)]uint64 moved [len(inputWords)]uint64
@@ -64,7 +64,7 @@ type answeredAt struct {
// inputWords name the shadow's inputs, in Run's order, as an answer says which of them woke it. // inputWords name the shadow's inputs, in Run's order, as an answer says which of them woke it.
var inputWords = [...]string{"the candidate", "placement", "records", "modules", "values", "haves", "wants"} var inputWords = [...]string{"the candidate", "placement", "records", "modules", "values", "haves", "wants"}
// wokenBy says what moved between the answer a group stands at (had, when done) and the one about to be composed (at): // wokenBy says what moved between the answer a proposal stands at (had, when done) and the one about to be composed (at):
// its first answer, a retry, its candidate's revisions, or the inputs that changed, by their words. // its first answer, a retry, its candidate's revisions, or the inputs that changed, by their words.
func wokenBy(had answeredAt, done, retry bool, at answeredAt) string { func wokenBy(had answeredAt, done, retry bool, at answeredAt) string {
switch { switch {
@@ -157,7 +157,7 @@ func (s *Shadow) Run(ctx context.Context) error {
{ResolvedBucket, node + ".>", (*Composer).takeValue, nil}, {ResolvedBucket, node + ".>", (*Composer).takeValue, nil},
// What this node runs moved: an apply ends in its module's have. // What this node runs moved: an apply ends in its module's have.
{delivery.HaveBucket, node + ".>", nil, nil}, {delivery.HaveBucket, node + ".>", nil, nil},
// Its wants: a head deleting its placement returns it to the controller's compositions at them. // Its wants: a proposal deleting its placement returns it to the controller's compositions at them.
{delivery.AssignerWantsBucket, "want." + node + ".>", nil, nil}, {delivery.AssignerWantsBucket, "want." + node + ".>", nil, nil},
} }
type tagged struct { type tagged struct {
@@ -187,7 +187,7 @@ func (s *Shadow) Run(ctx context.Context) error {
} }
caught := make([]bool, len(ins)) caught := make([]bool, len(ins))
ready, dirty := false, false ready, dirty := false, false
// Not ready after a minute is said, naming each input not read yet, in the journal and in an answer to each group it // Not ready after a minute is said, naming each input not read yet, in the journal and in an answer to each proposal it
// knows: a shadow that waits is never silent (issue 1126: ace, g14 and shanks answered nothing to seed #531). // knows: a shadow that waits is never silent (issue 1126: ace, g14 and shanks answered nothing to seed #531).
waiting := time.NewTimer(notReadyAfter) waiting := time.NewTimer(notReadyAfter)
defer waiting.Stop() defer waiting.Stop()
@@ -306,7 +306,7 @@ func (s *Shadow) entry(bucket, key string) (v any, has, known bool) {
return nil, false, false return nil, false, false
} }
// revsOf is a group's keys' revisions, by node, as one word. // revsOf is a proposal's keys' revisions, by node, as one word.
func revsOf(keys map[string]candidateKey) string { func revsOf(keys map[string]candidateKey) string {
nodes := make([]string, 0, len(keys)) nodes := make([]string, 0, len(keys))
for n := range keys { for n := range keys {
@@ -322,35 +322,35 @@ func revsOf(keys map[string]candidateKey) string {
// takeCandidate takes one candidate key, and says whether it changed anything. // takeCandidate takes one candidate key, and says whether it changed anything.
func (s *Shadow) takeCandidate(u agent.Update) bool { func (s *Shadow) takeCandidate(u agent.Update) bool {
group, node, err := delivery.ParsePlacementCandidateKey(u.Key) proposal, node, err := delivery.ParsePlacementCandidateKey(u.Key)
if err != nil { if err != nil {
return false return false
} }
s.mu.Lock() s.mu.Lock()
defer s.mu.Unlock() defer s.mu.Unlock()
if u.Deleted { if u.Deleted {
if _, ok := s.candidates[group][node]; !ok { if _, ok := s.candidates[proposal][node]; !ok {
return false return false
} }
delete(s.candidates[group], node) delete(s.candidates[proposal], node)
if len(s.candidates[group]) == 0 { if len(s.candidates[proposal]) == 0 {
delete(s.candidates, group) delete(s.candidates, proposal)
} }
return true return true
} }
c, err := delivery.DecodePlacementCandidate(u.Value) c, err := delivery.DecodePlacementCandidate(u.Value)
if err != nil || c.Group != group || (c.Placement != nil && c.Placement.Node != node) { if err != nil || c.Proposal != proposal || (c.Placement != nil && c.Placement.Node != node) {
// Not one: read as gone. // Not one: read as gone.
if _, ok := s.candidates[group][node]; ok { if _, ok := s.candidates[proposal][node]; ok {
delete(s.candidates[group], node) delete(s.candidates[proposal], node)
return true return true
} }
return false return false
} }
if s.candidates[group] == nil { if s.candidates[proposal] == nil {
s.candidates[group] = map[string]candidateKey{} s.candidates[proposal] = map[string]candidateKey{}
} }
s.candidates[group][node] = candidateKey{c: c, rev: u.Revision} s.candidates[proposal][node] = candidateKey{c: c, rev: u.Revision}
return true return true
} }
@@ -372,10 +372,10 @@ func (s *Shadow) takeRecord(u agent.Update) bool {
return had != has || (has && !delivery.SameNodeRecord(before, after)) return had != has || (has && !delivery.SameNodeRecord(before, after))
} }
// How long the shadow waits before it says it is not ready, and how long a group's answer runs before it is said slow. // How long the shadow waits before it says it is not ready, and how long a proposal's answer runs before it is said slow.
var notReadyAfter, slowAfter = time.Minute, time.Minute var notReadyAfter, slowAfter = time.Minute, time.Minute
// notReady says which inputs the shadow has not read yet, and answers each group it knows refused for it. // notReady says which inputs the shadow has not read yet, and answers each proposal it knows refused for it.
func (s *Shadow) notReady(ctx context.Context, ins []input, caught []bool) { func (s *Shadow) notReady(ctx context.Context, ins []input, caught []bool) {
var waits []string var waits []string
for i, in := range ins { for i, in := range ins {
@@ -387,15 +387,12 @@ func (s *Shadow) notReady(ctx context.Context, ins []input, caught []bool) {
s.c.env.Say("this node answers no placement candidate yet: " + why) s.c.env.Say("this node answers no placement candidate yet: " + why)
s.mu.Lock() s.mu.Lock()
next := map[string]delivery.CandidateAnswer{} next := map[string]delivery.CandidateAnswer{}
for group, keys := range s.candidates { for proposal, keys := range s.candidates {
a := delivery.CandidateAnswer{Refused: why, Verdict: delivery.CandidateRefused} own, ok := keys[s.c.env.Node]
for n, k := range keys { if !ok {
a.Head = k.c.Pull.Head continue // no key of this node's: nothing asks it to answer
if n == s.c.env.Node {
a.Revision = k.rev
}
} }
next[group] = a next[proposal] = delivery.CandidateAnswer{Proposal: proposal, Revision: own.rev, Refused: why, Verdict: delivery.CandidateRefused}
} }
s.answers = next s.answers = next
s.mu.Unlock() s.mu.Unlock()
@@ -404,9 +401,9 @@ func (s *Shadow) notReady(ctx context.Context, ins []input, caught []bool) {
} }
} }
// answerAll answers every group, the newest candidate first, and hands the answers each time one changes (issue 1126, // answerAll answers every proposal, the newest candidate first, and hands the answers each time one changes (issue 1126,
// T3: ace answered a closed pull request's group for 22 minutes before the newer one, all handed at the end). Newest // T3: ace answered a closed pull request's group for 22 minutes before the newer one, all handed at the end). Newest
// is the highest revision of the group's keys: a closed pull request's candidate is the older and waits; the indexer // is the highest revision of the proposal's keys: a withdrawn proposal's candidate is the older and waits; the seat
// drops a closed one's keys, which ends its answer here (a node does not read the forge). Each answer says when it // drops a closed one's keys, which ends its answer here (a node does not read the forge). Each answer says when it
// starts, every progressEvery while it runs what it waits on, and when it ends (T2: the node said nothing for 22 // starts, every progressEvery while it runs what it waits on, and when it ends (T2: the node said nothing for 22
// minutes). // minutes).
@@ -415,38 +412,38 @@ func (s *Shadow) answerAll(ctx context.Context) {
node := c.env.Node node := c.env.Node
s.mu.Lock() s.mu.Lock()
type listed struct { type listed struct {
group string proposal string
rev uint64 rev uint64
at answeredAt at answeredAt
} }
var groups []listed var proposals []listed
for g, keys := range s.candidates { for g, keys := range s.candidates {
l := listed{group: g, at: answeredAt{revs: revsOf(keys), moved: s.moved}} l := listed{proposal: g, at: answeredAt{revs: revsOf(keys), moved: s.moved}}
for _, k := range keys { for _, k := range keys {
l.rev = max(l.rev, k.rev) l.rev = max(l.rev, k.rev)
} }
groups = append(groups, l) proposals = append(proposals, l)
} }
// A group whose keys' revisions and the node's inputs are those its answer stands at is not answered again. // A proposal whose keys' revisions and the node's inputs are those its answer stands at is not answered again.
for g := range s.answered { for g := range s.answered {
if _, ok := s.candidates[g]; !ok { if _, ok := s.candidates[g]; !ok {
delete(s.answered, g) delete(s.answered, g)
} }
} }
// Each group's last answer stands until it is answered again; a group whose keys went is dropped. // Each proposal's last answer stands until it is answered again; a proposal whose keys went is dropped.
next := map[string]delivery.CandidateAnswer{} next := map[string]delivery.CandidateAnswer{}
for _, l := range groups { for _, l := range proposals {
if a, ok := s.answers[l.group]; ok { if a, ok := s.answers[l.proposal]; ok {
next[l.group] = a next[l.proposal] = a
} }
} }
s.owed = false s.owed = false
s.mu.Unlock() s.mu.Unlock()
sort.Slice(groups, func(i, j int) bool { sort.Slice(proposals, func(i, j int) bool {
if groups[i].rev != groups[j].rev { if proposals[i].rev != proposals[j].rev {
return groups[i].rev > groups[j].rev return proposals[i].rev > proposals[j].rev
} }
return groups[i].group < groups[j].group return proposals[i].proposal < proposals[j].proposal
}) })
hand := func() { hand := func() {
s.mu.Lock() s.mu.Lock()
@@ -470,23 +467,23 @@ func (s *Shadow) answerAll(ctx context.Context) {
given[g] = a given[g] = a
} }
s.answer(ctx, given) s.answer(ctx, given)
c.env.Say(fmt.Sprintf("%s answered %d placement candidate group(s)", node, len(given))) c.env.Say(fmt.Sprintf("%s answered %d placement candidate proposal(s)", node, len(given)))
} }
} }
for _, l := range groups { for _, l := range proposals {
if ctx.Err() != nil { if ctx.Err() != nil {
return return
} }
group := l.group proposal := l.proposal
s.mu.Lock() s.mu.Lock()
at, done := s.answered[group] at, done := s.answered[proposal]
retry := s.retry[group] retry := s.retry[proposal]
s.mu.Unlock() s.mu.Unlock()
if done && at == l.at { if done && at == l.at {
continue continue
} }
began := time.Now() began := time.Now()
c.env.Say(fmt.Sprintf("answering placement candidate %s (revision %d; woken by %s)", group, l.rev, wokenBy(at, done, retry, l.at))) c.env.Say(fmt.Sprintf("answering placement candidate %s (revision %d; woken by %s)", proposal, l.rev, wokenBy(at, done, retry, l.at)))
stop := make(chan struct{}) stop := make(chan struct{})
go func() { go func() {
tick := time.NewTicker(progressEvery) tick := time.NewTicker(progressEvery)
@@ -496,18 +493,18 @@ func (s *Shadow) answerAll(ctx context.Context) {
case <-stop: case <-stop:
return return
case <-tick.C: case <-tick.C:
c.env.Say(fmt.Sprintf("the answer to %s has run %s and waits on %s", group, c.env.Say(fmt.Sprintf("the answer to %s has run %s and waits on %s", proposal,
time.Since(began).Round(time.Second), c.waitingOn())) time.Since(began).Round(time.Second), c.waitingOn()))
} }
} }
}() }()
missed := c.missed.Load() missed := c.missed.Load()
a, ok := s.answerGroup(ctx, group) a, ok := s.answerProposal(ctx, proposal)
close(stop) close(stop)
provisional := c.missed.Load() > missed && ctx.Err() == nil provisional := c.missed.Load() > missed && ctx.Err() == nil
took := time.Since(began) took := time.Since(began)
if took > slowAfter { if took > slowAfter {
c.env.Say(fmt.Sprintf("the answer to %s took %s", group, took.Round(time.Second))) c.env.Say(fmt.Sprintf("the answer to %s took %s", proposal, took.Round(time.Second)))
if ok { if ok {
// Words that do not move with the time it took, so an answer the same as before is not written again. // Words that do not move with the time it took, so an answer the same as before is not written again.
a.Note = strings.TrimPrefix(a.Note+"; ", "; ") + "this answer took over " + slowAfter.String() + " to compose" a.Note = strings.TrimPrefix(a.Note+"; ", "; ") + "this answer took over " + slowAfter.String() + " to compose"
@@ -515,59 +512,59 @@ func (s *Shadow) answerAll(ctx context.Context) {
} }
if ok { if ok {
a = delivery.FitAnswer(a, delivery.MaxAnswerBytes) a = delivery.FitAnswer(a, delivery.MaxAnswerBytes)
next[group] = a next[proposal] = a
c.env.Say(fmt.Sprintf("answered placement candidate %s in %s: %s", group, took.Round(time.Millisecond), a.Verdict)) c.env.Say(fmt.Sprintf("answered placement candidate %s in %s: %s", proposal, took.Round(time.Millisecond), a.Verdict))
} else { } else {
delete(next, group) delete(next, proposal)
c.env.Say(fmt.Sprintf("placement candidate %s does not place this node: nothing to answer", group)) c.env.Say(fmt.Sprintf("placement candidate %s does not place this node: nothing to answer", proposal))
} }
s.mu.Lock() s.mu.Lock()
if provisional { if provisional {
// What the bus did not answer is no answer: handed as it stands, and composed again after a bound. // What the bus did not answer is no answer: handed as it stands, and composed again after a bound.
s.owed = true s.owed = true
s.retry[group] = true s.retry[proposal] = true
delete(s.answered, group) delete(s.answered, proposal)
c.env.Say(fmt.Sprintf("the answer to %s met a read on the bus that did not answer: it is answered again", group)) c.env.Say(fmt.Sprintf("the answer to %s met a read on the bus that did not answer: it is answered again", proposal))
} else { } else {
s.answered[group] = l.at s.answered[proposal] = l.at
delete(s.retry, group) delete(s.retry, proposal)
} }
s.mu.Unlock() s.mu.Unlock()
hand() hand()
} }
hand() // a group whose keys went, when no group is left to answer hand() // a proposal whose keys went, when no proposal is left to answer
} }
// How often a running answer says what it waits on. // How often a running answer says what it waits on.
var progressEvery = time.Minute var progressEvery = time.Minute
// answerGroup is the node's answer to one group, and false when the node has nothing to answer: neither the trunk nor // answerProposal is the node's answer to one proposal, and false when the node has nothing to answer: neither the trunk nor
// the group places it. // the proposal places it.
func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.CandidateAnswer, bool) { func (s *Shadow) answerProposal(ctx context.Context, proposal string) (delivery.CandidateAnswer, bool) {
c := s.c c := s.c
node := c.env.Node node := c.env.Node
defer c.waiting("composing")() defer c.waiting("composing")()
s.mu.Lock() s.mu.Lock()
keys := s.candidates[group] keys := s.candidates[proposal]
given := make(map[string]*delivery.Placement, len(keys)) given := make(map[string]*delivery.Placement, len(keys))
var head string
for n, k := range keys { for n, k := range keys {
if k.c.Reached && k.c.Placement == nil {
continue // reached, with no key of its own: nothing laid over it
}
given[n] = k.c.Placement given[n] = k.c.Placement
head = k.c.Pull.Head
} }
// The revision answered is the node's own key's, 0 when the group has none for it (judged by the head alone). // The revision answered is the node's own candidate key's; a node with none is not asked.
own, mine := keys[node] own, mine := keys[node]
rev := own.rev rev := own.rev
s.mu.Unlock() s.mu.Unlock()
c.mu.Lock() c.mu.Lock()
_, placed := c.placements[node]
modules := make(map[string]delivery.Module, len(c.modules)) modules := make(map[string]delivery.Module, len(c.modules))
for k, v := range c.modules { for k, v := range c.modules {
modules[k] = v modules[k] = v
} }
c.mu.Unlock() c.mu.Unlock()
if !placed && !mine { if !mine {
return delivery.CandidateAnswer{}, false return delivery.CandidateAnswer{}, false
} }
@@ -575,8 +572,8 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid
if c.env.Running != nil { if c.env.Running != nil {
running = c.env.Running() running = c.env.Running()
} }
if p, ok := given[node]; ok && p == nil { if p, ok := given[node]; ok && p == nil && !own.c.Reached {
// The head deletes this node's placement file: the node then returns to the controller's compositions, as one the // 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 // 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"). // seed's revert, whose check read "would stop running node").
s.mu.Lock() s.mu.Lock()
@@ -585,7 +582,7 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid
wants[m] = w wants[m] = w
} }
s.mu.Unlock() s.mu.Unlock()
return returning(ctx, c.get, node, wants, running, head, rev), true return returning(ctx, c.get, node, wants, running, proposal, rev), true
} }
placement := given[node] placement := given[node]
@@ -617,7 +614,7 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid
c.wants = held c.wants = held
c.mu.Unlock() c.mu.Unlock()
a := frombus.Answer(ctx, reader{c: c}, c.env.Mesh, frombus.Scope{Node: node, KeptAt: kept, Placements: given, InOwn: inOwn, a := frombus.Answer(ctx, reader{c: c}, c.env.Mesh, frombus.Scope{Node: node, KeptAt: kept, Placements: given, InOwn: inOwn,
SameValue: c.sameValue}, fps, running, head, rev) SameValue: c.sameValue}, fps, running, proposal, rev)
s.mu.Lock() s.mu.Lock()
ran := make(map[string]string, len(s.ran)) ran := make(map[string]string, len(s.ran))
for m, d := range s.ran { for m, d := range s.ran {
@@ -727,13 +724,13 @@ func (s *Shadow) takeWant(u agent.Update) bool {
return !had || old.FP != w.FP return !had || old.FP != w.FP
} }
// returning is the node's answer to a head deleting its placement: it returns to the controller's compositions, each // 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 // 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 // (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. // 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, 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, head string, rev uint64) delivery.CandidateAnswer { wants map[string]delivery.Want, running []delivery.Composed, proposal string, rev uint64) delivery.CandidateAnswer {
a := delivery.CandidateAnswer{Head: head, Revision: rev} a := delivery.CandidateAnswer{Proposal: proposal, Revision: rev}
bucket, err := delivery.ComposedBucketFor(node) bucket, err := delivery.ComposedBucketFor(node)
if err != nil { if err != nil {
a.Refused, a.Verdict = err.Error(), delivery.CandidateRefused a.Refused, a.Verdict = err.Error(), delivery.CandidateRefused