Merge branch 'feat/delivery-have-broker' into feat/delivery-have
This commit is contained in:
@@ -5,6 +5,7 @@ import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
@@ -189,14 +190,15 @@ func IsControllerBucket(bucket string) bool {
|
||||
bucket == ConditionHistoryBucket || bucket == LeaseBucket || bucket == AskedBucket ||
|
||||
bucket == AssignmentsBucket || bucket == DeclaredConfigBucket || bucket == PredefinedBucket ||
|
||||
bucket == MergeWindowBucket || bucket == RolloutSilentBucket || bucket == CheckRunningBucket ||
|
||||
bucket == DeliveryMergesBucket || bucket == BuildsBucket || bucket == RolloutsBucket
|
||||
bucket == DeliveryMergesBucket || bucket == BuildsBucket || bucket == RolloutsBucket ||
|
||||
bucket == delivery.HaveBucket
|
||||
}
|
||||
|
||||
// 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,
|
||||
DeliveryMergesBucket, BuildsBucket, RolloutsBucket}
|
||||
DeliveryMergesBucket, BuildsBucket, RolloutsBucket, delivery.HaveBucket}
|
||||
}
|
||||
|
||||
// ControllerBucketsAsserter is what raising the controller's buckets needs of a connection.
|
||||
@@ -370,6 +372,12 @@ func (j *JetStream) EnsureControllerBuckets() error {
|
||||
return fmt.Errorf("asserting bucket %s: %w", c.bucket, err)
|
||||
}
|
||||
}
|
||||
// **The have bucket: what each machine has of each module** (ADR 0429 §1, issue 1136). Twenty revisions of each key and
|
||||
// no age. Asserted by the controller, written by each machine's node-engine under its own node's keys and by nobody
|
||||
// else: the controller's grant leaves it out.
|
||||
if _, err := js.CreateOrUpdateKeyValue(ctx, HaveBucketConfig()); err != nil {
|
||||
return fmt.Errorf("asserting bucket %s: %w", delivery.HaveBucket, err)
|
||||
}
|
||||
// **The assignments keep the marker age they had as a mirror.** A stream that allows messages an age cannot be
|
||||
// made to disallow them (the server refuses: "message TTL status can not be disabled"), so the bucket that was
|
||||
// the assignments mirror is asserted with the age it was made with. Nothing is purged from it, so no marker is ever
|
||||
@@ -439,3 +447,18 @@ func EnsureLeaseBucket(ctx context.Context, js jetstream.JetStream) error {
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// HaveBucketConfig is the options of the have bucket (novox/hq ADR 0429 §1): delivery.HaveHistory revisions of each
|
||||
// key, no age, a value under a kilobyte (a have is JSON of about 500 bytes), 64 MiB on disk like the assignments.
|
||||
func HaveBucketConfig() jetstream.KeyValueConfig {
|
||||
return jetstream.KeyValueConfig{
|
||||
Bucket: delivery.HaveBucket,
|
||||
Description: "what each machine has of each module, one key `<node>.<module>` (and `<node>.node` for the machine's " +
|
||||
"own facts) (novox/hq ADR 0429, design 47): written by each machine's node-engine under its own node's keys " +
|
||||
"alone, never by the controller; read by the delivery view and the watchman",
|
||||
History: delivery.HaveHistory,
|
||||
MaxValueSize: 4 << 10,
|
||||
MaxBytes: 64 << 20,
|
||||
Storage: jetstream.FileStorage,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,154 @@
|
||||
//go:build integration
|
||||
|
||||
package broker
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"github.com/nats-io/nats-server/v2/server"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
"golang.org/x/crypto/bcrypt"
|
||||
)
|
||||
|
||||
// **Each node's credential writes its own have keys and no other** (novox/hq ADR 0429 §1, issue 1136, 1136/AC4): on a
|
||||
// real server whose users are composed as the controller composes them, node alpha's put of alpha.zsh is taken, its
|
||||
// put of beta.zsh is refused by the server, a module reading node-engine.have is refused any put and its watch sees
|
||||
// alpha.zsh.
|
||||
func TestNatsANodePutsOnlyItsOwnHaveKeys(t *testing.T) { // 1136/AC4
|
||||
hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
users, err := Users(Records{Nodes: []string{"alpha", "beta"}, Assigned: map[string][]Declared{
|
||||
"alpha": {{Module: RuntimeModule}, {Module: "mesh-delivery-view", Reads: []string{"node-engine.have"}}},
|
||||
"beta": {{Module: RuntimeModule}},
|
||||
}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for i := range users {
|
||||
users[i].PasswordHash = string(hash)
|
||||
}
|
||||
accounts, err := ComposeAccounts(users)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
conf := filepath.Join(t.TempDir(), "bus.conf")
|
||||
if err := os.WriteFile(conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: "+
|
||||
`"`+t.TempDir()+`"`+" }\n"+accounts), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
opts, err := server.ProcessConfigFile(conf)
|
||||
if err != nil {
|
||||
t.Fatalf("the composed accounts do not parse: %v", err)
|
||||
}
|
||||
opts.NoLog, opts.NoSigs = true, true
|
||||
s, err := server.NewServer(opts)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
go s.Start()
|
||||
if !s.ReadyForConnections(10 * time.Second) {
|
||||
t.Fatal("the bus did not come up")
|
||||
}
|
||||
t.Cleanup(s.Shutdown)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
connect := func(user string) (*nats.Conn, chan error) {
|
||||
t.Helper()
|
||||
refused := make(chan error, 8)
|
||||
nc, err := nats.Connect(s.ClientURL(), nats.UserInfo(user, "pw"), nats.CustomInboxPrefix("_INBOX."+user),
|
||||
nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refused <- err }))
|
||||
if err != nil {
|
||||
t.Fatalf("%s cannot connect: %v", user, err)
|
||||
}
|
||||
t.Cleanup(nc.Close)
|
||||
return nc, refused
|
||||
}
|
||||
controller, _ := connect(ControllerName)
|
||||
cjs, err := jetstream.New(controller)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := cjs.CreateOrUpdateKeyValue(ctx, HaveBucketConfig()); err != nil {
|
||||
t.Fatalf("the controller cannot make %s: %v", delivery.HaveBucket, err)
|
||||
}
|
||||
|
||||
alpha, _ := connect("node.alpha")
|
||||
ajs, err := jetstream.New(alpha)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
akv, err := ajs.KeyValue(ctx, delivery.HaveBucket)
|
||||
if err != nil {
|
||||
t.Fatalf("node alpha cannot bind %s: %v", delivery.HaveBucket, err)
|
||||
}
|
||||
if _, err := akv.Put(ctx, "alpha.zsh", []byte(`{"v":1}`)); err != nil {
|
||||
t.Fatalf("node alpha cannot put alpha.zsh: %v", err)
|
||||
}
|
||||
if e, err := akv.Get(ctx, "alpha.zsh"); err != nil || string(e.Value()) != `{"v":1}` {
|
||||
t.Errorf("node alpha reads %v (%v) of its own key", e, err)
|
||||
}
|
||||
|
||||
reader, readerRefused := connect("alpha.mesh-delivery-view")
|
||||
for _, other := range []struct {
|
||||
name string
|
||||
nc *nats.Conn
|
||||
ch chan error
|
||||
key string
|
||||
}{{"node alpha putting beta.zsh", alpha, nil, "beta.zsh"},
|
||||
{"the reading module putting alpha.zsh", reader, readerRefused, "alpha.zsh"}} {
|
||||
ch := other.ch
|
||||
if ch == nil {
|
||||
// alpha's own refusals arrive on its own handler: reconnect with one that is watched.
|
||||
var nc *nats.Conn
|
||||
nc, ch = connect("node.alpha")
|
||||
other.nc = nc
|
||||
}
|
||||
if err := other.nc.Publish("$KV."+delivery.HaveBucket+"."+other.key, []byte(`{"v":99}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = other.nc.Flush()
|
||||
select {
|
||||
case err := <-ch:
|
||||
if !errors.Is(err, nats.ErrPermissionViolation) {
|
||||
t.Errorf("%s: the server said %v, want a permissions violation", other.name, err)
|
||||
}
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Errorf("the server did not refuse %s", other.name)
|
||||
}
|
||||
}
|
||||
|
||||
// The reader watches and sees alpha.zsh, and beta.zsh was never taken.
|
||||
rjs, err := jetstream.New(reader)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rkv, err := rjs.KeyValue(ctx, delivery.HaveBucket)
|
||||
if err != nil {
|
||||
t.Fatalf("a module reading node-engine.have cannot bind it: %v", err)
|
||||
}
|
||||
w, err := rkv.WatchAll(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("the reading module cannot watch: %v", err)
|
||||
}
|
||||
defer w.Stop()
|
||||
seen := map[string]string{}
|
||||
for e := range w.Updates() {
|
||||
if e == nil {
|
||||
break
|
||||
}
|
||||
seen[e.Key()] = string(e.Value())
|
||||
}
|
||||
if seen["alpha.zsh"] != `{"v":1}` || len(seen) != 1 {
|
||||
t.Errorf("the reading module's watch sees %v, want alpha.zsh alone", seen)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,106 @@
|
||||
package broker
|
||||
|
||||
import (
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
// 1136/AC1: the controller's start asserts node-engine_have with the history delivery keeps, and the bucket is one of
|
||||
// the controller's own, so raising the buckets does not name it undeclared (ADR 0429 §1).
|
||||
func TestTheHaveBucketIsTheControllersAndKeepsTwentyRevisions(t *testing.T) { // 1136/AC1
|
||||
cfg := HaveBucketConfig()
|
||||
if cfg.Bucket != delivery.HaveBucket || cfg.History != delivery.HaveHistory || delivery.HaveHistory != 20 {
|
||||
t.Fatalf("the have bucket is asserted as %q with history %d, want %q with 20", cfg.Bucket, cfg.History, delivery.HaveBucket)
|
||||
}
|
||||
if cfg.Storage != jetstream.FileStorage || cfg.MaxBytes <= 0 || cfg.MaxValueSize <= 0 {
|
||||
t.Fatalf("the have bucket is not bounded on disk: %+v", cfg)
|
||||
}
|
||||
if !IsControllerBucket(delivery.HaveBucket) || !slices.Contains(ControllerBuckets(), delivery.HaveBucket) {
|
||||
t.Fatalf("%s is not among the controller's buckets", delivery.HaveBucket)
|
||||
}
|
||||
undeclared, err := RaiseBuckets(fakeBuckets{names: []string{delivery.HaveBucket}}, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(undeclared) != 0 {
|
||||
t.Fatalf("%v are raised as undeclared", undeclared)
|
||||
}
|
||||
}
|
||||
|
||||
// 1136/AC2: a node may write and read the keys of its own node of the have bucket, with what a get, create, update and
|
||||
// watch of them need, and no other key of it; the controller does not write it; an unsafe node name is refused.
|
||||
func TestANodeWritesAndReadsOnlyItsOwnHaveKeys(t *testing.T) { // 1136/AC2
|
||||
perms, err := PermissionsFor(Principal{Kind: KindNode, Node: "one", PasswordHash: "x"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b, stream := delivery.HaveBucket, "KV_"+delivery.HaveBucket
|
||||
for _, want := range []string{
|
||||
"$KV." + b + ".one.>",
|
||||
"$JS.API.STREAM.INFO." + stream,
|
||||
"$JS.API.DIRECT.GET." + stream + ".$KV." + b + ".one.>",
|
||||
"$JS.API.CONSUMER.CREATE." + stream + ".*.$KV." + b + ".one.>",
|
||||
"$JS.API.CONSUMER.DELETE." + stream + ".>",
|
||||
"$JS.FC." + stream + ".>",
|
||||
} {
|
||||
has(t, perms.Publish, want)
|
||||
}
|
||||
for _, p := range perms.Publish {
|
||||
if !strings.Contains(p, delivery.HaveBucket) {
|
||||
continue
|
||||
}
|
||||
if strings.Contains(p, "KV."+b+".>") || strings.Contains(p, "KV."+b+".two") || strings.HasSuffix(p, "."+b+".*") ||
|
||||
strings.Contains(p, "DIRECT.GET."+stream+".>") || strings.Contains(p, "CONSUMER.CREATE."+stream+".>") {
|
||||
t.Errorf("the node is granted %q, which reaches keys of other nodes", p)
|
||||
}
|
||||
}
|
||||
if err := CheckWriters(Principal{Kind: KindNode, Node: "one"}, perms.Publish); err != nil {
|
||||
t.Fatalf("the node's own have keys are refused as a second writer: %v", err)
|
||||
}
|
||||
controller, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, p := range controller.Publish {
|
||||
if SubjectsOverlap(p, "$KV."+b+".one.zsh") {
|
||||
t.Errorf("the controller may write the have keys through %q", p)
|
||||
}
|
||||
}
|
||||
for _, bad := range []string{"on.e", "one.>", "*", "o>ne"} {
|
||||
if _, err := PermissionsFor(Principal{Kind: KindNode, Node: bad, PasswordHash: "x"}); err == nil {
|
||||
t.Errorf("node %q was accepted as part of a have subject", bad)
|
||||
}
|
||||
}
|
||||
// A principal that is not a node writes no have key: the writers table refuses it.
|
||||
if err := CheckWriters(Principal{Kind: KindModule, Module: "billing", Node: "one"}, []string{"$KV." + b + ".one.>"}); err == nil {
|
||||
t.Error("a module granted the have keys of a node is not refused as a second writer")
|
||||
}
|
||||
if err := CheckWriters(Principal{Kind: KindNode, Node: "two"}, []string{"$KV." + b + ".one.>"}); err == nil {
|
||||
t.Error("a node granted another node's have keys is not refused as a second writer")
|
||||
}
|
||||
}
|
||||
|
||||
// 1136/AC3: a module whose manifest reads node-engine.have, mesh-controller.assignments and mesh-controller.builds may
|
||||
// watch and get every key of the three and write none.
|
||||
func TestAModuleReadingHaveAssignmentsAndBuildsReadsAndWritesNone(t *testing.T) { // 1136/AC3
|
||||
perms, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "mesh-delivery-view", PasswordHash: "x",
|
||||
Reads: []string{"node-engine.have", "mesh-controller.assignments", "mesh-controller.builds"}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, b := range []string{delivery.HaveBucket, AssignmentsBucket, BuildsBucket} {
|
||||
stream := "KV_" + b
|
||||
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."+b+".any.key") {
|
||||
t.Errorf("the reading module may write %s through %q", b, p)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+55
-2
@@ -21,6 +21,7 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"github.com/novox/mesh-controller/internal/lease"
|
||||
)
|
||||
|
||||
@@ -712,6 +713,9 @@ func PermissionsFor(p Principal) (Permissions, error) {
|
||||
// hand, which it alone writes. A put is a publish to the bucket's subject, which `$JS.API.>`
|
||||
// does not cover; each bucket named, not `$KV.>`, which would let it write any module's state.
|
||||
for _, bucket := range ControllerBuckets() {
|
||||
if bucket == delivery.HaveBucket {
|
||||
continue // each node's node-engine writes its own keys of it, the controller none (ADR 0429 §1)
|
||||
}
|
||||
pub = append(pub, "$KV."+bucket+".>")
|
||||
}
|
||||
// **And what the bus says about itself, read-only** (novox/hq to-be 45 §3, S9): a durable
|
||||
@@ -800,6 +804,13 @@ func PermissionsFor(p Principal) (Permissions, error) {
|
||||
// contract is lease/witness.go): it asks its own machine's node tools PING, and, where the
|
||||
// machine runs the controller, reads the lease's one key — read, never written.
|
||||
pub = append(pub, WitnessSubjects(p)...)
|
||||
// And the keys of its own node in the have bucket, which it alone writes (novox/hq ADR 0429 §1, issue 1136):
|
||||
// what the node has of each module, read by the view and the watchman. Never the controller's.
|
||||
haveKeys, err := NodeKeysGrants(delivery.HaveBucket, p.Node)
|
||||
if err != nil {
|
||||
return Permissions{}, err
|
||||
}
|
||||
pub = append(pub, haveKeys...)
|
||||
// And, while the vault serves it, the vault's move-in in its own name and nobody else's (issue 676): the
|
||||
// values sealed to this machine, which only it opens, handed in sealed to the vault's key for the call.
|
||||
if p.MovesIntoTheVault {
|
||||
@@ -812,6 +823,11 @@ func PermissionsFor(p Principal) (Permissions, error) {
|
||||
}
|
||||
|
||||
case KindModule:
|
||||
// 0. The watchman alone hears who connects and disconnects, and asks who is connected (ADR 0431 §2).
|
||||
if p.Module == WatchmanModule {
|
||||
sub = append(sub, BusConnectSubject, BusDisconnectSubject)
|
||||
pub = append(pub, BusConnzSubject)
|
||||
}
|
||||
// 1. Its own namespace: it publishes its events there and serves its tools there. Nothing
|
||||
// else may publish into it, so an event's source is a fact the server enforces rather
|
||||
// than a claim in the body (design 29 §2).
|
||||
@@ -1436,7 +1452,9 @@ func ComposeAccounts(principals []Principal) (string, error) {
|
||||
// JetStream is enabled per account once accounts exist at all: with only the global block set,
|
||||
// a user in MESH is told "JetStream not enabled for account" the first time it binds a
|
||||
// consumer, which is the first thing every host does (2026-09-28).
|
||||
b.WriteString("accounts {\n MESH {\n jetstream: enabled\n users = [\n")
|
||||
b.WriteString("accounts {\n MESH {\n jetstream: enabled\n")
|
||||
b.WriteString(meshSystemImports)
|
||||
b.WriteString(" users = [\n")
|
||||
for _, p := range sorted {
|
||||
perms, err := PermissionsFor(p)
|
||||
if err != nil {
|
||||
@@ -1457,10 +1475,45 @@ func ComposeAccounts(principals []Principal) (string, error) {
|
||||
}
|
||||
b.WriteString(" } }\n")
|
||||
}
|
||||
b.WriteString(" ]\n }\n}\n")
|
||||
b.WriteString(" ]\n }\n")
|
||||
b.WriteString(systemAccount)
|
||||
b.WriteString("}\n")
|
||||
return b.String(), nil
|
||||
}
|
||||
|
||||
// WatchmanModule is the one module whose principal hears the bus's connect and disconnect events and asks its list of
|
||||
// connections (novox/hq ADR 0431 §2, issue 1147): the watchman of the repository mesh-delivery, which writes presence
|
||||
// from them. No other principal is granted the subjects below, and nothing is a wildcard over them.
|
||||
const WatchmanModule = "mesh-delivery-watchman"
|
||||
|
||||
// The bus's own events, imported into MESH from the default system account `$SYS` under subjects of MESH's own
|
||||
// (novox/hq ADR 0431 §1, issue 1147): the server says an account's connections opening and closing only to the system
|
||||
// account, and the list of an account's connections only to a request made there. `$SYS` holds no user, so nobody
|
||||
// gains its other requests and events; it exports exactly these three, to MESH alone.
|
||||
const (
|
||||
BusConnectSubject = "mesh.bus.connect"
|
||||
BusDisconnectSubject = "mesh.bus.disconnect"
|
||||
BusConnzSubject = "mesh.bus.connz"
|
||||
)
|
||||
|
||||
// meshSystemImports is what MESH imports from $SYS: the two event streams and the one service.
|
||||
const meshSystemImports = " imports = [\n" +
|
||||
" { stream: { account: \"$SYS\", subject: \"$SYS.ACCOUNT.MESH.CONNECT\" }, to: \"" + BusConnectSubject + "\" }\n" +
|
||||
" { stream: { account: \"$SYS\", subject: \"$SYS.ACCOUNT.MESH.DISCONNECT\" }, to: \"" + BusDisconnectSubject + "\" }\n" +
|
||||
" { service: { account: \"$SYS\", subject: \"$SYS.REQ.ACCOUNT.MESH.CONNZ\" }, to: \"" + BusConnzSubject + "\" }\n" +
|
||||
" ]\n"
|
||||
|
||||
// systemAccount is the account `$SYS`, the server's default system account by its own name (a config that declares an
|
||||
// account of that name makes it the system account), with no user and the three exports to MESH. Quoted, because an
|
||||
// unquoted `$` begins a variable in the server's config grammar.
|
||||
const systemAccount = " \"$SYS\" {\n" +
|
||||
" exports = [\n" +
|
||||
" { stream: \"$SYS.ACCOUNT.MESH.CONNECT\", accounts: [\"MESH\"] }\n" +
|
||||
" { stream: \"$SYS.ACCOUNT.MESH.DISCONNECT\", accounts: [\"MESH\"] }\n" +
|
||||
" { service: \"$SYS.REQ.ACCOUNT.MESH.CONNZ\", accounts: [\"MESH\"] }\n" +
|
||||
" ]\n" +
|
||||
" }\n"
|
||||
|
||||
// unique is a sorted list with each subject once. Two carried modules holding seats with the same
|
||||
// verb, or the runtime module itself carried beside the others, would otherwise write a grant twice
|
||||
// — harmless to the server, and noise in a file that is read as the mesh's authority model.
|
||||
|
||||
@@ -28,6 +28,8 @@ func TestTheComposedConfigMatchesTheGolden(t *testing.T) {
|
||||
Emits: []string{"order.placed"}, PasswordHash: "$2a$11$ssssssssssssssssssssss"},
|
||||
{Kind: KindModule, Node: "two", Module: "audit",
|
||||
Consumes: []string{"shop.order.placed"}, PasswordHash: "$2a$11$aaaaaaaaaaaaaaaaaaaaaa"},
|
||||
// The watchman alone hears the bus's connections (novox/hq ADR 0431 §2, issue 1147).
|
||||
{Kind: KindModule, Node: "one", Module: WatchmanModule, PasswordHash: "$2a$11$wwwwwwwwwwwwwwwwwwwwww"},
|
||||
// The bus's own module: the snapshot API and its inbox, nothing else (novox/hq ADR 0235).
|
||||
{Kind: KindModule, Node: "one", Module: "nats", SnapshotsTheBus: true,
|
||||
Serves: []string{"nats_streams"}, PasswordHash: "$2a$11$bbbbbbbbbbbbbbbbbbbbbb"},
|
||||
|
||||
+32
-16
@@ -65,6 +65,37 @@ func bucketOfRead(read string) (string, bool) {
|
||||
return BucketName(module, name), true
|
||||
}
|
||||
|
||||
// readGrants is what reaching a bucket's keys needs: binding to it, a direct get, an ordered consumer for listing and
|
||||
// watching and the deletion of that consumer and its flow control. key is one key's subject tail, a pattern such as
|
||||
// `one.>` for a prefix of keys, or empty for every key. The bucket is made with AllowDirect, so a get is the direct get
|
||||
// and never the stream's message get.
|
||||
func readGrants(bucket, key string) []string {
|
||||
stream := "KV_" + bucket
|
||||
out := []string{"$JS.API.STREAM.INFO." + stream}
|
||||
if key == "" {
|
||||
out = append(out,
|
||||
"$JS.API.DIRECT.GET."+stream+".>",
|
||||
"$JS.API.CONSUMER.CREATE."+stream+".>")
|
||||
} else {
|
||||
out = append(out,
|
||||
"$JS.API.DIRECT.GET."+stream+".$KV."+bucket+"."+key,
|
||||
"$JS.API.CONSUMER.CREATE."+stream+".*.$KV."+bucket+"."+key)
|
||||
}
|
||||
return append(out,
|
||||
"$JS.API.CONSUMER.DELETE."+stream+".>",
|
||||
"$JS.FC."+stream+".>")
|
||||
}
|
||||
|
||||
// NodeKeysGrants is what a node's node-engine is granted of a bucket it writes by node, `<node>.>` and no other key
|
||||
// (novox/hq ADR 0429 §1: each key one writer, enforced by credentials): the writes of those keys and what a get,
|
||||
// create, update and watch of them need. The node name is checked as every subject token is.
|
||||
func NodeKeysGrants(bucket, node string) ([]string, error) {
|
||||
if !safeSubject.MatchString(node) {
|
||||
return nil, fmt.Errorf("%q cannot be part of a subject: a permission is a subject pattern, and this would widen it", node)
|
||||
}
|
||||
return append([]string{"$KV." + bucket + "." + node + ".>"}, readGrants(bucket, node+".>")...), nil
|
||||
}
|
||||
|
||||
// stateGrants is what a principal publishes to reach the state its modules keep and read: for every
|
||||
// bucket, binding to it, reading a key directly, and an ordered consumer for listing and watching,
|
||||
// created and deleted on the bucket's own stream, with its flow control answered; for a bucket an
|
||||
@@ -84,22 +115,7 @@ func bucketOfRead(read string) (string, bool) {
|
||||
// consumer (named at random by the client) and flow control; none of them reads or writes a value.
|
||||
func stateGrants(a stateAccess) []string {
|
||||
var out []string
|
||||
read := func(bucket, key string) {
|
||||
stream := "KV_" + bucket
|
||||
out = append(out, "$JS.API.STREAM.INFO."+stream)
|
||||
if key == "" {
|
||||
out = append(out,
|
||||
"$JS.API.DIRECT.GET."+stream+".>",
|
||||
"$JS.API.CONSUMER.CREATE."+stream+".>")
|
||||
} else {
|
||||
out = append(out,
|
||||
"$JS.API.DIRECT.GET."+stream+".$KV."+bucket+"."+key,
|
||||
"$JS.API.CONSUMER.CREATE."+stream+".*.$KV."+bucket+"."+key)
|
||||
}
|
||||
out = append(out,
|
||||
"$JS.API.CONSUMER.DELETE."+stream+".>",
|
||||
"$JS.FC."+stream+".>")
|
||||
}
|
||||
read := func(bucket, key string) { out = append(out, readGrants(bucket, key)...) }
|
||||
machineKey := ""
|
||||
if safeSubject.MatchString(a.Node) {
|
||||
machineKey = a.Node
|
||||
|
||||
@@ -0,0 +1,164 @@
|
||||
//go:build integration
|
||||
|
||||
package broker
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats-server/v2/server"
|
||||
"github.com/nats-io/nats.go"
|
||||
"golang.org/x/crypto/bcrypt"
|
||||
)
|
||||
|
||||
// **The system account's exports reach MESH on a reload, with no restart** (novox/hq ADR 0431 §3, issue 1147,
|
||||
// 1147/AC3): the embedded pinned server starts with today's composed accounts (no `$SYS`, no import); the new file is
|
||||
// written and the server reloaded; the reload succeeds; a user with the watchman's grants hears another user's connect
|
||||
// and disconnect and gets an answer to `mesh.bus.connz` that lists the open connections; another module's user is
|
||||
// refused a subscribe on `mesh.bus.connect`.
|
||||
func TestNatsTheSystemAccountsExportsReachMeshOnAReload(t *testing.T) { // 1147/AC3
|
||||
hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
users, err := Users(Records{Nodes: []string{"one"}, Assigned: map[string][]Declared{
|
||||
"one": {{Module: RuntimeModule}, {Module: WatchmanModule}, {Module: "gitea"}, {Module: "shop"}},
|
||||
}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for i := range users {
|
||||
users[i].PasswordHash = string(hash)
|
||||
}
|
||||
accounts, err := ComposeAccounts(users)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Today's file: the same accounts without the system account's exports and MESH's imports.
|
||||
today := strings.Replace(strings.Replace(accounts, meshSystemImports, "", 1), systemAccount, "", 1)
|
||||
if today == accounts || strings.Contains(today, "$SYS") {
|
||||
t.Fatal("today's composed file could not be derived from the new one")
|
||||
}
|
||||
dir := t.TempDir()
|
||||
conf := filepath.Join(dir, "bus.conf")
|
||||
write := func(accounts string) {
|
||||
t.Helper()
|
||||
if err := os.WriteFile(conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: "+`"`+dir+`/js"`+" }\n"+accounts), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
write(today)
|
||||
opts, err := server.ProcessConfigFile(conf)
|
||||
if err != nil {
|
||||
t.Fatalf("today's accounts do not parse: %v", err)
|
||||
}
|
||||
opts.NoLog, opts.NoSigs = true, true
|
||||
s, err := server.NewServer(opts)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
go s.Start()
|
||||
if !s.ReadyForConnections(10 * time.Second) {
|
||||
t.Fatal("the bus did not come up")
|
||||
}
|
||||
t.Cleanup(s.Shutdown)
|
||||
|
||||
connect := func(user string) (*nats.Conn, chan error) {
|
||||
t.Helper()
|
||||
refused := make(chan error, 8)
|
||||
nc, err := nats.Connect(s.ClientURL(), nats.UserInfo(user, "pw"), nats.CustomInboxPrefix("_INBOX."+user),
|
||||
nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refused <- err }))
|
||||
if err != nil {
|
||||
t.Fatalf("%s cannot connect: %v", user, err)
|
||||
}
|
||||
t.Cleanup(nc.Close)
|
||||
return nc, refused
|
||||
}
|
||||
// A connection open across the reload: the reload must not drop it.
|
||||
shop, _ := connect("one.shop")
|
||||
|
||||
write(accounts)
|
||||
if err := s.Reload(); err != nil {
|
||||
t.Fatalf("the server does not take the system account's exports on a reload: %v", err)
|
||||
}
|
||||
if !shop.IsConnected() {
|
||||
t.Error("the reload dropped a connection that was open")
|
||||
}
|
||||
|
||||
watchman, _ := connect("one." + WatchmanModule)
|
||||
connects := make(chan *nats.Msg, 16)
|
||||
disconnects := make(chan *nats.Msg, 16)
|
||||
if _, err := watchman.ChanSubscribe(BusConnectSubject, connects); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := watchman.ChanSubscribe(BusDisconnectSubject, disconnects); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := watchman.Flush(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Another user connects and disconnects: the watchman hears both, and the events name that user.
|
||||
gitea, _ := connect("one.gitea")
|
||||
hears := func(ch chan *nats.Msg, what string) {
|
||||
t.Helper()
|
||||
select {
|
||||
case m := <-ch:
|
||||
if !strings.Contains(string(m.Data), "one.gitea") {
|
||||
t.Errorf("the %s event does not name the user that connected: %s", what, m.Data)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Errorf("the watchman did not hear another user's %s", what)
|
||||
}
|
||||
}
|
||||
hears(connects, "connect")
|
||||
// The answer to the connections request lists the open connections, the shop's among them.
|
||||
reply, err := watchman.Request(BusConnzSubject, []byte(`{}`), 5*time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("mesh.bus.connz is not answered: %v", err)
|
||||
}
|
||||
var answer struct {
|
||||
Data struct {
|
||||
Connections []struct {
|
||||
AuthorizedUser string `json:"authorized_user"`
|
||||
} `json:"connections"`
|
||||
} `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(reply.Data, &answer); err != nil {
|
||||
t.Fatalf("the connections answer is not JSON: %v: %s", err, reply.Data)
|
||||
}
|
||||
var listed bool
|
||||
for _, c := range answer.Data.Connections {
|
||||
listed = listed || c.AuthorizedUser == "one.shop"
|
||||
}
|
||||
if !listed {
|
||||
t.Errorf("the connections answer does not list the shop's connection: %s", reply.Data)
|
||||
}
|
||||
gitea.Close()
|
||||
hears(disconnects, "disconnect")
|
||||
|
||||
// Another module's user is refused a subscribe on the events and a publish on the request.
|
||||
other, refused := connect("one.shop")
|
||||
for _, act := range []func() error{
|
||||
func() error { _, err := other.SubscribeSync(BusConnectSubject); return err },
|
||||
func() error { _, err := other.SubscribeSync(BusDisconnectSubject); return err },
|
||||
func() error { return other.Publish(BusConnzSubject, []byte(`{}`)) },
|
||||
} {
|
||||
if err := act(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = other.Flush()
|
||||
select {
|
||||
case err := <-refused:
|
||||
if !errors.Is(err, nats.ErrPermissionViolation) {
|
||||
t.Errorf("the server said %v, want a permissions violation", err)
|
||||
}
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Error("the server did not refuse another module's user the bus's connection subjects")
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,145 @@
|
||||
package broker
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/nats-io/nats-server/v2/server"
|
||||
)
|
||||
|
||||
// 1147/AC1: the composed accounts file declares $SYS with no user, exporting only the connect and disconnect events of
|
||||
// MESH and the request listing MESH's connections, and MESH imports exactly those three as mesh.bus.connect,
|
||||
// mesh.bus.disconnect and mesh.bus.connz (novox/hq ADR 0431 §1); the server's own grammar takes it with $SYS as its
|
||||
// system account; the golden file shows it (TestTheComposedConfigMatchesTheGolden).
|
||||
func TestTheComposedAccountsExportMeshsConnectionEventsFromTheSystemAccount(t *testing.T) { // 1147/AC1
|
||||
text, err := ComposeAccounts([]Principal{{Kind: KindNode, Node: "one", PasswordHash: "x"}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sys := between(text, `"$SYS" {`, "\n }\n")
|
||||
if sys == "" {
|
||||
t.Fatalf("no account $SYS in the composed accounts:\n%s", text)
|
||||
}
|
||||
if strings.Contains(sys, "users") || strings.Contains(sys, "user:") {
|
||||
t.Errorf("$SYS holds a user:\n%s", sys)
|
||||
}
|
||||
for _, want := range []string{
|
||||
`{ stream: "$SYS.ACCOUNT.MESH.CONNECT", accounts: ["MESH"] }`,
|
||||
`{ stream: "$SYS.ACCOUNT.MESH.DISCONNECT", accounts: ["MESH"] }`,
|
||||
`{ service: "$SYS.REQ.ACCOUNT.MESH.CONNZ", accounts: ["MESH"] }`,
|
||||
} {
|
||||
if !strings.Contains(sys, want) {
|
||||
t.Errorf("$SYS does not export %s:\n%s", want, sys)
|
||||
}
|
||||
}
|
||||
if n := strings.Count(sys, "{ stream:") + strings.Count(sys, "{ service:"); n != 3 {
|
||||
t.Errorf("$SYS exports %d things, want exactly 3:\n%s", n, sys)
|
||||
}
|
||||
mesh := between(text, " MESH {", " users = [")
|
||||
for _, want := range []string{
|
||||
`{ stream: { account: "$SYS", subject: "$SYS.ACCOUNT.MESH.CONNECT" }, to: "mesh.bus.connect" }`,
|
||||
`{ stream: { account: "$SYS", subject: "$SYS.ACCOUNT.MESH.DISCONNECT" }, to: "mesh.bus.disconnect" }`,
|
||||
`{ service: { account: "$SYS", subject: "$SYS.REQ.ACCOUNT.MESH.CONNZ" }, to: "mesh.bus.connz" }`,
|
||||
} {
|
||||
if !strings.Contains(mesh, want) {
|
||||
t.Errorf("MESH does not import %s:\n%s", want, mesh)
|
||||
}
|
||||
}
|
||||
if n := strings.Count(mesh, "{ stream:") + strings.Count(mesh, "{ service:"); n != 3 {
|
||||
t.Errorf("MESH imports %d things, want exactly 3:\n%s", n, mesh)
|
||||
}
|
||||
|
||||
// The pinned server's grammar takes it, and makes $SYS its system account.
|
||||
path := filepath.Join(t.TempDir(), "accounts.conf")
|
||||
if err := os.WriteFile(path, []byte(text), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
opts, err := server.ProcessConfigFile(path)
|
||||
if err != nil {
|
||||
t.Fatalf("the server refuses the composed accounts: %v", err)
|
||||
}
|
||||
var found bool
|
||||
for _, a := range opts.Accounts {
|
||||
if a.Name != "$SYS" {
|
||||
continue
|
||||
}
|
||||
found = true
|
||||
if !a.IsExportService("$SYS.REQ.ACCOUNT.MESH.CONNZ") {
|
||||
t.Error("$SYS does not export the connections request as a service")
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Error("the server read no account $SYS")
|
||||
}
|
||||
for _, u := range opts.Users {
|
||||
if u.Account != nil && u.Account.Name == "$SYS" {
|
||||
t.Errorf("user %s is in $SYS", u.Username)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// between is the text from the first start to the first end after it, or empty.
|
||||
func between(s, start, end string) string {
|
||||
i := strings.Index(s, start)
|
||||
if i < 0 {
|
||||
return ""
|
||||
}
|
||||
rest := s[i:]
|
||||
j := strings.Index(rest[len(start):], end)
|
||||
if j < 0 {
|
||||
return rest
|
||||
}
|
||||
return rest[:len(start)+j]
|
||||
}
|
||||
|
||||
// 1147/AC2: only mesh-delivery-watchman's principal may subscribe to mesh.bus.connect and mesh.bus.disconnect and
|
||||
// publish to mesh.bus.connz; no other principal's grant covers them by a wildcard (ADR 0431 §2).
|
||||
func TestOnlyTheWatchmanHearsTheBusConnectionEvents(t *testing.T) { // 1147/AC2
|
||||
builder := Seat{Name: "node-build-agent", Scope: "node", Accepts: []string{"build"}, Emits: []string{"built"}}
|
||||
watchman := Principal{Kind: KindModule, Node: "one", Module: "mesh-delivery-watchman", PasswordHash: "x"}
|
||||
others := []Principal{
|
||||
{Kind: KindController, PasswordHash: "x"},
|
||||
{Kind: KindNode, Node: "one", PasswordHash: "x"},
|
||||
{Kind: KindEnrolment, Node: "two", PasswordHash: "x"},
|
||||
{Kind: KindPerson, Module: "jochen", Invokes: []string{"*"}, PasswordHash: "x"},
|
||||
{Kind: KindProvisioner, Node: "one", PasswordHash: "x"},
|
||||
{Kind: KindModule, Node: "one", Module: "gitea", Emits: []string{"pull.merged"}, PasswordHash: "x"},
|
||||
{Kind: KindModule, Node: "one", Module: "mesh-delivery-view", PasswordHash: "x",
|
||||
Reads: []string{"node-engine.have"}, Consumes: []string{"mesh-delivery-watchman.presence"}},
|
||||
{Kind: KindModule, Node: "one", Module: "build-agent", Holds: []Seat{builder}, PasswordHash: "x"},
|
||||
{Kind: KindModule, Node: "one", Module: "nats", SnapshotsTheBus: true, PasswordHash: "x"},
|
||||
{Kind: KindNodeTools, Node: "one", Module: RuntimeModule, PasswordHash: "x", Carries: []Declared{
|
||||
{Module: "mesh-delivery-watchman"}, {Module: "build-agent", Holds: []Seat{builder}}}},
|
||||
}
|
||||
heard := []string{"mesh.bus.connect", "mesh.bus.disconnect"}
|
||||
perms, err := PermissionsFor(watchman)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, s := range heard {
|
||||
has(t, perms.Subscribe, s)
|
||||
}
|
||||
has(t, perms.Publish, "mesh.bus.connz")
|
||||
for _, s := range append(heard, "mesh.bus.connz") {
|
||||
for _, p := range perms.Subscribe {
|
||||
if p != s && SubjectsOverlap(p, s) && !strings.HasPrefix(p, "_INBOX.") {
|
||||
t.Errorf("the watchman's subscription %q reaches %s beyond naming it", p, s)
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, p := range others {
|
||||
perms, err := PermissionsFor(p)
|
||||
if err != nil {
|
||||
t.Fatalf("%s: %v", p.Username(), err)
|
||||
}
|
||||
for _, s := range append(heard, "mesh.bus.connz") {
|
||||
for _, g := range append(append([]string(nil), perms.Publish...), perms.Subscribe...) {
|
||||
if SubjectsOverlap(g, s) {
|
||||
t.Errorf("%s is granted %q, which covers %s", p.Username(), g, s)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+18
-1
@@ -22,6 +22,11 @@ jetstream {
|
||||
accounts {
|
||||
MESH {
|
||||
jetstream: enabled
|
||||
imports = [
|
||||
{ stream: { account: "$SYS", subject: "$SYS.ACCOUNT.MESH.CONNECT" }, to: "mesh.bus.connect" }
|
||||
{ stream: { account: "$SYS", subject: "$SYS.ACCOUNT.MESH.DISCONNECT" }, to: "mesh.bus.disconnect" }
|
||||
{ service: { account: "$SYS", subject: "$SYS.REQ.ACCOUNT.MESH.CONNZ" }, to: "mesh.bus.connz" }
|
||||
]
|
||||
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_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-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"] }
|
||||
@@ -33,10 +38,15 @@ accounts {
|
||||
subscribe: { allow: ["_INBOX.enrol.one.>"] }
|
||||
} }
|
||||
{ user: "node.one", password: "$2a$11$nnnnnnnnnnnnnnnnnnnnnn", permissions: {
|
||||
publish: { allow: ["$JS.ACK.NODES.one.>", "$JS.API.CONSUMER.INFO.NODES.one", "$SRV.PING.node-tools.one", "mesh.control.one.>"] }
|
||||
publish: { allow: ["$JS.ACK.NODES.one.>", "$JS.API.CONSUMER.CREATE.KV_node-engine_have.*.$KV.node-engine_have.one.>", "$JS.API.CONSUMER.DELETE.KV_node-engine_have.>", "$JS.API.CONSUMER.INFO.NODES.one", "$JS.API.DIRECT.GET.KV_node-engine_have.$KV.node-engine_have.one.>", "$JS.API.STREAM.INFO.KV_node-engine_have", "$JS.FC.KV_node-engine_have.>", "$KV.node-engine_have.one.>", "$SRV.PING.node-tools.one", "mesh.control.one.>"] }
|
||||
subscribe: { allow: ["_DELIVER.one", "_DELIVER.one.>", "_INBOX.node.one.>", "mesh.node.one.ask.hand-over", "mesh.node.one.ask.judge-directories", "mesh.node.one.ask.move-secret", "mesh.node.one.ask.report", "mesh.node.one.ask.setuid-search", "mesh.node.one.declare"] }
|
||||
allow_responses: { max: 1, ttl: "1m" }
|
||||
} }
|
||||
{ user: "one.mesh-delivery-watchman", password: "$2a$11$wwwwwwwwwwwwwwwwwwwwww", permissions: {
|
||||
publish: { allow: ["$JS.ACK.EVENTS.one_mesh-delivery-watchman.>", "$JS.API.CONSUMER.INFO.EVENTS.one_mesh-delivery-watchman", "$JS.API.CONSUMER.MSG.NEXT.EVENTS.one_mesh-delivery-watchman", "$JS.API.DIRECT.GET.ASSIGNMENTS.mesh.assignment.one.mesh-delivery-watchman", "mesh.bus.connz"] }
|
||||
subscribe: { allow: ["$SRV.INFO", "$SRV.INFO.mesh-delivery-watchman", "$SRV.INFO.mesh-delivery-watchman.>", "$SRV.PING", "$SRV.PING.mesh-delivery-watchman", "$SRV.PING.mesh-delivery-watchman.>", "$SRV.STATS", "$SRV.STATS.mesh-delivery-watchman", "$SRV.STATS.mesh-delivery-watchman.>", "_INBOX.one.mesh-delivery-watchman.>", "mesh.assignment.one.mesh-delivery-watchman", "mesh.bus.connect", "mesh.bus.disconnect", "mesh.mod.mesh-delivery-watchman.call.>", "mesh.mod.mesh-delivery-watchman.tool.>"] }
|
||||
allow_responses: { max: 1, ttl: "1m" }
|
||||
} }
|
||||
{ user: "one.nats", password: "$2a$11$bbbbbbbbbbbbbbbbbbbbbb", permissions: {
|
||||
publish: { allow: ["$JS.API.STREAM.INFO.*", "$JS.API.STREAM.NAMES", "$JS.API.STREAM.SNAPSHOT.*", "$JS.SNAPSHOT.ACK.>"] }
|
||||
subscribe: { allow: ["_INBOX.one.nats.>"] }
|
||||
@@ -58,4 +68,11 @@ accounts {
|
||||
} }
|
||||
]
|
||||
}
|
||||
"$SYS" {
|
||||
exports = [
|
||||
{ stream: "$SYS.ACCOUNT.MESH.CONNECT", accounts: ["MESH"] }
|
||||
{ stream: "$SYS.ACCOUNT.MESH.DISCONNECT", accounts: ["MESH"] }
|
||||
{ service: "$SYS.REQ.ACCOUNT.MESH.CONNZ", accounts: ["MESH"] }
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,6 +4,8 @@ import (
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// The writers table (novox/hq to-be 45 §1, ADR 0227 rule 1), compiled in.
|
||||
@@ -92,6 +94,13 @@ func ownProvisioner(p Principal, pattern string) bool {
|
||||
return p.Kind == KindProvisioner && len(tokens) > 3 && tokens[2] == p.Node
|
||||
}
|
||||
|
||||
// ownNodeKeys is a node writing the have keys of its own node: `$KV.<bucket>.<node>.…`, the node in the third token
|
||||
// and no wildcard there (novox/hq ADR 0429 §1).
|
||||
func ownNodeKeys(p Principal, pattern string) bool {
|
||||
tokens := strings.Split(pattern, ".")
|
||||
return p.Kind == KindNode && len(tokens) > 2 && tokens[2] == p.Node
|
||||
}
|
||||
|
||||
// kvOf is a bucket's write subjects.
|
||||
func kvOf(bucket string) []string { return []string{"$KV." + bucket + ".>"} }
|
||||
|
||||
@@ -205,6 +214,11 @@ var WritersTable = []WriterRow{
|
||||
{State: "the rollouts of the chain", Writer: "controller (lease holder), compare-and-set on the key's revision",
|
||||
KeptIn: "key-value " + RolloutsBucket + ", sixty-four revisions of each key, none removed",
|
||||
Others: "read through `reads` as mesh-controller." + ChainRollouts, Subjects: kvOf(RolloutsBucket), Writes: isController},
|
||||
// What each machine has of each module (novox/hq ADR 0429 §1, design 47): written by that machine's node-engine under
|
||||
// its own node's keys, never by the controller; the view and the watchman read it.
|
||||
{State: "a machine's have", Writer: "the node-engine of the machine, under its own keys `<node>.>`",
|
||||
KeptIn: "key-value " + delivery.HaveBucket + ", twenty revisions of each key",
|
||||
Others: "read through `reads` as node-engine.have", Subjects: kvOf(delivery.HaveBucket), Writes: ownNodeKeys},
|
||||
{State: "the declared-config mirror", Writer: "controller (lease holder), from its store, only on a change",
|
||||
KeptIn: "key-value " + DeclaredConfigBucket, Others: "read through `reads` as mesh-controller." + MirrorDeclaredConfig,
|
||||
Subjects: kvOf(DeclaredConfigBucket), Writes: isController},
|
||||
|
||||
@@ -41,6 +41,8 @@ var designRows = []string{
|
||||
// The chain beside the assignments, the record itself and no mirror of one (ADR 0408, design 47, issue 1089).
|
||||
"the builds of the chain",
|
||||
"the rollouts of the chain",
|
||||
// What each machine has of each module, written by its node-engine under its own keys (ADR 0429 §1, issue 1136).
|
||||
"a machine's have",
|
||||
"the declared-config mirror",
|
||||
"the predefined-keys mirror",
|
||||
// The provisioner client's word on what it placed (novox/hq ADR 0349 §4, design 57 P6).
|
||||
|
||||
@@ -0,0 +1,378 @@
|
||||
// Package delivery is the one home of the keys and values the parts of delivery read and write on the bus
|
||||
// (novox/hq ADR 0429 §2: the key layouts another repository reads live in mesh-sdk; research 063 section 01). The
|
||||
// node-engine writes the have keys, the watchman of the repository mesh-delivery writes presence and pulse, the
|
||||
// view reads all of them, and the controller grants them; every one imports this package, and nothing else defines
|
||||
// these names.
|
||||
//
|
||||
// **Buckets** are named `<owner module>_<name>` as the broker names a module's state (provider.BucketOf): the
|
||||
// node-engine's `have`, and the watchman's `presence` and `pulse`.
|
||||
//
|
||||
// **Keys.** A have key is `<node>.<module>`, and a node's own facts are `<node>.node`, so a module may not be named
|
||||
// `node`. A presence key is `<node>` for a node and `part.<part>.<instance>` for a part's instance, so a node may
|
||||
// not be named `part`. A pulse key is `<bucket>.<key>`: the full name of the bucket watched, then that bucket's key,
|
||||
// which may hold dots, so it splits at the first dot. A name is refused when it is empty or holds a dot, a wildcard,
|
||||
// whitespace or a character a bus key refuses, and the error names it.
|
||||
//
|
||||
// **Values** are JSON of under about 500 bytes, each carrying its version in `v`. A reader refuses a value whose
|
||||
// version is missing or one it does not know, with an error that is ErrUnknownVersion (research 063 section 05
|
||||
// question 9 item 7): a writer writes a new version only once every reader's have shows a version that reads it. A
|
||||
// writer writes only when the new value differs from the one before (the Same functions).
|
||||
//
|
||||
// Standard library only.
|
||||
package delivery
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The buckets of delivery's parts. Each is written by one owner: every node's node-engine writes the have keys of
|
||||
// its own node only, and the watchman writes presence and pulse.
|
||||
const (
|
||||
HaveBucket = "node-engine_have"
|
||||
PresenceBucket = "mesh-delivery-watchman_presence"
|
||||
PulseBucket = "mesh-delivery-watchman_pulse"
|
||||
)
|
||||
|
||||
// HaveHistory and PresenceHistory are the revisions the have and presence buckets keep per key (research 063
|
||||
// section 01).
|
||||
const (
|
||||
HaveHistory = 20
|
||||
PresenceHistory = 20
|
||||
)
|
||||
|
||||
// PulseMoveBound is how long a move may take: a pulse key lives for this time minus the move's age, and the bucket
|
||||
// expires it on its own (the server writes a delete marker), so a move that overruns it is seen to stall.
|
||||
const PulseMoveBound = 120 * time.Second
|
||||
|
||||
// Version is the version of every value this package writes and the only one it reads.
|
||||
const Version = 1
|
||||
|
||||
// ErrUnknownVersion is what a decode returns for a value whose version is missing or not Version. Test it with
|
||||
// errors.Is; the message names the version seen and the one read.
|
||||
var ErrUnknownVersion = errors.New("delivery: unknown value version")
|
||||
|
||||
// Phase is how far a module's change has come on a node.
|
||||
type Phase string
|
||||
|
||||
// The phases of a module's change on a node.
|
||||
const (
|
||||
PhaseBackingUp Phase = "backing-up"
|
||||
PhaseApplying Phase = "applying"
|
||||
PhaseChecking Phase = "checking"
|
||||
PhaseRunning Phase = "running"
|
||||
PhaseFailed Phase = "failed"
|
||||
)
|
||||
|
||||
// Health is what the module's own check says.
|
||||
type Health string
|
||||
|
||||
// The healths a have carries.
|
||||
const (
|
||||
HealthOK Health = "ok"
|
||||
HealthFailed Health = "failed"
|
||||
HealthWaits Health = "waits"
|
||||
HealthUnknown Health = "unknown"
|
||||
)
|
||||
|
||||
var nameOK = regexp.MustCompile(`^[-/_=a-zA-Z0-9]+$`)
|
||||
|
||||
// checkName refuses a name that is empty or holds a dot, a wildcard, whitespace or a character a bus key refuses.
|
||||
func checkName(kind, name string) error {
|
||||
if name == "" {
|
||||
return fmt.Errorf("delivery: the %s name is empty", kind)
|
||||
}
|
||||
if !nameOK.MatchString(name) {
|
||||
return fmt.Errorf("delivery: the %s name %q holds a dot, a wildcard, whitespace or a character a key refuses", kind, name)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// HaveKey is the key, in HaveBucket, of a module's have on a node: `<node>.<module>`. A module named `node` is
|
||||
// refused, since `<node>.node` is the node's own facts.
|
||||
func HaveKey(node, module string) (string, error) {
|
||||
if err := checkName("node", node); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if err := checkName("module", module); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if module == "node" {
|
||||
return "", fmt.Errorf(`delivery: the module name "node" is refused: %s.node is the node's own key`, node)
|
||||
}
|
||||
return node + "." + module, nil
|
||||
}
|
||||
|
||||
// HaveNodeKey is the key, in HaveBucket, of a node's own facts: `<node>.node`.
|
||||
func HaveNodeKey(node string) (string, error) {
|
||||
if err := checkName("node", node); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return node + ".node", nil
|
||||
}
|
||||
|
||||
// ParseHaveKey is the inverse of HaveKey and HaveNodeKey: isNode is true for `<node>.node`, with module empty.
|
||||
func ParseHaveKey(key string) (node, module string, isNode bool, err error) {
|
||||
node, module, ok := strings.Cut(key, ".")
|
||||
if !ok {
|
||||
return "", "", false, fmt.Errorf("delivery: %q is not a have key, <node>.<module>", key)
|
||||
}
|
||||
if err := checkName("node", node); err != nil {
|
||||
return "", "", false, fmt.Errorf("delivery: %q is not a have key: %w", key, err)
|
||||
}
|
||||
if err := checkName("module", module); err != nil {
|
||||
return "", "", false, fmt.Errorf("delivery: %q is not a have key: %w", key, err)
|
||||
}
|
||||
if module == "node" {
|
||||
return node, "", true, nil
|
||||
}
|
||||
return node, module, false, nil
|
||||
}
|
||||
|
||||
// PresenceNodeKey is the key, in PresenceBucket, of a node: `<node>`. A node named `part` is refused.
|
||||
func PresenceNodeKey(node string) (string, error) {
|
||||
if err := checkName("node", node); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if node == "part" {
|
||||
return "", errors.New(`delivery: the node name "part" is refused: part.<part>.<instance> is a part's key`)
|
||||
}
|
||||
return node, nil
|
||||
}
|
||||
|
||||
// PresencePartKey is the key, in PresenceBucket, of a part's instance: `part.<part>.<instance>`.
|
||||
func PresencePartKey(part, instance string) (string, error) {
|
||||
if err := checkName("part", part); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if err := checkName("instance", instance); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return "part." + part + "." + instance, nil
|
||||
}
|
||||
|
||||
// ParsePresenceKey is the inverse of PresenceNodeKey and PresencePartKey: for a node's key part and instance are
|
||||
// empty, for a part's key node is.
|
||||
func ParsePresenceKey(key string) (node, part, instance string, err error) {
|
||||
if !strings.Contains(key, ".") {
|
||||
node, err = PresenceNodeKey(key)
|
||||
if err != nil {
|
||||
return "", "", "", fmt.Errorf("delivery: %q is not a presence key: %w", key, err)
|
||||
}
|
||||
return node, "", "", nil
|
||||
}
|
||||
fields := strings.Split(key, ".")
|
||||
if len(fields) != 3 || fields[0] != "part" {
|
||||
return "", "", "", fmt.Errorf("delivery: %q is not a presence key, <node> or part.<part>.<instance>", key)
|
||||
}
|
||||
if _, err := PresencePartKey(fields[1], fields[2]); err != nil {
|
||||
return "", "", "", fmt.Errorf("delivery: %q is not a presence key: %w", key, err)
|
||||
}
|
||||
return "", fields[1], fields[2], nil
|
||||
}
|
||||
|
||||
// PulseKey is the key, in PulseBucket, of the move watched at key of bucket: `<bucket>.<key>`. The bucket is the
|
||||
// full bucket name and holds no dot; the key may.
|
||||
func PulseKey(bucket, key string) (string, error) {
|
||||
if err := checkName("bucket", bucket); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if key == "" || strings.HasPrefix(key, ".") || strings.HasSuffix(key, ".") || strings.Contains(key, "..") {
|
||||
return "", fmt.Errorf("delivery: the pulse key %q is empty or has an empty part", key)
|
||||
}
|
||||
for _, part := range strings.Split(key, ".") {
|
||||
if err := checkName("key", part); err != nil {
|
||||
return "", err
|
||||
}
|
||||
}
|
||||
return bucket + "." + key, nil
|
||||
}
|
||||
|
||||
// ParsePulseKey is the inverse of PulseKey, splitting at the first dot.
|
||||
func ParsePulseKey(pulseKey string) (bucket, key string, err error) {
|
||||
bucket, key, ok := strings.Cut(pulseKey, ".")
|
||||
if !ok {
|
||||
return "", "", fmt.Errorf("delivery: %q is not a pulse key, <bucket>.<key>", pulseKey)
|
||||
}
|
||||
if _, err := PulseKey(bucket, key); err != nil {
|
||||
return "", "", fmt.Errorf("delivery: %q is not a pulse key: %w", pulseKey, err)
|
||||
}
|
||||
return bucket, key, nil
|
||||
}
|
||||
|
||||
// PulseTTL is the time to live of a pulse key for a move of the given age: PulseMoveBound minus the age. At or past
|
||||
// the bound the move is stalled, with no time left and nothing to write: the watchman raises it at once.
|
||||
func PulseTTL(moveAge time.Duration) (ttl time.Duration, stalled bool) {
|
||||
if moveAge >= PulseMoveBound {
|
||||
return 0, true
|
||||
}
|
||||
return PulseMoveBound - moveAge, false
|
||||
}
|
||||
|
||||
// Have is a module's state on a node, written by that node's node-engine under HaveKey. FP is the module's
|
||||
// fingerprint once the indexer exists (design 47 release 3); until then Build is the build the node was sent, and
|
||||
// the view compares by build.
|
||||
type Have struct {
|
||||
V int `json:"v"`
|
||||
FP string `json:"fp,omitempty"`
|
||||
Build string `json:"build,omitempty"`
|
||||
Phase Phase `json:"phase"`
|
||||
Health Health `json:"health"`
|
||||
Detail string `json:"detail,omitempty"`
|
||||
}
|
||||
|
||||
// HaveNode is a node's own facts, written under HaveNodeKey. Each field is the node-engine's own JSON, carried whole:
|
||||
// this package names the facts and does not type them.
|
||||
type HaveNode struct {
|
||||
V int `json:"v"`
|
||||
Windows json.RawMessage `json:"windows,omitempty"`
|
||||
Busy json.RawMessage `json:"busy,omitempty"`
|
||||
Strays json.RawMessage `json:"strays,omitempty"`
|
||||
Filters json.RawMessage `json:"filters,omitempty"`
|
||||
Profile json.RawMessage `json:"profile,omitempty"`
|
||||
}
|
||||
|
||||
// Presence is whether a node or a part's instance is online; the entry's own time says since when. The watchman
|
||||
// writes it.
|
||||
type Presence struct {
|
||||
V int `json:"v"`
|
||||
Online bool `json:"online"`
|
||||
}
|
||||
|
||||
// Pulse is the value of a pulse key: the watched key mirrored, its revision, and who owes the next move (a node or
|
||||
// a part instance). The watchman writes it with the time to live PulseTTL gives.
|
||||
type Pulse struct {
|
||||
V int `json:"v"`
|
||||
Bucket string `json:"bucket"`
|
||||
Key string `json:"key"`
|
||||
Revision uint64 `json:"revision"`
|
||||
Owes string `json:"owes,omitempty"`
|
||||
}
|
||||
|
||||
func checkVersion(data []byte) error {
|
||||
var probe struct {
|
||||
V *int `json:"v"`
|
||||
}
|
||||
if err := json.Unmarshal(data, &probe); err != nil {
|
||||
return fmt.Errorf("delivery: value is not JSON: %w", err)
|
||||
}
|
||||
if probe.V == nil {
|
||||
return fmt.Errorf("%w: no version in the value, this reader reads %d", ErrUnknownVersion, Version)
|
||||
}
|
||||
if *probe.V != Version {
|
||||
return fmt.Errorf("%w: the value is version %d, this reader reads %d", ErrUnknownVersion, *probe.V, Version)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func decode(data []byte, into any) error {
|
||||
if err := checkVersion(data); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := json.Unmarshal(data, into); err != nil {
|
||||
return fmt.Errorf("delivery: value does not fit its shape: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h Have) check() error {
|
||||
switch h.Phase {
|
||||
case PhaseBackingUp, PhaseApplying, PhaseChecking, PhaseRunning, PhaseFailed:
|
||||
default:
|
||||
return fmt.Errorf("delivery: phase %q is not backing-up, applying, checking, running or failed", h.Phase)
|
||||
}
|
||||
switch h.Health {
|
||||
case HealthOK, HealthFailed, HealthWaits, HealthUnknown:
|
||||
default:
|
||||
return fmt.Errorf("delivery: health %q is not ok, failed, waits or unknown", h.Health)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// EncodeHave is h as JSON at Version; it refuses a phase or health outside the lists.
|
||||
func EncodeHave(h Have) ([]byte, error) {
|
||||
if err := h.check(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
h.V = Version
|
||||
return json.Marshal(h)
|
||||
}
|
||||
|
||||
// DecodeHave reads a have; it refuses an unknown version (ErrUnknownVersion) and a phase or health outside the lists.
|
||||
func DecodeHave(data []byte) (Have, error) {
|
||||
var h Have
|
||||
if err := decode(data, &h); err != nil {
|
||||
return Have{}, err
|
||||
}
|
||||
if err := h.check(); err != nil {
|
||||
return Have{}, err
|
||||
}
|
||||
return h, nil
|
||||
}
|
||||
|
||||
// EncodeHaveNode is n as JSON at Version.
|
||||
func EncodeHaveNode(n HaveNode) ([]byte, error) {
|
||||
n.V = Version
|
||||
return json.Marshal(n)
|
||||
}
|
||||
|
||||
// DecodeHaveNode reads a node's facts; it refuses an unknown version (ErrUnknownVersion).
|
||||
func DecodeHaveNode(data []byte) (HaveNode, error) {
|
||||
var n HaveNode
|
||||
err := decode(data, &n)
|
||||
return n, err
|
||||
}
|
||||
|
||||
// EncodePresence is p as JSON at Version.
|
||||
func EncodePresence(p Presence) ([]byte, error) {
|
||||
p.V = Version
|
||||
return json.Marshal(p)
|
||||
}
|
||||
|
||||
// DecodePresence reads a presence; it refuses an unknown version (ErrUnknownVersion).
|
||||
func DecodePresence(data []byte) (Presence, error) {
|
||||
var p Presence
|
||||
err := decode(data, &p)
|
||||
return p, err
|
||||
}
|
||||
|
||||
// EncodePulse is p as JSON at Version.
|
||||
func EncodePulse(p Pulse) ([]byte, error) {
|
||||
p.V = Version
|
||||
return json.Marshal(p)
|
||||
}
|
||||
|
||||
// DecodePulse reads a pulse; it refuses an unknown version (ErrUnknownVersion).
|
||||
func DecodePulse(data []byte) (Pulse, error) {
|
||||
var p Pulse
|
||||
err := decode(data, &p)
|
||||
return p, err
|
||||
}
|
||||
|
||||
// SameHave says whether two haves are the same value, so a writer writes only when they differ.
|
||||
func SameHave(a, b Have) bool { return a == b }
|
||||
|
||||
// SameHaveNode says whether two node facts are the same value, comparing each carried JSON without its spacing.
|
||||
func SameHaveNode(a, b HaveNode) bool {
|
||||
return a.V == b.V && sameJSON(a.Windows, b.Windows) && sameJSON(a.Busy, b.Busy) &&
|
||||
sameJSON(a.Strays, b.Strays) && sameJSON(a.Filters, b.Filters) && sameJSON(a.Profile, b.Profile)
|
||||
}
|
||||
|
||||
// SamePresence says whether two presences are the same value.
|
||||
func SamePresence(a, b Presence) bool { return a == b }
|
||||
|
||||
// SamePulse says whether two pulses are the same value.
|
||||
func SamePulse(a, b Pulse) bool { return a == b }
|
||||
|
||||
func sameJSON(a, b json.RawMessage) bool {
|
||||
var ca, cb bytes.Buffer
|
||||
if json.Compact(&ca, a) != nil || json.Compact(&cb, b) != nil {
|
||||
return bytes.Equal(a, b)
|
||||
}
|
||||
return bytes.Equal(ca.Bytes(), cb.Bytes())
|
||||
}
|
||||
Vendored
+1
@@ -1,6 +1,7 @@
|
||||
# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018164924-2b420603fe83
|
||||
## explicit; go 1.22
|
||||
git.novox.be/novox/mesh-sdk/go/asks
|
||||
git.novox.be/novox/mesh-sdk/go/delivery
|
||||
git.novox.be/novox/mesh-sdk/go/seal
|
||||
# github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op
|
||||
## explicit; go 1.24.0
|
||||
|
||||
Reference in New Issue
Block a user