The grants part publishes busgrants (the removals' grants, the ended controller's leftovers, what the watchman hears); the watchman takes them and the self-check's checked from the seat, provisions from meshprovisionerclient (ADR 0634 §4, ADR 0635 §4; issue 1966)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
@@ -26,10 +26,10 @@ import (
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
|
||||
"git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
|
||||
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
|
||||
"git.novox.be/novox/mesh-sdk/go/provisions"
|
||||
|
||||
"mesh-delivery-watchman/internal/socket"
|
||||
"mesh-delivery-watchman/internal/watch"
|
||||
@@ -382,21 +382,21 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
provisionsWatch := func(kv jetstream.KeyValue) (jetstream.KeyWatcher, error) {
|
||||
pw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { w.Provision(e.Key(), e.Value(), opOf(e)) })
|
||||
if err == nil {
|
||||
w.ReadWhole(provisions.ProvisionsBucket)
|
||||
w.ReadWhole(meshprovisionerclient.ProvisionsBucket)
|
||||
}
|
||||
return pw, err
|
||||
}
|
||||
var provWatch jetstream.KeyWatcher
|
||||
if kv, err := js.KeyValue(ctx, provisions.ProvisionsBucket); err == nil {
|
||||
if kv, err := js.KeyValue(ctx, meshprovisionerclient.ProvisionsBucket); err == nil {
|
||||
if provWatch, err = provisionsWatch(kv); err != nil {
|
||||
process.NotWatched(logger.Printf, provisions.ProvisionsBucket, err, "no provision waiting for the operator is said until the watchman starts again")
|
||||
process.NotWatched(logger.Printf, meshprovisionerclient.ProvisionsBucket, err, "no provision waiting for the operator is said until the watchman starts again")
|
||||
}
|
||||
} else {
|
||||
process.AwaitBucket(ctx, js, conn, provisions.ProvisionsBucket, "no provision waiting for the operator is said",
|
||||
process.AwaitBucket(ctx, js, conn, meshprovisionerclient.ProvisionsBucket, "no provision waiting for the operator is said",
|
||||
logger.Printf).Ready(func(kv jetstream.KeyValue) {
|
||||
pw, err := provisionsWatch(kv)
|
||||
if err != nil {
|
||||
process.NotWatched(logger.Printf, provisions.ProvisionsBucket, err, "no provision waiting for the operator is said until the watchman starts again")
|
||||
process.NotWatched(logger.Printf, meshprovisionerclient.ProvisionsBucket, err, "no provision waiting for the operator is said until the watchman starts again")
|
||||
return
|
||||
}
|
||||
go func() {
|
||||
|
||||
@@ -12,8 +12,9 @@ import (
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshselfcheck"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/busgrants"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
|
||||
"mesh-delivery-watchman/internal/deadletters"
|
||||
"mesh-delivery-watchman/internal/socket"
|
||||
@@ -22,11 +23,11 @@ import (
|
||||
|
||||
// The subjects the watchman hears for the signals table's rows (novox/hq ADR 0564 §4, Limits 3 and 11), each on the bus,
|
||||
// none in the controller's store. Its grant names them (mesh-sdk buscompose, the watchman's principal).
|
||||
const CheckedSubject = "mesh.seat." + health.SelfCheckModule + ".event." + health.EventChecked
|
||||
const CheckedSubject = "mesh.seat." + meshselfcheck.Seat + ".event." + meshselfcheck.EventChecked
|
||||
|
||||
// HeardSubjects is every subject the watchman subscribes for its rows, beside the bus's connect and disconnect: the ones
|
||||
// its grant lets it hear (buscompose.WatchmanHears), so the two cannot drift.
|
||||
var HeardSubjects = buscompose.WatchmanHears()
|
||||
// its grant lets it hear (busgrants.WatchmanHears), so the two cannot drift.
|
||||
var HeardSubjects = busgrants.WatchmanHears()
|
||||
|
||||
// The rows a heard subject goes to.
|
||||
const (
|
||||
|
||||
@@ -3,6 +3,9 @@ module mesh-delivery-watchman
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshselfcheck v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-delivery/modules/mesh-grants/busgrants v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient v0.0.0-00010101000000-000000000000
|
||||
|
||||
@@ -17,8 +17,8 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient"
|
||||
"git.novox.be/novox/mesh-delivery/sdk/heard"
|
||||
"git.novox.be/novox/mesh-sdk/go/provisions"
|
||||
)
|
||||
|
||||
func mustEncode(t *testing.T, f Finding) []byte {
|
||||
@@ -474,20 +474,20 @@ func TestTheWatchmansFindings(t *testing.T) {
|
||||
}, map[string]string{"module.fail2ban.novox.unhealthy": Warning}},
|
||||
{"1954/D2: a provision in waits is its consumer's needs-operator; provided, it ends", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
enc := func(state, why string) []byte {
|
||||
b, err := provisions.Encode(provisions.Provision{Provision: "oidc", Consumer: provisions.End{Module: "photos", Node: "ace"},
|
||||
Provider: provisions.End{Module: "keycloak", Node: "novox"}, State: state, Why: why, At: t0})
|
||||
b, err := meshprovisionerclient.Encode(meshprovisionerclient.Provision{Provision: "oidc", Consumer: meshprovisionerclient.End{Module: "photos", Node: "ace"},
|
||||
Provider: meshprovisionerclient.End{Module: "keycloak", Node: "novox"}, State: state, Why: why, At: t0})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return b
|
||||
}
|
||||
w.Provision("ace.photos.oidc", enc(provisions.Waits, "the operator gives the client's redirect"), Put)
|
||||
w.Provision("ace.photos.db", enc(provisions.Provided, ""), Put)
|
||||
w.Provision("ace.photos.oidc", enc(meshprovisionerclient.Waits, "the operator gives the client's redirect"), Put)
|
||||
w.Provision("ace.photos.db", enc(meshprovisionerclient.Provided, ""), Put)
|
||||
if f := o.findings["module.photos.ace.needs-operator"]; !strings.Contains(f.Summary, "redirect") || f.Machine != "ace" {
|
||||
t.Fatalf("findings = %v", o.findings)
|
||||
}
|
||||
w.Provision("ace.photos.oidc", enc(provisions.Provided, ""), Put)
|
||||
w.Provision("ace.web.oidc", enc(provisions.Waits, "x"), Put)
|
||||
w.Provision("ace.photos.oidc", enc(meshprovisionerclient.Provided, ""), Put)
|
||||
w.Provision("ace.web.oidc", enc(meshprovisionerclient.Waits, "x"), Put)
|
||||
w.Provision("ace.web.oidc", nil, Delete)
|
||||
}, map[string]string{}},
|
||||
{"774: a node's own report: its network, the resolver file rewritten, failed units, a relogin, a directory used as found, a login fallen back", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
@@ -674,10 +674,10 @@ func TestTheWatchmansFindings(t *testing.T) {
|
||||
w.SetStepBound(30 * time.Minute)
|
||||
held, _ := meshwantsclient.EncodeWant(meshwantsclient.Want{Fp: fpOld, Self: true, Newest: fpNew, Rule: meshwantsclient.RuleCanary, Words: "the canary ace does not run it yet"})
|
||||
released, _ := meshwantsclient.EncodeWant(meshwantsclient.Want{Fp: fpNew, Self: true})
|
||||
asked, _ := provisions.Encode(provisions.Provision{Provision: "oidc", Consumer: provisions.End{Module: "photos", Node: "ace"},
|
||||
Provider: provisions.End{Module: "keycloak", Node: "novox"}, State: provisions.Asked, At: t0})
|
||||
provided, _ := provisions.Encode(provisions.Provision{Provision: "oidc", Consumer: provisions.End{Module: "photos", Node: "ace"},
|
||||
Provider: provisions.End{Module: "keycloak", Node: "novox"}, State: provisions.Provided, At: t0})
|
||||
asked, _ := meshprovisionerclient.Encode(meshprovisionerclient.Provision{Provision: "oidc", Consumer: meshprovisionerclient.End{Module: "photos", Node: "ace"},
|
||||
Provider: meshprovisionerclient.End{Module: "keycloak", Node: "novox"}, State: meshprovisionerclient.Asked, At: t0})
|
||||
provided, _ := meshprovisionerclient.Encode(meshprovisionerclient.Provision{Provision: "oidc", Consumer: meshprovisionerclient.End{Module: "photos", Node: "ace"},
|
||||
Provider: meshprovisionerclient.End{Module: "keycloak", Node: "novox"}, State: meshprovisionerclient.Provided, At: t0})
|
||||
bkey, _ := builds.BuilderBuildKey("web", fpNew)
|
||||
queued, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildQueued, Waits: "mesh-sdk@v0.1.14"})
|
||||
building, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildBuilding, By: "n1"})
|
||||
@@ -685,15 +685,15 @@ func TestTheWatchmansFindings(t *testing.T) {
|
||||
w.Want("want.ace.web", held, Put, t0)
|
||||
w.Provision("ace.photos.oidc", asked, Put)
|
||||
w.Build(bkey, queued, Put, t0)
|
||||
waits := map[string]string{AssignerWantsBucket: "want.ace.web", provisions.ProvisionsBucket: "ace.photos.oidc", BuilderBuildsBucket: bkey}
|
||||
waits := map[string]string{AssignerWantsBucket: "want.ace.web", meshprovisionerclient.ProvisionsBucket: "ace.photos.oidc", BuilderBuildsBucket: bkey}
|
||||
if len(o.pulses) != 3 || len(o.findings) != 0 {
|
||||
t.Fatalf("pulses %v findings %v", o.pulses, o.findings)
|
||||
}
|
||||
// The keys are written again as their parts move on, with new words and a new time: the wait still began at t0.
|
||||
c.t = c.t.Add(20 * time.Minute)
|
||||
held2, _ := meshwantsclient.EncodeWant(meshwantsclient.Want{Fp: fpOld, Self: true, Newest: fpNew, Rule: meshwantsclient.RuleCanary, Words: "the canary ace runs it, not healthy yet"})
|
||||
asked2, _ := provisions.Encode(provisions.Provision{Provision: "oidc", Consumer: provisions.End{Module: "photos", Node: "ace"},
|
||||
Provider: provisions.End{Module: "keycloak", Node: "novox"}, State: provisions.Asked, Why: "the provider is applying", At: c.t})
|
||||
asked2, _ := meshprovisionerclient.Encode(meshprovisionerclient.Provision{Provision: "oidc", Consumer: meshprovisionerclient.End{Module: "photos", Node: "ace"},
|
||||
Provider: meshprovisionerclient.End{Module: "keycloak", Node: "novox"}, State: meshprovisionerclient.Asked, Why: "the provider is applying", At: c.t})
|
||||
w.Want("want.ace.web", held2, Put, c.t)
|
||||
w.Provision("ace.photos.oidc", asked2, Put)
|
||||
w.Build(bkey, queued, Put, c.t)
|
||||
|
||||
@@ -11,7 +11,7 @@ import (
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
|
||||
"git.novox.be/novox/mesh-sdk/go/provisions"
|
||||
"git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient"
|
||||
)
|
||||
|
||||
// The rows of the signals table the watchman holds (novox/hq ADR 0564 §4, Limits 3 and 11). Each is a key of the pulse
|
||||
@@ -582,7 +582,7 @@ const ControlNodeModule = "nats"
|
||||
// why; the finding ends when no provision of that consumer on that node waits. A key deleted, or one that cannot be read,
|
||||
// waits no more.
|
||||
func (w *Watchman) Provision(key string, value []byte, op Op) {
|
||||
node, consumer, _, _, err := provisions.ParseKey(key)
|
||||
node, consumer, _, _, err := meshprovisionerclient.ParseKey(key)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@@ -594,9 +594,9 @@ func (w *Watchman) Provision(key string, value []byte, op Op) {
|
||||
}
|
||||
delete(w.waiting[id], key)
|
||||
if op == Put {
|
||||
if p, err := provisions.Decode(value); err != nil {
|
||||
if p, err := meshprovisionerclient.Decode(value); err != nil {
|
||||
w.say("the provision %s cannot be read, and is taken as not waiting: %v", key, err)
|
||||
} else if p.State == provisions.Waits {
|
||||
} else if p.State == meshprovisionerclient.Waits {
|
||||
w.waiting[id][key] = fmt.Sprintf("%s from %s on %s: %s", p.Provision, p.Provider.Module, p.Provider.Node, p.Why)
|
||||
}
|
||||
}
|
||||
@@ -620,7 +620,7 @@ func (w *Watchman) Provision(key string, value []byte, op Op) {
|
||||
// holder wrote it, bounded as a step is, and provided, failed, waiting for the operator (its own finding), retiring or deleted
|
||||
// ends it. Called with the lock held.
|
||||
func (w *Watchman) askedWait(key string, value []byte, op Op) {
|
||||
bucket := provisions.ProvisionsBucket
|
||||
bucket := meshprovisionerclient.ProvisionsBucket
|
||||
if op == Expired {
|
||||
return
|
||||
}
|
||||
@@ -629,8 +629,8 @@ func (w *Watchman) askedWait(key string, value []byte, op Op) {
|
||||
w.moved(bucket, key)
|
||||
return
|
||||
}
|
||||
p, err := provisions.Decode(value)
|
||||
if err != nil || p.State != provisions.Asked {
|
||||
p, err := meshprovisionerclient.Decode(value)
|
||||
if err != nil || p.State != meshprovisionerclient.Asked {
|
||||
w.moved(bucket, key)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -0,0 +1,131 @@
|
||||
// Package busgrants is what the grants part grants other parts' accounts that those parts' own code needs to state
|
||||
// as well: the whole authority of an account whose bus declares removals (the nats keeper's tools), and the subjects the
|
||||
// watchman hears (its code subscribes the very ones its grant lets it hear). The part composes every grant; a part that
|
||||
// needs the same words reads them here, the grants part's own published package (novox/hq ADR 0634 §4, ADR 0635 §4).
|
||||
package busgrants
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"regexp"
|
||||
"slices"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshselfcheck"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/declaration"
|
||||
"git.novox.be/novox/mesh-sdk/go/membership"
|
||||
)
|
||||
|
||||
const (
|
||||
// EngineModule is the node-engine's manifest's module name, and EngineActsBucket the bucket of its state acts.
|
||||
EngineModule = "mesh-host"
|
||||
EngineActsBucket = EngineModule + "_acts"
|
||||
|
||||
// DeadLetterNoticesStream is the stream of the notices a consumer gave up on a message, and DeadLetterPrefix the
|
||||
// subjects the copies kept are on.
|
||||
DeadLetterNoticesStream = "DEAD_LETTER_NOTICES"
|
||||
DeadLetterPrefix = "mesh.events.dead."
|
||||
// ProvAppliedSubjects is every machine's provisioner client's reports: `prov.applied.<node>.<module>`.
|
||||
ProvAppliedSubjects = "prov.applied.>"
|
||||
|
||||
// ControllerName and MergesConsumerName are the names of the ended controller's consumers.
|
||||
ControllerName = "controller"
|
||||
MergesConsumerName = "controller-merges"
|
||||
// endedController is the name the ended controller had as a module: its buckets are named for it (ADR 0590 §6).
|
||||
endedController = "mesh-controller"
|
||||
)
|
||||
|
||||
var safeSubject = regexp.MustCompile(`^[A-Za-z0-9_-]+$`)
|
||||
|
||||
// ControllerLeftoverConsumers are the ended controller's durable consumers left on the bus, one leftover of the module
|
||||
// mesh-delivery placed nowhere (issue 1126, milestone section 3), and mesh-catalog's of the controller's built-before,
|
||||
// which its manifest no longer consumes and no part composes; each `<stream>.<consumer>`.
|
||||
var ControllerLeftoverConsumers = []string{
|
||||
"EVENTS." + ControllerName, "EVENTS." + MergesConsumerName, "EVENTS.novox_mesh-delivery", "EVENTS.novox_mesh-catalog", "EVENTS.novox_gitea",
|
||||
"CONTROL." + ControllerName, DeadLetterNoticesStream + "." + ControllerName,
|
||||
// The controller's consumers on three modules' buckets (issue 1126): the take-in on the builder's builds and on the
|
||||
// indexer's modules, and the config follow on the resolved configuration.
|
||||
"KV_mesh-delivery-builder_builds.mesh-controller-take-in", "KV_mesh-delivery-indexer_modules.mesh-controller-take-in",
|
||||
"KV_mesh-config_resolved.mesh-controller-config-follow",
|
||||
}
|
||||
|
||||
// ControllerLeftoverConsumersOn are ControllerLeftoverConsumers and, for each node given, the consumer the controller
|
||||
// made for the node's engine (`EVENTS.<node>_mesh-host`, of restic's backed-up): the engine hears that event without a
|
||||
// durable consumer, and the grants part composes none for it, so nothing reads it (issue 1126).
|
||||
func ControllerLeftoverConsumersOn(nodes []string) []string {
|
||||
out := slices.Clone(ControllerLeftoverConsumers)
|
||||
for _, n := range nodes {
|
||||
if safeSubject.MatchString(n) {
|
||||
out = append(out, "EVENTS."+n+"_"+EngineModule)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// ControllerLeftoverBuckets are the ended controller's buckets, each by its local name (`KV_mesh-controller_<name>`).
|
||||
var ControllerLeftoverBuckets = []string{"asked", "assignments", "builds", "calls", "checks-running", "composed",
|
||||
"condition-history", "conditions", "declared-config", "declared-packages", "delivery-merges", "grants", "hand-acts",
|
||||
"lease", "predefined", "rollouts", "states-announced"}
|
||||
|
||||
// ControllerLeftoverStreams are the ended controller's streams, by their names: CONTROL, ASSIGNMENTS, each of its buckets'
|
||||
// and, for each node given, the declaration files' object bucket the controller wrote (ADR 0486 §2).
|
||||
func ControllerLeftoverStreams(nodes []string) []string {
|
||||
out := []string{"CONTROL", membership.AssignmentsStream}
|
||||
for _, b := range ControllerLeftoverBuckets {
|
||||
out = append(out, "KV_"+endedController+"_"+b)
|
||||
}
|
||||
for _, n := range nodes {
|
||||
if safeSubject.MatchString(n) {
|
||||
out = append(out, "OBJ_"+declaration.FileBucketPrefix+n)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// BusRemovalGrants are the whole authority of an account whose bus declares removals (ADR 0564 Limits 4, ADR 0590 clause
|
||||
// 6), each removal its grant: the node-engine's acts bucket read (STREAM.INFO with a subject filter lists what a removal
|
||||
// would take, so the tools read before they remove) and its keys removed, which nothing else may empty when it fills; and
|
||||
// the ended controller's leftovers, each consumer and each stream read and deleted by its name, for the nodes given (the
|
||||
// declaration files' object buckets). A removal outside the set is refused, as the indexer's judge refuses it in a
|
||||
// manifest (novox/hq ADR 0578 §1).
|
||||
func BusRemovalGrants(removes, nodes []string) ([]string, error) {
|
||||
var out []string
|
||||
for _, r := range removes {
|
||||
switch r {
|
||||
case catalogue.RemovesEngineActs:
|
||||
out = append(out, "$JS.API.STREAM.INFO.KV_"+EngineActsBucket, "$KV."+EngineActsBucket+".>")
|
||||
case catalogue.RemovesControllerLeftovers:
|
||||
for _, c := range ControllerLeftoverConsumersOn(nodes) {
|
||||
out = append(out, "$JS.API.CONSUMER.INFO."+c, "$JS.API.CONSUMER.DELETE."+c)
|
||||
}
|
||||
for _, s := range ControllerLeftoverStreams(nodes) {
|
||||
out = append(out, "$JS.API.STREAM.INFO."+s, "$JS.API.STREAM.DELETE."+s)
|
||||
}
|
||||
default:
|
||||
return nil, fmt.Errorf("a bus account removes %q, which is not one of %s (novox/hq ADR 0564 Limits 4, ADR 0590)", r,
|
||||
strings.Join([]string{catalogue.RemovesEngineActs, catalogue.RemovesControllerLeftovers}, ", "))
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// WatchmanHears are the core subjects the watchman subscribes, with no durable consumer (ADR 0564 clauses 4, 5 and
|
||||
// Limits 3, 11): the power seat's words (S1, ADR 0211), the providers' failing and recovered words (S8), each tool
|
||||
// runner's beat (S11), the self-check's checked (S10), a consumer deleted and the bus module's slow consumers and closed
|
||||
// connections (S9), the provisioner clients' reports (S21), the dead letters kept, and the lease write's events. Never the bus's API audit
|
||||
// (`$JS.EVENT.ADVISORY.API`): it carries every API call's request and answer in the account, the vault's hashes among
|
||||
// them (the lead, 2026-10-29).
|
||||
func WatchmanHears() []string {
|
||||
return []string{
|
||||
"mesh.mod.power.event.*",
|
||||
"mesh.mod.*.event.provisioner.failing",
|
||||
"mesh.mod.*.event.provisioner.recovered",
|
||||
"mesh.control.*.tools-alive",
|
||||
"mesh.seat." + meshselfcheck.Seat + ".event." + meshselfcheck.EventChecked,
|
||||
"$JS.EVENT.ADVISORY.CONSUMER.DELETED.>",
|
||||
"mesh.mod.nats.event.bus.slow-consumer",
|
||||
"mesh.mod.nats.event.bus.connection-closed",
|
||||
ProvAppliedSubjects,
|
||||
DeadLetterPrefix + ">",
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
module git.novox.be/novox/mesh-delivery/modules/mesh-grants/busgrants
|
||||
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshselfcheck v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000
|
||||
)
|
||||
@@ -5,10 +5,10 @@ go 1.26.0
|
||||
require (
|
||||
git.novox.be/novox/mesh-catalog/modules/disk-load/diskreadings v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-networking/modules/mesh-wireguard/go v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-delivery/modules/mesh-grants/busgrants v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhealers v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshselfcheck v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/identity-and-access/go/clients/meshvaultclient v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/placement/go/clients/meshwantsclient v0.0.0-00010101000000-000000000000
|
||||
|
||||
@@ -1252,4 +1252,6 @@ func theBusModulesProcessKeepsBusState(t *testing.T) {
|
||||
t.Errorf("%s may create mod-x's bus state", user)
|
||||
}
|
||||
}
|
||||
t.Run("the names are the manifest's", namesAreTheManifests)
|
||||
t.Run("the bus grants are the declared", busGrantsAreTheDeclared)
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
nodeengineseat "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
|
||||
"git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/busgrants"
|
||||
)
|
||||
|
||||
// The node seat node-engine (novox/hq ADR 0563 §1): every node's engine holds it on its own bus account, and no module
|
||||
@@ -16,7 +17,7 @@ const EngineSeat = nodeengineseat.Seat
|
||||
// node seat and its own asks are kept in (ADR 0563 §2, §3: "the engine's own state"): the bucket mesh-host_acts, keys
|
||||
// `<node>.act.<id>`, `<node>.open.<act>.<digest>` and `<node>.ask.<id>`, each written by that node's engine alone.
|
||||
const (
|
||||
EngineModule = "mesh-host"
|
||||
EngineModule = busgrants.EngineModule
|
||||
EngineActsState = "acts"
|
||||
)
|
||||
|
||||
|
||||
@@ -1,13 +1,12 @@
|
||||
package buscompose
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/busgrants"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhealers"
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshselfcheck"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
@@ -25,80 +24,20 @@ import (
|
||||
// HealersWould is what mesh-healers would have done while a controller was placed: read by nobody (ADR 0564 clause 8).
|
||||
var HealersWould = meshhealers.Seat + "." + meshhealers.StateWould
|
||||
|
||||
// ControllerLeftoverConsumers are the ended controller's durable consumers left on the bus, one leftover of the module
|
||||
// mesh-delivery placed nowhere (issue 1126, milestone section 3), and mesh-catalog's of the controller's built-before,
|
||||
// which its manifest no longer consumes and no part composes; each `<stream>.<consumer>`.
|
||||
var ControllerLeftoverConsumers = []string{
|
||||
"EVENTS." + ControllerName, "EVENTS." + MergesConsumerName, "EVENTS.novox_mesh-delivery", "EVENTS.novox_mesh-catalog", "EVENTS.novox_gitea",
|
||||
"CONTROL." + ControllerName, DeadLetterNoticesStream + "." + ControllerName,
|
||||
// The controller's consumers on three modules' buckets (issue 1126): the take-in on the builder's builds and on the
|
||||
// indexer's modules, and the config follow on the resolved configuration.
|
||||
"KV_mesh-delivery-builder_builds.mesh-controller-take-in", "KV_mesh-delivery-indexer_modules.mesh-controller-take-in",
|
||||
"KV_mesh-config_resolved.mesh-controller-config-follow",
|
||||
}
|
||||
// The ended controller's leftovers an account may remove, the grants of an account that declares removals and the subjects the
|
||||
// watchman hears are the grants part's published package busgrants, which the parts that state them too read.
|
||||
var (
|
||||
ControllerLeftoverConsumers = busgrants.ControllerLeftoverConsumers
|
||||
ControllerLeftoverBuckets = busgrants.ControllerLeftoverBuckets
|
||||
)
|
||||
|
||||
// ControllerLeftoverConsumersOn are ControllerLeftoverConsumers and, for each node given, the consumer the controller
|
||||
// made for the node's engine (`EVENTS.<node>_mesh-host`, of restic's backed-up): the engine hears that event without a
|
||||
// durable consumer, and the grants part composes none for it, so nothing reads it (issue 1126).
|
||||
func ControllerLeftoverConsumersOn(nodes []string) []string {
|
||||
out := slices.Clone(ControllerLeftoverConsumers)
|
||||
for _, n := range nodes {
|
||||
if safeSubject.MatchString(n) {
|
||||
out = append(out, "EVENTS."+n+"_"+EngineModule)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// endedController is the name the ended controller had as a module: its buckets are named for it (ADR 0590 §6).
|
||||
const endedController = "mesh-controller"
|
||||
|
||||
// ControllerLeftoverBuckets are the ended controller's buckets, each by its local name (`KV_mesh-controller_<name>`).
|
||||
var ControllerLeftoverBuckets = []string{"asked", "assignments", "builds", "calls", "checks-running", "composed",
|
||||
"condition-history", "conditions", "declared-config", "declared-packages", "delivery-merges", "grants", "hand-acts",
|
||||
"lease", "predefined", "rollouts", "states-announced"}
|
||||
|
||||
// ControllerLeftoverStreams are the ended controller's streams, by their names: CONTROL, ASSIGNMENTS, each of its buckets'
|
||||
// and, for each node given, the declaration files' object bucket the controller wrote (ADR 0486 §2).
|
||||
func ControllerLeftoverStreams(nodes []string) []string {
|
||||
out := []string{"CONTROL", AssignmentsStream}
|
||||
for _, b := range ControllerLeftoverBuckets {
|
||||
out = append(out, "KV_"+BucketName(endedController, b))
|
||||
}
|
||||
for _, n := range nodes {
|
||||
if safeSubject.MatchString(n) {
|
||||
out = append(out, "OBJ_"+DeclarationFilesBucket+n)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// BusRemovalGrants are the whole authority of an account whose bus declares removals (ADR 0564 Limits 4, ADR 0590 clause
|
||||
// 6), each removal its grant: the node-engine's acts bucket read (STREAM.INFO with a subject filter lists what a removal
|
||||
// would take, so the tools read before they remove) and its keys removed, which nothing else may empty when it fills; and
|
||||
// the ended controller's leftovers, each consumer and each stream read and deleted by its name, for the nodes given (the
|
||||
// declaration files' object buckets). A removal outside the set is refused, as the indexer's judge refuses it in a
|
||||
// manifest (novox/hq ADR 0578 §1).
|
||||
func BusRemovalGrants(removes, nodes []string) ([]string, error) {
|
||||
var out []string
|
||||
for _, r := range removes {
|
||||
switch r {
|
||||
case catalogue.RemovesEngineActs:
|
||||
out = append(out, "$JS.API.STREAM.INFO.KV_"+EngineActsBucket, "$KV."+EngineActsBucket+".>")
|
||||
case catalogue.RemovesControllerLeftovers:
|
||||
for _, c := range ControllerLeftoverConsumersOn(nodes) {
|
||||
out = append(out, "$JS.API.CONSUMER.INFO."+c, "$JS.API.CONSUMER.DELETE."+c)
|
||||
}
|
||||
for _, s := range ControllerLeftoverStreams(nodes) {
|
||||
out = append(out, "$JS.API.STREAM.INFO."+s, "$JS.API.STREAM.DELETE."+s)
|
||||
}
|
||||
default:
|
||||
return nil, fmt.Errorf("a bus account removes %q, which is not one of %s (novox/hq ADR 0564 Limits 4, ADR 0590)", r,
|
||||
strings.Join([]string{catalogue.RemovesEngineActs, catalogue.RemovesControllerLeftovers}, ", "))
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
// ControllerLeftoverConsumersOn, ControllerLeftoverStreams, BusRemovalGrants and WatchmanHears are busgrants'.
|
||||
var (
|
||||
ControllerLeftoverConsumersOn = busgrants.ControllerLeftoverConsumersOn
|
||||
ControllerLeftoverStreams = busgrants.ControllerLeftoverStreams
|
||||
BusRemovalGrants = busgrants.BusRemovalGrants
|
||||
WatchmanHears = busgrants.WatchmanHears
|
||||
)
|
||||
|
||||
// removesLeftover is an account declared to remove the ended controller's leftovers deleting one of its streams: the
|
||||
// writers table's one exemption from stream definitions being the grants part's (ADR 0590 clause 6).
|
||||
@@ -149,27 +88,6 @@ func WatchmanDeadLetterGrants(sources []string) []string {
|
||||
return out
|
||||
}
|
||||
|
||||
// WatchmanHears are the core subjects the watchman subscribes, with no durable consumer (ADR 0564 clauses 4, 5 and
|
||||
// Limits 3, 11): the power seat's words (S1, ADR 0211), the providers' failing and recovered words (S8), each tool
|
||||
// runner's beat (S11), the self-check's checked (S10), a consumer deleted and the bus module's slow consumers and closed
|
||||
// connections (S9), the provisioner clients' reports (S21), the dead letters kept, and the lease write's events. Never the bus's API audit
|
||||
// (`$JS.EVENT.ADVISORY.API`): it carries every API call's request and answer in the account, the vault's hashes among
|
||||
// them (the lead, 2026-10-29).
|
||||
func WatchmanHears() []string {
|
||||
return []string{
|
||||
"mesh.mod.power.event.*",
|
||||
"mesh.mod.*.event.provisioner.failing",
|
||||
"mesh.mod.*.event.provisioner.recovered",
|
||||
"mesh.control.*.tools-alive",
|
||||
"mesh.seat." + meshselfcheck.Seat + ".event." + meshselfcheck.EventChecked,
|
||||
"$JS.EVENT.ADVISORY.CONSUMER.DELETED.>",
|
||||
"mesh.mod.nats.event.bus.slow-consumer",
|
||||
"mesh.mod.nats.event.bus.connection-closed",
|
||||
ProvAppliedSubjects,
|
||||
deadLetterPrefix + ">",
|
||||
}
|
||||
}
|
||||
|
||||
// DeadLetterSources are the streams of names a dead letter may come from, sorted, none twice: each a plain name, never a
|
||||
// bucket's (KV_) or an object store's (OBJ_) stream, the dead letters' own two streams, or a pattern.
|
||||
func DeadLetterSources(names []string) []string {
|
||||
|
||||
@@ -2,16 +2,17 @@ package buscompose
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"os"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// TestTheNamesAreTheManifests: the holders and states of other parts that the grants part names are the ones its manifest
|
||||
// namesAreTheManifests (a subtest of TestTheCopiedComposerComposesAsTheControllers): the holders and states of other parts that the grants part names are the ones its manifest
|
||||
// lists in `reads` and `invokes` (no contract names a holder: novox/hq ADR 0634 §1), and a state's bucket is its holder's
|
||||
// name and the state's.
|
||||
func TestTheNamesAreTheManifests(t *testing.T) {
|
||||
func namesAreTheManifests(t *testing.T) {
|
||||
raw, err := os.ReadFile("../../module.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -48,3 +49,30 @@ func TestTheNamesAreTheManifests(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// busGrantsAreTheDeclared (a subtest of TestTheCopiedComposerComposesAsTheControllers): a removal is its grants, the leftovers' for
|
||||
// the nodes given too, one outside the set is refused, and the watchman hears the self-check's word and the dead letters;
|
||||
// the functions are the grants part's published package busgrants'.
|
||||
func busGrantsAreTheDeclared(t *testing.T) {
|
||||
got, err := BusRemovalGrants([]string{catalogue.RemovesEngineActs, catalogue.RemovesControllerLeftovers}, []string{"novox", "a b"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, want := range []string{"$KV.mesh-host_acts.>", "$JS.API.CONSUMER.DELETE.EVENTS.controller", "$JS.API.CONSUMER.DELETE.EVENTS.novox_mesh-host",
|
||||
"$JS.API.STREAM.DELETE.OBJ_declaration-files-novox", "$JS.API.STREAM.DELETE.KV_mesh-controller_lease", "$JS.API.STREAM.DELETE.ASSIGNMENTS"} {
|
||||
if !slices.Contains(got, want) {
|
||||
t.Errorf("the grants lack %s", want)
|
||||
}
|
||||
}
|
||||
if slices.ContainsFunc(got, func(s string) bool {
|
||||
return s == "$JS.API.STREAM.DELETE.OBJ_declaration-files-a b" || s == "$JS.API.CONSUMER.DELETE.EVENTS.a b_mesh-host"
|
||||
}) {
|
||||
t.Errorf("a node name that is not a subject token widened the grants: %v", got)
|
||||
}
|
||||
if _, err := BusRemovalGrants([]string{"everything"}, nil); err == nil {
|
||||
t.Error("a removal outside the set was granted")
|
||||
}
|
||||
if hears := WatchmanHears(); !slices.Contains(hears, "mesh.seat.mesh-self-check.event.checked") || !slices.Contains(hears, "mesh.events.dead.>") {
|
||||
t.Errorf("the watchman hears %v", hears)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ package buscompose
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/busgrants"
|
||||
"sort"
|
||||
"strings"
|
||||
)
|
||||
@@ -91,11 +92,11 @@ const WorkQueueMetadata = "mesh-work-queue"
|
||||
// the subject, how often it was handed over and when it was given up. It stays there until a person
|
||||
// delivers it again or drops it, with why; a condition is open for as long as it does.
|
||||
const (
|
||||
DeadLetterNoticesStream = "DEAD_LETTER_NOTICES"
|
||||
DeadLetterNoticesStream = busgrants.DeadLetterNoticesStream
|
||||
DeadLettersStream = "DEAD_LETTERS"
|
||||
// MaxDeliveriesAdvisories is the subject the server says a given-up message on.
|
||||
MaxDeliveriesAdvisories = "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>"
|
||||
deadLetterPrefix = "mesh.events.dead."
|
||||
deadLetterPrefix = busgrants.DeadLetterPrefix
|
||||
againPrefix = "mesh.again."
|
||||
// DeadLettersBytes bounds the kept copies. Full, the stream refuses the next copy, which is said
|
||||
// as the consumer's condition; it never drops one it holds.
|
||||
@@ -221,7 +222,7 @@ const ProvAppliedStream = "PROV_APPLIED"
|
||||
|
||||
// ProvAppliedSubjects is every machine's client's reports: `prov.applied.<node>.<module>`, a module's name dots and
|
||||
// all (ProvisionerAppliedSubject).
|
||||
const ProvAppliedSubjects = "prov.applied.>"
|
||||
const ProvAppliedSubjects = busgrants.ProvAppliedSubjects
|
||||
|
||||
// DeliverSubjectFor is where a push consumer's messages land.
|
||||
//
|
||||
@@ -277,8 +278,8 @@ func Overlaps() []string {
|
||||
// removal (ControllerLeftoverConsumers) and the watchman can read whether `controller` on DEAD_LETTER_NOTICES still
|
||||
// stands.
|
||||
const (
|
||||
ControllerName = "controller"
|
||||
MergesConsumerName = "controller-merges"
|
||||
ControllerName = busgrants.ControllerName
|
||||
MergesConsumerName = busgrants.MergesConsumerName
|
||||
)
|
||||
|
||||
// AsksSeat is the seat an ask is made on and its warrant heard from (novox/hq ADR 0259): the router's.
|
||||
|
||||
@@ -147,6 +147,15 @@
|
||||
"grantsstate"
|
||||
],
|
||||
"series": "0.1"
|
||||
},
|
||||
{
|
||||
"name": "busgrants",
|
||||
"kind": "package",
|
||||
"language": "go",
|
||||
"paths": [
|
||||
"busgrants"
|
||||
],
|
||||
"series": "0.1"
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user