diff --git a/cmd/mesh-controller/declared_packages_integration_test.go b/cmd/mesh-controller/declared_packages_integration_test.go new file mode 100644 index 00000000..c99ca3da --- /dev/null +++ b/cmd/mesh-controller/declared_packages_integration_test.go @@ -0,0 +1,89 @@ +//go:build integration + +package main + +import ( + "encoding/json" + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "github.com/novox/mesh-controller/internal/inventory" +) + +// 1840/AC5 (ADR 0507 §5): the source of the declared-packages mirror composes each node and names, for a package two modules +// declare and one declared absent, the modules that declare it; the same records give the same bytes on a second pass, so +// the mirror writes nothing; a node with no package has a key that says so. +func TestTheDeclaredPackagesMirrorNamesWhatEachNodesModulesDeclare(t *testing.T) { + open := aMesh(t) + ctx := t.Context() + register(t, open, catalogue.Manifest{Module: "viewer", Version: "1", Resources: []map[string]any{ + {"id": "pkg-curl", "type": "package", "package": "curl"}, + {"id": "pkg-nano", "type": "package", "package": "nano", "absent": true}}}) + register(t, open, catalogue.Manifest{Module: "fetcher", Version: "1", Resources: []map[string]any{ + {"id": "pkg-curl", "type": "package", "package": "curl"}}}) + for _, module := range []string{"viewer", "fetcher"} { + if _, err := assign(ctx, open, "anchor", module); err != nil { + t.Fatalf("assign anchor %s: %v", module, err) + } + } + source := declaredPackagesMirror(open) + first, err := source(ctx) + if err != nil { + t.Fatal(err) + } + var got inventory.DeclaredPackagesMirrored + if err := json.Unmarshal(first["anchor"], &got); err != nil { + t.Fatalf("anchor has no value: %v", err) + } + // Every node also runs the mesh's own modules, which declare packages of their own: curl and nano are looked + // up by name, and no package names viewer or fetcher but those they declare. + byName := map[string]inventory.PackageDeclared{} + for _, p := range got.Packages { + byName[p.Package] = p + } + same := func(a, b []string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true + } + for _, p := range []inventory.PackageDeclared{{Package: "curl", Modules: []string{"fetcher", "viewer"}}, + {Package: "nano", Modules: []string{"viewer"}, Absent: true}} { + if g, ok := byName[p.Package]; !ok || g.Absent != p.Absent || !same(g.Modules, p.Modules) { + t.Errorf("anchor declares %s as %+v, want %+v (all: %+v)", p.Package, g, p, got.Packages) + } + } + for _, p := range got.Packages { + if p.Package == "curl" || p.Package == "nano" { + continue + } + for _, m := range p.Modules { + if m == "viewer" || m == "fetcher" { + t.Errorf("%s is said to be declared by %s, which does not declare it", p.Package, m) + } + } + } + var laptop inventory.DeclaredPackagesMirrored + if err := json.Unmarshal(first["laptop"], &laptop); err != nil || laptop.Node != "laptop" || laptop.Packages == nil { + t.Errorf("a node assigned neither module has no value of its own: %s (%v)", first["laptop"], err) + } + for _, p := range laptop.Packages { + if p.Package == "curl" || p.Package == "nano" { + t.Errorf("laptop is assigned neither module and declares %+v", p) + } + } + second, err := source(ctx) + if err != nil { + t.Fatal(err) + } + for node, value := range first { + if string(second[node]) != string(value) { + t.Errorf("the value of %s changed between two passes over the same records:\n%s\n%s", node, value, second[node]) + } + } +} diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index 2769c902..61cfcb9f 100644 --- a/cmd/mesh-controller/push.go +++ b/cmd/mesh-controller/push.go @@ -281,7 +281,7 @@ func serve(ctx context.Context) (err error) { // The provisioner's plan of each handed module, as its machine was last sent it (design 57 P6), in the module's // assignment. inv.SetProvisioningPlans(inv.SentProvisioningPlans) - go mirroringRecords(ctx, inv, bus.Conn, bus) + go mirroringRecords(ctx, open, inv, bus.Conn, bus) // And the bucket of the files that travel by digest, judged against what each machine's latest declaration names (ADR 0486 §3). go sweepingDeclarationFiles(ctx, inv, bus.Conn) // And the key of each plan in its build phase, which holds the build seat's checks while a delivery builds (ADR 0483 clause 2). diff --git a/cmd/mesh-controller/record_mirrors.go b/cmd/mesh-controller/record_mirrors.go index 25fd9d15..0f02cded 100644 --- a/cmd/mesh-controller/record_mirrors.go +++ b/cmd/mesh-controller/record_mirrors.go @@ -7,6 +7,7 @@ import ( "sort" "time" + "git.novox.be/novox/mesh-sdk/go/compose" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -29,7 +30,7 @@ import ( const mirrorEvery = 10 * time.Second // mirroringRecords keeps the mirrors current for as long as this controller serves: ctx ends when it stops. -func mirroringRecords(ctx context.Context, inv *inventory.Inventory, conn *nats.Conn, announce recordmirror.Announcer) { +func mirroringRecords(ctx context.Context, open *stores, inv *inventory.Inventory, conn *nats.Conn, announce recordmirror.Announcer) { api, err := jetstream.New(conn) if err != nil { fmt.Printf("the record mirrors are not kept: %v\n", err) @@ -44,8 +45,9 @@ func mirroringRecords(ctx context.Context, inv *inventory.Inventory, conn *nats. return e } sources := map[string]func(context.Context) (map[string][]byte, error){ - recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror, - recordmirror.Predefined.Name: inv.PredefinedMirror, + recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror, + recordmirror.Predefined.Name: inv.PredefinedMirror, + recordmirror.DeclaredPackages.Name: declaredPackagesMirror(open), } writers := map[string]*recordmirror.Writer{} failing := map[string]string{} @@ -176,3 +178,36 @@ func keepingAssignments(ctx context.Context, inv *inventory.Inventory, announce } events.known = now } + +// declaredPackagesMirror is the source of the declared-packages mirror (ADR 0507 §5): each node's composition, asked as a +// question that makes nothing, and the packages it declares with the modules that declare them. A node whose own set does +// not compose has no key, and any other failure to read is the pass's error, so a key is never purged for a store that +// could not be read. +func declaredPackagesMirror(open *stores) func(context.Context) (map[string][]byte, error) { + return func(ctx context.Context) (map[string][]byte, error) { + in, err := inputFor(ctx, open) + if err != nil { + return nil, err + } + // Which module owns each resource is carried only for a node-engine that reads it, so every node is asked as one that does. + asked := in.Input + asked.Nodes = append([]compose.Node(nil), in.Nodes...) + for i := range asked.Nodes { + asked.Nodes[i].ReadsModules = true + } + out := map[string][]byte{} + for _, n := range in.sortedNodes() { + d, _, err := compose.ComposeIn(asked, n.Name, compose.Foreseeing) + if compose.IsUnresolvable(err) { + continue + } + if err != nil { + return nil, fmt.Errorf("the packages %s declares cannot be read: %w", n.Name, err) + } + if out[n.Name], err = inventory.DeclaredPackagesValue(n.Name, d.Resources, d.ModuleOf); err != nil { + return nil, err + } + } + return out, nil + } +} diff --git a/go.mod b/go.mod index 23a119b3..6fbf57ab 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/novox/mesh-controller go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161855-e76a5a31a07b + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876 github.com/jackc/pgx/v5 v5.10.0 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 diff --git a/go.sum b/go.sum index de414370..60fa576d 100644 --- a/go.sum +++ b/go.sum @@ -6,6 +6,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022152521-dfc2c858d94b h1:cfJ2LK/e git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022152521-dfc2c858d94b/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161855-e76a5a31a07b h1:OjlhfiGr+wKLWxafV+BRjZx3TbBtwRYcm9mgoam/jv8= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161855-e76a5a31a07b/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876 h1:dtiIJzrX7ey1wmsdzapiFgCKDllCph7/vxfUp1ib/t8= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876/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/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= diff --git a/internal/broker/announced.go b/internal/broker/announced.go index 9d1fdcae..5f559b88 100644 --- a/internal/broker/announced.go +++ b/internal/broker/announced.go @@ -16,10 +16,10 @@ import ( // StatesAnnouncedBucket is the controller's bucket of the last announcement of each module (announce.KeptBucket). const StatesAnnouncedBucket = announce.KeptBucket -// controllerReadable are the controller's states others may read: the record mirrors, the chain's buckets, the checks -// running and the plans building. +// controllerReadable are the controller's states others may read: the record mirrors (the packages each node declares +// among them, which the pacman module reads, ADR 0507 §5), the chain's buckets, the checks running and the plans building. var controllerReadable = []string{MirrorAssignments, MirrorDeclaredConfig, MirrorPredefined, ChainBuilds, ChainRollouts, - ChecksRunning, PlansBuilding} + ChecksRunning, PlansBuilding, MirrorDeclaredPackages} // ControllerAnnouncement is what the controller announces it keeps (ADR 0495 clause 2): every bucket of its own by its local // name, the ones others may read marked readable. It announces itself, as any module does from its code. diff --git a/internal/broker/announced_test.go b/internal/broker/announced_test.go index cca553df..8b4c1167 100644 --- a/internal/broker/announced_test.go +++ b/internal/broker/announced_test.go @@ -24,8 +24,9 @@ func TestAReadIsGrantedFromItsOwnersAnnouncement(t *testing.T) { } } -// D1 (ADR 0495 clause 2): the controller announces every bucket of its own, and readable exactly the record mirrors, -// the chain's buckets, the checks running and the plans building. +// D1 (ADR 0495 clause 2): the controller announces every bucket of its own, and readable exactly the record mirrors (the +// declared packages the pacman module reads among them, ADR 0507 §5), the chain's buckets, the checks running and the +// plans building. func TestTheControllerAnnouncesWhatItKeeps(t *testing.T) { s := ControllerAnnouncement("novox") if _, err := announce.Encode(s); err != nil { @@ -38,7 +39,7 @@ func TestTheControllerAnnouncesWhatItKeeps(t *testing.T) { } } want := []string{ChainBuilds, ChecksRunning, MirrorDeclaredConfig, PlansBuilding, MirrorPredefined, ChainRollouts, - MirrorAssignments} + MirrorAssignments, MirrorDeclaredPackages} slices.Sort(want) if !slices.Equal(readable, want) { t.Fatalf("readable %v, want %v", readable, want) diff --git a/internal/broker/controller_buckets.go b/internal/broker/controller_buckets.go index f3aad638..0462f01d 100644 --- a/internal/broker/controller_buckets.go +++ b/internal/broker/controller_buckets.go @@ -104,6 +104,9 @@ var ( AssignmentsBucket = BucketName(ControllerSeat, MirrorAssignments) DeclaredConfigBucket = BucketName(ControllerSeat, MirrorDeclaredConfig) PredefinedBucket = BucketName(ControllerSeat, MirrorPredefined) + // DeclaredPackagesBucket holds, keyed by the node, each package the modules assigned to it declare, present or absent, + // with the declaring modules: what the pacman module's verbs read to say a package is declared (ADR 0507 §5). + DeclaredPackagesBucket = BucketName(ControllerSeat, MirrorDeclaredPackages) ) // The mirrors' local names, as a reader names them in `reads` after `mesh-controller.`. @@ -111,6 +114,8 @@ const ( MirrorAssignments = "assignments" MirrorDeclaredConfig = "declared-config" MirrorPredefined = "predefined" + // MirrorDeclaredPackages is the local name of the packages each node's modules declare (ADR 0507 §5, issue 1840/S3). + MirrorDeclaredPackages = "declared-packages" ) // The chain's buckets (novox/hq ADR 0408, issue 1089): builds, rollouts and assignments are one chain kept on the bus @@ -216,6 +221,7 @@ func IsControllerBucket(bucket string) bool { return bucket == CallsBucket || bucket == HandActsBucket || bucket == ConditionsBucket || bucket == ConditionHistoryBucket || bucket == LeaseBucket || bucket == AskedBucket || bucket == AssignmentsBucket || bucket == DeclaredConfigBucket || bucket == PredefinedBucket || + bucket == DeclaredPackagesBucket || bucket == MergeWindowBucket || bucket == RolloutSilentBucket || bucket == CheckRunningBucket || bucket == DeliveryMergesBucket || bucket == BuildsBucket || bucket == RolloutsBucket || bucket == PlansBuildingBucket || bucket == delivery.HaveBucket || bucket == StatesAnnouncedBucket @@ -224,7 +230,7 @@ func IsControllerBucket(bucket string) bool { // ControllerBuckets are the controller's own buckets, in the order they are asserted. func ControllerBuckets() []string { return []string{LeaseBucket, CallsBucket, HandActsBucket, ConditionsBucket, ConditionHistoryBucket, AskedBucket, - AssignmentsBucket, DeclaredConfigBucket, PredefinedBucket, MergeWindowBucket, RolloutSilentBucket, CheckRunningBucket, + AssignmentsBucket, DeclaredConfigBucket, PredefinedBucket, DeclaredPackagesBucket, MergeWindowBucket, RolloutSilentBucket, CheckRunningBucket, DeliveryMergesBucket, PlansBuildingBucket, BuildsBucket, RolloutsBucket, delivery.HaveBucket, StatesAnnouncedBucket} } @@ -452,6 +458,9 @@ func (j *JetStream) EnsureControllerBuckets() error { {PredefinedBucket, "the list of predefined keys the mesh published, the newest trunk build's of the " + "catalogue, one key `list` (novox/hq ADR 0348 §7, design 57 K3): written by the controller alone from " + "its store when a newer list is kept, read through `reads`"}, + {DeclaredPackagesBucket, "each package the modules assigned to a node declare, present or absent, with the " + + "declaring modules, keyed by the node (novox/hq ADR 0507 §5): written by the controller alone when it " + + "composes the node, only on a change, read through `reads`"}, } { if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{ Bucket: m.bucket, diff --git a/internal/broker/declared_packages_mirror_test.go b/internal/broker/declared_packages_mirror_test.go new file mode 100644 index 00000000..01b5b239 --- /dev/null +++ b/internal/broker/declared_packages_mirror_test.go @@ -0,0 +1,41 @@ +package broker + +import ( + "slices" + "testing" +) + +// Issue 1840/S3 (ADR 0507 §5), 1840/AC5: a module whose manifest reads mesh-controller.declared-packages may name it in +// `reads`, and reads the bucket and writes none; the controller writes it, so it is no mirror written by another, and it is +// one of the controller's own buckets, asserted at its start. +func TestAModuleReadingDeclaredPackagesReadsAndWritesNone(t *testing.T) { + if !slices.Contains(controllerReadableStates(), "declared-packages") || + BucketName(ControllerSeat, MirrorDeclaredPackages) != DeclaredPackagesBucket { + t.Fatalf("a module cannot name %s in `reads`: %v", DeclaredPackagesBucket, controllerReadableStates()) + } + if _, other := MirrorsWrittenByAnother[MirrorDeclaredPackages]; other { + t.Error("the controller writes the declared packages itself") + } + if !IsControllerBucket(DeclaredPackagesBucket) || !slices.Contains(ControllerBuckets(), DeclaredPackagesBucket) { + t.Errorf("%s is not among the controller's buckets", DeclaredPackagesBucket) + } + perms, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "pacman", PasswordHash: "x", + Reads: []string{"mesh-controller.declared-packages"}}) + if err != nil { + t.Fatal(err) + } + stream := "KV_" + DeclaredPackagesBucket + has(t, perms.Publish, "$JS.API.STREAM.INFO."+stream) + has(t, perms.Publish, "$JS.API.DIRECT.GET."+stream+".>") + has(t, perms.Publish, "$JS.API.CONSUMER.CREATE."+stream+".>") + for _, p := range perms.Publish { + if SubjectsOverlap(p, "$KV."+DeclaredPackagesBucket+".one") { + t.Errorf("the reading module may write %s through %q", DeclaredPackagesBucket, p) + } + } + controller, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"}) + if err != nil { + t.Fatal(err) + } + has(t, controller.Publish, "$KV."+DeclaredPackagesBucket+".>") +} diff --git a/internal/broker/nats.go b/internal/broker/nats.go index e61dfbd0..a6980a5d 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -440,6 +440,11 @@ var DegradableGrants = []DegradableGrant{ Reason: "refused: the controller then says no plan builds, and no check is held while a delivery builds — a check run " + "wrongly costs a delivery a place, a check held wrongly costs a pull request its turn (ADR 0483 clause 2, issue 1687)", }, + { + Subject: "$KV." + DeclaredPackagesBucket + ".>", + Reason: "carried only at genesis: the controller composes its own permissions and pushes them to the bus at its " + + "first push, and keeps the packages each node declares only after that, holding the lease (ADR 0507 §5, issue 1840)", + }, { Subject: "mesh.seat." + ControllerSeat + ".event.sent", Reason: "refused: the controller says a tier sent to a machine only for outsiders, and a send that could not " + diff --git a/internal/broker/process_state_watch_test.go b/internal/broker/process_state_watch_test.go new file mode 100644 index 00000000..3f9b2081 --- /dev/null +++ b/internal/broker/process_state_watch_test.go @@ -0,0 +1,41 @@ +package broker + +import "testing" + +// 1840/AC9: a process account declaring the state `x` may watch and list the keys of its module's own bucket (a consumer +// created and deleted on KV__x), read and write it, and has no grant on another module's bucket or on a mirror of the +// controller's, whatever it is named after. +func TestAProcessMayWatchItsOwnStateAndNoOthers(t *testing.T) { + p := Principal{Kind: KindProcess, Node: "novox", Module: "mesh-mail", Process: "send", State: []string{"x"}} + got := perms(t, p) + for _, s := range []string{ + "$JS.API.CONSUMER.CREATE.KV_mesh-mail_x", + "$JS.API.CONSUMER.CREATE.KV_mesh-mail_x.watcher", + "$JS.API.CONSUMER.CREATE.KV_mesh-mail_x.watcher.$KV.mesh-mail_x.>", + "$JS.API.CONSUMER.DELETE.KV_mesh-mail_x.watcher", + "$JS.API.DIRECT.GET.KV_mesh-mail_x.$KV.mesh-mail_x.key", + "$JS.API.STREAM.INFO.KV_mesh-mail_x", + "$JS.FC.KV_mesh-mail_x.watcher", + "$KV.mesh-mail_x.key", + } { + if !allowed(got.Publish, s) { + t.Errorf("the process may not publish %s, so it cannot watch or list its own state", s) + } + } + for _, other := range []string{"mesh-other_x", "mesh-mail_y", DeclaredPackagesBucket, AssignmentsBucket} { + for _, s := range []string{ + "$JS.API.CONSUMER.CREATE.KV_" + other, + "$JS.API.CONSUMER.CREATE.KV_" + other + ".watcher", + "$JS.API.CONSUMER.CREATE.KV_" + other + ".watcher.$KV." + other + ".>", + "$JS.API.CONSUMER.DELETE.KV_" + other + ".watcher", + "$JS.API.DIRECT.GET.KV_" + other + ".$KV." + other + ".key", + "$JS.API.STREAM.INFO.KV_" + other, + "$JS.FC.KV_" + other + ".watcher", + "$KV." + other + ".key", + } { + if allowed(got.Publish, s) { + t.Errorf("the process may publish %s, on a bucket that is not its own state", s) + } + } + } +} diff --git a/internal/broker/testdata/composed.conf b/internal/broker/testdata/composed.conf index c3f0da3e..dfc1d983 100644 --- a/internal/broker/testdata/composed.conf +++ b/internal/broker/testdata/composed.conf @@ -29,7 +29,7 @@ accounts { ] users = [ { user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: { - publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_checks-running.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_plans-building.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollout-silent.>", "$KV.mesh-controller_rollouts.>", "$KV.mesh-controller_states-announced.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.mod.mesh-controller.event.states", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.event.sent", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-controller.tool.push", "mesh.seat.mesh-controller.tool.upgrade", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.check.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] } + publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_checks-running.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_declared-packages.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_plans-building.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollout-silent.>", "$KV.mesh-controller_rollouts.>", "$KV.mesh-controller_states-announced.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.mod.mesh-controller.event.states", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.event.sent", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-controller.tool.push", "mesh.seat.mesh-controller.tool.upgrade", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.check.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] } subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] } allow_responses: { max: 1, ttl: "1m" } } } diff --git a/internal/broker/writers.go b/internal/broker/writers.go index b31d0192..73dd8369 100644 --- a/internal/broker/writers.go +++ b/internal/broker/writers.go @@ -260,6 +260,9 @@ var WritersTable = []WriterRow{ {State: "the predefined-keys mirror", Writer: "controller (lease holder), from its store, only on a change", KeptIn: "key-value " + PredefinedBucket, Others: "read through `reads` as mesh-controller." + MirrorPredefined, Subjects: kvOf(PredefinedBucket), Writes: isController}, + {State: "the declared-packages mirror", Writer: "controller (lease holder), from the composition of each node, only on a change", + KeptIn: "key-value " + DeclaredPackagesBucket, Others: "read through `reads` as mesh-controller." + MirrorDeclaredPackages, + Subjects: kvOf(DeclaredPackagesBucket), Writes: isController}, // What a machine's provisioner client placed of a module's files (novox/hq ADR 0349 §4, design 57 P6): its own // machine's word alone, which a walk's end and a gate wait for (ADR 0362, ADR 0349 §15). Kept newest per module // in the stream ProvAppliedStream, and by the client in the provisioner server's `applied` bucket. diff --git a/internal/broker/writers_test.go b/internal/broker/writers_test.go index d01005ce..aed01b84 100644 --- a/internal/broker/writers_test.go +++ b/internal/broker/writers_test.go @@ -52,6 +52,7 @@ var designRows = []string{ "a declaration's large files", "the declared-config mirror", "the predefined-keys mirror", + "the declared-packages mirror", // The provisioner client's word on what it placed (novox/hq ADR 0349 §4, design 57 P6). "a module's files placed on a machine", } diff --git a/internal/inventory/declared_packages_test.go b/internal/inventory/declared_packages_test.go new file mode 100644 index 00000000..51edea22 --- /dev/null +++ b/internal/inventory/declared_packages_test.go @@ -0,0 +1,83 @@ +package inventory + +import ( + "bytes" + "encoding/json" + "fmt" + "reflect" + "slices" + "testing" +) + +// aNodesPackages is the resources of a node whose modules declare packages: one package two modules declare present, one +// declared absent, and a resource that is no package. +func aNodesPackages() ([]map[string]any, map[string]string) { + return []map[string]any{ + {"id": "web.pkg-curl", "type": "package", "package": "curl"}, + {"id": "backup.pkg-curl", "type": "package", "package": "curl"}, + {"id": "web.pkg-nano", "type": "package", "package": "nano", "absent": true}, + {"id": "web.conf", "type": "file", "path": "/etc/web.conf"}, + }, + map[string]string{"web.pkg-curl": "web", "backup.pkg-curl": "backup", "web.pkg-nano": "web", "web.conf": "web"} +} + +// 1840/AC5: the declared-packages mirror's value for a node names each package with the modules that declare it and whether +// they declare it absent, and the same declaration is the same bytes however its resources are ordered, so a pass over an +// unchanged node writes nothing. +func TestTheDeclaredPackagesValueNamesEachPackageWithItsModules(t *testing.T) { + resources, owner := aNodesPackages() + got, err := DeclaredPackagesValue("one", resources, owner) + if err != nil { + t.Fatal(err) + } + want := `{"node":"one","packages":[{"package":"curl","modules":["backup","web"],"absent":false},` + + `{"package":"nano","modules":["web"],"absent":true}]}` + if string(got) != want { + t.Errorf("the value is\n%s\nwant\n%s", got, want) + } + reversed := slices.Clone(resources) + slices.Reverse(reversed) + again, err := DeclaredPackagesValue("one", reversed, owner) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, again) { + t.Errorf("the same declaration in another order is other bytes:\n%s\n%s", got, again) + } + if got := jsonFields(reflect.TypeOf(DeclaredPackagesMirrored{})); !slices.Equal(got, []string{"node", "packages"}) { + t.Errorf("the value carries %v", got) + } +} + +// A node declaring no package has a value that says so, an empty list and not null. +func TestANodeDeclaringNoPackageHasAnEmptyList(t *testing.T) { + got, err := DeclaredPackagesValue("one", nil, nil) + if err != nil || string(got) != `{"node":"one","packages":[]}` { + t.Errorf("the value is %s (%v)", got, err) + } +} + +// 1840/S3: a node's value stays under the 64 KiB a mirror's key holds at far more packages than a node declares: 600 +// packages, each declared by four modules, is a few times what the largest node declares today. +func TestTheDeclaredPackagesValueOfABigNodeFitsAKey(t *testing.T) { + var resources []map[string]any + owner := map[string]string{} + for p := 0; p < 600; p++ { + for m := 0; m < 4; m++ { + id := fmt.Sprintf("module-%d.pkg-%d", m, p) + resources = append(resources, map[string]any{"id": id, "type": "package", "package": fmt.Sprintf("package-%d", p)}) + owner[id] = fmt.Sprintf("module-%d", m) + } + } + got, err := DeclaredPackagesValue("one", resources, owner) + if err != nil { + t.Fatal(err) + } + var v DeclaredPackagesMirrored + if err := json.Unmarshal(got, &v); err != nil || len(v.Packages) != 600 { + t.Fatalf("the value holds %d packages (%v)", len(v.Packages), err) + } + if len(got) >= 64<<10 { + t.Errorf("the value is %d bytes, over the 64 KiB a key holds", len(got)) + } +} diff --git a/internal/inventory/mirrors.go b/internal/inventory/mirrors.go index 5836e921..11c1f7ca 100644 --- a/internal/inventory/mirrors.go +++ b/internal/inventory/mirrors.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "fmt" + "sort" "time" "git.novox.be/novox/mesh-sdk/go/catalogue" @@ -191,3 +192,65 @@ func DeclaredOf(settings map[string]catalogue.SettingDeclaration) (map[string]De } return declaredMirrored(raw) } + +// PackageDeclared is one package a node's modules declare, as the declared-packages mirror holds it (ADR 0507 §5): its +// name, the modules that declare it that way, sorted, and whether they declare it absent. A package some modules declare +// present and others absent is two entries; the composition refuses that pair, so it is seen only while it is refused. +type PackageDeclared struct { + Package string `json:"package"` + Modules []string `json:"modules"` + Absent bool `json:"absent"` +} + +// DeclaredPackagesMirrored is one node's value of the declared-packages mirror, keyed by the node: its packages sorted by +// name (absent after present). Fixed fields and no time, so the same declaration is the same bytes (ADR 0289 §1). +type DeclaredPackagesMirrored struct { + Node string `json:"node"` + Packages []PackageDeclared `json:"packages"` +} + +// DeclaredPackagesValue is the mirror's value for one node, from the resources its composition declares and the module that +// owns each (Composed.Owner, by resource id). A resource that is no package, or names none, is left out. +func DeclaredPackagesValue(node string, resources []map[string]any, owner map[string]string) ([]byte, error) { + type key struct { + name string + absent bool + } + by := map[key]map[string]bool{} + for _, r := range resources { + if fmt.Sprint(r["type"]) != "package" { + continue + } + name, _ := r["package"].(string) + if name == "" { + continue + } + k := key{name, r["absent"] == true} + if by[k] == nil { + by[k] = map[string]bool{} + } + if module := owner[fmt.Sprint(r["id"])]; module != "" { + by[k][module] = true + } + } + keys := make([]key, 0, len(by)) + for k := range by { + keys = append(keys, k) + } + sort.Slice(keys, func(i, j int) bool { + if keys[i].name != keys[j].name { + return keys[i].name < keys[j].name + } + return !keys[i].absent && keys[j].absent + }) + v := DeclaredPackagesMirrored{Node: node, Packages: []PackageDeclared{}} + for _, k := range keys { + modules := make([]string, 0, len(by[k])) + for m := range by[k] { + modules = append(modules, m) + } + sort.Strings(modules) + v.Packages = append(v.Packages, PackageDeclared{Package: k.name, Modules: modules, Absent: k.absent}) + } + return json.Marshal(v) +} diff --git a/internal/recordmirror/mirror.go b/internal/recordmirror/mirror.go index f330ed4e..db9a3ce0 100644 --- a/internal/recordmirror/mirror.go +++ b/internal/recordmirror/mirror.go @@ -57,10 +57,13 @@ var ( // mesh-config publishes as its own (design 57 K3). Predefined = Mirror{Name: broker.MirrorPredefined, Bucket: broker.PredefinedBucket, Event: link.KeyPredefinedChanged} + // DeclaredPackages is each node's packages its modules declare, keyed by the node (ADR 0507 §5). It raises no event: + // the seat's emits, which the SDK states, do not name one, and a reader reads the bucket when it is asked. + DeclaredPackages = Mirror{Name: broker.MirrorDeclaredPackages, Bucket: broker.DeclaredPackagesBucket} ) // Mirrors is every mirror this package writes, in the order they are written. -func Mirrors() []Mirror { return []Mirror{DeclaredConfig, Predefined} } +func Mirrors() []Mirror { return []Mirror{DeclaredConfig, Predefined, DeclaredPackages} } // The changes an event says. const ( @@ -251,7 +254,7 @@ func (w *Writer) put(ctx context.Context, key string, value []byte, before kept, // say raises one write's event, tried once more when refused and then counted as lost. func (w *Writer) say(ctx context.Context, key, change string, fields []string, revision uint64) { - if w.Announce == nil { + if w.Announce == nil || w.Mirror.Event == "" { return } e := Event{Mirror: w.Mirror.Name, Key: key, Change: change, Fields: fields, Revision: revision, At: w.now().UTC()} diff --git a/internal/recordmirror/mirror_integration_test.go b/internal/recordmirror/mirror_integration_test.go index b4a1bebd..64a63d27 100644 --- a/internal/recordmirror/mirror_integration_test.go +++ b/internal/recordmirror/mirror_integration_test.go @@ -9,6 +9,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/testbus" ) @@ -246,3 +247,49 @@ func TestNatsARemovalWithoutTheLeaseRemovesNothing(t *testing.T) { t.Errorf("the key was removed without the lease: %v", err) } } + +// 1840/AC5 (ADR 0507 §5, ADR 0289 §1): the declared-packages mirror, keyed by the node, writes a node's value once; a second +// pass over the same declaration writes nothing (the bucket takes no message); a package newly declared changes the one key; +// a node gone removes its key. It raises no event: the seat's emits name none for it. +func TestNatsTheDeclaredPackagesMirrorWritesOnlyOnAChange(t *testing.T) { + js, api := aMirrorBus(t) + events := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event.>") + w := aWriter(t, js, api, DeclaredPackages) + ctx := t.Context() + value := func(node string, resources ...map[string]any) []byte { + t.Helper() + owner := map[string]string{} + for _, r := range resources { + owner[r["id"].(string)] = "web" + } + v, err := inventory.DeclaredPackagesValue(node, resources, owner) + if err != nil { + t.Fatal(err) + } + return v + } + curl := map[string]any{"id": "web.curl", "type": "package", "package": "curl"} + nano := map[string]any{"id": "web.nano", "type": "package", "package": "nano", "absent": true} + + if r, err := w.Sync(ctx, map[string][]byte{"one": value("one", curl), "two": value("two")}); err != nil || + r != (Result{Written: 2}) { + t.Fatalf("the first pass wrote %+v (%v)", r, err) + } + before := writes(t, api, DeclaredPackages.Bucket) + if r, err := w.Sync(ctx, map[string][]byte{"one": value("one", curl), "two": value("two")}); err != nil || r != (Result{}) { + t.Errorf("a pass over the same declarations wrote %+v (%v)", r, err) + } + if got := writes(t, api, DeclaredPackages.Bucket) - before; got != 0 { + t.Errorf("a pass over the same declarations put %d messages in the bucket", got) + } + if r, err := w.Sync(ctx, map[string][]byte{"one": value("one", curl, nano), "two": value("two")}); err != nil || + r != (Result{Written: 1}) { + t.Errorf("a package newly declared wrote %+v (%v)", r, err) + } + if r, err := w.Sync(ctx, map[string][]byte{"one": value("one", curl, nano)}); err != nil || r != (Result{Removed: 1}) { + t.Errorf("a node gone wrote %+v (%v)", r, err) + } + if said := settle(events, 1); len(said) != 0 { + t.Errorf("the declared-packages mirror raised %+v", said) + } +} diff --git a/modules/route-proxy/go.mod b/modules/route-proxy/go.mod index 178b9e94..a1f29bc9 100644 --- a/modules/route-proxy/go.mod +++ b/modules/route-proxy/go.mod @@ -3,7 +3,7 @@ module route-proxy go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022075101-c61212eba77c + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876 github.com/nats-io/nats.go v1.54.0 golang.org/x/crypto v0.57.0 ) diff --git a/modules/route-proxy/go.sum b/modules/route-proxy/go.sum index 1db4a20f..b72ac074 100644 --- a/modules/route-proxy/go.sum +++ b/modules/route-proxy/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022075101-c61212eba77c h1:YdSbkI42rlVmClybH8BS717ut0vouKIMavO691fpdps= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022075101-c61212eba77c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876 h1:dtiIJzrX7ey1wmsdzapiFgCKDllCph7/vxfUp1ib/t8= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/announce/announce.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/announce/announce.go new file mode 100644 index 00000000..37dfd21d --- /dev/null +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/announce/announce.go @@ -0,0 +1,126 @@ +// Package announce is how a module says on the bus which states it keeps and which of them others may read (novox/hq +// ADR 0495 clauses 2 and 3; issue 1790): the subject it says it on, the value it says, and where the controller keeps +// the last one said by each module. What a module announced is the truth of what it keeps; a reader is granted a read +// from it, never from a list or from the owner's manifest. +// +// A module announces from its code (process.Conn.Keeps, stdio.Keeps, the TypeScript SDK's keeps), never from its +// manifest, which only the core reads (clause 1). A module that keeps no state on the bus announces nothing. +package announce + +import ( + "encoding/json" + "fmt" + "regexp" + "sort" +) + +// Version is the version of the value this package writes. +const Version = 1 + +// Event is the event type of an announcement, under the module's own namespace. +const Event = "states" + +// Subject is where module announces what it keeps: its own event subject, mesh.mod..event.states, which its +// account may already publish to and the events stream keeps. +func Subject(module string) string { return "mesh.mod." + module + ".event." + Event } + +// SubjectAll is every module's announcement, what the controller consumes. +const SubjectAll = "mesh.mod.*.event." + Event + +// KeptBucket is the controller's bucket of the last announcement of each module, by Key. +const KeptBucket = "mesh-controller_states-announced" + +var name = regexp.MustCompile(`^[a-z0-9][a-z0-9-]*$`) + +// Key is module's key in KeptBucket: the module itself, which must be one plain name, since a module's name is part of +// its buckets' names (catalogue.StateProblems). +func Key(module string) (string, error) { + if !name.MatchString(module) { + return "", fmt.Errorf("announce: %q is not a module that keeps state: lower-case letters, digits and hyphens", module) + } + return module, nil +} + +// State is one state a module keeps, by its local name, and whether others may read it. Others never write it: a +// state another module writes is the owner's own code's business. +type State struct { + Name string `json:"name"` + Readable bool `json:"readable,omitempty"` +} + +// States is an announcement: the module, the machine whose instance said it, and every state the module keeps, sorted +// by name. An empty list says the module keeps nothing. +type States struct { + V int `json:"v"` + Module string `json:"module"` + Node string `json:"node,omitempty"` + States []State `json:"states"` +} + +// Keeps says whether the module announced keeping the state. +func (s States) Keeps(state string) bool { + for _, k := range s.States { + if k.Name == state { + return true + } + } + return false +} + +// Readable says whether the module announced the state readable by others. +func (s States) Readable(state string) bool { + for _, k := range s.States { + if k.Name == state { + return k.Readable + } + } + return false +} + +func (s *States) check() error { + if s.Module == "" { + return fmt.Errorf("announce: an announcement names no module") + } + if _, err := Key(s.Module); err != nil { + return err + } + seen := map[string]bool{} + for _, k := range s.States { + if !name.MatchString(k.Name) { + return fmt.Errorf("announce: %s announces the state %q, and a state is named locally: lower-case letters, digits and hyphens", s.Module, k.Name) + } + if seen[k.Name] { + return fmt.Errorf("announce: %s announces the state %q twice", s.Module, k.Name) + } + seen[k.Name] = true + } + return nil +} + +// Encode is s as JSON at Version, its states sorted; it refuses a module or a state no bucket could be named by and a +// state announced twice, naming it. +func Encode(s States) ([]byte, error) { + if err := s.check(); err != nil { + return nil, err + } + s.V = Version + s.States = append([]State{}, s.States...) + sort.Slice(s.States, func(i, j int) bool { return s.States[i].Name < s.States[j].Name }) + return json.Marshal(s) +} + +// Decode reads an announcement; it refuses another version and what Encode refuses. +func Decode(data []byte) (States, error) { + var s States + if err := json.Unmarshal(data, &s); err != nil { + return States{}, fmt.Errorf("announce: the announcement is not JSON: %w", err) + } + if s.V != Version { + return States{}, fmt.Errorf("announce: the announcement is version %d, and this reads version %d", s.V, Version) + } + if err := s.check(); err != nil { + return States{}, err + } + sort.Slice(s.States, func(i, j int) bool { return s.States[i].Name < s.States[j].Name }) + return s, nil +} diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/keeps.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/keeps.go new file mode 100644 index 00000000..cffdf9c7 --- /dev/null +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/keeps.go @@ -0,0 +1,80 @@ +package stdio + +import ( + "encoding/json" + "fmt" + "os" + "strings" + "sync" + + "git.novox.be/novox/mesh-sdk/go/announce" +) + +// What this bundle said it keeps (novox/hq ADR 0495 clause 2; issue 1790). +var ( + keepsMu sync.Mutex + kept map[string]bool // nil until Keeps: no state is refused then + announced []byte // the announcement, published once the bundle serves +) + +// Keeps says on the bus, as this bundle's module, every state it keeps and which of them others may read (novox/hq +// ADR 0495 clause 2): what a reader is granted is what the owner announced (clause 3). Call it before Serve; the +// announcement is published through the runtime once the bundle serves, or at once when it already does. From then +// on a state of the module's own that it did not announce is refused at the call, naming it, so code and announcement +// cannot drift (clause 5). It refuses a state no bucket could be named by and a state given twice. +func Keeps(states ...announce.State) error { + module := os.Getenv("MESH_MODULE") + if module == "" { + module = os.Getenv(ServedAs) + } + data, err := announce.Encode(announce.States{Module: module, Node: os.Getenv("MESH_NODE"), States: states}) + if err != nil { + return err + } + names := map[string]bool{} + for _, s := range states { + names[s.Name] = true + } + keepsMu.Lock() + kept, announced = names, data + keepsMu.Unlock() + currentMu.Lock() + serving := current != nil + currentMu.Unlock() + if serving { + return announce1() + } + return nil +} + +// announce1 publishes the announcement Keeps made, if any. +func announce1() error { + keepsMu.Lock() + data := announced + keepsMu.Unlock() + if data == nil { + return nil + } + return Emit(announce.Event, json.RawMessage(data)) +} + +// refusedUnannounced is the error of a call on a state of the module's own that Keeps did not name, or nil. A read of +// another module's state (`.`) is the grant's business. +func refusedUnannounced(name string) error { + if strings.Contains(name, ".") { + return nil + } + keepsMu.Lock() + defer keepsMu.Unlock() + if kept != nil && !kept[name] { + return fmt.Errorf("the state %q is not among the states this module announced it keeps (stdio.Keeps)", name) + } + return nil +} + +// forgetKeeps is for a test: nothing announced. +func forgetKeeps() { + keepsMu.Lock() + kept, announced = nil, nil + keepsMu.Unlock() +} diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/merge-check.sh b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/merge-check.sh index 92e9d9cd..00587326 100644 --- a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/merge-check.sh +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/merge-check.sh @@ -10,13 +10,17 @@ # 1. the unit level: `go test ./...` untagged, no bus or other process, timed and held to a bound # (unit_bound below); over it the check fails and names the slowest tests (issue 895); # 2. the integration level: the whole suite with `-tags integration`, each test against a bus of its -# own (internal/testbus, which issue 1171 brings with the first test that needs a bus) so nothing is +# own (testbus, which issue 1171 brings with the first test that needs a bus) so nothing is # serialised. With no tagged test yet, the level runs the unit tests again and says their count. # # **A test that starts a bus or another process carries `//go:build integration`** as the first line of # its file; a file that held both kinds is two files (`_integration_test.go` beside `_test.go`). # Each level says its test count and its time. # +# **A test that needs bubblewrap carries `//go:build integration && isolation`** (issue 1309's proof that a run step +# is shut off): this script vets it but does not run it, since the toolchain it runs in has no bubblewrap; the checker +# runs the `isolation` tag on the build nodes, which have it (R4e). +# # No dependency is committed (no vendor directory: .gitignore; the operator, 2026-10-19; issue 1304). Every dependency # is resolved by the version and checksum go.sum names (novox/hq ADR 0405 §5), from its public source: this binding # requires nothing of the mesh's own forge. `go mod verify` fails by module name when a downloaded dependency does not @@ -98,6 +102,7 @@ fi go vet ./... # The files tagged `integration` are vetted too, or nothing would read them. go vet -tags integration ./... +go vet -tags integration,isolation ./... # NOT RACE-CHECKED here: the SDK's suite is small and carries no store or other shared state between # packages, so the race detector is not required of this check (novox/hq issue 1175); a module that needs @@ -117,7 +122,7 @@ if [ "$status" -ne 0 ]; then fi judge_unit_run "$work/unit.out" "$unit_bound" -# The integration level: the whole suite with the tag, each test against its own bus (internal/testbus). +# The integration level: the whole suite with the tag, each test against its own bus (testbus). integration_start=$(date +%s) status=0 go test -tags integration -count=1 -v ./... >"$work/integration.out" 2>&1 || status=$? diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/conn.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/conn.go index 58efb40c..ca469221 100644 --- a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/conn.go +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/conn.go @@ -24,6 +24,12 @@ type Conn struct { mu sync.Mutex next uint64 pending map[string]map[uint64]context.CancelCauseFunc + + // What Keeps announced (keeps.go): the states' names (nil before Keeps), the announcement, and whether a + // reconnect announces it again. + kept map[string]bool + announced []byte + reannounces bool } // errPermissionRefused marks a context cancelled because the bus refused this account's publish of the diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/keeps.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/keeps.go new file mode 100644 index 00000000..22a1df79 --- /dev/null +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/keeps.go @@ -0,0 +1,71 @@ +package process + +import ( + "encoding/json" + "fmt" + + "github.com/nats-io/nats.go" + + "git.novox.be/novox/mesh-sdk/go/announce" +) + +// Keeps says on the bus, as this process's module, every state it keeps and which of them others may read (novox/hq +// ADR 0495 clause 2; issue 1790), on announce.Subject; what a reader is granted is what the owner announced (clause 3). +// It publishes at once and again after every reconnect, so the controller hears it whenever it missed it. From then on +// a state of the module that it did not announce is refused at the call, naming it, so code and announcement cannot +// drift (clause 5). It refuses a state no bucket could be named by and a state given twice. +func (c *Conn) Keeps(states ...announce.State) error { + if err := c.remember(states); err != nil { + return err + } + c.mu.Lock() + first := !c.reannounces + c.reannounces = true + c.mu.Unlock() + if first { + before := c.nc.Opts.ReconnectedCB + c.nc.SetReconnectHandler(func(nc *nats.Conn) { + if before != nil { + before(nc) + } + _ = c.publishKeeps() + }) + } + return c.publishKeeps() +} + +// remember makes the announcement of states and binds only them from now on. +func (c *Conn) remember(states []announce.State) error { + data, err := announce.Encode(announce.States{Module: c.account.Module, Node: c.account.Node, States: states}) + if err != nil { + return err + } + names := map[string]bool{} + for _, s := range states { + names[s.Name] = true + } + c.mu.Lock() + c.kept, c.announced = names, data + c.mu.Unlock() + return nil +} + +func (c *Conn) publishKeeps() error { + c.mu.Lock() + data := c.announced + c.mu.Unlock() + if err := c.Emit(announce.Event, json.RawMessage(data)); err != nil { + return fmt.Errorf("announcing what %s keeps: %w", c.account.Module, err) + } + return nil +} + +// refusedUnannounced is the error of a bind of a state Keeps did not name, or nil (also before Keeps). +func (c *Conn) refusedUnannounced(name string) error { + c.mu.Lock() + defer c.mu.Unlock() + if c.kept != nil && !c.kept[name] { + return fmt.Errorf("the state %q is not among the states %s announced it keeps (Conn.Keeps)", name, c.account.Module) + } + return nil +} diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/state.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/state.go index 43221fac..777a9fd1 100644 --- a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/state.go +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/state.go @@ -5,6 +5,8 @@ import ( "encoding/json" "errors" "fmt" + "sort" + "strings" "time" "github.com/nats-io/nats.go" @@ -27,6 +29,8 @@ type State struct { // kv binds this state's underlying key-value store: a live connection's own (Conn.State), or, in a // unit test, a fake that never reaches a bus. kv func(ctx context.Context) (jetstream.KeyValue, error) + // closed says the connection is gone, so a watch ends; nil in a unit test. + closed func() bool } // bucket is this state's name on the bus. @@ -38,7 +42,10 @@ func (s State) bucket() string { return s.Module + "_" + s.Name } // not granted is refused there, promptly, as nats.ErrPermissionViolation. func (c *Conn) State(name string) State { bucket := c.account.Module + "_" + name - return State{Module: c.account.Module, Name: name, kv: func(ctx context.Context) (jetstream.KeyValue, error) { + return State{Module: c.account.Module, Name: name, closed: func() bool { return c.nc.IsClosed() }, kv: func(ctx context.Context) (jetstream.KeyValue, error) { + if err := c.refusedUnannounced(name); err != nil { + return nil, err + } js, err := jetstream.New(c.nc) if err != nil { return nil, err @@ -128,3 +135,148 @@ func (s State) Update(key string, value any, revision uint64) (uint64, error) { } return rev, nil } + +// Put writes a key whole, whatever it held, and returns the revision written. +func (s State) Put(key string, value any) (uint64, error) { + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + kv, err := s.store(ctx) + if err != nil { + return 0, err + } + data, err := json.Marshal(value) + if err != nil { + return 0, err + } + rev, err := kv.Put(ctx, key, data) + if err != nil { + return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err) + } + return rev, nil +} + +// Keys is every key with a current value, sorted; none is an empty list. +func (s State) Keys() ([]string, error) { + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + kv, err := s.store(ctx) + if err != nil { + return nil, err + } + keys, err := kv.Keys(ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return []string{}, nil + } + if err != nil { + return nil, fmt.Errorf("%s's state %s, keys: %w", s.Module, s.Name, err) + } + sort.Strings(keys) + return keys, nil +} + +// watchPattern is a key narrowing as the bus names it: `*` stays one name, `**` is the rest, empty is every key. +func watchPattern(key string) string { + if key == "" { + return jetstream.AllKeys + } + tokens := strings.Split(key, ".") + for i, t := range tokens { + if t == "**" { + tokens[i] = ">" + } + } + return strings.Join(tokens, ".") +} + +// Watch hands handle every current value — none that is deleted, each with Current set — and then every +// change in order, and returns once the current values have all been handled. key narrows it: a key, with +// `*` for one name and `**` for the rest; empty is every key. The changes go on for the life of the +// connection; an error from handle, or the connection closing, ends them. Before the current values are +// handled, either is returned as the error. +func (s State) Watch(key string, handle func(stdio.StateChange) error) error { + // the watch outlives this call, so its context is not the call's timeout: the connection ends it. + ctx, cancel := context.WithCancel(context.Background()) + bind, stopBind := context.WithTimeout(ctx, stateTimeout) + kv, err := s.store(bind) + stopBind() + if err != nil { + cancel() + return err + } + w, err := kv.Watch(ctx, watchPattern(key)) + if err != nil { + cancel() + return fmt.Errorf("%s's state %s, watching %q: %w", s.Module, s.Name, key, err) + } + end := func() { w.Stop(); cancel() } + next := func() (jetstream.KeyValueEntry, bool, error) { + tick := time.NewTicker(250 * time.Millisecond) + defer tick.Stop() + for { + select { + case e, open := <-w.Updates(): + if !open { + return nil, false, fmt.Errorf("%s's state %s: the watch ended", s.Module, s.Name) + } + return e, true, nil + case <-tick.C: + if s.closed != nil && s.closed() { + return nil, false, fmt.Errorf("%s's state %s: the connection closed", s.Module, s.Name) + } + } + } + } + deliver := func(e jetstream.KeyValueEntry, current bool) error { + c := stdio.StateChange{State: s.Name, Key: e.Key(), Op: "put", Revision: e.Revision(), Current: current} + if e.Operation() == jetstream.KeyValuePut { + c.Value = json.RawMessage(e.Value()) + } else { + c.Op = "delete" + } + return handle(c) + } + for { + e, _, err := next() + if err != nil { + end() + return err + } + if e == nil { + break + } + if e.Operation() != jetstream.KeyValuePut { + continue + } + if err := deliver(e, true); err != nil { + end() + return err + } + } + go func() { + defer end() + for { + e, _, err := next() + if err != nil || e == nil { + return + } + if deliver(e, false) != nil { + return + } + } + }() + return nil +} + +// Delete removes a key. A key that was not there is not an error. +func (s State) Delete(key string) error { + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + kv, err := s.store(ctx) + if err != nil { + return err + } + if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + return fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err) + } + return nil +} diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/state.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/state.go index 70e1b35c..56490b6e 100644 --- a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/state.go +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/state.go @@ -50,6 +50,9 @@ func State(name string) KeptState { return KeptState{name: name} } const stateTimeout = 30 * time.Second func (s KeptState) ask(verb string, params map[string]any) (json.RawMessage, error) { + if err := refusedUnannounced(s.name); err != nil { + return nil, err + } c, err := live() if err != nil { return nil, err diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/stdio.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/stdio.go index a61f4dec..a072f40d 100644 --- a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/stdio.go +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/stdio.go @@ -180,6 +180,9 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error { currentMu.Lock() current = c currentMu.Unlock() + // What the bundle said it keeps is announced once it serves: the runtime answers the publish on this channel, so + // it is asked from beside the loop that reads the answer. + go func() { _ = announce1() }() scanner := bufio.NewScanner(in) scanner.Buffer(make([]byte, 1<<20), 1<<24) for scanner.Scan() { diff --git a/modules/route-proxy/vendor/modules.txt b/modules/route-proxy/vendor/modules.txt index 909cdc3f..c65612ed 100644 --- a/modules/route-proxy/vendor/modules.txt +++ b/modules/route-proxy/vendor/modules.txt @@ -1,6 +1,7 @@ -# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022075101-c61212eba77c +# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876 ## explicit; go 1.26.0 git.novox.be/novox/mesh-sdk/go +git.novox.be/novox/mesh-sdk/go/announce git.novox.be/novox/mesh-sdk/go/membership git.novox.be/novox/mesh-sdk/go/process git.novox.be/novox/mesh-sdk/go/provider diff --git a/testdata/beside/CAPTURED b/testdata/beside/CAPTURED index e8914254..e9c65105 100644 --- a/testdata/beside/CAPTURED +++ b/testdata/beside/CAPTURED @@ -9,7 +9,7 @@ a second place to clean besides the catalogue itself. mesh-catalog e0bc919d4945bb5f13f971f74643a0b714cf45da modules/*/module.json, modules/nats/Dockerfile mesh-host 9d9e3a46c4f5943cfffb413d962ce48b75e2fd39 examples/foundation-first-node.lock -mesh-sdk e76a5a31a07bedb0cfa1e1244e54b391a33f9b75 conformance/events/module-event.json +mesh-sdk 875c680d9876863a3817bbf253d5fe4e54d62c2e conformance/events/module-event.json To move them, from this repository's root, with the two repositories checked out beside it: diff --git a/vendor/git.novox.be/novox/mesh-sdk/go/buildstep/narrow.go b/vendor/git.novox.be/novox/mesh-sdk/go/buildstep/narrow.go new file mode 100644 index 00000000..9bc6e00f --- /dev/null +++ b/vendor/git.novox.be/novox/mesh-sdk/go/buildstep/narrow.go @@ -0,0 +1,69 @@ +package buildstep + +import ( + "fmt" + "os" + "path" + "path/filepath" + "strings" +) + +// narrowKept are the files of a module's directory a narrowed build keeps beside its named paths: its manifest and +// its lock files. +var narrowKept = []string{"module.json", "package.json", "package-lock.json", "go.mod", "go.sum"} + +// Narrow leaves in the fetched tree only what a build of an artifact that names its paths may see (novox/hq ADR 0499 +// clause 4): r.Dir's named paths, its manifest and lock files, and .git; everything else is removed, so a file outside +// them cannot change what is built without changing its key. A named path absent at the tree is an error naming it, +// and nothing is removed then. With no paths in r it changes nothing. Both build modules' fetch steps call it. +func Narrow(tree string, r FetchRequest) error { + if len(r.Paths) == 0 { + return nil + } + keep := map[string]bool{".git": true} + for _, p := range r.Paths { + rel := path.Join(r.Dir, p) + if strings.HasPrefix(rel, "../") || rel == ".." || path.IsAbs(p) { + return fmt.Errorf("the named path %q leaves the module's directory", p) + } + if _, err := os.Lstat(filepath.Join(tree, filepath.FromSlash(rel))); err != nil { + return fmt.Errorf("the named path %q is not in the fetched tree: %w", p, err) + } + keep[rel] = true + } + for _, f := range narrowKept { + keep[path.Join(r.Dir, f)] = true + } + return prune(tree, "", keep) +} + +// prune removes from dir (rel within the tree) every entry that is neither kept nor on the way to one. +func prune(root, rel string, keep map[string]bool) error { + entries, err := os.ReadDir(filepath.Join(root, filepath.FromSlash(rel))) + if err != nil { + return err + } + for _, e := range entries { + p := path.Join(rel, e.Name()) + if keep[p] { + continue + } + ancestor := false + for k := range keep { + if strings.HasPrefix(k, p+"/") { + ancestor = true + break + } + } + if ancestor && e.IsDir() { + if err := prune(root, p, keep); err != nil { + return err + } + continue + } + if err := os.RemoveAll(filepath.Join(root, filepath.FromSlash(p))); err != nil { + return err + } + } + return nil +} diff --git a/vendor/git.novox.be/novox/mesh-sdk/go/compose/network.go b/vendor/git.novox.be/novox/mesh-sdk/go/compose/network.go index 1e66c516..afc8363c 100644 --- a/vendor/git.novox.be/novox/mesh-sdk/go/compose/network.go +++ b/vendor/git.novox.be/novox/mesh-sdk/go/compose/network.go @@ -74,7 +74,10 @@ func (c *composer) overlayRange() string { } // network builds the private network over the machines that resolved the module for it. A machine is on it because it -// was given the module, or because it is joining with a key and an address (novox/hq ADR 0169). +// was given the module, or because it is joining with a key and an address (novox/hq ADR 0169), or because its own set +// could not be resolved and it holds a place (a key and an address): that place is its last good one on the network, and +// it does not change because its modules failed, so every other machine keeps its path to it -- the hub above all -- and +// only its own declaration is refused (issue 1833). A machine refused with no place is left off. func (c *composer) network(on map[string]bool, refused map[string]string) (*overlay.Generator, error) { tunnels := c.in.Tunnels carried := c.in.CarriedPeers @@ -85,9 +88,15 @@ func (c *composer) network(on map[string]bool, refused map[string]string) (*over var nodes []overlay.Node for _, node := range c.in.sorted() { p := node.Place - if !on[node.Name] && !(isJoining[node.Name] && p.Key != "" && p.Address != "") { + placed := p.Key != "" && p.Address != "" + _, wasRefused := refused[node.Name] + if !on[node.Name] && !((isJoining[node.Name] || wasRefused) && placed) { continue } + if !on[node.Name] && wasRefused && !isJoining[node.Name] { + c.say("%s could not be resolved and keeps its place on the private network at %s, so the others keep their path to it", + node.Name, p.Address) + } n := overlay.Node{ Name: node.Name, Key: p.Key, Endpoint: p.Endpoint, Site: p.Site, Hub: p.Hub, Address: p.Address, @@ -233,7 +242,9 @@ func (c *composer) whereEveryoneIs() (map[string]string, error) { } // onTheNetwork is every placed machine that resolves the private network: has an address AND runs what puts it there -// (novox/hq issue 079). A machine that does not resolve is not counted, and said. +// (novox/hq issue 079). A machine that does not resolve keeps its place when it holds one (a key and an address), by the +// same rule as network(), so its name and its address stay where the tunnel to it does (issue 1833); one without is not +// counted, and said. func (c *composer) onTheNetwork() []Node { var out []Node for _, n := range c.in.sorted() { @@ -247,6 +258,12 @@ func (c *composer) onTheNetwork() []Node { catalogue.Node{Name: n.Name, Site: n.Place.Site, Capabilities: n.Capabilities}, catalogue.World{Unchecked: true, Holdings: c.in.Holdings}) if err != nil { + if n.Place.Key != "" { + c.say("%s could not be resolved and keeps its place on the private network at %s, so the others keep their path to it", + n.Name, n.Place.Address) + out = append(out, n) + continue + } c.say("%s is not counted as on the network: it does not resolve: %v", n.Name, err) continue } diff --git a/vendor/git.novox.be/novox/mesh-sdk/go/process/state.go b/vendor/git.novox.be/novox/mesh-sdk/go/process/state.go index 8e294e31..777a9fd1 100644 --- a/vendor/git.novox.be/novox/mesh-sdk/go/process/state.go +++ b/vendor/git.novox.be/novox/mesh-sdk/go/process/state.go @@ -5,6 +5,8 @@ import ( "encoding/json" "errors" "fmt" + "sort" + "strings" "time" "github.com/nats-io/nats.go" @@ -27,6 +29,8 @@ type State struct { // kv binds this state's underlying key-value store: a live connection's own (Conn.State), or, in a // unit test, a fake that never reaches a bus. kv func(ctx context.Context) (jetstream.KeyValue, error) + // closed says the connection is gone, so a watch ends; nil in a unit test. + closed func() bool } // bucket is this state's name on the bus. @@ -38,7 +42,7 @@ func (s State) bucket() string { return s.Module + "_" + s.Name } // not granted is refused there, promptly, as nats.ErrPermissionViolation. func (c *Conn) State(name string) State { bucket := c.account.Module + "_" + name - return State{Module: c.account.Module, Name: name, kv: func(ctx context.Context) (jetstream.KeyValue, error) { + return State{Module: c.account.Module, Name: name, closed: func() bool { return c.nc.IsClosed() }, kv: func(ctx context.Context) (jetstream.KeyValue, error) { if err := c.refusedUnannounced(name); err != nil { return nil, err } @@ -131,3 +135,148 @@ func (s State) Update(key string, value any, revision uint64) (uint64, error) { } return rev, nil } + +// Put writes a key whole, whatever it held, and returns the revision written. +func (s State) Put(key string, value any) (uint64, error) { + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + kv, err := s.store(ctx) + if err != nil { + return 0, err + } + data, err := json.Marshal(value) + if err != nil { + return 0, err + } + rev, err := kv.Put(ctx, key, data) + if err != nil { + return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err) + } + return rev, nil +} + +// Keys is every key with a current value, sorted; none is an empty list. +func (s State) Keys() ([]string, error) { + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + kv, err := s.store(ctx) + if err != nil { + return nil, err + } + keys, err := kv.Keys(ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return []string{}, nil + } + if err != nil { + return nil, fmt.Errorf("%s's state %s, keys: %w", s.Module, s.Name, err) + } + sort.Strings(keys) + return keys, nil +} + +// watchPattern is a key narrowing as the bus names it: `*` stays one name, `**` is the rest, empty is every key. +func watchPattern(key string) string { + if key == "" { + return jetstream.AllKeys + } + tokens := strings.Split(key, ".") + for i, t := range tokens { + if t == "**" { + tokens[i] = ">" + } + } + return strings.Join(tokens, ".") +} + +// Watch hands handle every current value — none that is deleted, each with Current set — and then every +// change in order, and returns once the current values have all been handled. key narrows it: a key, with +// `*` for one name and `**` for the rest; empty is every key. The changes go on for the life of the +// connection; an error from handle, or the connection closing, ends them. Before the current values are +// handled, either is returned as the error. +func (s State) Watch(key string, handle func(stdio.StateChange) error) error { + // the watch outlives this call, so its context is not the call's timeout: the connection ends it. + ctx, cancel := context.WithCancel(context.Background()) + bind, stopBind := context.WithTimeout(ctx, stateTimeout) + kv, err := s.store(bind) + stopBind() + if err != nil { + cancel() + return err + } + w, err := kv.Watch(ctx, watchPattern(key)) + if err != nil { + cancel() + return fmt.Errorf("%s's state %s, watching %q: %w", s.Module, s.Name, key, err) + } + end := func() { w.Stop(); cancel() } + next := func() (jetstream.KeyValueEntry, bool, error) { + tick := time.NewTicker(250 * time.Millisecond) + defer tick.Stop() + for { + select { + case e, open := <-w.Updates(): + if !open { + return nil, false, fmt.Errorf("%s's state %s: the watch ended", s.Module, s.Name) + } + return e, true, nil + case <-tick.C: + if s.closed != nil && s.closed() { + return nil, false, fmt.Errorf("%s's state %s: the connection closed", s.Module, s.Name) + } + } + } + } + deliver := func(e jetstream.KeyValueEntry, current bool) error { + c := stdio.StateChange{State: s.Name, Key: e.Key(), Op: "put", Revision: e.Revision(), Current: current} + if e.Operation() == jetstream.KeyValuePut { + c.Value = json.RawMessage(e.Value()) + } else { + c.Op = "delete" + } + return handle(c) + } + for { + e, _, err := next() + if err != nil { + end() + return err + } + if e == nil { + break + } + if e.Operation() != jetstream.KeyValuePut { + continue + } + if err := deliver(e, true); err != nil { + end() + return err + } + } + go func() { + defer end() + for { + e, _, err := next() + if err != nil || e == nil { + return + } + if deliver(e, false) != nil { + return + } + } + }() + return nil +} + +// Delete removes a key. A key that was not there is not an error. +func (s State) Delete(key string) error { + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + kv, err := s.store(ctx) + if err != nil { + return err + } + if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + return fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err) + } + return nil +} diff --git a/vendor/modules.txt b/vendor/modules.txt index 7ba7a3f9..ca0e1ff9 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -1,4 +1,4 @@ -# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161855-e76a5a31a07b +# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876 ## explicit; go 1.26.0 git.novox.be/novox/mesh-sdk/go git.novox.be/novox/mesh-sdk/go/announce