Copy the controller's hand acts into mesh-hand-acts' log once, and let the bus module's tools delete the ASSIGNMENTS stream at the lease write (issue 774, research 072 rows 26, 33)

2026-10-31 09:15:53 +01:00
parent a68b39f7b5
commit 52c875f870
20 changed files with 523 additions and 19 deletions
@@ -112,12 +112,12 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
 	}
 	// The tools that change the bus (issue 1952 N1-N3): listed, their audit events declared, and the credential
 	// they dial with the module's own secret `broker`.
-	for _, want := range []string{"nats_consumer_remove", "nats_bucket_remove"} {
+	for _, want := range []string{"nats_consumer_remove", "nats_bucket_remove", "nats_stream_remove"} {
 		if !slices.Contains(m.Tools, want) {
 			t.Errorf("tools do not list %s", want)
 		}
 	}
-	for _, want := range []string{"consumer-removed", "bucket-keys-removed"} {
+	for _, want := range []string{"consumer-removed", "bucket-keys-removed", "stream-removed"} {
 		if !slices.Contains(m.Emits, want) {
 			t.Errorf("emits do not list %s: the removal tools say it as the audit line", want)
 		}
@@ -1,7 +1,7 @@
 package main

-// The two tools of this module that change the bus (novox/hq ADR 0564 clause 5, issue 1952 notes 6, 7 and 9):
-// remove one named consumer, and remove one node's keys under a prefix of a key-value bucket. Everything else
+// The three tools of this module that change the bus (novox/hq ADR 0564 clause 5, issue 1952 notes 6, 7 and 9):
+// remove one named consumer, remove one named stream, and remove one node's keys under a prefix of a key-value bucket. Everything else
 // here reads.
 //
 // **Narrow by what the tool needs, and narrower by what the bus grants.** The tools reach the bus on the
@@ -34,6 +34,7 @@ type Said func(event string, body any) error
 const (
 	consumerRemoved   = "consumer-removed"
 	bucketKeysRemoved = "bucket-keys-removed"
+	streamRemoved     = "stream-removed"
 	// maxKeysShown bounds the keys an answer lists; the count is always whole.
 	maxKeysShown = 200
 )
@@ -227,7 +228,85 @@ func RemoveKeys(ctx context.Context, bus Bus, said Said, a map[string]any) (any,
 	return out, nil
 }

-// removalTools are the module's two tools that change the bus; every other tool here reads.
+// StreamRemoved is what nats_stream_remove answers and says.
+type StreamRemoved struct {
+	Stream    string    `json:"stream"`
+	Why       string    `json:"why"`
+	DryRun    bool      `json:"dry_run"`
+	Present   bool      `json:"present"`
+	Messages  uint64    `json:"messages"`
+	Bytes     uint64    `json:"bytes"`
+	Consumers int       `json:"consumers"`
+	Removed   bool      `json:"removed"`
+	Said      string    `json:"said"`
+	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
+// 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) {
+	w, err := why(a)
+	if err != nil {
+		return nil, err
+	}
+	stream := arg(a, "stream")
+	if !name.MatchString(stream) {
+		return nil, errors.New("stream is required, one stream's name (letters, digits, - and _): nothing was removed")
+	}
+	dry, _ := a["dry_run"].(bool)
+	js, closeBus, err := bus(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer closeBus()
+	out := StreamRemoved{Stream: stream, Why: w, DryRun: dry, At: time.Now().UTC()}
+	sctx, cancel := step(ctx)
+	s, err := js.Stream(sctx, stream)
+	cancel()
+	if errors.Is(err, jetstream.ErrStreamNotFound) {
+		out.Said = stream + " is not there: nothing to remove"
+		return out, nil
+	}
+	if err != nil {
+		return nil, unanswered(err, fmt.Sprintf("the stream %s could not be read, so it was not removed", stream))
+	}
+	sctx, cancel = step(ctx)
+	info, err := s.Info(sctx)
+	cancel()
+	if err != nil {
+		return nil, unanswered(err, fmt.Sprintf("the stream %s could not be read, so it was not removed", stream))
+	}
+	out.Present, out.Messages, out.Bytes, out.Consumers = true, info.State.Msgs, info.State.Bytes, info.State.Consumers
+	if info.State.Consumers > 0 {
+		return nil, fmt.Errorf("%s has %d consumers: a reader is attached, so it was not removed", stream, info.State.Consumers)
+	}
+	if dry {
+		out.Said = fmt.Sprintf("%s holds %d messages (%d bytes) and no consumer; it would be removed", stream, out.Messages, out.Bytes)
+		return out, nil
+	}
+	sctx, cancel = step(ctx)
+	err = js.DeleteStream(sctx, stream)
+	cancel()
+	if err != nil {
+		return nil, unanswered(err, fmt.Sprintf("stream %s was not removed", stream))
+	}
+	sctx, cancel = step(ctx)
+	_, err = js.Stream(sctx, stream)
+	cancel()
+	if err == nil {
+		return nil, fmt.Errorf("stream %s is still there after the removal was accepted", stream)
+	}
+	out.Removed = true
+	out.Said = fmt.Sprintf("%s removed: %d messages (%d bytes) and no consumer", stream, out.Messages, out.Bytes)
+	if err := said(streamRemoved, out); err != nil {
+		return nil, fmt.Errorf("stream %s WAS removed (%s) but the audit event could not be said: %w", stream, w, err)
+	}
+	return out, nil
+}
+
+// removalTools are the module's three tools that change the bus; every other tool here reads.
 func removalTools(bus Bus, said Said) []stdio.Tool {
 	run := func(f func(context.Context, Bus, Said, map[string]any) (any, error)) func(map[string]any) (any, error) {
 		return func(a map[string]any) (any, error) { return f(context.Background(), bus, said, a) }
@@ -237,6 +316,10 @@ func removalTools(bus Bus, said Said) []stdio.Tool {
 			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.",
 			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.",
+			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",
 			Description: "Removes (purges) keys from a key-value bucket: one node's keys (`node`, the keys starting `<node>.`) or the keys under a `prefix`, e.g. a full acts bucket of the node-engine. Needs `why`, kept in the audit trail; never the whole bucket. Counts the keys first and lists them (`dry_run` only counts; `expect` refuses when the count differs); removes only those keys, and their history. See the buckets and their keys with nats_buckets. The bus refuses any bucket the module's credential is not granted.",
 			Input: map[string]any{"bucket": str("the bucket as nats_buckets lists it, e.g. mesh-host_acts"), "node": str("a node whose keys (`<node>.`) go"), "prefix": str("key prefix instead of node"),
@@ -31,6 +31,8 @@ accounts { MESH { jetstream: enabled, users: [
      publish: { allow: [
        "$JS.API.CONSUMER.INFO.DEAD_LETTER_NOTICES.controller",
        "$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller",
+        "$JS.API.STREAM.INFO.ASSIGNMENTS",
+        "$JS.API.STREAM.DELETE.ASSIGNMENTS",
        "$JS.API.STREAM.INFO.KV_mesh-host_acts",
        "$KV.mesh-host_acts.>" ] },
      subscribe: { allow: ["_INBOX.nats-tools.>"] } } }
@@ -73,6 +75,14 @@ accounts { MESH { jetstream: enabled, users: [
 	if err != nil {
 		t.Fatal(err)
 	}
+	if _, err := ajs.CreateStream(ctx, jetstream.StreamConfig{Name: "ASSIGNMENTS", Subjects: []string{"mesh.assignment.*.*"}, MaxMsgsPerSubject: 1}); err != nil {
+		t.Fatal(err)
+	}
+	for _, sub := range []string{"mesh.assignment.g14.a", "mesh.assignment.ace.b"} {
+		if _, err := ajs.Publish(ctx, sub, []byte("{}")); err != nil {
+			t.Fatal(err)
+		}
+	}
 	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"))
@@ -131,6 +141,30 @@ accounts { MESH { jetstream: enabled, users: [
 		t.Fatalf("a consumer already gone: %v", err)
 	}

+	// Research 072 row 33: the stream ASSIGNMENTS goes whole, once; read first, and asking again is no error; a stream
+	// outside the grant is not touched.
+	if _, err := RemoveStream(ctx, bus, say, map[string]any{"stream": "ASSIGNMENTS"}); err == nil || !strings.Contains(err.Error(), "why is required") {
+		t.Fatalf("a stream without why: %v", err)
+	}
+	if _, err := RemoveStream(ctx, bus, say, map[string]any{"stream": "DEAD_LETTER_NOTICES", "why": "x"}); err == nil || !strings.Contains(err.Error(), "grant") {
+		t.Fatalf("a stream outside the grant: %v", err)
+	}
+	if d, err := RemoveStream(ctx, bus, say, map[string]any{"stream": "ASSIGNMENTS", "why": "look", "dry_run": true}); err != nil || d.(StreamRemoved).Messages != 2 || d.(StreamRemoved).Removed {
+		t.Fatalf("a dry run: %+v %v", d, err)
+	}
+	if _, err := ajs.Stream(ctx, "ASSIGNMENTS"); err != nil {
+		t.Fatalf("a dry run removed the stream: %v", err)
+	}
+	if r, err := RemoveStream(ctx, bus, say, map[string]any{"stream": "ASSIGNMENTS", "why": "the lease write"}); err != nil || !r.(StreamRemoved).Removed || r.(StreamRemoved).Messages != 2 {
+		t.Fatalf("the removal: %+v %v", r, err)
+	}
+	if _, err := ajs.Stream(ctx, "ASSIGNMENTS"); err == nil {
+		t.Fatal("ASSIGNMENTS is still there")
+	}
+	if r, err := RemoveStream(ctx, bus, say, map[string]any{"stream": "ASSIGNMENTS", "why": "again"}); err != nil || r.(StreamRemoved).Removed || r.(StreamRemoved).Present {
+		t.Fatalf("a second removal: %+v %v", r, 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)
@@ -167,7 +201,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,bucket-keys-removed" {
+	if strings.Join(said, ",") != "consumer-removed,stream-removed,bucket-keys-removed" {
 		t.Fatalf("the audit events said: %v", said)
 	}
 }
@@ -234,6 +234,10 @@ func TestTheRemovalToolsRefuseBeforeTheyReachTheBus(t *testing.T) {
 		{"nats_bucket_remove", map[string]any{"bucket": "mesh-host_acts", "node": "g14", "prefix": "g", "why": "x"}, "exactly one of node"},
 		{"nats_bucket_remove", map[string]any{"bucket": "mesh-host_acts", "prefix": "g*", "why": "x"}, "without wildcards"},
 		{"nats_bucket_remove", map[string]any{"bucket": "", "node": "g14", "why": "x"}, "bucket is required"},
+		{"nats_stream_remove", map[string]any{"stream": "ASSIGNMENTS"}, "why is required"},
+		{"nats_stream_remove", map[string]any{"stream": "ASSIGNMENTS", "why": " "}, "why is required"},
+		{"nats_stream_remove", map[string]any{"stream": "A.>", "why": "x"}, "stream is required"},
+		{"nats_stream_remove", map[string]any{"stream": "", "why": "x"}, "stream is required"},
 	} {
 		var run func(map[string]any) (any, error)
 		for _, tl := range removalTools(bus, said) {
@@ -3,7 +3,7 @@ module nats-tools
 go 1.26.0

 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030001116-2da0de75c3f0
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031081231-d352fe51b455
 	github.com/klauspost/compress v1.20.0
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
@@ -1,5 +1,7 @@
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030001116-2da0de75c3f0 h1:lKU3HjNXGiVbLW2hUgFjU/TMTUYMlNnnmZxG5T7biQA=
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030001116-2da0de75c3f0/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031081231-d352fe51b455 h1:1rAqVPyu07xkacqJrDaqUtoqLQJmklfyXC/5erRJnEk=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031081231-d352fe51b455/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
 github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
@@ -30,7 +30,8 @@
  ],
  "emits": [
    "consumer-removed",
-    "bucket-keys-removed"
+    "bucket-keys-removed",
+    "stream-removed"
  ],
  "consumes": [],
  "listens": [
@@ -54,7 +55,8 @@
    "nats_user_can",
    "nats_snapshot_verify",
    "nats_consumer_remove",
-    "nats_bucket_remove"
+    "nats_bucket_remove",
+    "nats_stream_remove"
  ],
  "guards": [
    8222
@@ -208,7 +210,8 @@
        ],
        "bus": {
          "removes": [
-            "DEAD_LETTER_NOTICES.controller"
+            "DEAD_LETTER_NOTICES.controller",
+            "ASSIGNMENTS"
          ]
        },
        "env": {