Merge remote-tracking branch 'origin/main' into feat/try-speed
This commit is contained in:
@@ -6,6 +6,8 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"os"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
@@ -153,6 +155,9 @@ type state struct {
|
||||
// unplaced is a module whose last fill failed, for want of the node-engine's write socket (issue 1165) or by the
|
||||
// engine's own refusal (issue 1873): filled again when the socket is made again.
|
||||
unplaced map[string]bool
|
||||
// ownComposed is the digest of the node's own composition the node-engine last said it runs (its have
|
||||
// `<node>.node`): when it moves, every module is filled again (issue 1945).
|
||||
ownComposed string
|
||||
// bound is each provider's bucket of entries this client watches, by key: this machine's consumers' entries alone;
|
||||
// boundReady is a bucket whose current entries are all read, and boundUsers the modules filled from each.
|
||||
bound map[string]map[string][]byte
|
||||
@@ -225,6 +230,21 @@ func (c *Client) Run(ctx context.Context) error {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// Its node's own have, one key: the node-engine moving to a new composition of the node's own fills every module
|
||||
// again, since an apply may have left a filled file as the composition had it (issue 1945, live on 2026-10-25). A
|
||||
// bucket not there yet (a node-engine before release 6) is said, and nothing waits for it.
|
||||
if haveKV, err := c.JS.KeyValue(ctx, delivery.HaveBucket); err == nil {
|
||||
ownKey, err := delivery.HaveNodeKey(c.Node)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
kvs[delivery.HaveBucket] = haveKV
|
||||
if err := watch(delivery.HaveBucket, ownKey); err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
fmt.Fprintf(os.Stderr, "the node's own have is not watched, so a new composition of the node's own fills nothing again: %v\n", err)
|
||||
}
|
||||
// A provider's bucket of entries, watched once a plan here names that provider (consumers.go): this machine's
|
||||
// consumers' keys alone. A bucket not there yet is tried again at the next fill.
|
||||
watchedBound := map[string]bool{}
|
||||
@@ -310,6 +330,13 @@ func (c *Client) Run(ctx context.Context) error {
|
||||
if u.socketUp {
|
||||
continue // the first fill below tries every file
|
||||
}
|
||||
if u.bucket == delivery.HaveBucket {
|
||||
// The composition the node runs now: the first fill below fills every module anyway.
|
||||
if u.entry != nil {
|
||||
s.ownComposed = ownComposedOf(u.entry)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if u.changed != nil {
|
||||
// Nothing is held yet: every secret is asked for when it is first filled.
|
||||
continue
|
||||
@@ -355,9 +382,27 @@ func (c *Client) Run(ctx context.Context) error {
|
||||
case u.asked != nil:
|
||||
c.respond(s, u.asked)
|
||||
case u.socketUp:
|
||||
// Every module, not only those whose fill failed: the node-engine that made the socket may have
|
||||
// applied a composition that left a filled file as the composition had it (issue 1945).
|
||||
for m := range s.unplaced {
|
||||
dirty[m] = true
|
||||
}
|
||||
for m := range s.plans {
|
||||
dirty[m] = true
|
||||
}
|
||||
case u.bucket == delivery.HaveBucket:
|
||||
if u.entry == nil {
|
||||
break
|
||||
}
|
||||
now := ownComposedOf(u.entry)
|
||||
if now != "" && s.ownComposed != "" && now != s.ownComposed {
|
||||
for m := range s.plans {
|
||||
dirty[m] = true
|
||||
}
|
||||
}
|
||||
if now != "" {
|
||||
s.ownComposed = now
|
||||
}
|
||||
case u.changed != nil:
|
||||
for _, m := range c.secretChanged(s, u.changed) {
|
||||
dirty[m] = true
|
||||
@@ -397,6 +442,19 @@ func (c *Client) Run(ctx context.Context) error {
|
||||
|
||||
func hasPlan(s *state, m string) bool { _, ok := s.plans[m]; return ok }
|
||||
|
||||
// ownComposedOf is the digest of the node's own composition a have `<node>.node` says the node-engine runs; empty for a
|
||||
// deleted key or a value it cannot read.
|
||||
func ownComposedOf(e jetstream.KeyValueEntry) string {
|
||||
if e.Operation() != jetstream.KeyValuePut {
|
||||
return ""
|
||||
}
|
||||
h, err := delivery.DecodeHaveNode(e.Value())
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
return h.Composed
|
||||
}
|
||||
|
||||
// told says what an update moved of a module's config, or of its placement, which decides whether it is answered
|
||||
// at all (asks.go).
|
||||
func (c *Client) told(u update) {
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
@@ -81,7 +82,7 @@ func startBus(t *testing.T) *aBus {
|
||||
}
|
||||
b := &aBus{url: s.ClientURL(), domain: js, conn: nc, kv: map[string]jetstream.KeyValue{}}
|
||||
for _, name := range []string{AssignmentsBucket, ConfigBucket, RegistryBucket, NeedsBucket, AppliedBucket,
|
||||
CollectBucket} {
|
||||
CollectBucket, delivery.HaveBucket} {
|
||||
kv, err := js.CreateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: name})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -741,3 +742,39 @@ func (b *aBus) defaults(t *testing.T, module string, manifest []byte) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The node-engine moving to a new composition of the node's own fills every module again: an apply may have left a
|
||||
// filled file as the composition had it (issue 1945, live on 2026-10-25: release 6's node-engines wrote filled files
|
||||
// empty, and the client, which fills at its start and on a change of a value, filled nothing again).
|
||||
func TestNatsTheNodesOwnHaveMovingFillsEveryModuleAgain(t *testing.T) {
|
||||
b := startBus(t)
|
||||
e := startEngine(t, map[string]string{dunstrc: "owned"})
|
||||
own := func(digit string) {
|
||||
raw, err := delivery.EncodeHaveNode(delivery.HaveNode{Composed: "sha256:" + strings.Repeat(digit, 64)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.put(t, delivery.HaveBucket, node+".node", string(raw))
|
||||
}
|
||||
own("a")
|
||||
b.config(t, "dunst", "font-size", "13")
|
||||
b.assign(t, "dunst", dunst(t))
|
||||
aClient(t, b, e, t.TempDir())
|
||||
eventually(t, "dunstrc is filled", func() bool { return strings.Contains(e.read(t, dunstrc), "Monospace 13") })
|
||||
|
||||
// The node's own composition empties it, and the node-engine says it runs that composition.
|
||||
if err := os.WriteFile(e.at(dunstrc), nil, 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
own("b")
|
||||
eventually(t, "dunstrc is filled again", func() bool { return strings.Contains(e.read(t, dunstrc), "Monospace 13") })
|
||||
|
||||
// The same composition said again fills nothing.
|
||||
_, before := e.counts(dunstrc)
|
||||
own("b")
|
||||
b.config(t, "dunst", "font-size", "15")
|
||||
eventually(t, "the change is written", func() bool { return strings.Contains(e.read(t, dunstrc), "Monospace 15") })
|
||||
if _, after := e.counts(dunstrc); after-before != 1 {
|
||||
t.Errorf("dunstrc was asked %d times for one change", after-before)
|
||||
}
|
||||
}
|
||||
|
||||
+9
-9
@@ -41,7 +41,10 @@ func TestNatsAFillThatFailedForWantOfTheSocketIsTriedWhenTheSocketReturns(t *tes
|
||||
})
|
||||
}
|
||||
|
||||
func TestNatsAReturningSocketAsksNothingOfAModuleThatWasPlaced(t *testing.T) {
|
||||
// A returning socket fills every module again, a placed one too: the node-engine that made it may have applied a
|
||||
// composition that left a filled file as the composition had it (issue 1945, live on 2026-10-25: release 6's node-engines
|
||||
// wrote claude-code's, node-env's and power's filled files empty, and nothing filled them again).
|
||||
func TestNatsAReturningSocketFillsAPlacedModuleAgain(t *testing.T) {
|
||||
b := startBus(t)
|
||||
e := newEngine(t, map[string]string{dunstrc: "owned"})
|
||||
stop := e.listen(t)
|
||||
@@ -49,17 +52,14 @@ func TestNatsAReturningSocketAsksNothingOfAModuleThatWasPlaced(t *testing.T) {
|
||||
b.assign(t, "dunst", dunst(t))
|
||||
aClient(t, b, e, t.TempDir())
|
||||
eventually(t, "dunstrc is filled", func() bool { return strings.Contains(e.read(t, dunstrc), "Monospace 13") })
|
||||
_, before := e.counts(dunstrc)
|
||||
|
||||
// The node-engine replaces itself: the socket goes and returns, and a later change proves the loop ran past it.
|
||||
// The node-engine replaces itself and, before it returns, the file is emptied.
|
||||
stop()
|
||||
e.listen(t)
|
||||
b.config(t, "dunst", "font-size", "15")
|
||||
eventually(t, "the change is written", func() bool { return strings.Contains(e.read(t, dunstrc), "Monospace 15") })
|
||||
_, after := e.counts(dunstrc)
|
||||
if after-before != 1 {
|
||||
t.Errorf("dunstrc was asked %d times for one change", after-before)
|
||||
if err := os.WriteFile(e.at(dunstrc), nil, 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
e.listen(t)
|
||||
eventually(t, "dunstrc is filled again", func() bool { return strings.Contains(e.read(t, dunstrc), "Monospace 13") })
|
||||
}
|
||||
|
||||
func TestWatchSocketWakesWhenTheSocketIsMadeAndWhenItsDirectoryReturns(t *testing.T) {
|
||||
|
||||
+2
-2
@@ -1,7 +1,7 @@
|
||||
# The provisioner client's account as mesh-controller composes it (internal/broker ComposeAccounts, feature branch
|
||||
# 57-config-and-provisioning with issue 647, which grants it the vault's secret-changed, issue 685, which lets its
|
||||
# own machine's tool runner ask it, and issue 679, which lets it read the whole assignments mirror and its machine's
|
||||
# keys of a provider's bucket of entries, here postgres_bound, and for gitea's route, route-adapter_bound and route-proxy_bound, issue 792; node-tools_bound for claude-code's console port, issue 761), the tool runner's account of the same machine
|
||||
# keys of a provider's bucket of entries, here postgres_bound, and for gitea's route, route-adapter_bound and route-proxy_bound, issue 792; node-tools_bound for claude-code's console port, issue 761; its node's own have, node-engine_have's g14.node, issue 1945), the tool runner's account of the same machine
|
||||
# carrying claude-code, and a `domain` user granted everything, standing for the controller, mesh-config and the
|
||||
# vault, which own the buckets the client reads and answer its asks. Composed again if the client's grants change.
|
||||
|
||||
@@ -11,7 +11,7 @@ accounts {
|
||||
users = [
|
||||
{ user: "domain", password: "x" }
|
||||
{ user: "provisioner.g14", password: "x", permissions: {
|
||||
publish: { allow: ["$JS.API.CONSUMER.CREATE.KV_mesh-config_resolved.*.$KV.mesh-config_resolved.g14.>", "$JS.API.CONSUMER.CREATE.KV_mesh-controller_assignments.>", "$JS.API.CONSUMER.CREATE.KV_mesh-provisioner_applied.*.$KV.mesh-provisioner_applied.g14.>", "$JS.API.CONSUMER.CREATE.KV_mesh-provisioner_collect.*.$KV.mesh-provisioner_collect.g14.>", "$JS.API.CONSUMER.CREATE.KV_mesh-provisioner_needs.*.$KV.mesh-provisioner_needs.g14.>", "$JS.API.CONSUMER.CREATE.KV_mesh-provisioner_registry.>", "$JS.API.CONSUMER.CREATE.KV_postgres_bound.*.$KV.postgres_bound.g14.>", "$JS.API.CONSUMER.CREATE.KV_route-adapter_bound.*.$KV.route-adapter_bound.g14.>", "$JS.API.CONSUMER.CREATE.KV_route-proxy_bound.*.$KV.route-proxy_bound.g14.>", "$JS.API.CONSUMER.CREATE.KV_node-tools_bound.*.$KV.node-tools_bound.g14.>", "$JS.API.CONSUMER.DELETE.KV_mesh-config_resolved.>", "$JS.API.CONSUMER.DELETE.KV_mesh-controller_assignments.>", "$JS.API.CONSUMER.DELETE.KV_mesh-provisioner_applied.>", "$JS.API.CONSUMER.DELETE.KV_mesh-provisioner_collect.>", "$JS.API.CONSUMER.DELETE.KV_mesh-provisioner_needs.>", "$JS.API.CONSUMER.DELETE.KV_mesh-provisioner_registry.>", "$JS.API.CONSUMER.DELETE.KV_postgres_bound.>", "$JS.API.CONSUMER.DELETE.KV_route-adapter_bound.>", "$JS.API.CONSUMER.DELETE.KV_route-proxy_bound.>", "$JS.API.CONSUMER.DELETE.KV_node-tools_bound.>", "$JS.API.DIRECT.GET.KV_mesh-config_resolved.$KV.mesh-config_resolved.g14.>", "$JS.API.DIRECT.GET.KV_mesh-controller_assignments.>", "$JS.API.DIRECT.GET.KV_mesh-provisioner_applied.$KV.mesh-provisioner_applied.g14.>", "$JS.API.DIRECT.GET.KV_mesh-provisioner_collect.$KV.mesh-provisioner_collect.g14.>", "$JS.API.DIRECT.GET.KV_mesh-provisioner_needs.$KV.mesh-provisioner_needs.g14.>", "$JS.API.DIRECT.GET.KV_mesh-provisioner_registry.>", "$JS.API.DIRECT.GET.KV_postgres_bound.$KV.postgres_bound.g14.>", "$JS.API.DIRECT.GET.KV_route-adapter_bound.$KV.route-adapter_bound.g14.>", "$JS.API.DIRECT.GET.KV_route-proxy_bound.$KV.route-proxy_bound.g14.>", "$JS.API.DIRECT.GET.KV_node-tools_bound.$KV.node-tools_bound.g14.>", "$JS.API.STREAM.INFO.KV_mesh-config_resolved", "$JS.API.STREAM.INFO.KV_mesh-controller_assignments", "$JS.API.STREAM.INFO.KV_mesh-provisioner_applied", "$JS.API.STREAM.INFO.KV_mesh-provisioner_collect", "$JS.API.STREAM.INFO.KV_mesh-provisioner_needs", "$JS.API.STREAM.INFO.KV_mesh-provisioner_registry", "$JS.API.STREAM.INFO.KV_postgres_bound", "$JS.API.STREAM.INFO.KV_route-adapter_bound", "$JS.API.STREAM.INFO.KV_route-proxy_bound", "$JS.API.STREAM.INFO.KV_node-tools_bound", "$JS.FC.KV_mesh-config_resolved.>", "$JS.FC.KV_mesh-controller_assignments.>", "$JS.FC.KV_mesh-provisioner_applied.>", "$JS.FC.KV_mesh-provisioner_collect.>", "$JS.FC.KV_mesh-provisioner_needs.>", "$JS.FC.KV_mesh-provisioner_registry.>", "$JS.FC.KV_postgres_bound.>", "$JS.FC.KV_route-adapter_bound.>", "$JS.FC.KV_route-proxy_bound.>", "$JS.FC.KV_node-tools_bound.>", "$KV.mesh-provisioner_applied.g14.>", "$KV.mesh-provisioner_collect.g14.>", "$KV.mesh-provisioner_needs.g14.>", "mesh.seat.mesh-vault.call.get.provisioner~g14", "prov.applied.g14.>", "prov.changed.g14.>"] }
|
||||
publish: { allow: ["$JS.API.CONSUMER.CREATE.KV_mesh-config_resolved.*.$KV.mesh-config_resolved.g14.>", "$JS.API.CONSUMER.CREATE.KV_mesh-controller_assignments.>", "$JS.API.CONSUMER.CREATE.KV_mesh-provisioner_applied.*.$KV.mesh-provisioner_applied.g14.>", "$JS.API.CONSUMER.CREATE.KV_mesh-provisioner_collect.*.$KV.mesh-provisioner_collect.g14.>", "$JS.API.CONSUMER.CREATE.KV_mesh-provisioner_needs.*.$KV.mesh-provisioner_needs.g14.>", "$JS.API.CONSUMER.CREATE.KV_mesh-provisioner_registry.>", "$JS.API.CONSUMER.CREATE.KV_node-engine_have.*.$KV.node-engine_have.g14.node", "$JS.API.CONSUMER.DELETE.KV_node-engine_have.>", "$JS.API.DIRECT.GET.KV_node-engine_have.$KV.node-engine_have.g14.node", "$JS.API.STREAM.INFO.KV_node-engine_have", "$JS.FC.KV_node-engine_have.>", "$JS.API.CONSUMER.CREATE.KV_postgres_bound.*.$KV.postgres_bound.g14.>", "$JS.API.CONSUMER.CREATE.KV_route-adapter_bound.*.$KV.route-adapter_bound.g14.>", "$JS.API.CONSUMER.CREATE.KV_route-proxy_bound.*.$KV.route-proxy_bound.g14.>", "$JS.API.CONSUMER.CREATE.KV_node-tools_bound.*.$KV.node-tools_bound.g14.>", "$JS.API.CONSUMER.DELETE.KV_mesh-config_resolved.>", "$JS.API.CONSUMER.DELETE.KV_mesh-controller_assignments.>", "$JS.API.CONSUMER.DELETE.KV_mesh-provisioner_applied.>", "$JS.API.CONSUMER.DELETE.KV_mesh-provisioner_collect.>", "$JS.API.CONSUMER.DELETE.KV_mesh-provisioner_needs.>", "$JS.API.CONSUMER.DELETE.KV_mesh-provisioner_registry.>", "$JS.API.CONSUMER.DELETE.KV_postgres_bound.>", "$JS.API.CONSUMER.DELETE.KV_route-adapter_bound.>", "$JS.API.CONSUMER.DELETE.KV_route-proxy_bound.>", "$JS.API.CONSUMER.DELETE.KV_node-tools_bound.>", "$JS.API.DIRECT.GET.KV_mesh-config_resolved.$KV.mesh-config_resolved.g14.>", "$JS.API.DIRECT.GET.KV_mesh-controller_assignments.>", "$JS.API.DIRECT.GET.KV_mesh-provisioner_applied.$KV.mesh-provisioner_applied.g14.>", "$JS.API.DIRECT.GET.KV_mesh-provisioner_collect.$KV.mesh-provisioner_collect.g14.>", "$JS.API.DIRECT.GET.KV_mesh-provisioner_needs.$KV.mesh-provisioner_needs.g14.>", "$JS.API.DIRECT.GET.KV_mesh-provisioner_registry.>", "$JS.API.DIRECT.GET.KV_postgres_bound.$KV.postgres_bound.g14.>", "$JS.API.DIRECT.GET.KV_route-adapter_bound.$KV.route-adapter_bound.g14.>", "$JS.API.DIRECT.GET.KV_route-proxy_bound.$KV.route-proxy_bound.g14.>", "$JS.API.DIRECT.GET.KV_node-tools_bound.$KV.node-tools_bound.g14.>", "$JS.API.STREAM.INFO.KV_mesh-config_resolved", "$JS.API.STREAM.INFO.KV_mesh-controller_assignments", "$JS.API.STREAM.INFO.KV_mesh-provisioner_applied", "$JS.API.STREAM.INFO.KV_mesh-provisioner_collect", "$JS.API.STREAM.INFO.KV_mesh-provisioner_needs", "$JS.API.STREAM.INFO.KV_mesh-provisioner_registry", "$JS.API.STREAM.INFO.KV_postgres_bound", "$JS.API.STREAM.INFO.KV_route-adapter_bound", "$JS.API.STREAM.INFO.KV_route-proxy_bound", "$JS.API.STREAM.INFO.KV_node-tools_bound", "$JS.FC.KV_mesh-config_resolved.>", "$JS.FC.KV_mesh-controller_assignments.>", "$JS.FC.KV_mesh-provisioner_applied.>", "$JS.FC.KV_mesh-provisioner_collect.>", "$JS.FC.KV_mesh-provisioner_needs.>", "$JS.FC.KV_mesh-provisioner_registry.>", "$JS.FC.KV_postgres_bound.>", "$JS.FC.KV_route-adapter_bound.>", "$JS.FC.KV_route-proxy_bound.>", "$JS.FC.KV_node-tools_bound.>", "$KV.mesh-provisioner_applied.g14.>", "$KV.mesh-provisioner_collect.g14.>", "$KV.mesh-provisioner_needs.g14.>", "mesh.seat.mesh-vault.call.get.provisioner~g14", "prov.applied.g14.>", "prov.changed.g14.>"] }
|
||||
subscribe: { allow: ["_INBOX.provisioner.g14.>", "mesh.mod.mesh-vault.event.secret-changed", "prov.ask.g14"] }
|
||||
allow_responses: { max: 1, ttl: "1m" }
|
||||
} }
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-provisioner-client
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021114014-66e8fdaa62e2
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024234636-10b6e202320f
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
golang.org/x/sys v0.48.0
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021114014-66e8fdaa62e2 h1:Cdl5N56QPfmFBoLHrDfvvyc5Q+PUnTjf3fgYD7imcco=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021114014-66e8fdaa62e2/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024234636-10b6e202320f h1:lUEyNPPF3w4Q0LIgcUUD6ofkKVhCF0n7dVjMoixQmXo=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024234636-10b6e202320f/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=
|
||||
|
||||
Reference in New Issue
Block a user