fix(idempotency): sweep with a same-session lock and bounded batches

docs/billing-idempotency.md

@@ -594,11 +594,123 @@ UPDATE usage_events SET status = 'applied' WHERE stellar_tx_hash IS NOT NULL;
 
 6. **Deploy and monitor** for duplicate request rate
 
+## Idempotency Store Maintenance (Sweeper)
+
+### Overview
+
+The **idempotency sweeper** is a background job that removes expired rows from `idempotency_store` to prevent unbounded table growth. It runs at regular intervals and is designed to be safe for concurrent deployments.
+
+### How It Works
+
+The sweeper uses PostgreSQL advisory locks and batched deletes to:
+
+1. **Acquire a transaction-scoped advisory lock** — ensures only one sweeper instance runs at a time across all replicas
+2. **Delete in batches** — splits large deletes into smaller transactions to avoid locking the table for extended periods
+3. **Respect time budgets** — stops early if a per-run time limit is reached, ensuring the next tick can start
+4. **Automatically release locks** — uses transaction-scoped locks (`pg_try_advisory_xact_lock`) so locks are released on COMMIT or ROLLBACK, preventing leaks
+
+### Configuration
+
+Control the sweeper via environment variables:
+
+| Variable | Type | Default | Description |
+|----------|------|---------|-------------|
+| `IDEMPOTENCY_SWEEPER_ENABLED` | bool | `true` | Enable/disable the sweeper job |
+| `IDEMPOTENCY_SWEEPER_INTERVAL_MS` | int | `60000` | Run interval in milliseconds (60 seconds) |
+| `IDEMPOTENCY_SWEEPER_BATCH_SIZE` | int | `5000` | Rows deleted per transaction batch |
+| `IDEMPOTENCY_SWEEPER_TIME_BUDGET_MS` | int | `5000` | Max wall-clock time (ms) per run |
+
+### Example
+
+```bash
+# Run the sweeper every 30 seconds, deleting 1000 rows per batch, with a 3-second time budget
+IDEMPOTENCY_SWEEPER_ENABLED=true
+IDEMPOTENCY_SWEEPER_INTERVAL_MS=30000
+IDEMPOTENCY_SWEEPER_BATCH_SIZE=1000
+IDEMPOTENCY_SWEEPER_TIME_BUDGET_MS=3000
+```
+
+### Metrics
+
+**Gauge: `idempotency_store_rows`**
+
+- Help: "Estimated number of rows in the idempotency_store table (from pg_class.reltuples; may lag behind true count until next ANALYZE). Updated by the idempotency sweeper on each run."
+- Updated on every sweep run
+- Uses `pg_class.reltuples` for efficiency (cheaper than `COUNT(*)`)
+- Falls back to `COUNT(*)` if reltuples is unavailable
+
+### Monitoring
+
+#### Healthy Operation
+
+1. **Rows remain in expected range** — gauge stays within your retention window
+2. **Sweeper runs regularly** — logs show "Sweep complete" messages at expected intervals
+3. **No lock contention** — if a second instance tries to sweep, it logs "Another instance owns the sweeper lock"
+4. **Batches complete within time budget** — logs show multiple batches if rows exceed batch size
+
+#### Troubleshooting
+
+**Issue: Gauge stuck at 0 or growing unbounded**
+
+Diagnosis:
+```sql
+SELECT COUNT(*) FROM idempotency_store;
+SELECT reltuples FROM pg_class WHERE relname = 'idempotency_store';
+```
+
+Solution:
+- If `COUNT(*)` > 0 but `reltuples` ≈ 0, run `ANALYZE idempotency_store` to update statistics
+- If sweeper logs show "Sweep failed", check for table locks or permissions
+
+**Issue: Sweeper logs don't appear**
+
+Diagnosis:
+- Check if `IDEMPOTENCY_SWEEPER_ENABLED` is set to `false`
+- Verify the job was wired in `src/index.ts` and started at boot
+
+Solution:
+- Set `IDEMPOTENCY_SWEEPER_ENABLED=true` and restart the server
+- Confirm startup logs include "[idempotencySweeper] Started with interval..."
+
+**Issue: High delete latency / table lock contention**
+
+Diagnosis:
+- Check logs for "Time budget reached" — indicates batches are taking longer than expected
+- Query `pg_stat_user_tables` to see if `idempotency_store` has high sequential scan count
+
+Solution:
+- Reduce `IDEMPOTENCY_SWEEPER_BATCH_SIZE` (e.g., 5000 → 1000)
+- Increase `IDEMPOTENCY_SWEEPER_TIME_BUDGET_MS` if the table is very large
+- Ensure the `expires_at` index exists and is not bloated
+
+### Test Scenarios
+
+The sweeper is tested with:
+
+1. **Single sweep** — lock acquired, rows deleted, lock released (no leak)
+2. **Concurrent sweeps** — two instances contend for lock; one deletes, other skips
+3. **Batching** — 25k rows with 5k batches → 5 batches, all removed
+4. **Time budget** — tight budget stops early; next run continues
+5. **Error recovery** — delete error rolls back and releases lock/client
+6. **Regression** — no lock leaks after multiple sweeps
+
+See `tests/integration/idempotencySweeper.test.ts` for detailed integration tests using testcontainers.
+
+### Performance Notes
+
+- **Lock scope**: Transaction-scoped (released automatically on transaction end)
+- **Lock contention**: Minimal — only one instance holds the lock per sweep
+- **Table locking**: Row-level locks are brief; transactions commit after each batch
+- **Metrics cost**: Uses `pg_class.reltuples` (free) instead of `COUNT(*)` (expensive)
+- **Scale**: Tested with 25k rows and configurable batch sizes
+
 ## References
 
 - [Idempotency Keys - Stripe Documentation](https://stripe.com/docs/api/idempotent_requests)
 - [PostgreSQL Unique Constraints](https://www.postgresql.org/docs/current/ddl-constraints.html#DDL-CONSTRAINTS-UNIQUE-CONSTRAINTS)
-- (Database Transaction Isolation](https://www.postgresql.org/docs/current/transaction-iso.html)
+- [Database Transaction Isolation](https://www.postgresql.org/docs/current/transaction-iso.html)
+- [PostgreSQL Advisory Locks](https://www.postgresql.org/docs/current/explicit-locking.html#ADVISORY-LOCKS)
+- [PostgreSQL CTID](https://www.postgresql.org/docs/current/ddl-system-columns.html)
 
 ## HTTP idempotency leases and response replay
 

src/config/env.ts

@@ -349,11 +349,11 @@
     STELLAR_NETWORK: stellarNetworkSchema.optional(),
     SOROBAN_NETWORK: stellarNetworkSchema.optional(),
 
     STELLAR_TESTNET_HORIZON_URL: z
       .string()
       .url()
       .default("https://horizon-testnet.stellar.org"),
     STELLAR_MAINNET_HORIZON_URL: z
       .string()
       .url()
       .default("https://horizon.stellar.org"),
@@ -366,8 +366,8 @@
       .url()
       .default("https://soroban-mainnet.stellar.org"),
 
     STELLAR_TESTNET_VAULT_CONTRACT_ID: contractId.optional(),
     STELLAR_MAINNET_VAULT_CONTRACT_ID: contractId.optional(),
     STELLAR_TESTNET_SETTLEMENT_CONTRACT_ID: contractId.optional(),
     STELLAR_MAINNET_SETTLEMENT_CONTRACT_ID: contractId.optional(),
     VAULT_SNAPSHOT_MAX_AGE_MS: z.coerce.number().int().positive().default(60_000),
@@ -501,11 +501,26 @@
       .int()
       .positive()
       .default(86400),
+    IDEMPOTENCY_SWEEPER_ENABLED: z
+      .string()
+      .optional()
+      .transform((v) => v !== "false")
+      .default(true),
     IDEMPOTENCY_SWEEPER_INTERVAL_MS: z.coerce
       .number()
       .int()
       .positive()
       .default(60_000),
+    IDEMPOTENCY_SWEEPER_BATCH_SIZE: z.coerce
+      .number()
+      .int()
+      .positive()
+      .default(5000),
+    IDEMPOTENCY_SWEEPER_TIME_BUDGET_MS: z.coerce
+      .number()
+      .int()
+      .positive()
+      .default(5000),
 
     // Slow query alerting
     SLOW_QUERY_ALERT_WEBHOOK_URL: z.string().url().optional(),

src/config/index.ts

@@ -74,7 +74,7 @@
   networkPassphrase: TESTNET_NETWORK_PASPPRASE,
   vaultContractId: env.STELLAR_TESTNET_VAULT_CONTRACT_ID,
   settlementContractId: env.STELLAR_TESTNET_SETTLEMENT_CONTRACT_ID,
   revenuePoolContractId: env.STELLAR_TESTNET_REVENUE_POOL_CONTRACT_ID,
   xusdContractId: env.STELLAR_TESTNET_USDC_CONTRACT_ID,
 } as StellarNetworkConfig;
 
@@ -344,6 +344,13 @@
     intervalMs: env.MONTHLY_INVOICE_JOB_INTERVAL_MS,
   },
 
+  idempotencySweeper: {
+    enabled: env.IDEMPOTENCY_SWEEPER_ENABLED,
+    intervalMs: env.IDEMPOTENCY_SWEEPER_INTERVAL_MS,
+    batchSize: env.IDEMPOTENCY_SWEEPER_BATCH_SIZE,
+    timeBudgetMs: env.IDEMPOTENCY_SWEEPER_TIME_BUDGET_MS,
+  },
+
   sloAlert: {
     enabled:
       Boolean(env.SLO_ALERT_WEBHOOK_URL) && env.SLO_ROUTE_CONFIGS.length > 0,

src/index.ts

@@ -3,10 +3,11 @@ import type { Socket } from "net";
 import { createApp } from "./app.js";
 import { config } from "./config/index.js";
 import { initializeDb, closeDb } from "./db/index.js";
-import { closePgPool } from "./db.js";
+import { closePgPool, pool } from "./db.js";
 import { closeDbPool } from "./config/health.js";
 import { disconnectPrisma } from "./lib/prisma.js";
 import { createGracefulShutdownHandler } from "./lifecycle/shutdown.js";
+import { createIdempotencySweeperJob } from "./services/idempotencySweeper.js";
 
 // Re-exported for backward compatibility with existing consumers/tests.
 export {
@@ -44,14 +45,39 @@ async function startServer(): Promise<void> {
       socket.once("close", () => activeConnections.delete(socket));
     });
 
+    // Create and start the idempotency sweeper job (if enabled)
+    let sweeperJob = null;
+    const subsystems = [];
+
+    if (config.idempotencySweeper.enabled) {
+      sweeperJob = createIdempotencySweeperJob(pool, {
+        intervalMs: config.idempotencySweeper.intervalMs,
+        batchSize: config.idempotencySweeper.batchSize,
+        timeBudgetMs: config.idempotencySweeper.timeBudgetMs,
+      });
+      sweeperJob.start();
+      console.log(
+        `[idempotencySweeper] Started with interval ${config.idempotencySweeper.intervalMs}ms, batch size ${config.idempotencySweeper.batchSize}, time budget ${config.idempotencySweeper.timeBudgetMs}ms`,
+      );
+
+      // Register sweeper as a graceful subsystem
+      subsystems.push({
+        name: "idempotency-sweeper",
+        beginShutdown: () => {
+          sweeperJob!.beginShutdown();
+        },
+        awaitIdle: () => sweeperJob!.awaitIdle(),
+      });
+    }
+
     const gracefulShutdown = createGracefulShutdownHandler({
       server,
       activeConnections,
       closeDatabase: async () => {
         await closeDb();
         await Promise.allSettled([closePgPool(), disconnectPrisma(), closeDbPool()]);
       },
-      subsystems: [],
+      subsystems,
       timeoutMs: 30_000,
     });
 

src/metrics.ts

@@ -681,7 +681,7 @@ const proxyPrematureAbortsTotal = new client.Counter({
 
 const idempotencyStoreRows = new client.Gauge({
   name: 'idempotency_store_rows',
-  help: 'Current number of rows in the idempotency_store table',
+  help: 'Estimated number of rows in the idempotency_store table (from pg_class.reltuples; may lag behind true count until next ANALYZE). Updated by the idempotency sweeper on each run.',
 });
 
 const endpointThroughputSaturationRatio = new client.Gauge({