Merge pull request 'The node agent answers a placement candidate in shadow, applies nothing, and keeps its own sealed values in its have (issue 1126, ADR 0559 §4)' (#151) from feat/1126-l2b into main
This commit was merged in pull request #151.
This commit is contained in:
@@ -56,6 +56,14 @@ func (w bucketWatch) Watch(ctx context.Context, pattern string) (<-chan agent.Up
|
|||||||
type sessionHaves struct{ session *link.Session }
|
type sessionHaves struct{ session *link.Session }
|
||||||
|
|
||||||
func (h sessionHaves) Read(ctx context.Context, module string) (agent.Have, bool) {
|
func (h sessionHaves) Read(ctx context.Context, module string) (agent.Have, bool) {
|
||||||
|
if module == agent.NodeModule {
|
||||||
|
// The node's own key: the digest of its own composition it runs, and what that carries sealed.
|
||||||
|
got, ok := h.session.ReadNodeHave(ctx)
|
||||||
|
if !ok || got.Composed == "" {
|
||||||
|
return agent.Have{}, false
|
||||||
|
}
|
||||||
|
return agent.Have{Module: module, Composed: got.Composed, Sealed: got.Sealed}, true
|
||||||
|
}
|
||||||
got, ok := h.session.ReadHave(ctx, module)
|
got, ok := h.session.ReadHave(ctx, module)
|
||||||
if !ok {
|
if !ok {
|
||||||
return agent.Have{}, false
|
return agent.Have{}, false
|
||||||
@@ -68,7 +76,7 @@ func (h sessionHaves) Write(ctx context.Context, x agent.Have) {
|
|||||||
if x.Module == agent.NodeModule {
|
if x.Module == agent.NodeModule {
|
||||||
// The node's own key holds its facts and the digest of the composition it runs.
|
// The node's own key holds its facts and the digest of the composition it runs.
|
||||||
if x.Phase == "" || x.Phase == string(delivery.PhaseRunning) {
|
if x.Phase == "" || x.Phase == string(delivery.PhaseRunning) {
|
||||||
h.session.NoteNode(ctx, x.Composed)
|
h.session.NoteNode(ctx, x.Composed, x.Sealed)
|
||||||
}
|
}
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -511,6 +519,10 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
|
|||||||
unseal: sealOpener(opts.state),
|
unseal: sealOpener(opts.state),
|
||||||
// And what each module's have says it runs sealed: after a restart the agent has applied nothing yet.
|
// And what each module's have says it runs sealed: after a restart the agent has applied nothing yet.
|
||||||
held: func(module string) []string {
|
held: func(module string) []string {
|
||||||
|
if module == agent.NodeModule {
|
||||||
|
got, _ := session.ReadNodeHave(linked)
|
||||||
|
return got.Sealed
|
||||||
|
}
|
||||||
got, ok := session.ReadHave(linked, module)
|
got, ok := session.ReadHave(linked, module)
|
||||||
if !ok {
|
if !ok {
|
||||||
return nil
|
return nil
|
||||||
@@ -523,6 +535,15 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
|
|||||||
}
|
}
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
// And its answer to every placement candidate, placed or not, applying nothing (novox/hq ADR 0559 §4): written into
|
||||||
|
// its own key, `<node>.node`, under candidates.
|
||||||
|
shadowHooks := hooks
|
||||||
|
shadowHooks.refused = nil
|
||||||
|
shadow := selfcompose.NewShadow(composerEnv(session, node, say, shadowHooks),
|
||||||
|
func(ctx context.Context, answers map[string]delivery.CandidateAnswer) {
|
||||||
|
session.NoteCandidates(ctx, answers)
|
||||||
|
})
|
||||||
|
go shadow.Keep(linked, link.Backoff)
|
||||||
if composer := selfComposer(linked, session, node, say, hooks); composer != nil {
|
if composer := selfComposer(linked, session, node, say, hooks); 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
|
||||||
} else {
|
} else {
|
||||||
@@ -838,7 +859,13 @@ func selfComposer(ctx context.Context, session *link.Session, node string, say f
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
say("this node composes its own modules: the catalogue places it (" + key + ")")
|
say("this node composes its own modules: the catalogue places it (" + key + ")")
|
||||||
return selfcompose.New(selfcompose.Env{Node: node,
|
return selfcompose.New(composerEnv(session, node, say, hooks))
|
||||||
|
}
|
||||||
|
|
||||||
|
// composerEnv is what the node's composers read of the bus, the vault and the agent: its own composer and its answer
|
||||||
|
// to placement candidates (selfcompose.Shadow) alike.
|
||||||
|
func composerEnv(session *link.Session, node string, say func(string), hooks composerHooks) selfcompose.Env {
|
||||||
|
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) {
|
||||||
return bucketWatch{session, bucket}.Watch(ctx, pattern)
|
return bucketWatch{session, bucket}.Watch(ctx, pattern)
|
||||||
},
|
},
|
||||||
@@ -857,7 +884,7 @@ func selfComposer(ctx context.Context, session *link.Session, node string, say f
|
|||||||
},
|
},
|
||||||
// And the vault's word that a secret changed, on which it composes again (ADR 0547 §2).
|
// And the vault's word that a secret changed, on which it composes again (ADR 0547 §2).
|
||||||
Hear: session.Hear,
|
Hear: session.Hear,
|
||||||
Running: hooks.running, Unseal: hooks.unseal, Refused: hooks.refused, Held: hooks.held})
|
Running: hooks.running, Unseal: hooks.unseal, Refused: hooks.refused, Held: hooks.held}
|
||||||
}
|
}
|
||||||
|
|
||||||
// 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
|
||||||
|
|||||||
@@ -113,7 +113,7 @@ func (b *itBed) start(trustOperator bool) (stop func()) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
b.session = &link.Session{Membership: link.Membership{Node: itNode}, Timeout: 2 * time.Second,
|
b.session = &link.Session{Membership: link.Membership{Node: itNode}, Timeout: 2 * time.Second,
|
||||||
Say: b.say}
|
Say: b.say, Keys: func() link.RecordKeys { return link.RecordKeys{} }}
|
||||||
b.session.Attach(context.Background(), link.OverNATS{Conn: b.conn, JS: b.js, Node: itNode})
|
b.session.Attach(context.Background(), link.OverNATS{Conn: b.conn, JS: b.js, Node: itNode})
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
done := make(chan struct{})
|
done := make(chan struct{})
|
||||||
@@ -316,7 +316,8 @@ func TestNatsTheHandPathAppliesACompositionTheOperatorSigned(t *testing.T) {
|
|||||||
// Issue 1126 (ADR 0547 §2, the milestone's H8): the node agent started whole against a real bus, once the catalogue
|
// Issue 1126 (ADR 0547 §2, the milestone's H8): the node agent started whole against a real bus, once the catalogue
|
||||||
// places its node (the placement written after it started), composes a module from its want, the indexer's manifest, its record and its mesh-config value, and applies it,
|
// places its node (the placement written after it started), composes a module from its want, the indexer's manifest, its record and its mesh-config value, and applies it,
|
||||||
// with no key in the controller's compositions; a changed config value is composed again on its event and applied as a
|
// with no key in the controller's compositions; a changed config value is composed again on its event and applied as a
|
||||||
// revision of what runs.
|
// revision of what runs. And a catalogue pull request's placement candidate equal to its placement is answered "same"
|
||||||
|
// in the node's own key, and nothing is applied for it (ADR 0559 §4, subtask 10 of the milestone).
|
||||||
func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) {
|
func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) {
|
||||||
b := newItBed(t)
|
b := newItBed(t)
|
||||||
kvOf := func(bucket string) nats.KeyValue {
|
kvOf := func(bucket string) nats.KeyValue {
|
||||||
@@ -329,6 +330,11 @@ func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) {
|
|||||||
placement, modules, manifests := kvOf(delivery.PlacementBucket), kvOf(delivery.ModulesBucket), kvOf(delivery.ManifestsBucket)
|
placement, modules, manifests := kvOf(delivery.PlacementBucket), kvOf(delivery.ModulesBucket), kvOf(delivery.ManifestsBucket)
|
||||||
resolved := kvOf(selfcompose.ResolvedBucket)
|
resolved := kvOf(selfcompose.ResolvedBucket)
|
||||||
kvOf(delivery.BuilderBuildsBucket)
|
kvOf(delivery.BuilderBuildsBucket)
|
||||||
|
candidates := kvOf(delivery.PlacementCandidatesBucket)
|
||||||
|
// The controller's stream the node's facts go into, so the node's key carries the facts it says.
|
||||||
|
if _, err := b.js.AddStream(&nats.StreamConfig{Name: "CONTROL", Subjects: []string{link.ReportSubject(itNode)}}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
path := filepath.Join(b.dir, "web.conf")
|
path := filepath.Join(b.dir, "web.conf")
|
||||||
man := `{"module":"web","version":"1","config":{"greeting":{"type":"string","description":"what the file says","default":"hello"}},` +
|
man := `{"module":"web","version":"1","config":{"greeting":{"type":"string","description":"what the file says","default":"hello"}},` +
|
||||||
`"resources":[{"id":"conf","type":"file","path":"` + path + `","mode":"0644","content":"greeting=${config:greeting}\n"}]}`
|
`"resources":[{"id":"conf","type":"file","path":"` + path + `","mode":"0644","content":"greeting=${config:greeting}\n"}]}`
|
||||||
@@ -385,4 +391,37 @@ func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) {
|
|||||||
h := b.have("web")
|
h := b.have("web")
|
||||||
return h.FP == fp && h.Composed != first && h.Phase == delivery.PhaseRunning
|
return h.FP == fp && h.Composed != first && h.Phase == delivery.PhaseRunning
|
||||||
})
|
})
|
||||||
|
|
||||||
|
// The node says its facts (its record, as the test wrote it), then a candidate equal to its placement is written: the
|
||||||
|
// node answers "same" at the candidate's head and revision in its own key, and applies nothing.
|
||||||
|
b.session.SayFacts(context.Background(), link.Report{Profile: map[string]any{"capabilities": []any{
|
||||||
|
map[string]any{"name": "service-manager", "present": true}}}})
|
||||||
|
second := b.have("web").Composed
|
||||||
|
same, _ := delivery.DecodePlacement(place)
|
||||||
|
craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "feat/x",
|
||||||
|
Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h1"}, Placement: &same})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
ckey, _ := delivery.PlacementCandidateKey("feat/x", itNode)
|
||||||
|
crev, err := candidates.Put(ckey, craw)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
var answer delivery.CandidateAnswer
|
||||||
|
within(t, 10*time.Second, "the node did not answer its candidate", func() bool {
|
||||||
|
e, err := b.haves.Get(hkey)
|
||||||
|
if err != nil {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
n, err := delivery.DecodeHaveNode(e.Value())
|
||||||
|
answer = n.Candidates["feat/x"]
|
||||||
|
return err == nil && answer.Head == "h1" && n.Record != nil
|
||||||
|
})
|
||||||
|
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)
|
||||||
|
}
|
||||||
|
if h := b.have("web"); h.Composed != second || b.file("web") != "greeting=salut\n" {
|
||||||
|
t.Errorf("a candidate moved what runs: %+v, %q", h, b.file("web"))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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.20261026210326-979644420117
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f
|
||||||
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
|
||||||
|
|||||||
@@ -1,5 +1,15 @@
|
|||||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk=
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk=
|
||||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||||
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa h1:tT6sl3vY0iVAXw+mqy3iGKUBUjVBRWvQvKUOTmTyWwk=
|
||||||
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||||
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef h1:uMqybf/sTAjapyllVjy8mjCi8evpcOmgnjmqRdJRsFo=
|
||||||
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||||
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56 h1:PUc5s3oaWEg0uyVmXhJSSK3x0IXZBXWGqq30U/CSx30=
|
||||||
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||||
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f h1:AKzV973UTWe0OYFoIjgGu0f+B26xh8QsAHzCpsTIjM8=
|
||||||
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||||
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERguQmCGYVNdRTrFcZ1dhpU6GWo+Q3EDhHOrVk=
|
||||||
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/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=
|
||||||
|
|||||||
@@ -135,7 +135,7 @@ type Env struct {
|
|||||||
// Policy is a module's upgrade policy as its manifest derives it (catalogue.DerivedUpgrade); with SelfComposed, a
|
// Policy is a module's upgrade policy as its manifest derives it (catalogue.DerivedUpgrade); with SelfComposed, a
|
||||||
// move of a module of policy record that recreates a container or moves its data waits for a person's yes in its
|
// move of a module of policy record that recreates a container or moves its data waits for a person's yes in its
|
||||||
// want (ADR 0547 §3). Nil judges no move.
|
// want (ADR 0547 §3). Nil judges no move.
|
||||||
Policy func(module string) string
|
Policy func(ctx context.Context, module string) string
|
||||||
}
|
}
|
||||||
|
|
||||||
// transientFor is how many transient retries say the phase applying before the have says failed: past them the bus has
|
// transientFor is how many transient retries say the phase applying before the have says failed: past them the bus has
|
||||||
@@ -377,8 +377,8 @@ func (a *Agent) NotComposed(ctx context.Context, why string) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// writeHave writes a module's have; on a node composing its own, with the sealed values of the composition it names
|
// writeHave writes a module's have (the node's own key for NodeModule); on a node composing its own, with the sealed
|
||||||
// when the caller gave none (Have.Sealed): those of the composition the agent applied or holds at that digest, else
|
// values of the composition it names when the caller gave none (Have.Sealed): those of the composition the agent applied or holds at that digest, else
|
||||||
// those the module's last have said at that digest.
|
// those the module's last have said at that digest.
|
||||||
//
|
//
|
||||||
// A writer that names no composition (one written from the agent's running entry before this agent ran anything of the
|
// A writer that names no composition (one written from the agent's running entry before this agent ran anything of the
|
||||||
@@ -394,7 +394,9 @@ func (a *Agent) writeHave(ctx context.Context, h Have) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if a.env.SelfComposed && h.Sealed == nil && h.Module != NodeModule && h.Composed != "" {
|
// The node's own composition too: its core boxes are sealed again by the vault at a restart, and kept in its key the
|
||||||
|
// composer reuses them, so the restart applies nothing again (review of L2, round 3, finding 1).
|
||||||
|
if a.env.SelfComposed && h.Sealed == nil && h.Composed != "" {
|
||||||
h.Sealed = a.sealedOf(ctx, h.Module, h.FP, h.Composed)
|
h.Sealed = a.sealedOf(ctx, h.Module, h.FP, h.Composed)
|
||||||
}
|
}
|
||||||
a.env.Haves.Write(ctx, h)
|
a.env.Haves.Write(ctx, h)
|
||||||
@@ -1101,7 +1103,7 @@ func short(s string) string {
|
|||||||
// for this composition; the have says phase waiting-yes with the judgement, and the assigner asks a person. A composition
|
// for this composition; the have says phase waiting-yes with the judgement, and the assigner asks a person. A composition
|
||||||
// equal to what runs is no move. True when the module waits.
|
// equal to what runs is no move. True when the module waits.
|
||||||
func (a *Agent) waitsForYes(ctx context.Context, module string, w wantNow, comp compNow) bool {
|
func (a *Agent) waitsForYes(ctx context.Context, module string, w wantNow, comp compNow) bool {
|
||||||
if module == NodeModule || a.env.Policy == nil || a.env.Policy(module) != catalogue.PolicyRecord || w.Want.Yes == comp.Dig {
|
if module == NodeModule || a.env.Policy == nil || a.env.Policy(ctx, module) != catalogue.PolicyRecord || w.Want.Yes == comp.Dig {
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
a.mu.Lock()
|
a.mu.Lock()
|
||||||
|
|||||||
@@ -964,7 +964,7 @@ func TestAModuleRunningFromTheKeptDeclarationKeepsItsFilledPaths(t *testing.T) {
|
|||||||
func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes(t *testing.T) {
|
func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes(t *testing.T) {
|
||||||
b := newBed(t)
|
b := newBed(t)
|
||||||
b.env.SelfComposed, b.env.Trusted = true, nil
|
b.env.SelfComposed, b.env.Trusted = true, nil
|
||||||
b.env.Policy = func(m string) string {
|
b.env.Policy = func(_ context.Context, m string) string {
|
||||||
if m == "db" || m == "rec" {
|
if m == "db" || m == "rec" {
|
||||||
return catalogue.PolicyRecord
|
return catalogue.PolicyRecord
|
||||||
}
|
}
|
||||||
@@ -1001,6 +1001,21 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes(
|
|||||||
// The same composition handed again is no move.
|
// The same composition handed again is no move.
|
||||||
b.comps.put(ck, craw)
|
b.comps.put(ck, craw)
|
||||||
time.Sleep(50 * time.Millisecond)
|
time.Sleep(50 * time.Millisecond)
|
||||||
|
// The node's own composition carries its sealed values in its key, as a module's have does, so a restart reuses its
|
||||||
|
// core boxes and applies nothing again (review of L2, round 3, finding 1).
|
||||||
|
box, _ := json.Marshal(map[string]any{"id": "core-box", "type": "file", "path": "/tmp/r6/core-box", "sealed": "core"})
|
||||||
|
own := delivery.Composed{Node: node, Resources: []json.RawMessage{box}}
|
||||||
|
nk, _ := delivery.ComposedNodeKey(node)
|
||||||
|
nraw, _ := delivery.EncodeComposed(own)
|
||||||
|
ndig, _ := delivery.ComposedDigestOf(own)
|
||||||
|
b.comps.put(nk, nraw)
|
||||||
|
nwk, _ := delivery.WantNodeKey(node)
|
||||||
|
nwraw, _ := delivery.EncodeNodeWant(delivery.NodeWant{Self: true})
|
||||||
|
b.wants.put(nwk, nwraw)
|
||||||
|
eventually(t, "the node's own key does not carry its sealed values", func() bool {
|
||||||
|
h := b.haves.last(NodeModule)
|
||||||
|
return h.Composed == ndig && len(h.Sealed) == 1 && h.Sealed[0] == "core"
|
||||||
|
})
|
||||||
// The node cannot compose (a stand-in refused): each wanted module's have says so, failed, naming why, and still
|
// The node cannot compose (a stand-in refused): each wanted module's have says so, failed, naming why, and still
|
||||||
// names what runs; composed again, the have is what it runs. web was applied by this agent; idle was not since the
|
// names what runs; composed again, the have is what it runs. web was applied by this agent; idle was not since the
|
||||||
// agent's start (the switch and every delivery of the node-engine restart it), so what runs is what its have said.
|
// agent's start (the switch and every delivery of the node-engine restart it), so what runs is what its have said.
|
||||||
@@ -1035,7 +1050,7 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes(
|
|||||||
// A revision at the same build (a config value changed) is applied.
|
// A revision at the same build (a config value changed) is applied.
|
||||||
_, craw2, dig2 := unsigned("web", fp, file("web", "x", "2"))
|
_, craw2, dig2 := unsigned("web", fp, file("web", "x", "2"))
|
||||||
b.comps.put(ck, craw2)
|
b.comps.put(ck, craw2)
|
||||||
eventually(t, "the revision was not applied", func() bool { return len(b.appliedList()) == 2 && b.haves.last("web").Composed == dig2 })
|
eventually(t, "the revision was not applied", func() bool { return len(b.appliedList()) == 3 && b.haves.last("web").Composed == dig2 })
|
||||||
|
|
||||||
// A record module: its first composition is no move; a new image recreates its container and waits for a yes.
|
// A record module: its first composition is no move; a new image recreates its container and waits for a yes.
|
||||||
container := func(image string) json.RawMessage {
|
container := func(image string) json.RawMessage {
|
||||||
@@ -1046,7 +1061,7 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes(
|
|||||||
dk, draw, _ := unsigned("db", fp, container("db@sha256:"+strings.Repeat("1", 64)))
|
dk, draw, _ := unsigned("db", fp, container("db@sha256:"+strings.Repeat("1", 64)))
|
||||||
b.comps.put(dk, draw)
|
b.comps.put(dk, draw)
|
||||||
want("db", fp, "")
|
want("db", fp, "")
|
||||||
eventually(t, "the record module's first composition was not applied", func() bool { return len(b.appliedList()) == 3 })
|
eventually(t, "the record module's first composition was not applied", func() bool { return len(b.appliedList()) == 4 })
|
||||||
fpB := fpOf("b")
|
fpB := fpOf("b")
|
||||||
dk2, draw2, ddig2 := unsigned("db", fpB, container("db@sha256:"+strings.Repeat("2", 64)))
|
dk2, draw2, ddig2 := unsigned("db", fpB, container("db@sha256:"+strings.Repeat("2", 64)))
|
||||||
// rec is a record module this agent has run nothing of since its start (the switch and every delivery of the
|
// rec is a record module this agent has run nothing of since its start (the switch and every delivery of the
|
||||||
|
|||||||
+66
-2
@@ -47,6 +47,11 @@ type haveState struct {
|
|||||||
node *delivery.HaveNode
|
node *delivery.HaveNode
|
||||||
// nodeComposed is the digest of the node's own composition it runs (ADR 0539 §9), kept beside the node's facts.
|
// nodeComposed is the digest of the node's own composition it runs (ADR 0539 §9), kept beside the node's facts.
|
||||||
nodeComposed string
|
nodeComposed string
|
||||||
|
// 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 []string
|
||||||
|
// nodeCandidates is the node's answer to each placement candidate group (novox/hq ADR 0559 §4), written into its key.
|
||||||
|
nodeCandidates map[string]delivery.CandidateAnswer
|
||||||
// written is the build of each module this node-engine wrote a have key for and that the declaration it applied
|
// written is the build of each module this node-engine wrote a have key for and that the declaration it applied
|
||||||
// still holds; removed is each such module the latest declaration no longer holds, with the build its key last
|
// still holds; removed is each such module the latest declaration no longer holds, with the build its key last
|
||||||
// said, until the key is written once with the phase removed (issue 1369, novox/hq ADR 0408 §4: never deleted).
|
// said, until the key is written once with the phase removed (issue 1369, novox/hq ADR 0408 §4: never deleted).
|
||||||
@@ -416,6 +421,8 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) {
|
|||||||
var raw []byte
|
var raw []byte
|
||||||
facts := *s.node
|
facts := *s.node
|
||||||
facts.Composed = s.nodeComposed
|
facts.Composed = s.nodeComposed
|
||||||
|
facts.Sealed = s.nodeSealed
|
||||||
|
facts.Candidates = s.nodeCandidates
|
||||||
if raw, err = delivery.EncodeHaveNode(facts); err == nil {
|
if raw, err = delivery.EncodeHaveNode(facts); err == nil {
|
||||||
want := facts
|
want := facts
|
||||||
if !q.putHave(write, bucket, key, raw, func(current []byte) bool {
|
if !q.putHave(write, bucket, key, raw, func(current []byte) bool {
|
||||||
@@ -703,12 +710,14 @@ func (q *Session) NoteRemoved(ctx context.Context, module, build string) {
|
|||||||
q.writeHaves(ctx, bus)
|
q.writeHaves(ctx, bus)
|
||||||
}
|
}
|
||||||
|
|
||||||
// NoteNode takes the digest of the node's own composition it runs, and writes the node's key where it differs.
|
// NoteNode takes the digest of the node's own composition it runs and the values it carries sealed to the node's key
|
||||||
func (q *Session) NoteNode(ctx context.Context, composed string) {
|
// (nil for a composition the node did not make itself), and writes the node's key where it differs.
|
||||||
|
func (q *Session) NoteNode(ctx context.Context, composed string, sealed []string) {
|
||||||
q.init()
|
q.init()
|
||||||
s := &q.haves
|
s := &q.haves
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
s.nodeComposed = composed
|
s.nodeComposed = composed
|
||||||
|
s.nodeSealed = sealed
|
||||||
if s.node == nil {
|
if s.node == nil {
|
||||||
n := delivery.HaveNode{V: delivery.Version}
|
n := delivery.HaveNode{V: delivery.Version}
|
||||||
s.node = &n
|
s.node = &n
|
||||||
@@ -720,6 +729,61 @@ func (q *Session) NoteNode(ctx context.Context, composed string) {
|
|||||||
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
|
||||||
|
// 0559 §4), and writes the node's key where it differs; a group left out is no longer answered.
|
||||||
|
func (q *Session) NoteCandidates(ctx context.Context, answers map[string]delivery.CandidateAnswer) {
|
||||||
|
q.init()
|
||||||
|
s := &q.haves
|
||||||
|
s.mu.Lock()
|
||||||
|
s.nodeCandidates = answers
|
||||||
|
if len(answers) == 0 {
|
||||||
|
s.nodeCandidates = nil
|
||||||
|
}
|
||||||
|
// Before the node's facts are first said, its key is not written for an answer alone: it would carry no record. The
|
||||||
|
// answers ride with the first facts.
|
||||||
|
known := s.node != nil
|
||||||
|
s.mu.Unlock()
|
||||||
|
if !known {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
q.mu.Lock()
|
||||||
|
bus := q.bus
|
||||||
|
q.mu.Unlock()
|
||||||
|
q.writeHaves(ctx, bus)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ReadNodeHave is the node's own key as the bucket holds it, and whether it holds one: what the node's own composition
|
||||||
|
// it last ran is and carries sealed, read after a restart (ADR 0547 §3).
|
||||||
|
func (q *Session) ReadNodeHave(ctx context.Context) (delivery.HaveNode, bool) {
|
||||||
|
q.init()
|
||||||
|
q.mu.Lock()
|
||||||
|
bus := q.bus
|
||||||
|
q.mu.Unlock()
|
||||||
|
hb, ok := bus.(HaveBus)
|
||||||
|
if !ok || bus == nil {
|
||||||
|
return delivery.HaveNode{}, false
|
||||||
|
}
|
||||||
|
key, err := delivery.HaveNodeKey(q.Membership.Node)
|
||||||
|
if err != nil {
|
||||||
|
return delivery.HaveNode{}, false
|
||||||
|
}
|
||||||
|
read, cancel := context.WithTimeout(ctx, q.timeout())
|
||||||
|
defer cancel()
|
||||||
|
bucket, err := hb.HaveBind(read)
|
||||||
|
if err != nil {
|
||||||
|
return delivery.HaveNode{}, false
|
||||||
|
}
|
||||||
|
value, _, err := bucket.Get(read, key)
|
||||||
|
if err != nil {
|
||||||
|
return delivery.HaveNode{}, false
|
||||||
|
}
|
||||||
|
n, err := delivery.DecodeHaveNode(value)
|
||||||
|
if err != nil {
|
||||||
|
return delivery.HaveNode{}, false
|
||||||
|
}
|
||||||
|
return n, true
|
||||||
|
}
|
||||||
|
|
||||||
// ReadHave is the have this node-engine last wrote for a module, and whether the bucket holds one. A bucket that cannot
|
// ReadHave is the have this node-engine last wrote for a module, and whether the bucket holds one. A bucket that cannot
|
||||||
// be read answers as none: the agent then compares with the declaration it kept, and applies only what differs.
|
// be read answers as none: the agent then compares with the declaration it kept, and applies only what differs.
|
||||||
func (q *Session) ReadHave(ctx context.Context, module string) (delivery.Have, bool) {
|
func (q *Session) ReadHave(ctx context.Context, module string) (delivery.Have, bool) {
|
||||||
|
|||||||
@@ -582,9 +582,22 @@ func TestAnAppliedModulesHaveNamesTheFingerprintAndTheDigestAndFollowsItsHealth(
|
|||||||
if got := l.have(t, "n1.web"); got.Phase != delivery.PhaseRemoved {
|
if got := l.have(t, "n1.web"); got.Phase != delivery.PhaseRemoved {
|
||||||
t.Fatalf("removed: %+v", got)
|
t.Fatalf("removed: %+v", got)
|
||||||
}
|
}
|
||||||
q.NoteNode(context.Background(), dig)
|
q.NoteNode(context.Background(), dig, []string{"core"})
|
||||||
if n, err := delivery.DecodeHaveNode(l.kv["n1.node"].value); err != nil || n.Composed != dig {
|
if n, err := delivery.DecodeHaveNode(l.kv["n1.node"].value); err != nil || n.Composed != dig || len(n.Sealed) != 1 || n.Sealed[0] != "core" {
|
||||||
t.Fatalf("the node key: %+v %v", n, err)
|
t.Fatalf("the node key, with its own composition's sealed values: %+v %v", n, err)
|
||||||
|
}
|
||||||
|
if n, ok := q.ReadNodeHave(context.Background()); !ok || n.Composed != dig || len(n.Sealed) != 1 {
|
||||||
|
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"}}}
|
||||||
|
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 {
|
||||||
|
t.Fatalf("the node key with its answers: %+v %v", n, err)
|
||||||
|
}
|
||||||
|
q.NoteCandidates(context.Background(), nil)
|
||||||
|
if n, _ := delivery.DecodeHaveNode(l.kv["n1.node"].value); n.Candidates != nil {
|
||||||
|
t.Fatalf("an answer left after its group went: %+v", n.Candidates)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -453,11 +453,10 @@ func (c *Composer) runningBox(box string) string {
|
|||||||
}
|
}
|
||||||
if c.env.Held != nil {
|
if c.env.Held != nil {
|
||||||
c.mu.Lock()
|
c.mu.Lock()
|
||||||
|
// The node's own key among them (agent.NodeModule): its core boxes, as its last run said them.
|
||||||
modules := make([]string, 0, len(c.wants))
|
modules := make([]string, 0, len(c.wants))
|
||||||
for m := range c.wants {
|
for m := range c.wants {
|
||||||
if m != agent.NodeModule {
|
modules = append(modules, m)
|
||||||
modules = append(modules, m)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
c.mu.Unlock()
|
c.mu.Unlock()
|
||||||
sort.Strings(modules)
|
sort.Strings(modules)
|
||||||
@@ -514,15 +513,16 @@ func (c *Composer) builtManifest(ctx context.Context, module, fp string) (string
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Policy is a module's upgrade policy as its manifest at the indexer's trunk derives it (catalogue.DerivedUpgrade), for
|
// Policy is a module's upgrade policy as its manifest at the indexer's trunk derives it (catalogue.DerivedUpgrade), for
|
||||||
// the agent's judgement of a move (ADR 0547 §3); empty when the manifest is not known or cannot be read.
|
// the agent's judgement of a move (ADR 0547 §3); empty when the manifest is not known or cannot be read. The manifest is
|
||||||
func (c *Composer) Policy(module string) string {
|
// read within ctx, the agent's (review of L2, round 1, finding 5), so a stopping agent stops the read.
|
||||||
|
func (c *Composer) Policy(ctx context.Context, module string) string {
|
||||||
c.mu.Lock()
|
c.mu.Lock()
|
||||||
m, ok := c.modules[module]
|
m, ok := c.modules[module]
|
||||||
c.mu.Unlock()
|
c.mu.Unlock()
|
||||||
if !ok || m.Manifest == "" {
|
if !ok || m.Manifest == "" {
|
||||||
return ""
|
return ""
|
||||||
}
|
}
|
||||||
raw, err := reader{c: c}.Manifest(context.Background(), m.Manifest)
|
raw, err := reader{c: c}.Manifest(ctx, m.Manifest)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return ""
|
return ""
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"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/delivery"
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
|
|
||||||
"github.com/novox/mesh-host/internal/agent"
|
"github.com/novox/mesh-host/internal/agent"
|
||||||
@@ -237,14 +238,18 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
|||||||
}
|
}
|
||||||
return sealed[0]
|
return sealed[0]
|
||||||
}
|
}
|
||||||
|
// And the node's own key (review of L2, round 3, finding 1): a core box its own composition carries is held there, and
|
||||||
|
// kept after a restart as a module's is, so the node's own composition does not move either.
|
||||||
for _, row := range []struct {
|
for _, row := range []struct {
|
||||||
name, held string
|
name, held, in string
|
||||||
kept bool
|
kept bool
|
||||||
}{{"the have holds the rotated value", boxOf(string(rotated.Value)), true}, {"the have holds the value before the rotation", boxOf(before), false}} {
|
}{{"the have holds the rotated value", boxOf(string(rotated.Value)), "keeper", true},
|
||||||
|
{"the have holds the value before the rotation", boxOf(before), "keeper", false},
|
||||||
|
{"the node's own key holds the rotated value", boxOf(string(rotated.Value)), agent.NodeModule, true}} {
|
||||||
restarted := env
|
restarted := env
|
||||||
restarted.Running = func() []delivery.Composed { return nil }
|
restarted.Running = func() []delivery.Composed { return nil }
|
||||||
restarted.Held = func(module string) []string {
|
restarted.Held = func(module string) []string {
|
||||||
if module == "keeper" {
|
if module == row.in {
|
||||||
return []string{row.held}
|
return []string{row.held}
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
@@ -258,7 +263,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
|||||||
return nil, fmt.Errorf("not sealed to n1: %s", box)
|
return nil, fmt.Errorf("not sealed to n1: %s", box)
|
||||||
}
|
}
|
||||||
again := New(restarted)
|
again := New(restarted)
|
||||||
again.wants = map[string]delivery.Want{"keeper": {FP: fp, Self: true}}
|
again.wants = map[string]delivery.Want{"keeper": {FP: fp, Self: true}, agent.NodeModule: {}}
|
||||||
rows, err := reader{c: again}.OwnSecrets(ctx, "n1")
|
rows, err := reader{c: again}.OwnSecrets(ctx, "n1")
|
||||||
if err != nil || len(rows) != 1 {
|
if err != nil || len(rows) != 1 {
|
||||||
t.Fatalf("%s: own secrets %+v, %v", row.name, rows, err)
|
t.Fatalf("%s: own secrets %+v, %v", row.name, rows, err)
|
||||||
@@ -313,4 +318,292 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
|||||||
if asked, _ := json.Marshal(sent["credentials"]); !strings.Contains(string(asked), `"role":"consumer"`) || !strings.Contains(string(asked), `"role":"provider"`) {
|
if asked, _ := json.Marshal(sent["credentials"]); !strings.Contains(string(asked), `"role":"consumer"`) || !strings.Contains(string(asked), `"role":"provider"`) {
|
||||||
t.Errorf("the vault was asked %s", asked)
|
t.Errorf("the vault was asked %s", asked)
|
||||||
}
|
}
|
||||||
|
// ADR 0559 §4: the node answers placement candidates in shadow, placed or not, applying nothing and handing no
|
||||||
|
// composition: the same; a module the group adds, its secret not yet issued; a module removed; a placement it cannot
|
||||||
|
// compose; a group whose keys went is no longer answered. What runs is never composed from a candidate: the
|
||||||
|
// Composer does not read the state.
|
||||||
|
for _, in := range New(env).inputs() {
|
||||||
|
if in.bucket == delivery.PlacementCandidatesBucket {
|
||||||
|
t.Fatal("the node's own composer reads placement candidates")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
sb := newPatternBus()
|
||||||
|
trunk := delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"}, Modules: []delivery.PlacedModule{{Module: "lemurs"}}}
|
||||||
|
mods := map[string]delivery.Module{"lemurs": {FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"},
|
||||||
|
"keeper": {FP2: fp, Manifest: kdigest, Repository: "novox/k", Commit: "c"}}
|
||||||
|
readOnly := &fakeBus{kv: map[string][]byte{delivery.ManifestsBucket + "/" + key: man, delivery.ManifestsBucket + "/" + kkey: kman}}
|
||||||
|
senv := readOnly.env()
|
||||||
|
senv.Watch = sb.watch
|
||||||
|
senv.Own = func(context.Context, map[string]any) (json.RawMessage, error) {
|
||||||
|
return json.RawMessage(`{"node":"n1","key":"sk","own":[]}`), nil // nothing issued for keeper: it is not placed here
|
||||||
|
}
|
||||||
|
// What n1 runs: lemurs and its own composition, as its composer composes them from the trunk.
|
||||||
|
running, _, err := frombus.Compose(ctx, fixedReader{placements: map[string]delivery.Placement{"n1": trunk},
|
||||||
|
records: map[string]delivery.NodeRecord{"n1": {Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}},
|
||||||
|
modules: mods, manifests: map[string][]byte{digest: []byte(lemurs), kdigest: []byte(keeper)}}, frombus.Mesh{}, frombus.Scope{Node: "n1"}, map[string]string{"lemurs": fp})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
senv.Running = func() []delivery.Composed {
|
||||||
|
var out []delivery.Composed
|
||||||
|
for _, c := range running {
|
||||||
|
out = append(out, c)
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
answered := make(chan map[string]delivery.CandidateAnswer, 8)
|
||||||
|
sh := NewShadow(senv, func(_ context.Context, a map[string]delivery.CandidateAnswer) { answered <- a })
|
||||||
|
sctx, scancel := context.WithCancel(ctx)
|
||||||
|
defer scancel()
|
||||||
|
// Review of L2b, round 1, finding 2: the candidates' state is not there when the node starts (a delivery before the
|
||||||
|
// indexer made it): the shadow tries again, and answers once it appears.
|
||||||
|
sb.missing(delivery.PlacementCandidatesBucket, true)
|
||||||
|
go sh.Keep(sctx, func(int) time.Duration { return 5 * time.Millisecond })
|
||||||
|
time.Sleep(30 * time.Millisecond)
|
||||||
|
sb.missing(delivery.PlacementCandidatesBucket, false)
|
||||||
|
sb.made(6)
|
||||||
|
rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
||||||
|
tp, _ := delivery.EncodePlacement(trunk)
|
||||||
|
sb.put(delivery.PlacementBucket, "placement.n1", tp, 1)
|
||||||
|
sb.put(delivery.HaveBucket, "n1.node", rec, 1)
|
||||||
|
for name, m := range mods {
|
||||||
|
raw, _ := delivery.EncodeModule(m)
|
||||||
|
sb.put(delivery.ModulesBucket, name, raw, 1)
|
||||||
|
}
|
||||||
|
candidate := func(group string, rev uint64, modules ...string) {
|
||||||
|
p := trunk
|
||||||
|
p.Modules = nil
|
||||||
|
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})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
k, _ := delivery.PlacementCandidateKey(group, "n1")
|
||||||
|
sb.put(delivery.PlacementCandidatesBucket, k, raw, rev)
|
||||||
|
}
|
||||||
|
candidate("same", 3, "lemurs")
|
||||||
|
candidate("adds", 4, "lemurs", "keeper")
|
||||||
|
candidate("drops", 5)
|
||||||
|
candidate("ghost", 6, "lemurs", "ghost")
|
||||||
|
sb.caught()
|
||||||
|
var answers map[string]delivery.CandidateAnswer
|
||||||
|
waitFor := func(why string, ok func(map[string]delivery.CandidateAnswer) bool) {
|
||||||
|
t.Helper()
|
||||||
|
deadline := time.After(10 * time.Second)
|
||||||
|
for answers == nil || !ok(answers) {
|
||||||
|
select {
|
||||||
|
case answers = <-answered:
|
||||||
|
case <-deadline:
|
||||||
|
t.Fatalf("%s: last answers %+v", why, answers)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
waitFor("every group answered", func(a map[string]delivery.CandidateAnswer) bool { return len(a) == 4 })
|
||||||
|
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
|
||||||
|
},
|
||||||
|
"adds": func(a delivery.CandidateAnswer) bool {
|
||||||
|
return a.Verdict == delivery.CandidateDiffers && len(a.Added) == 1 && a.Added[0].Module == "keeper" &&
|
||||||
|
len(a.Added[0].Unissued) == 1 && len(a.Refusals) == 0
|
||||||
|
},
|
||||||
|
"drops": func(a delivery.CandidateAnswer) bool {
|
||||||
|
return a.Verdict == delivery.CandidateDiffers && len(a.Removed) == 1 && a.Removed[0] == "lemurs"
|
||||||
|
},
|
||||||
|
"ghost": func(a delivery.CandidateAnswer) bool {
|
||||||
|
return a.Verdict == delivery.CandidateRefused && a.Refused != ""
|
||||||
|
},
|
||||||
|
} {
|
||||||
|
if !want(answers[group]) {
|
||||||
|
t.Errorf("%s: %+v", group, answers[group])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// The node's own key changing (its answers written) composes nothing again; a group's key deleted drops its answer.
|
||||||
|
sb.put(delivery.HaveBucket, "n1.node", rec, 2)
|
||||||
|
select {
|
||||||
|
case a := <-answered:
|
||||||
|
t.Errorf("answers handed again for nothing changed: %+v", a)
|
||||||
|
case <-time.After(100 * time.Millisecond):
|
||||||
|
}
|
||||||
|
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) == 3 })
|
||||||
|
|
||||||
|
// 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.
|
||||||
|
sb.missing(delivery.PlacementCandidatesBucket, true)
|
||||||
|
sb.end(delivery.ModulesBucket)
|
||||||
|
dropsKey, _ := delivery.PlacementCandidateKey("drops", "n1")
|
||||||
|
sb.del(delivery.PlacementCandidatesBucket, dropsKey)
|
||||||
|
time.Sleep(30 * time.Millisecond)
|
||||||
|
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) == 2
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// patternBus is a bus whose every watch gets the updates of its bucket matching its pattern, as the bus's own watches
|
||||||
|
// do: two watches of one bucket each see what matches theirs.
|
||||||
|
type patternBus struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
watches []patternWatch
|
||||||
|
absent map[string]bool
|
||||||
|
// kv is each bucket's keys now; once replay is set, a new watch gets those matching it and then Caught, as a watch
|
||||||
|
// of the bus's own does.
|
||||||
|
kv map[string]map[string]agent.Update
|
||||||
|
replay bool
|
||||||
|
}
|
||||||
|
|
||||||
|
// end ends every watch of a bucket, as a watch the bus drops, and from then on each new watch replays what is there.
|
||||||
|
func (b *patternBus) end(bucket string) {
|
||||||
|
b.mu.Lock()
|
||||||
|
defer b.mu.Unlock()
|
||||||
|
b.replay = true
|
||||||
|
kept := b.watches[:0]
|
||||||
|
for _, w := range b.watches {
|
||||||
|
if w.bucket == bucket {
|
||||||
|
close(w.ch)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
kept = append(kept, w)
|
||||||
|
}
|
||||||
|
b.watches = kept
|
||||||
|
}
|
||||||
|
|
||||||
|
// missing makes a bucket's watch fail, as one the bus holds no state for, or work again.
|
||||||
|
func (b *patternBus) missing(bucket string, gone bool) {
|
||||||
|
b.mu.Lock()
|
||||||
|
defer b.mu.Unlock()
|
||||||
|
if b.absent == nil {
|
||||||
|
b.absent = map[string]bool{}
|
||||||
|
}
|
||||||
|
b.absent[bucket] = gone
|
||||||
|
}
|
||||||
|
|
||||||
|
type patternWatch struct {
|
||||||
|
bucket, pattern string
|
||||||
|
ch chan agent.Update
|
||||||
|
}
|
||||||
|
|
||||||
|
func newPatternBus() *patternBus { return &patternBus{} }
|
||||||
|
|
||||||
|
func (b *patternBus) watch(_ context.Context, bucket, pattern string) (<-chan agent.Update, error) {
|
||||||
|
b.mu.Lock()
|
||||||
|
defer b.mu.Unlock()
|
||||||
|
if b.absent[bucket] {
|
||||||
|
return nil, fmt.Errorf("nats: bucket not found")
|
||||||
|
}
|
||||||
|
w := patternWatch{bucket: bucket, pattern: pattern, ch: make(chan agent.Update, 64)}
|
||||||
|
b.watches = append(b.watches, w)
|
||||||
|
if b.replay {
|
||||||
|
for key, u := range b.kv[bucket] {
|
||||||
|
if matches(pattern, key) {
|
||||||
|
w.ch <- u
|
||||||
|
}
|
||||||
|
}
|
||||||
|
w.ch <- agent.Update{Caught: true}
|
||||||
|
}
|
||||||
|
return w.ch, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func matches(pattern, key string) bool {
|
||||||
|
p, k := strings.Split(pattern, "."), strings.Split(key, ".")
|
||||||
|
for i, tok := range p {
|
||||||
|
if tok == ">" {
|
||||||
|
return len(k) > i
|
||||||
|
}
|
||||||
|
if i >= len(k) || (tok != "*" && tok != k[i]) {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return len(p) == len(k)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (b *patternBus) send(bucket string, u agent.Update) {
|
||||||
|
b.mu.Lock()
|
||||||
|
defer b.mu.Unlock()
|
||||||
|
if !u.Caught {
|
||||||
|
if b.kv == nil {
|
||||||
|
b.kv = map[string]map[string]agent.Update{}
|
||||||
|
}
|
||||||
|
if b.kv[bucket] == nil {
|
||||||
|
b.kv[bucket] = map[string]agent.Update{}
|
||||||
|
}
|
||||||
|
if u.Deleted {
|
||||||
|
delete(b.kv[bucket], u.Key)
|
||||||
|
} else {
|
||||||
|
b.kv[bucket][u.Key] = u
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for _, w := range b.watches {
|
||||||
|
if w.bucket == bucket && (u.Caught || matches(w.pattern, u.Key)) {
|
||||||
|
select {
|
||||||
|
case w.ch <- u:
|
||||||
|
default: // a watch of a run that ended reads no more
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (b *patternBus) put(bucket, key string, value []byte, rev uint64) {
|
||||||
|
b.send(bucket, agent.Update{Key: key, Value: value, Revision: rev})
|
||||||
|
}
|
||||||
|
|
||||||
|
func (b *patternBus) del(bucket, key string) { b.send(bucket, agent.Update{Key: key, Deleted: true}) }
|
||||||
|
|
||||||
|
// made waits until n watches are made, so nothing put is lost.
|
||||||
|
func (b *patternBus) made(n int) {
|
||||||
|
for {
|
||||||
|
b.mu.Lock()
|
||||||
|
got := len(b.watches)
|
||||||
|
b.mu.Unlock()
|
||||||
|
if got >= n {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
time.Sleep(5 * time.Millisecond)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// caught says every watch caught up.
|
||||||
|
func (b *patternBus) caught() {
|
||||||
|
b.mu.Lock()
|
||||||
|
defer b.mu.Unlock()
|
||||||
|
for _, w := range b.watches {
|
||||||
|
w.ch <- agent.Update{Caught: true}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// fixedReader is frombus's Reader over values held, for composing what a node runs.
|
||||||
|
type fixedReader struct {
|
||||||
|
placements map[string]delivery.Placement
|
||||||
|
records map[string]delivery.NodeRecord
|
||||||
|
modules map[string]delivery.Module
|
||||||
|
manifests map[string][]byte
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r fixedReader) Placements(context.Context) (map[string]delivery.Placement, error) {
|
||||||
|
return r.placements, nil
|
||||||
|
}
|
||||||
|
func (r fixedReader) Records(context.Context) (map[string]delivery.NodeRecord, error) {
|
||||||
|
return r.records, nil
|
||||||
|
}
|
||||||
|
func (r fixedReader) Modules(context.Context) (map[string]delivery.Module, error) {
|
||||||
|
return r.modules, nil
|
||||||
|
}
|
||||||
|
func (r fixedReader) Manifest(_ context.Context, d string) ([]byte, error) {
|
||||||
|
if raw, ok := r.manifests[d]; ok {
|
||||||
|
return raw, nil
|
||||||
|
}
|
||||||
|
return nil, fmt.Errorf("no manifest %s", d)
|
||||||
|
}
|
||||||
|
func (r fixedReader) Values(context.Context, string, string) (map[string]any, error) { return nil, nil }
|
||||||
|
func (r fixedReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, error) {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
func (r fixedReader) Credentials(context.Context, string, []compose.CredentialKey, []compose.CredentialKey) ([]compose.Credential, []compose.Issued, error) {
|
||||||
|
return nil, nil, nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,358 @@
|
|||||||
|
package selfcompose
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"sort"
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
|
||||||
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
|
|
||||||
|
"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
|
||||||
|
// 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.
|
||||||
|
//
|
||||||
|
// **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
|
||||||
|
// 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
|
||||||
|
answers map[string]delivery.CandidateAnswer
|
||||||
|
}
|
||||||
|
|
||||||
|
// candidateKey is one candidate key as last read: its value and revision.
|
||||||
|
type candidateKey struct {
|
||||||
|
c delivery.PlacementCandidate
|
||||||
|
rev uint64
|
||||||
|
}
|
||||||
|
|
||||||
|
// 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 {
|
||||||
|
return &Shadow{c: New(env), answer: answer, candidates: map[string]map[string]candidateKey{},
|
||||||
|
answers: map[string]delivery.CandidateAnswer{}}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Keep runs the shadow until ctx ends, beginning again after backoff(try) whenever it stops: an input it cannot watch
|
||||||
|
// yet (the candidates' state not made, or the node's grant of it not loaded at a delivery) or a watch that ended is
|
||||||
|
// tried again rather than leaving the node answering nothing for the rest of the link (review of L2b, round 1, finding
|
||||||
|
// 2). Each distinct reason is said once.
|
||||||
|
func (s *Shadow) Keep(ctx context.Context, backoff func(try int) time.Duration) {
|
||||||
|
said := ""
|
||||||
|
for try := 1; ctx.Err() == nil; try++ {
|
||||||
|
began := time.Now()
|
||||||
|
err := s.Run(ctx)
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if time.Since(began) > time.Minute {
|
||||||
|
try = 1 // it ran: a later stop begins the waits again
|
||||||
|
}
|
||||||
|
why := "it stopped"
|
||||||
|
if err != nil {
|
||||||
|
why = err.Error()
|
||||||
|
}
|
||||||
|
if why != said {
|
||||||
|
s.c.env.Say("this node answers no placement candidate for now, and tries again: " + why)
|
||||||
|
said = why
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return
|
||||||
|
case <-time.After(backoff(try)):
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Run answers until ctx ends or an input's watch ends.
|
||||||
|
func (s *Shadow) Run(ctx context.Context) error {
|
||||||
|
// Each run watches afresh and starts from nothing read: what changed while it was down (a candidate key deleted)
|
||||||
|
// is then not missed, and the watches of a run before are not left open (review of L2b, round 2).
|
||||||
|
ctx, cancel := context.WithCancel(ctx)
|
||||||
|
defer cancel()
|
||||||
|
s.mu.Lock()
|
||||||
|
s.candidates = map[string]map[string]candidateKey{}
|
||||||
|
s.c = New(s.c.env)
|
||||||
|
s.mu.Unlock()
|
||||||
|
c := s.c
|
||||||
|
node := c.env.Node
|
||||||
|
ins := []input{
|
||||||
|
{delivery.PlacementCandidatesBucket, ">", nil},
|
||||||
|
{delivery.PlacementBucket, ">", (*Composer).takePlacement},
|
||||||
|
{delivery.HaveBucket, "*." + delivery.HaveNodeWord, nil},
|
||||||
|
{delivery.ModulesBucket, ">", (*Composer).takeModule},
|
||||||
|
{ResolvedBucket, node + ".>", (*Composer).takeValue},
|
||||||
|
// What this node runs moved: an apply ends in its module's have.
|
||||||
|
{delivery.HaveBucket, node + ".*", nil},
|
||||||
|
}
|
||||||
|
type tagged struct {
|
||||||
|
in int
|
||||||
|
u agent.Update
|
||||||
|
ok bool
|
||||||
|
}
|
||||||
|
merged := make(chan tagged, 64)
|
||||||
|
for i, in := range ins {
|
||||||
|
ch, err := c.env.Watch(ctx, in.bucket, in.pattern)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("the node answers placement candidates from %s, which cannot be watched: %w", in.bucket, err)
|
||||||
|
}
|
||||||
|
go func(i int, ch <-chan agent.Update) {
|
||||||
|
for u := range ch {
|
||||||
|
select {
|
||||||
|
case merged <- tagged{in: i, u: u, ok: true}:
|
||||||
|
case <-ctx.Done():
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case merged <- tagged{in: i}:
|
||||||
|
case <-ctx.Done():
|
||||||
|
}
|
||||||
|
}(i, ch)
|
||||||
|
}
|
||||||
|
caught := make([]bool, len(ins))
|
||||||
|
ready, dirty := false, false
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return nil
|
||||||
|
case t := <-merged:
|
||||||
|
if !t.ok {
|
||||||
|
return fmt.Errorf("the watch of %s ended; the node answers no placement candidate until its agent starts again", ins[t.in].bucket)
|
||||||
|
}
|
||||||
|
switch {
|
||||||
|
case t.u.Caught:
|
||||||
|
caught[t.in] = true
|
||||||
|
case t.in == 0:
|
||||||
|
dirty = s.takeCandidate(t.u) || dirty
|
||||||
|
case t.in == 2:
|
||||||
|
dirty = s.takeRecord(t.u) || dirty
|
||||||
|
case t.in == 5:
|
||||||
|
dirty = !isNodeKey(t.u.Key) || dirty
|
||||||
|
default:
|
||||||
|
ins[t.in].take(c, t.u)
|
||||||
|
dirty = true
|
||||||
|
}
|
||||||
|
if len(merged) > 0 {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if !ready {
|
||||||
|
if ready = all(caught); !ready {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
dirty = true
|
||||||
|
}
|
||||||
|
if dirty {
|
||||||
|
dirty = false
|
||||||
|
s.answerAll(ctx)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func isNodeKey(key string) bool {
|
||||||
|
_, _, isNode, err := delivery.ParseHaveKey(key)
|
||||||
|
return err == nil && isNode
|
||||||
|
}
|
||||||
|
|
||||||
|
// 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)
|
||||||
|
if err != nil {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
if u.Deleted {
|
||||||
|
if _, ok := s.candidates[group][node]; !ok {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
delete(s.candidates[group], node)
|
||||||
|
if len(s.candidates[group]) == 0 {
|
||||||
|
delete(s.candidates, group)
|
||||||
|
}
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
c, err := delivery.DecodePlacementCandidate(u.Value)
|
||||||
|
if err != nil || c.Group != group || (c.Placement != nil && c.Placement.Node != node) {
|
||||||
|
// Not one: read as gone.
|
||||||
|
if _, ok := s.candidates[group][node]; ok {
|
||||||
|
delete(s.candidates[group], node)
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
if s.candidates[group] == nil {
|
||||||
|
s.candidates[group] = map[string]candidateKey{}
|
||||||
|
}
|
||||||
|
s.candidates[group][node] = candidateKey{c: c, rev: u.Revision}
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
// takeRecord takes a node's record, and says whether it changed: the node's own key changes with each answer written,
|
||||||
|
// which moves nothing composed.
|
||||||
|
func (s *Shadow) takeRecord(u agent.Update) bool {
|
||||||
|
node, _, isNode, err := delivery.ParseHaveKey(u.Key)
|
||||||
|
if err != nil || !isNode {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
c := s.c
|
||||||
|
c.mu.Lock()
|
||||||
|
before, had := c.records[node]
|
||||||
|
c.mu.Unlock()
|
||||||
|
c.takeRecord(u)
|
||||||
|
c.mu.Lock()
|
||||||
|
after, has := c.records[node]
|
||||||
|
c.mu.Unlock()
|
||||||
|
return had != has || (has && !delivery.SameNodeRecord(before, after))
|
||||||
|
}
|
||||||
|
|
||||||
|
// answerAll answers every group and hands the answers when any changed.
|
||||||
|
func (s *Shadow) answerAll(ctx context.Context) {
|
||||||
|
c := s.c
|
||||||
|
node := c.env.Node
|
||||||
|
s.mu.Lock()
|
||||||
|
groups := make([]string, 0, len(s.candidates))
|
||||||
|
for g := range s.candidates {
|
||||||
|
groups = append(groups, g)
|
||||||
|
}
|
||||||
|
s.mu.Unlock()
|
||||||
|
sort.Strings(groups)
|
||||||
|
next := map[string]delivery.CandidateAnswer{}
|
||||||
|
for _, group := range groups {
|
||||||
|
if a, ok := s.answerGroup(ctx, group); ok {
|
||||||
|
next[group] = a
|
||||||
|
}
|
||||||
|
}
|
||||||
|
s.mu.Lock()
|
||||||
|
same := len(next) == len(s.answers)
|
||||||
|
for g, a := range next {
|
||||||
|
if old, ok := s.answers[g]; !ok || !sameAnswer(old, a) {
|
||||||
|
same = false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if !same {
|
||||||
|
s.answers = next
|
||||||
|
}
|
||||||
|
s.mu.Unlock()
|
||||||
|
if !same && s.answer != nil {
|
||||||
|
out := make(map[string]delivery.CandidateAnswer, len(next))
|
||||||
|
for g, a := range next {
|
||||||
|
out[g] = a
|
||||||
|
}
|
||||||
|
s.answer(ctx, out)
|
||||||
|
c.env.Say(fmt.Sprintf("%s answered %d placement candidate group(s)", node, len(out)))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 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) {
|
||||||
|
c := s.c
|
||||||
|
node := c.env.Node
|
||||||
|
s.mu.Lock()
|
||||||
|
keys := s.candidates[group]
|
||||||
|
given := make(map[string]*delivery.Placement, len(keys))
|
||||||
|
var head string
|
||||||
|
for n, k := range keys {
|
||||||
|
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).
|
||||||
|
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 {
|
||||||
|
return delivery.CandidateAnswer{}, false
|
||||||
|
}
|
||||||
|
|
||||||
|
var running []delivery.Composed
|
||||||
|
if c.env.Running != nil {
|
||||||
|
running = c.env.Running()
|
||||||
|
}
|
||||||
|
if p, ok := given[node]; ok && p == nil {
|
||||||
|
// The head removes this node's placement file: everything it runs is removed.
|
||||||
|
a := delivery.CandidateAnswer{Head: head, Revision: rev}
|
||||||
|
for _, r := range running {
|
||||||
|
if r.Module == "" {
|
||||||
|
a.Removed = append(a.Removed, delivery.HaveNodeWord)
|
||||||
|
} else {
|
||||||
|
a.Removed = append(a.Removed, r.Module)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
sort.Strings(a.Removed)
|
||||||
|
a.Verdict = delivery.VerdictOf(a)
|
||||||
|
return a, true
|
||||||
|
}
|
||||||
|
|
||||||
|
// Each module at the build it runs; one the candidate adds at the trunk's build.
|
||||||
|
fps := map[string]string{}
|
||||||
|
kept := map[string]string{}
|
||||||
|
for _, r := range running {
|
||||||
|
if r.Module != "" && r.FP != "" {
|
||||||
|
fps[r.Module] = r.FP
|
||||||
|
}
|
||||||
|
}
|
||||||
|
placement := given[node]
|
||||||
|
if placement == nil {
|
||||||
|
c.mu.Lock()
|
||||||
|
p := c.placements[node]
|
||||||
|
c.mu.Unlock()
|
||||||
|
placement = &p
|
||||||
|
}
|
||||||
|
for _, m := range placement.Modules {
|
||||||
|
if _, runs := fps[m.Module]; !runs {
|
||||||
|
if mod, ok := modules[m.Module]; ok && mod.FP2 != "" {
|
||||||
|
fps[m.Module] = mod.FP2
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for module, fp := range fps {
|
||||||
|
if mod, ok := modules[module]; ok && mod.FP2 == fp {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
digest, err := c.builtManifest(ctx, module, fp)
|
||||||
|
if err != nil {
|
||||||
|
return delivery.CandidateAnswer{Head: head, Revision: rev, Verdict: delivery.CandidateRefused,
|
||||||
|
Refused: fmt.Sprintf("%s runs at %s, whose manifest cannot be read: %v", module, fp, err)}, true
|
||||||
|
}
|
||||||
|
kept[module] = digest
|
||||||
|
}
|
||||||
|
// The boxes it runs, by what each module's have and its own key say after a restart (Held), are reused for the
|
||||||
|
// modules it runs: a value sealed again is no difference.
|
||||||
|
held := map[string]delivery.Want{agent.NodeModule: {}}
|
||||||
|
for module := range fps {
|
||||||
|
held[module] = delivery.Want{FP: fps[module]}
|
||||||
|
}
|
||||||
|
c.mu.Lock()
|
||||||
|
c.wants = held
|
||||||
|
c.mu.Unlock()
|
||||||
|
a := frombus.Answer(ctx, reader{c: c}, c.env.Mesh, frombus.Scope{Node: node, Kept: kept, Placements: given}, fps, running, head, rev)
|
||||||
|
return a, true
|
||||||
|
}
|
||||||
|
|
||||||
|
func sameAnswer(a, b delivery.CandidateAnswer) bool {
|
||||||
|
x, errA := delivery.EncodeHaveNode(delivery.HaveNode{Candidates: map[string]delivery.CandidateAnswer{"g": a}})
|
||||||
|
y, errB := delivery.EncodeHaveNode(delivery.HaveNode{Candidates: map[string]delivery.CandidateAnswer{"g": b}})
|
||||||
|
return errA == nil && errB == nil && string(x) == string(y)
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user