Node agent: answer a proposal's candidate keys and follow the seat's placement, not the indexer's groups (issue 1126)
The answer names its proposal and the revision of the node's own candidate key; a node with no key in a proposal is not asked. The watch for its own placement key is tried again until the seat's state exists, so the node switches when its key is written.
This commit is contained in:
@@ -937,7 +937,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}
|
||||
}
|
||||
|
||||
// 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
|
||||
// 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.
|
||||
@@ -969,7 +969,7 @@ func endWhenUnplaced(ctx context.Context, session *link.Session, node string, st
|
||||
}
|
||||
}
|
||||
|
||||
// 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;
|
||||
// the node then switches at its next start.
|
||||
func endWhenPlaced(ctx context.Context, session *link.Session, node string, stop func(), say link.Announce) {
|
||||
@@ -977,10 +977,25 @@ func endWhenPlaced(ctx context.Context, session *link.Session, node string, stop
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
updates, err := bucketWatch{session, delivery.PlacementBucket}.Watch(ctx, key)
|
||||
if err != nil {
|
||||
say("whether the catalogue places this node is not followed (" + err.Error() + "): it composes its own from its next start")
|
||||
return
|
||||
// The seat's placement state appears when its holder first writes it (ADR 0567): until then the watch is refused, as a
|
||||
// read of it finds no bucket (Placed), and the node stays on the controller's compositions. The watch is tried again,
|
||||
// so the node switches when its key is written, not only at its next start.
|
||||
var updates <-chan agent.Update
|
||||
said := ""
|
||||
for try := 1; ; try++ {
|
||||
updates, err = bucketWatch{session, delivery.PlacementBucket}.Watch(ctx, key)
|
||||
if err == nil {
|
||||
break
|
||||
}
|
||||
if why := err.Error(); why != said {
|
||||
say("whether the catalogue places this node is not followed yet (" + why + "): the node stays on the controller's compositions and the watch is tried again")
|
||||
said = why
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(link.Backoff(try)):
|
||||
}
|
||||
}
|
||||
for u := range updates {
|
||||
if !u.Caught && !u.Deleted && len(u.Value) > 0 {
|
||||
|
||||
@@ -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,
|
||||
// issue 1126): kept in its key so a restart reuses them and applies nothing again.
|
||||
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
|
||||
// nodeComposes and nodeComposesWhy are whose compositions the node applies and why (HaveNode.Composes), written into
|
||||
// its key.
|
||||
@@ -733,8 +733,8 @@ func (q *Session) NoteNode(ctx context.Context, composed string, sealed delivery
|
||||
q.writeHaves(ctx, bus)
|
||||
}
|
||||
|
||||
// NoteCandidates takes the node's answers to the placement candidates, by group (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.
|
||||
// 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 proposal left out is no longer answered.
|
||||
func (q *Session) NoteCandidates(ctx context.Context, answers map[string]delivery.CandidateAnswer) {
|
||||
q.init()
|
||||
s := &q.haves
|
||||
|
||||
@@ -590,9 +590,9 @@ func TestAnAppliedModulesHaveNamesTheFingerprintAndTheDigestAndFollowsItsHealth(
|
||||
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.
|
||||
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)
|
||||
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)
|
||||
}
|
||||
q.NoteCandidates(context.Background(), nil)
|
||||
|
||||
@@ -89,12 +89,12 @@ func New(env Env) *Resync {
|
||||
func (r *Resync) Take(ctx context.Context, answers map[string]delivery.CandidateAnswer) []Outcome {
|
||||
seen := map[delivery.OwnValueDiffers]bool{}
|
||||
var drift []delivery.OwnValueDiffers
|
||||
groups := make([]string, 0, len(answers))
|
||||
proposals := make([]string, 0, len(answers))
|
||||
for g := range answers {
|
||||
groups = append(groups, g)
|
||||
proposals = append(proposals, g)
|
||||
}
|
||||
sort.Strings(groups)
|
||||
for _, g := range groups {
|
||||
sort.Strings(proposals)
|
||||
for _, g := range proposals {
|
||||
for _, o := range delivery.OwnValuesDiffer(answers[g]) {
|
||||
if !seen[o] {
|
||||
seen[o] = true
|
||||
|
||||
@@ -20,7 +20,7 @@ func TestTheNodeHandsTheVaultWhatRunsOfADriftedOwnSecret(t *testing.T) {
|
||||
digest := "sha256:" + strings.Repeat("a", 64)
|
||||
fp := "src2:" + strings.Repeat("b", 64)
|
||||
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{
|
||||
{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>"}`}}}}}}
|
||||
|
||||
@@ -1003,7 +1003,7 @@ func Placed(ctx context.Context, get func(ctx context.Context, bucket, key strin
|
||||
for try := 1; ; try++ {
|
||||
raw, err := get(ctx, delivery.PlacementBucket, key)
|
||||
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 {
|
||||
return len(raw) > 0, nil
|
||||
|
||||
@@ -762,8 +762,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
for _, m := range modules {
|
||||
p.Modules = append(p.Modules, delivery.PlacedModule{Module: m})
|
||||
}
|
||||
raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: group,
|
||||
Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h-" + group}, Placement: &p})
|
||||
raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: group, Placement: &p})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -776,8 +775,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
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
|
||||
// compositions, answered as such, not as every module removed.
|
||||
if raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "deletes",
|
||||
Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h-deletes"}}); err != nil {
|
||||
if raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "deletes"}); err != nil {
|
||||
t.Fatal(err)
|
||||
} else {
|
||||
k, _ := delivery.PlacementCandidateKey("deletes", "n1")
|
||||
@@ -814,7 +812,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
readOnly.mu.Unlock()
|
||||
for group, want := range map[string]func(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 {
|
||||
return a.Verdict == delivery.CandidateDiffers && len(a.Added) == 1 && a.Added[0].Module == "keeper" &&
|
||||
@@ -1008,13 +1006,11 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
raw, _ := delivery.EncodeModule(m)
|
||||
nb.put(delivery.ModulesBucket, name, raw, 1)
|
||||
}
|
||||
sameRaw, _ := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "same",
|
||||
Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h-same"}, Placement: &trunk})
|
||||
sameRaw, _ := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "same", Placement: &trunk})
|
||||
sameKey, _ := delivery.PlacementCandidateKey("same", "n1")
|
||||
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.
|
||||
oldRaw, _ := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "older",
|
||||
Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 2, Head: "h-old"}, Placement: &trunk})
|
||||
oldRaw, _ := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "older", Placement: &trunk})
|
||||
oldKey, _ := delivery.PlacementCandidateKey("older", "n1")
|
||||
nb.put(delivery.PlacementCandidatesBucket, oldKey, oldRaw, 2)
|
||||
for _, bucket := range []string{delivery.PlacementCandidatesBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket} {
|
||||
@@ -1022,7 +1018,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
}
|
||||
select {
|
||||
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)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
@@ -1072,7 +1068,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
|
||||
// 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{
|
||||
{Module: "xterm", Composed: "sha256:3c0a31df64283c0678b1cdbeff8ac030609cab65cc7e47faaf714da743d9340e", Why: "the fingerprint differs"},
|
||||
{Module: "zsh", Composed: "sha256:6093f31056d39eaf604410b7289ec6b5fe602b19c8547283562fcfc078c4844f", Resource: "zsh.bundle-tools", Why: "the resource differs"},
|
||||
@@ -1085,7 +1081,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 {
|
||||
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 {
|
||||
t.Errorf("a module the same as its have says is answered %+v", one)
|
||||
}
|
||||
@@ -1121,11 +1117,11 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
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, "h", 1)
|
||||
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{"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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -16,8 +16,8 @@ import (
|
||||
"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
|
||||
// the indexer's state `placement-candidates`, it composes this node in shadow from the trunk's placement with the group's
|
||||
// Shadow is the node's answer to placement candidates (novox/hq ADR 0559 §4, issue 1126): for every proposal with a key in
|
||||
// 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
|
||||
// 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.
|
||||
@@ -25,37 +25,37 @@ 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 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.
|
||||
type Shadow struct {
|
||||
c *Composer
|
||||
answer func(ctx context.Context, answers map[string]delivery.CandidateAnswer)
|
||||
|
||||
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
|
||||
// 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.
|
||||
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).
|
||||
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
|
||||
// every 13-17 s, woken by its own modules' haves whose health words changed at every look).
|
||||
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,
|
||||
// 2026-10-28, seed-9's probe: a safety check could not tell a loop from inputs moving).
|
||||
moved [len(inputWords)]uint64
|
||||
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
|
||||
// yet ... no responders" in its answer to seed-7).
|
||||
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 {
|
||||
revs string
|
||||
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.
|
||||
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.
|
||||
func wokenBy(had answeredAt, done, retry bool, at answeredAt) string {
|
||||
switch {
|
||||
@@ -157,7 +157,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 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},
|
||||
}
|
||||
type tagged struct {
|
||||
@@ -187,7 +187,7 @@ func (s *Shadow) Run(ctx context.Context) error {
|
||||
}
|
||||
caught := make([]bool, len(ins))
|
||||
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).
|
||||
waiting := time.NewTimer(notReadyAfter)
|
||||
defer waiting.Stop()
|
||||
@@ -306,7 +306,7 @@ func (s *Shadow) entry(bucket, key string) (v any, has, known bool) {
|
||||
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 {
|
||||
nodes := make([]string, 0, len(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.
|
||||
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 {
|
||||
return false
|
||||
}
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if u.Deleted {
|
||||
if _, ok := s.candidates[group][node]; !ok {
|
||||
if _, ok := s.candidates[proposal][node]; !ok {
|
||||
return false
|
||||
}
|
||||
delete(s.candidates[group], node)
|
||||
if len(s.candidates[group]) == 0 {
|
||||
delete(s.candidates, group)
|
||||
delete(s.candidates[proposal], node)
|
||||
if len(s.candidates[proposal]) == 0 {
|
||||
delete(s.candidates, proposal)
|
||||
}
|
||||
return true
|
||||
}
|
||||
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.
|
||||
if _, ok := s.candidates[group][node]; ok {
|
||||
delete(s.candidates[group], node)
|
||||
if _, ok := s.candidates[proposal][node]; ok {
|
||||
delete(s.candidates[proposal], node)
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
if s.candidates[group] == nil {
|
||||
s.candidates[group] = map[string]candidateKey{}
|
||||
if s.candidates[proposal] == nil {
|
||||
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
|
||||
}
|
||||
|
||||
@@ -372,10 +372,10 @@ func (s *Shadow) takeRecord(u agent.Update) bool {
|
||||
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
|
||||
|
||||
// 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) {
|
||||
var waits []string
|
||||
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.mu.Lock()
|
||||
next := map[string]delivery.CandidateAnswer{}
|
||||
for group, keys := range s.candidates {
|
||||
a := delivery.CandidateAnswer{Refused: why, Verdict: delivery.CandidateRefused}
|
||||
for n, k := range keys {
|
||||
a.Head = k.c.Pull.Head
|
||||
if n == s.c.env.Node {
|
||||
a.Revision = k.rev
|
||||
}
|
||||
for proposal, keys := range s.candidates {
|
||||
own, ok := keys[s.c.env.Node]
|
||||
if !ok {
|
||||
continue // no key of this node's: nothing asks it to answer
|
||||
}
|
||||
next[group] = a
|
||||
next[proposal] = delivery.CandidateAnswer{Proposal: proposal, Revision: own.rev, Refused: why, Verdict: delivery.CandidateRefused}
|
||||
}
|
||||
s.answers = next
|
||||
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
|
||||
// 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
|
||||
// starts, every progressEvery while it runs what it waits on, and when it ends (T2: the node said nothing for 22
|
||||
// minutes).
|
||||
@@ -415,38 +412,38 @@ func (s *Shadow) answerAll(ctx context.Context) {
|
||||
node := c.env.Node
|
||||
s.mu.Lock()
|
||||
type listed struct {
|
||||
group string
|
||||
rev uint64
|
||||
at answeredAt
|
||||
proposal string
|
||||
rev uint64
|
||||
at answeredAt
|
||||
}
|
||||
var groups []listed
|
||||
var proposals []listed
|
||||
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 {
|
||||
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 {
|
||||
if _, ok := s.candidates[g]; !ok {
|
||||
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{}
|
||||
for _, l := range groups {
|
||||
if a, ok := s.answers[l.group]; ok {
|
||||
next[l.group] = a
|
||||
for _, l := range proposals {
|
||||
if a, ok := s.answers[l.proposal]; ok {
|
||||
next[l.proposal] = a
|
||||
}
|
||||
}
|
||||
s.owed = false
|
||||
s.mu.Unlock()
|
||||
sort.Slice(groups, func(i, j int) bool {
|
||||
if groups[i].rev != groups[j].rev {
|
||||
return groups[i].rev > groups[j].rev
|
||||
sort.Slice(proposals, func(i, j int) bool {
|
||||
if proposals[i].rev != proposals[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() {
|
||||
s.mu.Lock()
|
||||
@@ -470,23 +467,23 @@ func (s *Shadow) answerAll(ctx context.Context) {
|
||||
given[g] = a
|
||||
}
|
||||
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 {
|
||||
return
|
||||
}
|
||||
group := l.group
|
||||
proposal := l.proposal
|
||||
s.mu.Lock()
|
||||
at, done := s.answered[group]
|
||||
retry := s.retry[group]
|
||||
at, done := s.answered[proposal]
|
||||
retry := s.retry[proposal]
|
||||
s.mu.Unlock()
|
||||
if done && at == l.at {
|
||||
continue
|
||||
}
|
||||
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{})
|
||||
go func() {
|
||||
tick := time.NewTicker(progressEvery)
|
||||
@@ -496,18 +493,18 @@ func (s *Shadow) answerAll(ctx context.Context) {
|
||||
case <-stop:
|
||||
return
|
||||
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()))
|
||||
}
|
||||
}
|
||||
}()
|
||||
missed := c.missed.Load()
|
||||
a, ok := s.answerGroup(ctx, group)
|
||||
a, ok := s.answerProposal(ctx, proposal)
|
||||
close(stop)
|
||||
provisional := c.missed.Load() > missed && ctx.Err() == nil
|
||||
took := time.Since(began)
|
||||
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 {
|
||||
// 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"
|
||||
@@ -515,59 +512,59 @@ func (s *Shadow) answerAll(ctx context.Context) {
|
||||
}
|
||||
if ok {
|
||||
a = delivery.FitAnswer(a, delivery.MaxAnswerBytes)
|
||||
next[group] = a
|
||||
c.env.Say(fmt.Sprintf("answered placement candidate %s in %s: %s", group, took.Round(time.Millisecond), a.Verdict))
|
||||
next[proposal] = a
|
||||
c.env.Say(fmt.Sprintf("answered placement candidate %s in %s: %s", proposal, took.Round(time.Millisecond), a.Verdict))
|
||||
} else {
|
||||
delete(next, group)
|
||||
c.env.Say(fmt.Sprintf("placement candidate %s does not place this node: nothing to answer", group))
|
||||
delete(next, proposal)
|
||||
c.env.Say(fmt.Sprintf("placement candidate %s does not place this node: nothing to answer", proposal))
|
||||
}
|
||||
s.mu.Lock()
|
||||
if provisional {
|
||||
// What the bus did not answer is no answer: handed as it stands, and composed again after a bound.
|
||||
s.owed = true
|
||||
s.retry[group] = true
|
||||
delete(s.answered, group)
|
||||
c.env.Say(fmt.Sprintf("the answer to %s met a read on the bus that did not answer: it is answered again", group))
|
||||
s.retry[proposal] = true
|
||||
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", proposal))
|
||||
} else {
|
||||
s.answered[group] = l.at
|
||||
delete(s.retry, group)
|
||||
s.answered[proposal] = l.at
|
||||
delete(s.retry, proposal)
|
||||
}
|
||||
s.mu.Unlock()
|
||||
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.
|
||||
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
|
||||
// the group places it.
|
||||
func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.CandidateAnswer, bool) {
|
||||
// answerProposal is the node's answer to one proposal, and false when the node has nothing to answer: neither the trunk nor
|
||||
// the proposal places it.
|
||||
func (s *Shadow) answerProposal(ctx context.Context, proposal string) (delivery.CandidateAnswer, bool) {
|
||||
c := s.c
|
||||
node := c.env.Node
|
||||
defer c.waiting("composing")()
|
||||
s.mu.Lock()
|
||||
keys := s.candidates[group]
|
||||
keys := s.candidates[proposal]
|
||||
given := make(map[string]*delivery.Placement, len(keys))
|
||||
var head string
|
||||
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
|
||||
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]
|
||||
rev := own.rev
|
||||
s.mu.Unlock()
|
||||
|
||||
c.mu.Lock()
|
||||
_, placed := c.placements[node]
|
||||
modules := make(map[string]delivery.Module, len(c.modules))
|
||||
for k, v := range c.modules {
|
||||
modules[k] = v
|
||||
}
|
||||
c.mu.Unlock()
|
||||
if !placed && !mine {
|
||||
if !mine {
|
||||
return delivery.CandidateAnswer{}, false
|
||||
}
|
||||
|
||||
@@ -575,8 +572,8 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid
|
||||
if c.env.Running != nil {
|
||||
running = c.env.Running()
|
||||
}
|
||||
if p, ok := given[node]; ok && p == nil {
|
||||
// The head deletes this node's placement file: the node then returns to the controller's compositions, as one the
|
||||
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()
|
||||
@@ -585,7 +582,7 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid
|
||||
wants[m] = w
|
||||
}
|
||||
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]
|
||||
@@ -617,7 +614,7 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid
|
||||
c.wants = held
|
||||
c.mu.Unlock()
|
||||
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()
|
||||
ran := make(map[string]string, len(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
|
||||
}
|
||||
|
||||
// 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
|
||||
// (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, head string, rev uint64) delivery.CandidateAnswer {
|
||||
a := delivery.CandidateAnswer{Head: head, Revision: rev}
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user