diff --git a/internal/data/ingest_store.go b/internal/data/ingest_store.go index 610a049cd..c2d700381 100644 --- a/internal/data/ingest_store.go +++ b/internal/data/ingest_store.go @@ -254,10 +254,19 @@ func (m *IngestStoreModel) GetLedgerGaps(ctx context.Context, startLedger, endLe return ledgerGaps, nil } +// GetOldestLedger returns the ledger_number of the oldest transaction still in the table, or 0 +// when it is empty. The hypertable's default partition-column index +// (transactions_ledger_created_at_idx) gives every uncompressed chunk an ordered path, so the +// planner serves this as an ordered ChunkAppend that opens the oldest chunk, takes one row and +// stops instead of reading the first row of every chunk. +// The sort key is ledger_created_at alone: stellar-core enforces strictly increasing ledger close +// times, so every row sharing a ledger_created_at belongs to one ledger and carries the same +// ledger_number. A to_id tie-break could not change the result, and would cost an incremental +// sort on top of the index's pathkeys. func (m *IngestStoreModel) GetOldestLedger(ctx context.Context) (uint32, error) { start := time.Now() oldest, err := db.QueryOne[uint32](ctx, m.DB, - `SELECT ledger_number FROM transactions ORDER BY ledger_created_at ASC, to_id ASC LIMIT 1`) + `SELECT ledger_number FROM transactions ORDER BY ledger_created_at ASC LIMIT 1`) duration := time.Since(start).Seconds() m.Metrics.QueryDuration.WithLabelValues("GetOldestLedger", "transactions").Observe(duration) if err != nil && !errors.Is(err, pgx.ErrNoRows) { diff --git a/internal/db/migrations/2025-06-10.2-transactions.sql b/internal/db/migrations/2025-06-10.2-transactions.sql index b3aae2416..39a6ce205 100644 --- a/internal/db/migrations/2025-06-10.2-transactions.sql +++ b/internal/db/migrations/2025-06-10.2-transactions.sql @@ -25,12 +25,6 @@ SELECT enable_chunk_skipping('transactions', 'to_id'); -- non-overlapping chunks at plan time. SELECT enable_chunk_skipping('transactions', 'ledger_number'); --- TimescaleDB's default single-column index on the partition column. Retention drops chunks by --- range metadata, and no query path filters this table by bare ledger_created_at (reads go through --- the primary key, idx_transactions_hash, or the transactions_accounts table), so it has zero --- consumers. -DROP INDEX IF EXISTS transactions_ledger_created_at_idx; - CREATE INDEX idx_transactions_hash ON transactions(hash); -- Table: transactions_accounts (TimescaleDB hypertable for automatic cleanup with retention) diff --git a/internal/ingest/timescaledb.go b/internal/ingest/timescaledb.go index 9c8e34db9..d9de2444f 100644 --- a/internal/ingest/timescaledb.go +++ b/internal/ingest/timescaledb.go @@ -26,9 +26,10 @@ var hypertables = data.BulkCopyTableNames() // run history survive repeated calls (e.g. every process restart). When // retention is enabled, the reconciliation job keeps oldest_ingest_ledger in // sync with the actual minimum ledger remaining after chunk drops. Compression -// schedule interval updates how frequently existing compression policy jobs -// run (does not create new policies). Compress after updates how long after a -// chunk closes before it becomes eligible for compression. +// schedule interval puts the existing compression policy jobs on staggered +// fixed schedules (does not create new policies) — see +// staggerCompressionJob. Compress after updates how long after a chunk closes +// before it becomes eligible for compression. func configureHypertableSettings(ctx context.Context, pool *pgxpool.Pool, chunkInterval, retentionPeriod, oldestCursorName, compressionScheduleInterval, compressAfter string, maxChunksToCompress int) error { for _, table := range hypertables { if _, err := pool.Exec(ctx, @@ -50,8 +51,8 @@ func configureHypertableSettings(ctx context.Context, pool *pgxpool.Pool, chunkI return fmt.Errorf("configuring reconciliation job: %w", err) } - if compressionScheduleInterval != "" { - for _, table := range hypertables { + if compressionScheduleInterval != "" || compressAfter != "" || maxChunksToCompress > 0 { + for i, table := range hypertables { var jobID int err := pool.QueryRow(ctx, `SELECT job_id FROM timescaledb_information.jobs @@ -60,71 +61,93 @@ func configureHypertableSettings(ctx context.Context, pool *pgxpool.Pool, chunkI table, ).Scan(&jobID) if err != nil { - log.Ctx(ctx).Warnf("No compression policy found for %s, skipping schedule update", table) + log.Ctx(ctx).Warnf("No compression policy found for %s, skipping compression settings", table) continue } - if _, err := pool.Exec(ctx, - "SELECT alter_job($1, schedule_interval => $2::interval)", - jobID, compressionScheduleInterval, - ); err != nil { - return fmt.Errorf("updating compression schedule interval on %s (job %d): %w", table, jobID, err) + if compressionScheduleInterval != "" { + if err := staggerCompressionJob(ctx, pool, jobID, table, compressionScheduleInterval, i, len(hypertables)); err != nil { + return err + } } - log.Ctx(ctx).Infof("Set compression schedule interval %q on %s (job %d)", compressionScheduleInterval, table, jobID) - } - } - if compressAfter != "" { - for _, table := range hypertables { - var jobID int - err := pool.QueryRow(ctx, - `SELECT job_id FROM timescaledb_information.jobs - WHERE proc_name = 'policy_compression' - AND hypertable_name = $1`, - table, - ).Scan(&jobID) - if err != nil { - log.Ctx(ctx).Warnf("No compression policy found for %s, skipping compress_after update", table) - continue + if compressAfter != "" { + if _, err := pool.Exec(ctx, + `SELECT alter_job($1, config => jsonb_set( + (SELECT config FROM timescaledb_information.jobs WHERE job_id = $1), + '{compress_after}', to_jsonb($2::text)))`, + jobID, compressAfter, + ); err != nil { + return fmt.Errorf("updating compress_after on %s (job %d): %w", table, jobID, err) + } + log.Ctx(ctx).Infof("Set compress_after %q on %s (job %d)", compressAfter, table, jobID) } - if _, err := pool.Exec(ctx, - `SELECT alter_job($1, config => jsonb_set( - (SELECT config FROM timescaledb_information.jobs WHERE job_id = $1), - '{compress_after}', to_jsonb($2::text)))`, - jobID, compressAfter, - ); err != nil { - return fmt.Errorf("updating compress_after on %s (job %d): %w", table, jobID, err) + if maxChunksToCompress > 0 { + if _, err := pool.Exec(ctx, + `SELECT alter_job($1, config => config || jsonb_build_object('maxchunks_to_compress', $2::int)) + FROM timescaledb_information.jobs WHERE job_id = $1`, + jobID, maxChunksToCompress, + ); err != nil { + return fmt.Errorf("updating maxchunks_to_compress on %s (job %d): %w", table, jobID, err) + } + log.Ctx(ctx).Infof("Set maxchunks_to_compress %d on %s (job %d)", maxChunksToCompress, table, jobID) } - log.Ctx(ctx).Infof("Set compress_after %q on %s (job %d)", compressAfter, table, jobID) } } - if maxChunksToCompress > 0 { - for _, table := range hypertables { - var jobID int - err := pool.QueryRow(ctx, - `SELECT job_id FROM timescaledb_information.jobs - WHERE proc_name = 'policy_compression' - AND hypertable_name = $1`, - table, - ).Scan(&jobID) - if err != nil { - log.Ctx(ctx).Warnf("No compression policy found for %s, skipping maxchunks_to_compress update", table) - continue - } + return nil +} - if _, err := pool.Exec(ctx, - `SELECT alter_job($1, config => config || jsonb_build_object('maxchunks_to_compress', $2::int)) - FROM timescaledb_information.jobs WHERE job_id = $1`, - jobID, maxChunksToCompress, - ); err != nil { - return fmt.Errorf("updating maxchunks_to_compress on %s (job %d): %w", table, jobID, err) - } - log.Ctx(ctx).Infof("Set maxchunks_to_compress %d on %s (job %d)", maxChunksToCompress, table, jobID) - } +// staggerCompressionJob converges a compression policy job onto a fixed +// schedule whose origin sits index/total of the way into the schedule +// interval. Each hypertable's policy is auto-created with an identical +// schedule, so without staggering all of them fire at the same instant and +// compress their just-closed chunks concurrently — an I/O storm that starves +// live ingestion's persist stage. Staggered, at most one policy comes due at +// a time (best effort: a job that overruns its slot still overlaps the next +// one). The origin is anchored to the interval grid (date_bin against the +// epoch), which makes the slot deterministic: a job already on its slot is +// left untouched, so next_start and run history survive process restarts. +// date_bin limits the schedule interval to month-free values (minutes, +// hours, days). +func staggerCompressionJob(ctx context.Context, pool *pgxpool.Pool, jobID int, table, scheduleInterval string, index, total int) error { + slotOffset := float64(index) / float64(total) + + var onSlot bool + if err := pool.QueryRow(ctx, + `SELECT COALESCE( + j.schedule_interval = $2::interval + AND j.fixed_schedule + AND js.next_start - $2::interval * $3::float8 + = date_bin($2::interval, js.next_start - $2::interval * $3::float8, 'epoch'::timestamptz), + false) + FROM timescaledb_information.jobs j + LEFT JOIN timescaledb_information.job_stats js USING (job_id) + WHERE j.job_id = $1`, + jobID, scheduleInterval, slotOffset, + ).Scan(&onSlot); err != nil { + return fmt.Errorf("checking compression schedule slot on %s (job %d): %w", table, jobID, err) + } + if onSlot { + return nil } + // The first run lands on the job's slot within the NEXT grid interval, + // which is always in the future; fixed_schedule keeps every later run on + // initial_start + n*interval, i.e. on the slot. + if _, err := pool.Exec(ctx, + `SELECT alter_job($1, + schedule_interval => $2::interval, + fixed_schedule => true, + initial_start => date_bin($2::interval, now(), 'epoch'::timestamptz) + + $2::interval + + $2::interval * $3::float8)`, + jobID, scheduleInterval, slotOffset, + ); err != nil { + return fmt.Errorf("updating compression schedule on %s (job %d): %w", table, jobID, err) + } + log.Ctx(ctx).Infof("Set compression schedule interval %q (slot %d/%d) on %s (job %d)", scheduleInterval, index+1, total, table, jobID) return nil } @@ -214,7 +237,7 @@ func configureReconciliationJob(ctx context.Context, pool *pgxpool.Pool, retenti stored INTEGER; BEGIN SELECT ledger_number INTO actual_min FROM transactions - ORDER BY ledger_created_at ASC, to_id ASC LIMIT 1; + ORDER BY ledger_created_at ASC LIMIT 1; IF actual_min IS NULL THEN RETURN; END IF; SELECT value::integer INTO stored FROM ingest_store WHERE key = config->>'cursor_name'; IF stored IS NULL OR actual_min <= stored THEN RETURN; END IF; @@ -231,9 +254,10 @@ func configureReconciliationJob(ctx context.Context, pool *pgxpool.Pool, retenti ).Scan(&jobID) switch { case errors.Is(err, pgx.ErrNoRows): - // Runs every 1 hour: cheap enough (oldest chunk metadata + 1 row from - // ingest_store) to not need coordination with the retention job's own - // schedule, and idempotent — a no-op once the cursor is already correct. + // Runs every 1 hour: cheap enough (an ordered ChunkAppend LIMIT 1 backed by the + // hypertable's partition-column index, plus 1 row from ingest_store) to not need + // coordination with the retention job's own schedule, and idempotent — a no-op + // once the cursor is already correct. if _, err = pool.Exec(ctx, ` SELECT add_job( 'reconcile_oldest_cursor', diff --git a/internal/ingest/timescaledb_test.go b/internal/ingest/timescaledb_test.go index 0cdd61f13..d9abe1b9e 100644 --- a/internal/ingest/timescaledb_test.go +++ b/internal/ingest/timescaledb_test.go @@ -218,6 +218,55 @@ func TestConfigureHypertableSettings(t *testing.T) { assert.Equal(t, 1, count, "expected exactly 1 reconciliation job") }) + t.Run("reconciliation_job_advances_cursor_to_oldest_ledger", func(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() + + err = configureHypertableSettings(ctx, dbConnectionPool, "1 day", "30 days", "oldest_ledger_cursor", "", "", 0) + require.NoError(t, err) + + // One transaction per chunk, inserted out of ledger order, so the lookup inside + // reconcile_oldest_cursor has to order across chunks rather than take the first + // row it reaches. + for _, tx := range []struct { + toID int64 + ledgerNumber int32 + createdAt string + }{ + {toID: 2, ledgerNumber: 150, createdAt: "2026-01-05T00:00:00Z"}, + {toID: 3, ledgerNumber: 200, createdAt: "2026-01-09T00:00:00Z"}, + {toID: 1, ledgerNumber: 100, createdAt: "2026-01-01T00:00:00Z"}, + } { + _, err = dbConnectionPool.Exec(ctx, + `INSERT INTO transactions (hash, to_id, fee_charged, result_code, ledger_number, ledger_created_at) + VALUES ($1, $2, 100, 'TransactionResultCodeTxSuccess', $3, $4::timestamptz)`, + []byte{byte(tx.toID)}, tx.toID, tx.ledgerNumber, tx.createdAt) + require.NoError(t, err) + } + + _, err = dbConnectionPool.Exec(ctx, + `INSERT INTO ingest_store (key, value) VALUES ('oldest_ledger_cursor', '50')`) + require.NoError(t, err) + + jobID, err := db.QueryOne[int](ctx, dbConnectionPool, + `SELECT job_id FROM timescaledb_information.jobs WHERE proc_name = 'reconcile_oldest_cursor'`, + ) + require.NoError(t, err) + + // run_job invokes the job exactly as the background scheduler does. + _, err = dbConnectionPool.Exec(ctx, "CALL run_job($1)", jobID) + require.NoError(t, err) + + cursor, err := db.QueryOne[string](ctx, dbConnectionPool, + `SELECT value FROM ingest_store WHERE key = 'oldest_ledger_cursor'`) + require.NoError(t, err) + assert.Equal(t, "100", cursor, "cursor should advance to the ledger of the oldest transaction") + }) + t.Run("reconciliation_job_idempotent", func(t *testing.T) { dbt := dbtest.Open(t) defer dbt.Close() @@ -490,6 +539,88 @@ func TestConfigureHypertableSettings(t *testing.T) { } }) + t.Run("compression_schedule_staggered", func(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() + + err = configureHypertableSettings(ctx, dbConnectionPool, "1 day", "", "oldest_ledger_cursor", "1 hour", "", 0) + require.NoError(t, err) + + // Each policy job must sit on its own slot of the hour grid: job i's + // next_start minus i/5 of the interval lands exactly on an hour + // boundary, and the run is in the future. + for i, table := range hypertables { + onSlot, err := db.QueryOne[bool](ctx, dbConnectionPool, + `SELECT j.fixed_schedule + AND js.next_start > now() + AND js.next_start - '1 hour'::interval * $2::float8 + = date_bin('1 hour'::interval, js.next_start - '1 hour'::interval * $2::float8, 'epoch'::timestamptz) + FROM timescaledb_information.jobs j + JOIN timescaledb_information.job_stats js USING (job_id) + WHERE j.proc_name = 'policy_compression' AND j.hypertable_name = $1`, + table, float64(i)/float64(len(hypertables)), + ) + require.NoError(t, err, "querying stagger slot for %s", table) + assert.True(t, onSlot, "compression job for %s should sit on slot %d of the hour grid", table, i) + } + + // All five slots are distinct, so no two policies come due together. + distinctStarts, err := db.QueryOne[int](ctx, dbConnectionPool, + `SELECT COUNT(DISTINCT js.next_start) + FROM timescaledb_information.jobs j + JOIN timescaledb_information.job_stats js USING (job_id) + WHERE j.proc_name = 'policy_compression'`, + ) + require.NoError(t, err) + assert.Equal(t, len(hypertables), distinctStarts, "each compression policy should own a distinct slot") + }) + + t.Run("compression_schedule_stagger_idempotent", func(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() + + err = configureHypertableSettings(ctx, dbConnectionPool, "1 day", "", "oldest_ledger_cursor", "1 hour", "", 0) + require.NoError(t, err) + + startsBefore := make(map[string]string) + for _, table := range hypertables { + nextStart, err := db.QueryOne[string](ctx, dbConnectionPool, + `SELECT js.next_start::text + FROM timescaledb_information.jobs j + JOIN timescaledb_information.job_stats js USING (job_id) + WHERE j.proc_name = 'policy_compression' AND j.hypertable_name = $1`, + table, + ) + require.NoError(t, err, "querying next_start for %s", table) + startsBefore[table] = nextStart + } + + // A job already on its slot is left untouched, so a restart must not + // move next_start (which would postpone compression by a cycle). + err = configureHypertableSettings(ctx, dbConnectionPool, "1 day", "", "oldest_ledger_cursor", "1 hour", "", 0) + require.NoError(t, err) + + for _, table := range hypertables { + nextStart, err := db.QueryOne[string](ctx, dbConnectionPool, + `SELECT js.next_start::text + FROM timescaledb_information.jobs j + JOIN timescaledb_information.job_stats js USING (job_id) + WHERE j.proc_name = 'policy_compression' AND j.hypertable_name = $1`, + table, + ) + require.NoError(t, err, "querying next_start for %s", table) + assert.Equal(t, startsBefore[table], nextStart, "re-application should not move %s's slot", table) + } + }) + t.Run("reconciliation_job_scheduled_after_retention", func(t *testing.T) { dbt := dbtest.Open(t) defer dbt.Close() diff --git a/internal/metrics/db.go b/internal/metrics/db.go index 19f046dba..8baba326f 100644 --- a/internal/metrics/db.go +++ b/internal/metrics/db.go @@ -34,9 +34,12 @@ type DBMetrics struct { func newDBMetrics(reg prometheus.Registerer) *DBMetrics { m := &DBMetrics{ QueryDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{ - Name: "wallet_db_query_duration_seconds", - Help: "Duration of database queries.", - Buckets: prometheus.ExponentialBuckets(0.0001, 2.5, 10), + Name: "wallet_db_query_duration_seconds", + Help: "Duration of database queries.", + // 0.1ms .. ~17.7s: the top buckets must cover multi-second bulk + // COPYs (transactions/operations/state_changes at high tx volume), + // or histogram_quantile clips every bulk write into +Inf. + Buckets: prometheus.ExponentialBuckets(0.0001, 3, 12), }, []string{"query_type", "table"}), QueriesTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "wallet_db_queries_total",