novox/mesh-catalog
18
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Let the bus tools remove the controller's leftovers by one word

Browse Source 
The controller ended by its unassignment, so its consumers, streams and
buckets may be removed as recorded hand acts (ADR 0590 §6). The nats tools'
account declares controller-leftovers in place of the lease write's two
words, and the removal test proves a consumer of EVENTS and a bucket's
stream go while EVENTS and its other consumers stay (issue 1126 B5).

Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
This commit is contained in:
jschoubben
2026-11-02 12:29:56 +01:00
parent 0bfc72d230
commit 40c4fabff6
4 changed files with 64 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/nats/cmd/nats-tools/manifest_test.go
+15 -9
View File
@@ -12,6 +12,13 @@ import (
// programs mounted read-only under /opt, unpacked by archive resources the container names under restart-on
// (the node-engine does not look inside a mounted directory), and the container the same one it was.
func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
type artifact struct {
Name, Kind, From, Language, System, Binary string
Env map[string]string
Bus struct {
Removes []string `json:"removes"`
} `json:"bus"`
}
raw, err := os.ReadFile("../../module.json")
if err != nil {
t.Fatal(err)
@@ -29,11 +36,8 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
Emits []string `json:"emits"`
Resources []map[string]any `json:"resources"`
Build struct {
On any `json:"on"`
Artifacts []struct {
Name, Kind, From, Language, System, Binary string
Env map[string]string
} `json:"artifacts"`
On any `json:"on"`
Artifacts []artifact `json:"artifacts"`
} `json:"build"`
}
if err := json.Unmarshal(raw, &m); err != nil {
@@ -42,10 +46,7 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
if m.Build.On != nil {
t.Error("build.on is declared: nothing is built on a base image any more")
}
artifacts := map[string]struct {
Name, Kind, From, Language, System, Binary string
Env map[string]string
}{}
artifacts := map[string]artifact{}
for _, a := range m.Build.Artifacts {
artifacts[a.Name] = a
if a.Kind == "image" {
@@ -123,6 +124,11 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
}
}
for _, a := range m.Build.Artifacts {
// Issue 1126 B5 (ADR 0590 §6): the tools may remove the controller's leftovers, by the one word for them, and
// nothing else of the controller's by a word of its own.
if a.Name == "tools" && !slices.Equal(a.Bus.Removes, []string{"controller-leftovers"}) {
t.Errorf("the tools' bus removes %v: the controller's leftovers by their one word, controller-leftovers", a.Bus.Removes)
}
if a.Name == "tools" && a.Env["NATS_TOOLS_CREDENTIAL"] != "${dir:tools-secret}/broker-nats-tools" {
t.Errorf("the tools' credential is %q: the process nats-tools' own secret broker-nats-tools, under a word of the bundle's own (the runtime strips MESH_BROKER_FILE)", a.Env["NATS_TOOLS_CREDENTIAL"])
}
modules/nats/cmd/nats-tools/removal.go
+5 -5
View File
@@ -9,7 +9,7 @@ package main
// granted, so a wrong stream, consumer or bucket is refused by the server and said here, not trusted to the
// arguments. Each tool refuses without `why`, reads what it would remove first (and says it, so a person can ask
// with dry_run and then for real), removes, reads that it is gone, and says what it did as an event of the module
// (the audit trail: who asked is in the call the controller audits, why and what are in the event).
// (the audit trail: who asked is in the call the tool runner audits, why and what are in the event).
import (
"context"
@@ -246,8 +246,8 @@ type StreamRemoved struct {
At time.Time `json:"at"`
}
// RemoveStream removes one named stream whole, after reading it, and says so (the lease write's step for the stream
// ASSIGNMENTS, novox/hq research 072 row 33). **A stream already gone is no error**: the removal is the same act
// RemoveStream removes one named stream whole, after reading it, and says so (the controller's leftovers: its streams
// CONTROL and ASSIGNMENTS, its buckets' and object stores' streams; novox/hq ADR 0590 §6). **A stream already gone is no error**: the removal is the same act
// however often it is asked, and says it found nothing. A stream with a consumer is refused, naming how many: a reader
// is still attached, and that is for a person to look at before the stream goes.
func RemoveStream(ctx context.Context, bus Bus, said Said, a map[string]any) (any, error) {
@@ -321,11 +321,11 @@ func removalTools(bus Bus, said Said) []stdio.Tool {
}
return []stdio.Tool{
{Name: "nats_consumer_remove",
Description: "Removes (deletes) one named JetStream consumer of one stream, e.g. the consumer `controller` of DEAD_LETTER_NOTICES at the lease write (novox/hq ADR 0564 clause 5). Needs `why`, kept in the audit trail; refuses an unknown stream or consumer; reads the consumer first and answers what it was (filters, pending, unacknowledged); the stream and its other consumers are not touched. The bus refuses any consumer the module's credential is not granted.",
Description: "Removes (deletes) one named JetStream consumer of one stream, e.g. a consumer the controller left, such as `controller` of DEAD_LETTER_NOTICES or `controller-merges` of EVENTS (novox/hq ADR 0590 §6, recorded as a hand act under issue 1126). Needs `why`, kept in the audit trail; refuses an unknown stream or consumer; reads the consumer first and answers what it was (filters, pending, unacknowledged); the stream and its other consumers are not touched. The bus refuses any consumer the module's credential is not granted.",
Input: map[string]any{"stream": str("the stream, e.g. DEAD_LETTER_NOTICES"), "consumer": str("the consumer's name, e.g. controller"), "why": str("why it is removed, for the audit trail (required)")},
Run: run(RemoveConsumer)},
{Name: "nats_stream_remove",
Description: "Removes (deletes) one named JetStream stream whole, e.g. ASSIGNMENTS at the lease write, beside the consumer of DEAD_LETTER_NOTICES (novox/hq research 072 row 33). Needs `why`, kept in the audit trail; reads the stream first and answers its messages, bytes and consumers; refuses a stream that still has a consumer; a stream already gone is answered as such and is no error, so asking again is safe. `dry_run` only reads. The bus refuses any stream the module's credential is not granted.",
Description: "Removes (deletes) one named JetStream stream whole, e.g. a stream the controller left: CONTROL, ASSIGNMENTS, one of its buckets as KV_<bucket> or a node's declaration files as OBJ_declaration-files-<node> (novox/hq ADR 0590 §6; a bucket's whole removal is this tool's, never nats_bucket_remove's). Needs `why`, kept in the audit trail; reads the stream first and answers its messages, bytes and consumers; refuses a stream that still has a consumer; a stream already gone is answered as such and is no error, so asking again is safe. `dry_run` only reads. The bus refuses any stream the module's credential is not granted.",
Input: map[string]any{"stream": str("the stream, e.g. ASSIGNMENTS"), "why": str("why it is removed, for the audit trail (required)"), "dry_run": map[string]any{"type": "boolean", "description": "read the stream, remove nothing"}},
Run: run(RemoveStream)},
{Name: "nats_bucket_remove",
modules/nats/cmd/nats-tools/removal_test.go
+43 -3
View File
@@ -17,9 +17,11 @@ import (
"github.com/nats-io/nats.go/jetstream"
)
// The two tools that change the bus, against a server shaped like the mesh's, on the credential the module's
// The tools that change the bus, against a server shaped like the mesh's, on the credential the module's
// grant proposal gives them and nothing wider (issue 1952 N2, N3): the bus itself refuses what the grant leaves out,
// so this is also the proof that the proposed grant is enough and no wider than it needs to be.
// so this is also the proof that the proposed grant is enough and no wider than it needs to be. Issue 1126 B5 (ADR 0590
// §6): the controller's leftovers go the same way, a consumer of EVENTS and a bucket's stream among them, while EVENTS
// and its other consumers stay.
func TestTheRemovalToolsChangeOnlyWhatTheyAreAskedAndTheGrantAllows(t *testing.T) {
dir := t.TempDir()
conf := filepath.Join(dir, "nats.conf")
@@ -33,6 +35,10 @@ accounts { MESH { jetstream: enabled, users: [
"$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller",
"$JS.API.STREAM.INFO.ASSIGNMENTS",
"$JS.API.STREAM.DELETE.ASSIGNMENTS",
"$JS.API.CONSUMER.INFO.EVENTS.controller-merges",
"$JS.API.CONSUMER.DELETE.EVENTS.controller-merges",
"$JS.API.STREAM.INFO.KV_mesh-controller_rollouts",
"$JS.API.STREAM.DELETE.KV_mesh-controller_rollouts",
"$JS.API.STREAM.INFO.KV_mesh-host_acts",
"$KV.mesh-host_acts.>" ] },
subscribe: { allow: ["_INBOX.nats-tools.>"] } } }
@@ -83,6 +89,19 @@ accounts { MESH { jetstream: enabled, users: [
t.Fatal(err)
}
}
if _, err := ajs.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.events.>"}}); err != nil {
t.Fatal(err)
}
for _, c := range []string{"controller-merges", "mesh-grants"} {
if _, err := ajs.CreateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: c, AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
t.Fatal(err)
}
}
rollouts, err := ajs.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "mesh-controller_rollouts"})
if err != nil {
t.Fatal(err)
}
rollouts.Put(ctx, "p-1", []byte("{}"))
other, _ := ajs.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "other_state"})
for _, k := range []string{"g14.1", "g14.2", "g14.sub.3", "g14x.4", "ace.5"} {
acts.Put(ctx, k, []byte("v"))
@@ -165,6 +184,27 @@ accounts { MESH { jetstream: enabled, users: [
t.Fatalf("a second removal: %+v %v", r, err)
}
// Issue 1126 B5: the controller's consumer of EVENTS goes and the stream's other consumer stays; its bucket's stream
// goes whole; EVENTS itself is outside the grant.
if _, err := RemoveConsumer(ctx, bus, say, map[string]any{"stream": "EVENTS", "consumer": "mesh-grants", "why": "x"}); err == nil {
t.Fatal("a consumer of EVENTS outside the grant was removed")
}
if _, err := RemoveConsumer(ctx, bus, say, map[string]any{"stream": "EVENTS", "consumer": "controller-merges", "why": "the controller ended"}); err != nil {
t.Fatalf("the controller's consumer of EVENTS: %v", err)
}
if _, err := ajs.Consumer(ctx, "EVENTS", "mesh-grants"); err != nil {
t.Fatalf("the other consumer of EVENTS went: %v", err)
}
if r, err := RemoveStream(ctx, bus, say, map[string]any{"stream": "KV_mesh-controller_rollouts", "why": "the controller ended"}); err != nil || !r.(StreamRemoved).Removed || r.(StreamRemoved).Messages != 1 {
t.Fatalf("the controller's bucket: %+v %v", r, err)
}
if _, err := RemoveStream(ctx, bus, say, map[string]any{"stream": "EVENTS", "why": "x"}); err == nil || !strings.Contains(err.Error(), "consumers") && !strings.Contains(err.Error(), "grant") {
t.Fatalf("EVENTS: %v", err)
}
if _, err := ajs.Stream(ctx, "EVENTS"); err != nil {
t.Fatalf("EVENTS went: %v", err)
}
// N3: only the node's keys go; counted first; refused without why, for a whole bucket, and outside the grant.
if _, err := RemoveKeys(ctx, bus, say, map[string]any{"bucket": "mesh-host_acts", "node": "g14"}); err == nil || !strings.Contains(err.Error(), "why is required") {
t.Fatalf("without why: %v", err)
@@ -201,7 +241,7 @@ accounts { MESH { jetstream: enabled, users: [
if k := keys(other); k != "g14.1" {
t.Fatalf("another bucket changed: %s", k)
}
if strings.Join(said, ",") != "consumer-removed,stream-removed,bucket-keys-removed" {
if strings.Join(said, ",") != "consumer-removed,stream-removed,consumer-removed,stream-removed,bucket-keys-removed" {
t.Fatalf("the audit events said: %v", said)
}
}
modules/nats/module.json
+1 -2
View File
@@ -215,8 +215,7 @@
],
"bus": {
"removes": [
"DEAD_LETTER_NOTICES.controller",
"ASSIGNMENTS"
"controller-leftovers"
]
},
"env": {