diff --git a/cmd/mesh-host/engineseat.go b/cmd/mesh-host/engineseat.go index 13afd82..c9ec77f 100644 --- a/cmd/mesh-host/engineseat.go +++ b/cmd/mesh-host/engineseat.go @@ -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 + } +} diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index 40325bf..97d0bc6 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -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) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 2cc0193..71a33d8 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -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") { diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index 4b36598..e6ce592 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -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". diff --git a/internal/link/enginepackages.go b/internal/link/enginepackages.go new file mode 100644 index 0000000..50e7adf --- /dev/null +++ b/internal/link/enginepackages.go @@ -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 +} diff --git a/internal/link/handacts.go b/internal/link/handacts.go index 6682982..2a0f538 100644 --- a/internal/link/handacts.go +++ b/internal/link/handacts.go @@ -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) diff --git a/internal/link/handover_test.go b/internal/link/handover_test.go index 63cdd81..27cb742 100644 --- a/internal/link/handover_test.go +++ b/internal/link/handover_test.go @@ -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, " ") diff --git a/internal/selfcompose/data_test.go b/internal/selfcompose/data_test.go index f613bd5..3b7112c 100644 --- a/internal/selfcompose/data_test.go +++ b/internal/selfcompose/data_test.go @@ -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() diff --git a/internal/selfcompose/packages.go b/internal/selfcompose/packages.go new file mode 100644 index 0000000..b06f61e --- /dev/null +++ b/internal/selfcompose/packages.go @@ -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) + } +} diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index cb6f1ea..101a511 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -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 } diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index fa5683c..1daff41 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -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") { diff --git a/module.json b/module.json index f0466ad..57270a3 100644 --- a/module.json +++ b/module.json @@ -23,6 +23,9 @@ }, { "name": "data" + }, + { + "name": "packages" } ], "config": { diff --git a/packaging/operator_keys_test.go b/packaging/operator_keys_test.go index 68a3d66..b86f6c4 100644 --- a/packaging/operator_keys_test.go +++ b/packaging/operator_keys_test.go @@ -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) } }