Fix lookup-table reaper deleting rows referenced by concurrent ingestion

CHANGELOG.md

@@ -3,6 +3,11 @@
 All notable changes to this project will be documented in this
 file. This project adheres to [Semantic Versioning](http://semver.org/).
 
+## Unreleased
+
+### Fixed
+- Fixed the history lookup-table reaper deleting rows (e.g. `history_accounts`) that live ingestion was concurrently inserting references to, which left dangling references and could make an account's history endpoints return 404 or silently omit records until a reingest ([#222](https://github.com/stellar/stellar-horizon/pull/222)).
+
 ## 28.0.0
 
 **This release adds support for Protocol 28.**

internal/db2/history/main.go

@@ -1106,6 +1106,34 @@
 }
 
 func (q *Q) deleteLookupTableRows(ctx context.Context, table string, ids []int64) (int64, error) {
+	// The two statements below must share a transaction: the FOR UPDATE lock taken
+	// by the first is only held until the transaction ends, and the DELETE relies
+	// on it still being held. Enforce it rather than fail silently — outside a
+	// transaction the lock would be released immediately and the orphan check could
+	// race concurrent ingestion. ReapLookupTable opens the transaction.
+	if q.GetTx() == nil {
+		return 0, errors.New("deleteLookupTableRows must be called within a transaction")
+	}
+
+	// Lock the candidate rows in a statement that is separate from the DELETE
+	// below. Live ingestion takes a FOR KEY SHARE lock on these rows before it
+	// inserts references to them, so this SELECT ... FOR UPDATE blocks until any
+	// such in-flight ingestion transaction commits. Crucially, because the DELETE
+	// is issued as a separate statement, it runs under a fresh READ COMMITTED
+	// snapshot taken *after* the lock wait resolves, so its NOT EXISTS checks
+	// observe the references committed by the ingestion transaction we blocked on.
+	//
+	// Combining the lock and the check into a single WITH ... FOR UPDATE ... DELETE
+	// statement (as this code previously did) is unsafe: that statement's snapshot
+	// is taken before the lock wait, so blocking on FOR UPDATE does not refresh the
+	// snapshot used by the NOT EXISTS sub-queries. The row would then be judged
+	// orphaned against a stale view and deleted while it is in fact referenced.
+	lockQuery := constructLockLookupTableRowsQuery(table, ids)
+	var lockedIDs []int64
+	if err := q.SelectRaw(ctx, &lockedIDs, lockQuery); err != nil {
+		return 0, fmt.Errorf("error running query %s : %w", lockQuery, err)
+	}
+
 	deleteQuery := constructDeleteLookupTableRowsQuery(table, ids)
 	result, err := q.ExecRaw(
 		context.WithValue(ctx, &db.QueryTypeContextKey, db.DeleteQueryType),
@@ -1122,83 +1150,82 @@
 	return deletedCount, nil
 }
 
-// constructDeleteLookupTableRowsQuery creates a query like (using history_claimable_balances
-// as an example):
-//
-//	WITH ha_batch AS (
-//		SELECT   id
-//		FROM     history_claimable_balances
-//		WHERE IN ($1, $2, ...) ORDER BY id asc FOR UPDATE
-//	) DELETE FROM history_claimable_balances WHERE id IN (
-//		SELECT e1.id as id FROM ha_batch e1
-//		WHERE NOT EXISTS (SELECT 1 FROM history_transaction_claimable_balances WHERE history_transaction_claimable_balances.history_claimable_balance_id = id limit 1)
-//		AND NOT EXISTS (SELECT 1 FROM history_operation_claimable_balances WHERE history_operation_claimable_balances.history_claimable_balance_id = id limit 1)
-//	 )
-//
-// It checks each of the candidate rows provided in the top level IN clause
-// and counts occurrences of each row in corresponding history tables.
-// If there are no history rows for a given id, the row in
-// history_claimable_balances is removed.
-//
-// Note that the rows are locked using via SELECT FOR UPDATE. The reason
-// for that is to maintain safety when ingestion is running concurrently.
-// The ingestion loaders will also lock rows from the history lookup tables
-// via SELECT FOR KEY SHARE. This will ensure that the reaping transaction
-// will block until the ingestion transaction commits (or vice-versa).
-func constructDeleteLookupTableRowsQuery(table string, ids []int64) string {
+// joinInt64s renders ids as a comma-separated list for inlining into an IN clause.
+func joinInt64s(ids []int64) string {
+	stringIds := make([]string, len(ids))
+	for i, id := range ids {
+		stringIds[i] = strconv.FormatInt(id, 10)
+	}
+	return strings.Join(stringIds, ", ")
+}
+
+// constructReapOrphanConditions returns the AND-joined NOT EXISTS clauses that
+// hold only when a lookup table row is referenced by none of the history tables
+// that can reference it. idColumn is the SQL expression that names the candidate
+// row's id in the surrounding query. It is shared by the find and delete reap
+// queries so both agree on what "orphaned" means.
+func constructReapOrphanConditions(table, idColumn string) string {
 	var conditions []string
 	for _, referencedTable := range historyLookupTables[table] {
 		conditions = append(
 			conditions,
 			fmt.Sprintf(
-				"NOT EXISTS ( SELECT 1 as row FROM %s WHERE %s.%s = id LIMIT 1)",
+				"NOT EXISTS ( SELECT 1 as row FROM %s WHERE %s.%s = %s LIMIT 1)",
 				referencedTable.name,
 				referencedTable.name, referencedTable.objectField,
+				idColumn,
 			),
 		)
 	}
+	return strings.Join(conditions, " AND ")
+}
 
-	stringIds := make([]string, len(ids))
-	for i, id := range ids {
-		stringIds[i] = strconv.FormatInt(id, 10)
-	}
-	innerQuery := fmt.Sprintf(
-		"SELECT id FROM %s WHERE id IN (%s) ORDER BY id asc FOR UPDATE",
+// constructLockLookupTableRowsQuery creates a query which locks the candidate
+// lookup table rows via SELECT ... FOR UPDATE, for example:
+//
+//	SELECT id FROM history_claimable_balances WHERE id IN ($1, $2, ...) ORDER BY id ASC FOR UPDATE
+//
+// It must run as its own statement, before the DELETE built by
+// constructDeleteLookupTableRowsQuery; see deleteLookupTableRows for why. The
+// rows are ordered by id to match the lock ordering used by the ingestion loaders
+// (SELECT ... FOR KEY SHARE) and avoid deadlocks.
+func constructLockLookupTableRowsQuery(table string, ids []int64) string {
+	return fmt.Sprintf(
+		"SELECT id FROM %s WHERE id IN (%s) ORDER BY id ASC FOR UPDATE",
 		table,
-		strings.Join(stringIds, ", "),
+		joinInt64s(ids),
 	)
+}
 
-	deleteQuery := fmt.Sprintf(
-		"WITH ha_batch AS (%s) DELETE FROM %s WHERE id IN ("+
-			"SELECT e1.id as id FROM ha_batch e1 WHERE %s)",
-		innerQuery,
+// constructDeleteLookupTableRowsQuery creates a query like (using history_claimable_balances
+// as an example):
+//
+//	DELETE FROM history_claimable_balances WHERE id IN ($1, $2, ...)
+//		AND NOT EXISTS (SELECT 1 FROM history_transaction_claimable_balances WHERE history_transaction_claimable_balances.history_claimable_balance_id = history_claimable_balances.id LIMIT 1)
+//		AND NOT EXISTS (SELECT 1 FROM history_operation_claimable_balances WHERE history_operation_claimable_balances.history_claimable_balance_id = history_claimable_balances.id LIMIT 1)
+//
+// It removes each candidate row from the top level IN clause that no history
+// table references. The candidates must already be locked by
+// constructLockLookupTableRowsQuery in the same transaction; see
+// deleteLookupTableRows.
+func constructDeleteLookupTableRowsQuery(table string, ids []int64) string {
+	return fmt.Sprintf(
+		"DELETE FROM %s WHERE id IN (%s) AND %s",
 		table,
-		strings.Join(conditions, " AND "),
+		joinInt64s(ids),
+		constructReapOrphanConditions(table, table+".id"),
 	)
-	return deleteQuery
 }
 
 func constructFindReapLookupTablesQuery(table string, batchSize int, offset int64) string {
-	var conditions []string
-
-	for _, referencedTable := range historyLookupTables[table] {
-		conditions = append(
-			conditions,
-			fmt.Sprintf(
-				"NOT EXISTS ( SELECT 1 as row FROM %s WHERE %s.%s = id LIMIT 1)",
-				referencedTable.name,
-				referencedTable.name, referencedTable.objectField,
-			),
-		)
-	}
-
 	return fmt.Sprintf(
 		"WITH ha_batch AS (SELECT id FROM %s WHERE id >= %d ORDER BY id ASC limit %d) "+
-			"SELECT e1.id as id FROM ha_batch e1 WHERE ",
+			"SELECT e1.id as id FROM ha_batch e1 WHERE %s",
 		table,
 		offset,
 		batchSize,
-	) + strings.Join(conditions, " AND ")
+		constructReapOrphanConditions(table, "id"),
+	)
 }
 
 // DeleteRangeAll deletes a range of rows from all history tables between

internal/db2/history/main_test.go

@@ -69,20 +69,29 @@
 	}
 }
 
+func TestConstructLockLookupTableRowsQuery(t *testing.T) {
+	query := constructLockLookupTableRowsQuery(
+		"history_accounts",
+		[]int64{100, 20, 30},
+	)
+
+	assert.Equal(t,
+		"SELECT id FROM history_accounts WHERE id IN (100, 20, 30) ORDER BY id ASC FOR UPDATE", query)
+}
+
 func TestConstructDeleteLookupTableRowsQuery(t *testing.T) {
 	query := constructDeleteLookupTableRowsQuery(
 		"history_accounts",
 		[]int64{100, 20, 30},
 	)
 
 	assert.Equal(t,
-		"WITH ha_batch AS (SELECT id FROM history_accounts WHERE id IN (100, 20, 30) ORDER BY id asc FOR UPDATE) "+
-			"DELETE FROM history_accounts WHERE id IN (SELECT e1.id as id FROM ha_batch e1 "+
-			"WHERE NOT EXISTS ( SELECT 1 as row FROM history_transaction_participants WHERE history_transaction_participants.history_account_id = id LIMIT 1) "+
-			"AND NOT EXISTS ( SELECT 1 as row FROM history_effects WHERE history_effects.history_account_id = id LIMIT 1) "+
-			"AND NOT EXISTS ( SELECT 1 as row FROM history_operation_participants WHERE history_operation_participants.history_account_id = id LIMIT 1) "+
-			"AND NOT EXISTS ( SELECT 1 as row FROM history_trades WHERE history_trades.base_account_id = id LIMIT 1) "+
-			"AND NOT EXISTS ( SELECT 1 as row FROM history_trades WHERE history_trades.counter_account_id = id LIMIT 1))", query)
+		"DELETE FROM history_accounts WHERE id IN (100, 20, 30) "+
+			"AND NOT EXISTS ( SELECT 1 as row FROM history_transaction_participants WHERE history_transaction_participants.history_account_id = history_accounts.id LIMIT 1) "+
+			"AND NOT EXISTS ( SELECT 1 as row FROM history_effects WHERE history_effects.history_account_id = history_accounts.id LIMIT 1) "+
+			"AND NOT EXISTS ( SELECT 1 as row FROM history_operation_participants WHERE history_operation_participants.history_account_id = history_accounts.id LIMIT 1) "+
+			"AND NOT EXISTS ( SELECT 1 as row FROM history_trades WHERE history_trades.base_account_id = history_accounts.id LIMIT 1) "+
+			"AND NOT EXISTS ( SELECT 1 as row FROM history_trades WHERE history_trades.counter_account_id = history_accounts.id LIMIT 1)", query)
 }
 
 func TestConstructReapLookupTablesQuery(t *testing.T) {

internal/db2/history/reap_concurrency_test.go

@@ -0,0 +1,88 @@
+package history
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/stellar/go-stellar-sdk/keypair"
+	"github.com/stellar/stellar-horizon/internal/test"
+)
+
+// TestReapDoesNotDeleteConcurrentlyReferencedRows is a regression test for a bug
+// where the lookup table reaper deleted a history_accounts row that live
+// ingestion was concurrently inserting a reference to.
+//
+// The reaper selects a candidate row while it is still orphaned, then, in a
+// separate transaction, re-checks that it is orphaned before deleting it. Live
+// ingestion locks the row FOR KEY SHARE and inserts a referencing child row in a
+// single transaction. The reaper's DELETE blocks on that lock, but under READ
+// COMMITTED blocking alone does not refresh the snapshot used by the orphan
+// check: if the lock and the NOT EXISTS check share one statement, the check
+// runs against the pre-commit snapshot, misses the new reference, and deletes the
+// row while it is in fact referenced. This test drives exactly that interleaving
+// and asserts the reaper leaves the row in place.
+func TestReapDoesNotDeleteConcurrentlyReferencedRows(t *testing.T) {
+	tt := test.Start(t)
+	defer tt.Finish()
+	test.ResetHorizonDB(t, tt.HorizonDB)
+
+	ctx := context.Background()
+	s1 := tt.HorizonSession()
+	s2 := s1.Clone()
+
+	// Create a bare account row (no references), which makes it a reap candidate.
+	address := keypair.MustRandom().Address()
+	setup := NewAccountLoader(ConcurrentDeletes)
+	setup.GetFuture(address)
+	assert.NoError(t, setup.Exec(ctx, s1))
+	accountID, err := setup.GetNow(address)
+	assert.NoError(t, err)
+
+	// Ingestion transaction: lock the account row FOR KEY SHARE, exactly as the
+	// loader does during live ingestion, then insert an operation participant
+	// referencing it. Do not commit yet, so the reference is still uncommitted
+	// while the reaper runs.
+	assert.NoError(t, s1.Begin(ctx))
+
+	ingestLoader := NewAccountLoader(ConcurrentDeletes)
+	future := ingestLoader.GetFuture(address)
+	assert.NoError(t, ingestLoader.Exec(ctx, s1))
+
+	q1 := &Q{s1}
+	opParticipants := q1.NewOperationParticipantBatchInsertBuilder()
+	assert.NoError(t, opParticipants.Add(int64(1), future))
+	assert.NoError(t, opParticipants.Exec(ctx, s1))
+
+	// Reaper transaction, running concurrently with the ingestion transaction.
+	assert.NoError(t, s2.Begin(ctx))
+	q2 := &Q{s2}
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		// Give the reaper time to start and block on the FOR KEY SHARE lock held
+		// by the ingestion transaction before we commit that transaction.
+		<-time.After(time.Second * 3)
+		assert.NoError(t, s1.Commit())
+	}()
+
+	// reapLookupTable blocks on the ingestion transaction's row lock. Once that
+	// transaction commits, the DELETE runs under a fresh snapshot and must observe
+	// the operation participant reference, so nothing is deleted.
+	deletedCount, err := q2.reapLookupTable(ctx, "history_accounts", []int64{accountID}, 1000)
+	assert.NoError(t, err)
+	assert.NoError(t, s2.Commit())
+	wg.Wait()
+
+	assert.Equal(t, int64(0), deletedCount)
+
+	// The account row must still exist and keep its original id.
+	var account Account
+	assert.NoError(t, q2.AccountByAddress(ctx, &account, address))
+	assert.Equal(t, accountID, account.ID)
+}