diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 8368dd3..09929b3 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -62,14 +62,14 @@ func (h sessionHaves) Read(ctx context.Context, module string) (agent.Have, bool if !ok || got.Composed == "" { return agent.Have{}, false } - return agent.Have{Module: module, Composed: got.Composed, Sealed: got.Sealed}, true + return agent.Have{Module: module, Composed: got.Composed, Sealed: got.SealedAt}, true } got, ok := h.session.ReadHave(ctx, module) if !ok { return agent.Have{}, false } return agent.Have{Module: module, FP: got.FP, Composed: got.Composed, Phase: string(got.Phase), Detail: got.Detail, - Sealed: got.Sealed}, true + Sealed: got.SealedAt}, true } // ReadErr is Read telling a have absent from one the bus could not read (agent.HavesReadErr). @@ -83,7 +83,7 @@ func (h sessionHaves) ReadErr(ctx context.Context, module string) (agent.Have, b return agent.Have{}, false, err } return agent.Have{Module: module, FP: got.FP, Composed: got.Composed, Phase: string(got.Phase), Detail: got.Detail, - Sealed: got.Sealed}, true, nil + Sealed: got.SealedAt}, true, nil } func (h sessionHaves) Write(ctx context.Context, x agent.Have) { @@ -535,16 +535,16 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess }, unseal: sealOpener(opts.state), // 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) (delivery.Seals, []string) { if module == agent.NodeModule { got, _ := session.ReadNodeHave(linked) - return got.Sealed + return got.SealedAt, got.Sealed } got, ok := session.ReadHave(linked, module) if !ok { - return nil + return nil, nil } - return got.Sealed + return got.SealedAt, got.Sealed }, refused: func(why string) { if now := own.get(); now != nil && run.Err() == nil { @@ -556,9 +556,15 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess // its own key, `.node`, under candidates. shadowHooks := hooks shadowHooks.refused = nil - shadow := selfcompose.NewShadow(composerEnv(session, node, say, shadowHooks), + shadowEnv := composerEnv(session, node, say, shadowHooks) + // And, after each answer, what runs of an own secret the answer says the vault holds another value of, handed to + // the vault while the node applies the controller's compositions (issue 1126, cause C4; ownresync). + composes := &composesNow{} + resync := ownResync(session, node, say, shadowEnv, composes) + shadow := selfcompose.NewShadow(shadowEnv, func(ctx context.Context, answers map[string]delivery.CandidateAnswer) { - session.NoteCandidates(ctx, answers) + session.NoteCandidates(ctx, delivery.WithValuesDiffer(answers)) + go resync.take(ctx, answers) }) go shadow.Keep(run, link.Backoff) // this run's alone: a switch begins a run, and its shadow, again composer, why := selfComposer(linked, session, node, say, hooks) @@ -569,12 +575,15 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess continue } if composer != nil { + composes.set(delivery.ComposesOwn) env.Composed, env.SelfComposed, env.Trusted, env.Policy = composer, true, nil, composer.Policy session.NoteComposes(linked, delivery.ComposesOwn, why) // And once the catalogue no longer places it, the agent begins again on the controller's compositions: the // placement key deleted ends this run, the way back (issue 1126, the seed's revert). go endWhenUnplaced(run, session, node, stopRun, say) } else { + composes.set(delivery.ComposesController) + go resync.again(run) session.NoteComposes(linked, delivery.ComposesController, why) // And once the catalogue places it, the agent begins again composing its own: the placement key appearing ends // this run, so a node does not wait for a restart to follow the seed of its placement (ADR 0549 §4). @@ -870,7 +879,7 @@ type composerHooks struct { running func() []delivery.Composed unseal func(string) ([]byte, error) refused func(string) - held func(module string) []string + held func(module string) (delivery.Seals, []string) } func selfComposer(ctx context.Context, session *link.Session, node string, say func(string), hooks composerHooks) (*selfcompose.Composer, string) { diff --git a/cmd/mesh-host/ownresync.go b/cmd/mesh-host/ownresync.go new file mode 100644 index 0000000..219c74c --- /dev/null +++ b/cmd/mesh-host/ownresync.go @@ -0,0 +1,61 @@ +package main + +import ( + "context" + "encoding/json" + "sync" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "github.com/novox/mesh-host/internal/identity" + "github.com/novox/mesh-host/internal/link" + "github.com/novox/mesh-host/internal/ownresync" + "github.com/novox/mesh-host/internal/selfcompose" +) + +// composesNow is whose compositions this run of the agent applies, once decided ("" before). +type composesNow struct { + mu sync.Mutex + now string +} + +func (c *composesNow) set(s string) { c.mu.Lock(); c.now = s; c.mu.Unlock() } +func (c *composesNow) get() string { c.mu.Lock(); defer c.mu.Unlock(); return c.now } + +// resyncStep is the step that hands the vault what runs of an own secret a shadow answer says another value of (issue +// 1126, cause C4; internal/ownresync): one pass at a time, and the last answers taken again once the agent has decided +// whose compositions it applies. +type resyncStep struct { + r *ownresync.Resync + one sync.Mutex + mu sync.Mutex + last map[string]delivery.CandidateAnswer +} + +func ownResync(session *link.Session, node string, say func(string), env selfcompose.Env, composes *composesNow) *resyncStep { + return &resyncStep{r: ownresync.New(ownresync.Env{Node: node, Composes: composes.get, Running: env.Running, + Get: env.Get, Unseal: env.Unseal, Seal: identity.Seal, Say: say, + Ask: func(ctx context.Context, verb string, args map[string]any) (json.RawMessage, error) { + return session.AskVault(ctx, verb, args) + }})} +} + +// take keeps answers as the last and takes them. +func (s *resyncStep) take(ctx context.Context, answers map[string]delivery.CandidateAnswer) { + s.mu.Lock() + s.last = answers + s.mu.Unlock() + s.again(ctx) +} + +// again takes the last answers. +func (s *resyncStep) again(ctx context.Context) { + s.one.Lock() + defer s.one.Unlock() + s.mu.Lock() + last := s.last + s.mu.Unlock() + if last != nil { + s.r.Take(ctx, last) + } +} diff --git a/go.mod b/go.mod index 1d9e140..7f2612f 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/novox/mesh-host go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028003907-362a3f1378aa + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028090029-3028c7ec4ed2 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 golang.org/x/crypto v0.57.0 diff --git a/go.sum b/go.sum index 55fd674..a460d12 100644 --- a/go.sum +++ b/go.sum @@ -1,35 +1,5 @@ -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.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= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027075026-05b678310366 h1:B7oQ82ar2McqZ9OrbJfyvG+50Ol6fBUoK3eYE2uTPvo= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027075026-05b678310366/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027195758-cebc6519d5b0 h1:9B7B38e3RtjbtX1uGP9dRgU9VY1/J2yAfnb5Ajoq4/s= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027195758-cebc6519d5b0/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027200317-538ffbaa085b h1:LuBK5dtBgm978krhj41N2MKA07Yv4zetRPOdHRTB6L8= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027200317-538ffbaa085b/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027201213-7a8e9f970648 h1:XoMsgg9kSgjQWa86NZM8h8OpxJ2kovxcMS2C0i6WSVU= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027201213-7a8e9f970648/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027221705-2fd5ffdd6263 h1:JZV8C+WR6NPiopj9ShzhbpjN0gqVQV60vOFxsxFwH00= -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/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= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027232145-7359d187db8d h1:DhR7DRKaF0OtA5l/ikIk7gZGeIM1lRz65PUE1DGEkvU= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027232145-7359d187db8d/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028000902-4e0c57771e8c h1:4lS82cVTtxls17bRn3551+m2xlZO+0WEtv0a7iME8Ec= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028000902-4e0c57771e8c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028003907-362a3f1378aa h1:DFYX7pIluzeeEyM+YXGeB6GvEnOYyRhz/1Ke/LFa8+8= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028003907-362a3f1378aa/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028090029-3028c7ec4ed2 h1:eSwObLdxs3jxPrlcwdaMo9hzIDTUz2ogWw8cjX8To7I= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028090029-3028c7ec4ed2/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/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 7c70a47..e802476 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -57,9 +57,10 @@ type Have struct { // waits for a person's yes (ADR 0547 §3). Judgement string Waits string - // Sealed is, on a node composing its own, each value the composition it names carries sealed to the node's key - // (delivery.Have.Sealed): kept in the have so the composer reuses them after a restart (ADR 0547 §3). - Sealed []string + // Sealed is, on a node composing its own, each value the composition it names carries sealed to the node's key, by + // slot (delivery.Have.SealedAt): kept in the have so the composer reuses them after a restart (ADR 0547 §3), each at + // its own slot alone (issue 1126, C1). + Sealed delivery.Seals } // Haves is the node's have bucket as the agent uses it. @@ -471,15 +472,15 @@ func (a *Agent) writeHave(ctx context.Context, h Have) { a.env.Haves.Write(ctx, h) } -func (a *Agent) sealedOf(ctx context.Context, module, fp, composed string) []string { +func (a *Agent) sealedOf(ctx context.Context, module, fp, composed string) delivery.Seals { a.mu.Lock() if c, ok := a.applied[module]; ok && c.Err == nil && c.Dig == composed { a.mu.Unlock() - return SealedIn(c.C) + return delivery.SealsIn(c.C) } if c, ok := a.compositionFor(module, fp); ok && c.Err == nil && c.Dig == composed { a.mu.Unlock() - return SealedIn(c.C) + return delivery.SealsIn(c.C) } a.mu.Unlock() if h, ok := a.env.Haves.Read(ctx, module); ok && h.Composed == composed { @@ -488,32 +489,6 @@ func (a *Agent) sealedOf(ctx context.Context, module, fp, composed string) []str return nil } -// SealedIn is each value a composition carries sealed to its node's key: a file's `sealed` and a vault box -// (catalogue.VaultSealedField), as their JSON text, in the order of its resources. -func SealedIn(c delivery.Composed) []string { - var out []string - for _, raw := range c.Resources { - var r map[string]json.RawMessage - if json.Unmarshal(raw, &r) != nil { - continue - } - for _, field := range []string{"sealed", catalogue.VaultSealedField} { - f, ok := r[field] - if !ok { - continue - } - var text string - if json.Unmarshal(f, &text) != nil { - text = string(f) // a vault box is carried as its JSON - } - if text != "" { - out = append(out, text) - } - } - } - return out -} - // Running is the compositions the node runs now: each module's that the agent applied or found already running, and // the node's own. What the node judges its health by and says of itself is built from them. func (a *Agent) Running() []delivery.Composed { diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index c3d4a56..115befc 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -5,6 +5,7 @@ import ( "crypto/ed25519" "encoding/json" "fmt" + "maps" "slices" "strings" "sync" @@ -1064,30 +1065,30 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes( 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" + return h.Composed == ndig && len(h.Sealed) == 1 && h.Sealed["core-box#sealed"] == "core" }) // 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 // agent's start (the switch and every delivery of the node-engine restart it), so what runs is what its have said. idleFP, idleDig := fpOf("c"), "sha256:"+strings.Repeat("9", 64) b.haves.Write(context.Background(), Have{Module: "idle", FP: idleFP, Composed: idleDig, Phase: string(delivery.PhaseRunning), - Sealed: []string{`{"box":"idle"}`}}) + Sealed: delivery.Seals{"idle.key#sealed": `{"box":"idle"}`}}) want("idle", idleFP, "") eventually(t, "idle's want was not taken", func() bool { b.ag.mu.Lock() defer b.ag.mu.Unlock() return b.ag.wants["idle"].Present }) - if h := b.haves.last("web"); len(h.Sealed) != 1 || h.Sealed[0] != "1" { + if h := b.haves.last("web"); len(h.Sealed) != 1 || h.Sealed["web.pw#sealed"] != "1" { t.Errorf("web's running have does not carry its sealed value: %+v", h) } b.ag.NotComposed(context.Background(), "a stand-in for web's password") for _, row := range []struct { - module, fp, composed, sealed string - }{{"web", fp, dig, "1"}, {"idle", idleFP, idleDig, `{"box":"idle"}`}} { + module, fp, composed, slot, sealed string + }{{"web", fp, dig, "web.pw#sealed", "1"}, {"idle", idleFP, idleDig, "idle.key#sealed", `{"box":"idle"}`}} { h := b.haves.last(row.module) if h.Phase != string(delivery.PhaseFailed) || h.FP != row.fp || h.Composed != row.composed || - len(h.Sealed) != 1 || h.Sealed[0] != row.sealed || !strings.Contains(h.Detail, "web's password") { + len(h.Sealed) != 1 || h.Sealed[row.slot] != row.sealed || !strings.Contains(h.Detail, "web's password") { t.Errorf("%s's have while the node cannot compose: %+v", row.module, h) } } @@ -1155,7 +1156,7 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes( // what runs (fingerprint, digest, sealed values) rather than naming nothing (review of L2, round 3). recFP, recDig := fpOf("d"), "sha256:"+strings.Repeat("8", 64) b.haves.Write(context.Background(), Have{Module: "rec", FP: recFP, Composed: recDig, Phase: string(delivery.PhaseRunning), - Sealed: []string{`{"box":"rec"}`}}) + Sealed: delivery.Seals{"rec.key#sealed": `{"box":"rec"}`}}) rk, rraw, rdig := unsigned("rec", fpB, container("rec@sha256:"+strings.Repeat("3", 64))) b.comps.put(dk2, draw2) b.comps.put(rk, rraw) @@ -1164,15 +1165,15 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes( applied := len(b.appliedList()) for _, row := range []struct { module, fp, composed, waits string - sealed []string - }{{"db", fp, "", ddig2, nil}, {"rec", recFP, recDig, rdig, []string{`{"box":"rec"}`}}} { + sealed delivery.Seals + }{{"db", fp, "", ddig2, nil}, {"rec", recFP, recDig, rdig, delivery.Seals{"rec.key#sealed": `{"box":"rec"}`}}} { eventually(t, row.module+"'s move did not wait for a yes", func() bool { h := b.haves.last(row.module) return h.Phase == string(delivery.PhaseWaitingYes) && h.Waits == row.waits }) h := b.haves.last(row.module) if h.FP != row.fp || (row.composed != "" && h.Composed != row.composed) || h.Composed == "" || - (row.sealed != nil && !slices.Equal(h.Sealed, row.sealed)) { + (row.sealed != nil && !maps.Equal(h.Sealed, row.sealed)) { t.Errorf("%s's have waiting for a yes does not keep what runs: %+v", row.module, h) } } diff --git a/internal/link/have.go b/internal/link/have.go index 82c41a9..d6df2f0 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -49,7 +49,7 @@ type haveState struct { 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 + nodeSealed delivery.Seals // 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 // nodeComposes and nodeComposesWhy are whose compositions the node applies and why (HaveNode.Composes), written into @@ -383,7 +383,7 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) { case f.Failed: h.Phase = delivery.PhaseFailed } - h.Sealed = f.Sealed + h.SealedAt = f.Sealed h.Health = s.healthOf(module, f) if h.Detail == "" { h.Detail = s.wordsOf(module, h.Health) @@ -424,7 +424,7 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) { var raw []byte facts := *s.node facts.Composed = s.nodeComposed - facts.Sealed = s.nodeSealed + facts.SealedAt = s.nodeSealed facts.Candidates = s.nodeCandidates facts.Composes, facts.ComposesWhy = s.nodeComposes, s.nodeComposesWhy if raw, err = delivery.EncodeHaveNode(facts); err == nil { @@ -716,7 +716,7 @@ func (q *Session) NoteRemoved(ctx context.Context, module, build string) { // NoteNode takes the digest of the node's own composition it runs and the values it carries sealed to the node's key // (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) { +func (q *Session) NoteNode(ctx context.Context, composed string, sealed delivery.Seals) { q.init() s := &q.haves s.mu.Lock() diff --git a/internal/link/have_test.go b/internal/link/have_test.go index 309fb4d..04ed949 100644 --- a/internal/link/have_test.go +++ b/internal/link/have_test.go @@ -582,11 +582,11 @@ func TestAnAppliedModulesHaveNamesTheFingerprintAndTheDigestAndFollowsItsHealth( if got := l.have(t, "n1.web"); got.Phase != delivery.PhaseRemoved { t.Fatalf("removed: %+v", got) } - q.NoteNode(context.Background(), dig, []string{"core"}) - 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, with its own composition's sealed values: %+v %v", n, err) + q.NoteNode(context.Background(), dig, delivery.Seals{"core-box#sealed": "core"}) + if n, err := delivery.DecodeHaveNode(l.kv["n1.node"].value); err != nil || n.Composed != dig || len(n.SealedAt) != 1 || n.SealedAt["core-box#sealed"] != "core" { + t.Fatalf("the node key, with its own composition's sealed values by slot: %+v %v", n, err) } - if n, ok := q.ReadNodeHave(context.Background()); !ok || n.Composed != dig || len(n.Sealed) != 1 { + if n, ok := q.ReadNodeHave(context.Background()); !ok || n.Composed != dig || len(n.SealedAt) != 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. diff --git a/internal/link/messages.go b/internal/link/messages.go index 36c3421..2582562 100644 --- a/internal/link/messages.go +++ b/internal/link/messages.go @@ -5,6 +5,7 @@ import ( "strings" "time" + "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/health" ) @@ -746,6 +747,6 @@ type ModuleFact struct { // Judgement and Waits are, in phase waiting-yes, the judgement of the move and the composition waiting (ADR 0547 §3). Judgement string Waits string - // Sealed is the values the composition it names carries sealed to this node's key (delivery.Have.Sealed). - Sealed []string + // Sealed is the values the composition it names carries sealed to this node's key, by slot (delivery.Have.SealedAt). + Sealed delivery.Seals } diff --git a/internal/ownresync/ownresync.go b/internal/ownresync/ownresync.go new file mode 100644 index 0000000..02599a3 --- /dev/null +++ b/internal/ownresync/ownresync.go @@ -0,0 +1,296 @@ +// Package ownresync hands the vault the value of this machine's own secret that runs, where the vault holds another +// (novox/hq issue 1126, cause C4). +// +// The vault's `move-in` never replaces a value it holds, so its first import of an own secret stays fixed; where the +// controller later made that secret anew (a rotation at start, a re-issue), the new value runs and the vault keeps the +// old one. The node's shadow composer sees it: composing from the vault's value, it opens both seals with this +// machine's key and its answer's shown body says delivery.MaskedAnotherValue. After each shadow answer this step asks the +// vault's `move-in` with replace_running: true and what runs, for each own secret so masked, once per running value, and +// says what it asked and what the vault answered. It is move-in's, the verb this node may ask already, so no seat and no +// bus grant changes for it. +// +// **Only while this node applies the controller's compositions**, and only for a module whose composition that runs is +// the one its want names: then the controller's value is what runs, and what runs is the truth the vault lacks. Once the +// node composes its own, the vault's value is the source and nothing is handed. **The switch** is the vault's +// (`own_take_switch`, a person's act): while it is off the vault refuses every ask, and this step says so once per value. +// The vault also refuses any value it changed in the last 15 minutes, so a re-issue the node has not applied yet is +// never put back. +package ownresync + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "sort" + "sync" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// Verb is the vault's verb this step asks, with ReplaceRunning true in both steps. +const Verb = "move-in" + +// ReplaceRunning is move-in's argument that makes it take the running value over the one the vault holds. +const ReplaceRunning = "replace_running" + +// Env is what the step reads and asks. +type Env struct { + Node string + // Composes is whose compositions the node applies now: delivery.ComposesController, delivery.ComposesOwn, or "" + // while its agent has not decided. The step hands nothing but for ComposesController. + Composes func() string + // Running is what the node runs now (agent.Everything). + Running func() []delivery.Composed + // Get reads one key's latest value; nil and no error when there is none. + Get func(ctx context.Context, bucket, key string) ([]byte, error) + // Unseal opens a value sealed to this machine; Seal seals a value to a raw X25519 key in base64 (identity.Seal). + Unseal func(sealed string) ([]byte, error) + Seal func(to string, value []byte) (string, error) + // Ask asks one of the vault's verbs in this engine's own name (link.Session.AskVault). + Ask func(ctx context.Context, verb string, args map[string]any) (json.RawMessage, error) + Say func(string) +} + +// Outcome is what the step did for one own secret: asked or not, and what was said. +type Outcome struct { + Module string + Name string + Asked bool + Said string +} + +// Resync is the step of one node agent's run. +type Resync struct { + env Env + + mu sync.Mutex + // done is, by module/name, the running value's fingerprint the vault was asked with or why it was not: each is + // asked, and each reason said, once. + done map[string]string +} + +// New is the step for one node. +func New(env Env) *Resync { + if env.Say == nil { + env.Say = func(string) {} + } + return &Resync{env: env, done: map[string]string{}} +} + +// Take reads the node's shadow answers and hands the vault what runs of each own secret they say another value of. +func (r *Resync) Take(ctx context.Context, answers map[string]delivery.CandidateAnswer) []Outcome { + seen := map[delivery.OwnValueDiffers]bool{} + var drift []delivery.OwnValueDiffers + groups := make([]string, 0, len(answers)) + for g := range answers { + groups = append(groups, g) + } + sort.Strings(groups) + for _, g := range groups { + for _, o := range delivery.OwnValuesDiffer(answers[g]) { + if !seen[o] { + seen[o] = true + drift = append(drift, o) + } + } + } + if len(drift) == 0 { + return nil + } + composes := "" + if r.env.Composes != nil { + composes = r.env.Composes() + } + var out []Outcome + for _, o := range drift { + res := r.one(ctx, composes, o) + if res.Said != "" { + out = append(out, res) + } + } + return out +} + +// one is the step for one own secret: Said is empty when it was done before for the same value and reason. +func (r *Resync) one(ctx context.Context, composes string, o delivery.OwnValueDiffers) Outcome { + res := Outcome{Module: o.Module, Name: o.Name} + key := o.Module + "/" + o.Name + if composes == "" { + return res // the agent has not decided whose compositions it applies: taken again once it has + } + if composes != delivery.ComposesController { + return r.said(key, "not:composes", res, fmt.Sprintf("%s holds another value in the vault than runs here, and "+ + "nothing is handed: this node composes its own (%q), so the vault's value is the source", key, composes)) + } + sealed, why := r.running(ctx, o) + if why != "" { + return r.said(key, "not:"+why, res, fmt.Sprintf("%s holds another value in the vault than runs here, and "+ + "nothing is handed: %s", key, why)) + } + sum := sha256.Sum256([]byte(sealed)) + fp := hex.EncodeToString(sum[:]) + r.mu.Lock() + asked := r.done[key] == fp + r.mu.Unlock() + if asked { + return res + } + value, err := r.env.Unseal(sealed) + if err != nil { + return r.said(key, "not:unseal:"+fp, res, fmt.Sprintf("%s: what runs does not open with this machine's key, so "+ + "nothing is handed: %v", key, err)) + } + defer wipe(value) + said, err := r.hand(ctx, o, value) + if err != nil { + if ctx.Err() != nil { + return res // the run ended: asked again in the next + } + said = fmt.Sprintf("%s: the vault's %s with replace_running refused or did not answer: %v", key, Verb, err) + } + // Once per running value, whatever the vault answered: a refusal is the vault's word, said once, and a new value + // that runs asks again. + r.mu.Lock() + r.done[key] = fp + r.mu.Unlock() + res.Asked, res.Said = true, said + r.env.Say(said) + return res +} + +// said says why once per reason, and keeps the reason as done. +func (r *Resync) said(key, reason string, res Outcome, why string) Outcome { + r.mu.Lock() + again := r.done[key] == reason + r.done[key] = reason + r.mu.Unlock() + if again { + return res + } + res.Said = why + r.env.Say(why) + return res +} + +// running is the sealed value of o in the composition that runs, when that composition is the one the module's want +// names; or why it is not handed. +func (r *Resync) running(ctx context.Context, o delivery.OwnValueDiffers) (string, string) { + var comp *delivery.Composed + if r.env.Running != nil { + for _, c := range r.env.Running() { + if c.Module == o.Module { + c := c + comp = &c + break + } + } + } + if comp == nil { + return "", o.Module + " runs no composition here" + } + wantKey, err := delivery.WantKey(r.env.Node, o.Module) + if err != nil { + return "", err.Error() + } + raw, err := r.env.Get(ctx, delivery.AssignerWantsBucket, wantKey) + if err != nil || raw == nil { + return "", fmt.Sprintf("its want %s could not be read (%v)", wantKey, err) + } + want, err := delivery.DecodeWant(raw) + if err != nil { + return "", fmt.Sprintf("its want %s is not one: %v", wantKey, err) + } + if want.Self || want.Removed || want.Composed == "" { + return "", fmt.Sprintf("its want %s names no composition of the controller's", wantKey) + } + // What runs is the composition the want names, as the module's have says: applied and running. + haveKey, err := delivery.HaveKey(r.env.Node, o.Module) + if err != nil { + return "", err.Error() + } + raw, err = r.env.Get(ctx, delivery.HaveBucket, haveKey) + if err != nil || raw == nil { + return "", fmt.Sprintf("its have %s could not be read (%v)", haveKey, err) + } + have, err := delivery.DecodeHave(raw) + if err != nil { + return "", fmt.Sprintf("its have %s is not one: %v", haveKey, err) + } + if have.Composed != want.Composed || have.Phase != delivery.PhaseRunning { + return "", fmt.Sprintf("what runs (%s, %s) is not yet the composition its want names (%s)", have.Composed, + have.Phase, want.Composed) + } + // On a node the resource's id carries its module's name (`only-office.needs-jwt`); delivery.OwnNeedName reads it. + id := o.Module + "." + catalogue.NeedID(o.Name) + for _, res := range comp.Resources { + var body struct { + ID string `json:"id"` + Sealed any `json:"sealed"` + } + if json.Unmarshal(res, &body) != nil { + continue + } + if name, own := delivery.OwnNeedName(o.Module, body.ID); !own || name != o.Name { + continue + } + if s, ok := body.Sealed.(string); ok && s != "" { + return s, "" + } + return "", fmt.Sprintf("its resource %s carries no value sealed to this machine", id) + } + return "", fmt.Sprintf("what runs holds no resource %s", id) +} + +// hand asks the vault's two steps with value, and checks what the vault holds after by its fingerprint. +func (r *Resync) hand(ctx context.Context, o delivery.OwnValueDiffers, value []byte) (string, error) { + raw, err := r.env.Ask(ctx, Verb, map[string]any{ReplaceRunning: true, "module": o.Module, "name": o.Name, + "why": "this node's shadow answer says the vault holds another value than runs here"}) + if err != nil { + return "", err + } + var opened struct { + Call string `json:"call"` + SealTo string `json:"seal_to"` + Secret string `json:"secret"` + } + if err := json.Unmarshal(raw, &opened); err != nil || opened.Call == "" || opened.SealTo == "" { + return "", fmt.Errorf("the vault answered no call and no key to seal to") + } + sealed, err := r.env.Seal(opened.SealTo, value) + if err != nil { + return "", fmt.Errorf("the vault's key for the call is not a sealing key: %w", err) + } + raw, err = r.env.Ask(ctx, Verb, map[string]any{ReplaceRunning: true, "call": opened.Call, "sealed": sealed}) + if err != nil { + return "", err + } + var kept struct { + Secret string `json:"secret"` + From int `json:"from_generation"` + To int `json:"to_generation"` + Unchanged bool `json:"unchanged"` + Fingerprint string `json:"fingerprint"` + Said string `json:"said"` + } + if err := json.Unmarshal(raw, &kept); err != nil || kept.Secret == "" { + return "", fmt.Errorf("the vault answered %s with no secret", opened.Secret) + } + sum := sha256.Sum256(value) + if kept.Fingerprint != "sha256:"+hex.EncodeToString(sum[:]) { + return "", fmt.Errorf("the vault says it holds %s at generation %d, but not the value handed in", kept.Secret, kept.To) + } + if kept.Unchanged { + return fmt.Sprintf("handed the vault what runs of %s: it held that value already (generation %d)", kept.Secret, kept.To), nil + } + return fmt.Sprintf("handed the vault what runs of %s: generation %d replaced by %d, the value that runs", kept.Secret, + kept.From, kept.To), nil +} + +func wipe(b []byte) { + for i := range b { + b[i] = 0 + } +} diff --git a/internal/ownresync/ownresync_test.go b/internal/ownresync/ownresync_test.go new file mode 100644 index 0000000..2ec1235 --- /dev/null +++ b/internal/ownresync/ownresync_test.go @@ -0,0 +1,104 @@ +package ownresync + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// Issue 1126, cause C4: after a shadow answer saying the vault holds another value of an own secret than runs, the node +// hands the vault what runs, in the verb's two steps, once per running value; and only while it applies the +// controller's compositions and runs the one its want names. +func TestTheNodeHandsTheVaultWhatRunsOfADriftedOwnSecret(t *testing.T) { + const running = "the-value-that-runs" + digest := "sha256:" + strings.Repeat("a", 64) + fp := "src2:" + strings.Repeat("b", 64) + sum := sha256.Sum256([]byte(running)) + answers := map[string]delivery.CandidateAnswer{"g": {Head: "h", Verdict: delivery.CandidateDiffers, + Differs: []delivery.ModuleDiffers{{Module: "only-office", Why: "the resource differs", Bodies: []delivery.ResourceBodies{ + {ID: "only-office.needs-jwt", Composed: `{"sealed":"` + delivery.MaskedAnotherValue + `"}`, Running: `{"sealed":"` + delivery.MaskedAnotherValue + `"}`}, + {ID: "only-office.needs-salt", Composed: `{"sealed":""}`}}}}}} + for name, row := range map[string]struct { + composes string + runs string // the have's composition digest + refuse string // the vault's refusal + asks int // the vault's verb asked, both passes + says string + }{ + "applies the controller's, runs its want": {delivery.ComposesController, digest, "", 2, "generation 1 replaced by 2"}, + "composes its own": {delivery.ComposesOwn, digest, "", 0, "composes its own"}, + "not decided yet": {"", digest, "", 0, ""}, + "runs not yet what its want names": {delivery.ComposesController, "sha256:" + strings.Repeat("c", 64), "", 0, "not yet the composition"}, + "the vault refuses": {delivery.ComposesController, digest, "switched off", 1, "switched off"}, + } { + var asked []map[string]any + var said []string + want, _ := delivery.EncodeWant(delivery.Want{FP: fp, Composed: digest}) + have, _ := delivery.EncodeHave(delivery.Have{FP: fp, Composed: row.runs, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}) + r := New(Env{Node: "novox", Composes: func() string { return row.composes }, + Running: func() []delivery.Composed { + return []delivery.Composed{{Node: "novox", Module: "only-office", FP: fp, Resources: []json.RawMessage{ + // As novox runs it (its shadow answer, 2026-10-28): the id carries the module's name. + json.RawMessage(`{"id":"only-office.needs-jwt","module":"only-office","type":"file",` + + `"path":"/var/lib/only-office/jwt.secret","sealed":"box-of-the-running-value"}`)}}} + }, + Get: func(_ context.Context, bucket, key string) ([]byte, error) { + switch { + case bucket == delivery.AssignerWantsBucket && key == "want.novox.only-office": + return want, nil + case bucket == delivery.HaveBucket && key == "novox.only-office": + return have, nil + } + return nil, nil + }, + Unseal: func(sealed string) ([]byte, error) { + if sealed != "box-of-the-running-value" { + return nil, errors.New("not this machine's") + } + return []byte(running), nil + }, + Seal: func(to string, value []byte) (string, error) { return to + "(" + string(value) + ")", nil }, + Ask: func(_ context.Context, verb string, args map[string]any) (json.RawMessage, error) { + asked = append(asked, args) + if verb != "move-in" || args["replace_running"] != true { + t.Errorf("%s: asked %s with %v, not move-in with replace_running", name, verb, args) + } + if row.refuse != "" { + return nil, errors.New(row.refuse) + } + if args["call"] == nil { + if args["module"] != "only-office" || args["name"] != "jwt" || args["node"] != nil { + t.Errorf("%s: the first step names %v", name, args) + } + return json.RawMessage(`{"call":"c1","seal_to":"key","secret":"own/novox/only-office/jwt"}`), nil + } + if args["sealed"] != "key("+running+")" { + t.Errorf("%s: handed %v, not what runs sealed to the call's key", name, args["sealed"]) + } + return json.RawMessage(`{"secret":"own/novox/only-office/jwt","from_generation":1,"to_generation":2,` + + `"fingerprint":"sha256:` + hex.EncodeToString(sum[:]) + `"}`), nil + }, + Say: func(s string) { said = append(said, s) }}) + first := r.Take(context.Background(), answers) + again := r.Take(context.Background(), answers) + if len(asked) != row.asks { + t.Errorf("%s: the vault asked %d times, want %d: %v", name, len(asked), row.asks, asked) + } + if len(again) != 0 { + t.Errorf("%s: the same answer did or said something again: %+v", name, again) + } + all := strings.Join(said, "\n") + if row.says == "" && (len(first) != 0 || all != "") || row.says != "" && (len(first) != 1 || !strings.Contains(all, row.says)) { + t.Errorf("%s: did %+v and said %q, want %q", name, first, all, row.says) + } + if strings.Contains(all, running) { + t.Errorf("%s: the value is in what the node said", name) + } + } +} diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 88ce8d8..0ecd181 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -11,7 +11,6 @@ package selfcompose import ( - "bytes" "context" "crypto/sha256" "encoding/hex" @@ -56,14 +55,18 @@ type Env struct { // have says it (agent.NotComposed): a refusal is loud where haves are read, not one journal line. Nil tells nobody. Refused func(why string) // Running is what the node runs now (agent.Everything), and Unseal opens a value sealed to this machine: a secret - // the vault seals again for each ask is kept as the box the node already runs when that box opens to the same - // value, so neither the switch to composing its own nor a restart changes a composition (ADR 0547 §3). Either nil - // keeps the vault's box. + // the vault seals again for each ask is kept as the box the node already runs at the same slot (the same field of + // the same resource of the same module, delivery.KeepSeals) when that box opens to the same value, so neither the + // switch to composing its own nor a restart changes a composition (ADR 0547 §3); a box held at another slot is + // never taken, though it opens to the same value (issue 1126, C1: a provider's grant and its consumer's secret). + // Either nil keeps the vault's box. Running func() []delivery.Composed Unseal func(sealed string) ([]byte, error) - // Held is the sealed values a module's have says it runs (delivery.Have.Sealed), kept by the agent on the bus: what - // runs after a restart, before the agent applied anything, and after the switch once the kept declaration is old. - Held func(module string) []string + // Held is the sealed values a module's have says it runs, kept by the agent on the bus: what runs after a restart, + // before the agent applied anything, and after the switch once the kept declaration is old. By slot + // (delivery.Have.SealedAt); legacy is the list a have written before slots holds (delivery.Have.Sealed), reused + // only within its own module. + Held func(module string) (slots delivery.Seals, legacy []string) // Fetch reads an archive at a URL in the artifact store: a moved module's templates archive, whose texts its built // manifest carries (issue 1126). Read once per digest. Nil reads none, and a moved module does not compose. Fetch func(ctx context.Context, url string) ([]byte, error) @@ -93,8 +96,12 @@ type Composer struct { sealedOwn map[string]ownSealed said string refusal string - caught bool - rev uint64 + // doing is what the composer waits on now (waiting), for a running answer to say (issue 1126, T2); its own lock, so + // it is read while c.mu is held. + doingMu sync.Mutex + doing string + caught bool + rev uint64 } // New is a composer for one node. @@ -399,11 +406,10 @@ func (c *Composer) takeValue(u agent.Update) { } c.mu.Lock() defer c.mu.Unlock() - var r struct { - Value json.RawMessage `json:"value"` - } - var v any - if u.Deleted || json.Unmarshal(u.Value, &r) != nil || len(r.Value) == 0 || json.Unmarshal(r.Value, &v) != nil { + // Only a value its source lays: a default is the composition's already, as the controller reads them + // (catalogue.LaidValue, issue 1126, C2: Element's links blanked by a laid default). + v, laid := catalogue.LaidValue(u.Value) + if u.Deleted || !laid { delete(c.values[module], key) return } @@ -479,6 +485,10 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { } c.refused("") c.sayOnce("") + // The boxes it runs, slot by slot: read before c.mu is taken, since the agent's lock is taken to read what runs. + for key, comp := range compositions { + compositions[key] = c.keepRunning(comp, running) + } c.mu.Lock() defer c.mu.Unlock() var out []agent.Update @@ -532,45 +542,69 @@ func (c *Composer) refused(why string) { } } -// runningBox is the box the node already runs that opens to the same value as box, or box: what the agent runs, and -// what each wanted module's have says it runs sealed (Held). Read before c.mu is taken, since the agent's lock is taken -// to read what runs. -func (c *Composer) runningBox(box string) string { - if c.env.Unseal == nil || (c.env.Running == nil && c.env.Held == nil) { - return box +// keepRunning is comp with each sealed value the node already runs at the same slot kept, where it opens to the same +// value (delivery.KeepSeals, issue 1126, C1): of what runs (running, the agent's), else of what its module's have says +// it runs (Held). A have written before slots names no slot: its boxes are reused within their own module alone, at a +// slot no box is kept for, and only one opening to the slot's value. +func (c *Composer) keepRunning(comp delivery.Composed, running []delivery.Composed) delivery.Composed { + if c.env.Unseal == nil { + return comp } - value, err := c.env.Unseal(box) - if err != nil { - return box - } - var held []string - if c.env.Running != nil { - for _, comp := range c.env.Running() { - held = append(held, agent.SealedIn(comp)...) - } + module := comp.Module + if module == "" { + module = agent.NodeModule } + kept := delivery.Seals{} + var legacy []string if c.env.Held != nil { - 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)) - for m := range c.wants { - modules = append(modules, m) + slots, old := c.env.Held(module) + for slot, box := range slots { + kept[slot] = box } - c.mu.Unlock() - sort.Strings(modules) - for _, m := range modules { - held = append(held, c.env.Held(m)...) + legacy = old + } + for _, r := range running { + if r.Module == comp.Module { + for slot, box := range delivery.SealsIn(r) { + kept[slot] = box + } } } - for _, h := range held { - if h == box { + for slot, box := range delivery.SealsIn(comp) { + if _, ok := kept[slot]; ok { continue } - if v, err := c.env.Unseal(h); err == nil && bytes.Equal(v, value) { - return h + for _, h := range legacy { + if same, known := c.sameValue(box, h); known && same { + kept[slot] = h + break + } } } - return box + return delivery.KeepSeals(comp, kept, c.sameValue) +} + +// waiting notes what the composer waits on until the returned func is called, which restores what it waited on before. +func (c *Composer) waiting(what string) func() { + c.doingMu.Lock() + before := c.doing + c.doing = what + c.doingMu.Unlock() + return func() { + c.doingMu.Lock() + c.doing = before + c.doingMu.Unlock() + } +} + +// waitingOn is what the composer waits on now, in words. +func (c *Composer) waitingOn() string { + c.doingMu.Lock() + defer c.doingMu.Unlock() + if c.doing == "" { + return "nothing it reads (composing)" + } + return c.doing } func (c *Composer) sayOnce(line string) { @@ -652,6 +686,7 @@ func (r reader) Manifest(ctx context.Context, digest string) ([]byte, error) { if err != nil { return nil, err } + defer r.c.waiting("the manifest " + digest + " (" + delivery.ManifestsBucket + ")")() value, err := r.c.env.Get(ctx, delivery.ManifestsBucket, key) if err != nil { return nil, err @@ -674,6 +709,7 @@ func (r reader) Build(ctx context.Context, module, fp string) (*delivery.Builder if err != nil { return nil, err } + defer r.c.waiting("the build of " + module + " at " + fp + " (" + delivery.BuilderBuildsBucket + ")")() raw, err := r.c.env.Get(ctx, delivery.BuilderBuildsBucket, key) if err != nil || raw == nil { return nil, err @@ -700,6 +736,7 @@ func (r reader) Archive(ctx context.Context, url, digest string) ([]byte, error) return nil, fmt.Errorf("this node reads no archive (%s)", url) } // Bounded: a store that does not answer must not hold the composer or the node's answer (issue 1126). + defer r.c.waiting("the archive " + url)() fetching, cancel := context.WithTimeout(ctx, archiveWithin) defer cancel() body, err := r.c.env.Fetch(fetching, url) @@ -719,17 +756,15 @@ func (r reader) Archive(ctx context.Context, url, digest string) ([]byte, error) // Served reads, of another node's module, the keys named of mesh-config's resolved values, one key each: the node is // granted those keys alone (buscompose Principal.ServedReads, issue 1126). func (r reader) Served(ctx context.Context, node, module string, keys []string) (map[string]any, error) { + defer r.c.waiting("what " + module + " serves on " + node + " (" + ResolvedBucket + ")")() out := map[string]any{} for _, k := range keys { raw, err := r.c.env.Get(ctx, ResolvedBucket, node+"."+module+"."+k) if err != nil { return nil, err } - var v struct { - Value json.RawMessage `json:"value"` - } - var value any - if raw == nil || json.Unmarshal(raw, &v) != nil || len(v.Value) == 0 || json.Unmarshal(v.Value, &value) != nil { + value, laid := catalogue.LaidValue(raw) + if raw == nil || !laid { continue } out[k] = value @@ -760,7 +795,9 @@ func (r reader) OwnSecrets(ctx context.Context, node string) ([]compose.OwnSecre if node != r.c.env.Node || r.c.env.Own == nil { return nil, nil } + done := r.c.waiting("the vault's own secrets") raw, err := r.c.env.Own(ctx, map[string]any{}) + done() if err != nil { return nil, fmt.Errorf("the vault's own: %w", err) } @@ -781,25 +818,16 @@ func (r reader) OwnSecrets(ctx context.Context, node string) ([]compose.OwnSecre if a.Node != node { return nil, fmt.Errorf("the vault answered %q's own secrets to %s", a.Node, node) } - ids := make([]string, len(a.Own)) - fresh := make([]string, len(a.Own)) - for i, o := range a.Own { - ids[i] = o.Module + "/" + o.Name - if !r.c.keeps(ids[i], o.Generation, a.Key) { - fresh[i] = r.c.runningBox(string(o.Sealed)) - } - } r.c.mu.Lock() defer r.c.mu.Unlock() out := make([]compose.OwnSecretRow, 0, len(a.Own)) - for i, o := range a.Own { - id := ids[i] + for _, o := range a.Own { + // The box first had is kept while its generation stands, so sealing it again moves no composition; the box the + // node runs at the same slot is kept after the composition (keepRunning). + id := o.Module + "/" + o.Name kept, ok := r.c.sealedOwn[id] if !ok || kept.generation != o.Generation || kept.key != a.Key { - kept = ownSealed{generation: o.Generation, key: a.Key, sealed: fresh[i]} - if kept.sealed == "" { - kept.sealed = string(o.Sealed) - } + kept = ownSealed{generation: o.Generation, key: a.Key, sealed: string(o.Sealed)} r.c.sealedOwn[id] = kept } out = append(out, compose.OwnSecretRow{Node: node, Module: o.Module, Name: o.Name, Origin: o.Origin, @@ -823,7 +851,9 @@ func (r reader) Credentials(ctx context.Context, node string, asConsumer, asProv for _, k := range asProvider { ask = append(ask, asked{Role: "provider", CredentialKey: k}) } + done := r.c.waiting("the vault's credentials") raw, err := r.c.env.Own(ctx, map[string]any{"credentials": ask}) + done() if err != nil { return nil, nil, fmt.Errorf("the vault's own: %w", err) } @@ -844,28 +874,17 @@ func (r reader) Credentials(ctx context.Context, node string, asConsumer, asProv if a.Node != node { return nil, nil, fmt.Errorf("the vault answered %q's credentials to %s", a.Node, node) } - ids := make([]string, len(a.Credentials)) - fresh := make([]string, len(a.Credentials)) - for i, c := range a.Credentials { - ids[i] = c.Role + "|" + c.Name + "|" + c.Provision + "|" + c.Consumer + "|" + c.ConsumerModule + "|" + c.Local - if !r.c.keeps(ids[i], c.Generation, a.Key) { - fresh[i] = r.c.runningBox(string(c.Sealed)) - } - } r.c.mu.Lock() defer r.c.mu.Unlock() var creds []compose.Credential var issued []compose.Issued - for i, c := range a.Credentials { - // The value first had is kept while its generation stands, so sealing it again moves no composition; and the - // first had is the box the node runs when it opens to the same value (runningBox). - id := ids[i] + for _, c := range a.Credentials { + // The value first had is kept while its generation stands, so sealing it again moves no composition; the box the + // node runs at the same slot is kept after the composition (keepRunning). + id := c.Role + "|" + c.Name + "|" + c.Provision + "|" + c.Consumer + "|" + c.ConsumerModule + "|" + c.Local kept, ok := r.c.sealedOwn[id] if !ok || kept.generation != c.Generation || kept.key != a.Key { - kept = ownSealed{generation: c.Generation, key: a.Key, sealed: fresh[i]} - if kept.sealed == "" { - kept.sealed = string(c.Sealed) - } + kept = ownSealed{generation: c.Generation, key: a.Key, sealed: string(c.Sealed)} r.c.sealedOwn[id] = kept } k := c.CredentialKey @@ -881,14 +900,6 @@ func (r reader) Credentials(ctx context.Context, node string, asConsumer, asProv return creds, issued, nil } -// keeps says the composer holds id's sealed value at this generation and key already. -func (c *Composer) keeps(id string, generation int, key string) bool { - c.mu.Lock() - defer c.mu.Unlock() - kept, ok := c.sealedOwn[id] - 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 diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index 9b6e264..b106ca4 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -306,8 +306,28 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { t.Error("Transient misjudges a bus that does not answer, or a refusal") } + // Issue 1126, C2: a config value is laid as the controller lays it (catalogue.LaidValue), from the mesh's, the node's + // or the module's own and never from a default or an unset one, both of this node's and of what another node serves: + // a default laid blanked Element's links on ace. + for _, row := range []struct { + source string + laid bool + }{{"mesh", true}, {"node", true}, {"module", true}, {"default", false}, {"unset", false}, {"", false}} { + probe := New(env) + value := []byte(`{"value":["x"],"source":"` + row.source + `"}`) + probe.takeValue(agent.Update{Key: "n1.lemurs.links", Value: value}) + _, laid := probe.values["lemurs"]["links"] + b.mu.Lock() + b.kv[ResolvedBucket+"/n2.lemurs.links"] = value + b.mu.Unlock() + served, err := reader{c: probe}.Served(ctx, "n2", "lemurs", []string{"links"}) + if laid != row.laid || err != nil || (served["links"] != nil) != row.laid { + t.Errorf("a value from %q: laid %v, served %v (%v), want laid %v", row.source, laid, served, err, row.laid) + } + } + // A config value of another module, and a value changing nothing composed, hand nothing. - b.put(t, ResolvedBucket, "n1.other.k", []byte(`{"value":1}`)) + b.put(t, ResolvedBucket, "n1.other.k", []byte(`{"value":1,"source":"node"}`)) quiet(t, out) // A want removed takes its composition back. @@ -367,7 +387,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { t.Fatalf("keeper's composition does not carry the sealed value: %s", handed) } before := handed - b.put(t, ResolvedBucket, "n1.other.k", []byte(`{"value":2}`)) + b.put(t, ResolvedBucket, "n1.other.k", []byte(`{"value":2,"source":"node"}`)) quiet(t, out) b.mu.Lock() generation++ @@ -383,32 +403,47 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { // node runs comes from keeper's have (Held); the vault seals the value again, other bytes, and the box the have holds // is kept, since it opens to the same value: the composition does not move. The box from before the rotation opens // to the old value, so the vault's is taken. - boxOf := func(raw string) string { + // Issue 1126, C1: a box is kept only at the slot that holds it (the same field of the same resource of the same + // module). A box another module holds (a provider's grant of the value its consumer holds as a secret, on one node) + // or another resource of keeper holds is never taken, though it opens to the same value; a have written before slots + // (a list) is reused within its own module alone. + boxOf := func(raw string) (string, string) { c, err := delivery.DecodeComposed([]byte(raw)) if err != nil { t.Fatal(err) } - sealed := agent.SealedIn(c) + sealed := delivery.SealsIn(c) if len(sealed) != 1 { t.Fatalf("keeper's composition carries %v", sealed) } - return sealed[0] + for slot, box := range sealed { + return slot, box + } + return "", "" } - // 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. + slot, rotatedBox := boxOf(string(rotated.Value)) + _, beforeBox := boxOf(before) for _, row := range []struct { - name, held, in string + name, in, held string + slots delivery.Seals kept bool - }{{"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}} { + }{ + {"the have holds the rotated value at its slot", "keeper", rotatedBox, delivery.Seals{slot: rotatedBox}, true}, + {"the have holds the value before the rotation", "keeper", beforeBox, delivery.Seals{slot: beforeBox}, false}, + {"a have written before slots holds the rotated value", "keeper", rotatedBox, nil, true}, + {"another module holds the same value", "lemurs", rotatedBox, delivery.Seals{"lemurs.grant-x#sealed": rotatedBox}, false}, + {"another resource of keeper holds the same value", "keeper", rotatedBox, delivery.Seals{"keeper.other#sealed": rotatedBox}, false}, + } { restarted := env restarted.Running = func() []delivery.Composed { return nil } - restarted.Held = func(module string) []string { - if module == row.in { - return []string{row.held} + restarted.Held = func(module string) (delivery.Seals, []string) { + if module != row.in { + return nil, nil } - return nil + if row.slots == nil { + return nil, []string{row.held} + } + return row.slots, nil } restarted.Unseal = func(box string) ([]byte, error) { b.mu.Lock() @@ -419,14 +454,56 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { return nil, fmt.Errorf("not sealed to n1: %s", box) } again := New(restarted) - again.wants = map[string]delivery.Want{"keeper": {FP: fp, Self: true}, agent.NodeModule: {}} - rows, err := reader{c: again}.OwnSecrets(ctx, "n1") - if err != nil || len(rows) != 1 { - t.Fatalf("%s: own secrets %+v, %v", row.name, rows, err) + c.mu.Lock() + for k, v := range c.wants { + again.wants[k] = v } - if kept := rows[0].Sealed == row.held; kept != row.kept { - t.Errorf("%s: composed with %s (held %s), kept %v, want %v", row.name, rows[0].Sealed, row.held, kept, row.kept) + for k, v := range c.placements { + again.placements[k] = v } + for k, v := range c.records { + again.records[k] = v + } + for k, v := range c.modules { + again.modules[k] = v + } + c.mu.Unlock() + var composed string + for _, u := range again.compose(ctx) { + if u.Key == keeperKey { + composed = string(u.Value) + } + } + if composed == "" { + t.Fatalf("%s: keeper was not composed", row.name) + } + if _, box := boxOf(composed); (box == row.held) != row.kept { + t.Errorf("%s: composed with %s (held %s), kept %v", row.name, box, row.held, row.kept) + } + } + // 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. + ownComp := delivery.Composed{Node: "n1", Resources: []json.RawMessage{json.RawMessage(`{"id":"core-box","sealed":"{\"ask\":999}"}`)}} + b.mu.Lock() + sealedAt[`{"ask":999}`] = generation + 1 + b.mu.Unlock() + coreEnv := env + coreEnv.Unseal = func(box string) ([]byte, error) { + b.mu.Lock() + defer b.mu.Unlock() + if g, ok := sealedAt[box]; ok { + return []byte(fmt.Sprintf("password-%d", g)), nil + } + return nil, fmt.Errorf("not sealed to n1: %s", box) + } + coreEnv.Held = func(module string) (delivery.Seals, []string) { + if module == agent.NodeModule { + return delivery.Seals{"core-box#sealed": rotatedBox}, nil + } + return nil, nil + } + if got := delivery.SealsIn(New(coreEnv).keepRunning(ownComp, nil))["core-box#sealed"]; got != rotatedBox { + t.Errorf("the node's own core box after a restart is %s, want the one its key holds %s", got, rotatedBox) } // The pair credentials asked (ADR 0550 §5): the keys go to the vault's own with their role, a consumer's comes back as @@ -460,10 +537,15 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { if err != nil || len(creds) != 1 || len(issued) != 1 { t.Fatalf("credentials %+v issued %+v, %v", creds, issued, err) } - if creds[0].State != compose.CredentialHeld || creds[0].Provider != "n2" || creds[0].Sealed != `{"run":1}` || + if creds[0].State != compose.CredentialHeld || creds[0].Provider != "n2" || creds[0].Sealed != `{"c":1}` || issued[0].Consumer != "n3" || issued[0].ForProvider != `{"p":1}` { t.Errorf("credentials %+v issued %+v", creds, issued) } + // What runs at the credential's slot is kept, after the composition (keepRunning, issue 1126, C1). + webComp := delivery.Composed{Node: "n1", Module: "web", Resources: []json.RawMessage{json.RawMessage(`{"id":"web.secret-pg","sealed":"{\"c\":1}"}`)}} + if got := delivery.SealsIn(pc.keepRunning(webComp, pairEnv.Running()))["web.secret-pg#sealed"]; got != `{"run":1}` { + t.Errorf("web's credential composed as %s, want the box that runs", got) + } // A node that cannot compose tells the agent once, and once again when it composes (the agent's haves say it). pc.refused("a stand-in for web's password") pc.refused("a stand-in for web's password") @@ -718,6 +800,11 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h-same"}, Placement: &trunk}) sameKey, _ := delivery.PlacementCandidateKey("same", "n1") nb.put(delivery.PlacementCandidatesBucket, sameKey, sameRaw, 3) + // Issue 1126, T3: an older group (a closed pull request's candidate, rev 2) is answered after the newest. + oldRaw, _ := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "older", + Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 2, Head: "h-old"}, Placement: &trunk}) + oldKey, _ := delivery.PlacementCandidateKey("older", "n1") + nb.put(delivery.PlacementCandidatesBucket, oldKey, oldRaw, 2) for _, bucket := range []string{delivery.PlacementCandidatesBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket} { nb.send(bucket, agent.Update{Caught: true}) } @@ -742,11 +829,31 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { if g := a["same"]; g.Verdict != delivery.CandidateSame || !strings.Contains(g.Note, "this answer took over") { t.Errorf("the shadow, once ready, answered %+v", g) } + // The newest first, handed at once: the older group still holds the answer it had. + if g := a["older"]; g.Verdict != delivery.CandidateRefused { + t.Errorf("the older group was answered before the newest was handed: %+v", g) + } + if g := (<-nanswered)["older"]; g.Verdict != delivery.CandidateSame { + t.Errorf("the older group, answered next: %+v", g) + } + // Issue 1126, T2: an answer says when it starts and when it ends (and, every minute between, what it waits on). nmu.Lock() - if !strings.Contains(strings.Join(nsaid, "\n"), "the answer to same took") { - t.Errorf("a slow answer was not said: %q", nsaid) + said := strings.Join(nsaid, "\n") + for _, line := range []string{"the answer to same took", "answering placement candidate same (revision 3)", + "answered placement candidate same in", "answering placement candidate older (revision 2)"} { + if !strings.Contains(said, line) { + t.Errorf("%q was not said: %q", line, nsaid) + } + } + if strings.Index(said, "answering placement candidate same") > strings.Index(said, "answering placement candidate older") { + t.Errorf("the older group was answered first: %q", nsaid) } nmu.Unlock() + done := nsh.c.waiting("the vault's own secrets") + if got := nsh.c.waitingOn(); got != "the vault's own secrets" { + t.Errorf("a running answer would say it waits on %q", got) + } + done() case <-time.After(10 * time.Second): t.Fatal("the shadow, once ready, answered nothing") } @@ -830,7 +937,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { buildKey, _ := delivery.BuilderBuildKey("lemurs", fp) b.mu.Lock() b.kv[delivery.BuilderBuildsBucket+"/"+buildKey] = built - b.kv[ResolvedBucket+"/n2.mailer.domain"] = []byte(`{"value":"example.org"}`) + b.kv[ResolvedBucket+"/n2.mailer.domain"] = []byte(`{"value":"example.org","source":"module"}`) b.reads = nil b.mu.Unlock() fetched := 0 diff --git a/internal/selfcompose/shadow.go b/internal/selfcompose/shadow.go index f5a3b9a..bcdc263 100644 --- a/internal/selfcompose/shadow.go +++ b/internal/selfcompose/shadow.go @@ -276,26 +276,92 @@ func (s *Shadow) notReady(ctx context.Context, ins []input, caught []bool) { } } -// answerAll answers every group and hands the answers when any changed. +// answerAll answers every group, the newest candidate first, and hands the answers each time one changes (issue 1126, +// T3: ace answered a closed pull request's group for 22 minutes before the newer one, all handed at the end). Newest +// is the highest revision of the group's keys: a closed pull request's candidate is the older and waits; the indexer +// drops a closed one's keys, which ends its answer here (a node does not read the forge). Each answer says when it +// starts, every progressEvery while it runs what it waits on, and when it ends (T2: the node said nothing for 22 +// minutes). 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) + type listed struct { + group string + rev uint64 + } + var groups []listed + for g, keys := range s.candidates { + l := listed{group: g} + for _, k := range keys { + l.rev = max(l.rev, k.rev) + } + groups = append(groups, l) + } + // Each group's last answer stands until it is answered again; a group whose keys went is dropped. + next := map[string]delivery.CandidateAnswer{} + for _, l := range groups { + if a, ok := s.answers[l.group]; ok { + next[l.group] = a + } } s.mu.Unlock() - sort.Strings(groups) - next := map[string]delivery.CandidateAnswer{} - for _, group := range groups { + sort.Slice(groups, func(i, j int) bool { + if groups[i].rev != groups[j].rev { + return groups[i].rev > groups[j].rev + } + return groups[i].group < groups[j].group + }) + hand := func() { + 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 + } + } + out := make(map[string]delivery.CandidateAnswer, len(next)) + for g, a := range next { + out[g] = a + } + if !same { + s.answers = out + } + s.mu.Unlock() + if !same && s.answer != nil { + given := make(map[string]delivery.CandidateAnswer, len(out)) + for g, a := range out { + given[g] = a + } + s.answer(ctx, given) + c.env.Say(fmt.Sprintf("%s answered %d placement candidate group(s)", node, len(given))) + } + } + for _, l := range groups { + if ctx.Err() != nil { + return + } + group := l.group began := time.Now() - slow := time.AfterFunc(slowAfter, func() { - c.env.Say(fmt.Sprintf("the answer to %s has run over %s, and goes on", group, slowAfter)) - }) + c.env.Say(fmt.Sprintf("answering placement candidate %s (revision %d)", group, l.rev)) + stop := make(chan struct{}) + go func() { + tick := time.NewTicker(progressEvery) + defer tick.Stop() + for { + select { + case <-stop: + return + case <-tick.C: + c.env.Say(fmt.Sprintf("the answer to %s has run %s and waits on %s", group, + time.Since(began).Round(time.Second), c.waitingOn())) + } + } + }() a, ok := s.answerGroup(ctx, group) - slow.Stop() - if took := time.Since(began); took > slowAfter { + close(stop) + took := time.Since(began) + if took > slowAfter { c.env.Say(fmt.Sprintf("the answer to %s took %s", group, took.Round(time.Second))) if ok { // Words that do not move with the time it took, so an answer the same as before is not written again. @@ -303,35 +369,27 @@ func (s *Shadow) answerAll(ctx context.Context) { } } if ok { + a = delivery.FitAnswer(a, delivery.MaxAnswerBytes) next[group] = a + c.env.Say(fmt.Sprintf("answered placement candidate %s in %s: %s", group, took.Round(time.Millisecond), a.Verdict)) + } else { + delete(next, group) + c.env.Say(fmt.Sprintf("placement candidate %s does not place this node: nothing to answer", group)) } + hand() } - 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))) - } + hand() // a group whose keys went, when no group is left to answer } +// How often a running answer says what it waits on. +var progressEvery = time.Minute + // answerGroup is the node's answer to one group, and false when the node has nothing to answer: neither the trunk nor // the group places it. func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.CandidateAnswer, bool) { c := s.c node := c.env.Node + defer c.waiting("composing")() s.mu.Lock() keys := s.candidates[group] given := make(map[string]*delivery.Placement, len(keys))