chore: run settlement/cache poll reads without stalling on DDL

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

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 {

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 {