diff --git a/internal/store/natsstore/bootstrap.go b/internal/store/natsstore/bootstrap.go index 92059222d75..2ac4cacb40b 100644 --- a/internal/store/natsstore/bootstrap.go +++ b/internal/store/natsstore/bootstrap.go @@ -31,9 +31,14 @@ type DriftError struct { Bucket string // Fields lists each mismatch as `name: have X, want Y`. Fields []string + // Remedy replaces the default advice, for buckets startup never reconciles. + Remedy string } func (e *DriftError) Error() string { + if e.Remedy != "" { + return fmt.Sprintf("bucket %s differs from the expected configuration (%s); %s", e.Bucket, strings.Join(e.Fields, "; "), e.Remedy) + } return fmt.Sprintf("bucket %s differs from the expected configuration (%s); update it to match or set %s=%s to let startup reconcile it", e.Bucket, strings.Join(e.Fields, "; "), EnvStoreCreate, CreateAuto) } @@ -75,7 +80,11 @@ func Bootstrap(ctx context.Context, nc *nats.Conn, settings StoreSettings) (*Boo buckets := make(map[string]jetstream.KeyValue) for _, spec := range settings.Buckets() { - kv, err := EnsureBucket(ctx, js, spec, settings.Create) + ensure := EnsureBucket + if spec.Name == settings.BucketName(BucketRefresh) { + ensure = EnsureRefreshBucket + } + kv, err := ensure(ctx, js, spec, settings.Create) if err != nil { return nil, err } diff --git a/internal/store/natsstore/refresh_bucket.go b/internal/store/natsstore/refresh_bucket.go new file mode 100644 index 00000000000..b866d7696a6 --- /dev/null +++ b/internal/store/natsstore/refresh_bucket.go @@ -0,0 +1,105 @@ +package natsstore + +import ( + "context" + "errors" + "fmt" + + "github.com/nats-io/nats.go/jetstream" +) + +// EnsureRefreshBucket makes the refresh journal bucket exist and verifies it is +// safe to claim grants in. Unlike the other buckets it is never reconciled once +// it exists: a journal that drifted may already have lost claims, so the +// replica refuses to start refresh workers and an operator decides. +// +// jetstream.KeyValueConfig has no field for DenyDelete, DenyPurge or +// AllowRollup, so the KV helpers cannot create the bucket the journal wants. +// CreateKeyValue already sets DenyDelete. In CreateAuto mode this updates the +// underlying KV_ stream to also deny purge and roll-ups, which is what stops +// a KV purge from wiping a claim, and the server refuses to cancel it later. +// In CreateNever mode the operator provisions the stream with those settings. +// +// Known gap: a KV delete is an ordinary publish carrying a delete marker header, +// so no stream setting prevents it. Only publish permissions on the bucket +// subjects can. The journal never deletes, claims treat a marker as an existing +// write, and reads fail closed, so a marker blocks the grant instead of +// reopening it. +func EnsureRefreshBucket(ctx context.Context, js jetstream.JetStream, spec BucketSpec, mode CreateMode) (jetstream.KeyValue, error) { + kv, err := js.KeyValue(ctx, spec.Name) + if errors.Is(err, jetstream.ErrBucketNotFound) { + if mode != CreateAuto { + return nil, fmt.Errorf("%w: %s; provision it or set %s=%s", ErrBucketMissing, spec.Name, EnvStoreCreate, CreateAuto) + } + kv, err = js.CreateOrUpdateKeyValue(ctx, spec.config()) + if err != nil { + // Another replica may have created and hardened it first, which + // makes the reconciling update fail. + var errOpen error + if kv, errOpen = js.KeyValue(ctx, spec.Name); errOpen != nil { + return nil, fmt.Errorf("failed to ensure bucket %s: %w", spec.Name, err) + } + } + } else if err != nil { + return nil, fmt.Errorf("failed to ensure bucket %s: %w", spec.Name, err) + } + + stream, err := js.Stream(ctx, kvStreamName(spec.Name)) + if err != nil { + return nil, fmt.Errorf("failed to read stream of bucket %s: %w", spec.Name, err) + } + cfg := stream.CachedInfo().Config + if mode == CreateAuto && (!cfg.DenyDelete || !cfg.DenyPurge || cfg.AllowRollup) { + cfg.DenyDelete, cfg.DenyPurge, cfg.AllowRollup = true, true, false + if _, err := js.UpdateStream(ctx, cfg); err != nil { + return nil, fmt.Errorf("failed to deny delete and purge on bucket %s: %w", spec.Name, err) + } + } + if err := verifyRefreshBucket(ctx, js, kv, spec); err != nil { + return nil, err + } + return kv, nil +} + +const journalRemedy = "refresh workers will not start; recreate the bucket as documented, since a journal that drifted may have lost claims" + +// kvStreamName is the stream JetStream keeps a bucket in. +func kvStreamName(bucket string) string { return "KV_" + bucket } + +// verifyRefreshBucket checks the shape every bucket has plus what only the +// journal needs: no limit that can discard a claim and no way to delete or purge one. +func verifyRefreshBucket(ctx context.Context, js jetstream.JetStream, kv jetstream.KeyValue, spec BucketSpec) error { + if err := verifyBucket(ctx, kv, spec); err != nil { + var drift *DriftError + if errors.As(err, &drift) { + drift.Remedy = journalRemedy + } + return err + } + stream, err := js.Stream(ctx, kvStreamName(spec.Name)) + if err != nil { + return fmt.Errorf("failed to read stream of bucket %s: %w", spec.Name, err) + } + info, err := stream.Info(ctx) + if err != nil { + return fmt.Errorf("failed to read stream of bucket %s: %w", spec.Name, err) + } + cfg := info.Config + + var drift []string + check := func(name string, have, want any) { + if have != want { + drift = append(drift, fmt.Sprintf("%s: have %v, want %v", name, have, want)) + } + } + check("max messages", cfg.MaxMsgs, int64(-1)) + check("max bytes", cfg.MaxBytes, int64(-1)) + check("per-message ttl", cfg.AllowMsgTTL, false) + check("deny delete", cfg.DenyDelete, true) + check("deny purge", cfg.DenyPurge, true) + check("allow roll-up", cfg.AllowRollup, false) + if len(drift) > 0 { + return &DriftError{Bucket: spec.Name, Fields: drift, Remedy: journalRemedy} + } + return nil +} diff --git a/internal/store/natsstore/refresh_bucket_test.go b/internal/store/natsstore/refresh_bucket_test.go new file mode 100644 index 00000000000..0e44438e5a6 --- /dev/null +++ b/internal/store/natsstore/refresh_bucket_test.go @@ -0,0 +1,146 @@ +package natsstore + +import ( + "errors" + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "github.com/router-for-me/CLIProxyAPI/v8/internal/store/natsstore/natstest" +) + +func refreshStreamConfig(t *testing.T, boot *Bootstrapped, settings StoreSettings) jetstream.StreamConfig { + t.Helper() + stream, err := boot.JS.Stream(testContext(t), kvStreamName(settings.BucketName(BucketRefresh))) + if err != nil { + t.Fatal(err) + } + info, err := stream.Info(testContext(t)) + if err != nil { + t.Fatal(err) + } + return info.Config +} + +func TestBootstrapDeniesDeleteAndPurgeOnJournal(t *testing.T) { + srv := natstest.Start(t) + settings := autoSettings("hardened") + ctx := testContext(t) + nc, err := connect(t, map[string]string{EnvURL: srv.URL}, 5*time.Second) + if err != nil { + t.Fatal(err) + } + boot, err := Bootstrap(ctx, nc, settings) + if err != nil { + t.Fatal(err) + } + cfg := refreshStreamConfig(t, boot, settings) + if !cfg.DenyDelete || !cfg.DenyPurge || cfg.AllowRollup { + t.Fatalf("journal stream deny delete %v, deny purge %v, allow roll-up %v", cfg.DenyDelete, cfg.DenyPurge, cfg.AllowRollup) + } + + kv := boot.Bucket(settings, BucketRefresh) + if _, err := kv.Create(ctx, "claim", []byte("pending")); err != nil { + t.Fatal(err) + } + if err := kv.Purge(ctx, "claim"); err == nil { + t.Fatal("a KV purge removed a claim") + } + stream, err := boot.JS.Stream(ctx, kvStreamName(settings.BucketName(BucketRefresh))) + if err != nil { + t.Fatal(err) + } + if err := stream.Purge(ctx); err == nil { + t.Fatal("a stream purge removed the journal") + } + if err := stream.DeleteMsg(ctx, 1); err == nil { + t.Fatal("a message delete removed a claim") + } + + // Starting again against the hardened bucket reconciles nothing and succeeds. + if _, err := Bootstrap(ctx, nc, settings); err != nil { + t.Fatalf("second bootstrap: %v", err) + } + if entry, err := kv.Get(ctx, "claim"); err != nil || string(entry.Value()) != "pending" { + t.Fatalf("claim after second bootstrap = %v, %v", entry, err) + } +} + +func TestEnsureRefreshBucketRefusesMisconfiguredJournal(t *testing.T) { + srv := natstest.Start(t) + nc, err := connect(t, map[string]string{EnvURL: srv.URL}, 5*time.Second) + if err != nil { + t.Fatal(err) + } + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + ctx := testContext(t) + want := autoSettings("probe").Buckets()[2] + if want.Name != "probe-refresh" { + t.Fatalf("unexpected bucket order: %s", want.Name) + } + + // A bucket the plain KV helpers create is missing only the purge denial. + good := func(name string) jetstream.KeyValueConfig { + cfg := want.config() + cfg.Bucket = name + return cfg + } + cases := []struct { + name string + mutate func(*jetstream.KeyValueConfig) + field string + }{ + {"ttl", func(c *jetstream.KeyValueConfig) { c.TTL = time.Hour }, "ttl"}, + {"memory", func(c *jetstream.KeyValueConfig) { c.Storage = jetstream.MemoryStorage }, "storage"}, + {"history", func(c *jetstream.KeyValueConfig) { c.History = 5 }, "history"}, + {"size limit", func(c *jetstream.KeyValueConfig) { c.MaxBytes = 1 << 20 }, "max bytes"}, + {"marker ttl", func(c *jetstream.KeyValueConfig) { c.LimitMarkerTTL = time.Minute }, "limit marker ttl"}, + {"purge", func(*jetstream.KeyValueConfig) {}, "deny purge"}, + } + for _, tc := range cases { + for _, mode := range []CreateMode{CreateNever, CreateAuto} { + name := strings.ReplaceAll(tc.name, " ", "") + string(mode) + cfg := good(name) + tc.mutate(&cfg) + if _, err := js.CreateKeyValue(ctx, cfg); err != nil { + t.Fatalf("%s: %v", tc.name, err) + } + spec := want + spec.Name = name + _, err := EnsureRefreshBucket(ctx, js, spec, mode) + var drift *DriftError + switch { + case tc.name == "purge" && mode == CreateAuto: + // Hardening is the one thing auto mode fixes. + if err != nil { + t.Errorf("%s/%s: %v", tc.name, mode, err) + } + case !errors.As(err, &drift) || !strings.Contains(strings.Join(drift.Fields, ";"), tc.field): + t.Errorf("%s/%s: err = %v, want drift on %s", tc.name, mode, err, tc.field) + case strings.Contains(drift.Error(), EnvStoreCreate): + t.Errorf("%s/%s: advice points at %s, which does not reconcile the journal: %v", tc.name, mode, EnvStoreCreate, err) + } + } + } + + spec := want + spec.Name = "absent-refresh" + if _, err := EnsureRefreshBucket(ctx, js, spec, CreateNever); !errors.Is(err, ErrBucketMissing) { + t.Fatalf("missing bucket in never mode = %v, want ErrBucketMissing", err) + } + // The replica count is part of the expected shape. + if _, err := js.CreateKeyValue(ctx, good("replicas")); err != nil { + t.Fatal(err) + } + spec = want + spec.Name = "replicas" + spec.Replicas = 3 + if _, err := EnsureRefreshBucket(ctx, js, spec, CreateNever); err == nil || !strings.Contains(err.Error(), "replicas") { + t.Fatalf("replica mismatch = %v, want drift on replicas", err) + } +}