diff --git a/modules/nats/cmd/nats-tools/expiry.go b/modules/nats/cmd/nats-tools/expiry.go new file mode 100644 index 000000000..217566575 --- /dev/null +++ b/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) + }}} +} diff --git a/modules/nats/cmd/nats-tools/expiry_test.go b/modules/nats/cmd/nats-tools/expiry_test.go new file mode 100644 index 000000000..cc1c6529e --- /dev/null +++ b/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) + } +} diff --git a/modules/nats/cmd/nats-tools/main.go b/modules/nats/cmd/nats-tools/main.go index 933abbeaa..ef376d511 100644 --- a/modules/nats/cmd/nats-tools/main.go +++ b/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) } diff --git a/modules/nats/cmd/nats-tools/manifest_test.go b/modules/nats/cmd/nats-tools/manifest_test.go index dc9f453a0..e6ad1b19b 100644 --- a/modules/nats/cmd/nats-tools/manifest_test.go +++ b/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) } diff --git a/modules/nats/module.json b/modules/nats/module.json index 81bfc6190..fc553dd8b 100644 --- a/modules/nats/module.json +++ b/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