From 9eace1409a1583abff2303b0eeccedf1beec3071 Mon Sep 17 00:00:00 2001 From: Aditya Vyas Date: Mon, 10 Aug 2026 15:54:38 -0400 Subject: [PATCH 1/3] fix(metrics): widen DB query-duration buckets to cover bulk COPYs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The wallet_db_query_duration_seconds histogram topped out at 0.38s (ExponentialBuckets(0.0001, 2.5, 10)), so every multi-second bulk COPY landed in +Inf and histogram_quantile panels clipped at ~0.38s, under-reporting exactly the queries being optimized. Widen to ExponentialBuckets(0.0001, 3, 12) — 0.1ms through ~17.7s. --- internal/metrics/db.go | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) 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", From 6e3811df5a2d411ceab42468873cf9aa5ab5b742 Mon Sep 17 00:00:00 2001 From: Aditya Vyas Date: Mon, 10 Aug 2026 16:06:14 -0400 Subject: [PATCH 2/3] fix(ingest): keep the ordered path to the oldest chunk for the oldest-ledger lookup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The oldest-ledger lookup (SELECT ledger_number FROM transactions ORDER BY ledger_created_at ASC LIMIT 1) has two consumers — backfill gap detection's left bound (GetOldestLedger) and the hourly reconcile_oldest_cursor TimescaleDB job — and neither had an ordered path: the transactions migration dropped TimescaleDB's default partition-column index as consumer-less, an audit that counted WHERE-clause consumers and missed that this query consumes the index's ordering. Without it the planner pulls the first row from every chunk instead of running an ordered ChunkAppend that stops at the oldest (12-187s on an 87GB DB; the hourly job alone accounted for 45% of all DB disk-read time and flushed shared_buffers every run). Keep the default index (fresh databases get it from create_hypertable; already-migrated environments need a one-time manual `CREATE INDEX transactions_ledger_created_at_idx ON transactions (ledger_created_at DESC)`) and drop the to_id tie-break from both query sites: close times increase strictly, so rows sharing a ledger_created_at carry the same ledger_number and the tie-break could only force an incremental sort on top of the index's pathkeys. Covered by a new test that runs the reconciliation job via run_job against out-of-order chunks. --- internal/data/ingest_store.go | 11 ++++- .../migrations/2025-06-10.2-transactions.sql | 6 --- internal/ingest/timescaledb.go | 9 ++-- internal/ingest/timescaledb_test.go | 49 +++++++++++++++++++ 4 files changed, 64 insertions(+), 11 deletions(-) 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..f42aec685 100644 --- a/internal/ingest/timescaledb.go +++ b/internal/ingest/timescaledb.go @@ -214,7 +214,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 +231,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..4607c6994 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() From 65a26233ca760080f6ec62d797407e2583298943 Mon Sep 17 00:00:00 2001 From: Aditya Vyas Date: Mon, 10 Aug 2026 23:20:53 -0400 Subject: [PATCH 3/3] perf(ingest): stagger the per-hypertable compression policies across the schedule interval MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each hypertable's columnstore policy is auto-created with an identical schedule, so all five fire at the same instant and compress their just-closed chunks concurrently — an I/O storm that starves the persist stage (measured on the loadtest rig: persist p50 0.68s -> 3.0s for 18 of every 60 minutes). Converge each policy onto a fixed schedule anchored to the interval grid with a distinct per-table slot offset, so at most one policy comes due at a time. Jobs already on their slot are left untouched across restarts, preserving next_start and run history. --- internal/ingest/timescaledb.go | 135 ++++++++++++++++------------ internal/ingest/timescaledb_test.go | 82 +++++++++++++++++ 2 files changed, 161 insertions(+), 56 deletions(-) diff --git a/internal/ingest/timescaledb.go b/internal/ingest/timescaledb.go index f42aec685..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 } diff --git a/internal/ingest/timescaledb_test.go b/internal/ingest/timescaledb_test.go index 4607c6994..d9abe1b9e 100644 --- a/internal/ingest/timescaledb_test.go +++ b/internal/ingest/timescaledb_test.go @@ -539,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()