From ff2f90ec187fe8d031dff65887e80f34c36cf5a3 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 06:22:54 +0100 Subject: [PATCH] GR13 purge: the named asks go at the first revision that leaves them, or by grants_named_purge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit While no entry of the revision invokes a queue of a build seat (seat:.accept., read by catalogue.QueueInvoke), the part composes no named worker, and at the first such pass it purges accept..* of each build seat's stream and deletes the named workers, saying each in its journal; a failed read of the catalog half purges nothing. Its account alone is granted the stream's purge and the named workers' delete. The hand tool grants_named_purge records its hand act first, then asks the process to do it. Builds: design 47 G §544, §600, §601, A-231 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- modules/mesh-grants/cmd/mesh-grants/create.go | 26 ++++ .../cmd/mesh-grants/integration_test.go | 25 ++++ modules/mesh-grants/cmd/mesh-grants/main.go | 2 +- modules/mesh-grants/cmd/mesh-grants/tools.go | 19 ++- .../internal/buscompose/named_asks_test.go | 26 ++++ .../mesh-grants/internal/buscompose/nats.go | 8 ++ .../internal/buscompose/seattraffic.go | 66 +++++++++- .../mesh-grants/internal/buscompose/users.go | 1 + .../mesh-grants/internal/grants/namedpurge.go | 121 ++++++++++++++++++ .../internal/grants/namedpurge_test.go | 96 ++++++++++++++ modules/mesh-grants/internal/grants/part.go | 18 ++- .../mesh-grants/internal/grants/records.go | 1 + modules/mesh-grants/internal/grants/tools.go | 2 +- modules/mesh-grants/module.json | 3 +- 14 files changed, 404 insertions(+), 10 deletions(-) create mode 100644 modules/mesh-grants/internal/grants/namedpurge.go create mode 100644 modules/mesh-grants/internal/grants/namedpurge_test.go diff --git a/modules/mesh-grants/cmd/mesh-grants/create.go b/modules/mesh-grants/cmd/mesh-grants/create.go index 23621ec8..a072cc46 100644 --- a/modules/mesh-grants/cmd/mesh-grants/create.go +++ b/modules/mesh-grants/cmd/mesh-grants/create.go @@ -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 +} diff --git a/modules/mesh-grants/cmd/mesh-grants/integration_test.go b/modules/mesh-grants/cmd/mesh-grants/integration_test.go index 4b5fc605..7d45118a 100644 --- a/modules/mesh-grants/cmd/mesh-grants/integration_test.go +++ b/modules/mesh-grants/cmd/mesh-grants/integration_test.go @@ -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). diff --git a/modules/mesh-grants/cmd/mesh-grants/main.go b/modules/mesh-grants/cmd/mesh-grants/main.go index a98c8d64..4e3654b3 100644 --- a/modules/mesh-grants/cmd/mesh-grants/main.go +++ b/modules/mesh-grants/cmd/mesh-grants/main.go @@ -169,7 +169,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro branches.Entries.Changed = func(meshcatalogclient.EntryRow) { f.Wake() } fv.Catalog = &grants.CatalogOf{Branches: branches, Context: ctx, Wake: f.Wake} 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 diff --git a/modules/mesh-grants/cmd/mesh-grants/tools.go b/modules/mesh-grants/cmd/mesh-grants/tools.go index 0ad31793..a78064cd 100644 --- a/modules/mesh-grants/cmd/mesh-grants/tools.go +++ b/modules/mesh-grants/cmd/mesh-grants/tools.go @@ -281,6 +281,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..* 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 " + @@ -291,9 +299,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. diff --git a/modules/mesh-grants/internal/buscompose/named_asks_test.go b/modules/mesh-grants/internal/buscompose/named_asks_test.go index 9361ac03..74ff72cf 100644 --- a/modules/mesh-grants/internal/buscompose/named_asks_test.go +++ b/modules/mesh-grants/internal/buscompose/named_asks_test.go @@ -111,3 +111,29 @@ func TestAVerbIsGrantedToItsEffectiveCallersAlone(t *testing.T) { } } } + +// 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") + } + } +} diff --git a/modules/mesh-grants/internal/buscompose/nats.go b/modules/mesh-grants/internal/buscompose/nats.go index 42d11be8..8c2abfea 100644 --- a/modules/mesh-grants/internal/buscompose/nats.go +++ b/modules/mesh-grants/internal/buscompose/nats.go @@ -113,6 +113,9 @@ type Seat struct { // 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:.accept.`), 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 @@ -152,6 +155,9 @@ type Principal struct { // ruling, registry milestone 13:0x): `mesh.seat.mesh-broker.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 // Process is the process resource's id, for a KindProcess principal alone (novox/hq ADR 0436 §2): // Username and inbox are ... Process string @@ -930,6 +936,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()...) } diff --git a/modules/mesh-grants/internal/buscompose/seattraffic.go b/modules/mesh-grants/internal/buscompose/seattraffic.go index 0cb21b5f..fc34cfe9 100644 --- a/modules/mesh-grants/internal/buscompose/seattraffic.go +++ b/modules/mesh-grants/internal/buscompose/seattraffic.go @@ -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..*` (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.`, 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) diff --git a/modules/mesh-grants/internal/buscompose/users.go b/modules/mesh-grants/internal/buscompose/users.go index c861f9d2..c3eefa31 100644 --- a/modules/mesh-grants/internal/buscompose/users.go +++ b/modules/mesh-grants/internal/buscompose/users.go @@ -378,6 +378,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 diff --git a/modules/mesh-grants/internal/grants/namedpurge.go b/modules/mesh-grants/internal/grants/namedpurge.go new file mode 100644 index 00000000..56f76982 --- /dev/null +++ b/modules/mesh-grants/internal/grants/namedpurge.go @@ -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:.accept.`, the part purges the subjects `accept..*` 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 +} diff --git a/modules/mesh-grants/internal/grants/namedpurge_test.go b/modules/mesh-grants/internal/grants/namedpurge_test.go new file mode 100644 index 00000000..26a1aaf1 --- /dev/null +++ b/modules/mesh-grants/internal/grants/namedpurge_test.go @@ -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) + } +} diff --git a/modules/mesh-grants/internal/grants/part.go b/modules/mesh-grants/internal/grants/part.go index 51ea62c9..5d95c2c3 100644 --- a/modules/mesh-grants/internal/grants/part.go +++ b/modules/mesh-grants/internal/grants/part.go @@ -132,6 +132,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). @@ -168,6 +170,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 @@ -186,6 +191,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. @@ -268,6 +276,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. @@ -505,7 +516,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 @@ -544,6 +555,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 diff --git a/modules/mesh-grants/internal/grants/records.go b/modules/mesh-grants/internal/grants/records.go index f2c32ade..721ced60 100644 --- a/modules/mesh-grants/internal/grants/records.go +++ b/modules/mesh-grants/internal/grants/records.go @@ -111,6 +111,7 @@ func records(in Inputs, union bool) (buscompose.Records, []Gap) { withEventSubjects(in.EventSubjects, &d) withAcceptCallers(in.AcceptCallers, &d) withVerbCallers(in.VerbCallers, &d) + withNoNamedAsks(NamedAsksUnused(in), &d) return d } d := declare(m) diff --git a/modules/mesh-grants/internal/grants/tools.go b/modules/mesh-grants/internal/grants/tools.go index 49fd88e0..cd57f538 100644 --- a/modules/mesh-grants/internal/grants/tools.go +++ b/modules/mesh-grants/internal/grants/tools.go @@ -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). diff --git a/modules/mesh-grants/module.json b/modules/mesh-grants/module.json index a5b98926..1edf4c56 100644 --- a/modules/mesh-grants/module.json +++ b/modules/mesh-grants/module.json @@ -84,7 +84,8 @@ "grants_seat_list", "grants_seat_list_again", "grants_callers", - "grants_why_refused" + "grants_why_refused", + "grants_named_purge" ], "state": [ {