Fix the DB query-duration buckets, the oldest-ledger index path, and compression scheduling

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) {

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)

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',

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()

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",