From 6c39173febb0ebbe3bec923385d194bc3b454ac1 Mon Sep 17 00:00:00 2001 From: Michael Buntarman Date: Wed, 1 Jul 2026 18:31:09 +0700 Subject: [PATCH] chore: run settlement/cache poll reads without stalling on DDL --- extensions/tn_cache/pool_config_test.go | 22 ++ extensions/tn_cache/tn_cache.go | 51 ++- extensions/tn_settlement/extension.go | 34 +- .../tn_settlement/internal/engine_ops.go | 301 +++++++----------- .../tn_settlement/internal/engine_ops_test.go | 61 ++-- .../tn_settlement/internal/read_pool.go | 197 ++++++++++++ .../internal/read_pool_pglive_test.go | 100 ++++++ .../tn_settlement/internal/read_pool_test.go | 42 +++ .../settlement_integration_test.go | 10 +- extensions/tn_settlement/tn_settlement.go | 21 +- 10 files changed, 603 insertions(+), 236 deletions(-) create mode 100644 extensions/tn_cache/pool_config_test.go create mode 100644 extensions/tn_settlement/internal/read_pool.go create mode 100644 extensions/tn_settlement/internal/read_pool_pglive_test.go create mode 100644 extensions/tn_settlement/internal/read_pool_test.go diff --git a/extensions/tn_cache/pool_config_test.go b/extensions/tn_cache/pool_config_test.go new file mode 100644 index 000000000..948b7b3d2 --- /dev/null +++ b/extensions/tn_cache/pool_config_test.go @@ -0,0 +1,22 @@ +package tn_cache + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +// TestBuildCachePoolConfig_SetsLockTimeout guards the cache half of the +// settlement/cache interpreter-deadlock fix. The cache's stream/taxonomy reads +// run through the engine interpreter on this independent pool, so a bounded +// lock_timeout on its connections is the only thing that prevents an in-block +// DDL (AccessExclusiveLock) on a cache-read table from deadlocking block +// production. If a future edit drops or clobbers the lock_timeout wiring, this +// fails loudly. +func TestBuildCachePoolConfig_SetsLockTimeout(t *testing.T) { + cfg, err := buildCachePoolConfig("host=127.0.0.1 port=5432 user=u database=d sslmode=disable") + require.NoError(t, err) + require.Equal(t, "3000", cfg.ConnConfig.RuntimeParams["lock_timeout"], + "cache pool must set lock_timeout=3000ms (3s) as a connection startup parameter") + require.Equal(t, cacheLockTimeoutMillis, cfg.ConnConfig.RuntimeParams["lock_timeout"]) +} diff --git a/extensions/tn_cache/tn_cache.go b/extensions/tn_cache/tn_cache.go index a418b4bca..287a3a09a 100644 --- a/extensions/tn_cache/tn_cache.go +++ b/extensions/tn_cache/tn_cache.go @@ -457,6 +457,46 @@ func SetupCacheExtension(ctx context.Context, config *config.ProcessedConfig, en return NewExtension(service.Logger, cacheDB, scheduler, syncChecker, metricsRecorder, engineOps, db, config.Enabled), nil } +// cacheLockTimeoutMillis bounds (in milliseconds) how long a cache read or +// write waits on a Postgres table lock before failing. See the lock_timeout +// rationale in createIndependentConnectionPool. +const cacheLockTimeoutMillis = "3000" + +// buildCachePoolConfig parses the connection string and applies the cache pool +// tuning and the bounded lock_timeout. It is separated from pool creation so a +// unit test can assert the lock_timeout is wired without needing a live database. +func buildCachePoolConfig(connStr string) (*pgxpool.Config, error) { + poolConfig, err := pgxpool.ParseConfig(connStr) + if err != nil { + return nil, fmt.Errorf("parse pool config: %w", err) + } + + // Configure pool specifically for cache operations + poolConfig.MaxConns = 10 // Dedicated connections for cache operations + poolConfig.MinConns = 2 // Keep minimum connections ready + poolConfig.MaxConnLifetime = 30 * time.Minute + poolConfig.MaxConnIdleTime = 5 * time.Minute + + // Bound how long a cache read/write waits on a Postgres table lock. Cache + // reads of streams/taxonomies run through the engine interpreter (to reuse + // composed-stream logic), so they hold the interpreter read lock for the + // duration of the query. If an in-block DDL (ALTER/DROP/etc.) holds an + // AccessExclusiveLock on one of those tables, an unbounded wait would keep the + // interpreter lock held and deadlock block production. A finite lock_timeout + // makes the read fail fast, release the interpreter lock, and retry next tick. + // Ordinary DML never conflicts with a SELECT's AccessShareLock, so this only + // trips during real DDL contention. Cache writes target ext_tn_cache (owned + // exclusively by this extension) and simply retry on the next refresh. + // Set as a startup parameter so it applies to every connection and stays + // introspectable for tests. + if poolConfig.ConnConfig.RuntimeParams == nil { + poolConfig.ConnConfig.RuntimeParams = map[string]string{} + } + poolConfig.ConnConfig.RuntimeParams["lock_timeout"] = cacheLockTimeoutMillis + + return poolConfig, nil +} + // createIndependentConnectionPool creates a dedicated connection pool for cache operations func createIndependentConnectionPool(ctx context.Context, service *common.Service, logger log.Logger) (*pgxpool.Pool, error) { // Check for test DB config override first @@ -470,18 +510,11 @@ func createIndependentConnectionPool(ctx context.Context, service *common.Servic connStr += " password=" + dbConfig.Pass } - // Parse configuration to customize pool settings - poolConfig, err := pgxpool.ParseConfig(connStr) + poolConfig, err := buildCachePoolConfig(connStr) if err != nil { - return nil, fmt.Errorf("parse pool config: %w", err) + return nil, err } - // Configure pool specifically for cache operations - poolConfig.MaxConns = 10 // Dedicated connections for cache operations - poolConfig.MinConns = 2 // Keep minimum connections ready - poolConfig.MaxConnLifetime = 30 * time.Minute - poolConfig.MaxConnIdleTime = 5 * time.Minute - // Create the pool pool, err := pgxpool.NewWithConfig(ctx, poolConfig) if err != nil { diff --git a/extensions/tn_settlement/extension.go b/extensions/tn_settlement/extension.go index a5514d939..9f3dc0fa9 100644 --- a/extensions/tn_settlement/extension.go +++ b/extensions/tn_settlement/extension.go @@ -31,6 +31,7 @@ type Extension struct { // lifecycle wiring engineOps *internal.EngineOperations + readPool *internal.ReadPool // independent read pool backing engineOps' poll reads; closed on shutdown service *common.Service // config snapshot @@ -111,18 +112,19 @@ func NewExtension(logger log.Logger, sched *scheduler.SettlementScheduler) *Exte } } -func (e *Extension) Logger() log.Logger { return e.logger } -func (e *Extension) Scheduler() *scheduler.SettlementScheduler { return e.scheduler } -func (e *Extension) IsLeader() bool { return e.isLeader.Load() } -func (e *Extension) setLeader(v bool) { e.isLeader.Store(v) } -func (e *Extension) EngineOps() *internal.EngineOperations { return e.engineOps } -func (e *Extension) SetEngineOps(ops *internal.EngineOperations) { e.engineOps = ops } -func (e *Extension) Service() *common.Service { return e.service } -func (e *Extension) SetService(s *common.Service) { e.service = s } -func (e *Extension) ConfigEnabled() bool { return e.enabled } -func (e *Extension) Schedule() string { return e.schedule } -func (e *Extension) MaxMarketsPerRun() int { return e.maxMarketsPerRun } -func (e *Extension) RetryAttempts() int { return e.retryAttempts } +func (e *Extension) Logger() log.Logger { return e.logger } +func (e *Extension) Scheduler() *scheduler.SettlementScheduler { return e.scheduler } +func (e *Extension) IsLeader() bool { return e.isLeader.Load() } +func (e *Extension) setLeader(v bool) { e.isLeader.Store(v) } +func (e *Extension) EngineOps() *internal.EngineOperations { return e.engineOps } +func (e *Extension) SetEngineOps(ops *internal.EngineOperations) { e.engineOps = ops } +func (e *Extension) SetReadPool(p *internal.ReadPool) { e.readPool = p } +func (e *Extension) Service() *common.Service { return e.service } +func (e *Extension) SetService(s *common.Service) { e.service = s } +func (e *Extension) ConfigEnabled() bool { return e.enabled } +func (e *Extension) Schedule() string { return e.schedule } +func (e *Extension) MaxMarketsPerRun() int { return e.maxMarketsPerRun } +func (e *Extension) RetryAttempts() int { return e.retryAttempts } func (e *Extension) SetScheduler(s *scheduler.SettlementScheduler) { e.mu.Lock() oldScheduler := e.scheduler @@ -150,8 +152,8 @@ func (e *Extension) SetConfig(enabled bool, schedule string, maxMarkets, retries func (e *Extension) SetReloadIntervalBlocks(v int64) { e.reloadIntervalBlocks = v } func (e *Extension) ReloadIntervalBlocks() int64 { return e.reloadIntervalBlocks } -func (e *Extension) SetLastCheckedHeight(h int64) { atomic.StoreInt64(&e.lastCheckedHeight, h) } -func (e *Extension) LastCheckedHeight() int64 { return atomic.LoadInt64(&e.lastCheckedHeight) } +func (e *Extension) SetLastCheckedHeight(h int64) { atomic.StoreInt64(&e.lastCheckedHeight, h) } +func (e *Extension) LastCheckedHeight() int64 { return atomic.LoadInt64(&e.lastCheckedHeight) } func (e *Extension) SetReloadRetryBackoff(d time.Duration) { e.reloadRetryBackoff = d } func (e *Extension) ReloadRetryBackoff() time.Duration { if e.reloadRetryBackoff == 0 { @@ -348,6 +350,10 @@ func (e *Extension) Close() { if e.scheduler != nil { _ = e.scheduler.Stop() } + // Release the independent read pool's connections. + if e.readPool != nil { + e.readPool.Close() + } // Cancel shutdown context to signal any operations using it e.mu.Lock() if e.shutdownCancel != nil { diff --git a/extensions/tn_settlement/internal/engine_ops.go b/extensions/tn_settlement/internal/engine_ops.go index 9bb8bfaa0..e818830a8 100644 --- a/extensions/tn_settlement/internal/engine_ops.go +++ b/extensions/tn_settlement/internal/engine_ops.go @@ -29,6 +29,7 @@ type EngineOperations struct { logger log.Logger db sql.DB dbPool sql.DelayedReadTxMaker // For fresh read transactions in background jobs + readDB sql.DB // Independent read handle for poll reads; bypasses the engine interpreter lock accounts common.Accounts } @@ -39,11 +40,19 @@ type UnsettledMarket struct { SettleTime int64 // Unix timestamp } -func NewEngineOperations(engine common.Engine, db sql.DB, dbPool sql.DelayedReadTxMaker, accounts common.Accounts, logger log.Logger) *EngineOperations { +// NewEngineOperations builds the settlement engine-ops wrapper. +// +// readDB is an independent read handle (a *ReadPool in production) used for all +// poll-time SELECTs. It runs plain SQL and does NOT go through the engine +// interpreter, so a settlement read can never hold the interpreter lock while +// waiting on a Postgres table lock — the deadlock this fixes. In tests readDB +// may be the platform tx so reads observe the test's uncommitted writes. +func NewEngineOperations(engine common.Engine, db sql.DB, dbPool sql.DelayedReadTxMaker, readDB sql.DB, accounts common.Accounts, logger log.Logger) *EngineOperations { return &EngineOperations{ engine: engine, db: db, dbPool: dbPool, + readDB: readDB, accounts: accounts, logger: logger.New("settlement_ops"), } @@ -59,73 +68,40 @@ func isAccountNotFoundError(err error) bool { return strings.Contains(msg, "not found") || strings.Contains(msg, "no rows") } -// getFreshReadTx returns a fresh database connection for read operations. -// This is critical for background scheduler operations where e.db may be stale/closed. -// Returns: (db, cleanup function, error) -// The cleanup function MUST be called (via defer) to rollback the read transaction. -func (e *EngineOperations) getFreshReadTx(ctx context.Context) (sql.DB, func(), error) { - if e.dbPool != nil { - readTx := e.dbPool.BeginDelayedReadTx() - cleanup := func() { - readTx.Rollback(ctx) - } - return readTx, cleanup, nil +// toInt64 coerces a decoded SQL integer value (INT/INT8/etc. decode to int64) +// to an int64. It returns false for NULLs or unexpected types. +func toInt64(v any) (int64, bool) { + switch n := v.(type) { + case int64: + return n, true + case int: + return int64(n), true + case int32: + return int64(n), true + default: + return 0, false } - // Fallback to stored db (may fail if tx is closed, but better than nothing) - e.logger.Warn("dbPool is nil, falling back to stored db connection (may be stale)") - return e.db, func() {}, nil } // LoadSettlementConfig reads the single-row settlement configuration // Returns (enabled, schedule, maxMarketsPerRun, retryAttempts) -// If table/row missing, returns false, "", 10, 3 and no error +// If table/row missing, returns false, "", 10, 3 and no error. +// +// The read runs on the independent readDB handle (plain SQL, no engine +// interpreter) so it cannot deadlock against in-block DDL. This method is +// invoked both from the background scheduler and, periodically, from within +// EndBlock; the readDB's bounded lock_timeout ensures the EndBlock call fails +// fast (surfacing an error to the caller's retry path) rather than hanging. func (e *EngineOperations) LoadSettlementConfig(ctx context.Context) (bool, string, int, int, error) { - var ( - enabled bool - schedule string - maxMarkets int - retries int - found bool - ) - - // Use fresh read transaction from pool to avoid stale connection issues - // in background scheduler contexts where e.db may be closed - db, cleanup, err := e.getFreshReadTx(ctx) - if err != nil { - return false, "", 10, 3, fmt.Errorf("get fresh read tx: %w", err) + if e.readDB == nil { + return false, "", 10, 3, fmt.Errorf("settlement read handle not initialized") } - defer cleanup() - // Read using engine without engine ctx (owner-level read) - err = e.engine.ExecuteWithoutEngineCtx(ctx, db, + rs, err := e.readDB.Execute(ctx, `SELECT enabled, settlement_schedule, max_markets_per_run, retry_attempts - FROM main.settlement_config WHERE id = 1`, nil, - func(row *common.Row) error { - if len(row.Values) >= 4 { - if v, ok := row.Values[0].(bool); ok { - enabled = v - } - if s, ok := row.Values[1].(string); ok { - schedule = s - } - if m, ok := row.Values[2].(int); ok { - maxMarkets = m - } else if m64, ok := row.Values[2].(int64); ok { - maxMarkets = int(m64) - } - if r, ok := row.Values[3].(int); ok { - retries = r - } else if r64, ok := row.Values[3].(int64); ok { - retries = int(r64) - } - found = true - } - return nil - }) - + FROM main.settlement_config WHERE id = 1`) if err != nil { // tolerate missing table; everything else should surface to caller - // TODO: Use typed error from engine API if available msg := strings.ToLower(err.Error()) if strings.Contains(msg, "settlement_config") && (strings.Contains(msg, "does not exist") || @@ -137,125 +113,96 @@ func (e *EngineOperations) LoadSettlementConfig(ctx context.Context) (bool, stri } return false, "", 10, 3, err } - if !found { + + if len(rs.Rows) == 0 || len(rs.Rows[0]) < 4 { return false, "", 10, 3, nil } + + row := rs.Rows[0] + enabled, _ := row[0].(bool) + schedule, _ := row[1].(string) + maxMarkets := 10 + if n, ok := toInt64(row[2]); ok { + maxMarkets = int(n) + } + retries := 3 + if n, ok := toInt64(row[3]); ok { + retries = int(n) + } return enabled, schedule, maxMarkets, retries, nil } // FindUnsettledMarkets queries for markets past settle_time that haven't been settled yet -// Uses the current Unix timestamp to determine which markets are ready +// Uses the current Unix timestamp to determine which markets are ready. +// +// Reads run on the independent readDB handle (plain SQL, positional params, +// main-schema-qualified) rather than through the engine interpreter. func (e *EngineOperations) FindUnsettledMarkets(ctx context.Context, limit int) ([]*UnsettledMarket, error) { - var markets []*UnsettledMarket + if e.readDB == nil { + return nil, fmt.Errorf("settlement read handle not initialized") + } // Get current Unix timestamp for comparison currentTime := time.Now().Unix() - // Query: SELECT id, hash, settle_time FROM ob_queries - // WHERE settled = FALSE AND settle_time <= $current_time - // ORDER BY settle_time ASC LIMIT $limit - query := ` - SELECT id, hash, settle_time - FROM ob_queries - WHERE settled = false AND settle_time <= $current_time - ORDER BY settle_time ASC - LIMIT $limit - ` - - // Use fresh read transaction from pool to avoid stale connection issues - db, cleanup, err := e.getFreshReadTx(ctx) - if err != nil { - return nil, fmt.Errorf("get fresh read tx: %w", err) - } - defer cleanup() - - err = e.engine.ExecuteWithoutEngineCtx(ctx, db, query, - map[string]any{ - "current_time": currentTime, - "limit": int64(limit), - }, - func(row *common.Row) error { - if len(row.Values) >= 3 { - // Extract values with type assertions - var id int - var hash []byte - var settleTime int64 - - // Handle id (can be int32 or int64) - switch v := row.Values[0].(type) { - case int: - id = v - case int32: - id = int(v) - case int64: - id = int(v) - default: - return fmt.Errorf("unexpected type for id: %T", v) - } - - // Handle hash - var ok bool - hash, ok = row.Values[1].([]byte) - if !ok { - return fmt.Errorf("unexpected type for hash: %T", row.Values[1]) - } - - // Handle settle_time - switch v := row.Values[2].(type) { - case int64: - settleTime = v - case int: - settleTime = int64(v) - default: - return fmt.Errorf("unexpected type for settle_time: %T", v) - } - - market := &UnsettledMarket{ - ID: id, - Hash: hash, - SettleTime: settleTime, - } - markets = append(markets, market) - } - return nil - }) - + rs, err := e.readDB.Execute(ctx, + `SELECT id, hash, settle_time + FROM main.ob_queries + WHERE settled = false AND settle_time <= $1 + ORDER BY settle_time ASC + LIMIT $2`, + currentTime, int64(limit)) if err != nil { return nil, fmt.Errorf("query unsettled markets: %w", err) } + markets := make([]*UnsettledMarket, 0, len(rs.Rows)) + for _, row := range rs.Rows { + if len(row) < 3 { + continue + } + + idVal, ok := toInt64(row[0]) + if !ok { + return nil, fmt.Errorf("unexpected type for id: %T", row[0]) + } + hash, ok := row[1].([]byte) + if !ok { + return nil, fmt.Errorf("unexpected type for hash: %T", row[1]) + } + settleTime, ok := toInt64(row[2]) + if !ok { + return nil, fmt.Errorf("unexpected type for settle_time: %T", row[2]) + } + + markets = append(markets, &UnsettledMarket{ + ID: int(idVal), + Hash: hash, + SettleTime: settleTime, + }) + } + return markets, nil } -// AttestationExists checks if a signed attestation exists for the given hash +// AttestationExists checks if a signed attestation exists for the given hash. +// +// Reads run on the independent readDB handle rather than the engine interpreter. func (e *EngineOperations) AttestationExists(ctx context.Context, marketHash []byte) (bool, error) { - var exists bool - - query := ` - SELECT 1 FROM attestations - WHERE attestation_hash = $hash AND signature IS NOT NULL - LIMIT 1 - ` - - // Use fresh read transaction from pool to avoid stale connection issues - db, cleanup, err := e.getFreshReadTx(ctx) - if err != nil { - return false, fmt.Errorf("get fresh read tx: %w", err) + if e.readDB == nil { + return false, fmt.Errorf("settlement read handle not initialized") } - defer cleanup() - - err = e.engine.ExecuteWithoutEngineCtx(ctx, db, query, - map[string]any{"hash": marketHash}, - func(row *common.Row) error { - exists = true - return nil - }) + rs, err := e.readDB.Execute(ctx, + `SELECT 1 FROM main.attestations + WHERE attestation_hash = $1 AND signature IS NOT NULL + LIMIT 1`, + marketHash) if err != nil { return false, fmt.Errorf("check attestation: %w", err) } - return exists, nil + return len(rs.Rows) > 0, nil } // BroadcastSettleMarketWithRetry broadcasts settle_market transaction with retry logic. @@ -417,46 +364,32 @@ func (e *EngineOperations) broadcastSettleMarketWithFreshNonce( return hash, nil } -// GetMarketQueryComponents fetches and decodes query_components for a market +// GetMarketQueryComponents fetches and decodes query_components for a market. +// +// Reads run on the independent readDB handle rather than the engine interpreter. func (e *EngineOperations) GetMarketQueryComponents(ctx context.Context, queryID int) (*QueryComponents, error) { - var queryComponentsBytes []byte - var foundRow bool - var foundData bool - - // Use fresh read transaction from pool to avoid stale connection issues - db, cleanup, err := e.getFreshReadTx(ctx) - if err != nil { - return nil, fmt.Errorf("get fresh read tx: %w", err) - } - defer cleanup() - - err = e.engine.ExecuteWithoutEngineCtx(ctx, db, - `SELECT query_components FROM ob_queries WHERE id = $query_id`, - map[string]any{"query_id": int64(queryID)}, - func(row *common.Row) error { - if len(row.Values) >= 1 { - foundRow = true - if row.Values[0] == nil { - return fmt.Errorf("query_components is NULL for query_id=%d", queryID) - } - bytes, ok := row.Values[0].([]byte) - if !ok { - return fmt.Errorf("unexpected query_components type: %T", row.Values[0]) - } - queryComponentsBytes = bytes - foundData = true - } - return nil - }) + if e.readDB == nil { + return nil, fmt.Errorf("settlement read handle not initialized") + } + rs, err := e.readDB.Execute(ctx, + `SELECT query_components FROM main.ob_queries WHERE id = $1`, + int64(queryID)) if err != nil { return nil, fmt.Errorf("fetch query_components: %w", err) } - if !foundRow { + + if len(rs.Rows) == 0 || len(rs.Rows[0]) < 1 { return nil, fmt.Errorf("market not found: query_id=%d", queryID) } - if !foundData { - return nil, fmt.Errorf("query_components missing or invalid for query_id=%d", queryID) + + raw := rs.Rows[0][0] + if raw == nil { + return nil, fmt.Errorf("query_components is NULL for query_id=%d", queryID) + } + queryComponentsBytes, ok := raw.([]byte) + if !ok { + return nil, fmt.Errorf("unexpected query_components type: %T", raw) } return decodeQueryComponents(queryComponentsBytes) diff --git a/extensions/tn_settlement/internal/engine_ops_test.go b/extensions/tn_settlement/internal/engine_ops_test.go index c87af40cc..aefcd0f4a 100644 --- a/extensions/tn_settlement/internal/engine_ops_test.go +++ b/extensions/tn_settlement/internal/engine_ops_test.go @@ -12,7 +12,7 @@ import ( gethAbi "github.com/ethereum/go-ethereum/accounts/abi" gethCommon "github.com/ethereum/go-ethereum/common" - "github.com/trufnetwork/kwil-db/common" + "github.com/stretchr/testify/require" "github.com/trufnetwork/kwil-db/core/crypto" "github.com/trufnetwork/kwil-db/core/crypto/auth" "github.com/trufnetwork/kwil-db/core/log" @@ -20,6 +20,29 @@ import ( "github.com/trufnetwork/kwil-db/node/types/sql" ) +// TestPollReads_FailClosedWhenReadDBNil asserts the settlement poll reads fail +// closed when the independent read handle was never built (e.g. the pool could +// not be created at startup). This is the fix's central safety property: reads +// must error out rather than silently fall back to the engine interpreter, which +// is what would reintroduce the deadlock this change eliminates. +func TestPollReads_FailClosedWhenReadDBNil(t *testing.T) { + ops := &EngineOperations{logger: log.New()} // readDB deliberately nil + ctx := context.Background() + const want = "settlement read handle not initialized" + + _, _, _, _, err := ops.LoadSettlementConfig(ctx) + require.ErrorContains(t, err, want, "LoadSettlementConfig must fail closed") + + _, err = ops.FindUnsettledMarkets(ctx, 10) + require.ErrorContains(t, err, want, "FindUnsettledMarkets must fail closed") + + _, err = ops.AttestationExists(ctx, []byte{0x01}) + require.ErrorContains(t, err, want, "AttestationExists must fail closed") + + _, err = ops.GetMarketQueryComponents(ctx, 1) + require.ErrorContains(t, err, want, "GetMarketQueryComponents must fail closed") +} + // ============================================================================= // Mock implementations for testing // ============================================================================= @@ -568,14 +591,14 @@ func TestRequestAttestationForMarket_VerifyTransactionStructure(t *testing.T) { t.Fatalf("Failed to encode query components: %v", err) } - mockEngine := &mockEngineForQueryComponents{ + readDB := &mockReadDBForQueryComponents{ queryComponents: queryComponents, } ops := &EngineOperations{ logger: log.New(), accounts: accounts, - engine: mockEngine, + readDB: readDB, } ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second) @@ -650,29 +673,23 @@ func encodeQueryComponentsForTest(dataProvider, streamID, actionName string, arg return args.Pack(addr, streamIDBytes32, actionName, argsBytes) } -// Mock engine for query components test -type mockEngineForQueryComponents struct { +// Mock read handle for query components test. GetMarketQueryComponents now reads +// via the independent readDB handle (plain SQL), so the test supplies the row +// directly rather than through the engine. +type mockReadDBForQueryComponents struct { queryComponents []byte } -// Compile-time check that mockEngineForQueryComponents implements common.Engine -var _ common.Engine = (*mockEngineForQueryComponents)(nil) +// Compile-time check that mockReadDBForQueryComponents implements sql.DB +var _ sql.DB = (*mockReadDBForQueryComponents)(nil) -func (m *mockEngineForQueryComponents) ExecuteWithoutEngineCtx(ctx context.Context, db sql.DB, stmt string, params map[string]any, fn func(*common.Row) error) error { - // Return mock query_components - row := &common.Row{ - Values: []any{m.queryComponents}, - } - return fn(row) +func (m *mockReadDBForQueryComponents) Execute(ctx context.Context, stmt string, args ...any) (*sql.ResultSet, error) { + return &sql.ResultSet{ + Columns: []string{"query_components"}, + Rows: [][]any{{m.queryComponents}}, + }, nil } -// Satisfy the Engine interface - these are not used in our tests -func (m *mockEngineForQueryComponents) Call(ctx *common.EngineContext, db sql.DB, namespace, action string, args []any, fn func(*common.Row) error) (*common.CallResult, error) { - return nil, nil -} -func (m *mockEngineForQueryComponents) CallWithoutEngineCtx(ctx context.Context, db sql.DB, namespace, action string, args []any, fn func(*common.Row) error) (*common.CallResult, error) { - return nil, nil -} -func (m *mockEngineForQueryComponents) Execute(ctx *common.EngineContext, db sql.DB, stmt string, params map[string]any, fn func(*common.Row) error) error { - return nil +func (m *mockReadDBForQueryComponents) BeginTx(ctx context.Context) (sql.Tx, error) { + return nil, fmt.Errorf("mock read handle does not support transactions") } diff --git a/extensions/tn_settlement/internal/read_pool.go b/extensions/tn_settlement/internal/read_pool.go new file mode 100644 index 000000000..b7353f5f9 --- /dev/null +++ b/extensions/tn_settlement/internal/read_pool.go @@ -0,0 +1,197 @@ +package internal + +import ( + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "github.com/trufnetwork/kwil-db/common" + "github.com/trufnetwork/kwil-db/core/log" + "github.com/trufnetwork/kwil-db/node/pg" + "github.com/trufnetwork/kwil-db/node/types/sql" +) + +// settlementReadLockTimeout bounds how long a settlement poll read waits on a +// Postgres table lock before it fails. +// +// A settlement read only ever blocks when a block is holding an +// AccessExclusiveLock (in-block DDL: ALTER/DROP/etc.) on a table it reads — +// ordinary DML never conflicts with a plain SELECT's AccessShareLock. Bounding +// the wait lets such a read fail fast and release its goroutine instead of +// deadlocking block production against the engine interpreter lock. +const settlementReadLockTimeout = 3 * time.Second + +// ReadPool is a dedicated, read-only Postgres connection pool for the settlement +// scheduler's poll-time reads. +// +// It deliberately bypasses the engine interpreter: reads run as plain SQL via +// Execute on its own connections, so a settlement read can never hold the +// interpreter lock while it waits on a Postgres table lock — the condition that +// previously deadlocked the chain against in-block DDL. Every connection also +// carries a bounded lock_timeout (see settlementReadLockTimeout) so a read that +// runs inside EndBlock (config reload) fails fast rather than self-blocking on +// the block's own uncommitted AccessExclusiveLock. +type ReadPool struct { + pool *pgxpool.Pool +} + +// ReadPool satisfies sql.DB so it can be handed to EngineOperations as its +// read handle. Only Execute is used; BeginTx is intentionally unsupported. +var _ sql.DB = (*ReadPool)(nil) + +// NewReadPool builds an independent read-only connection pool from the node's DB +// config, applying a bounded lock_timeout on every connection. +func NewReadPool(ctx context.Context, service *common.Service, logger log.Logger) (*ReadPool, error) { + dbConfig := service.LocalConfig.DB + + // Build the DSN without the password, then set it on the parsed ConnConfig. + // A password containing spaces or special characters cannot be safely + // interpolated into a key=value connection string. + connStr := fmt.Sprintf("host=%s port=%s user=%s database=%s sslmode=disable", + dbConfig.Host, dbConfig.Port, dbConfig.User, dbConfig.DBName) + + poolConfig, err := buildReadPoolConfig(connStr) + if err != nil { + return nil, err + } + if dbConfig.Pass != "" { + poolConfig.ConnConfig.Password = dbConfig.Pass + } + + return newReadPoolFromConfig(ctx, poolConfig, logger) +} + +// lockTimeoutMillis renders settlementReadLockTimeout as a Postgres +// millisecond string for use as a connection startup parameter. +func lockTimeoutMillis() string { + return fmt.Sprintf("%d", settlementReadLockTimeout.Milliseconds()) +} + +// buildReadPoolConfig parses the connection string and applies the pool tuning +// and the bounded lock_timeout. It is separated from pool creation so a unit +// test can assert the lock_timeout is wired without needing a live database. +func buildReadPoolConfig(connStr string) (*pgxpool.Config, error) { + poolConfig, err := pgxpool.ParseConfig(connStr) + if err != nil { + return nil, fmt.Errorf("parse settlement read pool config: %w", err) + } + + // The settlement poll reads are low-frequency (a handful per scheduler tick), + // so a small dedicated pool is plenty and keeps the connection budget modest. + poolConfig.MaxConns = 4 + poolConfig.MinConns = 1 + poolConfig.MaxConnLifetime = 30 * time.Minute + poolConfig.MaxConnIdleTime = 5 * time.Minute + + // Apply the bounded lock_timeout to every connection as a startup parameter so + // a read blocked by an in-block AccessExclusiveLock fails fast instead of + // hanging. A startup RuntimeParam (rather than an AfterConnect hook) keeps the + // setting introspectable, so a unit test can assert it is present. + if poolConfig.ConnConfig.RuntimeParams == nil { + poolConfig.ConnConfig.RuntimeParams = map[string]string{} + } + poolConfig.ConnConfig.RuntimeParams["lock_timeout"] = lockTimeoutMillis() + + return poolConfig, nil +} + +// newReadPoolFromConnString builds a ReadPool from a pgx connection string. +// Used by tests to exercise the same pool tuning + lock_timeout as NewReadPool. +func newReadPoolFromConnString(ctx context.Context, connStr string, logger log.Logger) (*ReadPool, error) { + poolConfig, err := buildReadPoolConfig(connStr) + if err != nil { + return nil, err + } + return newReadPoolFromConfig(ctx, poolConfig, logger) +} + +// newReadPoolFromConfig creates the pgxpool from an already-built config and +// verifies connectivity before returning. +func newReadPoolFromConfig(ctx context.Context, poolConfig *pgxpool.Config, logger log.Logger) (*ReadPool, error) { + pool, err := pgxpool.NewWithConfig(ctx, poolConfig) + if err != nil { + return nil, fmt.Errorf("create settlement read pool: %w", err) + } + + // Verify connectivity before returning, so a misconfigured pool fails loudly + // here rather than on the first poll read. + conn, err := pool.Acquire(ctx) + if err != nil { + pool.Close() + return nil, fmt.Errorf("test settlement read pool connection: %w", err) + } + conn.Release() + + logger.Info("created independent read pool for settlement", + "max_conns", poolConfig.MaxConns, + "lock_timeout", settlementReadLockTimeout.String()) + + return &ReadPool{pool: pool}, nil +} + +// Execute runs a read-only statement directly on the pool, bypassing the engine +// interpreter. It mirrors the decoding done by node/extensions/tn_local's +// PoolDBWrapper so the returned values match what the engine would yield. +func (p *ReadPool) Execute(ctx context.Context, stmt string, args ...any) (*sql.ResultSet, error) { + conn, err := p.pool.Acquire(ctx) + if err != nil { + return nil, fmt.Errorf("acquire connection: %w", err) + } + defer conn.Release() + + rows, err := conn.Query(ctx, stmt, args...) + if err != nil { + return nil, fmt.Errorf("execute query: %w", err) + } + defer rows.Close() + + oidToDataType := pg.OidTypesMap(conn.Conn().TypeMap()) + + fields := rows.FieldDescriptions() + columns := make([]string, len(fields)) + oids := make([]uint32, len(fields)) + for i, field := range fields { + columns[i] = string(field.Name) + oids[i] = field.DataTypeOID + } + + var resultRows [][]any + for rows.Next() { + values, err := rows.Values() + if err != nil { + return nil, fmt.Errorf("scan row values: %w", err) + } + decodedValues, err := pg.DecodeFromPG(values, oids, oidToDataType) + if err != nil { + return nil, fmt.Errorf("decode values: %w", err) + } + resultRows = append(resultRows, decodedValues) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("rows iteration: %w", err) + } + + return &sql.ResultSet{ + Columns: columns, + Rows: resultRows, + Status: sql.CommandTag{ + Text: rows.CommandTag().String(), + RowsAffected: rows.CommandTag().RowsAffected(), + }, + }, nil +} + +// BeginTx implements sql.TxMaker. The settlement read path issues single- +// statement reads via Execute and never opens transactions, so this is +// intentionally unsupported. +func (p *ReadPool) BeginTx(ctx context.Context) (sql.Tx, error) { + return nil, fmt.Errorf("settlement read pool does not support transactions") +} + +// Close closes the underlying connection pool. +func (p *ReadPool) Close() { + if p != nil && p.pool != nil { + p.pool.Close() + } +} diff --git a/extensions/tn_settlement/internal/read_pool_pglive_test.go b/extensions/tn_settlement/internal/read_pool_pglive_test.go new file mode 100644 index 000000000..8c613f6fd --- /dev/null +++ b/extensions/tn_settlement/internal/read_pool_pglive_test.go @@ -0,0 +1,100 @@ +//go:build pglive + +package internal + +import ( + "context" + "fmt" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/require" + "github.com/trufnetwork/kwil-db/core/log" +) + +// testConnString returns the pgx connection string for the liveness test. It is +// overridable via TN_TEST_PG_CONNSTRING and defaults to the parameters of the +// node's kwil-testing Postgres container. +func testConnString() string { + if s := os.Getenv("TN_TEST_PG_CONNSTRING"); s != "" { + return s + } + return "host=127.0.0.1 port=52853 user=kwild database=kwil_test_db sslmode=disable" +} + +// TestReadPool_LockTimeoutBreaksBlockedRead proves the core liveness property of +// the settlement/cache interpreter-deadlock fix: a ReadPool SELECT blocked by +// another connection's AccessExclusiveLock — exactly what an in-block DDL +// (ALTER/DROP/etc.) holds — fails fast via lock_timeout instead of hanging +// forever. Under the old code the equivalent read (through the engine +// interpreter) would hold the interpreter lock until block commit, deadlocking +// the chain. +// +// Requires a reachable Postgres; skips cleanly otherwise. Run with: +// +// go test -tags 'kwiltest pglive' ./extensions/tn_settlement/internal/ \ +// -run TestReadPool_LockTimeoutBreaksBlockedRead -count=1 +func TestReadPool_LockTimeoutBreaksBlockedRead(t *testing.T) { + ctx := context.Background() + connStr := testConnString() + + probe, err := pgx.Connect(ctx, connStr) + if err != nil { + t.Skipf("no reachable Postgres (%v); set TN_TEST_PG_CONNSTRING to run", err) + } + defer probe.Close(context.Background()) + + const tbl = "tn_settlement_locktimeout_probe" + mustExec := func(conn *pgx.Conn, sqlText string) { + _, e := conn.Exec(ctx, sqlText) + require.NoError(t, e) + } + mustExec(probe, fmt.Sprintf("DROP TABLE IF EXISTS %s", tbl)) + mustExec(probe, fmt.Sprintf("CREATE TABLE %s (id int)", tbl)) + mustExec(probe, fmt.Sprintf("INSERT INTO %s (id) VALUES (1)", tbl)) + defer func() { + _, _ = probe.Exec(context.Background(), fmt.Sprintf("DROP TABLE IF EXISTS %s", tbl)) + }() + + // The pool under test. + pool, err := newReadPoolFromConnString(ctx, connStr, log.New()) + require.NoError(t, err) + defer pool.Close() + + // Sanity: an unobstructed read succeeds. + rs, err := pool.Execute(ctx, fmt.Sprintf("SELECT id FROM %s", tbl)) + require.NoError(t, err) + require.Len(t, rs.Rows, 1) + + // Hold an AccessExclusiveLock on the table from a separate connection — as an + // in-block ALTER/DROP would — without committing. + locker, err := pgx.Connect(ctx, connStr) + require.NoError(t, err) + defer locker.Close(context.Background()) + lockTx, err := locker.Begin(ctx) + require.NoError(t, err) + defer lockTx.Rollback(context.Background()) + _, err = lockTx.Exec(ctx, fmt.Sprintf("LOCK TABLE %s IN ACCESS EXCLUSIVE MODE", tbl)) + require.NoError(t, err) + + // The blocked read must return (fail-fast) rather than hang. lock_timeout is + // settlementReadLockTimeout; allow a generous ceiling before declaring a hang. + done := make(chan error, 1) + start := time.Now() + go func() { + _, e := pool.Execute(context.Background(), fmt.Sprintf("SELECT id FROM %s", tbl)) + done <- e + }() + + select { + case e := <-done: + require.Error(t, e, "read should fail on lock_timeout while the table is exclusively locked") + require.Contains(t, e.Error(), "lock timeout") + require.GreaterOrEqual(t, time.Since(start), settlementReadLockTimeout-time.Second, + "read returned before lock_timeout could have elapsed") + case <-time.After(settlementReadLockTimeout + 10*time.Second): + t.Fatal("ReadPool read did not return: lock_timeout is not bounding the wait (deadlock reproduced)") + } +} diff --git a/extensions/tn_settlement/internal/read_pool_test.go b/extensions/tn_settlement/internal/read_pool_test.go new file mode 100644 index 000000000..b86ec15de --- /dev/null +++ b/extensions/tn_settlement/internal/read_pool_test.go @@ -0,0 +1,42 @@ +package internal + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + "github.com/trufnetwork/kwil-db/core/log" +) + +// TestNewReadPool_FailsClosedOnUnreachableDB verifies that when the independent +// read pool cannot be built (no reachable database), NewReadPool returns an +// error rather than panicking or hanging. The caller (engineReadyHook) treats +// this as fail-closed: readDB stays nil and settlement reads error out, instead +// of silently falling back to the engine interpreter and reintroducing the +// deadlock this pool exists to prevent. +func TestNewReadPool_FailsClosedOnUnreachableDB(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + // Port 1 has nothing listening: connection is refused immediately. + _, err := newReadPoolFromConnString(ctx, + "host=127.0.0.1 port=1 user=nobody database=nodb sslmode=disable connect_timeout=1", + log.New()) + require.Error(t, err) +} + +// TestBuildReadPoolConfig_SetsLockTimeout guards the settlement half of the +// deadlock fix: the read pool's connections must carry a bounded lock_timeout so +// a poll read blocked by in-block DDL fails fast instead of deadlocking block +// production. This is a no-DB regression guard (the pglive liveness test that +// exercises the real behaviour is not run in the default CI tag set), so if a +// future edit drops or clobbers the lock_timeout wiring, this fails loudly. +func TestBuildReadPoolConfig_SetsLockTimeout(t *testing.T) { + cfg, err := buildReadPoolConfig("host=127.0.0.1 port=5432 user=u database=d sslmode=disable") + require.NoError(t, err) + require.Equal(t, "3000", cfg.ConnConfig.RuntimeParams["lock_timeout"], + "settlement read pool must set lock_timeout=3000ms (3s) as a connection startup parameter") + // Keep the wired value in sync with the source-of-truth constant. + require.Equal(t, lockTimeoutMillis(), cfg.ConnConfig.RuntimeParams["lock_timeout"]) +} diff --git a/extensions/tn_settlement/settlement_integration_test.go b/extensions/tn_settlement/settlement_integration_test.go index 224ddd297..c29e6e37f 100644 --- a/extensions/tn_settlement/settlement_integration_test.go +++ b/extensions/tn_settlement/settlement_integration_test.go @@ -132,7 +132,7 @@ func testFindUnsettledMarkets(t *testing.T) func(context.Context, *kwilTesting.P // Test FindUnsettledMarkets accts, err := accounts.InitializeAccountStore(ctx, platform.DB, log.New()) require.NoError(t, err) - ops := internal.NewEngineOperations(platform.Engine, platform.DB, nil, accts, log.New()) + ops := internal.NewEngineOperations(platform.Engine, platform.DB, nil, platform.DB, accts, log.New()) markets, err := ops.FindUnsettledMarkets(ctx, 10) require.NoError(t, err) @@ -205,7 +205,7 @@ func testAttestationExists(t *testing.T) func(context.Context, *kwilTesting.Plat // Test AttestationExists accts, err := accounts.InitializeAccountStore(ctx, platform.DB, log.New()) require.NoError(t, err) - ops := internal.NewEngineOperations(platform.Engine, platform.DB, nil, accts, log.New()) + ops := internal.NewEngineOperations(platform.Engine, platform.DB, nil, platform.DB, accts, log.New()) exists, err := ops.AttestationExists(ctx, attestationHash) require.NoError(t, err) @@ -323,7 +323,7 @@ func testLoadSettlementConfig(t *testing.T) func(context.Context, *kwilTesting.P // Test LoadSettlementConfig (table exists from migration with seeded defaults) accts, err := accounts.InitializeAccountStore(ctx, platform.DB, log.New()) require.NoError(t, err) - ops := internal.NewEngineOperations(platform.Engine, platform.DB, nil, accts, log.New()) + ops := internal.NewEngineOperations(platform.Engine, platform.DB, nil, platform.DB, accts, log.New()) enabled, schedule, maxMarkets, retries, err := ops.LoadSettlementConfig(ctx) require.NoError(t, err) @@ -403,7 +403,7 @@ func testSkipMarketWithoutAttestation(t *testing.T) func(context.Context, *kwilT // Test AttestationExists should return false accts, err := accounts.InitializeAccountStore(ctx, platform.DB, log.New()) require.NoError(t, err) - ops := internal.NewEngineOperations(platform.Engine, platform.DB, nil, accts, log.New()) + ops := internal.NewEngineOperations(platform.Engine, platform.DB, nil, platform.DB, accts, log.New()) exists, err := ops.AttestationExists(ctx, attestationHash) require.NoError(t, err) @@ -492,7 +492,7 @@ func testMultipleMarketsProcessing(t *testing.T) func(context.Context, *kwilTest // Test FindUnsettledMarkets with limit accts, err := accounts.InitializeAccountStore(ctx, platform.DB, log.New()) require.NoError(t, err) - ops := internal.NewEngineOperations(platform.Engine, platform.DB, nil, accts, log.New()) + ops := internal.NewEngineOperations(platform.Engine, platform.DB, nil, platform.DB, accts, log.New()) markets, err := ops.FindUnsettledMarkets(ctx, 2) require.NoError(t, err) diff --git a/extensions/tn_settlement/tn_settlement.go b/extensions/tn_settlement/tn_settlement.go index f986edaee..36f467170 100644 --- a/extensions/tn_settlement/tn_settlement.go +++ b/extensions/tn_settlement/tn_settlement.go @@ -58,9 +58,25 @@ func engineReadyHook(ctx context.Context, app *common.App) error { logger.Warn("app.DB is nil; settlement extension may not be fully operational") } + // Build an independent read-only pool for the scheduler's poll reads. These + // reads bypass the engine interpreter so a settlement read can never hold the + // interpreter lock while waiting on a Postgres table lock — the condition that + // previously deadlocked the chain against in-block DDL. + // + // Fail-closed: if the pool can't be built, readDB stays nil and settlement + // reads return errors (the scheduler skips and retries) rather than falling + // back to the interpreter and reintroducing the deadlock. + var readDB sql.DB + readPool, poolErr := internal.NewReadPool(ctx, app.Service, logger) + if poolErr != nil { + logger.Warn("failed to create settlement read pool; settlement reads disabled until restart", "error", poolErr) + } else { + readDB = readPool + } + // Build engine operations wrapper - // Pass DBPool for fresh read transactions in background jobs - engOps := internal.NewEngineOperations(app.Engine, db, app.Service.DBPool, app.Accounts, app.Service.Logger) + // Pass DBPool for fresh read transactions in background jobs and readDB for poll reads + engOps := internal.NewEngineOperations(app.Engine, db, app.Service.DBPool, readDB, app.Accounts, app.Service.Logger) // Load schedule from config; fall back to defaults if absent enabled, schedule, maxMarkets, retries, _ := engOps.LoadSettlementConfig(ctx) @@ -79,6 +95,7 @@ func engineReadyHook(ctx context.Context, app *common.App) error { ext.logger = logger ext.SetService(app.Service) ext.SetEngineOps(engOps) + ext.SetReadPool(readPool) ext.SetConfig(enabled, schedule, maxMarkets, retries) // Load config from node TOML [extensions.tn_settlement]