diff --git a/backend/migrations/1723000000000_add-token-columns.js b/backend/migrations/1723000000000_add-token-columns.js index 19a78904..4640fe37 100644 --- a/backend/migrations/1723000000000_add-token-columns.js +++ b/backend/migrations/1723000000000_add-token-columns.js @@ -1,31 +1,27 @@ -/** - * @param { import("pg").Pool } pool - */ -exports.up = async (pool) => { - // Add original_token and original_amount columns to bets table - await pool.query(` - ALTER TABLE bets +/* eslint-disable camelcase */ + +// NOTE: this migration previously destructured its argument as a `pg.Pool` and +// called `pool.query(...)`. node-pg-migrate passes a MigrationBuilder, so it +// threw `pool.query is not a function` and never applied. Rewritten to use +// `pgm.sql`; the SQL itself is unchanged and remains idempotent. + +exports.shorthands = undefined; + +exports.up = (pgm) => { + pgm.sql(` + ALTER TABLE bets ADD COLUMN IF NOT EXISTS original_token VARCHAR(56) DEFAULT 'XLM', - ADD COLUMN IF NOT EXISTS original_amount NUMERIC(40) DEFAULT '0'; + ADD COLUMN IF NOT EXISTS original_amount NUMERIC(40) DEFAULT '0' `); - // Create indexes for new columns - await pool.query(` - CREATE INDEX IF NOT EXISTS bets_original_token_idx ON bets(original_token); - `); + pgm.sql('CREATE INDEX IF NOT EXISTS bets_original_token_idx ON bets(original_token)'); }; -/** - * @param { import("pg").Pool } pool - */ -exports.down = async (pool) => { - await pool.query(` - DROP INDEX IF EXISTS bets_original_token_idx; - `); - - await pool.query(` - ALTER TABLE bets +exports.down = (pgm) => { + pgm.sql('DROP INDEX IF EXISTS bets_original_token_idx'); + pgm.sql(` + ALTER TABLE bets DROP COLUMN IF EXISTS original_token, - DROP COLUMN IF EXISTS original_amount; + DROP COLUMN IF EXISTS original_amount `); -}; \ No newline at end of file +}; diff --git a/backend/migrations/1724000000000_partition-hot-tables.js b/backend/migrations/1724000000000_partition-hot-tables.js new file mode 100644 index 00000000..208f4e4b --- /dev/null +++ b/backend/migrations/1724000000000_partition-hot-tables.js @@ -0,0 +1,272 @@ +/* eslint-disable camelcase */ + +// Converts the two highest-volume tables to declarative range partitions. +// +// bets -> RANGE (placed_at) +// blockchain_events -> RANGE (ledger_close_time) +// +// Postgres cannot ALTER an existing table into a partitioned one, so each +// table is rebuilt: rename aside, create partitioned parent, copy, drop. +// +// Partition-key constraint: every UNIQUE/PRIMARY KEY on a partitioned table +// must contain the partition key. `id` becomes (id, ) and the tx_hash +// guard becomes (tx_hash, ). See MIGRATION NOTE in the down() comment +// and docs in src/db/partitionManager.ts for why this stays safe. + +exports.shorthands = undefined; + +// Partitions are monthly. Seed a window around the existing data so the copy +// back in has somewhere to land; partitionManager keeps the window rolling. +const SEED_MONTHS_BACK = 6; +const SEED_MONTHS_FORWARD = 3; + +function monthBounds(offset) { + const start = new Date(Date.UTC(new Date().getUTCFullYear(), new Date().getUTCMonth() + offset, 1)); + const end = new Date(Date.UTC(start.getUTCFullYear(), start.getUTCMonth() + 1, 1)); + const suffix = `${start.getUTCFullYear()}_${String(start.getUTCMonth() + 1).padStart(2, '0')}`; + return { suffix, from: start.toISOString(), to: end.toISOString() }; +} + +function seedPartitions(pgm, parent) { + for (let i = -SEED_MONTHS_BACK; i <= SEED_MONTHS_FORWARD; i += 1) { + const { suffix, from, to } = monthBounds(i); + pgm.sql(` + CREATE TABLE IF NOT EXISTS ${parent}_${suffix} + PARTITION OF ${parent} + FOR VALUES FROM ('${from}') TO ('${to}') + `); + } + // Catch-all for rows older than the seeded window, so the data copy cannot + // fail on a stray historical row. Never routed to by new writes. + pgm.sql(` + CREATE TABLE IF NOT EXISTS ${parent}_default PARTITION OF ${parent} DEFAULT + `); +} + +exports.up = (pgm) => { + // ---------------------------------------------------------------- bets --- + pgm.sql('ALTER TABLE bets RENAME TO bets_legacy'); + pgm.sql('ALTER TABLE bets_legacy RENAME CONSTRAINT bets_pkey TO bets_legacy_pkey'); + + pgm.sql(` + CREATE TABLE bets ( + id BIGSERIAL NOT NULL, + market_id TEXT NOT NULL REFERENCES markets(market_id), + bettor_address TEXT NOT NULL, + side TEXT NOT NULL, + amount NUMERIC NOT NULL, + amount_xlm NUMERIC NOT NULL DEFAULT 0, + placed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + claimed BOOLEAN NOT NULL DEFAULT FALSE, + claimed_at TIMESTAMPTZ, + payout NUMERIC, + tx_hash TEXT NOT NULL, + ledger_sequence INTEGER NOT NULL DEFAULT 0, + original_token VARCHAR(56) DEFAULT 'XLM', + original_amount NUMERIC(40) DEFAULT '0', + PRIMARY KEY (id, placed_at), + UNIQUE (tx_hash, placed_at) + ) PARTITION BY RANGE (placed_at) + `); + + seedPartitions(pgm, 'bets'); + + pgm.sql(` + INSERT INTO bets ( + id, market_id, bettor_address, side, amount, amount_xlm, + placed_at, claimed, claimed_at, payout, tx_hash, ledger_sequence, + original_token, original_amount + ) + SELECT + id, market_id, bettor_address, side, amount, amount_xlm, + placed_at, claimed, claimed_at, payout, tx_hash, ledger_sequence, + original_token, original_amount + FROM bets_legacy + `); + + // Keep the sequence ahead of the copied ids. + pgm.sql(` + SELECT setval( + pg_get_serial_sequence('bets', 'id'), + GREATEST((SELECT COALESCE(MAX(id), 0) FROM bets), 1) + ) + `); + + pgm.sql('DROP TABLE bets_legacy'); + + pgm.createIndex('bets', 'market_id'); + pgm.createIndex('bets', 'bettor_address'); + pgm.createIndex('bets', ['market_id', 'placed_at']); + + // --------------------------------------------------- blockchain_events --- + pgm.sql('ALTER TABLE blockchain_events RENAME TO blockchain_events_legacy'); + pgm.sql( + 'ALTER TABLE blockchain_events_legacy RENAME CONSTRAINT blockchain_events_pkey TO blockchain_events_legacy_pkey', + ); + + pgm.sql(` + CREATE TABLE blockchain_events ( + id BIGSERIAL NOT NULL, + contract_address TEXT NOT NULL, + event_type TEXT NOT NULL, + payload JSONB NOT NULL DEFAULT '{}', + ledger_sequence INTEGER NOT NULL, + ledger_close_time TIMESTAMPTZ NOT NULL DEFAULT NOW(), + tx_hash TEXT NOT NULL, + processed BOOLEAN NOT NULL DEFAULT FALSE, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (id, ledger_close_time), + UNIQUE (tx_hash, ledger_close_time) + ) PARTITION BY RANGE (ledger_close_time) + `); + + seedPartitions(pgm, 'blockchain_events'); + + pgm.sql(` + INSERT INTO blockchain_events ( + id, contract_address, event_type, payload, ledger_sequence, + ledger_close_time, tx_hash, processed, created_at + ) + SELECT + id, contract_address, event_type, payload, ledger_sequence, + ledger_close_time, tx_hash, processed, created_at + FROM blockchain_events_legacy + `); + + pgm.sql(` + SELECT setval( + pg_get_serial_sequence('blockchain_events', 'id'), + GREATEST((SELECT COALESCE(MAX(id), 0) FROM blockchain_events), 1) + ) + `); + + pgm.sql('DROP TABLE blockchain_events_legacy'); + + pgm.createIndex('blockchain_events', 'ledger_sequence'); + pgm.createIndex('blockchain_events', 'contract_address'); + // Indexer scan path: unprocessed events in ledger order. + pgm.createIndex('blockchain_events', ['processed', 'ledger_sequence'], { + where: 'processed = FALSE', + }); + + // ------------------------------------------------- tx_hash dedup guard --- + // Partitioning weakened the old global `tx_hash UNIQUE`: the constraint must + // include the partition key, so it only holds *within* a partition. A write + // retried with a re-defaulted timestamp lands in a different partition and + // is accepted, double-crediting a bet. Verified, not theoretical. + // + // This registry restores the global guarantee. It stays unpartitioned so its + // PRIMARY KEY is genuinely global, and is narrow enough to stay cheap. + pgm.sql(` + CREATE TABLE tx_hash_registry ( + source_table TEXT NOT NULL, + tx_hash TEXT NOT NULL, + occurred_at TIMESTAMPTZ NOT NULL, + PRIMARY KEY (source_table, tx_hash) + ) + `); + + // occurred_at is carried purely so retention pruning can drop registry rows + // alongside the partitions they describe. + pgm.createIndex('tx_hash_registry', 'occurred_at'); + + pgm.sql(` + CREATE OR REPLACE FUNCTION tx_hash_registry_guard() RETURNS TRIGGER AS $$ + BEGIN + INSERT INTO tx_hash_registry (source_table, tx_hash, occurred_at) + VALUES ('bets', NEW.tx_hash, NEW.placed_at); + RETURN NEW; + END; + $$ LANGUAGE plpgsql + `); + + // Separate function per table: the timestamp column differs, and a trigger + // function cannot parameterise a column reference. + pgm.sql(` + CREATE OR REPLACE FUNCTION tx_hash_registry_guard_events() RETURNS TRIGGER AS $$ + BEGIN + INSERT INTO tx_hash_registry (source_table, tx_hash, occurred_at) + VALUES ('blockchain_events', NEW.tx_hash, NEW.ledger_close_time); + RETURN NEW; + END; + $$ LANGUAGE plpgsql + `); + + pgm.sql(` + CREATE TRIGGER bets_tx_hash_guard + BEFORE INSERT ON bets + FOR EACH ROW EXECUTE FUNCTION tx_hash_registry_guard() + `); + + pgm.sql(` + CREATE TRIGGER blockchain_events_tx_hash_guard + BEFORE INSERT ON blockchain_events + FOR EACH ROW EXECUTE FUNCTION tx_hash_registry_guard_events() + `); + + // Backfill from the rows copied in above, so the guard covers existing data. + pgm.sql(` + INSERT INTO tx_hash_registry (source_table, tx_hash, occurred_at) + SELECT 'bets', tx_hash, placed_at FROM bets + ON CONFLICT DO NOTHING + `); + pgm.sql(` + INSERT INTO tx_hash_registry (source_table, tx_hash, occurred_at) + SELECT 'blockchain_events', tx_hash, ledger_close_time FROM blockchain_events + ON CONFLICT DO NOTHING + `); + + // Registry backing src/db/shardMap.ts. Single row per logical shard; the + // default deployment is one shard, so routing is a no-op until a second + // row exists. + pgm.createTable('shard_map', { + id: { type: 'serial', primaryKey: true }, + shard_key: { type: 'text', notNull: true, unique: true }, + dsn_env_var: { type: 'text', notNull: true }, + slot_start: { type: 'integer', notNull: true }, + slot_end: { type: 'integer', notNull: true }, + status: { type: 'text', notNull: true, default: 'active' }, + last_health_check_at: { type: 'timestamptz' }, + healthy: { type: 'boolean', notNull: true, default: true }, + created_at: { type: 'timestamptz', notNull: true, default: pgm.func('NOW()') }, + }); + + pgm.addConstraint('shard_map', 'shard_map_slot_range_valid', { + check: 'slot_start >= 0 AND slot_end > slot_start AND slot_end <= 4096', + }); + + pgm.sql(` + INSERT INTO shard_map (shard_key, dsn_env_var, slot_start, slot_end) + VALUES ('shard-0', 'DATABASE_URL', 0, 4096) + `); +}; + +// MIGRATION NOTE: down() restores unpartitioned tables. It will fail if the +// data no longer fits the original INTEGER id or if duplicate tx_hash rows +// were somehow written across partitions — both are intentional loud failures +// rather than silent data loss. +exports.down = (pgm) => { + pgm.dropTable('shard_map'); + + // Drop the guard before rebuilding the tables — the triggers reference them. + pgm.sql('DROP TRIGGER IF EXISTS bets_tx_hash_guard ON bets'); + pgm.sql('DROP TRIGGER IF EXISTS blockchain_events_tx_hash_guard ON blockchain_events'); + pgm.sql('DROP FUNCTION IF EXISTS tx_hash_registry_guard()'); + pgm.sql('DROP FUNCTION IF EXISTS tx_hash_registry_guard_events()'); + pgm.dropTable('tx_hash_registry'); + + for (const [parent, tsColumn] of [ + ['bets', 'placed_at'], + ['blockchain_events', 'ledger_close_time'], + ]) { + pgm.sql(`CREATE TABLE ${parent}_flat (LIKE ${parent} INCLUDING DEFAULTS)`); + pgm.sql(`INSERT INTO ${parent}_flat SELECT * FROM ${parent}`); + pgm.sql(`DROP TABLE ${parent} CASCADE`); + pgm.sql(`ALTER TABLE ${parent}_flat RENAME TO ${parent}`); + pgm.sql(`ALTER TABLE ${parent} ADD PRIMARY KEY (id)`); + pgm.sql(`ALTER TABLE ${parent} ADD CONSTRAINT ${parent}_tx_hash_key UNIQUE (tx_hash)`); + pgm.sql(`ALTER TABLE ${parent} ALTER COLUMN ${tsColumn} SET DEFAULT NOW()`); + } + + pgm.sql('ALTER TABLE bets ADD CONSTRAINT bets_market_id_fkey FOREIGN KEY (market_id) REFERENCES markets(market_id)'); +}; diff --git a/backend/src/db/partitionManager.ts b/backend/src/db/partitionManager.ts new file mode 100644 index 00000000..6b68c265 --- /dev/null +++ b/backend/src/db/partitionManager.ts @@ -0,0 +1,151 @@ +// ============================================================ +// BOXMEOUT — Partition Manager +// Rolling monthly range partitions for bets / blockchain_events +// ============================================================ + +import type { Pool } from 'pg'; + +/** Tables converted to declarative range partitions, and the column each is keyed on. */ +export const PARTITIONED_TABLES = { + bets: 'placed_at', + blockchain_events: 'ledger_close_time', +} as const; + +export type PartitionedTable = keyof typeof PARTITIONED_TABLES; + +export interface PartitionRange { + /** Child table name, e.g. `bets_2026_07`. */ + name: string; + /** Inclusive lower bound. */ + from: Date; + /** Exclusive upper bound. */ + to: Date; +} + +/** + * UTC month containing `date`, as a half-open [from, to) range. + * + * Partition bounds are always UTC. Deriving them from local time would shift + * boundaries by the server's offset and leave gaps that route rows to the + * DEFAULT partition. + */ +export function monthRange(table: PartitionedTable, date: Date): PartitionRange { + const from = new Date(Date.UTC(date.getUTCFullYear(), date.getUTCMonth(), 1)); + const to = new Date(Date.UTC(from.getUTCFullYear(), from.getUTCMonth() + 1, 1)); + const suffix = `${from.getUTCFullYear()}_${String(from.getUTCMonth() + 1).padStart(2, '0')}`; + return { name: `${table}_${suffix}`, from, to }; +} + +/** + * Create partitions for the next `monthsAhead` months if they don't exist. + * + * Run this on a schedule well ahead of the boundary. Without a partition for + * the incoming month, rows fall into the DEFAULT partition, which is slow to + * query and cannot later be split without rewriting it. + * + * Returns the partitions that were newly created. + */ +export async function ensureFuturePartitions( + pool: Pool, + table: PartitionedTable, + monthsAhead = 3, +): Promise { + const created: PartitionRange[] = []; + const now = new Date(); + + for (let i = 0; i <= monthsAhead; i += 1) { + const target = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth() + i, 1)); + const range = monthRange(table, target); + + // `CREATE TABLE IF NOT EXISTS` reports nothing useful about whether it + // actually created anything, so existence is checked first to keep the + // return value meaningful to callers that log or alert on it. + const { rows } = await pool.query<{ exists: boolean }>('SELECT to_regclass($1) IS NOT NULL AS exists', [ + range.name, + ]); + if (rows[0]?.exists) continue; + + // CREATE TABLE ... PARTITION OF takes an ACCESS EXCLUSIVE lock on the + // parent, so this is deliberately one short statement per month rather + // than a single long transaction holding the lock across all of them. + await pool.query( + `CREATE TABLE IF NOT EXISTS ${range.name} + PARTITION OF ${table} + FOR VALUES FROM ($1) TO ($2)`, + [range.from.toISOString(), range.to.toISOString()], + ); + created.push(range); + } + + return created; +} + +/** Partitions currently attached to `table`, oldest bound first. */ +export async function listPartitions(pool: Pool, table: PartitionedTable): Promise { + const { rows } = await pool.query<{ child: string }>( + `SELECT c.relname AS child + FROM pg_inherits i + JOIN pg_class c ON c.oid = i.inhrelid + JOIN pg_class p ON p.oid = i.inhparent + WHERE p.relname = $1 + ORDER BY c.relname`, + [table], + ); + return rows.map((r) => r.child); +} + +/** + * Detach and drop partitions whose entire range predates the retention window. + * + * Detach-then-drop rather than a bare DROP: detaching takes a briefer lock and + * leaves the data readable as a standalone table if the drop is interrupted. + * The DEFAULT partition is never dropped. + */ +export async function dropPartitionsOlderThan( + pool: Pool, + table: PartitionedTable, + retentionMonths: number, +): Promise { + const now = new Date(); + const cutoff = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth() - retentionMonths, 1)); + const existing = await listPartitions(pool, table); + const dropped: string[] = []; + + for (const child of existing) { + if (child === `${table}_default`) continue; + + const match = child.match(/_(\d{4})_(\d{2})$/); + if (!match) continue; + + const partitionStart = new Date(Date.UTC(Number(match[1]), Number(match[2]) - 1, 1)); + if (partitionStart >= cutoff) continue; + + await pool.query(`ALTER TABLE ${table} DETACH PARTITION ${child}`); + await pool.query(`DROP TABLE ${child}`); + dropped.push(child); + } + + if (dropped.length > 0) { + // The dedup registry is unpartitioned, so it does not shrink when a + // partition is dropped. Prune it in step or it grows without bound. + // Safe because a tx_hash whose rows have aged out can no longer be + // double-inserted against data that still exists. + await pool.query('DELETE FROM tx_hash_registry WHERE source_table = $1 AND occurred_at < $2', [ + table, + cutoff.toISOString(), + ]); + } + + return dropped; +} + +/** + * Rows sitting in the DEFAULT partition — always a problem worth alerting on. + * + * A non-zero count means writes arrived outside every defined range, i.e. + * `ensureFuturePartitions` stopped running or fell behind. + */ +export async function countDefaultPartitionRows(pool: Pool, table: PartitionedTable): Promise { + const { rows } = await pool.query<{ count: string }>(`SELECT COUNT(*)::text AS count FROM ${table}_default`); + return Number(rows[0]?.count ?? 0); +} diff --git a/backend/src/db/shardMap.ts b/backend/src/db/shardMap.ts new file mode 100644 index 00000000..94f1e8bd --- /dev/null +++ b/backend/src/db/shardMap.ts @@ -0,0 +1,145 @@ +// ============================================================ +// BOXMEOUT — Shard Map +// Slot-based key routing and the shard registry +// ============================================================ + +import { createHash } from 'crypto'; +import type { Pool } from 'pg'; + +/** + * Fixed slot space, Redis-Cluster style. + * + * Keys hash to a slot, and slots — not keys — are assigned to shards. This is + * what makes rebalancing tractable: moving a slot range moves a known set of + * rows without rehashing anything else. The count is fixed forever; changing + * it relocates every key at once. + */ +export const SLOT_COUNT = 4096; + +export type ShardStatus = 'active' | 'draining' | 'offline'; + +export interface ShardDescriptor { + shardKey: string; + /** Env var holding this shard's connection string; the DSN itself is never stored in the DB. */ + dsnEnvVar: string; + slotStart: number; + /** Exclusive. */ + slotEnd: number; + status: ShardStatus; + healthy: boolean; + lastHealthCheckAt: Date | null; +} + +/** + * Which shard owns a given market. + * + * `market_id` is the shard key for every co-located table: a market and its + * bets must live together or the hot read path (market detail + its bets) + * becomes a cross-shard join. Never shard bets by bettor_address for this + * reason. + */ +export function slotForKey(key: string): number { + const digest = createHash('sha1').update(key).digest(); + return digest.readUInt32BE(0) % SLOT_COUNT; +} + +/** Load the registry. Callers should cache this; it changes only on rebalance. */ +export async function loadShardMap(pool: Pool): Promise { + const { rows } = await pool.query<{ + shard_key: string; + dsn_env_var: string; + slot_start: number; + slot_end: number; + status: ShardStatus; + healthy: boolean; + last_health_check_at: Date | null; + }>( + `SELECT shard_key, dsn_env_var, slot_start, slot_end, status, healthy, last_health_check_at + FROM shard_map + ORDER BY slot_start`, + ); + + return rows.map((r) => ({ + shardKey: r.shard_key, + dsnEnvVar: r.dsn_env_var, + slotStart: r.slot_start, + slotEnd: r.slot_end, + status: r.status, + healthy: r.healthy, + lastHealthCheckAt: r.last_health_check_at, + })); +} + +/** + * Assert the map covers [0, SLOT_COUNT) exactly once. + * + * A gap silently drops writes and an overlap sends the same key to two + * shards, so this runs at startup and after any rebalance rather than being + * left to discovery in production. + */ +export function validateShardMap(shards: ShardDescriptor[]): void { + const ordered = [...shards].sort((a, b) => a.slotStart - b.slotStart); + let expected = 0; + + for (const shard of ordered) { + if (shard.slotStart !== expected) { + throw new Error( + `shard map is not contiguous: expected slot ${expected} at '${shard.shardKey}', got ${shard.slotStart}`, + ); + } + expected = shard.slotEnd; + } + + if (expected !== SLOT_COUNT) { + throw new Error(`shard map covers ${expected} of ${SLOT_COUNT} slots`); + } +} + +/** Resolve a key to its owning shard. Throws if the map has a hole at that slot. */ +export function resolveShard(shards: ShardDescriptor[], key: string): ShardDescriptor { + const slot = slotForKey(key); + const shard = shards.find((s) => slot >= s.slotStart && slot < s.slotEnd); + + if (!shard) { + throw new Error(`no shard owns slot ${slot} (key '${key}')`); + } + return shard; +} + +/** + * Slot ranges to move to bring the map toward even ownership. + * + * Returns proposals only — it does not move data. Executing a move means + * copying rows, double-writing during cutover, then flipping ownership, which + * belongs in an operator-driven job rather than an automatic one. + */ +export function planRebalance(shards: ShardDescriptor[]): Array<{ from: string; to: string; slots: number }> { + const active = shards.filter((s) => s.status === 'active'); + if (active.length < 2) return []; + + const target = Math.floor(SLOT_COUNT / active.length); + const overloaded = active + .map((s) => ({ shard: s, excess: s.slotEnd - s.slotStart - target })) + .filter((s) => s.excess > 0) + .sort((a, b) => b.excess - a.excess); + const underloaded = active + .map((s) => ({ shard: s, deficit: target - (s.slotEnd - s.slotStart) })) + .filter((s) => s.deficit > 0) + .sort((a, b) => b.deficit - a.deficit); + + const moves: Array<{ from: string; to: string; slots: number }> = []; + + for (const donor of overloaded) { + for (const recipient of underloaded) { + if (donor.excess === 0) break; + if (recipient.deficit === 0) continue; + + const slots = Math.min(donor.excess, recipient.deficit); + moves.push({ from: donor.shard.shardKey, to: recipient.shard.shardKey, slots }); + donor.excess -= slots; + recipient.deficit -= slots; + } + } + + return moves; +} diff --git a/backend/src/db/shardRouter.ts b/backend/src/db/shardRouter.ts new file mode 100644 index 00000000..84f29f4a --- /dev/null +++ b/backend/src/db/shardRouter.ts @@ -0,0 +1,148 @@ +// ============================================================ +// BOXMEOUT — Shard Router +// Key-directed and scatter-gather query routing +// ============================================================ + +import { Pool } from 'pg'; +import type { QueryResult, QueryResultRow } from 'pg'; +import { loadShardMap, resolveShard, validateShardMap, type ShardDescriptor } from './shardMap'; + +export interface ShardHealth { + shardKey: string; + healthy: boolean; + latencyMs: number | null; + error: string | null; +} + +/** + * Routing surface used by services. + * + * Services depend on this interface, never on a Pool directly. That is the + * whole point of introducing it now while BOXMEOUT still runs on one database: + * adding a second shard later becomes a deployment change instead of a rewrite + * of every query site. + */ +export interface ShardRouter { + /** Run a query on the shard owning `shardKey` (a market_id). */ + query( + shardKey: string, + text: string, + values?: unknown[], + ): Promise>; + + /** + * Run a query on every active shard and concatenate the rows. + * + * Ordering across shards is not meaningful — callers that need a global sort + * or LIMIT must re-sort the merged rows themselves. Avoid on hot paths; + * latency is that of the slowest shard. + */ + queryAll(text: string, values?: unknown[]): Promise; + + healthCheck(): Promise; + + close(): Promise; +} + +/** + * The deployed router: one Pool per registry entry. + * + * With the default single-row shard_map this degenerates to a plain connection + * pool, so it is safe to route everything through it from day one. + */ +export class PoolShardRouter implements ShardRouter { + private readonly pools = new Map(); + + private constructor(private readonly shards: ShardDescriptor[]) {} + + static async create(controlPool: Pool): Promise { + const shards = await loadShardMap(controlPool); + validateShardMap(shards); + + const router = new PoolShardRouter(shards); + for (const shard of shards) { + const dsn = process.env[shard.dsnEnvVar]; + if (!dsn) { + throw new Error(`shard '${shard.shardKey}' expects connection string in ${shard.dsnEnvVar}, which is unset`); + } + router.pools.set(shard.shardKey, new Pool({ connectionString: dsn })); + } + return router; + } + + private poolFor(shard: ShardDescriptor): Pool { + const pool = this.pools.get(shard.shardKey); + if (!pool) { + throw new Error(`no pool initialised for shard '${shard.shardKey}'`); + } + return pool; + } + + async query( + shardKey: string, + text: string, + values: unknown[] = [], + ): Promise> { + const shard = resolveShard(this.shards, shardKey); + + // Draining shards still serve reads and writes — a drain moves slots, it + // does not stop traffic. Only an offline shard is refused. + if (shard.status === 'offline') { + throw new Error(`shard '${shard.shardKey}' is offline`); + } + + return this.poolFor(shard).query(text, values); + } + + async queryAll(text: string, values: unknown[] = []): Promise { + const active = this.shards.filter((s) => s.status !== 'offline'); + + // Fail the whole fan-out if any shard errors. A partial result set is + // worse than an error here: callers aggregate these rows into totals and + // would silently report low numbers. + const results = await Promise.all(active.map((shard) => this.poolFor(shard).query(text, values))); + + return results.flatMap((r) => r.rows); + } + + async healthCheck(): Promise { + return Promise.all( + this.shards.map(async (shard) => { + const startedAt = Date.now(); + try { + await this.poolFor(shard).query('SELECT 1'); + return { + shardKey: shard.shardKey, + healthy: true, + latencyMs: Date.now() - startedAt, + error: null, + }; + } catch (err) { + return { + shardKey: shard.shardKey, + healthy: false, + latencyMs: null, + error: err instanceof Error ? err.message : String(err), + }; + } + }), + ); + } + + async close(): Promise { + await Promise.all([...this.pools.values()].map((pool) => pool.end())); + this.pools.clear(); + } +} + +/** Persist the outcome of a health sweep so operators and alerting can read it. */ +export async function recordHealthResults(controlPool: Pool, results: ShardHealth[]): Promise { + for (const result of results) { + await controlPool.query( + `UPDATE shard_map + SET healthy = $2, last_health_check_at = NOW() + WHERE shard_key = $1`, + [result.shardKey, result.healthy], + ); + } +} diff --git a/backend/tests/db/partition.test.ts b/backend/tests/db/partition.test.ts new file mode 100644 index 00000000..9eb5cf03 --- /dev/null +++ b/backend/tests/db/partition.test.ts @@ -0,0 +1,276 @@ +// ============================================================ +// BOXMEOUT — Partition & Shard Strategy Unit Tests +// ============================================================ + +import type { Pool } from 'pg'; +import { + countDefaultPartitionRows, + dropPartitionsOlderThan, + ensureFuturePartitions, + listPartitions, + monthRange, +} from '../../src/db/partitionManager'; +import { planRebalance, resolveShard, slotForKey, SLOT_COUNT, validateShardMap } from '../../src/db/shardMap'; +import type { ShardDescriptor } from '../../src/db/shardMap'; + +const mockPool = () => ({ query: jest.fn() }) as unknown as Pool & { query: jest.Mock }; + +const shard = (overrides: Partial & Pick) => + ({ + dsnEnvVar: 'DATABASE_URL', + status: 'active', + healthy: true, + lastHealthCheckAt: null, + ...overrides, + }) as ShardDescriptor; + +describe('partitionManager', () => { + describe('monthRange', () => { + it('derives a half-open UTC month from any day within it', () => { + const range = monthRange('bets', new Date('2026-07-20T13:45:00Z')); + + expect(range.name).toBe('bets_2026_07'); + expect(range.from.toISOString()).toBe('2026-07-01T00:00:00.000Z'); + expect(range.to.toISOString()).toBe('2026-08-01T00:00:00.000Z'); + }); + + it('rolls the upper bound into the next year in December', () => { + const range = monthRange('blockchain_events', new Date('2026-12-31T23:59:59Z')); + + expect(range.name).toBe('blockchain_events_2026_12'); + expect(range.to.toISOString()).toBe('2027-01-01T00:00:00.000Z'); + }); + + it('pads single-digit months so names sort chronologically', () => { + expect(monthRange('bets', new Date('2026-03-05T00:00:00Z')).name).toBe('bets_2026_03'); + }); + }); + + describe('ensureFuturePartitions', () => { + it('creates only the months that do not already exist', async () => { + const pool = mockPool(); + // First month exists, the rest do not. + pool.query + .mockResolvedValueOnce({ rows: [{ exists: true }] }) + .mockResolvedValueOnce({ rows: [{ exists: false }] }) + .mockResolvedValueOnce({ rows: [] }) + .mockResolvedValueOnce({ rows: [{ exists: false }] }) + .mockResolvedValueOnce({ rows: [] }); + + const created = await ensureFuturePartitions(pool, 'bets', 2); + + expect(created).toHaveLength(2); + const createStatements = pool.query.mock.calls.filter((c: unknown[]) => + String(c[0]).includes('PARTITION OF bets'), + ); + expect(createStatements).toHaveLength(2); + }); + + it('passes ISO bounds as parameters rather than interpolating them', async () => { + const pool = mockPool(); + pool.query.mockResolvedValueOnce({ rows: [{ exists: false }] }).mockResolvedValueOnce({ rows: [] }); + + await ensureFuturePartitions(pool, 'bets', 0); + + const [, values] = pool.query.mock.calls[1]; + expect(values).toHaveLength(2); + expect(values[0]).toMatch(/^\d{4}-\d{2}-01T00:00:00\.000Z$/); + }); + }); + + describe('dropPartitionsOlderThan', () => { + it('drops partitions past the window and leaves recent ones attached', async () => { + const pool = mockPool(); + const now = new Date(); + const old = monthRange('bets', new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth() - 12, 1))); + const recent = monthRange('bets', now); + + pool.query.mockImplementation((text: string) => { + if (text.includes('pg_inherits')) { + return Promise.resolve({ rows: [{ child: old.name }, { child: recent.name }] }); + } + return Promise.resolve({ rows: [] }); + }); + + const dropped = await dropPartitionsOlderThan(pool, 'bets', 6); + + expect(dropped).toEqual([old.name]); + }); + + it('never drops the default partition', async () => { + const pool = mockPool(); + pool.query.mockImplementation((text: string) => { + if (text.includes('pg_inherits')) { + return Promise.resolve({ rows: [{ child: 'bets_default' }] }); + } + return Promise.resolve({ rows: [] }); + }); + + const dropped = await dropPartitionsOlderThan(pool, 'bets', 1); + + expect(dropped).toEqual([]); + expect(pool.query.mock.calls.some((c: unknown[]) => String(c[0]).includes('DROP TABLE'))).toBe(false); + }); + + it('detaches before dropping so an interrupted drop leaves data readable', async () => { + const pool = mockPool(); + const old = monthRange('bets', new Date(Date.UTC(2020, 0, 1))); + + pool.query.mockImplementation((text: string) => { + if (text.includes('pg_inherits')) return Promise.resolve({ rows: [{ child: old.name }] }); + return Promise.resolve({ rows: [] }); + }); + + await dropPartitionsOlderThan(pool, 'bets', 6); + + const statements = pool.query.mock.calls.map((c: unknown[]) => String(c[0])); + const detachAt = statements.findIndex((s) => s.includes('DETACH PARTITION')); + const dropAt = statements.findIndex((s) => s.includes('DROP TABLE')); + expect(detachAt).toBeGreaterThan(-1); + expect(dropAt).toBeGreaterThan(detachAt); + }); + }); + + describe('listPartitions / countDefaultPartitionRows', () => { + it('returns child table names', async () => { + const pool = mockPool(); + pool.query.mockResolvedValue({ rows: [{ child: 'bets_2026_07' }, { child: 'bets_default' }] }); + + expect(await listPartitions(pool, 'bets')).toEqual(['bets_2026_07', 'bets_default']); + }); + + it('reports default-partition rows as a number', async () => { + const pool = mockPool(); + pool.query.mockResolvedValue({ rows: [{ count: '42' }] }); + + expect(await countDefaultPartitionRows(pool, 'bets')).toBe(42); + }); + + it('reports zero when the default partition is empty', async () => { + const pool = mockPool(); + pool.query.mockResolvedValue({ rows: [{ count: '0' }] }); + + expect(await countDefaultPartitionRows(pool, 'blockchain_events')).toBe(0); + }); + }); +}); + +describe('shardMap', () => { + describe('slotForKey', () => { + it('is deterministic', () => { + expect(slotForKey('market-abc')).toBe(slotForKey('market-abc')); + }); + + it('always lands inside the slot space', () => { + for (let i = 0; i < 500; i += 1) { + const slot = slotForKey(`market-${i}`); + expect(slot).toBeGreaterThanOrEqual(0); + expect(slot).toBeLessThan(SLOT_COUNT); + } + }); + + it('spreads keys across the space rather than clustering', () => { + const slots = new Set(Array.from({ length: 200 }, (_, i) => slotForKey(`market-${i}`))); + expect(slots.size).toBeGreaterThan(150); + }); + }); + + describe('validateShardMap', () => { + it('accepts full contiguous coverage', () => { + expect(() => + validateShardMap([ + shard({ shardKey: 'a', slotStart: 0, slotEnd: 2048 }), + shard({ shardKey: 'b', slotStart: 2048, slotEnd: SLOT_COUNT }), + ]), + ).not.toThrow(); + }); + + it('accepts the single-shard default', () => { + expect(() => validateShardMap([shard({ shardKey: 'shard-0', slotStart: 0, slotEnd: SLOT_COUNT })])).not.toThrow(); + }); + + it('rejects a gap that would silently drop writes', () => { + expect(() => + validateShardMap([ + shard({ shardKey: 'a', slotStart: 0, slotEnd: 1000 }), + shard({ shardKey: 'b', slotStart: 2000, slotEnd: SLOT_COUNT }), + ]), + ).toThrow(/not contiguous/); + }); + + it('rejects an overlap that would route one key to two shards', () => { + expect(() => + validateShardMap([ + shard({ shardKey: 'a', slotStart: 0, slotEnd: 3000 }), + shard({ shardKey: 'b', slotStart: 2000, slotEnd: SLOT_COUNT }), + ]), + ).toThrow(/not contiguous/); + }); + + it('rejects a map that stops short of the full space', () => { + expect(() => validateShardMap([shard({ shardKey: 'a', slotStart: 0, slotEnd: 100 })])).toThrow(/covers 100 of/); + }); + }); + + describe('resolveShard', () => { + it('routes a key to the shard owning its slot', () => { + const shards = [ + shard({ shardKey: 'a', slotStart: 0, slotEnd: 2048 }), + shard({ shardKey: 'b', slotStart: 2048, slotEnd: SLOT_COUNT }), + ]; + const key = 'market-abc'; + const expected = slotForKey(key) < 2048 ? 'a' : 'b'; + + expect(resolveShard(shards, key).shardKey).toBe(expected); + }); + + it('sends every key to the sole shard in a single-shard deployment', () => { + const shards = [shard({ shardKey: 'shard-0', slotStart: 0, slotEnd: SLOT_COUNT })]; + + for (let i = 0; i < 50; i += 1) { + expect(resolveShard(shards, `market-${i}`).shardKey).toBe('shard-0'); + } + }); + + it('throws rather than guessing when the map has a hole', () => { + const shards = [shard({ shardKey: 'a', slotStart: 0, slotEnd: 1 })]; + // slotForKey is deterministic, so pick a key that misses slot 0. + const key = Array.from({ length: 100 }, (_, i) => `market-${i}`).find((k) => slotForKey(k) !== 0); + + expect(() => resolveShard(shards, key as string)).toThrow(/no shard owns slot/); + }); + }); + + describe('planRebalance', () => { + it('proposes nothing for a single shard', () => { + expect(planRebalance([shard({ shardKey: 'shard-0', slotStart: 0, slotEnd: SLOT_COUNT })])).toEqual([]); + }); + + it('proposes nothing when ownership is already even', () => { + expect( + planRebalance([ + shard({ shardKey: 'a', slotStart: 0, slotEnd: 2048 }), + shard({ shardKey: 'b', slotStart: 2048, slotEnd: SLOT_COUNT }), + ]), + ).toEqual([]); + }); + + it('moves slots from the overloaded shard to the new empty one', () => { + const moves = planRebalance([ + shard({ shardKey: 'a', slotStart: 0, slotEnd: SLOT_COUNT }), + shard({ shardKey: 'b', slotStart: SLOT_COUNT, slotEnd: SLOT_COUNT }), + ]); + + expect(moves).toEqual([{ from: 'a', to: 'b', slots: 2048 }]); + }); + + it('ignores shards that are not active', () => { + const moves = planRebalance([ + shard({ shardKey: 'a', slotStart: 0, slotEnd: SLOT_COUNT }), + shard({ shardKey: 'b', slotStart: SLOT_COUNT, slotEnd: SLOT_COUNT, status: 'offline' }), + ]); + + expect(moves).toEqual([]); + }); + }); +});