Merge pull request 'Keep each node's declared packages in a state the pacman module reads (upgrade safety, issue 1840, ADR 0507 §5)' (#439) from feat/upgrade-safety into main
This commit was merged in pull request #439.
This commit is contained in:
@@ -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])
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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).
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.20261022132648-bd7731f3588c
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161324-8892a1176800
|
||||
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
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5 h1:E5stfIeCN6uYgA9TCv+4K8JgaURg3zQppvUgJLaDMP4=
|
||||
git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5/go.mod h1:4wyoCr6IIIIarEUuROThWDj7DAvJvIidqMjWcpSnOwE=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022132648-bd7731f3588c h1:UKlwMw6fmesDqe0aL0lfI/ivhorsGjAvh9XefFjhNI4=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022132648-bd7731f3588c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161324-8892a1176800 h1:Q29osS06HHcXH1WXURDyITbOLnJG7XJYgCzmZD++Nws=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161324-8892a1176800/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=
|
||||
|
||||
@@ -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
|
||||
@@ -153,7 +158,7 @@ func ChainBuckets() []string { return []string{BuildsBucket, RolloutsBucket, Ass
|
||||
// ControllerMirrors are the local names of the controller's buckets a module may read: the record mirrors and the
|
||||
// chain's buckets.
|
||||
func ControllerMirrors() []string {
|
||||
return []string{MirrorAssignments, MirrorDeclaredConfig, MirrorPredefined, ChainBuilds, ChainRollouts, ChecksRunning, PlansBuilding}
|
||||
return []string{MirrorAssignments, MirrorDeclaredConfig, MirrorPredefined, ChainBuilds, ChainRollouts, ChecksRunning, PlansBuilding, MirrorDeclaredPackages}
|
||||
}
|
||||
|
||||
// MirrorsWrittenByAnother are the controller's mirrors whose writer is not the controller, each with its reason: the rule
|
||||
@@ -222,6 +227,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
|
||||
@@ -230,7 +236,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}
|
||||
}
|
||||
|
||||
@@ -446,6 +452,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,
|
||||
|
||||
@@ -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(ControllerMirrors(), "declared-packages") ||
|
||||
BucketName(ControllerSeat, MirrorDeclaredPackages) != DeclaredPackagesBucket {
|
||||
t.Fatalf("a module cannot name %s in `reads`: %v", DeclaredPackagesBucket, ControllerMirrors())
|
||||
}
|
||||
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+".>")
|
||||
}
|
||||
@@ -429,6 +429,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 " +
|
||||
|
||||
@@ -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_<module>_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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -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.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "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.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "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" }
|
||||
} }
|
||||
|
||||
@@ -236,6 +236,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.
|
||||
|
||||
@@ -49,6 +49,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",
|
||||
}
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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()}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
Vendored
+1
-1
@@ -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 bd7731f3588cf336620526b032bbfe9867f7837e conformance/events/module-event.json
|
||||
mesh-sdk 8892a1176800cb27e818af8076cf05f890a84807 conformance/events/module-event.json
|
||||
|
||||
To move them, from this repository's root, with the two repositories checked out beside it:
|
||||
|
||||
|
||||
@@ -134,8 +134,8 @@ func StateProblems(m Manifest) []string {
|
||||
// second way and granted to the module's own principal as well. Named here so a read of one is a read of
|
||||
// kept state. The bus's list is broker.ControllerMirrors; a test keeps the two agreeing.
|
||||
// plans-building is the plans in their build phase, read by the checker to hold checks while a delivery builds (ADR 0483
|
||||
// clause 2, issue 1687).
|
||||
var ControllerMirrors = []string{"assignments", "declared-config", "predefined", "builds", "rollouts", "checks-running", "plans-building"}
|
||||
// clause 2, issue 1687). declared-packages is the packages a node's declaration names (ADR 0507, issue 1840/S1c).
|
||||
var ControllerMirrors = []string{"assignments", "declared-config", "predefined", "builds", "rollouts", "checks-running", "plans-building", "declared-packages"}
|
||||
|
||||
// StateReadsNothingDeclares is every read across a catalogue whose owner is present and declares no
|
||||
// such state. An absent owner says nothing — a module may be installed long before the one whose
|
||||
|
||||
+150
-1
@@ -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) {
|
||||
js, err := jetstream.New(c.nc)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -128,3 +132,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
|
||||
}
|
||||
|
||||
Vendored
+1
-1
@@ -1,4 +1,4 @@
|
||||
# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022132648-bd7731f3588c
|
||||
# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161324-8892a1176800
|
||||
## explicit; go 1.26.0
|
||||
git.novox.be/novox/mesh-sdk/go
|
||||
git.novox.be/novox/mesh-sdk/go/asks
|
||||
|
||||
Reference in New Issue
Block a user