Write a provisioner client's files only on a change over the bus, never on a timer (issue 1129, ADR 0349 clause 4)
mesh/merge-gate head e7ff7c3045fc merged into trunk afb94d925e94 — pass: builds mesh-provisioner-client → ace, g14, novox, shanks; no bus step; every machi…
mesh/repo-check head e7ff7c3045fc merged into trunk afb94d925e94 — pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-…
mesh/delivery delivered

This commit is contained in:
jochen
2026-10-18 18:08:50 +02:00
parent cfb90d7b2f
commit e7ff7c3045
20 changed files with 190 additions and 119 deletions
+2 -1
View File
@@ -48,7 +48,8 @@ and the placement values its templates name — the client:
- has the node-engine write each changed file through its local write socket (`/run/nox-mesh-host/fill.sock`,
mesh-host `internal/filling`), which writes it as the declaration says (an `owned` file whole, a `seed` only when
absent, a `managed` file's own keys or block) and restarts or reloads its program as the declaration says;
- writes every file again every minute, so the node-engine puts back an `owned` file changed on disk;
- writes again on every change it watches and when its node comes back, never on a timer (novox/hq ADR 0349 clause
4): every file of the module goes through the node-engine again, so an `owned` file changed on disk is put back;
- reports `prov.applied.<node>.<module>` and keeps the same report in `mesh-provisioner_applied` (`<node>.<module>`):
`applied` with each file's action and digest, `waiting`, `failed` with the reason, or `removed`;
- when a module is taken off the node, removes its needs, its items and its report (the node-engine removes the files its
@@ -102,7 +102,7 @@ func TestNatsAModulesCodeReadsItsConfigAndIsToldOfAChange(t *testing.T) {
b.config(t, "claude-code", "managed_settings", `{"spinnerTipsEnabled":false}`)
runner := aToolRunner(t, b)
heard := runner.hear()
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
// The value as mesh-config resolved it.
if a := runner.askConfig("claude-code", "managed_settings"); a.Error != "" || string(a.Value) != `{"spinnerTipsEnabled":false}` {
@@ -134,7 +134,7 @@ func TestNatsWhatAModulesCodeMayNotReadIsRefusedWithTheReason(t *testing.T) {
// Resolved for a module the composition does not place on this machine.
b.config(t, "dunst", "font-size", "13")
runner := aToolRunner(t, b)
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
for _, c := range []struct{ ask, refused string }{
{`{"module":"claude-code","kind":"config","key":"no_such_key"}`, "declares no config value no_such_key"},
@@ -5,7 +5,6 @@ import (
"os"
"path/filepath"
"testing"
"time"
)
// The bar files i3status-rust's bars come to, filled by this client from power's items of the collection
@@ -74,7 +73,7 @@ func TestNatsTheBarsFilledFromPowersItemsAreWhatTheOldContributionMade(t *testin
e := startEngine(t, map[string]string{"/bars/bottom.toml": "owned", "/bars/top.toml": "owned"})
b.assign(t, "i3status-rust", bars)
b.assign(t, "power", contrib)
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
for _, which := range []string{"bottom", "top"} {
golden, err := os.ReadFile(filepath.Join("testdata", "bars", name+"."+which+"-bar.toml"))
if err != nil {
@@ -5,7 +5,6 @@ import (
"os"
"strings"
"testing"
"time"
"github.com/nats-io/nats.go/jetstream"
)
@@ -56,7 +55,7 @@ func giteaRun(t *testing.T, entry string, extra string) (*engine, *aBus, *report
}
b.kv["postgres_bound"] = kv
b.put(t, "postgres_bound", node+".gitea.postgres-database", entry)
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
return e, b, heard
}
@@ -109,7 +108,7 @@ func TestNatsAHandedGiteasOriginsAreFilledFromTheRouteProvidersEntry(t *testing.
b.put(t, "route-adapter_bound", node+".gitea.route", `{"provider":"novox","provider-module":"route-adapter",`+
`"provision":"route","consumer":"g14","module":"gitea","as":"mesh_g14_gitea",`+
`"values":{"name":"git.example.org","internal-name":"git.novox.internal","name-web":"git.example.org"}}`)
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
eventually(t, "origins is filled to the controller's text", func() bool {
return e.read(t, originsPath) == "https://git.example.org\n"
})
@@ -139,7 +138,7 @@ func TestNatsAHandedGiteasOriginsAreFilledFromRouteProxysEntryOnNovox(t *testing
b.put(t, "route-proxy_bound", node+".gitea.route", `{"provider":"novox","provider-module":"route-proxy",`+
`"provision":"route","consumer":"g14","module":"gitea","as":"mesh_g14_gitea",`+
`"values":{"name":"git.example.org","internal-name":"git.novox.internal","name-web":"git.example.org"}}`)
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
eventually(t, "origins is filled to the controller's text", func() bool {
return e.read(t, originsPath) == "https://git.example.org\n"
})
@@ -3,7 +3,6 @@ package main
import (
"bytes"
"context"
"crypto/sha256"
"encoding/json"
"errors"
"fmt"
@@ -25,10 +24,10 @@ import (
// current values first), fills the templates only once every value is there — otherwise it keeps the files there
// are and reports what it waits for — has the node-engine write each changed file through its narrow local verb,
// which makes the program take the change as the file declares, and reports `prov.applied` (applied, or failed
// with the reason). It does so again on every change it watches, writes every file again on a drift pass so the
// node-engine puts an owned file back, and when its node comes back it starts from the current values. A module
// taken off the node loses its needs and its report; the node-engine removes the files the declaration no longer
// names.
// with the reason). It writes again on every change it watches and when its node comes back, never on a timer: each
// time, every file of the module goes through the node-engine again, which puts an owned file changed on disk
// back. A module taken off the node loses its needs and its report; the node-engine removes the files the
// declaration no longer names.
//
// The owners it fills from are placement (the controller's, in the module's assignment), config
// (mesh-config_resolved) and the vault, asked for each secret the module's plan names (secrets.go), and for each
@@ -118,19 +117,12 @@ type Client struct {
Secrets Secrets
// Templates fetches and keeps the templates archives.
Templates *templateStore
// Drift is how often every applied module's files are written again, so an owned file changed on disk is put
// back; zero is a minute.
Drift time.Duration
// Now dates the reports; nil is time.Now.
Now func() time.Time
// watchBound watches a provider's bucket of entries for this machine's keys, once per run (consumers.go); set by
// Run.
watchBound func(bucket string) error
// What it last had written, by path: the content's digest. Kept across runs of Run, so a node coming back
// writes only what moved — and the drift pass writes everything again anyway.
written map[string][32]byte
}
// state is what one run holds, from its watches.
@@ -173,9 +165,6 @@ type update struct {
// Run serves until ctx ends or a watch stops, which it returns: run again, it starts from the current values.
func (c *Client) Run(ctx context.Context) error {
if c.written == nil {
c.written = map[string][32]byte{}
}
ctx, cancel := context.WithCancel(ctx)
defer cancel()
kvs := map[string]jetstream.KeyValue{}
@@ -310,7 +299,7 @@ func (c *Client) Run(ctx context.Context) error {
}
// Contributors first, so a collector starting with its node is filled from the items there are now.
for _, m := range contributorsFirst(s.plans) {
c.reconcile(ctx, kvs, s, m, false)
c.reconcile(ctx, kvs, s, m)
}
for _, m := range sortedKeys(s.refused) {
c.report(ctx, kvs, s, Report{Module: m, Status: StatusFailed, Reason: s.refused[m]})
@@ -324,22 +313,12 @@ func (c *Client) Run(ctx context.Context) error {
c.tell(m, Changed{Kind: AskConfig})
}
drift := c.Drift
if drift <= 0 {
drift = time.Minute
}
tick := time.NewTicker(drift)
defer tick.Stop()
for {
select {
case <-ctx.Done():
return ctx.Err()
case err := <-stopped:
return err
case <-tick.C:
for _, m := range sortedKeys(s.plans) {
c.reconcile(ctx, kvs, s, m, true)
}
case u := <-updates:
// What moved together is filled together: every update waiting is taken before anything is filled.
dirty := map[string]bool{}
@@ -375,7 +354,7 @@ func (c *Client) Run(ctx context.Context) error {
case s.refused[m] != "":
c.report(ctx, kvs, s, Report{Module: m, Status: StatusFailed, Reason: s.refused[m]})
case hasPlan(s, m):
c.reconcile(ctx, kvs, s, m, false)
c.reconcile(ctx, kvs, s, m)
default:
c.leave(ctx, kvs, s, m)
}
@@ -597,10 +576,10 @@ func (c *Client) leave(ctx context.Context, kvs map[string]jetstream.KeyValue, s
removed)
}
// reconcile fills one module's files from the current values, and says what came of it. force writes every file
// again, whether it moved or not: the drift pass.
func (c *Client) reconcile(ctx context.Context, kvs map[string]jetstream.KeyValue, s *state, module string,
force bool) {
// reconcile fills one module's files from the current values, and says what came of it. Every file of the module
// goes through the node-engine again, whether it moved or not (ADR 0349 clause 5), which puts an owned file changed
// on disk back.
func (c *Client) reconcile(ctx context.Context, kvs map[string]jetstream.KeyValue, s *state, module string) {
p := s.plans[module]
r := Report{Module: module, Templates: p.Templates.Digest}
texts, err := c.Templates.load(ctx, p.Templates)
@@ -682,23 +661,15 @@ func (c *Client) reconcile(ctx context.Context, kvs map[string]jetstream.KeyValu
moved := false
for _, f := range targets {
content := []byte(render(byFile[f.Path], values))
sum := sha256.Sum256(content)
fa := FileApplied{Path: f.Path, Takes: f.Takes}
if !force && c.written[f.Path] == sum {
fa.Action = "unchanged"
r.Files = append(r.Files, fa)
continue
}
a, err := c.Fill.Fill(ctx, f.Path, content)
if err == nil {
err = errFill(a)
}
if err != nil {
delete(c.written, f.Path)
failed = append(failed, fmt.Sprintf("%s: %v", f.Path, err))
continue
}
c.written[f.Path] = sum
fa.Action, fa.Digest = a.Action, a.Digest
if a.Action != "unchanged" && a.Action != "kept" {
moved = true
@@ -711,12 +682,8 @@ func (c *Client) reconcile(ctx context.Context, kvs map[string]jetstream.KeyValu
} else {
r.Status = StatusApplied
}
if force && !moved && r.Status == StatusApplied {
// The drift pass found every file as it should be: nothing to say.
return
}
if force || moved {
// Said again, even when it reads the same: a file was put back.
if moved {
// Said again, even when it reads the same as the last report: a file was put back.
delete(s.reported, module)
}
c.report(ctx, kvs, s, r)
@@ -330,7 +330,7 @@ func templates(t *testing.T, files map[string]string) (Templates, *httptest.Serv
}
// aClient runs a client on g14's account until the returned stop is called.
func aClient(t *testing.T, b *aBus, e *engine, state string, drift time.Duration) (*Client, func()) {
func aClient(t *testing.T, b *aBus, e *engine, state string) (*Client, func()) {
t.Helper()
nc, err := nats.Connect(b.url, nats.UserInfo("provisioner."+node, "x"),
nats.CustomInboxPrefix("_INBOX.provisioner."+node),
@@ -349,7 +349,7 @@ func aClient(t *testing.T, b *aBus, e *engine, state string, drift time.Duration
t.Fatal(err)
}
secrets.Within = 2 * time.Second
c := &Client{Node: node, JS: js, Conn: nc, Drift: drift, Secrets: secrets,
c := &Client{Node: node, JS: js, Conn: nc, Secrets: secrets,
Fill: SocketFiller{Socket: e.socket, Busy: 10 * time.Millisecond, Tries: 3},
Templates: &templateStore{dir: state, fetch: HTTPFetch}}
ctx, cancel := context.WithCancel(context.Background())
@@ -399,7 +399,7 @@ func TestNatsATemplateFillsFromEachOwnersKeys(t *testing.T) {
e := startEngine(t, map[string]string{dunstrc: "owned"})
b.config(t, "dunst", "font-size", "13")
b.assign(t, "dunst", dunst(t))
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
want := "[global]\nfont = Monospace 13\nicons = /home/nox/.icons\n"
eventually(t, "dunstrc is filled", func() bool { return e.read(t, dunstrc) == want })
@@ -427,7 +427,7 @@ func TestNatsAMissingValueKeepsTheOldFileAndSaysWhatItWaitsFor(t *testing.T) {
heard := b.listen(t)
e := startEngine(t, map[string]string{dunstrc: "owned"})
b.assign(t, "dunst", dunst(t))
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
// No value yet: nothing written, and the wait said.
eventually(t, "prov.applied says it waits", func() bool {
@@ -463,7 +463,7 @@ func TestNatsAChangeRefillsWritesOnceAndTheProgramTakesIt(t *testing.T) {
e := startEngine(t, map[string]string{dunstrc: "owned"})
b.config(t, "dunst", "font-size", "13")
b.assign(t, "dunst", dunst(t))
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
eventually(t, "dunstrc is filled", func() bool { return strings.Contains(e.read(t, dunstrc), "Monospace 13") })
b.config(t, "dunst", "font-size", "15")
@@ -472,13 +472,14 @@ func TestNatsAChangeRefillsWritesOnceAndTheProgramTakesIt(t *testing.T) {
r, _ := heard.last("dunst")
return r.Status == StatusApplied && len(r.Files) == 1 && r.Files[0].Action == "updated"
})
// The same value again, and a key the template does not name: nothing written.
// The same value again, and a key the template does not name: the next fill asks the node-engine for the file
// again (ADR 0349 clause 5), which answers unchanged, so the program is not reloaded for it.
b.config(t, "dunst", "font-size", "15")
b.config(t, "dunst", "geometry", `"300x5"`)
time.Sleep(300 * time.Millisecond)
if moved, asked := e.counts(dunstrc); moved != 2 || asked != 2 {
t.Errorf("two values wrote dunstrc %d times (asked %d); want 2 and 2: the program is reloaded once a change",
moved, asked)
if moved, asked := e.counts(dunstrc); moved != 2 || asked != 3 {
t.Errorf("two values wrote dunstrc %d times (asked %d); want 2 and 3: the program is reloaded once a change, "+
"the file asked again each fill", moved, asked)
}
}
@@ -494,7 +495,7 @@ func TestNatsAnOwnedFileIsWrittenBackASeedLeftAndAManagedFileKeepsTheProgramsLin
{ID: "rc", Path: dunstrc, Template: "dunstrc", OwnerAtRuntime: "owned"},
{ID: "first", Path: seed, Template: "first.conf", OwnerAtRuntime: "seed"},
{ID: "settings", Path: managed, Template: "settings.ini", OwnerAtRuntime: "managed", Takes: "read-on-use"}}})
aClient(t, b, e, t.TempDir(), 200*time.Millisecond)
aClient(t, b, e, t.TempDir())
eventually(t, "all three are placed", func() bool {
r, _ := heard.last("app")
return r.Status == StatusApplied
@@ -507,24 +508,79 @@ func TestNatsAnOwnedFileIsWrittenBackASeedLeftAndAManagedFileKeepsTheProgramsLin
t.Fatal(err)
}
}
eventually(t, "the owned file is written back", func() bool { return e.read(t, dunstrc) == "font = 13\n" })
// 1129/AC3: with nothing reaching the module over the bus, an owned file changed on disk stays as the program
// left it (well over any former drift interval).
time.Sleep(1500 * time.Millisecond)
if got := e.read(t, dunstrc); got != "font = 99\n" {
t.Errorf("an owned file changed on disk, with nothing over the bus, reads %q, want the program's", got)
}
// 1129/AC2: a config value of the module changes over the bus, so every file of the module is written again
// through the node-engine (ADR 0349 clause 5): the owned file is put back, the seed keeps the program's text,
// and the managed file keeps the program's lines with the mesh's part.
b.config(t, "app", "font-size", "14")
eventually(t, "the owned file is written back", func() bool { return e.read(t, dunstrc) == "font = 14\n" })
eventually(t, "prov.applied says it was put back", func() bool {
r, _ := heard.last("app")
return r.Status == StatusApplied && len(r.Files) == 3 && r.Files[0].Action == "corrected"
})
time.Sleep(500 * time.Millisecond)
if got := e.read(t, seed); got != "start = mine\n" {
t.Errorf("the seed file is %q: written once, then the program's", got)
}
if got := e.read(t, managed); !strings.Contains(got, "theme = dark\n") || !strings.Contains(got, "recent = a,b\n") ||
!strings.Contains(got, "size = 13\n") {
!strings.Contains(got, "size = 14\n") {
t.Errorf("the managed file is %q: the mesh's block and the program's own lines", got)
}
// The drift pass that finds every file as it should be says nothing.
_, before := heard.last("app")
// A further reconcile that moves nothing is said, because it reads unchanged where the last read corrected; the
// one after that, which reads the same as it, is not said again.
_, afterCorrected := heard.last("app")
b.config(t, "app", "geometry", `"300x5"`) // a value the templates do not name: a reason to look, not to move
eventually(t, "a reconcile that moves nothing is said once, reading unchanged", func() bool {
_, n := heard.last("app")
return n == afterCorrected+1
})
_, afterUnchanged := heard.last("app")
b.config(t, "app", "geometry", `"300x5"`)
time.Sleep(600 * time.Millisecond)
if _, after := heard.last("app"); after != before {
t.Errorf("drift passes that moved nothing said %d reports", after-before)
if _, after := heard.last("app"); after != afterUnchanged {
t.Errorf("a reconcile whose report reads the same as the last one said %d more reports", after-afterUnchanged)
}
}
// 1129/AC1: with every module filled and nothing reaching the client over the bus, it asks the node-engine for no
// write, asks the vault nothing, and publishes no report — there is no timer that asks again on its own.
func TestNatsWithNothingOverTheBusItAsksForNoWriteNoSecretAndNoReport(t *testing.T) {
b := startBus(t)
heard := b.listen(t)
vault := startVault(t, b)
vault.keep("own/g14/app/app-token", "tok", false)
vault.keep("own/g14/app/db-password", "pw", false)
e := startEngine(t, map[string]string{appConf: "owned", appDB: "owned"})
b.config(t, "app", "font-size", "13")
b.assign(t, "app", app(t))
aClient(t, b, e, t.TempDir())
eventually(t, "app is filled", func() bool {
r, _ := heard.last("app")
return r.Status == StatusApplied
})
_, askedBefore := e.counts(appConf)
vaultBefore := vault.times("own/g14/app/app-token") + vault.times("own/g14/app/db-password")
_, reportsBefore := heard.last("app")
// Well over any former drift interval used in these tests (100ms, 200ms): nothing reaches the client.
time.Sleep(1500 * time.Millisecond)
if _, askedAfter := e.counts(appConf); askedAfter != askedBefore {
t.Errorf("the node-engine was asked to write %s again, with nothing over the bus", appConf)
}
if n := vault.times("own/g14/app/app-token") + vault.times("own/g14/app/db-password"); n != vaultBefore {
t.Error("the vault was asked again, with nothing over the bus")
}
if _, after := heard.last("app"); after != reportsBefore {
t.Error("a report was published again, with nothing over the bus")
}
}
@@ -534,7 +590,7 @@ func TestNatsAModuleTakenOffTheNodeLosesItsNeedsAndItsReport(t *testing.T) {
e := startEngine(t, map[string]string{dunstrc: "owned"})
b.config(t, "dunst", "font-size", "13")
b.assign(t, "dunst", dunst(t))
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
eventually(t, "the needs are recorded", func() bool { return len(b.keys(t, NeedsBucket, node+".dunst.")) == 2 })
b.del(t, AssignmentsBucket, node+".dunst")
@@ -557,7 +613,7 @@ func TestNatsANodeComingBackFillsFromTheCurrentValues(t *testing.T) {
b.assign(t, "zsh", Plan{Templates: tpl, Files: []PlannedFile{{ID: "zshrc", Path: zshrc, Template: "zshrc",
OwnerAtRuntime: "owned"}}})
state := t.TempDir()
_, stop := aClient(t, b, e, state, time.Hour)
_, stop := aClient(t, b, e, state)
eventually(t, "both are filled", func() bool {
return strings.Contains(e.read(t, dunstrc), "Monospace 13") && e.read(t, zshrc) == "PS1=%\n"
})
@@ -568,7 +624,7 @@ func TestNatsANodeComingBackFillsFromTheCurrentValues(t *testing.T) {
b.del(t, AssignmentsBucket, node+".zsh")
heard := b.listen(t)
aClient(t, b, e, state, time.Hour)
aClient(t, b, e, state)
eventually(t, "dunstrc has the value set while it was away", func() bool {
return strings.Contains(e.read(t, dunstrc), "Monospace 17")
})
@@ -585,7 +641,7 @@ func TestNatsWhatCannotBeFilledFailsWithTheReason(t *testing.T) {
b := startBus(t)
heard := b.listen(t)
e := startEngine(t, map[string]string{dunstrc: "owned", "/etc/app.conf": "owned"})
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
b.put(t, AssignmentsBucket, node+".broken", `{"node":"g14","module":"broken","provisioning":{"templates":`+
`{"source":"x","digest":"sha256:00"},"files":[],"surprise":1}}`)
@@ -638,7 +694,7 @@ func TestNatsResticsFilesFillFromTheBackupPlacement(t *testing.T) {
Placement: map[string]string{"backup:lines": lines, "backup:data": declared}}
}
b.assign(t, "restic", plan("# a\n/srv/a\n# b\n/srv/b\n", "# a\nstate\n"))
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
head := "# What the modules on this machine back up, composed by the mesh. Do not edit.\n"
eventually(t, "backups.conf is filled", func() bool { return e.read(t, backups) == head+"# a\n/srv/a\n# b\n/srv/b\n" })
eventually(t, "data.conf is filled", func() bool {
@@ -3,7 +3,6 @@ package main
import (
"strings"
"testing"
"time"
)
// An item may carry attributes and a module may give several named items to one collection; a placeholder selecting
@@ -44,7 +43,7 @@ func TestNatsASelectingPlaceholderTakesOnlyTheItemsCarryingItsAttributesInOrder(
b.assign(t, "another", blocksOf(t, map[string]PlannedItem{
"disk": {Collection: "blocks", Template: "disk", Attributes: map[string]string{"bar": "bottom", "place": "status", "order": "20"}},
}, map[string]string{"disk": "disk\n"}))
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
want := "[bottom]\n# another\ndisk\n# power\ncpu\n# power\nbattery\n" +
"[top]\n# power\nclock\n" +
@@ -5,7 +5,6 @@ import (
"encoding/json"
"strings"
"testing"
"time"
)
// A collection's test of novox/hq ADR 0349 ("How it is checked", design 57 L1, issue 684): zsh's file holds every
@@ -40,7 +39,7 @@ func TestNatsZshsFileHoldsEveryPlacedContributorsItemAndLosesOneWhoseContributor
b.config(t, "fzf", "binding", `"^T"`)
fzf := contributor(t, "zsh-widgets", "bindkey ${config:binding} fzf-file-widget\n")
b.assign(t, "fzf", fzf)
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
every := "# zsh\n# mesh-cli\nalias nox=mesh-cli\n# powerlevel10k\nsource /home/nox/.p10k.zsh\n# then\n" +
"bindkey ^T fzf-file-widget\n# end\n"
@@ -95,13 +94,13 @@ func TestNatsACollectorStartingWithItsNodeHoldsTheItemsAlreadyThereAndOnesRecord
b.assign(t, "zsh", zshCollector(t))
b.assign(t, "mesh-cli", contributor(t, "shell:zsh:normal", "alias nox=mesh-cli\n"))
state := t.TempDir()
_, stop := aClient(t, b, e, state, time.Hour)
_, stop := aClient(t, b, e, state)
eventually(t, "zshrc holds mesh-cli's item", func() bool { return strings.Contains(e.read(t, zshrc), "alias nox") })
stop()
b.del(t, AssignmentsBucket, node+".mesh-cli")
b.assign(t, "zsh-autosuggestions", contributor(t, "shell:zsh:normal", "source suggestions.zsh\n"))
aClient(t, b, e, state, time.Hour)
aClient(t, b, e, state)
want := "# zsh\n# zsh-autosuggestions\nsource suggestions.zsh\n# then\n# end\n"
eventually(t, "zshrc is filled from the items now", func() bool { return e.read(t, zshrc) == want })
eventually(t, "the item of the module that left while away is removed", func() bool {
@@ -119,7 +118,7 @@ func TestNatsACollectionNotPlannedAndAnItemNamingASecretFail(t *testing.T) {
p.Collections = p.Collections[:1]
b.assign(t, "zsh", p)
b.assign(t, "leaky", contributor(t, "shell:zsh:normal", "export TOKEN=${secret:token}\n"))
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
for module, why := range map[string]string{"zsh": "the plan names no collection zsh-widgets",
"leaky": "where no secret is"} {
eventually(t, module+" fails", func() bool {
@@ -107,7 +107,7 @@ func TestNatsAProvidersConsumersAreFilledFromTheVaultAndFollowEachBindAndRemake(
ProviderModule: "postgres", CredentialForm: "shared"})
e := startEngine(t, map[string]string{consumersFile: "owned"})
b.assign(t, "postgres", postgresPlan(t, "${consumers:postgres-database}"))
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
letta := listed{Consumer: "ace", Module: "letta", Credential: pairOf("ace", "letta", ""),
Value: "letta-pw-never-on-the-bus"}
@@ -174,7 +174,7 @@ func TestNatsAProviderWithNoConsumerIsFilledWithAnEmptyList(t *testing.T) {
startVault(t, b)
e := startEngine(t, map[string]string{consumersFile: "owned"})
b.assign(t, "postgres", postgresPlan(t, "${consumers:postgres-database}"))
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
eventually(t, "the file holds an empty list", func() bool { return e.read(t, consumersFile) == "[]" })
}
@@ -203,7 +203,7 @@ func TestNatsAConsumersBoundValuesAreFilledFromItsProvidersEntry(t *testing.T) {
heard := b.listen(t)
e := startEngine(t, map[string]string{appBound: "owned"})
b.assign(t, "app", appPlan(t))
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
// No bucket of the provider's yet: waited for, and said.
eventually(t, "the consumer waits for its provider's bucket", func() bool {
@@ -217,7 +217,7 @@ func TestNatsAConsumersBoundValuesAreFilledFromItsProvidersEntry(t *testing.T) {
b.kv["postgres_bound"] = kv
// The provider has not answered it yet: waited for.
b.put(t, "postgres_bound", "ace.app.postgres-database", `{"values":{"host":"elsewhere"}}`)
b.assign(t, "app", appPlan(t)) // a change of its own, so it looks again before the drift pass
b.assign(t, "app", appPlan(t)) // a change of its own is the only reason it looks again: there is no timer
eventually(t, "the consumer waits for its own entry", func() bool {
r, _ := heard.last("app")
return r.Status == StatusWaiting && strings.Contains(strings.Join(r.WaitsFor, " "), "postgres on novox")
@@ -254,7 +254,7 @@ func TestNatsAProviderListsAConsumerWhoseEntryNamesItsLoginFile(t *testing.T) {
ProviderModule: "postgres", CredentialForm: "issued", Path: "/s/forge.secret"})
e := startEngine(t, map[string]string{consumersFile: "owned"})
b.assign(t, "postgres", postgresPlan(t, "${consumers:postgres-database}"))
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
eventually(t, "the consumer with a login file is listed, the per-holder one is not", func() bool {
return sameList(consumersIn(t, e), listed{Consumer: "ace", Module: "nextcloud",
Credential: pairOf("ace", "nextcloud", ""), Value: "nc-pw-never-on-the-bus"})
@@ -0,0 +1,52 @@
//go:build integration
package main
import (
"context"
"encoding/json"
"os"
"path/filepath"
"strings"
"testing"
"time"
)
// 1129/AC5 (integration): the client as main/serve would run it — reading its credential from disk, dialling a
// real bus, and asking a real node-engine-shaped socket to write — fills a module once and then asks for no write
// for three seconds: there is no timer that asks again on its own (ADR 0349 clause 4). Compiled (`go vet -tags
// integration ./...`), never run here: the lead runs it against a real bus through mesh-controller.try.
func TestIntegrationServeFillsOnceAndThenAsksForNoWrite(t *testing.T) {
b := startBus(t)
e := startEngine(t, map[string]string{dunstrc: "owned"})
b.config(t, "dunst", "font-size", "13")
b.assign(t, "dunst", dunst(t))
dir := t.TempDir()
credential := filepath.Join(dir, "credential.json")
body, err := json.Marshal(Credential{URL: b.url, Node: node, User: "provisioner." + node, Password: "x"})
if err != nil {
t.Fatal(err)
}
if err := os.WriteFile(credential, body, 0o600); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithCancel(context.Background())
done := make(chan error, 1)
go func() { done <- serve(ctx, credential, e.socket, filepath.Join(dir, "state")) }()
t.Cleanup(func() {
cancel()
<-done
})
eventually(t, "dunstrc is filled", func() bool { return strings.Contains(e.read(t, dunstrc), "Monospace 13") })
_, asked := e.counts(dunstrc)
// Well over any former drift interval (a minute by default, before this issue removed the --drift flag).
time.Sleep(3 * time.Second)
if _, after := e.counts(dunstrc); after != asked {
t.Errorf("the node-engine was asked to write %s again %d time(s) within three seconds, with nothing over "+
"the bus: there is no timer", dunstrc, after-asked)
}
}
@@ -17,7 +17,7 @@ func TestNatsAnEndedAssignmentTakesAModuleOffTheNodeLikeARemovedKey(t *testing.T
b.config(t, "dunst", "font-size", "13")
p := dunst(t)
b.assign(t, "dunst", p)
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
eventually(t, "the needs are recorded", func() bool { return len(b.keys(t, NeedsBucket, node+".dunst.")) == 2 })
b.put(t, AssignmentsBucket, node+".dunst", fmt.Sprintf(`{"node":%q,"module":"dunst","assigned":"2026-10-14T20:00:00Z",`+
@@ -37,7 +37,7 @@ func TestNatsAnAssignmentWithTheNewFieldsIsFilledUnderTheSameKey(t *testing.T) {
b.config(t, "dunst", "font-size", "13")
p := dunst(t)
b.assign(t, "dunst", p)
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
eventually(t, "the needs are recorded", func() bool { return len(b.keys(t, NeedsBucket, node+".dunst.")) == 2 })
b.put(t, AssignmentsBucket, node+".dunst", fmt.Sprintf(`{"node":%q,"module":"dunst","assigned":"2026-10-14T20:00:00Z",`+
`"build":"b-1","artifact":{"name":"dunst","version":"1"},"commit":"abc","rollout":"r-1",`+
@@ -6,7 +6,6 @@ import (
"path/filepath"
"strings"
"testing"
"time"
)
// The session environment's collection (design 57 L3, issue 795), filled by this client from typed items, is what the
@@ -87,7 +86,7 @@ func TestNatsTheEnvironmentFilledFromTypedItemsIsWhatTheOldContributionMade(t *t
for m, p := range contributors {
b.assign(t, m, p)
}
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
for file, golden := range map[string]string{"/env/environment.sh": "posix", "/env/50-mesh.conf": "systemd"} {
want, err := os.ReadFile(filepath.Join("testdata", "environment", name+"."+golden))
if err != nil {
@@ -170,7 +169,7 @@ func TestNatsTheEnvironmentOfTheCataloguesModulesIsWhatTheOldFormComposed(t *tes
tp, _ := templates(t, texts)
b.assign(t, m, Plan{Templates: tp, Items: planned, Placement: facts})
}
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
for file, golden := range map[string]string{"/env/environment.sh": "posix", "/env/50-mesh.conf": "systemd"} {
want, err := os.ReadFile(filepath.Join("testdata", "environment", name+"."+golden))
if err != nil {
@@ -6,7 +6,6 @@ import (
"path/filepath"
"strings"
"testing"
"time"
)
// i3's window-manager configuration, filled by this client from the items of the collection
@@ -73,7 +72,7 @@ func TestNatsI3sConfigFilledFromItsContributorsItemsIsWhatTheOldContributionsMad
tp, _ := templates(t, texts)
b.assign(t, m, Plan{Templates: tp, Items: planned, Placement: facts})
}
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
want, err := os.ReadFile(filepath.Join("testdata", "i3", name+".config"))
if err != nil {
t.Fatal(err)
@@ -82,7 +82,6 @@ func main() {
socket := flag.String("socket", DefaultSocket, "the node-engine's socket for writing a filled file")
stateDir := flag.String("state", os.Getenv("MESH_PROVISIONER_STATE"),
"where the client keeps the templates it fetched (MESH_PROVISIONER_STATE)")
drift := flag.Duration("drift", time.Minute, "how often every filled file is written again")
flag.Parse()
if *credential == "" || *stateDir == "" {
logf("usage: mesh-provisioner-client -credential <file> -state <dir>")
@@ -90,7 +89,7 @@ func main() {
}
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
if err := serve(ctx, *credential, *socket, *stateDir, *drift); err != nil && ctx.Err() == nil {
if err := serve(ctx, *credential, *socket, *stateDir); err != nil && ctx.Err() == nil {
logf("[provisioner-client] %v", err)
os.Exit(1)
}
@@ -98,7 +97,7 @@ func main() {
// serve connects and runs the client until ctx ends. A run that stops — a watch ended, the bus came back — is run
// again from the current values, after a wait that grows to a minute.
func serve(ctx context.Context, credential, socket, stateDir string, drift time.Duration) error {
func serve(ctx context.Context, credential, socket, stateDir string) error {
c, err := readCredential(credential)
if err != nil {
return err
@@ -134,7 +133,7 @@ func serve(ctx context.Context, credential, socket, stateDir string, drift time.
if err != nil {
return err
}
client := &Client{Node: c.Node, JS: js, Conn: nc, Drift: drift, Secrets: secrets,
client := &Client{Node: c.Node, JS: js, Conn: nc, Secrets: secrets,
Fill: SocketFiller{Socket: socket, Busy: time.Second, Tries: 10},
Templates: &templateStore{dir: filepath.Join(stateDir, "templates"), fetch: HTTPFetch}}
wait := time.Second
@@ -6,7 +6,6 @@ import (
"path/filepath"
"regexp"
"testing"
"time"
)
// The collectors design 57 L2 moves (issue 681): zsh, power and triggerhappy, each with every contributor the
@@ -93,7 +92,7 @@ func TestNatsTheCollectorsL2MovedFillFromTheirContributorsItemsAsTheyWereCompose
contributors++
}
}
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
eventually(t, c.module+"'s file as the controller composed it", func() bool {
return e.read(t, path) == want
})
@@ -5,7 +5,6 @@ import (
"encoding/json"
"os"
"testing"
"time"
"github.com/nats-io/nats.go/jetstream"
)
@@ -93,7 +92,7 @@ func TestNatsAMovedGiteaAndClaudeCodeHaveTheirFilesFilledFromTheBus(t *testing.T
b.put(t, "node-tools_bound", node+".claude-code.mcp-endpoint", `{"provider":"g14","provider-module":"node-tools",`+
`"provision":"mcp-endpoint","consumer":"g14","module":"claude-code","as":"mesh_g14_claude-code",`+
`"values":{"port":4270}}`)
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
for path, want := range map[string]string{
serverEnv: giteaBoundExpected("tok-never-on-the-bus", "pw-never-on-the-bus"),
@@ -5,7 +5,6 @@ import (
"fmt"
"strings"
"testing"
"time"
)
// A module only pulled in by another on this node has a value of its own in the assignments mirror (mesh-controller
@@ -53,7 +52,7 @@ func TestNatsAPulledInModuleIsFilledWithItsSecretAndItsItemIsRecorded(t *testing
b.pullIn(t, "app", "desk", app(t))
b.assign(t, "zsh", zshCollector(t))
b.pullIn(t, "mesh-cli", "desk", contributor(t, "shell:zsh:normal", "alias nox=mesh-cli\n"))
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
eventually(t, "the pulled-in app.conf is filled with its secret", func() bool {
return e.read(t, appConf) == "token = "+token+"\nfont = 13\n"
@@ -6,7 +6,6 @@ import (
"strings"
"sync"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/provider"
"github.com/nats-io/nats.go/jetstream"
@@ -91,7 +90,7 @@ func routeRun(t *testing.T, module, id, path, templatePath, entryKey, firstEntry
if firstEntry != "" {
b.put(t, "route-proxy_bound", node+"."+entryKey, firstEntry)
}
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
return e, b, heard
}
@@ -160,7 +160,7 @@ func TestNatsAMovedModulesFileIsFilledWithItsSecretFromTheVault(t *testing.T) {
e := startEngine(t, map[string]string{appConf: "owned", appDB: "owned"})
b.config(t, "app", "font-size", "13")
b.assign(t, "app", app(t))
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
eventually(t, "app.conf is filled with its secret", func() bool {
return e.read(t, appConf) == "token = "+token+"\nfont = 13\n"
@@ -197,7 +197,7 @@ func TestNatsASecretRotatedInTheVaultReachesItsFileWithoutANewDeclaration(t *tes
e := startEngine(t, map[string]string{appConf: "owned", appDB: "owned"})
b.config(t, "app", "font-size", "13")
b.assign(t, "app", app(t))
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
eventually(t, "app.conf is filled", func() bool { return strings.Contains(e.read(t, appConf), "token = first\n") })
assigned, err := b.kv[AssignmentsBucket].Get(context.Background(), node+".app")
if err != nil {
@@ -243,7 +243,7 @@ func TestNatsASecretTheVaultDoesNotHoldYetIsWaitedFor(t *testing.T) {
e := startEngine(t, map[string]string{appConf: "owned", appDB: "owned"})
b.config(t, "app", "font-size", "13")
b.assign(t, "app", app(t))
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
eventually(t, "prov.applied says it waits for the secret", func() bool {
r, _ := heard.last("app")
@@ -262,6 +262,8 @@ func TestNatsASecretTheVaultDoesNotHoldYetIsWaitedFor(t *testing.T) {
})
}
// 1129/AC4: a secret the vault refuses is asked for once and not again while nothing about it changes, even
// though several unrelated changes over the bus each reconcile the module again (never a timer).
func TestNatsARefusalFailsTheModuleWithTheVaultsReasonAndIsNotAskedEveryPass(t *testing.T) {
b := startBus(t)
heard := b.listen(t)
@@ -272,15 +274,20 @@ func TestNatsARefusalFailsTheModuleWithTheVaultsReasonAndIsNotAskedEveryPass(t *
e := startEngine(t, map[string]string{appConf: "owned", appDB: "owned"})
b.config(t, "app", "font-size", "13")
b.assign(t, "app", app(t))
aClient(t, b, e, t.TempDir(), 100*time.Millisecond)
aClient(t, b, e, t.TempDir())
eventually(t, "prov.applied says it failed, with the vault's reason", func() bool {
r, _ := heard.last("app")
return r.Status == StatusFailed && strings.Contains(r.Reason, "names no secret app-token")
})
time.Sleep(600 * time.Millisecond)
// Several changes reach the module, each a reason to reconcile it again.
for _, size := range []string{"14", "15", "16"} {
b.config(t, "app", "font-size", size)
time.Sleep(100 * time.Millisecond)
}
time.Sleep(200 * time.Millisecond)
if n := vault.times("own/g14/app/app-token"); n != 1 {
t.Errorf("a refused secret was asked %d times in six drift passes, want once", n)
t.Errorf("a refused secret was asked %d times across several reconciles, want once", n)
}
if _, asked := e.counts(appConf); asked != 0 {
t.Error("a file whose secret was refused was written")
@@ -293,7 +300,7 @@ func TestNatsAVaultThatDoesNotAnswerIsWaitedFor(t *testing.T) {
e := startEngine(t, map[string]string{appConf: "owned", appDB: "owned"})
b.config(t, "app", "font-size", "13")
b.assign(t, "app", app(t))
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
eventually(t, "prov.applied says it waits for the vault", func() bool {
r, _ := heard.last("app")
return r.Status == StatusWaiting && len(r.WaitsFor) == 2 && strings.Contains(r.WaitsFor[0], "nothing answers")
@@ -355,7 +362,7 @@ func TestNatsAModulesRequiredCredentialIssuedForItIsFilledFromTheVaultAndFollows
vault.keep(forgeHolder, first, false)
e := startEngine(t, map[string]string{forgeFile: "owned", settingsFile: "owned"})
b.assign(t, "claude-code", claudeCode(t))
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
eventually(t, "gitea-read's file holds the value issued for claude-code here, alone", func() bool {
return e.read(t, forgeFile) == first
@@ -400,7 +407,7 @@ func TestNatsARequiredCredentialTheVaultRefusesFailsTheModule(t *testing.T) {
vault.refuse = "provisioner.g14 may not ask for " + forgeHolder + ": claude-code's plan on g14 names no credential of gitea-read from novox"
e := startEngine(t, map[string]string{forgeFile: "owned", settingsFile: "owned"})
b.assign(t, "claude-code", claudeCode(t))
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
eventually(t, "prov.applied says it failed, with the vault's reason", func() bool {
r, _ := heard.last("claude-code")
return r.Status == StatusFailed && strings.Contains(r.Reason, "names no credential of gitea-read")
@@ -471,7 +478,7 @@ func TestNatsAModulesOwnLoginFromAPerLoginProviderIsFilledFromTheVaultAndFollows
vault.keep(dbHolder, first, false)
e := startEngine(t, map[string]string{dbFile: "owned", dbConf: "owned"})
b.assign(t, "nextcloud", nextcloud(t))
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
eventually(t, "the secrets file holds the module's own login, alone", func() bool {
return e.read(t, dbFile) == first
@@ -5,7 +5,6 @@ import (
"os"
"path/filepath"
"testing"
"time"
)
// The graphical session's two files, xinitrc and xresources, filled by this client from the items of the collections
@@ -82,7 +81,7 @@ func TestNatsTheSessionFilesFilledFromTheirContributorsItemsAreWhatTheOldFormCom
tp, _ := templates(t, texts)
b.assign(t, m, Plan{Templates: tp, Items: planned, Placement: facts})
}
aClient(t, b, e, t.TempDir(), time.Hour)
aClient(t, b, e, t.TempDir())
for _, f := range []string{"xinitrc", "xresources"} {
want, err := os.ReadFile(filepath.Join("testdata", "xorg", name+"."+f))
if err != nil {