diff --git a/docs/configuration.md b/docs/configuration.md index 6523d3592..73fb4f4e6 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -20,6 +20,7 @@ - [Storage Dialect](#storage-dialect) - [Storage Connection Pool](#storage-connection-pool) - [Spirit Run Settings](#spirit-run-settings) +- [Postgres](#postgres) - [PlanetScale mTLS](#planetscale-mtls) - [Storage Schema Changes](#storage-schema-changes) - [Support Channel](#support-channel) @@ -581,6 +582,37 @@ These settings only apply where this server constructs the Spirit engine itself — local-mode MySQL databases. Databases routed to a remote deployment over gRPC run with that deployment's engine settings. +## Postgres + +The `postgres:` block sets the largest table on which the PostgreSQL engine +will execute native-safe DDL. The limit is expressed in bytes and defaults to +1 GiB: + +```yaml +postgres: + native_safe_table_size_limit_bytes: 4294967296 +``` + +The ceiling is SchemaBot's own conservatism about how much work to attempt +under an exclusive lock, not a PostgreSQL limit. A native-safe `ALTER` that +rewrites a table rebuilds its heap, its indexes, and its TOAST data while +holding `ACCESS EXCLUSIVE`, and the size compared against the ceiling is the +total relation size summed across the table's partition tree — indexes and +TOAST included — so an index-heavy table cannot slip under it. Raising the +ceiling converts an up-front refusal into a bounded attempt, not an unbounded +lock: every apply still runs under short `lock_timeout` and +`statement_timeout` budgets, so above the ceiling it is the statement +timeout, not the ceiling, that stops a runaway rewrite. + +The server fails startup validation when +`native_safe_table_size_limit_bytes` is zero or negative. + +The ceiling is process-wide: every PostgreSQL database this server drives +shares the same value, and a database cannot override it in its own metadata. +These settings only apply where this server constructs the PostgreSQL engine +itself — local-mode PostgreSQL databases. Databases routed to a remote +deployment over gRPC run with that deployment's engine settings. + ## PlanetScale mTLS Some PlanetScale-compatible endpoints require mutual TLS: every MySQL diff --git a/pkg/api/config.go b/pkg/api/config.go index a52bcd3af..2359fda27 100644 --- a/pkg/api/config.go +++ b/pkg/api/config.go @@ -17,6 +17,7 @@ import ( "unicode/utf8" "github.com/block/schemabot/pkg/engine" + postgresengine "github.com/block/schemabot/pkg/engine/postgres" "github.com/block/schemabot/pkg/engine/spirit" "github.com/block/schemabot/pkg/inventory" "github.com/block/schemabot/pkg/pendingdrops" @@ -185,6 +186,10 @@ type ServerConfig struct { // Vitess database this server drives, unlike the per-database tls block, // which only covers statically registered databases. PlanetScale PlanetScaleConfig `yaml:"planetscale,omitempty"` + + // Postgres configures process-wide behavior for every PostgreSQL database + // this server drives directly. + Postgres PostgresConfig `yaml:"postgres,omitempty"` } // PendingDropsConfig configures the pending drops quarantine for MySQL/Spirit @@ -1099,6 +1104,29 @@ type PlanetScaleMTLSConfig struct { ClientKey string `yaml:"client_key"` } +// PostgresConfig holds process-wide settings for the PostgreSQL engine. +type PostgresConfig struct { + // NativeSafeTableSizeLimitBytes is the largest table on which the engine + // will execute native-safe DDL. When unset, + // postgres.DefaultNativeSafeTableSizeLimitBytes applies. + NativeSafeTableSizeLimitBytes *int64 `yaml:"native_safe_table_size_limit_bytes,omitempty"` +} + +// NativeSafeTableSizeLimit returns the configured limit or its default. +func (c PostgresConfig) NativeSafeTableSizeLimit() int64 { + if c.NativeSafeTableSizeLimitBytes == nil { + return postgresengine.DefaultNativeSafeTableSizeLimitBytes + } + return *c.NativeSafeTableSizeLimitBytes +} + +func (c PostgresConfig) validate() error { + if c.NativeSafeTableSizeLimitBytes != nil && *c.NativeSafeTableSizeLimitBytes <= 0 { + return fmt.Errorf("postgres.native_safe_table_size_limit_bytes must be positive, got %d", *c.NativeSafeTableSizeLimitBytes) + } + return nil +} + // validate checks that an mtls block, when present, names all three // certificate paths. File readability is checked at startup registration, not // here, so config validation stays filesystem-independent. @@ -1331,6 +1359,9 @@ func (c *ServerConfig) Validate() error { if err := c.PlanetScale.validate(); err != nil { return err } + if err := c.Postgres.validate(); err != nil { + return err + } if err := c.validateRequiredChecksNotAggregate(); err != nil { return err } diff --git a/pkg/api/config_test.go b/pkg/api/config_test.go index e81390223..7733103aa 100644 --- a/pkg/api/config_test.go +++ b/pkg/api/config_test.go @@ -4238,3 +4238,62 @@ planetscale: ClientKey: "/etc/secrets/pca/tls.key", }, cfg.PlanetScale.MTLS) } + +func TestLoadServerConfigPostgresNativeSafeTableSizeLimit(t *testing.T) { + t.Run("unset uses default", func(t *testing.T) { + cfg := PostgresConfig{} + assert.Equal(t, int64(1<<30), cfg.NativeSafeTableSizeLimit()) + }) + + t.Run("configured bytes", func(t *testing.T) { + path := filepath.Join(t.TempDir(), "config.yaml") + require.NoError(t, os.WriteFile(path, []byte(` +databases: + mydb: + type: postgres + environments: + staging: + dsn: postgres://localhost/mydb +postgres: + native_safe_table_size_limit_bytes: 4294967296 +`), 0o600)) + + cfg, err := LoadServerConfigFromFile(path) + require.NoError(t, err) + assert.Equal(t, int64(4<<30), cfg.Postgres.NativeSafeTableSizeLimit()) + }) + + t.Run("non-positive fails validation", func(t *testing.T) { + limit := int64(0) + cfg := ServerConfig{Databases: map[string]DatabaseConfig{ + "mydb": { + Type: storage.DatabaseTypePostgres, + Environments: map[string]EnvironmentConfig{ + "staging": {DSN: "postgres://localhost/mydb"}, + }, + }, + }} + cfg.Postgres.NativeSafeTableSizeLimitBytes = &limit + + err := cfg.Validate() + require.ErrorContains(t, err, "postgres.native_safe_table_size_limit_bytes must be positive") + }) + + t.Run("unparseable fails config load", func(t *testing.T) { + path := filepath.Join(t.TempDir(), "config.yaml") + require.NoError(t, os.WriteFile(path, []byte(` +databases: + mydb: + type: postgres + environments: + staging: + dsn: postgres://localhost/mydb +postgres: + native_safe_table_size_limit_bytes: 4GiB +`), 0o600)) + + _, err := LoadServerConfigFromFile(path) + require.ErrorContains(t, err, "parse config file") + require.ErrorContains(t, err, "cannot unmarshal") + }) +} diff --git a/pkg/api/service.go b/pkg/api/service.go index 7887b895c..5e2a1b324 100644 --- a/pkg/api/service.go +++ b/pkg/api/service.go @@ -600,12 +600,13 @@ func (s *Service) newLocalTernClient(key, database, dbType string, envConfig Env } maps.Copy(metadata, directMetadata) client, err := tern.NewLocalClient(tern.LocalConfig{ - Database: database, - Type: dbType, - TargetDSN: targetDSN, - Metadata: metadata, - WakeOperator: s.wakeOperator, - EngineFactories: s.engineFactories, + Database: database, + Type: dbType, + TargetDSN: targetDSN, + Metadata: metadata, + PostgresNativeSafeTableSizeLimitBytes: s.config.Postgres.NativeSafeTableSizeLimit(), + WakeOperator: s.wakeOperator, + EngineFactories: s.engineFactories, }, s.storage, s.logger) if err != nil { return nil, fmt.Errorf("create local tern client for %s: %w", key, err) diff --git a/pkg/api/service_test.go b/pkg/api/service_test.go index ff696ea98..781bb456b 100644 --- a/pkg/api/service_test.go +++ b/pkg/api/service_test.go @@ -10,6 +10,7 @@ import ( "github.com/stretchr/testify/require" "github.com/block/schemabot/pkg/engine" + postgresengine "github.com/block/schemabot/pkg/engine/postgres" ternv1 "github.com/block/schemabot/pkg/proto/ternv1" "github.com/block/schemabot/pkg/storage" "github.com/block/schemabot/pkg/tern" @@ -468,3 +469,22 @@ func TestNewLocalTernClient_AcceptsWellFormedTokenReference(t *testing.T) { require.NoError(t, err) assert.NotNil(t, client) } + +// The server-level postgres ceiling reaches the engine of every local client +// the control plane builds, so a configured value governs native-safe DDL +// instead of silently reverting to the default. +func TestNewLocalTernClient_ConfiguresPostgresTableSizeLimit(t *testing.T) { + limit := int64(4 << 30) + cfg := &ServerConfig{Postgres: PostgresConfig{NativeSafeTableSizeLimitBytes: &limit}} + service := New(nil, cfg, nil, slog.New(slog.NewTextHandler(io.Discard, nil))) + + envConfig := EnvironmentConfig{DSN: "postgres://localhost:5432/orders"} + client, err := service.newLocalTernClient("orders-staging", "orders", storage.DatabaseTypePostgres, envConfig) + require.NoError(t, err) + + lc, ok := client.(*tern.LocalClient) + require.True(t, ok) + eng, ok := lc.Engine().(*postgresengine.Engine) + require.True(t, ok) + assert.Equal(t, limit, eng.TableSizeLimit()) +} diff --git a/pkg/engine/postgres/apply.go b/pkg/engine/postgres/apply.go index 7b2aa21b8..cf43b6478 100644 --- a/pkg/engine/postgres/apply.go +++ b/pkg/engine/postgres/apply.go @@ -19,7 +19,6 @@ import ( ) const ( - optimisticTableSizeLimit = int64(1 << 30) optimisticLockTimeout = 3 * time.Second optimisticStatementLimit = 30 * time.Second @@ -122,7 +121,7 @@ func (e *Engine) runOptimisticApply(ctx context.Context, conn targetConn, change // survive the request), so boundedness comes from the ceiling instead. ctx, cancel := context.WithTimeout(ctx, optimisticApplyCeiling) defer cancel() - err := executeOptimistic(ctx, conn, change) + err := executeOptimistic(ctx, conn, change, e.tableSizeLimit) if err == nil { e.publishProgress(key, progressResult(engine.StateCompleted, "completed", started, change, ""), logger) return @@ -204,7 +203,7 @@ func classifyRefusal(err error, table string) *refusal { return nil } -func executeOptimistic(ctx context.Context, conn targetConn, change nativeApply) error { +func executeOptimistic(ctx context.Context, conn targetConn, change nativeApply, tableSizeLimit int64) error { poolCfg, err := spritePoolConfig(conn.dsn, conn.caCertPath) if err != nil { return fmt.Errorf("prepare pg-sprite apply pool for table %q: %w", change.table, err) @@ -226,7 +225,7 @@ func executeOptimistic(ctx context.Context, conn targetConn, change nativeApply) if _, err := preflight.CheckPrivileges(ctx, pool, change.namespace, change.table, preflight.Requirement{Tier: tier}); err != nil { return fmt.Errorf("check privileges for PostgreSQL table %q: %w", change.table, err) } - table, err := preflight.CheckTable(ctx, pool, change.namespace, change.table, optimisticTableSizeLimit) + table, err := preflight.CheckTable(ctx, pool, change.namespace, change.table, tableSizeLimit) if err != nil { return fmt.Errorf("preflight PostgreSQL table %q: %w", change.table, err) } diff --git a/pkg/engine/postgres/apply_test.go b/pkg/engine/postgres/apply_test.go index ec4c35499..50748d6c6 100644 --- a/pkg/engine/postgres/apply_test.go +++ b/pkg/engine/postgres/apply_test.go @@ -175,7 +175,7 @@ func TestExecuteOptimisticRefusesUnreadableCABundle(t *testing.T) { caCertPath: filepath.Join(t.TempDir(), "missing.pem"), } - err := executeOptimistic(t.Context(), conn, nativeApply{namespace: "public", table: "widgets", sql: "CREATE TABLE widgets (id bigint PRIMARY KEY)"}) + err := executeOptimistic(t.Context(), conn, nativeApply{namespace: "public", table: "widgets", sql: "CREATE TABLE widgets (id bigint PRIMARY KEY)"}, DefaultNativeSafeTableSizeLimitBytes) require.Error(t, err) assert.Contains(t, err.Error(), "open pg-sprite apply pool") diff --git a/pkg/engine/postgres/postgres.go b/pkg/engine/postgres/postgres.go index cfad460ea..535b37bcb 100644 --- a/pkg/engine/postgres/postgres.go +++ b/pkg/engine/postgres/postgres.go @@ -31,13 +31,36 @@ type Engine struct { // progressKey (the apply's ResumeState.MigrationContext). One engine is // shared for the lifetime of a target, so Progress must answer for the // apply the caller identifies — never for whichever apply wrote last. - progress *engine.ProgressResult - progressKey string + progress *engine.ProgressResult + progressKey string + tableSizeLimit int64 } +// DefaultNativeSafeTableSizeLimitBytes preserves the native-safe execution +// ceiling when the server does not configure one. +const DefaultNativeSafeTableSizeLimitBytes = int64(1 << 30) + // New creates a new PostgreSQL engine. func New() *Engine { - return &Engine{} + return NewWithTableSizeLimit(DefaultNativeSafeTableSizeLimitBytes) +} + +// NewWithTableSizeLimit creates a PostgreSQL engine with the native-safe +// table size ceiling expressed in bytes. Zero means unset and adopts +// DefaultNativeSafeTableSizeLimitBytes. A negative value is kept as-is +// rather than silently replaced with a ceiling the caller did not choose: +// the preflight check rejects a non-positive limit loudly at apply time, +// and server config validation rejects it at startup. +func NewWithTableSizeLimit(tableSizeLimit int64) *Engine { + if tableSizeLimit == 0 { + tableSizeLimit = DefaultNativeSafeTableSizeLimitBytes + } + return &Engine{tableSizeLimit: tableSizeLimit} +} + +// TableSizeLimit exposes the native-safe ceiling for wiring verification and observability. +func (e *Engine) TableSizeLimit() int64 { + return e.tableSizeLimit } // Name returns the engine identifier. diff --git a/pkg/engine/postgres/postgres_integration_test.go b/pkg/engine/postgres/postgres_integration_test.go index 3d03220fe..9fd33ee12 100644 --- a/pkg/engine/postgres/postgres_integration_test.go +++ b/pkg/engine/postgres/postgres_integration_test.go @@ -110,6 +110,23 @@ func TestEngineApplyTableNotFoundRefusal(t *testing.T) { assert.Contains(t, progress.ErrorMessage, "missing_users") } +// TestEngineApplyTableSizeRefusal proves the configured native-safe ceiling +// reaches preflight and permanently refuses an ALTER when the table exceeds it. +func TestEngineApplyTableSizeRefusal(t *testing.T) { + dsn, db := testutil.StartPostgres(t, "size_limit_test") + _, err := db.ExecContext(t.Context(), "CREATE TABLE public.users (id bigint PRIMARY KEY)") + require.NoError(t, err) + + eng := NewWithTableSizeLimit(1) + _, err = eng.Apply(t.Context(), applyRequest(dsn, "users", "ALTER TABLE public.users ADD COLUMN email text")) + require.NoError(t, err) + progress := awaitPostgresProgress(t, eng, "users") + assert.Equal(t, engine.StateFailed, progress.State) + assert.Equal(t, "refused", progress.Metadata["phase"]) + assert.False(t, progress.Retryable, "a size refusal is permanent until the ceiling or target changes") + assert.Contains(t, progress.ErrorMessage, "1-byte threshold") +} + // TestEngineApplyOperationalFailure proves an execution-path error — here an // unreachable target — remains a failed operation rather than being // misclassified as a safety refusal. diff --git a/pkg/engine/postgres/postgres_test.go b/pkg/engine/postgres/postgres_test.go index 067934bd7..876f6cd2a 100644 --- a/pkg/engine/postgres/postgres_test.go +++ b/pkg/engine/postgres/postgres_test.go @@ -249,3 +249,10 @@ func TestLifecycleControlsDeclineAsUnsupported(t *testing.T) { }) } } + +// A zero ceiling means unset and adopts the default, so a zero-valued client +// config preserves the stock ceiling instead of disabling the size guard. +func TestNewWithTableSizeLimitTreatsZeroAsUnset(t *testing.T) { + assert.Equal(t, DefaultNativeSafeTableSizeLimitBytes, NewWithTableSizeLimit(0).TableSizeLimit()) + assert.Equal(t, int64(42), NewWithTableSizeLimit(42).TableSizeLimit()) +} diff --git a/pkg/serve/serve.go b/pkg/serve/serve.go index 2be742663..7f4e1eca3 100644 --- a/pkg/serve/serve.go +++ b/pkg/serve/serve.go @@ -301,6 +301,14 @@ func Build(ctx context.Context, cfg *api.ServerConfig, opts ...Option) (*Server, return nil, err } + // The postgres ceiling only surfaces at apply time (a refusal names it; + // an attempt below it names nothing), so state the effective value once + // at startup where operators can correlate it across pods and config + // revisions. + logger.Info("PostgreSQL engine native-safe table size ceiling in effect", + "limit_bytes", cfg.Postgres.NativeSafeTableSizeLimit(), + "configured", cfg.Postgres.NativeSafeTableSizeLimitBytes != nil) + // Get storage DSN from config (with fallback to the STORAGE_DSN env var, // then MYSQL_DSN) dsn, err := cfg.StorageDSN() @@ -777,6 +785,7 @@ func grpcLocalClientFactory(config *api.ServerConfig, wakeOperator func(applyIde if cfg.Metadata == nil { cfg.Metadata = map[string]string{} } + cfg.PostgresNativeSafeTableSizeLimitBytes = config.Postgres.NativeSafeTableSizeLimit() // Stated either way rather than only when disabled: a data plane that // predates the opt-in default reads an absent key as "quarantine", so // leaving it out during a rolling deploy would quarantine on a diff --git a/pkg/serve/serve_engine_test.go b/pkg/serve/serve_engine_test.go index 97cb0722c..b18a701c2 100644 --- a/pkg/serve/serve_engine_test.go +++ b/pkg/serve/serve_engine_test.go @@ -10,6 +10,8 @@ import ( "github.com/block/schemabot/pkg/api" "github.com/block/schemabot/pkg/engine" + postgresengine "github.com/block/schemabot/pkg/engine/postgres" + "github.com/block/schemabot/pkg/storage" "github.com/block/schemabot/pkg/storage/mysqlstore" "github.com/block/schemabot/pkg/tern" ) @@ -81,6 +83,29 @@ func TestGRPCLocalClientFactoryMergesIntoExistingFactories(t *testing.T) { "a non-nil per-config factory map must not drop the embedder registry") } +// The data-plane client factory applies the server-level postgres ceiling to +// every LocalClient it builds, so a configured value governs native-safe DDL +// on the gRPC/router path instead of silently reverting to the default. +func TestGRPCLocalClientFactoryConfiguresPostgresTableSizeLimit(t *testing.T) { + limit := int64(4 << 30) + factory := grpcLocalClientFactory(&api.ServerConfig{ + Postgres: api.PostgresConfig{NativeSafeTableSizeLimitBytes: &limit}, + }, nil, nil) + + client, err := factory(tern.LocalConfig{ + Database: "orders", + Type: storage.DatabaseTypePostgres, + TargetDSN: "postgres://localhost:5432/orders", + }, mysqlstore.New(nil), slog.New(slog.DiscardHandler)) + require.NoError(t, err) + + lc, ok := client.(*tern.LocalClient) + require.True(t, ok) + eng, ok := lc.Engine().(*postgresengine.Engine) + require.True(t, ok) + assert.Equal(t, limit, eng.TableSizeLimit()) +} + // Without a registered engine, the data-plane client factory fails closed for a // custom database type rather than building a client with no engine. func TestGRPCLocalClientFactoryFailsClosedForUnregisteredType(t *testing.T) { diff --git a/pkg/tern/local_client.go b/pkg/tern/local_client.go index 4a32f0c64..061757c51 100644 --- a/pkg/tern/local_client.go +++ b/pkg/tern/local_client.go @@ -135,6 +135,10 @@ type LocalConfig struct { // TargetDSN is the connection string to the target database for schema changes. TargetDSN string + // PostgresNativeSafeTableSizeLimitBytes is the maximum table size in bytes + // for PostgreSQL native-safe execution. Zero uses the engine default. + PostgresNativeSafeTableSizeLimitBytes int64 + // Metadata holds engine-specific configuration as key-value pairs. // The tern layer does not interpret these — it passes them through to the // engine via Credentials.Metadata and reads specific keys as needed. @@ -308,7 +312,7 @@ func NewLocalClient(cfg LocalConfig, stor storage.Storage, logger *slog.Logger) Settings: spiritSettings, }), planetscaleEngine: psEngine, - postgresEngine: postgres.New(), + postgresEngine: postgres.NewWithTableSizeLimit(cfg.PostgresNativeSafeTableSizeLimitBytes), customEngine: customEngine, psClientFunc: psClientFunc, logger: logger, @@ -2783,6 +2787,13 @@ func (c *LocalClient) getEngine() engine.Engine { } } +// Engine returns the engine that drives this client's database type, exposed +// so callers that assemble a LocalClient can verify the engine settings they +// configured actually reached it. +func (c *LocalClient) Engine() engine.Engine { + return c.getEngine() +} + // Progress returns detailed progress for an active schema change. // Returns ALL tasks for the current apply: completed, running, and pending. // req.ApplyId is required so progress is always scoped to a single apply. diff --git a/pkg/tern/local_client_test.go b/pkg/tern/local_client_test.go index 4fc5ec287..4cbd2b0a4 100644 --- a/pkg/tern/local_client_test.go +++ b/pkg/tern/local_client_test.go @@ -17,6 +17,7 @@ import ( "github.com/block/schemabot/pkg/ddl" "github.com/block/schemabot/pkg/engine" + postgresengine "github.com/block/schemabot/pkg/engine/postgres" ternv1 "github.com/block/schemabot/pkg/proto/ternv1" "github.com/block/schemabot/pkg/psclient" "github.com/block/schemabot/pkg/schema" @@ -3764,6 +3765,18 @@ func TestNewLocalClientUsesPostgresEngine(t *testing.T) { assert.Equal(t, ternv1.Engine_ENGINE_POSTGRES, c.protoEngine()) } +func TestNewLocalClientConfiguresPostgresTableSizeLimit(t *testing.T) { + c, err := NewLocalClient(LocalConfig{ + Database: "orders", + Type: storage.DatabaseTypePostgres, + PostgresNativeSafeTableSizeLimitBytes: 4 << 30, + }, nil, slog.Default()) + require.NoError(t, err) + eng, ok := c.getEngine().(*postgresengine.Engine) + require.True(t, ok) + assert.Equal(t, int64(4<<30), eng.TableSizeLimit()) +} + // A type with no built-in engine and no registered factory fails closed. func TestNewLocalClientErrorsWhenEngineUnregistered(t *testing.T) { _, err := NewLocalClient(LocalConfig{Database: "db", Type: "customengine"}, nil, slog.Default())