feat(natsstore): refuse a refresh journal bucket that can lose claims

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
 		}

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
+}

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)
+	}
+}