Merge pull request 'Each node's engine writes its own declared packages and answers packages (ADR 0565 §6, issue 1953)' (#193) from feat/node-packages into main

This commit was merged in pull request #193.
This commit is contained in:
2026-11-01 17:02:53 +00:00
15 changed files with 470 additions and 14 deletions
+35
View File
@@ -329,3 +329,38 @@ func composeFrom(current *agentNow, composer *atomic.Pointer[selfcompose.Compose
return out, nil
}
}
// packagesReadFor bounds the verb packages' reads and its write.
const packagesReadFor = 20 * time.Second
// packagesFrom answers the verb packages (novox/hq ADR 0565 §6, §15): this node's key of the engine's state packages
// beside the controller's mirror's key for it and their difference; with write, the key written again from what the
// composer that runs now handed, refused in words while the node composes none of its own.
func packagesFrom(node string, session *link.Session, composer *atomic.Pointer[selfcompose.Composer]) func(write bool) (any, error) {
return func(write bool) (any, error) {
ctx, cancel := context.WithTimeout(context.Background(), packagesReadFor)
defer cancel()
var wrote *bool
if write {
c := composer.Load()
if c == nil {
return nil, errors.New("this node composes none of its own modules now (it applies the controller's " +
"compositions, or its node agent is starting), so it has no packages of its own to write")
}
w, err := c.WritePackages(ctx)
if err != nil {
return nil, fmt.Errorf("this node's key of %s is not written: %w", link.PackagesBucket, err)
}
wrote = &w
}
key, keyErr := link.EnginePackages{Session: session}.Get(ctx, node)
var mirror []byte
h, mirrorErr := session.KVHistory(ctx, link.MirrorPackagesBucket, node)
if mirrorErr == nil && len(h) > 0 && !h[len(h)-1].Deleted {
mirror = h[len(h)-1].Value
}
v := link.PackagesAnswer(node, key, mirror, keyErr, mirrorErr)
v.Wrote = wrote
return v, nil
}
}
+1
View File
@@ -1143,6 +1143,7 @@ func runLink(ctx context.Context, opts options) error {
current := &agentNow{}
queue.Acts.Composition = compositionFrom(current)
queue.Acts.Compose = composeFrom(current, &composerNow)
queue.Acts.Packages = packagesFrom(mine.Node, queue, &composerNow)
setRunsHere(runsFromAgent(current))
factsOfNow := func(ctx context.Context, ag *agent.Agent) {
sayFacts(ctx, opts, mine.Membership.Signer, queue, say, ag)
+3
View File
@@ -1024,6 +1024,9 @@ func selfComposer(ctx context.Context, session *link.Session, node string, say f
// And it keeps each placed module's data items in its own state data, which the node's disk-load watches (novox/hq
// ADR 0577 clause 1); the shadow composes for candidates and keeps none.
env.Data = link.EngineData{Session: session}
// And the packages its compositions declare in its own key of the state packages, which pacman reads (novox/hq ADR
// 0565 §6).
env.Packages = link.EnginePackages{Session: session}
unreadable := ""
placed, err := selfcompose.Placed(ctx, env.Get, node, link.Backoff, func(m string) {
if strings.Contains(m, "cannot be read") {
+22 -6
View File
@@ -405,7 +405,19 @@ func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) {
placement, resolved, candidates := st.placement, st.resolved, st.candidates
hkey, place, fp := st.hkey, st.place, st.fp
within(t, 5*time.Second, "the node did not compose and apply its own module once placed", func() bool { return b.file("web") == "greeting=bonjour\n" })
// ADR 0565 §6 (1953): the node's engine writes its own key of the state packages, under the grant the grants part
// composes for it, with each package its compositions declare and the module declaring it; once, while nothing changes.
wantPkgs := `{"node":"` + itNode + `","packages":[{"package":"web-tools","modules":["web"],"absent":false}]}`
within(t, 5*time.Second, "the node did not write its key of the state packages", func() bool {
e, err := st.packages.Get(itNode)
return err == nil && string(e.Value()) == wantPkgs
})
written, _ := st.packages.Get(itNode)
h8rest(t, b, resolved, candidates, hkey, place, fp)
// web composed again for its changed config value declares the same packages: the key is not written again.
if e, _ := st.packages.Get(itNode); e == nil || written == nil || e.Revision() != written.Revision() {
t.Errorf("the node's packages key was written again with no package changed: %v, was %v", e, written)
}
t.Run("one answer per candidate revision", func(t *testing.T) { h8once(t, b, resolved, candidates, place) })
h8back(t, b, placement, resolved, candidates, hkey)
}
@@ -552,9 +564,11 @@ type Have1126 struct {
// h8state is what h8 set up that the rest of H8 writes to.
type h8state struct {
placement, resolved, candidates nats.KeyValue
hkey string
place []byte
fp string
// packages is the engine's state packages (ADR 0565 §6), as the grants part creates it from the engine's manifest.
packages nats.KeyValue
hkey string
place []byte
fp string
}
// h8 sets H8's bus up and starts the agent, then places the node. With checkBuild, the placed module's build key is
@@ -572,13 +586,15 @@ func h8(t *testing.T, b *itBed, checkBuild bool) (func(), h8state) {
resolved := kvOf(selfcompose.ResolvedBucket)
builds := kvOf(delivery.BuilderBuildsBucket)
candidates := kvOf(delivery.PlacementCandidatesBucket)
packages := kvOf(link.PackagesBucket)
// 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")
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"},` +
`{"id":"tools","type":"package","package":"web-tools"}],` +
`"build":{"artifacts":[{"name":"files","kind":"archive","from":"files"}]}}`
digest := delivery.ManifestDigest([]byte(man))
mraw, _ := delivery.EncodeManifest(delivery.Manifest{Module: "web", JSON: man})
@@ -586,7 +602,7 @@ func h8(t *testing.T, b *itBed, checkBuild bool) (func(), h8state) {
fp := fpIt("a")
modv, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/web", Commit: strings.Repeat("1", 40)})
place, _ := delivery.EncodePlacement(delivery.Placement{Node: itNode, Modules: []delivery.PlacedModule{{Module: "web"}}})
rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: itNode, Capabilities: []string{"service-manager"}}})
rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: itNode, Capabilities: []string{"service-manager", "package-manager"}}})
// it1 in one key per module and node (research 072 row 37), it2 in the form before: a node reads both.
pkey, _ := delivery.PlacementNodeKey(itNode)
places, _ := delivery.PlacementValues(delivery.Placement{Node: itNode, Modules: []delivery.PlacedModule{{Module: "web"}}})
@@ -659,7 +675,7 @@ func h8(t *testing.T, b *itBed, checkBuild bool) (func(), h8state) {
t.Fatal(err)
}
}
return stop, h8state{placement: placement, resolved: resolved, candidates: candidates, hkey: hkey, place: place, fp: fp}
return stop, h8state{placement: placement, resolved: resolved, candidates: candidates, packages: packages, hkey: hkey, place: place, fp: fp}
}
// h8rest is H8 once the node composes its own: the have, a config revision, and the candidate answered "same".
+1 -1
View File
@@ -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.20261101022206-8af6e7266915
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101151945-b2a27ed5cec4
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
+2 -2
View File
@@ -1,5 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101022206-8af6e7266915 h1:x8k7IeMirMTxUo8BH+xUO+MDnHYgU+636jl1NIw1wvs=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101022206-8af6e7266915/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101151945-b2a27ed5cec4 h1:xeuMBwOHNKrVUdGVg7Bdv5tkdueBvngwJFyk6n10rp0=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101151945-b2a27ed5cec4/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=
+133
View File
@@ -0,0 +1,133 @@
package link
import (
"context"
"encoding/json"
"errors"
"fmt"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go"
)
// ---- the packages this node's compositions declare, in the bus's own storage (novox/hq ADR 0565 §6) ---------------
// PackagesBucket is the node-engine's declared state packages, created by the grants part from the engine's manifest as
// its states health and data are: one key, the node's name, written by that node's engine alone, read by pacman as
// mesh-host.packages (delivery.EnginePackagesRead).
const PackagesBucket = "mesh-host_" + delivery.EnginePackagesState
// MirrorPackagesBucket is the controller's declared-packages mirror (ADR 0507 §5), which the engine reads its own node's
// key of until the lease write, to compare (the node seat's verb packages).
const MirrorPackagesBucket = "mesh-controller_declared-packages"
// packagesStore binds PackagesBucket over the link open now: bound, never created.
func (l *natsLink) packagesStore() (nats.KeyValue, error) {
kv, err := l.js.KeyValue(PackagesBucket)
if err != nil {
return nil, fmt.Errorf("the node-engine's state %s cannot be bound (it is created from the node-engine's "+
"manifest; written again at the next composition that changes, or by the node seat's packages with write): %w",
PackagesBucket, err)
}
return kv, nil
}
// packagesStore binds PackagesBucket over a link made of a connection the caller holds (OverNATS), as natsLink does.
func (b OverNATS) packagesStore() (nats.KeyValue, error) {
kv, err := b.JS.KeyValue(PackagesBucket)
if err != nil {
return nil, fmt.Errorf("the node-engine's state %s cannot be bound: %w", PackagesBucket, err)
}
return kv, nil
}
// EnginePackages is this node's key of PackagesBucket over whichever link the session has open
// (selfcompose.PackagesStore); with none open every read and write is refused, and nothing is kept anywhere else.
type EnginePackages struct{ Session *Session }
func (p EnginePackages) kv() (nats.KeyValue, error) {
p.Session.init()
p.Session.mu.Lock()
bus := p.Session.bus
p.Session.mu.Unlock()
l, ok := bus.(interface{ packagesStore() (nats.KeyValue, error) })
if bus == nil || !ok {
return nil, ErrNoLink
}
return l.packagesStore()
}
// Get is the key's value, nil when it has none.
func (p EnginePackages) Get(_ context.Context, key string) ([]byte, error) {
kv, err := p.kv()
if err != nil {
return nil, err
}
e, err := kv.Get(key)
if errors.Is(err, nats.ErrKeyNotFound) {
return nil, nil
}
if err != nil {
return nil, err
}
return e.Value(), nil
}
// Put writes the key.
func (p EnginePackages) Put(_ context.Context, key string, value []byte) error {
kv, err := p.kv()
if err != nil {
return err
}
_, err = kv.Put(key, value)
return err
}
// PackagesView is the node seat's verb packages' answer (ADR 0565 §6): this node's key of the engine's state packages,
// and until the lease write the controller's mirror's key for the node and every package they hold differently.
type PackagesView struct {
Node string `json:"node"`
// Key is the node's key as the bus holds it (Source); null when the engine has written none.
Source string `json:"source"`
Key *delivery.DeclaredPackages `json:"key"`
KeyError string `json:"key_error,omitempty"`
// Mirror is the controller's mirror's key for the node (MirrorSource); null when it holds none or cannot be read.
MirrorSource string `json:"mirror_source"`
Mirror *delivery.DeclaredPackages `json:"mirror"`
MirrorError string `json:"mirror_error,omitempty"`
// Differences is every package the two hold differently, read only when both are; Agree when there is none.
Differences []delivery.PackageDifference `json:"differences"`
Agree bool `json:"agree"`
// Wrote says, for write, whether the key was written again (false: it already held what the engine composed).
Wrote *bool `json:"wrote,omitempty"`
Note string `json:"note,omitempty"`
}
// PackagesAnswer is the verb's answer from the two keys as read (nil: no key), each with its read's error.
func PackagesAnswer(node string, key, mirror []byte, keyErr, mirrorErr error) PackagesView {
v := PackagesView{Node: node, Source: delivery.EnginePackagesRead, MirrorSource: "mesh-controller.declared-packages",
Differences: []delivery.PackageDifference{}}
read := func(raw []byte, err error, into **delivery.DeclaredPackages, said *string) {
if err == nil && raw != nil {
var d delivery.DeclaredPackages
if err = json.Unmarshal(raw, &d); err == nil {
*into = &d
}
}
if err != nil {
*said = err.Error()
}
}
read(key, keyErr, &v.Key, &v.KeyError)
read(mirror, mirrorErr, &v.Mirror, &v.MirrorError)
switch {
case v.Key != nil && v.Mirror != nil:
v.Differences = delivery.DiffDeclaredPackages(*v.Key, *v.Mirror)
v.Agree = len(v.Differences) == 0
case v.Key == nil && v.KeyError == "":
v.Note = "the engine has written no key for this node yet: pacman reads the controller's mirror for it"
case v.Mirror == nil && v.MirrorError == "":
v.Note = "the controller's mirror holds no key for this node: nothing to compare"
}
return v
}
+15
View File
@@ -42,6 +42,9 @@ const (
ActComposition = "composition"
// ActCompose composes the node now, the recovery of a failed composition (issue 774).
ActCompose = "compose"
// ActPackages reads the packages this node's compositions declare, and with write writes its key again (novox/hq
// ADR 0565 §6, §15).
ActPackages = "packages"
)
// The states of an act.
@@ -137,6 +140,9 @@ type HandActs struct {
// Compose makes a pass of the node's composition now and says how it went; with module, its apply is also tried
// again (the verb compose, issue 774).
Compose func(module string) (any, error)
// Packages answers the node's key of the engine's state packages beside the controller's mirror's key and their
// difference, and with write writes the key again from what the engine composed (the verb packages, ADR 0565 §6, §15).
Packages func(write bool) (any, error)
// Say says one of the seat's events, asked, done or refused; nil says nothing.
Say func(event string, body any)
Now func() time.Time
@@ -200,6 +206,15 @@ func (h *HandActs) Call(c EngineCall) (any, error) {
return nil, errors.New("this engine composes nothing on a call yet: its node agent has not started")
}
return h.Compose(arg("module"))
case ActPackages:
if h.Packages == nil {
return nil, errors.New("this engine keeps no packages yet: its node agent has not started")
}
write := arg("write") == "true"
if b, ok := c.Args["write"].(bool); ok {
write = b
}
return h.Packages(write)
case "acts":
if id := arg("id"); id != "" {
return h.Act(id)
+20 -1
View File
@@ -240,7 +240,20 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) {
Composition: func(module, digest string, diff bool) (any, error) {
return map[string]any{"module": module, "digest": digest, "diff": diff}, nil
},
Compose: func(module string) (any, error) { return map[string]any{"composed": true, "retried": module}, nil }}
Compose: func(module string) (any, error) { return map[string]any{"composed": true, "retried": module}, nil },
// ADR 0565 §6, §15: the node's key beside the controller's mirror's, the mirror's from its writer's test.
Packages: func(write bool) (any, error) {
key := []byte(`{"node":"n1","packages":[{"package":"curl","modules":["backup","web"],"absent":false}]}`)
mirror := []byte(`{"node":"n1","packages":[{"package":"curl","modules":["backup","web"],"absent":false},` +
`{"package":"nano","modules":["web"],"absent":true}]}`)
if !write {
return PackagesAnswer("n1", key, mirror, nil, nil), nil
}
v := PackagesAnswer("n1", nil, nil, nil, errors.New("not granted"))
wrote := true
v.Wrote = &wrote
return v, nil
}}
for _, row := range []struct{ verb, subject, want string }{
{"judge-directories", "mesh.seat.node-engine.call.judge-directories.n1.shanks~node-tools", `"said":"judged for shanks.node-tools"`},
{"judge-directories", EngineToolSubject("judge-directories", "n1"), `"said":"judged for unproven"`},
@@ -252,6 +265,12 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) {
// The recovery verb compose (issue 774) makes a pass now, with a module's apply tried again when named.
{"compose", EngineToolSubject("compose", "n1"), `"composed":true,"retried":""`},
{"compose", EngineToolSubject("compose", "n1") + " " + `{"module":"route-proxy"}`, `"retried":"route-proxy"`},
// The read verb packages (ADR 0565 §6) names both sources and their difference; with write, what it wrote, and a
// key the engine has not written, or a mirror it may not read, is said and not an error.
{"packages", EngineToolSubject("packages", "n1"), `"differences":[{"package":"nano","absent":true,"engine":[],"mirror":["web"]}],"agree":false`},
{"packages", EngineToolSubject("packages", "n1"), `"source":"mesh-host.packages","key":{"node":"n1"`},
{"packages", EngineToolSubject("packages", "n1") + " " + `{"write":true}`, `"mirror":null,"mirror_error":"not granted","differences":[],"agree":false,"wrote":true`},
{"packages", EngineToolSubject("packages", "n1") + " " + `{"write":"true"}`, `"note":"the engine has written no key for this node yet`},
{"unserved", "mesh.seat.node-engine.call.unserved.n1.x", `"error":"the node seat node-engine serves no verb \"unserved\""`},
} {
subject, body, ok := strings.Cut(row.subject, " ")
+50
View File
@@ -177,6 +177,56 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) {
}
})
}
// ADR 0565 §6 (1953): the node's key of the engine's state packages is written after a pass that handed something new
// when the packages its compositions declare differ, and only then: a composition adding a package writes it once,
// one that changes no package writes nothing; a write refused is owed after the growing bound; the verb's write reads
// the key again, so a key lost on the bus is written back. Another node's key is never touched.
pkgStore := &fakeData{kv: map[string][]byte{"n2": []byte(`{"node":"n2","packages":[]}`)}}
pc := New(Env{Node: "n1", Packages: pkgStore})
web := func(pkgs ...string) delivery.Composed {
comp := delivery.Composed{Node: "n1", Module: "web"}
for _, p := range pkgs {
comp.Resources = append(comp.Resources, []byte(`{"id":"`+p+`","type":"package","package":"`+p+`"}`))
}
comp.Resources = append(comp.Resources, []byte(`{"id":"conf","type":"file","path":"/etc/web.conf"}`))
return packagesOf(comp)
}
for _, step := range []struct {
name string
comp delivery.Composed
fail error
puts int
owed bool
holds string
}{
{name: "the run's first pass writes the key", comp: web("curl"), puts: 1,
holds: `{"node":"n1","packages":[{"package":"curl","modules":["web"],"absent":false}]}`},
{name: "a pass that changes no package writes nothing", comp: web("curl"), puts: 1},
{name: "a refused write is owed after the bound", comp: web("curl", "jq"), fail: errors.New("no grant yet"), puts: 1, owed: true},
{name: "a composition adding a package writes the key once", comp: web("curl", "jq"), puts: 2,
holds: `{"node":"n1","packages":[{"package":"curl","modules":["web"],"absent":false},{"package":"jq","modules":["web"],"absent":false}]}`},
} {
pkgStore.mu.Lock()
pkgStore.fail = step.fail
pkgStore.mu.Unlock()
pc.paid()
pc.mu.Lock()
pc.pkgs, pc.pkgsComposed, pc.pkgsDue = map[string]delivery.Composed{webKey: step.comp}, true, true
pc.mu.Unlock()
pc.publishPackages(context.Background())
if len(pkgStore.puts) != step.puts || pc.owed.Load() != step.owed ||
(step.holds != "" && string(pkgStore.kv["n1"]) != step.holds) || string(pkgStore.kv["n2"]) != `{"node":"n2","packages":[]}` {
t.Errorf("%s: %d writes, owed %v, n1 holds %s", step.name, len(pkgStore.puts), pc.owed.Load(), pkgStore.kv["n1"])
}
}
delete(pkgStore.kv, "n1")
if wrote, err := pc.WritePackages(context.Background()); !wrote || err != nil || pkgStore.kv["n1"] == nil {
t.Errorf("the verb's write of a lost key wrote %v (%v)", wrote, err)
}
if wrote, _ := pc.WritePackages(context.Background()); wrote {
t.Error("the verb's write of a key holding what was composed wrote it again")
}
// The items are worked out from what the pass read: a done build is read over the bus once, not again for the data
// items after the pass (a module in a running build is read again, since it may land).
b := newFakeBus()
+152
View File
@@ -0,0 +1,152 @@
package selfcompose
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"sort"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// **The packages this node's compositions declare, kept by its engine** (novox/hq ADR 0565 §6, §15; issue 1953). The
// composer keeps the package resources of each composition it handed, and after a pass that handed something new (or
// the run's first) it works out the node's key of the engine's state packages from them (delivery.DeclaredPackagesOf,
// the controller mirror's layout and order) and writes it only when the value differs from what the key holds (ADR 0289
// §1): no timer, and no composing of another node. A module held keeps the packages of what was handed of it, as it keeps
// running what it runs. pacman reads the key as mesh-host.packages; the node seat's verb packages answers it beside the
// controller's mirror until the lease write, and with write writes it again (ADR 0565 §15).
// PackagesStore is this node's key of the engine's state packages (link.EnginePackages).
type PackagesStore interface {
// Get is the key's value, nil when it has none.
Get(ctx context.Context, key string) ([]byte, error)
Put(ctx context.Context, key string, value []byte) error
}
// packagesOf is a composition's package resources alone, with its module: what the node's key is worked out from.
func packagesOf(comp delivery.Composed) delivery.Composed {
out := delivery.Composed{Node: comp.Node, Module: comp.Module}
for _, raw := range comp.Resources {
var head struct {
Type string `json:"type"`
}
if json.Unmarshal(raw, &head) == nil && head.Type == "package" {
out.Resources = append(out.Resources, raw)
}
}
return out
}
// DeclaredPackages is the node's key as this composer would write it now, from what it handed.
func (c *Composer) DeclaredPackages() (delivery.DeclaredPackages, error) {
c.mu.Lock()
keys := make([]string, 0, len(c.pkgs))
for k := range c.pkgs {
keys = append(keys, k)
}
sort.Strings(keys)
comps := make([]delivery.Composed, 0, len(keys))
for _, k := range keys {
comps = append(comps, c.pkgs[k])
}
composed := c.pkgsComposed
c.mu.Unlock()
if !composed {
return delivery.DeclaredPackages{}, errors.New("this node's composer has handed no composition yet in this run, so it " +
"declares no packages yet")
}
return delivery.DeclaredPackagesOf(c.env.Node, comps)
}
// publishPackages writes the node's key of the state packages when a pass asked it (pkgsDue) and the value differs.
func (c *Composer) publishPackages(ctx context.Context) {
c.mu.Lock()
due := c.pkgsDue
c.pkgsDue = false
c.mu.Unlock()
if !due || c.env.Packages == nil {
return
}
wrote, err := c.writePackages(ctx, false)
if err != nil {
c.sayPackages("this node's packages are not kept in " + enginePackagesBucket + "; " + c.owePackages() + ": " + err.Error())
return
}
c.mu.Lock()
c.pkgsTries = 0
c.mu.Unlock()
c.sayPackages("")
if wrote {
c.env.Say("this node's packages: its key of " + enginePackagesBucket + " written")
}
}
// enginePackagesBucket is the engine's state packages as the bus names it, for what the composer says.
const enginePackagesBucket = "mesh-host_" + delivery.EnginePackagesState
// WritePackages writes the node's key again from what the composer handed, when it differs from what the bus holds
// now (the node seat's verb packages with write, ADR 0565 §15): the key is read again rather than trusted, so a key lost
// or changed on the bus is written back. It answers whether it wrote.
func (c *Composer) WritePackages(ctx context.Context) (bool, error) {
if c.env.Packages == nil {
return false, errors.New("this engine keeps no packages key")
}
return c.writePackages(ctx, true)
}
// writePackages writes the key when the value worked out differs from what it holds: read once per run (pkgsKept), or
// again when fresh says so.
func (c *Composer) writePackages(ctx context.Context, fresh bool) (bool, error) {
v, err := c.DeclaredPackages()
if err != nil {
return false, err
}
raw, err := json.Marshal(v)
if err != nil {
return false, err
}
c.pkgsWrite.Lock()
defer c.pkgsWrite.Unlock()
if fresh || !c.pkgsRead {
kept, err := c.env.Packages.Get(ctx, c.env.Node)
if err != nil {
return false, fmt.Errorf("%s cannot be read: %w", c.env.Node, err)
}
c.pkgsKept, c.pkgsRead = kept, true
}
if bytes.Equal(c.pkgsKept, raw) {
return false, nil
}
if err := c.env.Packages.Put(ctx, c.env.Node, raw); err != nil {
return false, err
}
c.pkgsKept = raw
return true, nil
}
// owePackages keeps the packages due and has the loop make a pass again after the growing bound, as oweData does for the
// data items: the first write after a delivery may be refused while the state or its grant is not on the bus yet.
func (c *Composer) owePackages() string {
c.mu.Lock()
defer c.mu.Unlock()
c.pkgsDue = true
c.wait = LaterWait(c.pkgsTries)
c.pkgsTries++
c.owed.Store(true)
return fmt.Sprintf("tried again at %s (try %d)", time.Now().Add(c.wait).Format("15:04:05"), c.pkgsTries)
}
// sayPackages says a line about the packages once until it changes; "" clears it.
func (c *Composer) sayPackages(line string) {
c.mu.Lock()
same := c.pkgsSaid == line
c.pkgsSaid = line
c.mu.Unlock()
if !same && line != "" {
c.env.Say(line)
}
}
+21 -1
View File
@@ -90,6 +90,9 @@ type Env struct {
// Data is this node's keys of the engine's state data (novox/hq ADR 0577 clause 1), where each placed module's data
// items are kept after a pass that composed something new. Nil keeps none.
Data DataStore
// Packages is this node's key of the engine's state packages (novox/hq ADR 0565 §6), written after a pass that
// handed something new when the packages its compositions declare differ. Nil keeps none.
Packages PackagesStore
}
// VaultSecretChanged is the vault's event that a secret changed: a name and its generation, never a value.
@@ -161,6 +164,17 @@ type Composer struct {
dataSaid string
dataScope frombus.Scope
dataKept map[string][]byte
// pkgs is the package resources of each composition handed, by its composed key, and pkgsComposed whether this run
// handed any (ADR 0565 §6); pkgsDue, pkgsTries and pkgsSaid are as the data items' are. Under mu. pkgsKept is what
// the node's key of the state packages held when last read or written (pkgsRead), under pkgsWrite.
pkgs map[string]delivery.Composed
pkgsComposed bool
pkgsDue bool
pkgsTries int
pkgsSaid string
pkgsWrite sync.Mutex
pkgsKept []byte
pkgsRead bool
}
// Pass is how one pass went, as the node-engine's verb compose answers it.
@@ -241,7 +255,7 @@ func New(env Env) *Composer {
return &Composer{env: env, wants: map[string]delivery.Want{}, placements: map[string]delivery.Placement{}, placementKeys: delivery.NewPlacementKeys(),
records: map[string]delivery.NodeRecord{}, inOwn: map[string]bool{}, modules: map[string]delivery.Module{}, values: map[string]map[string]any{},
manifests: map[string][]byte{}, archives: map[string][]byte{}, builds: map[string]delivery.BuilderBuild{}, handed: map[string]string{}, sealedOwn: map[string]ownSealed{},
asked: make(chan chan Pass, 8)}
pkgs: map[string]delivery.Composed{}, asked: make(chan chan Pass, 8)}
}
// ComposeNow asks the composer for a pass now (the node-engine's verb compose, issue 774): the growing bound after a seat
@@ -453,6 +467,7 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er
}
}
c.publishData(ctx)
c.publishPackages(ctx)
if c.owed.Load() {
wait, _ := c.waitNow()
c.env.Say(fmt.Sprintf("the node composes again in %s: a read on the bus did not answer", wait))
@@ -903,6 +918,7 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
}
c.rev++
c.handed[key] = dig
c.pkgs[key] = packagesOf(comp)
out = append(out, agent.Update{Key: key, Value: raw, Revision: c.rev})
}
for key := range c.handed {
@@ -914,12 +930,16 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
if !now[key] {
c.rev++
delete(c.handed, key)
delete(c.pkgs, key)
out = append(out, agent.Update{Key: key, Deleted: true, Revision: c.rev})
}
}
pass.Handed = len(out)
// The data items are kept again after a pass that handed something new, and after the run's first (ADR 0577 clause 1).
c.dataDue = c.dataDue || len(out) > 0 || c.dataKept == nil
// And the node's packages after the same passes (ADR 0565 §6): written only when they differ.
c.pkgsDue = c.pkgsDue || len(out) > 0 || !c.pkgsComposed
c.pkgsComposed = true
return out
}
+8 -1
View File
@@ -393,7 +393,8 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
t.Fatal(err)
}
proxy := `{"module":"proxy","version":"1","capabilities":["systemd"],"provides":[{"name":"route","scope":"mesh"},{"name":"smtp","scope":"mesh"}],` +
`"resources":[{"id":"service","type":"service","unit":"proxy.service","state":"running","boot":"enabled"}]}`
`"resources":[{"id":"service","type":"service","unit":"proxy.service","state":"running","boot":"enabled"},` +
`{"id":"pkg","type":"package","package":"haproxy"}]}`
c3, told := fresh(env.Get)
for m, raw := range map[string]string{"de-spiegel": string(spiegel), "proxy": proxy} {
dig := delivery.ManifestDigest([]byte(raw))
@@ -430,6 +431,12 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
if u, ok := handed3[spiegelKey]; ok {
t.Errorf("held de-spiegel is handed %+v, want what was handed of it kept", u)
}
// ADR 0565 §6 (1953): the pass keeps the packages each handed composition declares, with their module, for the node's
// key of the engine's state packages.
if d, err := c3.DeclaredPackages(); err != nil || len(d.Packages) != 1 || d.Packages[0].Package != "haproxy" ||
strings.Join(d.Packages[0].Modules, ",") != "proxy" || d.Node != "n1" {
t.Errorf("the node's packages after the pass are %+v (%v)", d, err)
}
if len(*told) != 1 || !strings.HasPrefix((*told)[0], "held de-spiegel: ") || !strings.Contains((*told)[0], "${config:praktijk-name}") ||
!strings.Contains((*told)[0], "without what it contributes to route") || strings.Contains((*told)[0], "held proxy") ||
strings.Contains((*told)[0], "held lemurs") {
+3
View File
@@ -23,6 +23,9 @@
},
{
"name": "data"
},
{
"name": "packages"
}
],
"config": {
+4 -2
View File
@@ -79,8 +79,10 @@ func TestTheManifestConsumesTheBackupEvent(t *testing.T) {
// written again (no TTL: a node that stops writing keeps its last word, which the watchman reads with its time).
// ADR 0577 clause 1: and the state data, one key per module placed on the node that declares data items, kept until
// the module leaves the node; created by the grants part from this manifest as health is.
if len(parsed.State) != 2 || parsed.State[0].Name != "health" || parsed.State[0].TTLSeconds != 0 ||
parsed.State[1].Name != "data" || parsed.State[1].TTLSeconds != 0 {
// ADR 0565 clause 6: and the state packages, one key per node, its compositions' declared packages, read by pacman.
if len(parsed.State) != 3 || parsed.State[0].Name != "health" || parsed.State[0].TTLSeconds != 0 ||
parsed.State[1].Name != "data" || parsed.State[1].TTLSeconds != 0 || parsed.State[2].Name != "packages" ||
parsed.State[2].TTLSeconds != 0 {
t.Fatalf("the manifest declares the state %+v", parsed.State)
}
}