The bus module sets and restores a key-value bucket's expiry (ADR 0582)

nats_bucket_expiry reads a bucket's stream, sets MaxAge (a duration, or none), reads it back, records a hand act
through mesh-hand-acts and says bucket-expiry-set, so the lease write can take the 15 seconds off the lease bucket and
the hand-back can restore them. Refuses a missing why and a name that is no bucket. Two new unit tests (stand-in
embedded server) in a module at 26 of 500; no test removed, none needed to stay in budget.

(issue 774, 774/S6)

Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL

modules/nats/cmd/nats-tools/expiry.go

@@ -0,0 +1,158 @@
+package main
+
+// The tool of this module that sets and restores a key-value bucket's expiry (novox/hq ADR 0582, "How it is
+// checked": the bus module's expiry tool sets and restores a bucket's expiry and records the act; issue 774). The
+// grants part calls it at the lease write, to take the 15 seconds off the lease bucket, and at the hand-back, to put
+// them back. Like the removal tools it refuses without `why`, reads what it changes first, reads it back after, and
+// says what it did as an event of the module; it also records the act through the seat mesh-hand-acts (ADR 0565
+// clause 11), because it changes what the mesh keeps.
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/handacts"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+const bucketExpirySet = "bucket-expiry-set"
+
+// kvPrefix is the name a bucket's stream carries.
+const kvPrefix = "KV_"
+
+// Recorded records one hand act through the seat mesh-hand-acts.
+type Recorded func(handacts.Request) error
+
+// RecordedBySeat records an act through the runtime, to the seat mesh-hand-acts alone.
+func RecordedBySeat(q handacts.Request) error {
+	_, err := handacts.Recorder{Ask: stdio.Ask}.Record(q)
+	return err
+}
+
+// BucketExpirySet is what nats_bucket_expiry answers and says.
+type BucketExpirySet struct {
+	Bucket  string    `json:"bucket"`
+	Was     string    `json:"was"`
+	Now     string    `json:"now"`
+	Why     string    `json:"why"`
+	Warrant string    `json:"warrant,omitempty"`
+	DryRun  bool      `json:"dry_run,omitempty"`
+	Changed bool      `json:"changed"`
+	Said    string    `json:"said"`
+	At      time.Time `json:"at"`
+}
+
+// expiryText names an expiry: a duration, or "none".
+func expiryText(d time.Duration) string {
+	if d == 0 {
+		return "none"
+	}
+	return d.String()
+}
+
+// parseExpiry reads `max_age`: "none" removes the expiry, otherwise a positive duration such as 15s.
+func parseExpiry(s string) (time.Duration, error) {
+	s = strings.TrimSpace(s)
+	if s == "none" {
+		return 0, nil
+	}
+	d, err := time.ParseDuration(s)
+	if err != nil || d <= 0 {
+		return 0, fmt.Errorf("max_age is required: a positive duration such as 15s, or none to remove the expiry (got %q): nothing was changed", s)
+	}
+	return d, nil
+}
+
+// SetBucketExpiry sets one key-value bucket's expiry (MaxAge of its stream) after reading it, reads it back, records
+// the act and says so. A bucket already at that expiry is answered as such and records nothing: asking again is safe.
+// A name that is no key-value bucket is refused and nothing changes.
+func SetBucketExpiry(ctx context.Context, bus Bus, said Said, record Recorded, a map[string]any) (any, error) {
+	w, err := why(a)
+	if err != nil {
+		return nil, err
+	}
+	bucket := arg(a, "bucket")
+	if !name.MatchString(bucket) {
+		return nil, errors.New("bucket is required, one bucket's name as nats_buckets lists it (letters, digits, - and _): nothing was changed")
+	}
+	maxAge, err := parseExpiry(arg(a, "max_age"))
+	if err != nil {
+		return nil, err
+	}
+	dry, _ := a["dry_run"].(bool)
+	js, closeBus, err := bus(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer closeBus()
+	stream := kvPrefix + bucket
+	out := BucketExpirySet{Bucket: bucket, Why: w, Warrant: strings.TrimSpace(arg(a, "warrant")), DryRun: dry, At: time.Now().UTC()}
+	sctx, cancel := step(ctx)
+	s, err := js.Stream(sctx, stream)
+	cancel()
+	if errors.Is(err, jetstream.ErrStreamNotFound) {
+		return nil, fmt.Errorf("%s is no key-value bucket (there is no stream %s): nothing was changed", bucket, stream)
+	}
+	if err != nil {
+		return nil, unanswered(err, fmt.Sprintf("the bucket %s could not be read, so its expiry was not changed", bucket))
+	}
+	cfg := s.CachedInfo().Config
+	out.Was, out.Now = expiryText(cfg.MaxAge), expiryText(maxAge)
+	if cfg.MaxAge == maxAge {
+		out.Said = fmt.Sprintf("%s already expires after %s: nothing to change", bucket, out.Now)
+		return out, nil
+	}
+	if dry {
+		out.Said = fmt.Sprintf("%s expires after %s; it would be set to %s", bucket, out.Was, out.Now)
+		return out, nil
+	}
+	cfg.MaxAge = maxAge
+	// A stream refuses a window of duplicates longer than its expiry; a bucket made with 15 s carries 15 s.
+	if maxAge > 0 && (cfg.Duplicates == 0 || cfg.Duplicates > maxAge) {
+		cfg.Duplicates = maxAge
+	}
+	sctx, cancel = step(ctx)
+	_, err = js.UpdateStream(sctx, cfg)
+	cancel()
+	if err != nil {
+		return nil, unanswered(err, fmt.Sprintf("the expiry of %s was not changed", bucket))
+	}
+	sctx, cancel = step(ctx)
+	s, err = js.Stream(sctx, stream)
+	cancel()
+	if err != nil {
+		return nil, unanswered(err, fmt.Sprintf("the new expiry of %s was accepted but could not be read back", bucket))
+	}
+	if got := s.CachedInfo().Config.MaxAge; got != maxAge {
+		return nil, fmt.Errorf("%s reads back with an expiry of %s after the change to %s was accepted", bucket, expiryText(got), out.Now)
+	}
+	out.Changed = true
+	out.Said = fmt.Sprintf("%s expiry set from %s to %s", bucket, out.Was, out.Now)
+	what := fmt.Sprintf("set the expiry of the bucket %s from %s to %s", bucket, out.Was, out.Now)
+	if out.Warrant != "" {
+		what += " (warrant " + out.Warrant + ")"
+	}
+	if err := record(handacts.Request{What: what, Why: w, Cause: "bucket-expiry"}); err != nil {
+		return nil, fmt.Errorf("the expiry of %s WAS changed (%s) but the hand act could not be recorded: %w", bucket, what, err)
+	}
+	if err := said(bucketExpirySet, out); err != nil {
+		return nil, fmt.Errorf("the expiry of %s WAS changed (%s) but the audit event could not be said: %w", bucket, w, err)
+	}
+	return out, nil
+}
+
+// expiryTools is the module's tool that sets a bucket's expiry.
+func expiryTools(bus Bus, said Said, record Recorded) []stdio.Tool {
+	return []stdio.Tool{{Name: "nats_bucket_expiry",
+		Description: "Sets or restores the expiry of one key-value bucket: `max_age` is a duration such as 15s, or `none` to keep keys with no expiry (e.g. the lease bucket of mesh-controller at the lease write, back to 15s at the hand-back; novox/hq ADR 0582). Needs `why`, kept in the audit trail and recorded as a hand act through mesh-hand-acts (`warrant` names the operator's ask when there is one); refuses a name that is no key-value bucket; reads the expiry first and back after, and a bucket already at that expiry is answered as such. `dry_run` only reads. 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-controller_lease"), "max_age": str("a duration such as 15s, or none to remove the expiry"),
+			"why": str("why it is changed, for the audit trail (required)"), "warrant": str("the operator's ask this follows, when there is one"),
+			"dry_run": map[string]any{"type": "boolean", "description": "read the expiry, change nothing"}},
+		Run: func(a map[string]any) (any, error) {
+			return SetBucketExpiry(context.Background(), bus, said, record, a)
+		}}}
+}

modules/nats/cmd/nats-tools/expiry_test.go

@@ -0,0 +1,123 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/handacts"
+	"github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// A stand-in bus: the upstream server the mesh runs, in this process, with a bucket made as the controller makes its
+// lease bucket (15 seconds) and a plain stream that is no bucket.
+func expiryBus(t *testing.T) (Bus, jetstream.JetStream) {
+	t.Helper()
+	s, err := server.NewServer(&server.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	go s.Start()
+	if !s.ReadyForConnections(5 * time.Second) {
+		t.Fatal("the server did not start")
+	}
+	t.Cleanup(s.Shutdown)
+	nc, err := nats.Connect(s.ClientURL())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(nc.Close)
+	js, _ := jetstream.New(nc)
+	ctx := context.Background()
+	if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "mesh-controller_lease", TTL: 15 * time.Second}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "ASSIGNMENTS", Subjects: []string{"a.>"}}); err != nil {
+		t.Fatal(err)
+	}
+	return func(context.Context) (jetstream.JetStream, func(), error) { return js, func() {}, nil }, js
+}
+
+func maxAge(t *testing.T, js jetstream.JetStream, stream string) time.Duration {
+	t.Helper()
+	s, err := js.Stream(context.Background(), stream)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return s.CachedInfo().Config.MaxAge
+}
+
+// 774/S6/AC1: a bucket with a 15 s expiry set to none reads back with none, one hand act and bucket-expiry-set;
+// restored to 15 s it reads back with 15 s again; asking for what it already has records nothing.
+func TestABucketsExpiryIsSetAndRestoredAndRecorded(t *testing.T) {
+	bus, js := expiryBus(t)
+	var acts []handacts.Request
+	var events []string
+	record := func(q handacts.Request) error { acts = append(acts, q); return nil }
+	said := func(e string, _ any) error { events = append(events, e); return nil }
+	call := func(max string, extra map[string]any) (BucketExpirySet, error) {
+		a := map[string]any{"bucket": "mesh-controller_lease", "max_age": max, "why": "the lease write", "warrant": "ask-1"}
+		for k, v := range extra {
+			a[k] = v
+		}
+		out, err := SetBucketExpiry(context.Background(), bus, said, record, a)
+		if err != nil {
+			return BucketExpirySet{}, err
+		}
+		return out.(BucketExpirySet), nil
+	}
+	if out, err := call("none", map[string]any{"dry_run": true}); err != nil || out.Changed || maxAge(t, js, "KV_mesh-controller_lease") != 15*time.Second {
+		t.Fatalf("a dry run changed the bucket or failed: %+v %v", out, err)
+	}
+	out, err := call("none", nil)
+	if err != nil || !out.Changed || out.Was != "15s" || out.Now != "none" || maxAge(t, js, "KV_mesh-controller_lease") != 0 {
+		t.Fatalf("none: %+v %v", out, err)
+	}
+	if len(acts) != 1 || !strings.Contains(acts[0].What, "mesh-controller_lease from 15s to none") || !strings.Contains(acts[0].What, "ask-1") || acts[0].Why != "the lease write" {
+		t.Errorf("the hand act is %+v", acts)
+	}
+	if len(events) != 1 || events[0] != "bucket-expiry-set" {
+		t.Errorf("events are %v", events)
+	}
+	if out, err = call("none", nil); err != nil || out.Changed || len(acts) != 1 || len(events) != 1 {
+		t.Errorf("asking again for none changed or recorded something: %+v %v acts %d events %d", out, err, len(acts), len(events))
+	}
+	if out, err = call("15s", nil); err != nil || !out.Changed || maxAge(t, js, "KV_mesh-controller_lease") != 15*time.Second || len(acts) != 2 {
+		t.Errorf("restore: %+v %v", out, err)
+	}
+	// A hand act the seat does not take is said loudly, not as success.
+	_, err = SetBucketExpiry(context.Background(), bus, said, func(handacts.Request) error { return errors.New("no responders") }, map[string]any{"bucket": "mesh-controller_lease", "max_age": "none", "why": "x"})
+	if err == nil || !strings.Contains(err.Error(), "WAS changed") {
+		t.Errorf("an unrecorded act was answered %v", err)
+	}
+}
+
+// 774/S6/AC2: no why, a stream that is no bucket, and a bad expiry change nothing and say why.
+func TestABucketExpiryRefusesWithoutWhyOrABucket(t *testing.T) {
+	bus, js := expiryBus(t)
+	record := func(handacts.Request) error { t.Fatal("an act was recorded for a refused call"); return nil }
+	said := func(string, any) error { t.Fatal("an event was said for a refused call"); return nil }
+	for _, c := range []struct {
+		args map[string]any
+		want string
+	}{
+		{map[string]any{"bucket": "mesh-controller_lease", "max_age": "none"}, "why is required"},
+		{map[string]any{"bucket": "mesh-controller_lease", "max_age": "none", "why": " "}, "why is required"},
+		{map[string]any{"bucket": "ASSIGNMENTS", "max_age": "none", "why": "x"}, "no key-value bucket"},
+		{map[string]any{"bucket": "nothing", "max_age": "none", "why": "x"}, "no key-value bucket"},
+		{map[string]any{"bucket": "a.b", "max_age": "none", "why": "x"}, "bucket is required"},
+		{map[string]any{"bucket": "mesh-controller_lease", "max_age": "", "why": "x"}, "max_age is required"},
+		{map[string]any{"bucket": "mesh-controller_lease", "max_age": "-5s", "why": "x"}, "max_age is required"},
+	} {
+		if _, err := SetBucketExpiry(context.Background(), bus, said, record, c.args); err == nil || !strings.Contains(err.Error(), c.want) {
+			t.Errorf("%v: %v, want a refusal saying %q", c.args, err, c.want)
+		}
+	}
+	if got := maxAge(t, js, "KV_mesh-controller_lease"); got != 15*time.Second {
+		t.Errorf("a refused call left the expiry at %s", got)
+	}
+}

modules/nats/cmd/nats-tools/main.go

@@ -152,7 +152,7 @@ func tools(monitor Monitor, users func(context.Context) ([]byte, error), written
 }
 
 func main() {
-	if err := stdio.Serve("", append(tools(LiveMonitor(), ReadUsers, UsersWritten), removalTools(LiveBus(), SaidByTheRuntime)...)); err != nil {
+	if err := stdio.Serve("", append(append(tools(LiveMonitor(), ReadUsers, UsersWritten), removalTools(LiveBus(), SaidByTheRuntime)...), expiryTools(LiveBus(), SaidByTheRuntime, RecordedBySeat)...)); err != nil {
 		fmt.Fprintln(os.Stderr, err)
 		os.Exit(1)
 	}

modules/nats/cmd/nats-tools/manifest_test.go

@@ -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", "nats_stream_remove"} {
+	for _, want := range []string{"nats_consumer_remove", "nats_bucket_remove", "nats_stream_remove", "nats_bucket_expiry"} {
 		if !slices.Contains(m.Tools, want) {
 			t.Errorf("tools do not list %s", want)
 		}
 	}
-	for _, want := range []string{"consumer-removed", "bucket-keys-removed", "stream-removed"} {
+	for _, want := range []string{"consumer-removed", "bucket-keys-removed", "stream-removed", "bucket-expiry-set"} {
 		if !slices.Contains(m.Emits, want) {
 			t.Errorf("emits do not list %s: the removal tools say it as the audit line", want)
 		}

modules/nats/module.json

@@ -31,7 +31,11 @@
   "emits": [
     "consumer-removed",
     "bucket-keys-removed",
-    "stream-removed"
+    "stream-removed",
+    "bucket-expiry-set"
+  ],
+  "invokes": [
+    "seat:mesh-hand-acts.hand-act"
   ],
   "consumes": [],
   "listens": [
@@ -56,7 +60,8 @@
     "nats_snapshot_verify",
     "nats_consumer_remove",
     "nats_bucket_remove",
-    "nats_stream_remove"
+    "nats_stream_remove",
+    "nats_bucket_expiry"
   ],
   "guards": [
     8222