services/horizon: Reap history object tables when ingestion is idle

services/horizon/internal/db2/history/main.go

@@ -261,6 +261,7 @@ type IngestionQ interface {
  NewTradeBatchInsertBuilder(maxBatchSize int) TradeBatchInsertBuilder
  RebuildTradeAggregationTimes(ctx context.Context, from, to strtime.Millis, roundingSlippageFilter int) error
  RebuildTradeAggregationBuckets(ctx context.Context, fromLedger, toLedger uint32, roundingSlippageFilter int) error
+ ReapLookupTables(ctx context.Context, offsets map[string]int64) (map[string]int64, error)
  CreateAssets(ctx context.Context, assets []xdr.Asset, batchSize int) (map[string]Asset, error)
  QTransactions
  QTrustLines
@@ -825,6 +826,145 @@ func (q *Q) CloneIngestionQ() IngestionQ {
  return &Q{q.Clone()}
 }
 
+type tableObjectFieldPair struct {
+ // name is a table name of history table
+ name string
+ // objectField is a name of object field in history table which uses
+ // the lookup table.
+ objectField string
+}
+
+// ReapLookupTables removes rows from lookup tables like history_claimable_balances
+// which aren't used (orphaned), i.e. history entries for them were reaped.
+// This method must be executed inside ingestion transaction. Otherwise it may
+// create invalid state in lookup and history tables.
+func (q Q) ReapLookupTables(ctx context.Context, offsets map[string]int64) (map[string]int64, error) {
+ if q.GetTx() == nil {
+ return nil, errors.New("cannot be called outside of an ingestion transaction")
+ }
+
+ const batchSize = 10000
+
+ if offsets == nil {
+ offsets = make(map[string]int64)
+ }
+
+ for table, historyTables := range map[string][]tableObjectFieldPair{
+ "history_claimable_balances": {
+ {
+ name: "history_operation_claimable_balances",
+ objectField: "history_claimable_balance_id",
+ },
+ {
+ name: "history_transaction_claimable_balances",
+ objectField: "history_claimable_balance_id",
+ },
+ },
+ "history_liquidity_pools": {
+ {
+ name: "history_operation_liquidity_pools",
+ objectField: "history_liquidity_pool_id",
+ },
+ {
+ name: "history_transaction_liquidity_pools",
+ objectField: "history_liquidity_pool_id",
+ },
+ },
+ } {
+ query, err := constructReapLookupTablesQuery(table, historyTables, batchSize, offsets[table])
+ if err != nil {
+ return nil, errors.Wrap(err, "error constructing a query")
+ }
+
+ _, err = q.ExecRaw(
+ context.WithValue(ctx, &db.QueryTypeContextKey, db.DeleteQueryType),
+ query,
+ )
+ if err != nil {
+ return nil, errors.Wrapf(err, "error running query: %s", query)
+ }
+
+ offsets[table] += batchSize
+
+ // Check if offset exceeds table size and then reset it
+ var count int64
+ err = q.GetRaw(ctx, &count, fmt.Sprintf("SELECT COUNT(*) FROM %s", table))
+ if err != nil {
+ return nil, err
+ }
+
+ if offsets[table] > count {
+ offsets[table] = 0
+ }
+ }
+ return offsets, nil
+}
+
+// constructReapLookupTablesQuery creates a query like (using history_claimable_balances
+// as an example):
+//
+// delete from history_claimable_balances where id in
+// (select id from
+// (select id,
+// (select count(*) from history_operation_claimable_balances
+// where history_claimable_balance_id = hcb.id) as c1,
+// (select count(*) from history_transaction_claimable_balances
+// where history_claimable_balance_id = hcb.id) as c2,
+// 1 as cx,
+// from history_claimable_balances hcb order by id limit 100 offset 1000)
+// as sub where c1 = 0 and c2 = 0 and 1=1);
+//
+// In short it checks the 100 rows omiting 1000 row of history_claimable_balances
+// and counts occurences 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.
+//
+// The offset param should be increased before each execution. Given that
+// some rows will be removed and some will be added by ingestion it's
+// possible that rows will be skipped from deletion. But offset is reset
+// when it reaches the table size so eventually all orphaned rows are
+// deleted.
+func constructReapLookupTablesQuery(table string, historyTables []tableObjectFieldPair, batchSize, offset int64) (string, error) {
+ var sb strings.Builder
+ var err error
+ _, err = fmt.Fprintf(&sb, "delete from %s where id IN (select id from (select id, ", table)
+ if err != nil {
+ return "", err
+ }
+
+ for i, historyTable := range historyTables {
+ _, err = fmt.Fprintf(
+ &sb,
+ `(select count(*) from %s where %s = hcb.id) as c%d, `,
+ historyTable.name,
+ historyTable.objectField,
+ i,
+ )
+ if err != nil {
+ return "", err
+ }
+ }
+
+ _, err = fmt.Fprintf(&sb, "1 as cx from %s hcb order by id limit %d offset %d) as sub where ", table, batchSize, offset)
+ if err != nil {
+ return "", err
+ }
+
+ for i := range historyTables {
+ _, err = fmt.Fprintf(&sb, "c%d = 0 and ", i)
+ if err != nil {
+ return "", err
+ }
+ }
+
+ _, err = sb.WriteString("1=1);")
+ if err != nil {
+ return "", err
+ }
+
+ return sb.String(), nil
+}
+
 // DeleteRangeAll deletes a range of rows from all history tables between
 // `start` and `end` (exclusive).
 func (q *Q) DeleteRangeAll(ctx context.Context, start, end int64) error {

services/horizon/internal/db2/history/main_test.go

@@ -5,6 +5,8 @@ import (
  "time"
 
  "github.com/stellar/go/services/horizon/internal/test"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
 )
 
 func TestLatestLedger(t *testing.T) {
@@ -66,3 +68,45 @@ func TestElderLedger(t *testing.T) {
  tt.Assert.Equal(1, seq)
  }
 }
+
+func TestConstructReapLookupTablesQuery(t *testing.T) {
+ query, err := constructReapLookupTablesQuery(
+ "history_accounts",
+ []tableObjectFieldPair{
+ {
+ name: "history_effects",
+ objectField: "history_account_id",
+ },
+ {
+ name: "history_operation_participants",
+ objectField: "history_account_id",
+ },
+ {
+ name: "history_trades",
+ objectField: "base_account_id",
+ },
+ {
+ name: "history_trades",
+ objectField: "counter_account_id",
+ },
+ {
+ name: "history_transaction_participants",
+ objectField: "history_account_id",
+ },
+ },
+ 10,
+ 0,
+ )
+
+ require.NoError(t, err)
+ assert.Equal(t,
+ "delete from history_accounts where id IN "+
+ "(select id from "+
+ "(select id, (select count(*) from history_effects where history_account_id = hcb.id) as c0, "+
+ "(select count(*) from history_operation_participants where history_account_id = hcb.id) as c1, "+
+ "(select count(*) from history_trades where base_account_id = hcb.id) as c2, "+
+ "(select count(*) from history_trades where counter_account_id = hcb.id) as c3, "+
+ "(select count(*) from history_transaction_participants where history_account_id = hcb.id) as c4, "+
+ "1 as cx from history_accounts hcb order by id limit 10 offset 0) as sub "+
+ "where c0 = 0 and c1 = 0 and c2 = 0 and c3 = 0 and c4 = 0 and 1=1);", query)
+}

services/horizon/internal/db2/history/reap_test.go

@@ -0,0 +1,74 @@
+package history_test
+
+import (
+ "testing"
+
+ "github.com/stellar/go/services/horizon/internal/db2/history"
+ "github.com/stellar/go/services/horizon/internal/ledger"
+ "github.com/stellar/go/services/horizon/internal/reap"
+ "github.com/stellar/go/services/horizon/internal/test"
+)
+
+func TestReapLookupTables(t *testing.T) {
+ tt := test.Start(t)
+ defer tt.Finish()
+ ledgerState := &ledger.State{}
+ ledgerState.SetStatus(tt.Scenario("kahuna"))
+
+ db := tt.HorizonSession()
+
+ sys := reap.New(0, db, ledgerState)
+
+ var (
+ prevLedgers, curLedgers int
+ prevClaimableBalances, curClaimableBalances int
+ prevLiquidityPools, curLiquidityPools int
+ )
+
+ // Prev
+ {
+ err := db.GetRaw(tt.Ctx, &prevLedgers, `SELECT COUNT(*) FROM history_ledgers`)
+ tt.Require.NoError(err)
+ err = db.GetRaw(tt.Ctx, &prevClaimableBalances, `SELECT COUNT(*) FROM history_claimable_balances`)
+ tt.Require.NoError(err)
+ err = db.GetRaw(tt.Ctx, &prevLiquidityPools, `SELECT COUNT(*) FROM history_liquidity_pools`)
+ tt.Require.NoError(err)
+ }
+
+ ledgerState.SetStatus(tt.LoadLedgerStatus())
+ sys.RetentionCount = 1
+ err := sys.DeleteUnretainedHistory(tt.Ctx)
+ tt.Require.NoError(err)
+
+ q := &history.Q{tt.HorizonSession()}
+
+ err = q.Begin()
+ tt.Require.NoError(err)
+
+ newOffsets, err := q.ReapLookupTables(tt.Ctx, nil)
+ tt.Require.NoError(err)
+
+ err = q.Commit()
+ tt.Require.NoError(err)
+
+ // cur
+ {
+ err := db.GetRaw(tt.Ctx, &curLedgers, `SELECT COUNT(*) FROM history_ledgers`)
+ tt.Require.NoError(err)
+ err = db.GetRaw(tt.Ctx, &curClaimableBalances, `SELECT COUNT(*) FROM history_claimable_balances`)
+ tt.Require.NoError(err)
+ err = db.GetRaw(tt.Ctx, &curLiquidityPools, `SELECT COUNT(*) FROM history_liquidity_pools`)
+ tt.Require.NoError(err)
+ }
+
+ tt.Assert.Equal(61, prevLedgers, "prevLedgers")
+ tt.Assert.Equal(1, curLedgers, "curLedgers")
+ tt.Assert.Equal(1, prevClaimableBalances, "prevClaimableBalances")
+ tt.Assert.Equal(0, curClaimableBalances, "curClaimableBalances")
+ tt.Assert.Equal(1, prevLiquidityPools, "prevLiquidityPools")
+ tt.Assert.Equal(0, curLiquidityPools, "curLiquidityPools")
+
+ tt.Assert.Len(newOffsets, 2)
+ tt.Assert.Equal(int64(0), newOffsets["history_claimable_balances"])
+ tt.Assert.Equal(int64(0), newOffsets["history_liquidity_pools"])
+}

services/horizon/internal/ingest/fsm.go

@@ -523,6 +523,7 @@ func (r resumeState) run(s *system) (transition, error) {
  localLog.Info("Processed ledger")
 
  s.maybeVerifyState(ingestLedger)
+ s.maybeReapLookupTables(ingestLedger)
 
  return resumeImmediately(ingestLedger), nil
 }

services/horizon/internal/ingest/main.go

@@ -127,6 +127,10 @@ type Metrics struct {
  // duration of rebuilding trade aggregation buckets.
  LedgerIngestionTradeAggregationDuration prometheus.Summary
 
+ // LedgerIngestionReapLookupTablesDuration exposes timing metrics about the rate and
+ // duration of reaping lookup tables.
+ LedgerIngestionReapLookupTablesDuration prometheus.Summary
+
  // StateVerifyDuration exposes timing metrics about the rate and
  // duration of state verification.
  StateVerifyDuration prometheus.Summary
@@ -201,6 +205,8 @@ type system struct {
  disableStateVerification bool
 
  checkpointManager historyarchive.CheckpointManager
+
+ reapOffsets map[string]int64
 }
 
 func NewSystem(config Config) (System, error) {
@@ -313,6 +319,11 @@ func (s *system) initMetrics() {
  Help: "ledger ingestion trade aggregation rebuild durations, sliding window = 10m",
  })
 
+ s.metrics.LedgerIngestionReapLookupTablesDuration = prometheus.NewSummary(prometheus.SummaryOpts{
+ Namespace: "horizon", Subsystem: "ingest", Name: "ledger_ingestion_reap_lookup_tables_duration_seconds",
+ Help: "ledger ingestion reap lookup tables durations, sliding window = 10m",
+ })
+
  s.metrics.StateVerifyDuration = prometheus.NewSummary(prometheus.SummaryOpts{
  Namespace: "horizon", Subsystem: "ingest", Name: "state_verify_duration_seconds",
  Help: "state verification durations, sliding window = 10m",
@@ -445,6 +456,7 @@ func (s *system) RegisterMetrics(registry *prometheus.Registry) {
  registry.MustRegister(s.metrics.LocalLatestLedger)
  registry.MustRegister(s.metrics.LedgerIngestionDuration)
  registry.MustRegister(s.metrics.LedgerIngestionTradeAggregationDuration)
+ registry.MustRegister(s.metrics.LedgerIngestionReapLookupTablesDuration)
  registry.MustRegister(s.metrics.StateVerifyDuration)
  registry.MustRegister(s.metrics.StateInvalidGauge)
  registry.MustRegister(s.metrics.LedgerStatsCounter)
@@ -663,6 +675,56 @@ func (s *system) maybeVerifyState(lastIngestedLedger uint32) {
  }
 }
 
+func (s *system) maybeReapLookupTables(lastIngestedLedger uint32) {
+ // Check if lastIngestedLedger is the last one available in the backend
+ sequence, err := s.ledgerBackend.GetLatestLedgerSequence(s.ctx)
+ if err != nil {
+ log.WithField("err", err).Error("Error getting latest ledger sequence from backend")
+ return
+ }
+
+ if sequence != lastIngestedLedger {
+ // Catching up - skip reaping tables in this cycle.
+ return
+ }
+
+ err = s.historyQ.Begin()
+ if err != nil {
+ log.WithField("err", err).Error("Error starting a transaction")
+ return
+ }
+ defer s.historyQ.Rollback()
+
+ // If so block ingestion in the cluster to reap tables
+ _, err = s.historyQ.GetLastLedgerIngest(s.ctx)
+ if err != nil {
+ log.WithField("err", err).Error(getLastIngestedErrMsg)
+ return
+ }
+
+ // Make sure reaping will not take more than 5s, which is average ledger
+ // closing time.
+ ctx, cancel := context.WithTimeout(s.ctx, 5*time.Second)
+ defer cancel()
+
+ reapStart := time.Now()
+ newOffsets, err := s.historyQ.ReapLookupTables(ctx, s.reapOffsets)
+ if err != nil {
+ log.WithField("err", err).Warn("Error reaping lookup tables")
+ return
+ }
+
+ err = s.historyQ.Commit()
+ if err != nil {
+ log.WithField("err", err).Error("Error commiting a transaction")
+ return
+ }
+
+ s.reapOffsets = newOffsets
+ reapDuration := time.Since(reapStart).Seconds()
+ s.Metrics().LedgerIngestionReapLookupTablesDuration.Observe(float64(reapDuration))
+}
+
 func (s *system) incrementStateVerificationErrors() int {
  s.stateVerificationMutex.Lock()
  defer s.stateVerificationMutex.Unlock()