A failing placement read chooses neither composer, a deleting head is compared with the controller's compositions at the node's wants, and a placement watch that ends is said; pin mesh-sdk e32d777 (issue 1126, review of the shadow, round 1)
selfComposer took a KVHistory error for 'not placed': at any restart a timeout or a bus not ready flipped a self-composing node to the controller's compositions for a whole run. selfcompose.Placed chooses only on a read that succeeds (a missing placement state is one), and tries again otherwise, said once. The shadow's answer to a head deleting the placement reads the controller's composition at each want in the node's bucket and compares it with what runs; a want with none refuses, without returning.
This commit is contained in:
@@ -544,7 +544,14 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
|
|||||||
session.NoteCandidates(ctx, answers)
|
session.NoteCandidates(ctx, answers)
|
||||||
})
|
})
|
||||||
go shadow.Keep(linked, link.Backoff)
|
go shadow.Keep(linked, link.Backoff)
|
||||||
if composer, why := selfComposer(linked, session, node, say, hooks); composer != nil {
|
composer, why := selfComposer(linked, session, node, say, hooks)
|
||||||
|
if linked.Err() != nil {
|
||||||
|
// The link ended before the node knew whose compositions to apply: it chose neither, and waits for the next.
|
||||||
|
stopRun()
|
||||||
|
cancel()
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if composer != nil {
|
||||||
env.Composed, env.SelfComposed, env.Trusted, env.Policy = composer, true, nil, composer.Policy
|
env.Composed, env.SelfComposed, env.Trusted, env.Policy = composer, true, nil, composer.Policy
|
||||||
session.NoteComposes(linked, delivery.ComposesOwn, why)
|
session.NoteComposes(linked, delivery.ComposesOwn, why)
|
||||||
// And once the catalogue no longer places it, the agent begins again on the controller's compositions: the
|
// And once the catalogue no longer places it, the agent begins again on the controller's compositions: the
|
||||||
@@ -854,18 +861,19 @@ func selfComposer(ctx context.Context, session *link.Session, node string, say f
|
|||||||
say("this node applies the controller's compositions: " + err.Error())
|
say("this node applies the controller's compositions: " + err.Error())
|
||||||
return nil, err.Error()
|
return nil, err.Error()
|
||||||
}
|
}
|
||||||
got, err := session.KVHistory(ctx, delivery.PlacementBucket, key)
|
env := composerEnv(session, node, say, hooks)
|
||||||
if err != nil || len(got) == 0 || got[len(got)-1].Deleted {
|
placed, err := selfcompose.Placed(ctx, env.Get, node, link.Backoff, say)
|
||||||
|
if err != nil {
|
||||||
|
return nil, "its placement could not be read before the link ended: " + err.Error()
|
||||||
|
}
|
||||||
|
if !placed {
|
||||||
why := "the catalogue does not place it"
|
why := "the catalogue does not place it"
|
||||||
if err != nil {
|
|
||||||
why = "its placement cannot be read: " + err.Error()
|
|
||||||
}
|
|
||||||
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 catalogue places it (" + key + ")"
|
||||||
say("this node composes its own modules: " + why)
|
say("this node composes its own modules: " + why)
|
||||||
return selfcompose.New(composerEnv(session, node, say, hooks)), why
|
return selfcompose.New(env), why
|
||||||
}
|
}
|
||||||
|
|
||||||
// composerEnv is what the node's composers read of the bus, the vault and the agent: its own composer and its answer
|
// composerEnv is what the node's composers read of the bus, the vault and the agent: its own composer and its answer
|
||||||
@@ -922,6 +930,9 @@ func endWhenUnplaced(ctx context.Context, session *link.Session, node string, st
|
|||||||
placed = true
|
placed = true
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
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")
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// 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 indexer's placement (a value after the watch
|
||||||
@@ -944,4 +955,7 @@ func endWhenPlaced(ctx context.Context, session *link.Session, node string, stop
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
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")
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -406,8 +406,8 @@ func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) {
|
|||||||
h8back(t, b, placement, candidates, hkey)
|
h8back(t, b, placement, candidates, hkey)
|
||||||
}
|
}
|
||||||
|
|
||||||
// h8back is the way back (issue 1126, the seed's revert): a candidate deleting the node's placement is answered as
|
// h8back is the way back (issue 1126, the seed's revert): a candidate deleting the node's placement is compared with
|
||||||
// returning to the controller's compositions, and the placement key deleted returns the agent to them without a
|
// the controller's compositions at its wants (here none: refused, naming web), and the placement key deleted returns the agent to them without a
|
||||||
// restart, said in the journal and the node's have, keeping what runs.
|
// restart, said in the journal and the node's have, keeping what runs.
|
||||||
func h8back(t *testing.T, b *itBed, placement, candidates nats.KeyValue, hkey string) {
|
func h8back(t *testing.T, b *itBed, placement, candidates nats.KeyValue, hkey string) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
@@ -431,10 +431,12 @@ func h8back(t *testing.T, b *itBed, placement, candidates nats.KeyValue, hkey st
|
|||||||
if _, err := candidates.Put(ckey, craw); err != nil {
|
if _, err := candidates.Put(ckey, craw); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
within(t, 10*time.Second, "the node did not answer a candidate deleting its placement as returning", func() bool {
|
// 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).
|
||||||
|
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 == delivery.ReturnsToController && a.Verdict == delivery.CandidateSame &&
|
return a.Head == "h2" && a.Returns == "" && a.Verdict == delivery.CandidateRefused &&
|
||||||
len(a.Removed) == 0
|
strings.Contains(a.Refused, "web: the controller holds no composition")
|
||||||
})
|
})
|
||||||
ran := b.have("web").Composed
|
ran := b.have("web").Composed
|
||||||
pkey, _ := delivery.PlacementKey(itNode)
|
pkey, _ := delivery.PlacementKey(itNode)
|
||||||
|
|||||||
@@ -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.20261027222254-472d5f314466
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027223231-e32d77753024
|
||||||
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
|
||||||
|
|||||||
@@ -22,6 +22,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027221705-2fd5ffdd6263 h1:JZV8C+WR
|
|||||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027221705-2fd5ffdd6263/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027221705-2fd5ffdd6263/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027222254-472d5f314466 h1:xzbkI9dOkN/Iv78AS/fOwhjkgBEQOYtpbHzNYwb4eu0=
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027222254-472d5f314466 h1:xzbkI9dOkN/Iv78AS/fOwhjkgBEQOYtpbHzNYwb4eu0=
|
||||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027222254-472d5f314466/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027222254-472d5f314466/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||||
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027223231-e32d77753024 h1:MnmyspVoaCldYmm4Ut1O39RDR9WpiL4TfFgDEnY+QRg=
|
||||||
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027223231-e32d77753024/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=
|
||||||
|
|||||||
@@ -16,6 +16,7 @@ import (
|
|||||||
"crypto/sha256"
|
"crypto/sha256"
|
||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"sort"
|
"sort"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -27,6 +28,8 @@ import (
|
|||||||
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
|
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
|
|
||||||
|
"github.com/nats-io/nats.go"
|
||||||
|
|
||||||
"github.com/novox/mesh-host/internal/agent"
|
"github.com/novox/mesh-host/internal/agent"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -866,3 +869,38 @@ func (c *Composer) keeps(id string, generation int, key string) bool {
|
|||||||
kept, ok := c.sealedOwn[id]
|
kept, ok := c.sealedOwn[id]
|
||||||
return ok && kept.generation == generation && kept.key == key
|
return ok && kept.generation == generation && kept.key == key
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Placed reads whether the catalogue places node now, for the node agent choosing whose compositions it applies at
|
||||||
|
// each start: true when its placement key holds a value, false only when a read succeeds and finds no key or a delete
|
||||||
|
// marker. A read that fails chooses neither (review of the shadow, round 1: a timeout or a bus not ready at a restart
|
||||||
|
// flipped a self-composing node to the controller's compositions for a whole run): it is tried again after wait(try),
|
||||||
|
// each distinct failure said once, until ctx ends, which is then its error.
|
||||||
|
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) {
|
||||||
|
key, err := delivery.PlacementKey(node)
|
||||||
|
if err != nil {
|
||||||
|
return false, err
|
||||||
|
}
|
||||||
|
said := ""
|
||||||
|
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
|
||||||
|
}
|
||||||
|
if err == nil {
|
||||||
|
return len(raw) > 0, nil
|
||||||
|
}
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
return false, ctx.Err()
|
||||||
|
}
|
||||||
|
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))
|
||||||
|
said = why
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return false, ctx.Err()
|
||||||
|
case <-time.After(wait(try)):
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -16,6 +16,7 @@ import (
|
|||||||
"git.novox.be/novox/mesh-sdk/go/compose"
|
"git.novox.be/novox/mesh-sdk/go/compose"
|
||||||
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
|
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
|
"github.com/nats-io/nats.go"
|
||||||
|
|
||||||
"github.com/novox/mesh-host/internal/agent"
|
"github.com/novox/mesh-host/internal/agent"
|
||||||
)
|
)
|
||||||
@@ -440,6 +441,18 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
|||||||
}
|
}
|
||||||
return out
|
return out
|
||||||
}
|
}
|
||||||
|
// The controller's compositions at n1's wants, in n1's composition bucket: what a head deleting its placement
|
||||||
|
// returns it to (the seed's revert); here the ones it runs.
|
||||||
|
composedBucket, _ := delivery.ComposedBucketFor("n1")
|
||||||
|
for key, c := range running {
|
||||||
|
raw, err := delivery.EncodeComposed(c)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
readOnly.mu.Lock()
|
||||||
|
readOnly.kv[composedBucket+"/"+key] = raw
|
||||||
|
readOnly.mu.Unlock()
|
||||||
|
}
|
||||||
answered := make(chan map[string]delivery.CandidateAnswer, 8)
|
answered := make(chan map[string]delivery.CandidateAnswer, 8)
|
||||||
sh := NewShadow(senv, func(_ context.Context, a map[string]delivery.CandidateAnswer) { answered <- a })
|
sh := NewShadow(senv, func(_ context.Context, a map[string]delivery.CandidateAnswer) { answered <- a })
|
||||||
sctx, scancel := context.WithCancel(ctx)
|
sctx, scancel := context.WithCancel(ctx)
|
||||||
@@ -450,7 +463,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
|||||||
go sh.Keep(sctx, func(int) time.Duration { return 5 * time.Millisecond })
|
go sh.Keep(sctx, func(int) time.Duration { return 5 * time.Millisecond })
|
||||||
time.Sleep(30 * time.Millisecond)
|
time.Sleep(30 * time.Millisecond)
|
||||||
sb.missing(delivery.PlacementCandidatesBucket, false)
|
sb.missing(delivery.PlacementCandidatesBucket, false)
|
||||||
sb.made(6)
|
sb.made(7)
|
||||||
rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
||||||
tp, _ := delivery.EncodePlacement(trunk)
|
tp, _ := delivery.EncodePlacement(trunk)
|
||||||
sb.put(delivery.PlacementBucket, "placement.n1", tp, 1)
|
sb.put(delivery.PlacementBucket, "placement.n1", tp, 1)
|
||||||
@@ -459,6 +472,21 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
|||||||
raw, _ := delivery.EncodeModule(m)
|
raw, _ := delivery.EncodeModule(m)
|
||||||
sb.put(delivery.ModulesBucket, name, raw, 1)
|
sb.put(delivery.ModulesBucket, name, raw, 1)
|
||||||
}
|
}
|
||||||
|
wantLemurs, _ := delivery.WantKey("n1", "lemurs")
|
||||||
|
var lemursDigest string
|
||||||
|
for _, c := range running {
|
||||||
|
if c.Module == "lemurs" {
|
||||||
|
lemursDigest, _ = delivery.ComposedDigestOf(c)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
wantRaw, err := delivery.EncodeWant(delivery.Want{FP: fp, Composed: lemursDigest})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
sb.put(delivery.AssignerWantsBucket, wantLemurs, wantRaw, 1)
|
||||||
|
wantNode, _ := delivery.WantNodeKey("n1")
|
||||||
|
nodeRaw, _ := delivery.EncodeWant(delivery.Want{})
|
||||||
|
sb.put(delivery.AssignerWantsBucket, wantNode, nodeRaw, 1)
|
||||||
candidate := func(group string, rev uint64, modules ...string) {
|
candidate := func(group string, rev uint64, modules ...string) {
|
||||||
p := trunk
|
p := trunk
|
||||||
p.Modules = nil
|
p.Modules = nil
|
||||||
@@ -531,7 +559,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
|||||||
},
|
},
|
||||||
"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
|
len(a.Removed) == 0 && fmt.Sprint(a.Same) == "[lemurs node]"
|
||||||
},
|
},
|
||||||
} {
|
} {
|
||||||
if !want(answers[group]) {
|
if !want(answers[group]) {
|
||||||
@@ -562,6 +590,81 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
|||||||
return !ok && len(a) == 3
|
return !ok && len(a) == 3
|
||||||
})
|
})
|
||||||
|
|
||||||
|
// Whose compositions a node applies at a start (review of the shadow, round 1): only a read that succeeds chooses; a
|
||||||
|
// failing read is tried again, said once, and chooses neither, even when the link ends.
|
||||||
|
for name, row := range map[string]struct {
|
||||||
|
answers []error // per try; nil then the value
|
||||||
|
value []byte
|
||||||
|
end bool
|
||||||
|
placed bool
|
||||||
|
tries int
|
||||||
|
quiet bool // read and known: nothing to say
|
||||||
|
}{
|
||||||
|
"placed after two failures": {answers: []error{fmt.Errorf("nats: timeout"), fmt.Errorf("nats: timeout")}, value: []byte(`{}`), placed: true, tries: 3},
|
||||||
|
"no key": {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},
|
||||||
|
"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},
|
||||||
|
} {
|
||||||
|
lctx, lcancel := context.WithCancel(ctx)
|
||||||
|
tries := 0
|
||||||
|
var said []string
|
||||||
|
get := func(context.Context, string, string) ([]byte, error) {
|
||||||
|
tries++
|
||||||
|
if tries <= len(row.answers) {
|
||||||
|
if row.end && tries == len(row.answers) {
|
||||||
|
lcancel()
|
||||||
|
}
|
||||||
|
return nil, row.answers[tries-1]
|
||||||
|
}
|
||||||
|
return row.value, nil
|
||||||
|
}
|
||||||
|
placed, err := Placed(lctx, get, "n1", func(int) time.Duration { return time.Millisecond }, func(s string) { said = append(said, s) })
|
||||||
|
lcancel()
|
||||||
|
if placed != row.placed || tries != row.tries || (err != nil) != row.end || (len(row.answers) > 0 && !row.quiet) != (len(said) == 1) {
|
||||||
|
t.Errorf("%s: placed %v, err %v after %d reads, said %q", name, placed, err, tries, said)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// The way back, row by row (review of the shadow, round 1): the controller's compositions at n1's wants compared
|
||||||
|
// with what runs; one differing is answered so; one the controller holds none of refuses, without returning.
|
||||||
|
var runsList []delivery.Composed
|
||||||
|
for _, c := range running {
|
||||||
|
runsList = append(runsList, c)
|
||||||
|
}
|
||||||
|
lemursKey, _ = delivery.ComposedKey("n1", "lemurs", fp)
|
||||||
|
changed := running[lemursKey]
|
||||||
|
changed.Resources = []json.RawMessage{json.RawMessage(`{"id":"lemurs.service","type":"service","module":"lemurs","unit":"lemurs.service","state":"stopped","boot":"disabled"}`)}
|
||||||
|
changedRaw, _ := delivery.EncodeComposed(changed)
|
||||||
|
for name, row := range map[string]struct {
|
||||||
|
lemurs []byte
|
||||||
|
check func(delivery.CandidateAnswer) bool
|
||||||
|
}{
|
||||||
|
"the controller's differs": {changedRaw, func(a delivery.CandidateAnswer) bool {
|
||||||
|
return a.Verdict == delivery.CandidateDiffers && a.Returns == delivery.ReturnsToController &&
|
||||||
|
len(a.Differs) == 1 && a.Differs[0].Module == "lemurs" && fmt.Sprint(a.Same) == "[node]"
|
||||||
|
}},
|
||||||
|
"the controller holds none": {nil, func(a delivery.CandidateAnswer) bool {
|
||||||
|
return a.Verdict == delivery.CandidateRefused && a.Returns == "" && strings.Contains(a.Refused, "lemurs: the controller holds no composition")
|
||||||
|
}},
|
||||||
|
} {
|
||||||
|
get := func(_ context.Context, bucket, key string) ([]byte, error) {
|
||||||
|
if key == lemursKey {
|
||||||
|
return row.lemurs, nil
|
||||||
|
}
|
||||||
|
readOnly.mu.Lock()
|
||||||
|
defer readOnly.mu.Unlock()
|
||||||
|
return readOnly.kv[bucket+"/"+key], nil
|
||||||
|
}
|
||||||
|
a := returning(ctx, get, "n1", map[string]delivery.Want{"lemurs": {FP: fp}, agent.NodeModule: {}}, runsList, "h", 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 {
|
||||||
|
t.Errorf("%s: the answer is not one a have holds: %v", name, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Issue 1126, the seed's review: a module n1 runs inside its own composition (nats, at a build of the old form) is
|
// Issue 1126, the seed's review: a module n1 runs inside its own composition (nats, at a build of the old form) is
|
||||||
// composed there again, not at the trunk's build as one added; lemurs at the build it runs, keeper (added) at the
|
// composed there again, not at the trunk's build as one added; lemurs at the build it runs, keeper (added) at the
|
||||||
// trunk's.
|
// trunk's.
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
"sort"
|
"sort"
|
||||||
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -31,6 +32,9 @@ type Shadow struct {
|
|||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
candidates map[string]map[string]candidateKey // by group, by node
|
candidates map[string]map[string]candidateKey // by group, 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
|
||||||
|
// it to, the controller's compositions at them.
|
||||||
|
wants map[string]delivery.Want
|
||||||
}
|
}
|
||||||
|
|
||||||
// candidateKey is one candidate key as last read: its value and revision.
|
// candidateKey is one candidate key as last read: its value and revision.
|
||||||
@@ -42,7 +46,7 @@ type candidateKey struct {
|
|||||||
// NewShadow is the shadow composer of one node; answer writes the node's answers into its own have key.
|
// NewShadow is the shadow composer of one node; answer writes the node's answers into its own have key.
|
||||||
func NewShadow(env Env, answer func(ctx context.Context, answers map[string]delivery.CandidateAnswer)) *Shadow {
|
func NewShadow(env Env, answer func(ctx context.Context, answers map[string]delivery.CandidateAnswer)) *Shadow {
|
||||||
return &Shadow{c: New(env), answer: answer, candidates: map[string]map[string]candidateKey{},
|
return &Shadow{c: New(env), answer: answer, candidates: map[string]map[string]candidateKey{},
|
||||||
answers: map[string]delivery.CandidateAnswer{}}
|
answers: map[string]delivery.CandidateAnswer{}, wants: map[string]delivery.Want{}}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Keep runs the shadow until ctx ends, beginning again after backoff(try) whenever it stops: an input it cannot watch
|
// Keep runs the shadow until ctx ends, beginning again after backoff(try) whenever it stops: an input it cannot watch
|
||||||
@@ -84,6 +88,7 @@ func (s *Shadow) Run(ctx context.Context) error {
|
|||||||
defer cancel()
|
defer cancel()
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
s.candidates = map[string]map[string]candidateKey{}
|
s.candidates = map[string]map[string]candidateKey{}
|
||||||
|
s.wants = map[string]delivery.Want{}
|
||||||
s.c = New(s.c.env)
|
s.c = New(s.c.env)
|
||||||
s.mu.Unlock()
|
s.mu.Unlock()
|
||||||
c := s.c
|
c := s.c
|
||||||
@@ -96,6 +101,8 @@ 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.
|
||||||
|
{delivery.AssignerWantsBucket, "want." + node + ".>", nil, nil},
|
||||||
}
|
}
|
||||||
type tagged struct {
|
type tagged struct {
|
||||||
in int
|
in int
|
||||||
@@ -142,6 +149,8 @@ func (s *Shadow) Run(ctx context.Context) error {
|
|||||||
case t.in == 5:
|
case t.in == 5:
|
||||||
c.takeModuleHave(t.u)
|
c.takeModuleHave(t.u)
|
||||||
dirty = !isNodeKey(t.u.Key) || dirty
|
dirty = !isNodeKey(t.u.Key) || dirty
|
||||||
|
case t.in == 6:
|
||||||
|
dirty = s.takeWant(t.u) || dirty
|
||||||
default:
|
default:
|
||||||
ins[t.in].take(c, t.u)
|
ins[t.in].take(c, t.u)
|
||||||
dirty = true
|
dirty = true
|
||||||
@@ -293,10 +302,15 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid
|
|||||||
}
|
}
|
||||||
if p, ok := given[node]; ok && p == nil {
|
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
|
// The head deletes this node's placement file: the node then returns to the controller's compositions, as one the
|
||||||
// catalogue does not place (endWhenUnplaced), and keeps what runs; nothing is removed (issue 1126, the seed's
|
// catalogue does not place (endWhenUnplaced). Those at its wants are compared with what runs (issue 1126, the
|
||||||
// revert, whose check read "would stop running node").
|
// seed's revert, whose check read "would stop running node").
|
||||||
return delivery.CandidateAnswer{Head: head, Revision: rev, Returns: delivery.ReturnsToController,
|
s.mu.Lock()
|
||||||
Verdict: delivery.CandidateSame}, true
|
wants := make(map[string]delivery.Want, len(s.wants))
|
||||||
|
for m, w := range s.wants {
|
||||||
|
wants[m] = w
|
||||||
|
}
|
||||||
|
s.mu.Unlock()
|
||||||
|
return returning(ctx, c.env.Get, node, wants, running, head, rev), true
|
||||||
}
|
}
|
||||||
|
|
||||||
placement := given[node]
|
placement := given[node]
|
||||||
@@ -362,3 +376,88 @@ func sameAnswer(a, b delivery.CandidateAnswer) bool {
|
|||||||
y, errB := delivery.EncodeHaveNode(delivery.HaveNode{Candidates: map[string]delivery.CandidateAnswer{"g": b}})
|
y, errB := delivery.EncodeHaveNode(delivery.HaveNode{Candidates: map[string]delivery.CandidateAnswer{"g": b}})
|
||||||
return errA == nil && errB == nil && string(x) == string(y)
|
return errA == nil && errB == nil && string(x) == string(y)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// takeWant takes one of the node's wants, and says whether it changed.
|
||||||
|
func (s *Shadow) takeWant(u agent.Update) bool {
|
||||||
|
module := ""
|
||||||
|
var w delivery.Want
|
||||||
|
if _, err := delivery.ParseWantNodeKey(u.Key); err == nil {
|
||||||
|
module = agent.NodeModule
|
||||||
|
} else if node, m, err := delivery.ParseWantKey(u.Key); err == nil && node == s.c.env.Node {
|
||||||
|
module = m
|
||||||
|
if !u.Deleted {
|
||||||
|
if w, err = delivery.DecodeWant(u.Value); err != nil || w.Removed || w.FP == "" {
|
||||||
|
u.Deleted = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
old, had := s.wants[module]
|
||||||
|
if u.Deleted {
|
||||||
|
delete(s.wants, module)
|
||||||
|
return had
|
||||||
|
}
|
||||||
|
s.wants[module] = w
|
||||||
|
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
|
||||||
|
// 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}
|
||||||
|
bucket, err := delivery.ComposedBucketFor(node)
|
||||||
|
if err != nil {
|
||||||
|
a.Refused, a.Verdict = err.Error(), delivery.CandidateRefused
|
||||||
|
return a
|
||||||
|
}
|
||||||
|
modules := make([]string, 0, len(wants))
|
||||||
|
for m := range wants {
|
||||||
|
modules = append(modules, m)
|
||||||
|
}
|
||||||
|
sort.Strings(modules)
|
||||||
|
comps := map[string]delivery.Composed{}
|
||||||
|
var missing []string
|
||||||
|
for _, m := range modules {
|
||||||
|
var key string
|
||||||
|
if m == agent.NodeModule {
|
||||||
|
key, err = delivery.ComposedNodeKey(node)
|
||||||
|
} else {
|
||||||
|
key, err = delivery.ComposedKey(node, m, wants[m].FP)
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
missing = append(missing, m+": "+err.Error())
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
raw, err := get(ctx, bucket, key)
|
||||||
|
switch {
|
||||||
|
case err != nil:
|
||||||
|
missing = append(missing, fmt.Sprintf("%s: the controller's composition %s cannot be read: %v", m, key, err))
|
||||||
|
continue
|
||||||
|
case raw == nil:
|
||||||
|
missing = append(missing, fmt.Sprintf("%s: the controller holds no composition %s, which its want names", m, key))
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
comp, err := delivery.DecodeComposed(raw)
|
||||||
|
if err != nil {
|
||||||
|
missing = append(missing, fmt.Sprintf("%s: the controller's composition %s: %v", m, key, err))
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
comps[key] = comp
|
||||||
|
}
|
||||||
|
if len(missing) > 0 {
|
||||||
|
a.Refused = "returning to the controller's compositions, " + strings.Join(missing, "; ")
|
||||||
|
a.Verdict = delivery.CandidateRefused
|
||||||
|
return a
|
||||||
|
}
|
||||||
|
cmp := delivery.CompareComposed(comps, running)
|
||||||
|
a.Returns = delivery.ReturnsToController
|
||||||
|
a.Same, a.Differs, a.Added, a.Removed = cmp.Same, cmp.Differs, cmp.Added, cmp.Removed
|
||||||
|
a.Verdict = delivery.VerdictOf(a)
|
||||||
|
return a
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user