feat(postgres): add stock-PostgreSQL storage backend (KPEP-0001)

.gitignore

@@ -2,3 +2,6 @@
 # Test coverage artifacts
 coverage.html
 coverage.out
+
+# IDE / editor
+.idea/

backends/postgres/config.go

@@ -0,0 +1,94 @@
+package postgres
+
+import (
+	"context"
+	"fmt"
+	"runtime"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// Config configures a PostgreSQL backend. DSN is a libpq/pgx connection
+// string (or URL); pgx will honor multi-host DSNs by trying hosts in order.
+type Config struct {
+	DSN string
+
+	// Database, when set, overrides the database named in the DSN after the
+	// connection config is parsed. Leave empty to use the DSN's database.
+	Database string
+
+	// MaxConns caps the pgxpool size. Zero picks 4 * GOMAXPROCS.
+	MaxConns int32
+
+	// MaxConnLifetime bounds how long any pool connection stays open.
+	// Finite values are required so rolling restarts don't strand conns.
+	MaxConnLifetime time.Duration
+
+	// MaxConnLifetimeJitter randomizes lifetimes so a pool doesn't rotate
+	// its connections in lockstep. Recommended ~10% of MaxConnLifetime.
+	MaxConnLifetimeJitter time.Duration
+
+	// MaxConnIdleTime closes idle connections after this interval.
+	MaxConnIdleTime time.Duration
+
+	// HealthCheckPeriod controls how often the pool prunes broken conns.
+	HealthCheckPeriod time.Duration
+}
+
+// buildPoolConfig returns a pgxpool.Config with our defaults applied.
+func (c Config) buildPoolConfig() (*pgxpool.Config, error) {
+	if c.DSN == "" {
+		return nil, fmt.Errorf("postgres: DSN is required")
+	}
+	cfg, err := pgxpool.ParseConfig(c.DSN)
+	if err != nil {
+		return nil, fmt.Errorf("postgres: parse DSN: %w", err)
+	}
+	cfg.MaxConns = c.MaxConns
+	if cfg.MaxConns <= 0 {
+		cfg.MaxConns = int32(runtime.GOMAXPROCS(0) * 4)
+	}
+	cfg.MaxConnLifetime = c.MaxConnLifetime
+	if cfg.MaxConnLifetime <= 0 {
+		cfg.MaxConnLifetime = 5 * time.Minute
+	}
+	cfg.MaxConnLifetimeJitter = c.MaxConnLifetimeJitter
+	if cfg.MaxConnLifetimeJitter <= 0 {
+		cfg.MaxConnLifetimeJitter = 30 * time.Second
+	}
+	cfg.MaxConnIdleTime = c.MaxConnIdleTime
+	if cfg.MaxConnIdleTime <= 0 {
+		cfg.MaxConnIdleTime = 5 * time.Minute
+	}
+	cfg.HealthCheckPeriod = c.HealthCheckPeriod
+	if cfg.HealthCheckPeriod <= 0 {
+		cfg.HealthCheckPeriod = 30 * time.Second
+	}
+	if c.Database != "" {
+		cfg.ConnConfig.Database = c.Database
+	}
+	return cfg, nil
+}
+
+// NewPool constructs a pgxpool.Pool from the config. Callers own the pool
+// and must call Close when done.
+func (c Config) NewPool(ctx context.Context) (*pgxpool.Pool, error) {
+	poolCfg, err := c.buildPoolConfig()
+	if err != nil {
+		return nil, err
+	}
+	return pgxpool.NewWithConfig(ctx, poolCfg)
+}
+
+// ConnConfig returns the underlying single-connection config. The watch
+// feed uses this for a dedicated LISTEN connection that lives outside the
+// pool — a connection blocked in WaitForNotification can't be shared.
+func (c Config) ConnConfig() (*pgx.ConnConfig, error) {
+	poolCfg, err := c.buildPoolConfig()
+	if err != nil {
+		return nil, err
+	}
+	return poolCfg.ConnConfig, nil
+}

backends/postgres/conformance_test.go

@@ -0,0 +1,233 @@
+package postgres
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"k8s.io/apimachinery/pkg/api/apitesting"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apiserver/pkg/apis/example"
+	examplev1 "k8s.io/apiserver/pkg/apis/example/v1"
+	"k8s.io/apiserver/pkg/storage"
+	storagetesting "k8s.io/apiserver/pkg/storage/testing"
+	"k8s.io/apiserver/pkg/storage/value/encrypt/identity"
+)
+
+// Conformance tests run the upstream storage.Interface suite against the
+// PostgreSQL backend. Each TestConformance* delegates to a RunTest* from
+// staging/.../pkg/storage/testing — the same contract etcd3, Spanner, and
+// Cockroach ship against.
+
+var conformanceScheme = runtime.NewScheme()
+var conformanceCodecs = serializer.NewCodecFactory(conformanceScheme)
+
+func init() {
+	metav1.AddToGroupVersion(conformanceScheme, metav1.SchemeGroupVersion)
+	utilruntime.Must(example.AddToScheme(conformanceScheme))
+	utilruntime.Must(examplev1.AddToScheme(conformanceScheme))
+}
+
+// setupConformanceStore returns a Postgres-backed store configured the way
+// upstream conformance tests expect: Pod codec, "/pods/" resource prefix,
+// identity transformer, no wrapDecodedObject hook. Wired to a live feed so
+// watch tests see real events.
+func setupConformanceStore(t *testing.T) (context.Context, *store) {
+	t.Helper()
+	cfg := freshConfig(t)
+	ctx := context.Background()
+	if err := EnsureSchema(ctx, cfg); err != nil {
+		t.Fatalf("EnsureSchema: %v", err)
+	}
+	pool, err := cfg.NewPool(ctx)
+	if err != nil {
+		t.Fatalf("NewPool: %v", err)
+	}
+	t.Cleanup(pool.Close)
+
+	codec := apitesting.TestCodec(conformanceCodecs, examplev1.SchemeGroupVersion)
+	s := NewStore(
+		pool,
+		codec,
+		func() runtime.Object { return &example.Pod{} },
+		func() runtime.Object { return &example.PodList{} },
+		"/registry",
+		"/pods/",
+		identity.NewEncryptCheckTransformer(),
+		nil,
+	)
+
+	connCfg, err := cfg.ConnConfig()
+	if err != nil {
+		t.Fatalf("ConnConfig: %v", err)
+	}
+	feed := NewFeed(pool, connCfg)
+	feed.Start(context.Background())
+	s.SetFeed(feed)
+	t.Cleanup(feed.Stop)
+	waitForFeedReady(t, feed, 5*time.Second)
+
+	scanner := NewTTLScanner(pool, 0)
+	scanner.Start(context.Background())
+	t.Cleanup(scanner.Stop)
+
+	return ctx, s
+}
+
+// waitForFeedReady blocks until the feed has completed at least one poll,
+// i.e. its safe watermark has advanced past the empty log (the init sentinel
+// guarantees id >= 1). This proves the poll loop is live before a test writes.
+func waitForFeedReady(t *testing.T, f *Feed, d time.Duration) {
+	t.Helper()
+	deadline := time.Now().Add(d)
+	for time.Now().Before(deadline) {
+		if f.SafeRV() > 0 {
+			return
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	t.Fatalf("feed never reached ready state within %v", d)
+}
+
+// Suite-required hooks. The append-only log retains full history, so there
+// is no separate compaction to validate; the suite's compaction-dependent
+// subtests self-skip when compaction is nil.
+
+func noopKeyValidation(ctx context.Context, t *testing.T, key string) {}
+
+func noopIncreaseRV(ctx context.Context, t *testing.T) int64 { return 0 }
+
+func TestConformance_Create(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestCreate(ctx, t, s, noopKeyValidation)
+}
+
+func TestConformance_CreateWithTTL(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestCreateWithTTL(ctx, t, s)
+}
+
+func TestConformance_CreateWithKeyExist(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestCreateWithKeyExist(ctx, t, s)
+}
+
+func TestConformance_Get(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestGet(ctx, t, s)
+}
+
+func TestConformance_UnconditionalDelete(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestUnconditionalDelete(ctx, t, s)
+}
+
+func TestConformance_ConditionalDelete(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestConditionalDelete(ctx, t, s)
+}
+
+func TestConformance_DeleteWithSuggestion(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestDeleteWithSuggestion(ctx, t, s)
+}
+
+func TestConformance_DeleteWithSuggestionAndConflict(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestDeleteWithSuggestionAndConflict(ctx, t, s)
+}
+
+func TestConformance_DeleteWithConflict(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestDeleteWithConflict(ctx, t, s)
+}
+
+func TestConformance_GuaranteedUpdate(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestGuaranteedUpdate(ctx, t, &storeWithPrefixTransformer{Interface: s}, noopKeyValidation)
+}
+
+func TestConformance_GuaranteedUpdateWithTTL(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestGuaranteedUpdateWithTTL(ctx, t, s)
+}
+
+func TestConformance_GuaranteedUpdateWithConflict(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestGuaranteedUpdateWithConflict(ctx, t, s)
+}
+
+func TestConformance_GuaranteedUpdateWithSuggestionAndConflict(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestGuaranteedUpdateWithSuggestionAndConflict(ctx, t, s)
+}
+
+func TestConformance_GetListNonRecursive(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestGetListNonRecursive(ctx, t, noopIncreaseRV, s)
+}
+
+func TestConformance_GetListRecursivePrefix(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestGetListRecursivePrefix(ctx, t, s)
+}
+
+func TestConformance_Watch(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestWatch(ctx, t, s)
+}
+
+func TestConformance_WatchFromZero(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestWatchFromZero(ctx, t, s, nil)
+}
+
+func TestConformance_DeleteTriggerWatch(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestDeleteTriggerWatch(ctx, t, s)
+}
+
+func TestConformance_WatchFromNonZero(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestWatchFromNonZero(ctx, t, s)
+}
+
+func TestConformance_WatchContextCancel(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestWatchContextCancel(ctx, t, s)
+}
+
+func TestConformance_WatcherTimeout(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestWatcherTimeout(ctx, t, s)
+}
+
+func TestConformance_WatchDeleteEventObjectHaveLatestRV(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestWatchDeleteEventObjectHaveLatestRV(ctx, t, s)
+}
+
+func TestConformance_ClusterScopedWatch(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestClusterScopedWatch(ctx, t, s)
+}
+
+func TestConformance_NamespaceScopedWatch(t *testing.T) {
+	ctx, s := setupConformanceStore(t)
+	storagetesting.RunTestNamespaceScopedWatch(ctx, t, s)
+}
+
+// storeWithPrefixTransformer satisfies the suite's
+// InterfaceWithPrefixTransformer. We don't support hot-swapping the
+// transformer, so the modifier is a no-op; the suite's transformer-swap
+// tests are intentionally excluded from this battery.
+type storeWithPrefixTransformer struct {
+	storage.Interface
+}
+
+func (s *storeWithPrefixTransformer) UpdatePrefixTransformer(_ storagetesting.PrefixTransformerModifier) func() {
+	return func() {}
+}

backends/postgres/doc.go

@@ -0,0 +1,28 @@
+// Package postgres implements storage.Interface backed by a stock
+// PostgreSQL server (no extensions, no elevated replication privileges).
+//
+// Unlike the Cockroach and Spanner backends — which keep a single row per
+// key and lean on the engine's own MVCC for resource versions and
+// point-in-time reads — PostgreSQL exposes neither a client-visible commit
+// clock nor time-travel reads. This backend therefore stores an
+// append-only MVCC log: every mutation INSERTs a new row, and the row's
+// BIGSERIAL id IS the Kubernetes resource version. That single choice
+// covers three storage.Interface obligations at once:
+//
+//   - Monotonic resource versions come from the id sequence.
+//   - Point-in-time reads (Get/GetList at an explicit ResourceVersion) are
+//     reconstructed from the retained history via DISTINCT ON (name) with
+//     an `id <= rv` bound — no engine time-travel required.
+//   - Watch is a resumable tail of the log (`WHERE id > cursor`), woken
+//     promptly by LISTEN/NOTIFY and made gap-safe by a transaction-id
+//     watermark (see feed.go) so concurrently-committed rows are never
+//     skipped.
+//
+// Writes run in SERIALIZABLE transactions with an automatic retry on
+// serialization failure (see retry.go), the stock-Postgres equivalent of
+// Cockroach's crdbpgx.ExecuteTx. TTL is enforced by a per-process scanner
+// that appends tombstone rows (Postgres has no built-in row-level TTL).
+//
+// Minimum supported release: PostgreSQL 12 (for txid_snapshot_xmin and
+// aggregate FILTER). Primary test target: PostgreSQL 16.
+package postgres