Pre-prod hardening 3/4: ingestion lifecycle & resilience

internal/data/ingest_store.go

@@ -24,10 +24,12 @@ const (
 // not exist in ingest_store. The model layer keeps this distinct from the
 // ordinary value-mismatch race so callers can decide what to do — a missing
 // row may be operationally normal (cursor not yet initialized by protocol-setup
-// / protocol-migrate) or a real incident (dropped row, bad restore). The
-// service layer treats this as a soft skip: the `cursor_missing` query-error
-// metric and a per-ledger warn provide the observability signal without
-// killing live ingest. See ingest_live.go PersistLedgerData for the handling.
+// / protocol-migrate) or a real incident (dropped row, bad restore). Live
+// ingestion (see ingest_live.go's casProtocolCursor) only ever calls
+// CompareAndSwap for a cursor its protocolCursorSnapshot believes exists, so
+// from that caller this error always means the genuine incident case: the
+// `cursor_missing` query-error metric recorded below fires accordingly,
+// rather than on every not-yet-initialized ledger.
 var ErrCASCursorMissing = errors.New("ingest_store cursor row missing")
 
 type LedgerRange struct {
@@ -118,6 +120,48 @@ func (m *IngestStoreModel) Update(ctx context.Context, dbTx pgx.Tx, cursorName s
 	return nil
 }
 
+// ErrCursorGuardFailed is returned by UpdateGuarded when the cursor's current
+// value is neither ledger-1 nor ledger (see UpdateGuarded), or the row does
+// not exist. Either way, a writer other than the one calling UpdateGuarded
+// has moved the cursor past what it expected — most commonly a second live
+// ingestion instance that acquired the advisory lock after this session's
+// Postgres session died in a failover (see startLiveIngestion's
+// checkLockSession, which is the primary defense; this guard is the backstop
+// for the race window before that probe observes the dead session).
+var ErrCursorGuardFailed = errors.New("ingest_store guarded cursor update refused: cursor value not owned by this writer")
+
+// UpdateGuarded advances cursorName to ledger only if its current value is ledger-1 (the normal
+// sequential case: this writer is the sole owner and is advancing by exactly one ledger) or
+// ledger itself (the self-value case: the first ledger processed immediately after
+// startLiveIngestion's initializeCursors already set the cursor to this same starting ledger).
+// Any other current value — including a missing row — means a writer other than the caller has
+// moved the cursor, so the swap is refused with ErrCursorGuardFailed instead of silently
+// overwriting a value another instance already advanced (which a blind Update would do).
+func (m *IngestStoreModel) UpdateGuarded(ctx context.Context, dbTx pgx.Tx, cursorName string, ledger uint32) error {
+	const query = `
+		UPDATE ingest_store
+		SET value = $1
+		WHERE key = $2 AND value IN ($3, $4)
+	`
+	newValue := strconv.FormatUint(uint64(ledger), 10)
+	prevValue := strconv.FormatUint(uint64(ledger-1), 10)
+
+	start := time.Now()
+	tag, err := dbTx.Exec(ctx, query, newValue, cursorName, prevValue, newValue)
+	duration := time.Since(start).Seconds()
+	m.Metrics.QueryDuration.WithLabelValues("UpdateGuarded", "ingest_store").Observe(duration)
+	m.Metrics.QueriesTotal.WithLabelValues("UpdateGuarded", "ingest_store").Inc()
+	if err != nil {
+		m.Metrics.QueryErrors.WithLabelValues("UpdateGuarded", "ingest_store", utils.GetDBErrorType(err)).Inc()
+		return fmt.Errorf("guarded update for cursor %s to %d: %w", cursorName, ledger, err)
+	}
+	if tag.RowsAffected() == 0 {
+		m.Metrics.QueryErrors.WithLabelValues("UpdateGuarded", "ingest_store", "cursor_guard_failed").Inc()
+		return fmt.Errorf("guarded update for cursor %s to %d: %w", cursorName, ledger, ErrCursorGuardFailed)
+	}
+	return nil
+}
+
 func (m *IngestStoreModel) CompareAndSwap(ctx context.Context, dbTx pgx.Tx, cursorName string, expectedValue string, newValue string) (bool, error) {
 	// A plain UPDATE returns RowsAffected=0 for both "value mismatch" and
 	// "row missing" — callers treat false as a race loss and mark the

internal/data/ingest_store_test.go

@@ -227,6 +227,99 @@ func Test_IngestStoreModel_UpdateLatestLedgerSynced(t *testing.T) {
 	}
 }
 
+func Test_IngestStoreModel_UpdateGuarded(t *testing.T) {
+	dbt := dbtest.Open(t)
+	defer dbt.Close()
+	ctx := context.Background()
+	dbConnectionPool, err := db.OpenDBConnectionPool(ctx, dbt.DSN)
+	require.NoError(t, err)
+	defer dbConnectionPool.Close()
+
+	const key = "guarded_cursor"
+
+	testCases := []struct {
+		name          string
+		setupDB       func(t *testing.T) // nil means no row inserted
+		ledger        uint32
+		expectErr     bool
+		expectedValue string // DB value after the call; only checked when !expectErr
+	}{
+		{
+			// Normal sequential case: current value is ledger-1.
+			name: "advances_from_ledger_minus_one",
+			setupDB: func(t *testing.T) {
+				_, err := dbConnectionPool.Exec(ctx, `INSERT INTO ingest_store (key, value) VALUES ($1, '99')`, key)
+				require.NoError(t, err)
+			},
+			ledger:        100,
+			expectedValue: "100",
+		},
+		{
+			// Self-value case: the first ledger processed right after
+			// startLiveIngestion's initializeCursors already set the cursor to
+			// this same starting ledger.
+			name: "self_value_is_a_noop_success",
+			setupDB: func(t *testing.T) {
+				_, err := dbConnectionPool.Exec(ctx, `INSERT INTO ingest_store (key, value) VALUES ($1, '100')`, key)
+				require.NoError(t, err)
+			},
+			ledger:        100,
+			expectedValue: "100",
+		},
+		{
+			// Regression guard: a second writer already advanced the cursor past
+			// what this caller expected (e.g. a second live-ingestion instance
+			// that acquired the lock after this session's Postgres session died
+			// in a failover). Must refuse rather than overwrite.
+			name: "refuses_when_cursor_already_advanced_past_expected",
+			setupDB: func(t *testing.T) {
+				_, err := dbConnectionPool.Exec(ctx, `INSERT INTO ingest_store (key, value) VALUES ($1, '105')`, key)
+				require.NoError(t, err)
+			},
+			ledger:    100,
+			expectErr: true,
+		},
+		{
+			name:      "errors_when_row_missing",
+			ledger:    100,
+			expectErr: true,
+		},
+	}
+
+	dbMetrics := metrics.NewMetrics(prometheus.NewRegistry()).DB
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := dbConnectionPool.Exec(ctx, "DELETE FROM ingest_store")
+			require.NoError(t, err)
+
+			m := &IngestStoreModel{
+				DB:      dbConnectionPool,
+				Metrics: dbMetrics,
+			}
+			if tc.setupDB != nil {
+				tc.setupDB(t)
+			}
+
+			err = db.RunInTransaction(ctx, m.DB, func(dbTx pgx.Tx) error {
+				return m.UpdateGuarded(ctx, dbTx, key, tc.ledger)
+			})
+
+			if tc.expectErr {
+				require.Error(t, err)
+				assert.True(t, errors.Is(err, ErrCursorGuardFailed), "expected ErrCursorGuardFailed, got %v", err)
+				return
+			}
+			require.NoError(t, err)
+
+			var dbValue string
+			scanErr := m.DB.QueryRow(ctx, `SELECT value FROM ingest_store WHERE key = $1`, key).Scan(&dbValue)
+			require.NoError(t, scanErr)
+			assert.Equal(t, tc.expectedValue, dbValue)
+		})
+	}
+}
+
 func Test_IngestStoreModel_CompareAndSwap(t *testing.T) {
 	dbt := dbtest.Open(t)
 	defer dbt.Close()

internal/data/mocks.go

@@ -9,6 +9,7 @@ import (
 	"github.com/jackc/pgx/v5"
 	"github.com/stretchr/testify/mock"
 
+	"github.com/stellar/wallet-backend/internal/db"
 	"github.com/stellar/wallet-backend/internal/indexer/types"
 )
 
@@ -252,8 +253,8 @@ func (m *ProtocolWasmsModelMock) GetUnclassified(ctx context.Context) ([]Protoco
 	return args.Get(0).([]ProtocolWasms), args.Error(1)
 }
 
-func (m *ProtocolWasmsModelMock) GetClassifiedByHashes(ctx context.Context, dbTx pgx.Tx, hashes []types.HashBytea) (map[types.HashBytea]string, error) {
-	args := m.Called(ctx, dbTx, hashes)
+func (m *ProtocolWasmsModelMock) GetClassifiedByHashes(ctx context.Context, q db.Querier, hashes []types.HashBytea) (map[types.HashBytea]string, error) {
+	args := m.Called(ctx, q, hashes)
 	if args.Get(0) == nil {
 		return nil, args.Error(1)
 	}

internal/data/protocol_wasms.go

@@ -25,7 +25,7 @@ type ProtocolWasms struct {
 type ProtocolWasmsModelInterface interface {
 	BatchInsert(ctx context.Context, dbTx pgx.Tx, wasms []ProtocolWasms) error
 	GetUnclassified(ctx context.Context) ([]ProtocolWasms, error)
-	GetClassifiedByHashes(ctx context.Context, dbTx pgx.Tx, hashes []types.HashBytea) (map[types.HashBytea]string, error)
+	GetClassifiedByHashes(ctx context.Context, q db.Querier, hashes []types.HashBytea) (map[types.HashBytea]string, error)
 	BatchUpdateProtocolID(ctx context.Context, dbTx pgx.Tx, wasmHashes []types.HashBytea, protocolID string) error
 }
 
@@ -92,10 +92,12 @@ func (m *ProtocolWasmsModel) GetUnclassified(ctx context.Context) ([]ProtocolWas
 }
 
 // GetClassifiedByHashes returns wasm_hash → protocol_id for the supplied hashes
-// where protocol_id IS NOT NULL. The read runs inside the supplied dbTx so
-// callers on the live ingest path see classifications consistent with the
-// in-flight transaction.
-func (m *ProtocolWasmsModel) GetClassifiedByHashes(ctx context.Context, dbTx pgx.Tx, hashes []types.HashBytea) (map[types.HashBytea]string, error) {
+// where protocol_id IS NOT NULL. q accepts either the connection pool or a
+// pgx.Tx: this read is used both non-transactionally (classification
+// prefetch, before any ledger transaction opens — staleness here is harmless
+// since this-ledger uploads resolve from the buffer and prior rows are
+// immutable) and, where a caller still holds an open tx, against that tx.
+func (m *ProtocolWasmsModel) GetClassifiedByHashes(ctx context.Context, q db.Querier, hashes []types.HashBytea) (map[types.HashBytea]string, error) {
 	if len(hashes) == 0 {
 		return nil, nil
 	}
@@ -112,7 +114,7 @@ func (m *ProtocolWasmsModel) GetClassifiedByHashes(ctx context.Context, dbTx pgx
 	const query = `SELECT wasm_hash, protocol_id, created_at FROM protocol_wasms WHERE wasm_hash = ANY($1::bytea[]) AND protocol_id IS NOT NULL`
 
 	start := time.Now()
-	wasms, err := db.QueryMany[ProtocolWasms](ctx, dbTx, query, hashBytes)
+	wasms, err := db.QueryMany[ProtocolWasms](ctx, q, query, hashBytes)
 	duration := time.Since(start).Seconds()
 	m.Metrics.QueryDuration.WithLabelValues("GetClassifiedByHashes", "protocol_wasms").Observe(duration)
 	m.Metrics.QueriesTotal.WithLabelValues("GetClassifiedByHashes", "protocol_wasms").Inc()

internal/indexer/indexer_buffer.go

@@ -197,7 +197,9 @@ func (b *IndexerBuffer) GetTransactions() []*types.Transaction {
 	return txs
 }
 
-// GetTransactionsParticipants returns a map of transaction ToIDs to its participants.
+// GetTransactionsParticipants returns a map of transaction ToIDs to its
+// participants. The map itself is a shallow clone, but each set.Set[string]
+// value is the buffer's live object; callers must not modify the sets.
 func (b *IndexerBuffer) GetTransactionsParticipants() map[int64]set.Set[string] {
 	b.mu.RLock()
 	defer b.mu.RUnlock()
@@ -326,8 +328,8 @@ func (b *IndexerBuffer) PushTrustlineChange(trustlineChange types.TrustlineChang
 	pushWithTombstone(b.trustlineChangesByTrustlineKey, b.trustlineTombstones, changeKey, trustlineChange, trustlineOrder, trustlineIsNoopRemove)
 }
 
-// GetTrustlineChanges returns all trustline changes stored in the buffer.
-// Thread-safe: uses read lock.
+// GetTrustlineChanges returns the buffer's internal map of trustline changes;
+// callers must not modify it. Thread-safe: uses read lock.
 func (b *IndexerBuffer) GetTrustlineChanges() map[TrustlineChangeKey]types.TrustlineChange {
 	b.mu.RLock()
 	defer b.mu.RUnlock()
@@ -346,8 +348,8 @@ func (b *IndexerBuffer) PushAccountChange(accountChange types.AccountChange) {
 	pushWithTombstone(b.accountChangesByAccountID, b.accountTombstones, accountChange.AccountID, accountChange, accountOrder, accountIsNoopRemove)
 }
 
-// GetAccountChanges returns all account changes stored in the buffer.
-// Thread-safe: uses read lock.
+// GetAccountChanges returns the buffer's internal map of account changes;
+// callers must not modify it. Thread-safe: uses read lock.
 func (b *IndexerBuffer) GetAccountChanges() map[string]types.AccountChange {
 	b.mu.RLock()
 	defer b.mu.RUnlock()
@@ -370,8 +372,8 @@ func (b *IndexerBuffer) PushSACBalanceChange(sacBalanceChange types.SACBalanceCh
 	pushWithTombstone(b.sacBalanceChangesByKey, b.sacTombstones, key, sacBalanceChange, sacBalanceOrder, sacBalanceIsNoopRemove)
 }
 
-// GetSACBalanceChanges returns all SAC balance changes stored in the buffer.
-// Thread-safe: uses read lock.
+// GetSACBalanceChanges returns the buffer's internal map of SAC balance
+// changes; callers must not modify it. Thread-safe: uses read lock.
 func (b *IndexerBuffer) GetSACBalanceChanges() map[SACBalanceChangeKey]types.SACBalanceChange {
 	b.mu.RLock()
 	defer b.mu.RUnlock()
@@ -394,8 +396,9 @@ func (b *IndexerBuffer) PushLiquidityPoolShareChange(change types.LiquidityPoolS
 	pushWithTombstone(b.lpShareChangesByKey, b.lpShareTombstones, key, change, lpShareOrder, lpShareIsNoopRemove)
 }
 
-// GetLiquidityPoolShareChanges returns all pool-share balance changes stored in the buffer.
-// Thread-safe: uses read lock.
+// GetLiquidityPoolShareChanges returns the buffer's internal map of
+// pool-share balance changes; callers must not modify it. Thread-safe: uses
+// read lock.
 func (b *IndexerBuffer) GetLiquidityPoolShareChanges() map[LiquidityPoolShareChangeKey]types.LiquidityPoolShareChange {
 	b.mu.RLock()
 	defer b.mu.RUnlock()
@@ -414,8 +417,8 @@ func (b *IndexerBuffer) PushLiquidityPoolChange(change types.LiquidityPoolChange
 	pushWithTombstone(b.lpChangesByPoolID, b.lpTombstones, change.PoolID, change, lpOrder, lpIsNoopRemove)
 }
 
-// GetLiquidityPoolChanges returns all pool reserve changes stored in the buffer.
-// Thread-safe: uses read lock.
+// GetLiquidityPoolChanges returns the buffer's internal map of pool reserve
+// changes; callers must not modify it. Thread-safe: uses read lock.
 func (b *IndexerBuffer) GetLiquidityPoolChanges() map[string]types.LiquidityPoolChange {
 	b.mu.RLock()
 	defer b.mu.RUnlock()
@@ -447,7 +450,9 @@ func (b *IndexerBuffer) GetOperations() []*types.Operation {
 	return ops
 }
 
-// GetOperationsParticipants returns a map of operation IDs to its participants.
+// GetOperationsParticipants returns a map of operation IDs to its
+// participants. The map itself is a shallow clone, but each set.Set[string]
+// value is the buffer's live object; callers must not modify the sets.
 func (b *IndexerBuffer) GetOperationsParticipants() map[int64]set.Set[string] {
 	b.mu.RLock()
 	defer b.mu.RUnlock()
@@ -485,8 +490,8 @@ func (b *IndexerBuffer) PushStateChange(transaction types.Transaction, operation
 	}
 }
 
-// GetStateChanges returns a copy of all state changes stored in the buffer.
-// Thread-safe: uses read lock.
+// GetStateChanges returns the buffer's internal slice of state changes;
+// callers must not modify it. Thread-safe: uses read lock.
 func (b *IndexerBuffer) GetStateChanges() []types.StateChange {
 	b.mu.RLock()
 	defer b.mu.RUnlock()

internal/ingest/datastore_backend.go

@@ -43,6 +43,15 @@ const (
 	defaultRetryWait         = 5 * time.Second
 )
 
+// ErrBufferDead marks a GetLedger failure as terminal: the ledgerBuffer's
+// internal context was cancelled by the buffer itself (a worker exhausted
+// its download retry budget, or hit a hard NotExist on a bounded range),
+// independent of the caller's context lifecycle. The buffer never recovers
+// from this — every subsequent GetLedger call on this backend instance
+// returns the same error — so callers should treat it as permanent rather
+// than retrying against a dead buffer.
+var ErrBufferDead = errors.New("ledger buffer permanently cancelled")
+
 var _ ledgerbackend.LedgerBackend = (*datastoreBackend)(nil)
 
 // datastoreBackend implements ledgerbackend.LedgerBackend with optimizations
@@ -417,10 +426,21 @@ func (buf *ledgerBuffer) downloadAndDecode(sequence uint32) (xdr.LedgerCloseMeta
 
 // getNextBatch receives the next decoded batch from the buffer in sequence order.
 // After receiving, it enqueues a new download task to maintain the buffer invariant.
+//
+// buf.ctx is derived from the caller-supplied ctx via context.WithCancelCause, so
+// buf.ctx.Done() fires both when that parent ctx is cancelled (shutdown — its cause is
+// context.Canceled, matching close()'s own buf.cancel(context.Canceled)) and when
+// downloadAndStore gives up on a sequence and calls buf.cancel with a real error. Only the
+// latter means the buffer itself has permanently died; distinguishing on the cause lets
+// callers (the ledger-fetch retry ladder) tell "will never recover" apart from "shutting down".
 func (buf *ledgerBuffer) getNextBatch(ctx context.Context) (xdr.LedgerCloseMetaBatch, error) {
 	select {
 	case <-buf.ctx.Done():
-		return xdr.LedgerCloseMetaBatch{}, fmt.Errorf("buffer context cancelled: %w", context.Cause(buf.ctx))
+		cause := context.Cause(buf.ctx)
+		if errors.Is(cause, context.Canceled) {
+			return xdr.LedgerCloseMetaBatch{}, fmt.Errorf("buffer context cancelled: %w", cause)
+		}
+		return xdr.LedgerCloseMetaBatch{}, fmt.Errorf("%w: %w", ErrBufferDead, cause)
 	case <-ctx.Done():
 		return xdr.LedgerCloseMetaBatch{}, fmt.Errorf("caller context cancelled: %w", ctx.Err())
 	case batch := <-buf.batchQueue: