Node agent: follow the placement key at start only and say why when it cannot; test the reached candidate and the switched rule; word the log by the seat's state (issue 1126)
This commit is contained in:
@@ -483,7 +483,7 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
|
||||
if !placed {
|
||||
// The catalogue no longer places it: the agent returns to the controller's compositions (endWhenUnplaced), and
|
||||
// this is no refusal of what it runs (issue 1126, the seed's revert).
|
||||
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
|
||||
return nil
|
||||
}
|
||||
@@ -1018,11 +1018,11 @@ func Placed(ctx context.Context, get func(ctx context.Context, bucket, key strin
|
||||
return false, ctx.Err()
|
||||
}
|
||||
if !switched {
|
||||
say(fmt.Sprintf("whether the catalogue places this node 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))
|
||||
say(fmt.Sprintf("this node's key in the seat's placement state cannot be read (%s): it has never composed its own, so it applies the controller's compositions and switches when its key is read: %v", key, err))
|
||||
return false, nil
|
||||
}
|
||||
if why := err.Error(); why != said {
|
||||
say(fmt.Sprintf("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
|
||||
}
|
||||
select {
|
||||
@@ -1033,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.
|
||||
var composeAgain = []time.Duration{time.Second, 2 * time.Second, 4 * time.Second}
|
||||
|
||||
|
||||
@@ -781,6 +781,15 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
k, _ := delivery.PlacementCandidateKey("deletes", "n1")
|
||||
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()
|
||||
var answers map[string]delivery.CandidateAnswer
|
||||
waitFor := func(why string, ok func(map[string]delivery.CandidateAnswer) bool) {
|
||||
@@ -794,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).
|
||||
sb.mu.Lock()
|
||||
for _, w := range sb.watches {
|
||||
@@ -824,6 +833,10 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
"ghost": func(a delivery.CandidateAnswer) bool {
|
||||
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 {
|
||||
return a.Verdict == delivery.CandidateSame && a.Returns == delivery.ReturnsToController && a.Revision == 7 &&
|
||||
len(a.Removed) == 0 && fmt.Sprint(a.Same) == "[lemurs node]"
|
||||
@@ -933,7 +946,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
}
|
||||
k, _ := delivery.PlacementCandidateKey("same", "n1")
|
||||
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
|
||||
// again. The new run reads afresh, so that group is no longer answered.
|
||||
@@ -945,7 +958,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
sb.missing(delivery.PlacementCandidatesBucket, false)
|
||||
waitFor("a group deleted while the shadow was down is still answered", func(a map[string]delivery.CandidateAnswer) bool {
|
||||
_, 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
|
||||
@@ -987,6 +1000,25 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
// a refused answer to each group known; once read, the group is answered for real.
|
||||
setFor(t, ¬ReadyAfter, 30*time.Millisecond)
|
||||
|
||||
Reference in New Issue
Block a user