diff --git a/CHANGELOG.md b/CHANGELOG.md index 5603499c..ffadf9ef 100644 --- a/CHANGELOG.md +++ b/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.** diff --git a/internal/db2/history/main.go b/internal/db2/history/main.go index b4dddf34..c9e78d2e 100644 --- a/internal/db2/history/main.go +++ b/internal/db2/history/main.go @@ -1106,6 +1106,34 @@ var historyLookupTables = map[string][]tableObjectFieldPair{ } 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 @@ func (q *Q) deleteLookupTableRows(ctx context.Context, table string, ids []int64 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 diff --git a/internal/db2/history/main_test.go b/internal/db2/history/main_test.go index d29267bc..3f3f2565 100644 --- a/internal/db2/history/main_test.go +++ b/internal/db2/history/main_test.go @@ -69,6 +69,16 @@ func TestElderLedger(t *testing.T) { } } +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", @@ -76,13 +86,12 @@ func TestConstructDeleteLookupTableRowsQuery(t *testing.T) { ) 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) { diff --git a/internal/db2/history/reap_concurrency_test.go b/internal/db2/history/reap_concurrency_test.go new file mode 100644 index 00000000..530aa753 --- /dev/null +++ b/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) +}