Merge branch 'feat/47-G-fix11-b4e' into feat/47-G-join11
# Conflicts: # modules/mesh-grants/internal/buscompose/users.go
This commit is contained in:
@@ -132,3 +132,29 @@ func (c creating) AddFilters(stream, name string, add []string) error {
|
||||
_, err = c.js.UpdateConsumer(ctx, stream, cfg)
|
||||
return err
|
||||
}
|
||||
|
||||
// PurgeSubject purges the messages of one subject of a stream (design 47 G §544): the named asks of a build seat's queue.
|
||||
func (c creating) PurgeSubject(stream, subject string) error {
|
||||
ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
|
||||
defer cancel()
|
||||
st, err := c.js.Stream(ctx, stream)
|
||||
if errors.Is(err, jetstream.ErrStreamNotFound) {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return st.Purge(ctx, jetstream.WithPurgeSubject(subject))
|
||||
}
|
||||
|
||||
// DeleteConsumer deletes one consumer of a stream (design 47 G §544): a build seat's named worker; one the bus does not hold
|
||||
// is no error.
|
||||
func (c creating) DeleteConsumer(stream, name string) error {
|
||||
ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
|
||||
defer cancel()
|
||||
err := c.js.DeleteConsumer(ctx, stream, name)
|
||||
if errors.Is(err, jetstream.ErrConsumerNotFound) || errors.Is(err, jetstream.ErrStreamNotFound) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -623,6 +623,31 @@ func TestTheGrantsPartWhole(t *testing.T) {
|
||||
if again, _ := stream.Info(ctx); !slices.Equal(again.Config.Subjects, []string{"a.>", "b.>"}) {
|
||||
t.Errorf("subjects after adding a held one: %v", again.Config.Subjects)
|
||||
}
|
||||
// The named asks' purge takes one subject's messages and no other's, and a named worker's delete is said done also
|
||||
// where the bus holds none (design 47 G §544).
|
||||
for _, subj := range []string{"b.one", "b.two"} {
|
||||
if _, err := js.Publish(ctx, subj, []byte("{}")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
held, _ := stream.Info(ctx)
|
||||
if err := c.PurgeSubject("ADDS", "b.*"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if left, _ := stream.Info(ctx); left.State.Msgs != held.State.Msgs-2 {
|
||||
t.Errorf("the purge left %d messages of %d, want the two of b.* gone", left.State.Msgs, held.State.Msgs)
|
||||
}
|
||||
for i := 0; i < 2; i++ {
|
||||
if err := c.DeleteConsumer("ADDS", "idx"); err != nil {
|
||||
t.Fatalf("delete %d: %v", i, err)
|
||||
}
|
||||
}
|
||||
if _, err := js.Consumer(ctx, "ADDS", "idx"); err == nil {
|
||||
t.Error("the consumer is still held after its delete")
|
||||
}
|
||||
if err := c.PurgeSubject("NO_SUCH", "x.*"); err != nil {
|
||||
t.Errorf("the purge of a stream the bus lacks: %v", err)
|
||||
}
|
||||
})
|
||||
|
||||
// And what the part composes holds on a server that loads it (one integration test per module, the test budget).
|
||||
|
||||
@@ -178,7 +178,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}
|
||||
fv.Assignment = assignment
|
||||
f.Part = &grants.Part{Read: fv, Call: conn, Store: ownStore{conn}, Emit: emitting{conn}, Views: fv, FirstViewContext: ctx,
|
||||
Create: creating{ctx: ctx, js: js}, Consumers: creating{ctx: ctx, js: js},
|
||||
Create: creating{ctx: ctx, js: js}, Consumers: creating{ctx: ctx, js: js}, NamedPurge: creating{ctx: ctx, js: js},
|
||||
Objects: sealedObjects{ctx: ctx, js: js}, Purging: ownPurging{ctx: ctx, js: js, module: account.Module}, Logf: logger.Printf,
|
||||
Bus: grants.BusAddress{URL: account.URL, Fingerprint: account.Fingerprint}}
|
||||
f.Watch = fv
|
||||
@@ -192,6 +192,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
return grants.Unwrapped(conn.Call)(grants.CallBrokerList, map[string]any{})
|
||||
}}
|
||||
f.Part.Bindings.Start(ctx, f.Part)
|
||||
// The restore points' names (design 47 G §565): the seat mesh-recovery's kept-points, a kept call asked at each start and on the
|
||||
// seat's ended, point-removed and expiry-ended, until it answers.
|
||||
f.Part.KeptPoints = &grants.KeptPointsView{Changed: f.Wake, Logf: logger.Printf,
|
||||
Ask: func(context.Context) (json.RawMessage, error) {
|
||||
return grants.Unwrapped(conn.Call)(grants.CallKeptPoints, map[string]any{})
|
||||
}}
|
||||
f.Part.KeptPoints.Start(ctx, f.Part)
|
||||
// The seat list: read over the connection and the hub, given to catalogue at every change, and a pass composes again then.
|
||||
gate := newSeatGate()
|
||||
gate.Follow(seatlist.ProcessBus(conn.Call, hub), f.Wake, logger.Printf)
|
||||
|
||||
@@ -288,6 +288,14 @@ func toolsOf(p *grants.Part) []stdio.Tool {
|
||||
Run: func(a map[string]any) (any, error) {
|
||||
return p.ReleaseWait(stdio.CallerOf(a), grants.Answer(a, "module"), grants.Answer(a, "why"), recordHandAct)
|
||||
}},
|
||||
{Name: grants.ToolNamedPurge, Description: "Purge the build seats' named asks (accept.<queue>.* of each seat's stream) and delete " +
|
||||
"their named workers, by a person's act with why, recorded as a hand act (the seat mesh-hand-acts) before anything is " +
|
||||
"asked (design 47 G §544): the step where the grants part stopped before it purged them at the first revision in " +
|
||||
"which no entry invokes a queue of a build seat. Asks the part's process, whose journal names each purge and delete.",
|
||||
Input: schema(map[string]string{"why": "why, kept in the hand act"}, "why"),
|
||||
Run: func(a map[string]any) (any, error) {
|
||||
return p.AskNamedPurge(stdio.CallerOf(a), grants.Answer(a, "why"), recordHandActAs("named-purge"))
|
||||
}},
|
||||
{Name: grants.ToolRunning, Description: "What each node-engine said it runs (novox/hq design 47, ADR 0660 §2): the last event " +
|
||||
"running of each node and module, as the grants part keeps it in its state running (module, node, the build's " +
|
||||
"fingerprint, the digest of that build's manifest, when it was seen running healthy). A seat's holder is read from " +
|
||||
@@ -298,9 +306,14 @@ func toolsOf(p *grants.Part) []stdio.Tool {
|
||||
}
|
||||
|
||||
// recordHandAct records a person's act through the seat mesh-hand-acts.
|
||||
func recordHandAct(what, why string) error {
|
||||
_, err := meshhandactsclient.Recorder{Ask: stdio.Ask}.Record(meshhandactsclient.Request{What: what, Why: why, Cause: "first-view-again"})
|
||||
return err
|
||||
func recordHandAct(what, why string) error { return recordHandActAs("first-view-again")(what, why) }
|
||||
|
||||
// recordHandActAs records a person's act of the cause given through the seat mesh-hand-acts.
|
||||
func recordHandActAs(cause string) func(what, why string) error {
|
||||
return func(what, why string) error {
|
||||
_, err := meshhandactsclient.Recorder{Ask: stdio.Ask}.Record(meshhandactsclient.Request{What: what, Why: why, Cause: cause})
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// serveTools serves the tools to the runtime until it stops.
|
||||
|
||||
@@ -8,6 +8,7 @@ require (
|
||||
git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/go/seats/meshbroker v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/go/seats/meshrecovery 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/meshhandactsclient v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000
|
||||
|
||||
@@ -3081,8 +3081,8 @@ func TestGrants(t *testing.T) {
|
||||
t.Errorf("%s is no seat event of Consumes beside the verdicts and views", subject)
|
||||
}
|
||||
}
|
||||
if n := len(grants.SeatEventSubjects()); n != 6 {
|
||||
t.Errorf("the part hears %d seats' events, want the keeper's two, the vault's one and the broker's three", n)
|
||||
if n := len(grants.SeatEventSubjects()); n != 9 {
|
||||
t.Errorf("the part hears %d seats' events, want the keeper's two, the vault's one, the broker's three and the recovery's three", n)
|
||||
}
|
||||
for _, c := range b.calls {
|
||||
if strings.Contains(c, "mesh-controller") {
|
||||
|
||||
@@ -239,6 +239,16 @@ type boundGrants struct {
|
||||
// the bus being the mesh's `<module>_<local>` (BusStateStreams) and a stream's subjects under `mesh.bus.<module>.<local>`;
|
||||
// a binding naming any other is granted nothing.
|
||||
func boundOf(module string, bindings []membership.BusBinding) boundGrants {
|
||||
return boundOn(module, "", bindings)
|
||||
}
|
||||
|
||||
// FindingEventsLocal is the local name a part asks its finding stream under, by the template events (design 47 G §529).
|
||||
const FindingEventsLocal = "finding-events"
|
||||
|
||||
// boundOn is boundOf for the account of a module on node: the module's finding stream, whose subjects end in the machine
|
||||
// (`mesh.bus.<module>.finding-events.<event>.<node>`), is published on its own node's subjects alone (§529). An empty node
|
||||
// grants every machine's.
|
||||
func boundOn(module, node string, bindings []membership.BusBinding) boundGrants {
|
||||
var g boundGrants
|
||||
for _, b := range bindings {
|
||||
base := module + "_" + b.Local
|
||||
@@ -263,7 +273,11 @@ func boundOf(module string, bindings []membership.BusBinding) boundGrants {
|
||||
if b.Stream != base || b.Subject != "mesh.bus."+module+"."+b.Local {
|
||||
continue
|
||||
}
|
||||
g.Streams = append(g.Streams, b.Subject+".>",
|
||||
published := b.Subject + ".>"
|
||||
if node != "" && b.Local == FindingEventsLocal && safeSubject.MatchString(node) {
|
||||
published = b.Subject + ".*." + node
|
||||
}
|
||||
g.Streams = append(g.Streams, published,
|
||||
"$JS.API.STREAM.INFO."+base, "$JS.API.STREAM.MSG.GET."+base, "$JS.API.DIRECT.GET."+base+".>",
|
||||
"$JS.API.CONSUMER.CREATE."+base+".>", "$JS.API.CONSUMER.DELETE."+base+".>", "$JS.FC."+base+".>")
|
||||
}
|
||||
|
||||
@@ -67,3 +67,73 @@ func TestNamedAsks(t *testing.T) {
|
||||
t.Errorf("the work queue's consumers: %d, want 4 (a main and a named worker for each of two queues)", len(consumers))
|
||||
}
|
||||
}
|
||||
|
||||
// Design 47 G §610, A-179 (test 60's case of §271): a user of a seat is granted a verb's call subject only where the verb's
|
||||
// effective callers name it, so a kept-artifact call is composed for the module that invokes it and for no other user of
|
||||
// the build seat; `module:` and `seat:` entries name theirs, and a seat whose definition is unread grants its users no verb.
|
||||
func TestAVerbIsGrantedToItsEffectiveCallersAlone(t *testing.T) {
|
||||
build := Seat{Name: "node-build-go", Scope: "node", Serves: []string{"kept-artifact", "freed", "run"},
|
||||
VerbCallers: map[string][]string{
|
||||
"kept-artifact": {"invokers", "persons", "sessions"},
|
||||
"freed": {"module:a-checker"},
|
||||
"run": {"seat:mesh-pipeline-runner"},
|
||||
}}
|
||||
user := func(module string, invokes []string, held ...string) Principal {
|
||||
p := Principal{Kind: KindModule, Node: "one", Module: module, Uses: []Seat{build}, Invokes: invokes}
|
||||
for _, h := range held {
|
||||
p.Holds = append(p.Holds, Seat{Name: h})
|
||||
}
|
||||
return p
|
||||
}
|
||||
unread := build
|
||||
unread.VerbCallers = nil
|
||||
const base = "mesh.seat.node-build-go.tool."
|
||||
for _, c := range []struct {
|
||||
who Principal
|
||||
want map[string]bool
|
||||
}{
|
||||
{user("an-invoker", []string{"seat:node-build-go.kept-artifact"}), map[string]bool{
|
||||
base + "kept-artifact.ace": true, base + "freed.ace": false, base + "run.ace": false}},
|
||||
{user("a-runner", nil, "mesh-pipeline-runner"), map[string]bool{
|
||||
base + "kept-artifact.ace": false, base + "freed.ace": false, base + "run.ace": true}},
|
||||
{user("a-checker", nil), map[string]bool{base + "kept-artifact.ace": false, base + "freed.ace": true}},
|
||||
{Principal{Kind: KindModule, Node: "one", Module: "a-checker", Uses: []Seat{unread}}, map[string]bool{
|
||||
base + "freed.ace": false, base + "kept-artifact.ace": false}},
|
||||
} {
|
||||
perms, err := PermissionsFor(c.who)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for subject, want := range c.want {
|
||||
if got := MayPublish(perms, subject); got != want {
|
||||
t.Errorf("%s may publish %s: %v, want %v", c.who.Module, subject, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Design 47 G §544: the grants part's account alone is granted the purge of a build seat's stream and the delete of its
|
||||
// named workers, for every queue of a build seat a composed user holds.
|
||||
func TestTheGrantsPartAlonePurgesTheNamedAsks(t *testing.T) {
|
||||
held := Seat{Name: "node-build-go", Scope: "node", Accepts: []string{"check"}, PulledByKind: true}
|
||||
users, err := Users(Records{Nodes: []string{"one"}, Assigned: map[string][]Declared{"one": {
|
||||
{Module: GrantsModule}, {Module: "go-module", Holds: []Seat{held}}}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, u := range users {
|
||||
perms, err := PermissionsFor(u)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, subject := range []string{"$JS.API.STREAM.PURGE.SEAT_NODE_BUILD_GO",
|
||||
"$JS.API.CONSUMER.DELETE.SEAT_NODE_BUILD_GO.SEAT_NODE_BUILD_GO_CHECK_named_worker"} {
|
||||
if got, want := MayPublish(perms, subject), u.Module == GrantsModule && u.Kind == KindModule; got != want {
|
||||
t.Errorf("%s may publish %s: %v, want %v", u.Username(), subject, got, want)
|
||||
}
|
||||
}
|
||||
if u.Module == GrantsModule && MayPublish(perms, "$JS.API.CONSUMER.DELETE.SEAT_NODE_BUILD_GO.SEAT_NODE_BUILD_GO_CHECK_worker") {
|
||||
t.Errorf("the grants part may delete main's worker")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -28,6 +28,7 @@ import (
|
||||
sdkbc "git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/membership"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
)
|
||||
|
||||
// A Kind is what a principal is, which decides the shape of its authority rather than its
|
||||
@@ -110,6 +111,14 @@ type Seat struct {
|
||||
// entries of the callers of the accepted kinds the definition puts on that queue (design 47 G §543, §600). The asker
|
||||
// in a named ask's subject is the asking seat's name (A-163). Empty where the seat list is unread.
|
||||
AcceptCallers map[string][]string
|
||||
// VerbCallers are, by verb, the callers each verb the seat serves admits: its definition's effective list
|
||||
// (seatdef.Verb.EffectiveCallers, the class defaults of design 47 G §610). A user of the seat is granted a verb's call
|
||||
// subject only where that list names it (A-179), so the bus holds every other caller out. Empty where the seat list is
|
||||
// unread, and then no verb of the seat is granted to its users.
|
||||
VerbCallers map[string][]string
|
||||
// NoNamedAsks says no entry of the revision invokes a queue of a build seat (`seat:<seat>.accept.<queue>`), so its queues
|
||||
// have no named worker: the grants part purges their named asks and deletes those workers (design 47 G §544, A-231).
|
||||
NoNamedAsks bool
|
||||
// HeldByTheEngine says every node's engine holds the seat on its own account (catalogue.Seat.HeldByTheEngine, novox/hq
|
||||
// ADR 0563 §1): each of its events names the engine's machine as its last token (machineEvent).
|
||||
HeldByTheEngine bool
|
||||
@@ -149,6 +158,11 @@ type Principal struct {
|
||||
// ruling, registry milestone 13:0x): `mesh.seat.mesh-broker.event.<event>` for each event the seat's definition
|
||||
// names, as a holder's own account would be granted them; set on the keeper alone.
|
||||
SaysSeatEvents []string
|
||||
// PurgesNamed are the named asks of every queue of a build seat a composed user holds, for the grants part alone: the
|
||||
// purge of their subjects on the seat's stream and the delete of their named workers (design 47 G §544).
|
||||
PurgesNamed []NamedQueue
|
||||
// Recovers are the bus names of kept data a module holding the seat mesh-recovery is granted (RecoveryGrants), for a KindModule principal.
|
||||
Recovers []string
|
||||
// Process is the process resource's id, for a KindProcess principal alone (novox/hq ADR 0436 §2):
|
||||
// Username and inbox are <node>.<module>.<process id>.
|
||||
Process string
|
||||
@@ -940,6 +954,8 @@ func permissionsFor(p Principal) (Permissions, error) {
|
||||
pub = append(pub, CreateStreamGrants(p.CreatesStreams)...)
|
||||
// And creating each durable consumer an account needs, by its stream and name (ADR 0561 §6).
|
||||
pub = append(pub, CreateConsumerGrants(p.CreatesConsumers)...)
|
||||
// And the purge of the build seats' named asks and the delete of their named workers (design 47 G §544).
|
||||
pub = append(pub, NamedPurgeGrants(p.PurgesNamed)...)
|
||||
// And its own object bucket, where it writes the user list sealed to the bus node (ADR 0547 clause 7).
|
||||
pub = append(pub, GrantsUserListWrites()...)
|
||||
}
|
||||
@@ -968,6 +984,8 @@ func permissionsFor(p Principal) (Permissions, error) {
|
||||
// away — no other principal may subscribe this namespace, and a caller's authority is
|
||||
// still granted per tool, by name, on the publish side.
|
||||
sub = append(sub, own+".tool.>")
|
||||
// The holder of the seat mesh-recovery: the bus names of kept data it takes, restores and deletes (design 47 G §554).
|
||||
pub = append(pub, RecoveryGrants(p.Recovers)...)
|
||||
// It says what it serves (novox/hq ADR 0197): discovery for its own name and every seat it
|
||||
// holds a verb of, answered by the runtime that serves them.
|
||||
announced := []string{p.Module}
|
||||
@@ -1077,14 +1095,17 @@ func permissionsFor(p Principal) (Permissions, error) {
|
||||
pub = append(pub, seatSubject(s, "accept", a))
|
||||
}
|
||||
for _, t := range s.Serves {
|
||||
pub = append(pub, seatToolSubject(s, t, "*"))
|
||||
// Only a verb whose effective callers name this module (design 47 G A-179, §610).
|
||||
if callerOf(p, s, t) {
|
||||
pub = append(pub, seatToolSubject(s, t, "*"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 5. Its state, and the state of others it reads (novox/hq ADR 0201): every one read and
|
||||
// watched, its own written too.
|
||||
// Its bus state asked in `requires` is granted as the old sections are (boundOf, ADR 0595 §1, issue 1966).
|
||||
bound := boundOf(p.Module, p.Bound)
|
||||
bound := boundOn(p.Module, p.Node, p.Bound)
|
||||
kept := stateAccess{Module: p.Module, Node: p.Node, Keeps: append(slices.Clone(p.State), bound.States...), PerMachine: p.PerMachine,
|
||||
ReadOnly: p.ReadOnly, Reads: p.Reads, KeyedReads: p.KeyedReads, Nodes: p.Nodes}
|
||||
pub = append(pub, stateGrants(kept)...)
|
||||
@@ -1194,7 +1215,7 @@ func permissionsFor(p Principal) (Permissions, error) {
|
||||
// each assignment is issued, as it keeps each module's events under that module's own name.
|
||||
for _, d := range p.Carries {
|
||||
// Its bus state asked in `requires` as its old sections' (boundOf, ADR 0595 §1, §9, issue 1966).
|
||||
bound := boundOf(d.Module, d.Bindings)
|
||||
bound := boundOn(d.Module, p.Node, d.Bindings)
|
||||
pub = append(pub, stateGrants(stateAccess{Module: d.Module, Node: p.Node, Keeps: append(stateNames(d.State), bound.States...),
|
||||
PerMachine: perMachineNames(d.State), ReadOnly: readOnlyNames(d.State), Reads: d.Reads, KeyedReads: d.KeyedReads, Nodes: p.Nodes})...)
|
||||
pub = append(pub, objectGrants(append(objectNames(d.Objects), bound.Objects...), d.ReadsObjects)...)
|
||||
@@ -1381,6 +1402,27 @@ func seatSubject(s Seat, kind, verb string) string {
|
||||
return "mesh.seat." + s.Name + "." + kind + "." + verb
|
||||
}
|
||||
|
||||
// callerOf says whether a verb's effective callers (Seat.VerbCallers) name the module principal p: `invokers` where its
|
||||
// manifest invokes the verb (`seat:<seat>.<verb>`, or every tool by `*`), `module:<module>` naming it, or `seat:<seat>`
|
||||
// naming a seat it holds (design 47 G §336, §610, A-179). `operator`, `persons` and `sessions` name no module.
|
||||
func callerOf(p Principal, s Seat, verb string) bool {
|
||||
for _, c := range s.VerbCallers[verb] {
|
||||
switch {
|
||||
case c == seatdef.CallerInvokers:
|
||||
if slices.Contains(p.Invokes, "*") || slices.Contains(p.Invokes, "seat:"+s.Name+"."+verb) {
|
||||
return true
|
||||
}
|
||||
case c == seatdef.CallerModulePrefix+p.Module:
|
||||
return true
|
||||
}
|
||||
if seat, ok := strings.CutPrefix(c, seatdef.CallerSeatPrefix); ok &&
|
||||
slices.ContainsFunc(p.Holds, func(h Seat) bool { return h.Name == seat }) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// seatToolSubject is where a role's tool is asked. Mesh-wide for a mesh-scoped seat; a node-scoped
|
||||
// seat carries the node it is asked of, because a flat subject would reach every machine's holder
|
||||
// and the queue group would silently pick a winner (novox/hq ADR 0132, design 33 §4). A holder
|
||||
|
||||
@@ -0,0 +1,67 @@
|
||||
package buscompose
|
||||
|
||||
import (
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/meshrecovery"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// **The recovery module's account** (design 47 G §554, §565; ADR 0672 §4): the account of the module holding the seat mesh-recovery
|
||||
// (the seat, never a module's name) is granted the bus's stream snapshot and restore subjects with their chunk subjects, a stream's
|
||||
// deletion and a durable consumer's creation, only for the bus names of kept data: those of every kept data of every catalog graph
|
||||
// entry and of every entry a restore point names (an old section's name before a form change, a template's binding after), the
|
||||
// second read by the grants part through `kept-points`. The subjects its seat says, its hand-act log write, its consumes of
|
||||
// `node-backup` and its calls come from its manifest as any module's. No stream of the mesh's own is ever granted.
|
||||
|
||||
// RecoverySeat is the seat whose holder's account is granted the bus names of kept data.
|
||||
const RecoverySeat = meshrecovery.Seat
|
||||
|
||||
// meshStreams are the streams of the mesh's own, never granted to the recovery module's account (§554); every `SEAT_` stream and
|
||||
// every `KV_` bucket of a seat (KV_SEAT_…) are refused by their prefix.
|
||||
var meshStreams = []string{"CONTROL", "NODES", "EVENTS", "DEAD_LETTER_NOTICES", "DEAD_LETTERS", "PROV_APPLIED", "CHECKS_READY", "CHECKS_OTHER"}
|
||||
|
||||
// RecoveryNameRefused says a bus name is one of the mesh's own streams, or no plain subject token.
|
||||
func RecoveryNameRefused(name string) bool {
|
||||
return !safeSubject.MatchString(name) || strings.Contains(name, ".") || slices.Contains(meshStreams, name) ||
|
||||
strings.HasPrefix(name, "SEAT_") || strings.HasPrefix(name, "KV_SEAT_") || name == "KV_CONTROL" || name == "KV_NODES"
|
||||
}
|
||||
|
||||
// RecoveryGrants are what the account publishes for each bus name of kept data: the stream's snapshot and its chunks'
|
||||
// acknowledgements, its restore and the chunks sent to it, its deletion, and the creation of a durable consumer on it.
|
||||
func RecoveryGrants(names []string) []string {
|
||||
var out []string
|
||||
for _, n := range names {
|
||||
if RecoveryNameRefused(n) {
|
||||
continue
|
||||
}
|
||||
out = append(out, "$JS.API.STREAM.SNAPSHOT."+n, "$JS.SNAPSHOT.ACK."+n+".>", "$JS.API.STREAM.RESTORE."+n,
|
||||
"$JS.SNAPSHOT.RESTORE."+n+".>", "$JS.API.STREAM.DELETE."+n, "$JS.API.CONSUMER.DURABLE.CREATE."+n+".*")
|
||||
}
|
||||
return unique(out)
|
||||
}
|
||||
|
||||
// KeptBusNames are the bus names of every kept data a manifest declares: the old sections' (a state's `KV_<module>_<name>`, a log's,
|
||||
// a stream's, an object store's) and each entry of `requires` by `<module>_<local>` (BusStateStreams). Sorted, once each. Pure.
|
||||
func KeptBusNames(m catalogue.Manifest) []string {
|
||||
var out []string
|
||||
for _, s := range m.State {
|
||||
out = append(out, "KV_"+m.Module+"_"+s.Name)
|
||||
}
|
||||
for _, l := range m.Logs {
|
||||
out = append(out, "LOG_"+m.Module+"_"+l.Name)
|
||||
}
|
||||
for _, s := range m.Streams {
|
||||
out = append(out, s.BusName())
|
||||
}
|
||||
for _, o := range m.Objects {
|
||||
out = append(out, "OBJ_"+catalogue.ObjectsBucket(m.Module, o.Name))
|
||||
}
|
||||
for _, e := range m.BusState {
|
||||
out = append(out, BusStateStreams(m.Module, e)...)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return slices.Compact(out)
|
||||
}
|
||||
@@ -47,3 +47,40 @@ func TestTheNodeEngineIsGrantedWhatItNeedsOfTheRecoverySeat(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Design 47 G §554: the account of the module holding the seat mesh-recovery is granted the snapshot, the restore with its
|
||||
// chunks, the deletion and a durable consumer's creation for the bus names of kept data, and no stream of the mesh's own; no other
|
||||
// module is granted them.
|
||||
func TestTheRecoveryAccountIsGrantedTheBusNamesOfKeptDataAlone(t *testing.T) {
|
||||
names := []string{"KV_a_x", "OBJ_b_y", "EVENTS", "SEAT_node-build-go", "KV_SEAT_x", "CONTROL", "bad.name"}
|
||||
records := Records{Nodes: []string{"one"}, RecoveryNames: names,
|
||||
Assigned: map[string][]Declared{"one": {{Module: "recovery", HoldsRecovery: true}, {Module: "plain"}}}}
|
||||
users, err := Users(records)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
perms := map[string]Permissions{}
|
||||
for _, u := range users {
|
||||
if u.Kind == KindModule {
|
||||
if perms[u.Module], err = PermissionsFor(u); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, n := range []string{"KV_a_x", "OBJ_b_y"} {
|
||||
for _, want := range []string{"$JS.API.STREAM.SNAPSHOT." + n, "$JS.SNAPSHOT.ACK." + n + ".abc", "$JS.API.STREAM.RESTORE." + n,
|
||||
"$JS.SNAPSHOT.RESTORE." + n + ".abc", "$JS.API.STREAM.DELETE." + n, "$JS.API.CONSUMER.DURABLE.CREATE." + n + ".keeper"} {
|
||||
if !MayPublish(perms["recovery"], want) {
|
||||
t.Errorf("the recovery account may not publish %s", want)
|
||||
}
|
||||
if MayPublish(perms["plain"], want) {
|
||||
t.Errorf("a module that holds no recovery seat may publish %s", want)
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, n := range []string{"EVENTS", "SEAT_node-build-go", "KV_SEAT_x", "CONTROL"} {
|
||||
if MayPublish(perms["recovery"], "$JS.API.STREAM.DELETE."+n) || MayPublish(perms["recovery"], "$JS.API.STREAM.SNAPSHOT."+n) {
|
||||
t.Errorf("the recovery account may touch %s, a stream of the mesh's own", n)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,6 +2,8 @@ package buscompose
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"maps"
|
||||
"slices"
|
||||
"sort"
|
||||
)
|
||||
|
||||
@@ -41,6 +43,61 @@ func NamedWorkerName(seat, queue string) string {
|
||||
return "SEAT_" + upperSnake(seat) + "_" + upperSnake(queue) + "_named_worker"
|
||||
}
|
||||
|
||||
// NamedAsksSubject is the subject of every named ask of a build seat's queue, `accept.<queue>.*` (design 47 G §544, §600).
|
||||
func NamedAsksSubject(seat, queue string) string {
|
||||
return "mesh.seat." + seat + ".accept." + queue + ".*"
|
||||
}
|
||||
|
||||
// NamedQueue is one queue of a build seat: the stream its asks are on, the subject of its named asks and its named worker.
|
||||
type NamedQueue struct {
|
||||
Seat, Queue, Stream, Subject, Worker string
|
||||
}
|
||||
|
||||
// NamedQueuesOf are the queues of every pulled-by-kind seat a composed user holds, once each, sorted: what the grants part
|
||||
// purges at the first revision in which no entry invokes a queue of a build seat (design 47 G §544). Pure.
|
||||
func NamedQueuesOf(users []Principal) []NamedQueue {
|
||||
seen := map[string]NamedQueue{}
|
||||
add := func(holds []Seat) {
|
||||
for _, s := range holds {
|
||||
if !s.PulledByKind || !safeSubject.MatchString(s.Name) {
|
||||
continue
|
||||
}
|
||||
for _, q := range s.Accepts {
|
||||
if safeSubject.MatchString(q) {
|
||||
seen[s.Name+"."+q] = NamedQueue{Seat: s.Name, Queue: q, Stream: seatStreamName(s.Name),
|
||||
Subject: NamedAsksSubject(s.Name, q), Worker: NamedWorkerName(s.Name, q)}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, p := range users {
|
||||
switch p.Kind {
|
||||
case KindModule:
|
||||
add(p.Holds)
|
||||
case KindNodeTools:
|
||||
for _, d := range p.Carries {
|
||||
add(d.Holds)
|
||||
}
|
||||
}
|
||||
}
|
||||
out := make([]NamedQueue, 0, len(seen))
|
||||
for _, k := range slices.Sorted(maps.Keys(seen)) {
|
||||
out = append(out, seen[k])
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// NamedPurgeGrants are the subjects the purge of each queue's named asks and the delete of its named worker need: the stream's
|
||||
// purge (whose subject filter is the request's, so the grant is the stream's) and the one consumer's delete and info.
|
||||
func NamedPurgeGrants(queues []NamedQueue) []string {
|
||||
var out []string
|
||||
for _, q := range queues {
|
||||
out = append(out, "$JS.API.STREAM.PURGE."+q.Stream, "$JS.API.CONSUMER.DELETE."+q.Stream+"."+q.Worker,
|
||||
"$JS.API.CONSUMER.INFO."+q.Stream+"."+q.Worker)
|
||||
}
|
||||
return unique(out)
|
||||
}
|
||||
|
||||
// MainAskers are the seats that may publish a build seat's ask on main's subject `accept.<queue>`, which names no asker: the
|
||||
// checker and the builder, until main's subject goes (design 47 G §543, §600, A-184).
|
||||
var MainAskers = []string{"mesh-delivery-checker", "mesh-builds"}
|
||||
@@ -96,9 +153,12 @@ func SeatTrafficOf(module string, holds, uses, watches []Seat) SeatTraffic {
|
||||
for _, a := range s.Accepts {
|
||||
t.Workers = append(t.Workers, Worker{Stream: seatStreamName(s.Name), Consumer: WorkerName(s.Name, a),
|
||||
Filter: "mesh.seat." + s.Name + ".accept." + a})
|
||||
// Its named asks have a worker of their own (§544); a holder of G pulls it, one before G knows only main's.
|
||||
t.Workers = append(t.Workers, Worker{Stream: seatStreamName(s.Name), Consumer: NamedWorkerName(s.Name, a),
|
||||
Filter: "mesh.seat." + s.Name + ".accept." + a + ".*"})
|
||||
// Its named asks have a worker of their own (§544); a holder of G pulls it, one before G knows only main's. None
|
||||
// while no entry invokes a queue of a build seat, whose named asks the grants part purges.
|
||||
if !s.NoNamedAsks {
|
||||
t.Workers = append(t.Workers, Worker{Stream: seatStreamName(s.Name), Consumer: NamedWorkerName(s.Name, a),
|
||||
Filter: NamedAsksSubject(s.Name, a)})
|
||||
}
|
||||
}
|
||||
} else if len(s.Accepts) > 0 {
|
||||
stream := seatStreamName(s.Name)
|
||||
|
||||
@@ -36,3 +36,23 @@ func TestStateObjectsAndSentStores(t *testing.T) {
|
||||
t.Error("a reader of a sent store reads that one and writes none")
|
||||
}
|
||||
}
|
||||
|
||||
// A part's finding stream is published on its own node's subjects alone (design 47 G §529): the machine is the subject's last
|
||||
// token, and another machine's is refused.
|
||||
func TestFindingEventsOnOwnNodeOnly(t *testing.T) {
|
||||
bound := []membership.BusBinding{{Local: "finding-events", Storage: catalogue.StorageStream, Template: catalogue.TemplateEvents,
|
||||
Stream: "shop_finding-events", Subject: "mesh.bus.shop.finding-events"}}
|
||||
perms, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "shop", Bound: bound})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for subject, want := range map[string]bool{
|
||||
"mesh.bus.shop.finding-events.finding-written.one": true,
|
||||
"mesh.bus.shop.finding-events.finding-written.two": false,
|
||||
"mesh.bus.shop.finding-events.findings-stated.two": false,
|
||||
} {
|
||||
if got := MayPublish(perms, subject); got != want {
|
||||
t.Errorf("may publish %s: %v, want %v", subject, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -91,6 +91,8 @@ type Declared struct {
|
||||
// SnapshotsTheBus says the module holds mesh-broker — it is the bus — and so is the one module
|
||||
// granted the snapshot API, to copy the bus's streams for the night's backup (novox/hq ADR 0235).
|
||||
SnapshotsTheBus bool
|
||||
// HoldsRecovery says the module claims the seat mesh-recovery: its account is granted the bus names of kept data (recovery.go).
|
||||
HoldsRecovery bool
|
||||
// Checks are the module's own tools its health asks, each `<module>.<tool>` (novox/hq ADR 0240, to-be
|
||||
// 48 §3): the machine's node-engine asks them of its own node tools, and is granted that and no more.
|
||||
Checks []string
|
||||
@@ -158,6 +160,8 @@ type Records struct {
|
||||
// node-engine last said running, by `<node>.<module>` (design 46 approval mark §6, design 47 G §309).
|
||||
Readings map[string]RunnerReading
|
||||
Running map[string]string
|
||||
// RecoveryNames are the bus names of kept data the holder of the seat mesh-recovery is granted (recovery.go, design 47 G §554, §565).
|
||||
RecoveryNames []string
|
||||
// Kept is every seat event kept in a stream of its own (kept.go): the grants part makes each stream, so its principal is
|
||||
// granted the create of each by name.
|
||||
Kept []KeptEvent
|
||||
@@ -335,6 +339,9 @@ func Users(r Records) ([]Principal, error) {
|
||||
Logs: logNames(d.Logs), Streams: d.Streams, Hears: ComposedHears(d), Queues: d.Queues,
|
||||
Objects: objectNames(d.Objects), ReadsObjects: d.ReadsObjects, Bound: d.Bindings, AnswersSeat: d.ServesOwn,
|
||||
})
|
||||
if d.HoldsRecovery {
|
||||
out[len(out)-1].Recovers = r.RecoveryNames
|
||||
}
|
||||
// **The grants part creates what a manifest declares** (novox/hq ADR 0547 §5; the operator, 2026-10-26
|
||||
// 08:54): every placed module's buckets, logs and streams, each by its stream's name.
|
||||
if d.Module == WatchmanModule {
|
||||
@@ -389,6 +396,7 @@ func Users(r Records) ([]Principal, error) {
|
||||
// The grants part creates the durable consumers the accounts need (novox/hq ADR 0561 §6), by stream and name.
|
||||
if grants >= 0 {
|
||||
out[grants].CreatesConsumers = AccountConsumers(out)
|
||||
out[grants].PurgesNamed = NamedQueuesOf(out)
|
||||
}
|
||||
|
||||
// Refused here rather than discovered by the server. Two users with one name is a file the
|
||||
|
||||
@@ -38,7 +38,7 @@ type WriterRow struct {
|
||||
// that only adds subjects, never a delete (novox/hq ADR 0547 §5; the operator, 2026-10-26 08:54, 2026-11-05); or the bus module's tools deleting one of
|
||||
// the ended controller's leftover streams (ADR 0590 clause 6).
|
||||
func createsDeclared(p Principal, pattern string) bool {
|
||||
if removesLeftover(p, pattern) || keepsBusState(p, pattern) {
|
||||
if removesLeftover(p, pattern) || keepsBusState(p, pattern) || recoversKept(p, pattern) {
|
||||
return true
|
||||
}
|
||||
name, ok := strings.CutPrefix(pattern, "$JS.API.STREAM.CREATE.")
|
||||
@@ -54,6 +54,21 @@ func createsDeclared(p Principal, pattern string) bool {
|
||||
safeSubject.MatchString(name)
|
||||
}
|
||||
|
||||
// recoversKept is the account of the holder of the seat mesh-recovery deleting a stream of a kept data, or creating a durable
|
||||
// consumer on one, by its name alone (design 47 G §554, ADR 0672 §4: this changes ADR 0436 §3 and §5 for this account alone).
|
||||
func recoversKept(p Principal, pattern string) bool {
|
||||
if p.Kind != KindModule || len(p.Recovers) == 0 {
|
||||
return false
|
||||
}
|
||||
for _, prefix := range []string{"$JS.API.STREAM.DELETE.", "$JS.API.CONSUMER.DURABLE.CREATE."} {
|
||||
if rest, ok := strings.CutPrefix(pattern, prefix); ok {
|
||||
name, _, _ := strings.Cut(rest, ".")
|
||||
return slices.Contains(p.Recovers, name) && !RecoveryNameRefused(name)
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// ownMachine is a node writing a subject whose third token is its own name, and no wildcard there.
|
||||
func ownMachine(p Principal, pattern string) bool {
|
||||
tokens := strings.Split(pattern, ".")
|
||||
@@ -154,7 +169,7 @@ var WritersTable = []WriterRow{
|
||||
Subjects: []string{"$JS.API.STREAM.CREATE.>", "$JS.API.STREAM.UPDATE.>", "$JS.API.STREAM.DELETE.>",
|
||||
"$JS.API.CONSUMER.DURABLE.CREATE.>"},
|
||||
Writes: createsDeclared,
|
||||
Shared: "the grants part creates a stream a manifest declares, by its name, adds to a held one the subjects its manifest names, and never removes a subject or deletes a stream (novox/hq ADR 0547 §5; the operator, 2026-10-26 08:54, 2026-11-05); the bus module's tools account deletes the ended controller's leftover streams, each by its name, and no other (novox/hq ADR 0590 clause 6); the bus module's process creates, updates and deletes every module's bus state, each stream by its name (novox/hq ADR 0595 §4)"},
|
||||
Shared: "the grants part creates a stream a manifest declares, by its name, adds to a held one the subjects its manifest names, and never removes a subject or deletes a stream (novox/hq ADR 0547 §5; the operator, 2026-10-26 08:54, 2026-11-05); the bus module's tools account deletes the ended controller's leftover streams, each by its name, and no other (novox/hq ADR 0590 clause 6); the bus module's process creates, updates and deletes every module's bus state, each stream by its name (novox/hq ADR 0595 §4); the account of the holder of the seat mesh-recovery deletes the stream of a kept data and creates a durable consumer on it, each by its name (design 47 G §554, ADR 0672 §4)"},
|
||||
{State: "builds and their outcomes", Writer: "the build seat's holder", KeptIn: "its own state",
|
||||
Others: "the builder asks",
|
||||
Subjects: []string{"mesh.seat.mesh-build-machine.event.built"},
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
package grants
|
||||
|
||||
import (
|
||||
"slices"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// A `*` emitter before a finding event reaches both forms of a part's finding stream (design 47 G §386, §529): the old stream a
|
||||
// foundation part declares and the template stream another part asks for as `finding-events`.
|
||||
func TestTheKeepersConsumerOnEitherFindingForm(t *testing.T) {
|
||||
events := []string{"finding-written", "finding-cleared", "findings-stated"}
|
||||
declared := map[string]catalogue.Manifest{
|
||||
"old-part": {Module: "old-part", Streams: []catalogue.StreamDeclaration{{Name: "old-part-findings", Events: events}}},
|
||||
"new-part": {Module: "new-part", BusState: []catalogue.BusStateEntry{
|
||||
{Name: "findings", Storage: catalogue.StorageKeyValue, Template: catalogue.TemplateState},
|
||||
{Name: "finding-events", Storage: catalogue.StorageStream, Template: catalogue.TemplateEvents, Events: events}}},
|
||||
"quiet": {Module: "quiet", BusState: []catalogue.BusStateEntry{{Name: "other", Storage: catalogue.StorageStream, Template: catalogue.TemplateEvents,
|
||||
Events: []string{"tick.*"}}}},
|
||||
}
|
||||
keeper := catalogue.Manifest{Module: "keeper", Consumes: []string{"*.finding-written"}}
|
||||
heard, touches, _ := hearsOf(keeper, declared, "one", nil, keeper.Consumes, func(e string) bool { return e == "finding-written" })
|
||||
got := map[string][]string{}
|
||||
for _, h := range heard {
|
||||
got[h.Emitter] = h.Filters
|
||||
}
|
||||
if !touches || len(got) != 2 || !slices.Equal(got["new-part"], []string{"mesh.bus.new-part.finding-events.finding-written.*"}) || len(got["old-part"]) != 1 {
|
||||
t.Errorf("heard %+v", got)
|
||||
}
|
||||
if heard[len(heard)-1].Stream != "new-part_finding-events" {
|
||||
t.Errorf("the template stream's bus name: %+v", heard)
|
||||
}
|
||||
}
|
||||
@@ -401,6 +401,8 @@ type FirstViewsState struct {
|
||||
// Bindings is the first view of the bindings, asked of the seat mesh-broker's list, beside the others (design 25
|
||||
// bindings-first-view §6); absent where the part's process holds none.
|
||||
Bindings *BindingsStatus `json:"bindings,omitempty"`
|
||||
// KeptPoints is the kept call of the seat mesh-recovery's kept-points, `waiting` until it first answers after a start (design 47 G §565).
|
||||
KeptPoints *KeptPointsStatus `json:"kept-points,omitempty"`
|
||||
// Streams are the whole reads of streams kept last per subject at the part's start, `waiting` until each ends (design 47 G §556).
|
||||
Streams []StreamRead `json:"streams,omitempty"`
|
||||
// Catalog is the first view of each branch of the catalog half the part follows (design 47 G §614), and CatalogEntries the entries
|
||||
@@ -422,7 +424,7 @@ type CatalogEntry struct {
|
||||
// sayFirstViews keeps the first views' statuses in the part's own state (for grants_first_views) when they change, and
|
||||
// raises or clears FirstViewEmpty for each answered at revision 0. Each pass calls it; nil Views is none.
|
||||
func (p *Part) sayFirstViews(res *Result) {
|
||||
if p.Views == nil && p.Bindings == nil && len(p.StreamReads()) == 0 {
|
||||
if p.Views == nil && p.Bindings == nil && p.KeptPoints == nil && len(p.StreamReads()) == 0 {
|
||||
return
|
||||
}
|
||||
state := FirstViewsState{V: 1, Views: []views.Status{}}
|
||||
@@ -438,6 +440,11 @@ func (p *Part) sayFirstViews(res *Result) {
|
||||
b.Asked, b.Answered = b.Asked.Truncate(time.Second), b.Answered.Truncate(time.Second)
|
||||
state.Bindings = &b
|
||||
}
|
||||
if p.KeptPoints != nil {
|
||||
k := p.KeptPoints.Status()
|
||||
k.Asked, k.Answered = k.Asked.Truncate(time.Second), k.Answered.Truncate(time.Second)
|
||||
state.KeptPoints = &k
|
||||
}
|
||||
state.Streams = p.StreamReads()
|
||||
if p.Views != nil && p.Views.Catalog != nil {
|
||||
state.Catalog, state.CatalogWaiting = p.Views.Catalog.Statuses(), p.Views.Catalog.Waiting()
|
||||
|
||||
@@ -144,6 +144,11 @@ func (f *Follower) open() {
|
||||
f.Wake()
|
||||
return
|
||||
}
|
||||
// The recovery module's points changed: the kept call asks again (keptpoints.go).
|
||||
if f.Part.KeptPoints != nil && IsKeptPointsEvent(event, body) {
|
||||
f.Part.KeptPoints.Heard(event, body)
|
||||
return
|
||||
}
|
||||
// An account the vault minted.
|
||||
if IsAccountIssued(event) {
|
||||
f.Wake()
|
||||
|
||||
@@ -73,7 +73,7 @@ const (
|
||||
)
|
||||
|
||||
// Invokes is every tool the part calls.
|
||||
var Invokes = append([]string{CallUsers, CallUserCan, CallStreams, CallRootFree, CallHandAct, CallSeatDefinitions, CallBrokerList}, append(catalogHalfVerbs(), firstViewTools()...)...)
|
||||
var Invokes = append([]string{CallUsers, CallUserCan, CallStreams, CallRootFree, CallHandAct, CallSeatDefinitions, CallBrokerList, CallKeptPoints}, append(catalogHalfVerbs(), firstViewTools()...)...)
|
||||
|
||||
// CallSeatDefinitions reads the seat list (mesh-sdk seatlist): the verb of the seat mesh-catalog.
|
||||
const CallSeatDefinitions = "seat:mesh-catalog.seat-definitions"
|
||||
@@ -222,7 +222,10 @@ type Inputs struct {
|
||||
// build of it): what the seats it defines, the streams others hear from it and its reads are composed from while its
|
||||
// trunk manifest is refused or waits, so no other principal loses a grant to it (ADR 0601 clause 3).
|
||||
Held map[string]catalogue.Manifest
|
||||
Gaps []Gap
|
||||
// KeptNames are the bus names the restore points and asides of the recovery module name, as last read from `kept-points` (design 47 G
|
||||
// §565, A-230), kept in the part's own state: set by the pass, never read from another part.
|
||||
KeptNames []string
|
||||
Gaps []Gap
|
||||
}
|
||||
|
||||
// HeldHash is one key of the vault's bus-accounts: the hash, and what it is of. The hash is never said by the part, and
|
||||
|
||||
@@ -177,3 +177,13 @@ func withAcceptCallers(askers map[string]map[string][]string, d *buscompose.Decl
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// withVerbCallers gives each seat a declaration holds, uses or watches the effective callers of its verbs
|
||||
// (Seat.VerbCallers, design 47 G §610, A-179), as withAcceptCallers gives the askers of its queues.
|
||||
func withVerbCallers(callers map[string]map[string][]string, d *buscompose.Declared) {
|
||||
for _, list := range []*[]buscompose.Seat{&d.Holds, &d.Uses, &d.Watches} {
|
||||
for i := range *list {
|
||||
(*list)[i].VerbCallers = callers[(*list)[i].Name]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,315 @@
|
||||
package grants
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/meshrecovery"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"mesh-grants/internal/buscompose"
|
||||
)
|
||||
|
||||
// **The recovery module's restore points' names, read through `kept-points`** (design 47 G §565, A-230). The bus names of the
|
||||
// recovery module's account include those a restore point (or an aside) names. The part reads them by its `invokes` of
|
||||
// `seat:mesh-recovery.kept-points`, a kept call (ADR 0668 §1: made again until it answers), at its start and again on every `ended`
|
||||
// with outcome `done`, every `point-removed` and every `expiry-ended` of the seat, which it consumes. Until the call first answers
|
||||
// after a start, the part grants the names it last granted from it, kept in its state memberships (KeptNamesKey): it takes no name
|
||||
// away before that answer, and grants_first_views says the wait.
|
||||
|
||||
// KeptPointsVerb is the seat's verb as the manifest's invokes names it.
|
||||
const (
|
||||
CallKeptPoints = "seat:" + meshrecovery.Seat + "." + meshrecovery.VerbKeptPoints
|
||||
// KeptNamesKey is the key of the part's own state memberships that keeps the names last read from `kept-points`; never a
|
||||
// `<node>.<module>` key, since no machine is named `kept-names`.
|
||||
KeptNamesKey = "kept-names." + meshrecovery.Seat
|
||||
)
|
||||
|
||||
// KeptNames is the value of KeptNamesKey: every bus name a restore point or an aside of the recovery module names.
|
||||
type KeptNames struct {
|
||||
V int `json:"v"`
|
||||
Names []string `json:"names"`
|
||||
At time.Time `json:"at,omitzero"`
|
||||
}
|
||||
|
||||
// keptPointsAnswer is the part of kept-points' answer the part reads: `points[].names` and `asides[].names` (A-230). The contract's
|
||||
// Point gains `names` at the join, as its Aside has it; until then this is the wire's form.
|
||||
type keptPointsAnswer struct {
|
||||
Points []struct {
|
||||
Names []string `json:"names"`
|
||||
} `json:"points"`
|
||||
Asides []struct {
|
||||
Names []string `json:"names"`
|
||||
} `json:"asides"`
|
||||
}
|
||||
|
||||
// KeptPointsStatus is what grants_first_views says of the kept call.
|
||||
type KeptPointsStatus struct {
|
||||
Seat string `json:"seat"`
|
||||
Verb string `json:"verb"`
|
||||
Asked time.Time `json:"asked,omitzero"`
|
||||
Answered time.Time `json:"answered,omitzero"`
|
||||
Asks int `json:"asks"`
|
||||
// Names is how many names the last answer named (or, while waiting, the names kept from the last answer before the start).
|
||||
Names int `json:"names"`
|
||||
// Waiting says the call has not answered since the start: the part grants the names it kept.
|
||||
Waiting bool `json:"waiting"`
|
||||
Why string `json:"why,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// KeptPointsView is the kept call of `kept-points`.
|
||||
type KeptPointsView struct {
|
||||
// Ask calls the seat's verb and answers its reply (the process's connection).
|
||||
Ask func(ctx context.Context) (json.RawMessage, error)
|
||||
Changed func()
|
||||
Logf func(string, ...any)
|
||||
Now func() time.Time
|
||||
// Pause waits d between two asks that failed and answers false once ctx ended; nil waits on the context's bound.
|
||||
Pause func(ctx context.Context, d time.Duration) bool
|
||||
// Wait is the first wait between two asks that failed (ten seconds when zero), doubling to MaxWait (five minutes when zero).
|
||||
Wait time.Duration
|
||||
MaxWait time.Duration
|
||||
|
||||
mu sync.Mutex
|
||||
ctx context.Context
|
||||
part *Part
|
||||
st KeptPointsStatus
|
||||
asking bool
|
||||
again bool
|
||||
}
|
||||
|
||||
func (v *KeptPointsView) now() time.Time {
|
||||
if v.Now != nil {
|
||||
return v.Now().UTC()
|
||||
}
|
||||
return time.Now().UTC()
|
||||
}
|
||||
|
||||
func (v *KeptPointsView) logf(format string, args ...any) {
|
||||
if v.Logf != nil {
|
||||
v.Logf(format, args...)
|
||||
}
|
||||
}
|
||||
|
||||
// Status is the kept call's status now.
|
||||
func (v *KeptPointsView) Status() KeptPointsStatus {
|
||||
v.mu.Lock()
|
||||
defer v.mu.Unlock()
|
||||
s := v.st
|
||||
s.Seat, s.Verb = meshrecovery.Seat, meshrecovery.VerbKeptPoints
|
||||
return s
|
||||
}
|
||||
|
||||
// Start asks at the part's start, on a goroutine of its own, and until it answers.
|
||||
func (v *KeptPointsView) Start(ctx context.Context, p *Part) {
|
||||
v.mu.Lock()
|
||||
v.ctx, v.part = ctx, p
|
||||
v.st.Waiting = true
|
||||
v.mu.Unlock()
|
||||
if kept, err := p.keptNames(); err == nil {
|
||||
v.mu.Lock()
|
||||
v.st.Names = len(kept)
|
||||
v.mu.Unlock()
|
||||
}
|
||||
go v.askUntil(ctx, p, "start")
|
||||
}
|
||||
|
||||
// Heard asks again when an event heard is one that changes the points: `ended` with outcome `done`, `point-removed`, `expiry-ended`
|
||||
// of the seat mesh-recovery, named `mesh-recovery.<event>` or by its bus subject.
|
||||
func (v *KeptPointsView) Heard(event string, body []byte) {
|
||||
name, ok := recoveryEvent(event)
|
||||
if !ok || !changesPoints(name, body) {
|
||||
return
|
||||
}
|
||||
v.mu.Lock()
|
||||
ctx, p := v.ctx, v.part
|
||||
v.mu.Unlock()
|
||||
if p == nil {
|
||||
return // not started: the start's ask reads what the event changed
|
||||
}
|
||||
go v.askUntil(ctx, p, "the seat's "+name)
|
||||
}
|
||||
|
||||
func recoveryEvent(event string) (string, bool) {
|
||||
for _, prefix := range []string{meshrecovery.Seat + ".", "mesh.seat." + meshrecovery.Seat + ".event."} {
|
||||
if name, ok := strings.CutPrefix(event, prefix); ok {
|
||||
return name, true
|
||||
}
|
||||
}
|
||||
return "", false
|
||||
}
|
||||
|
||||
func changesPoints(name string, body []byte) bool {
|
||||
switch name {
|
||||
case meshrecovery.EventPointRemoved, meshrecovery.EventExpiryEnded:
|
||||
return true
|
||||
case meshrecovery.EventEnded:
|
||||
var e meshrecovery.EndedEvent
|
||||
return json.Unmarshal(body, &e) != nil || e.Outcome == meshrecovery.EndedEventOutcomeDone
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// IsKeptPointsEvent says an event heard is one the kept call asks again on.
|
||||
func IsKeptPointsEvent(event string, body []byte) bool {
|
||||
name, ok := recoveryEvent(event)
|
||||
return ok && changesPoints(name, body)
|
||||
}
|
||||
|
||||
func (v *KeptPointsView) pause(ctx context.Context, d time.Duration) bool {
|
||||
if v.Pause != nil {
|
||||
return v.Pause(ctx, d)
|
||||
}
|
||||
t, cancel := context.WithTimeout(ctx, d)
|
||||
<-t.Done()
|
||||
cancel()
|
||||
return ctx.Err() == nil
|
||||
}
|
||||
|
||||
// askUntil asks until the seat answers or ctx ends, from ten seconds, doubling, at most every five minutes. An ask made while one
|
||||
// runs is made once more when it ends, so an event heard meanwhile is never lost.
|
||||
func (v *KeptPointsView) askUntil(ctx context.Context, p *Part, why string) {
|
||||
v.mu.Lock()
|
||||
if v.asking {
|
||||
v.again = true
|
||||
v.mu.Unlock()
|
||||
return
|
||||
}
|
||||
v.asking = true
|
||||
v.st.Why = why
|
||||
v.mu.Unlock()
|
||||
wait, most := v.Wait, v.MaxWait
|
||||
if wait == 0 {
|
||||
wait = 10 * time.Second
|
||||
}
|
||||
if most == 0 {
|
||||
most = 5 * time.Minute
|
||||
}
|
||||
for {
|
||||
v.mu.Lock()
|
||||
v.st.Asked = v.now()
|
||||
v.st.Asks++
|
||||
v.mu.Unlock()
|
||||
raw, err := v.Ask(ctx)
|
||||
var n int
|
||||
if err == nil {
|
||||
n, err = p.applyKeptPoints(raw, v.wake)
|
||||
}
|
||||
if err != nil {
|
||||
v.mu.Lock()
|
||||
v.st.Error = err.Error()
|
||||
v.mu.Unlock()
|
||||
v.logf("%s is not answered (%v): asked again in %s; the names last kept stay granted", CallKeptPoints, err, wait)
|
||||
if !v.pause(ctx, wait) {
|
||||
v.mu.Lock()
|
||||
v.asking = false
|
||||
v.mu.Unlock()
|
||||
return
|
||||
}
|
||||
if wait *= 2; wait > most {
|
||||
wait = most
|
||||
}
|
||||
continue
|
||||
}
|
||||
v.mu.Lock()
|
||||
v.st.Answered, v.st.Waiting, v.st.Error, v.st.Names = v.now(), false, "", n
|
||||
again := v.again
|
||||
v.again = false
|
||||
if !again {
|
||||
v.asking = false
|
||||
}
|
||||
v.mu.Unlock()
|
||||
v.wake()
|
||||
if !again {
|
||||
return
|
||||
}
|
||||
wait = v.Wait
|
||||
if wait == 0 {
|
||||
wait = 10 * time.Second
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (v *KeptPointsView) wake() {
|
||||
if v.Changed != nil {
|
||||
v.Changed()
|
||||
}
|
||||
}
|
||||
|
||||
// keptNames are the names last read from kept-points, from the part's own state: none before the first answer ever.
|
||||
func (p *Part) keptNames() ([]string, error) {
|
||||
raw, _, found, err := p.Store.Get(StateMemberships, KeptNamesKey)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("its own %s %s cannot be read: %w", StateMemberships, KeptNamesKey, err)
|
||||
}
|
||||
if !found {
|
||||
return nil, nil
|
||||
}
|
||||
var k KeptNames
|
||||
if err := json.Unmarshal(raw, &k); err != nil {
|
||||
return nil, fmt.Errorf("its own %s %s does not read: %w", StateMemberships, KeptNamesKey, err)
|
||||
}
|
||||
return k.Names, nil
|
||||
}
|
||||
|
||||
// applyKeptPoints replaces the kept names with those of an answer (a name no point names any more is taken away), and calls wake
|
||||
// when they changed. It answers how many names the answer named.
|
||||
func (p *Part) applyKeptPoints(raw json.RawMessage, wake func()) (int, error) {
|
||||
var a keptPointsAnswer
|
||||
if err := json.Unmarshal(raw, &a); err != nil {
|
||||
return 0, fmt.Errorf("%s answered what it does not read: %w", CallKeptPoints, err)
|
||||
}
|
||||
var names []string
|
||||
for _, pt := range a.Points {
|
||||
names = append(names, pt.Names...)
|
||||
}
|
||||
for _, as := range a.Asides {
|
||||
names = append(names, as.Names...)
|
||||
}
|
||||
slices.Sort(names)
|
||||
names = slices.Compact(names)
|
||||
was, err := p.keptNames()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if slices.Equal(was, names) {
|
||||
return len(names), nil
|
||||
}
|
||||
body, _ := json.Marshal(KeptNames{V: 1, Names: names, At: p.now()})
|
||||
if _, err := p.Store.Put(StateMemberships, KeptNamesKey, body); err != nil {
|
||||
return 0, fmt.Errorf("its own %s %s cannot be written: %w", StateMemberships, KeptNamesKey, err)
|
||||
}
|
||||
if wake != nil {
|
||||
wake()
|
||||
}
|
||||
return len(names), nil
|
||||
}
|
||||
|
||||
// withKeptNames sets the names the recovery module's account is granted for restore points on in (Inputs.KeptNames), read from
|
||||
// the part's own state in one place for the pass and for its tools.
|
||||
func (p *Part) withKeptNames(in *Inputs) error {
|
||||
names, err := p.keptNames()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
in.KeptNames = names
|
||||
return nil
|
||||
}
|
||||
|
||||
// recoveryNamesOf are the bus names of kept data of every manifest of the catalog graph the part holds, and the names a restore
|
||||
// point or an aside names (Inputs.KeptNames): a name a manifest left stays while a point names it. Sorted, once each. Pure.
|
||||
func recoveryNamesOf(in Inputs) []string {
|
||||
names := slices.Clone(in.KeptNames)
|
||||
for _, group := range []map[string]catalogue.Manifest{in.Manifests, in.Running, in.Ran, in.RanByFingerprint, in.Had, in.Held} {
|
||||
for _, m := range group {
|
||||
names = append(names, buscompose.KeptBusNames(m)...)
|
||||
}
|
||||
}
|
||||
slices.Sort(names)
|
||||
return slices.Compact(names)
|
||||
}
|
||||
@@ -0,0 +1,78 @@
|
||||
package grants
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"slices"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// Design 47 G §565, A-230: the part reads points[].names and asides[].names of kept-points, keeps them in its state memberships,
|
||||
// grants the names it kept until the call first answers after a start, takes a name away only by an answer, and asks again on
|
||||
// the seat's ended (outcome done), point-removed and expiry-ended alone.
|
||||
func TestKeptPointsNamesAreKeptUntilTheCallAnswers(t *testing.T) {
|
||||
store := memStore{}
|
||||
answer := func(s string) func(context.Context) (json.RawMessage, error) {
|
||||
return func(context.Context) (json.RawMessage, error) { return json.RawMessage(s), nil }
|
||||
}
|
||||
p := &Part{Store: store, Now: func() time.Time { return time.Unix(100, 0) }}
|
||||
v := &KeptPointsView{Ask: answer(`{"points":[{"names":["KV_a_x","KV_b_y"]}],"asides":[{"names":["OBJ_c_z"]}]}`)}
|
||||
p.KeptPoints = v
|
||||
v.askUntil(context.Background(), p, "start")
|
||||
var in Inputs
|
||||
if err := p.withKeptNames(&in); err != nil || !slices.Equal(in.KeptNames, []string{"KV_a_x", "KV_b_y", "OBJ_c_z"}) {
|
||||
t.Fatalf("names after the first answer %v (%v)", in.KeptNames, err)
|
||||
}
|
||||
// A restart: the call fails, and the names last kept stay; the status says it waits.
|
||||
p2 := &Part{Store: store}
|
||||
failing := &KeptPointsView{Ask: func(context.Context) (json.RawMessage, error) { return nil, errors.New("no responders") },
|
||||
Pause: func(context.Context, time.Duration) bool { return false }}
|
||||
p2.KeptPoints = failing
|
||||
failing.Start(context.Background(), p2)
|
||||
failing.askUntil(context.Background(), p2, "start")
|
||||
in = Inputs{}
|
||||
if err := p2.withKeptNames(&in); err != nil || len(in.KeptNames) != 3 {
|
||||
t.Fatalf("names kept before the call answers %v (%v)", in.KeptNames, err)
|
||||
}
|
||||
if s := failing.Status(); !s.Waiting || s.Names != 3 || s.Error == "" {
|
||||
t.Errorf("status %+v, want waiting with 3 names and the error", s)
|
||||
}
|
||||
// An answer takes a name away that no point names any more.
|
||||
failing.Ask = answer(`{"points":[{"names":["KV_a_x"]}],"asides":[]}`)
|
||||
failing.askUntil(context.Background(), p2, "the seat's point-removed")
|
||||
in = Inputs{}
|
||||
_ = p2.withKeptNames(&in)
|
||||
if !slices.Equal(in.KeptNames, []string{"KV_a_x"}) || failing.Status().Waiting {
|
||||
t.Errorf("names after the answer %v, status %+v", in.KeptNames, failing.Status())
|
||||
}
|
||||
// Which events ask again.
|
||||
for event, body := range map[string]string{"mesh-recovery.ended": `{"outcome":"done"}`, "mesh.seat.mesh-recovery.event.point-removed": `{}`,
|
||||
"mesh-recovery.expiry-ended": `{}`} {
|
||||
if !IsKeptPointsEvent(event, []byte(body)) {
|
||||
t.Errorf("%s does not ask again", event)
|
||||
}
|
||||
}
|
||||
for event, body := range map[string]string{"mesh-recovery.ended": `{"outcome":"failed"}`, "mesh-recovery.restored": `{"outcome":"done"}`, "mesh-broker.created": `{}`} {
|
||||
if IsKeptPointsEvent(event, []byte(body)) {
|
||||
t.Errorf("%s asks again", event)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Design 47 G §554, §565: the names the recovery account is granted are those of every manifest of the catalog graph (an old
|
||||
// section's and a template entry's) and those a point names, once each; a name a manifest left stays while a point names it.
|
||||
func TestTheRecoveryNamesAreTheCatalogGraphsAndThePointsNames(t *testing.T) {
|
||||
trunk := catalogue.Manifest{Module: "a", State: []catalogue.StateDeclaration{{Name: "x"}}}
|
||||
old := catalogue.Manifest{Module: "a", State: []catalogue.StateDeclaration{{Name: "old"}},
|
||||
BusState: []catalogue.BusStateEntry{{Name: "tpl", Storage: catalogue.StorageKeyValue}}}
|
||||
got := recoveryNamesOf(Inputs{Manifests: map[string]catalogue.Manifest{"a": trunk}, RanByFingerprint: map[string]catalogue.Manifest{"fp": old},
|
||||
KeptNames: []string{"KV_a_gone", "KV_a_x"}})
|
||||
want := []string{"KV_a_gone", "KV_a_old", "KV_a_tpl", "KV_a_x"}
|
||||
if !slices.Equal(got, want) {
|
||||
t.Errorf("names %v, want %v", got, want)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,121 @@
|
||||
package grants
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
|
||||
"mesh-grants/internal/buscompose"
|
||||
)
|
||||
|
||||
// **The purge of the build seats' named asks** (design 47 G §544, A-231): at the first revision in which no entry invokes any
|
||||
// `seat:<seat>.accept.<queue>`, the part purges the subjects `accept.<queue>.*` of each build seat's stream and deletes the
|
||||
// named workers, within its pass, saying each in its journal; where that did not run (the part stopped first), the tool
|
||||
// grants_named_purge, written in the hand-act log first, asks the part's process to do it by hand. An ask is transport, not
|
||||
// kept data.
|
||||
|
||||
// ToolNamedPurge is the hand tool that purges the named asks and deletes the named workers.
|
||||
const ToolNamedPurge = "grants_named_purge"
|
||||
|
||||
// NamedPurger purges one subject of a stream and deletes one consumer: the process's connection; nil in the tools.
|
||||
type NamedPurger interface {
|
||||
PurgeSubject(stream, subject string) error
|
||||
// DeleteConsumer deletes a consumer; one the bus does not hold is no error.
|
||||
DeleteConsumer(stream, name string) error
|
||||
}
|
||||
|
||||
// NamedAsksUnused says the revision the part composes from holds entries and none invokes a queue of a build seat
|
||||
// (catalogue.QueueInvoke); false while the catalog half is not read, so a failed read purges nothing.
|
||||
func NamedAsksUnused(in Inputs) bool {
|
||||
if in.CatalogUnread != "" || len(in.Manifests) == 0 {
|
||||
return false
|
||||
}
|
||||
for _, m := range in.Manifests {
|
||||
for _, i := range m.Invokes {
|
||||
if _, _, ok := catalogue.QueueInvoke(i); ok {
|
||||
return false
|
||||
}
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// withNoNamedAsks marks each pulled-by-kind seat a declaration holds as having no named worker while unused.
|
||||
func withNoNamedAsks(unused bool, d *buscompose.Declared) {
|
||||
for _, list := range []*[]buscompose.Seat{&d.Holds, &d.Uses, &d.Watches} {
|
||||
for i := range *list {
|
||||
if (*list)[i].PulledByKind {
|
||||
(*list)[i].NoNamedAsks = unused
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// purgeNamed purges the named asks and deletes the named workers of every build seat's queue the records compose, at the
|
||||
// first pass whose revision leaves them unused, and on a person's ask whatever the revision says.
|
||||
func (p *Part) purgeNamed(res *Result, in Inputs, recs buscompose.Records, byHand string) {
|
||||
unused := NamedAsksUnused(in)
|
||||
due := unused && !p.namedPurged || byHand != ""
|
||||
p.namedPurged = unused
|
||||
if !due || p.NamedPurge == nil {
|
||||
return
|
||||
}
|
||||
users, err := buscompose.Users(recs)
|
||||
if err != nil {
|
||||
res.Errors = append(res.Errors, "the named asks could not be purged: "+err.Error())
|
||||
return
|
||||
}
|
||||
why := "no entry of the revision invokes a queue of a build seat"
|
||||
if byHand != "" {
|
||||
why = "by hand, for " + byHand
|
||||
}
|
||||
for _, q := range buscompose.NamedQueuesOf(users) {
|
||||
if err := p.NamedPurge.PurgeSubject(q.Stream, q.Subject); err != nil {
|
||||
res.Errors = append(res.Errors, fmt.Sprintf("the named asks %s of %s were not purged: %v", q.Subject, q.Stream, err))
|
||||
p.namedPurged = false
|
||||
} else {
|
||||
res.NamedPurged = append(res.NamedPurged, q.Stream+" "+q.Subject)
|
||||
p.logf("purged the named asks %s of %s (%s)", q.Subject, q.Stream, why)
|
||||
}
|
||||
if err := p.NamedPurge.DeleteConsumer(q.Stream, q.Worker); err != nil {
|
||||
res.Errors = append(res.Errors, fmt.Sprintf("the named worker %s.%s was not deleted: %v", q.Stream, q.Worker, err))
|
||||
p.namedPurged = false
|
||||
} else {
|
||||
res.NamedPurged = append(res.NamedPurged, q.Stream+"."+q.Worker)
|
||||
p.logf("deleted the named worker %s.%s (%s)", q.Stream, q.Worker, why)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (p *Part) logf(format string, args ...any) {
|
||||
if p.Logf != nil {
|
||||
p.Logf(format, args...)
|
||||
}
|
||||
}
|
||||
|
||||
// AskNamedPurge answers the tool grants_named_purge: it records the hand act through record (the seat mesh-hand-acts) and only
|
||||
// then asks the part's process to purge the named asks and delete the named workers of every build seat's queue.
|
||||
func (p *Part) AskNamedPurge(caller, why string, record func(what, why string) error) (any, error) {
|
||||
if strings.TrimSpace(why) == "" {
|
||||
return nil, errors.New(ToolNamedPurge + " needs why: it is recorded as a hand act")
|
||||
}
|
||||
if record == nil {
|
||||
return nil, errors.New("the hand act cannot be recorded here, so nothing was asked")
|
||||
}
|
||||
if err := record(ToolNamedPurge, why); err != nil {
|
||||
return nil, fmt.Errorf("the hand act could not be recorded, so nothing was asked: %w", err)
|
||||
}
|
||||
by := caller
|
||||
if by == "" {
|
||||
by = "a person"
|
||||
}
|
||||
ask, _ := json.Marshal(Ask{By: by, At: p.now(), NamedPurge: true})
|
||||
rev, err := p.Store.Put(StateUserList, KeyAsked, ask)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the ask could not be written: %w", err)
|
||||
}
|
||||
return map[string]any{"asked": rev, "says": "the part's process purges the named asks and deletes the named workers at this ask; its journal names each"}, nil
|
||||
}
|
||||
@@ -0,0 +1,96 @@
|
||||
package grants
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"slices"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
|
||||
"mesh-grants/internal/buscompose"
|
||||
)
|
||||
|
||||
type purged struct{ acts []string }
|
||||
|
||||
func (p *purged) PurgeSubject(stream, subject string) error {
|
||||
p.acts = append(p.acts, "purge "+stream+" "+subject)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *purged) DeleteConsumer(stream, name string) error {
|
||||
p.acts = append(p.acts, "delete "+stream+"."+name)
|
||||
return nil
|
||||
}
|
||||
|
||||
type memStore map[string][]byte
|
||||
|
||||
func (m memStore) Get(state, key string) ([]byte, uint64, bool, error) {
|
||||
v, ok := m[state+"/"+key]
|
||||
return v, uint64(len(m)), ok, nil
|
||||
}
|
||||
func (m memStore) Put(state, key string, value []byte) (uint64, error) {
|
||||
m[state+"/"+key] = value
|
||||
return uint64(len(m)), nil
|
||||
}
|
||||
func (m memStore) Keys(string) ([]string, error) { return nil, nil }
|
||||
func (m memStore) Delete(state, key string) error { delete(m, state+"/"+key); return nil }
|
||||
|
||||
// Design 47 G §544, A-231: at the first revision in which no entry invokes a queue of a build seat the part purges each
|
||||
// queue's named asks and deletes its named worker, once, and again only after they were used and left again; a failed read
|
||||
// of the catalog half purges nothing; grants_named_purge records its hand act first and then has the process purge.
|
||||
func TestTheNamedAsksArePurgedAtTheFirstRevisionThatLeavesThem(t *testing.T) {
|
||||
recs := buscompose.Records{Nodes: []string{"one"}, Assigned: map[string][]buscompose.Declared{"one": {{Module: "go-module",
|
||||
Holds: []buscompose.Seat{{Name: "node-build-go", Scope: "node", Accepts: []string{"check"}, PulledByKind: true}}}}}}
|
||||
unused := Inputs{Manifests: map[string]catalogue.Manifest{"runner": {Module: "runner", Invokes: []string{"seat:node-build-go.run"}}}}
|
||||
used := Inputs{Manifests: map[string]catalogue.Manifest{"runner": {Module: "runner", Invokes: []string{"seat:node-build-go.accept.check"}}}}
|
||||
unread := unused
|
||||
unread.CatalogUnread = "the catalog half of the branch main"
|
||||
bus := &purged{}
|
||||
p := &Part{NamedPurge: bus, Store: memStore{}}
|
||||
want := []string{"purge SEAT_NODE_BUILD_GO mesh.seat.node-build-go.accept.check.*",
|
||||
"delete SEAT_NODE_BUILD_GO.SEAT_NODE_BUILD_GO_CHECK_named_worker"}
|
||||
for i, c := range []struct {
|
||||
in Inputs
|
||||
byHand string
|
||||
acts int
|
||||
}{{unread, "", 0}, {used, "", 0}, {unused, "", 2}, {unused, "", 2}, {used, "", 2}, {unused, "", 4}, {used, "jochen@shanks", 6}} {
|
||||
var res Result
|
||||
p.purgeNamed(&res, c.in, recs, c.byHand)
|
||||
if len(bus.acts) != c.acts {
|
||||
t.Fatalf("pass %d: %d acts, want %d: %v", i, len(bus.acts), c.acts, bus.acts)
|
||||
}
|
||||
}
|
||||
if !slices.Equal(bus.acts[:2], want) {
|
||||
t.Errorf("purged %v, want %v", bus.acts[:2], want)
|
||||
}
|
||||
// The named workers are composed while an entry invokes a queue, and not while none does.
|
||||
for _, c := range []struct {
|
||||
in Inputs
|
||||
named bool
|
||||
}{{used, true}, {unused, false}, {unread, true}} {
|
||||
d := buscompose.Declared{Holds: slices.Clone(recs.Assigned["one"][0].Holds)}
|
||||
withNoNamedAsks(NamedAsksUnused(c.in), &d)
|
||||
tr := buscompose.SeatTrafficOf("go-module", d.Holds, nil, nil)
|
||||
has := slices.ContainsFunc(tr.Workers, func(w buscompose.Worker) bool { return w.Consumer == "SEAT_NODE_BUILD_GO_CHECK_named_worker" })
|
||||
if has != c.named {
|
||||
t.Errorf("a named worker composed %v, want %v (%v)", has, c.named, c.in.Manifests["runner"].Invokes)
|
||||
}
|
||||
}
|
||||
// The hand tool: nothing is asked when the hand act is not recorded; once it is, the process purges at its ask.
|
||||
if _, err := p.AskNamedPurge("jochen@shanks", "the part stopped first", func(string, string) error { return errors.New("down") }); err == nil {
|
||||
t.Error("asked though the hand act was not recorded")
|
||||
}
|
||||
if a := p.asked(); a.NamedPurge {
|
||||
t.Error("an ask was written though the hand act was not recorded")
|
||||
}
|
||||
var acts []string
|
||||
if _, err := p.AskNamedPurge("jochen@shanks", "the part stopped first", func(what, why string) error {
|
||||
acts = append(acts, what+": "+why)
|
||||
return nil
|
||||
}); err != nil || len(acts) != 1 {
|
||||
t.Fatalf("the hand act: %v %v", err, acts)
|
||||
}
|
||||
if a := p.asked(); !a.NamedPurge || a.By != "jochen@shanks" {
|
||||
t.Errorf("the ask read back: %+v", a)
|
||||
}
|
||||
}
|
||||
@@ -137,6 +137,8 @@ type Result struct {
|
||||
Sealed *SealedState `json:"sealed,omitempty"`
|
||||
// Consumers is what the pass found and created of the durable consumers an account needs (ADR 0561 §6).
|
||||
Consumers *ConsumersState `json:"consumers,omitempty"`
|
||||
// NamedPurged are the build seats' named asks purged and named workers deleted in this pass (design 47 G §544).
|
||||
NamedPurged []string `json:"named_purged,omitempty"`
|
||||
// Accounts is what the pass wrote of the wanted accounts the vault mints from (ADR 0561 §2).
|
||||
Accounts *AccountsWritten `json:"accounts,omitempty"`
|
||||
// Grants is what the pass wrote of its grants (grants.go).
|
||||
@@ -173,6 +175,9 @@ type Part struct {
|
||||
// Consumers creates an account's durable consumer (consumers.go): the process's connection; nil in the tools, which
|
||||
// compare only.
|
||||
Consumers ConsumerCreator
|
||||
// NamedPurge purges the build seats' named asks and deletes their named workers (namedpurge.go): the process's
|
||||
// connection; nil in the tools.
|
||||
NamedPurge NamedPurger
|
||||
// Bus is where the nodes reach the bus, which each wanted account carries (accounts.go): the process's own credential's;
|
||||
// empty in the tools, which write no account.
|
||||
Bus BusAddress
|
||||
@@ -191,6 +196,9 @@ type Part struct {
|
||||
made creations
|
||||
// askSeen is the revision of the last ask to compose (KeyAsked) this part acted on.
|
||||
askSeen uint64
|
||||
// namedPurged says the last pass found the build seats' named asks unused and purged them, so the next purges only
|
||||
// once they are used again and then left unused (namedpurge.go).
|
||||
namedPurged bool
|
||||
// heldSaid is the reason of the held-list condition last said (empty: none raised); heldKnown is whether one was said
|
||||
// since the start, so the first pass after a start always says its state (a condition raised before a restart is cleared
|
||||
// by it). holdMu guards both.
|
||||
@@ -229,6 +237,8 @@ type Part struct {
|
||||
fvSaid firstViewsSaid
|
||||
// Bindings is the first view of the bindings, asked of the seat mesh-broker's list (bindingsview.go); nil in the tools.
|
||||
Bindings *BindingsView
|
||||
// KeptPoints is the kept call of the seat mesh-recovery's kept-points (keptpoints.go); nil in the tools.
|
||||
KeptPoints *KeptPointsView
|
||||
// bindMu orders the writes of the kept bindings: an event's and a first view's answer (bindings.go).
|
||||
bindMu sync.Mutex
|
||||
// runningRead is the whole read of the stream running at the start (running.go), guarded by mu.
|
||||
@@ -273,6 +283,9 @@ type Ask struct {
|
||||
Consumer string `json:"consumer,omitempty"`
|
||||
// FirstView is, from grants_first_view_again, one first view to ask once more by hand (its state), or AllFirstViews.
|
||||
FirstView string `json:"first-view,omitempty"`
|
||||
// NamedPurge is, from grants_named_purge, the purge of the build seats' named asks and their named workers by hand
|
||||
// (design 47 G §544).
|
||||
NamedPurge bool `json:"named-purge,omitempty"`
|
||||
}
|
||||
|
||||
// AllFirstViews asks every first view once more.
|
||||
@@ -317,6 +330,9 @@ func (p *Part) Compose() Result {
|
||||
if err := p.withBound(&in); err != nil {
|
||||
res.Errors = append(res.Errors, err.Error())
|
||||
}
|
||||
if err := p.withKeptNames(&in); err != nil {
|
||||
res.Errors = append(res.Errors, err.Error())
|
||||
}
|
||||
for _, g := range in.Refused {
|
||||
res.Refused = append(res.Refused, g.Input+": "+g.Why)
|
||||
}
|
||||
@@ -511,7 +527,7 @@ func (p *Part) Compose() Result {
|
||||
p.clearFull()
|
||||
held, heldErr := HeldStreams(p.Call)
|
||||
var ask Ask
|
||||
if p.Create != nil || p.Consumers != nil {
|
||||
if p.Create != nil || p.Consumers != nil || p.NamedPurge != nil {
|
||||
ask = p.asked()
|
||||
}
|
||||
forced := ask.Assert
|
||||
@@ -550,6 +566,11 @@ func (p *Part) Compose() Result {
|
||||
}
|
||||
}
|
||||
p.consumers(&res, recs, ask.Consumer)
|
||||
byHand := ""
|
||||
if ask.NamedPurge {
|
||||
byHand = ask.By
|
||||
}
|
||||
p.purgeNamed(&res, in, recs, byHand)
|
||||
p.sayFirstViews(&res)
|
||||
if ask.FirstView != "" && p.FirstViewContext != nil {
|
||||
state := ask.FirstView
|
||||
|
||||
@@ -60,6 +60,9 @@ func records(in Inputs, union bool) (buscompose.Records, []Gap) {
|
||||
out.BusState[module] = m.BusState
|
||||
}
|
||||
}
|
||||
// The bus names of kept data the recovery module's account is granted (design 47 G §554, §565): those of every manifest of the
|
||||
// catalog graph the part holds (the trunk's, each node's wanted, ran and had builds), and those restore points name.
|
||||
out.RecoveryNames = recoveryNamesOf(in)
|
||||
// The holders on record, after the seats whose holder has left are passed (ADR 0660): the one rule of mesh-sdk, the
|
||||
// node-engine's too, read with the manifest each node runs, else the trunk's. Composed, never recorded.
|
||||
holders, passed := holdersAfterPass(in)
|
||||
@@ -117,6 +120,8 @@ func records(in Inputs, union bool) (buscompose.Records, []Gap) {
|
||||
d.Bindings = in.Bindings[m.Module]
|
||||
withEventSubjects(in.EventSubjects, &d)
|
||||
withAcceptCallers(in.AcceptCallers, &d)
|
||||
withVerbCallers(in.VerbCallers, &d)
|
||||
withNoNamedAsks(NamedAsksUnused(in), &d)
|
||||
return d
|
||||
}
|
||||
d := declare(m)
|
||||
@@ -149,6 +154,7 @@ func records(in Inputs, union bool) (buscompose.Records, []Gap) {
|
||||
State: bucketsFor(m, in.Nodes), Logs: logsOf(m), Streams: streamsOf(m)}
|
||||
withEventSubjects(in.EventSubjects, out.Engine)
|
||||
withAcceptCallers(in.AcceptCallers, out.Engine)
|
||||
withVerbCallers(in.VerbCallers, out.Engine)
|
||||
}
|
||||
return out, gaps
|
||||
}
|
||||
@@ -184,6 +190,7 @@ func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaratio
|
||||
d.ProvisionsNode = true
|
||||
}
|
||||
d.SnapshotsTheBus = m.ClaimsSeat(meshbroker.Seat)
|
||||
d.HoldsRecovery = m.ClaimsSeat(buscompose.RecoverySeat)
|
||||
for _, c := range m.Claims {
|
||||
if s, hasAProtocol := seats[c.Name]; hasAProtocol {
|
||||
held := asSeat(s, declarers[s.Name])
|
||||
@@ -527,6 +534,33 @@ func hearsOf(m catalogue.Manifest, declared map[string]catalogue.Manifest, node
|
||||
}
|
||||
out = append(out, h)
|
||||
}
|
||||
// **Either form of a part's finding stream** (design 47 G §386, §529): a `*` emitter before an event reaches, beside each
|
||||
// module's old stream, the stream a part asked for in `requires` by the template events (its `finding-events`) that carries
|
||||
// the event, one durable consumer on it, heard on the binding's subject `mesh.bus.<module>.<local>.<event>`.
|
||||
for _, c := range reading.Consumes {
|
||||
emitter, event, named := strings.Cut(c, ".")
|
||||
if !named || emitter != catalogue.OneName || event == "" || strings.ContainsAny(event, "*>") || defined == nil || !defined(event) {
|
||||
continue
|
||||
}
|
||||
for _, module := range slices.Sorted(maps.Keys(declared)) {
|
||||
for _, e := range declared[module].BusState {
|
||||
if e.Storage != catalogue.StorageStream || e.Template != catalogue.TemplateEvents {
|
||||
continue
|
||||
}
|
||||
h := buscompose.Heard{Stream: catalogue.BusName(module, e.Name), Emitter: module}
|
||||
for _, ev := range e.Events {
|
||||
// An event is said on its binding's subject with the machine last (§529): `<event>.<node>`.
|
||||
if ev == event || ev == event+".*" {
|
||||
h.Filters = append(h.Filters, "mesh.bus."+module+"."+e.Name+"."+event+".*")
|
||||
}
|
||||
}
|
||||
if len(h.Filters) > 0 {
|
||||
out = append(out, h)
|
||||
touches = true
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return out, touches, onEvents
|
||||
}
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/meshrecovery"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
|
||||
"slices"
|
||||
"sort"
|
||||
@@ -59,7 +60,8 @@ var seatListEvents = []string{meshcatalog.Seat + "." + meshcatalog.EventSeatDefi
|
||||
|
||||
// seatEvents are the events of seats the part follows its passes by, as `<seat>.<event>`.
|
||||
var seatEvents = []string{EventKeeperRaised, EventKeeperCleared, EventAccountIssued, meshbroker.Seat + "." + BrokerCreated,
|
||||
meshbroker.Seat + "." + BrokerDropped, meshbroker.Seat + "." + BrokerRecreated}
|
||||
meshbroker.Seat + "." + BrokerDropped, meshbroker.Seat + "." + BrokerRecreated,
|
||||
meshrecovery.Seat + "." + meshrecovery.EventEnded, meshrecovery.Seat + "." + meshrecovery.EventPointRemoved, meshrecovery.Seat + "." + meshrecovery.EventExpiryEnded}
|
||||
|
||||
// SeatEventSubjects are the subjects of the seats' events the part follows its passes by besides the verdicts' and the
|
||||
// views': the keeper's condition events, the vault's account-issued and the seat mesh-broker's, each `<seat>.<event>` of
|
||||
|
||||
@@ -6,6 +6,8 @@ import (
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
|
||||
"mesh-grants/internal/buscompose"
|
||||
)
|
||||
|
||||
// The sent stores a module reads follow the catalog graph (design 47 G §59, §124): the consumers of a module's events and the
|
||||
@@ -34,5 +36,11 @@ func TestSentStoresRead(t *testing.T) {
|
||||
if got := sentStoresRead(c.m, all, callers); !slices.Equal(got, c.want) {
|
||||
t.Errorf("%s: %v, want %v", c.name, got, c.want)
|
||||
}
|
||||
} // The same callers reach each seat a declaration holds, uses or watches, from which its users' call subjects are
|
||||
// composed (design 47 G A-179).
|
||||
d := buscompose.Declared{Module: "bank", Uses: []buscompose.Seat{{Name: "orders-seat"}}, Holds: []buscompose.Seat{{Name: "billing"}}}
|
||||
withVerbCallers(callers, &d)
|
||||
if !slices.Equal(d.Uses[0].VerbCallers["place"], callers["orders-seat"]["place"]) || d.Holds[0].VerbCallers != nil {
|
||||
t.Errorf("a used seat's verbs carry %v, a held seat no definition names %v", d.Uses[0].VerbCallers, d.Holds[0].VerbCallers)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -40,7 +40,7 @@ const (
|
||||
|
||||
// Tools is every tool, in the manifest's order.
|
||||
var Tools = []string{ToolGrants, ToolMembership, ToolUserList, ToolComposeNow, ToolAssert, ToolAccounts, ToolConsumerMatch,
|
||||
ToolAcceptHashes, ToolFirstViews, ToolFirstViewAgain, ToolRunning, ToolWaits, ToolRelease, ToolSeatList, ToolSeatListAgain, ToolCallers, ToolWhyRefused}
|
||||
ToolAcceptHashes, ToolFirstViews, ToolFirstViewAgain, ToolRunning, ToolWaits, ToolRelease, ToolSeatList, ToolSeatListAgain, ToolCallers, ToolWhyRefused, ToolNamedPurge}
|
||||
|
||||
// WritesGrants is what grants says of the part's own grants: written once the list the bus loaded carries a build's
|
||||
// accounts, and the one key the assigner reads (ADR 0571 clause 9, which changes ADR 0547 clause 5).
|
||||
@@ -119,6 +119,9 @@ func (p *Part) MembershipOf(node, module string) (any, error) {
|
||||
// What the bus module said it created, as the pass reads it, so the composed side carries the grants and bindings the
|
||||
// pass writes (issue 1966).
|
||||
boundErr := p.withBound(&in)
|
||||
if err := p.withKeptNames(&in); boundErr == nil {
|
||||
boundErr = err
|
||||
}
|
||||
recs, left := ListRecords(in)
|
||||
notFree := WithRootFree(&recs, p.Call)
|
||||
m, ok := Memberships(recs)[node+"."+module]
|
||||
|
||||
@@ -27,6 +27,9 @@
|
||||
"mesh-broker.created",
|
||||
"mesh-broker.dropped",
|
||||
"mesh-broker.recreated",
|
||||
"mesh-recovery.ended",
|
||||
"mesh-recovery.expiry-ended",
|
||||
"mesh-recovery.point-removed",
|
||||
"mesh-delivery-assigner.views.assignments",
|
||||
"mesh-delivery-assigner.views.wants",
|
||||
"mesh-delivery-indexer.views.manifests",
|
||||
@@ -61,7 +64,8 @@
|
||||
"seat:mesh-delivery.source",
|
||||
"seat:mesh-wants.revision",
|
||||
"seat:mesh-hand-acts.hand-act",
|
||||
"seat:mesh-broker.list"
|
||||
"seat:mesh-broker.list",
|
||||
"seat:mesh-recovery.kept-points"
|
||||
],
|
||||
"emits": [
|
||||
"asserted",
|
||||
@@ -87,7 +91,8 @@
|
||||
"grants_seat_list",
|
||||
"grants_seat_list_again",
|
||||
"grants_callers",
|
||||
"grants_why_refused"
|
||||
"grants_why_refused",
|
||||
"grants_named_purge"
|
||||
],
|
||||
"state": [
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user