From d053d0651133843e2168afe99697484af8c4ab0e Mon Sep 17 00:00:00 2001 From: Dan Anderson Date: Sun, 4 Oct 2026 12:51:14 -0400 Subject: [PATCH] Verify replay continuity and journal exact mutable state for reorgs --- apps/api/migrations-core/0100_reorg_undo.sql | 216 +++++++++++++++++ apps/api/src/indexer/events/context.ts | 2 +- apps/api/src/indexer/events/dispatch.ts | 2 + apps/api/src/indexer/reorg-undo.ts | 126 ++++++++++ apps/api/src/indexer/sync.ts | 63 ++++- apps/api/src/integrations/counterparty.ts | 5 +- apps/api/src/integrations/fresh-read.ts | 11 + apps/api/tests/core-foundation-writes.test.ts | 222 ++++++++++++++++-- apps/api/tests/node-test.d.ts | 1 + apps/api/tests/replay-verbose-parity.test.ts | 5 +- docs/reorg-recovery.md | 30 +++ 11 files changed, 653 insertions(+), 30 deletions(-) create mode 100644 apps/api/migrations-core/0100_reorg_undo.sql create mode 100644 apps/api/src/indexer/reorg-undo.ts create mode 100644 apps/api/src/integrations/fresh-read.ts create mode 100644 docs/reorg-recovery.md diff --git a/apps/api/migrations-core/0100_reorg_undo.sql b/apps/api/migrations-core/0100_reorg_undo.sql new file mode 100644 index 00000000..dfa7870d --- /dev/null +++ b/apps/api/migrations-core/0100_reorg_undo.sql @@ -0,0 +1,216 @@ +-- Bounded before-images for mutable chain state. The context row exists only +-- inside the same D1 transaction as the indexed writes, never between batches. +CREATE TABLE reorg_undo_context (singleton INTEGER PRIMARY KEY CHECK(singleton=1), block_index INTEGER NOT NULL); +CREATE TABLE reorg_undo_state (singleton INTEGER PRIMARY KEY CHECK(singleton=1), floor INTEGER NOT NULL); +CREATE TABLE reorg_undo (seq INTEGER PRIMARY KEY AUTOINCREMENT, block_index INTEGER NOT NULL, table_name TEXT NOT NULL, row_key TEXT NOT NULL, before_row TEXT); +CREATE INDEX idx_reorg_undo_block ON reorg_undo(block_index,seq); + +CREATE TRIGGER reorg_undo_assets_insert AFTER INSERT ON assets +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'assets',json_object('asset_id',NEW."asset_id"),NULL FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_assets_update AFTER UPDATE ON assets +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) AND (OLD."asset_id" IS NOT NEW."asset_id" OR OLD."asset_longname" IS NOT NEW."asset_longname" OR OLD."numeric_asset_id" IS NOT NEW."numeric_asset_id" OR OLD."type" IS NOT NEW."type" OR OLD."issuer_id" IS NOT NEW."issuer_id" OR OLD."owner_id" IS NOT NEW."owner_id" OR OLD."divisible" IS NOT NEW."divisible" OR OLD."locked" IS NOT NEW."locked" OR OLD."description_locked" IS NOT NEW."description_locked" OR OLD."supply" IS NOT NEW."supply" OR OLD."supply_normalized" IS NOT NEW."supply_normalized" OR OLD."description" IS NOT NEW."description" OR OLD."mime_type" IS NOT NEW."mime_type" OR OLD."first_issuance_block_index" IS NOT NEW."first_issuance_block_index" OR OLD."last_issuance_block_index" IS NOT NEW."last_issuance_block_index" OR OLD."first_issuance_block_time" IS NOT NEW."first_issuance_block_time" OR OLD."last_issuance_block_time" IS NOT NEW."last_issuance_block_time" OR OLD."updated_at" IS NOT NEW."updated_at") +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'assets',json_object('asset_id',OLD."asset_id"),json_object('asset_id',OLD."asset_id",'asset_longname',OLD."asset_longname",'numeric_asset_id',OLD."numeric_asset_id",'type',OLD."type",'issuer_id',OLD."issuer_id",'owner_id',OLD."owner_id",'divisible',OLD."divisible",'locked',OLD."locked",'description_locked',OLD."description_locked",'supply',OLD."supply",'supply_normalized',OLD."supply_normalized",'description',OLD."description",'mime_type',OLD."mime_type",'first_issuance_block_index',OLD."first_issuance_block_index",'last_issuance_block_index',OLD."last_issuance_block_index",'first_issuance_block_time',OLD."first_issuance_block_time",'last_issuance_block_time',OLD."last_issuance_block_time",'updated_at',OLD."updated_at") FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_assets_delete AFTER DELETE ON assets +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'assets',json_object('asset_id',OLD."asset_id"),json_object('asset_id',OLD."asset_id",'asset_longname',OLD."asset_longname",'numeric_asset_id',OLD."numeric_asset_id",'type',OLD."type",'issuer_id',OLD."issuer_id",'owner_id',OLD."owner_id",'divisible',OLD."divisible",'locked',OLD."locked",'description_locked',OLD."description_locked",'supply',OLD."supply",'supply_normalized',OLD."supply_normalized",'description',OLD."description",'mime_type',OLD."mime_type",'first_issuance_block_index',OLD."first_issuance_block_index",'last_issuance_block_index',OLD."last_issuance_block_index",'first_issuance_block_time',OLD."first_issuance_block_time",'last_issuance_block_time',OLD."last_issuance_block_time",'updated_at',OLD."updated_at") FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_orders_insert AFTER INSERT ON orders +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'orders',json_object('tx_index',NEW."tx_index"),NULL FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_orders_update AFTER UPDATE ON orders +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) AND (OLD."tx_index" IS NOT NEW."tx_index" OR OLD."tx_hash" IS NOT NEW."tx_hash" OR OLD."block_index" IS NOT NEW."block_index" OR OLD."block_time" IS NOT NEW."block_time" OR OLD."source_id" IS NOT NEW."source_id" OR OLD."give_asset_id" IS NOT NEW."give_asset_id" OR OLD."give_quantity" IS NOT NEW."give_quantity" OR OLD."give_remaining" IS NOT NEW."give_remaining" OR OLD."get_asset_id" IS NOT NEW."get_asset_id" OR OLD."get_quantity" IS NOT NEW."get_quantity" OR OLD."get_remaining" IS NOT NEW."get_remaining" OR OLD."expiration" IS NOT NEW."expiration" OR OLD."expire_index" IS NOT NEW."expire_index" OR OLD."fee_required" IS NOT NEW."fee_required" OR OLD."fee_required_remaining" IS NOT NEW."fee_required_remaining" OR OLD."fee_provided" IS NOT NEW."fee_provided" OR OLD."fee_provided_remaining" IS NOT NEW."fee_provided_remaining" OR OLD."status" IS NOT NEW."status" OR OLD."closed_block_index" IS NOT NEW."closed_block_index") +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'orders',json_object('tx_index',OLD."tx_index"),json_object('tx_index',OLD."tx_index",'tx_hash',CASE WHEN OLD."tx_hash" IS NULL THEN NULL ELSE hex(OLD."tx_hash") END,'block_index',OLD."block_index",'block_time',OLD."block_time",'source_id',OLD."source_id",'give_asset_id',OLD."give_asset_id",'give_quantity',OLD."give_quantity",'give_remaining',OLD."give_remaining",'get_asset_id',OLD."get_asset_id",'get_quantity',OLD."get_quantity",'get_remaining',OLD."get_remaining",'expiration',OLD."expiration",'expire_index',OLD."expire_index",'fee_required',OLD."fee_required",'fee_required_remaining',OLD."fee_required_remaining",'fee_provided',OLD."fee_provided",'fee_provided_remaining',OLD."fee_provided_remaining",'status',OLD."status",'closed_block_index',OLD."closed_block_index") FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_orders_delete AFTER DELETE ON orders +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'orders',json_object('tx_index',OLD."tx_index"),json_object('tx_index',OLD."tx_index",'tx_hash',CASE WHEN OLD."tx_hash" IS NULL THEN NULL ELSE hex(OLD."tx_hash") END,'block_index',OLD."block_index",'block_time',OLD."block_time",'source_id',OLD."source_id",'give_asset_id',OLD."give_asset_id",'give_quantity',OLD."give_quantity",'give_remaining',OLD."give_remaining",'get_asset_id',OLD."get_asset_id",'get_quantity',OLD."get_quantity",'get_remaining',OLD."get_remaining",'expiration',OLD."expiration",'expire_index',OLD."expire_index",'fee_required',OLD."fee_required",'fee_required_remaining',OLD."fee_required_remaining",'fee_provided',OLD."fee_provided",'fee_provided_remaining',OLD."fee_provided_remaining",'status',OLD."status",'closed_block_index',OLD."closed_block_index") FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_order_matches_insert AFTER INSERT ON order_matches +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'order_matches',json_object('tx0_index',NEW."tx0_index",'tx1_index',NEW."tx1_index"),NULL FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_order_matches_update AFTER UPDATE ON order_matches +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) AND (OLD."tx0_index" IS NOT NEW."tx0_index" OR OLD."tx1_index" IS NOT NEW."tx1_index" OR OLD."tx0_hash" IS NOT NEW."tx0_hash" OR OLD."tx1_hash" IS NOT NEW."tx1_hash" OR OLD."tx0_address_id" IS NOT NEW."tx0_address_id" OR OLD."tx1_address_id" IS NOT NEW."tx1_address_id" OR OLD."forward_asset_id" IS NOT NEW."forward_asset_id" OR OLD."forward_quantity" IS NOT NEW."forward_quantity" OR OLD."backward_asset_id" IS NOT NEW."backward_asset_id" OR OLD."backward_quantity" IS NOT NEW."backward_quantity" OR OLD."block_index" IS NOT NEW."block_index" OR OLD."block_time" IS NOT NEW."block_time" OR OLD."status" IS NOT NEW."status" OR OLD."match_expire_index" IS NOT NEW."match_expire_index" OR OLD."fee_paid" IS NOT NEW."fee_paid" OR OLD."tx0_block_index" IS NOT NEW."tx0_block_index" OR OLD."tx1_block_index" IS NOT NEW."tx1_block_index" OR OLD."tx0_expiration" IS NOT NEW."tx0_expiration" OR OLD."tx1_expiration" IS NOT NEW."tx1_expiration") +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'order_matches',json_object('tx0_index',OLD."tx0_index",'tx1_index',OLD."tx1_index"),json_object('tx0_index',OLD."tx0_index",'tx1_index',OLD."tx1_index",'tx0_hash',CASE WHEN OLD."tx0_hash" IS NULL THEN NULL ELSE hex(OLD."tx0_hash") END,'tx1_hash',CASE WHEN OLD."tx1_hash" IS NULL THEN NULL ELSE hex(OLD."tx1_hash") END,'tx0_address_id',OLD."tx0_address_id",'tx1_address_id',OLD."tx1_address_id",'forward_asset_id',OLD."forward_asset_id",'forward_quantity',OLD."forward_quantity",'backward_asset_id',OLD."backward_asset_id",'backward_quantity',OLD."backward_quantity",'block_index',OLD."block_index",'block_time',OLD."block_time",'status',OLD."status",'match_expire_index',OLD."match_expire_index",'fee_paid',OLD."fee_paid",'tx0_block_index',OLD."tx0_block_index",'tx1_block_index',OLD."tx1_block_index",'tx0_expiration',OLD."tx0_expiration",'tx1_expiration',OLD."tx1_expiration") FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_order_matches_delete AFTER DELETE ON order_matches +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'order_matches',json_object('tx0_index',OLD."tx0_index",'tx1_index',OLD."tx1_index"),json_object('tx0_index',OLD."tx0_index",'tx1_index',OLD."tx1_index",'tx0_hash',CASE WHEN OLD."tx0_hash" IS NULL THEN NULL ELSE hex(OLD."tx0_hash") END,'tx1_hash',CASE WHEN OLD."tx1_hash" IS NULL THEN NULL ELSE hex(OLD."tx1_hash") END,'tx0_address_id',OLD."tx0_address_id",'tx1_address_id',OLD."tx1_address_id",'forward_asset_id',OLD."forward_asset_id",'forward_quantity',OLD."forward_quantity",'backward_asset_id',OLD."backward_asset_id",'backward_quantity',OLD."backward_quantity",'block_index',OLD."block_index",'block_time',OLD."block_time",'status',OLD."status",'match_expire_index',OLD."match_expire_index",'fee_paid',OLD."fee_paid",'tx0_block_index',OLD."tx0_block_index",'tx1_block_index',OLD."tx1_block_index",'tx0_expiration',OLD."tx0_expiration",'tx1_expiration',OLD."tx1_expiration") FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_dispensers_insert AFTER INSERT ON dispensers +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'dispensers',json_object('tx_index',NEW."tx_index"),NULL FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_dispensers_update AFTER UPDATE ON dispensers +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) AND (OLD."tx_index" IS NOT NEW."tx_index" OR OLD."tx_hash" IS NOT NEW."tx_hash" OR OLD."block_index" IS NOT NEW."block_index" OR OLD."block_time" IS NOT NEW."block_time" OR OLD."source_id" IS NOT NEW."source_id" OR OLD."asset_id" IS NOT NEW."asset_id" OR OLD."give_quantity" IS NOT NEW."give_quantity" OR OLD."give_quantity_normalized" IS NOT NEW."give_quantity_normalized" OR OLD."escrow_quantity" IS NOT NEW."escrow_quantity" OR OLD."give_remaining" IS NOT NEW."give_remaining" OR OLD."give_remaining_normalized" IS NOT NEW."give_remaining_normalized" OR OLD."satoshirate" IS NOT NEW."satoshirate" OR OLD."satoshirate_normalized" IS NOT NEW."satoshirate_normalized" OR OLD."status" IS NOT NEW."status" OR OLD."oracle_address_id" IS NOT NEW."oracle_address_id" OR OLD."dispense_count" IS NOT NEW."dispense_count" OR OLD."closed_block_index" IS NOT NEW."closed_block_index" OR OLD."origin_id" IS NOT NEW."origin_id" OR OLD."last_status_tx_hash" IS NOT NEW."last_status_tx_hash") +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'dispensers',json_object('tx_index',OLD."tx_index"),json_object('tx_index',OLD."tx_index",'tx_hash',CASE WHEN OLD."tx_hash" IS NULL THEN NULL ELSE hex(OLD."tx_hash") END,'block_index',OLD."block_index",'block_time',OLD."block_time",'source_id',OLD."source_id",'asset_id',OLD."asset_id",'give_quantity',OLD."give_quantity",'give_quantity_normalized',OLD."give_quantity_normalized",'escrow_quantity',OLD."escrow_quantity",'give_remaining',OLD."give_remaining",'give_remaining_normalized',OLD."give_remaining_normalized",'satoshirate',OLD."satoshirate",'satoshirate_normalized',OLD."satoshirate_normalized",'status',OLD."status",'oracle_address_id',OLD."oracle_address_id",'dispense_count',OLD."dispense_count",'closed_block_index',OLD."closed_block_index",'origin_id',OLD."origin_id",'last_status_tx_hash',CASE WHEN OLD."last_status_tx_hash" IS NULL THEN NULL ELSE hex(OLD."last_status_tx_hash") END) FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_dispensers_delete AFTER DELETE ON dispensers +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'dispensers',json_object('tx_index',OLD."tx_index"),json_object('tx_index',OLD."tx_index",'tx_hash',CASE WHEN OLD."tx_hash" IS NULL THEN NULL ELSE hex(OLD."tx_hash") END,'block_index',OLD."block_index",'block_time',OLD."block_time",'source_id',OLD."source_id",'asset_id',OLD."asset_id",'give_quantity',OLD."give_quantity",'give_quantity_normalized',OLD."give_quantity_normalized",'escrow_quantity',OLD."escrow_quantity",'give_remaining',OLD."give_remaining",'give_remaining_normalized',OLD."give_remaining_normalized",'satoshirate',OLD."satoshirate",'satoshirate_normalized',OLD."satoshirate_normalized",'status',OLD."status",'oracle_address_id',OLD."oracle_address_id",'dispense_count',OLD."dispense_count",'closed_block_index',OLD."closed_block_index",'origin_id',OLD."origin_id",'last_status_tx_hash',CASE WHEN OLD."last_status_tx_hash" IS NULL THEN NULL ELSE hex(OLD."last_status_tx_hash") END) FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_fairminters_insert AFTER INSERT ON fairminters +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'fairminters',json_object('tx_index',NEW."tx_index"),NULL FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_fairminters_update AFTER UPDATE ON fairminters +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) AND (OLD."tx_index" IS NOT NEW."tx_index" OR OLD."tx_hash" IS NOT NEW."tx_hash" OR OLD."block_index" IS NOT NEW."block_index" OR OLD."block_time" IS NOT NEW."block_time" OR OLD."source_id" IS NOT NEW."source_id" OR OLD."asset_id" IS NOT NEW."asset_id" OR OLD."asset_parent_id" IS NOT NEW."asset_parent_id" OR OLD."asset_longname" IS NOT NEW."asset_longname" OR OLD."description" IS NOT NEW."description" OR OLD."price" IS NOT NEW."price" OR OLD."quantity_by_price" IS NOT NEW."quantity_by_price" OR OLD."hard_cap" IS NOT NEW."hard_cap" OR OLD."burn_payment" IS NOT NEW."burn_payment" OR OLD."max_mint_per_tx" IS NOT NEW."max_mint_per_tx" OR OLD."premint_quantity" IS NOT NEW."premint_quantity" OR OLD."start_block" IS NOT NEW."start_block" OR OLD."end_block" IS NOT NEW."end_block" OR OLD."minted_asset_commission_int" IS NOT NEW."minted_asset_commission_int" OR OLD."soft_cap" IS NOT NEW."soft_cap" OR OLD."soft_cap_deadline_block" IS NOT NEW."soft_cap_deadline_block" OR OLD."lock_description" IS NOT NEW."lock_description" OR OLD."lock_quantity" IS NOT NEW."lock_quantity" OR OLD."divisible" IS NOT NEW."divisible" OR OLD."pre_minted" IS NOT NEW."pre_minted" OR OLD."status" IS NOT NEW."status" OR OLD."max_mint_per_address" IS NOT NEW."max_mint_per_address" OR OLD."mime_type" IS NOT NEW."mime_type" OR OLD."earned_quantity" IS NOT NEW."earned_quantity" OR OLD."paid_quantity" IS NOT NEW."paid_quantity" OR OLD."pool_quantity" IS NOT NEW."pool_quantity" OR OLD."lp_asset" IS NOT NEW."lp_asset") +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'fairminters',json_object('tx_index',OLD."tx_index"),json_object('tx_index',OLD."tx_index",'tx_hash',CASE WHEN OLD."tx_hash" IS NULL THEN NULL ELSE hex(OLD."tx_hash") END,'block_index',OLD."block_index",'block_time',OLD."block_time",'source_id',OLD."source_id",'asset_id',OLD."asset_id",'asset_parent_id',OLD."asset_parent_id",'asset_longname',OLD."asset_longname",'description',OLD."description",'price',OLD."price",'quantity_by_price',OLD."quantity_by_price",'hard_cap',OLD."hard_cap",'burn_payment',OLD."burn_payment",'max_mint_per_tx',OLD."max_mint_per_tx",'premint_quantity',OLD."premint_quantity",'start_block',OLD."start_block",'end_block',OLD."end_block",'minted_asset_commission_int',OLD."minted_asset_commission_int",'soft_cap',OLD."soft_cap",'soft_cap_deadline_block',OLD."soft_cap_deadline_block",'lock_description',OLD."lock_description",'lock_quantity',OLD."lock_quantity",'divisible',OLD."divisible",'pre_minted',OLD."pre_minted",'status',OLD."status",'max_mint_per_address',OLD."max_mint_per_address",'mime_type',OLD."mime_type",'earned_quantity',OLD."earned_quantity",'paid_quantity',OLD."paid_quantity",'pool_quantity',OLD."pool_quantity",'lp_asset',OLD."lp_asset") FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_fairminters_delete AFTER DELETE ON fairminters +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'fairminters',json_object('tx_index',OLD."tx_index"),json_object('tx_index',OLD."tx_index",'tx_hash',CASE WHEN OLD."tx_hash" IS NULL THEN NULL ELSE hex(OLD."tx_hash") END,'block_index',OLD."block_index",'block_time',OLD."block_time",'source_id',OLD."source_id",'asset_id',OLD."asset_id",'asset_parent_id',OLD."asset_parent_id",'asset_longname',OLD."asset_longname",'description',OLD."description",'price',OLD."price",'quantity_by_price',OLD."quantity_by_price",'hard_cap',OLD."hard_cap",'burn_payment',OLD."burn_payment",'max_mint_per_tx',OLD."max_mint_per_tx",'premint_quantity',OLD."premint_quantity",'start_block',OLD."start_block",'end_block',OLD."end_block",'minted_asset_commission_int',OLD."minted_asset_commission_int",'soft_cap',OLD."soft_cap",'soft_cap_deadline_block',OLD."soft_cap_deadline_block",'lock_description',OLD."lock_description",'lock_quantity',OLD."lock_quantity",'divisible',OLD."divisible",'pre_minted',OLD."pre_minted",'status',OLD."status",'max_mint_per_address',OLD."max_mint_per_address",'mime_type',OLD."mime_type",'earned_quantity',OLD."earned_quantity",'paid_quantity',OLD."paid_quantity",'pool_quantity',OLD."pool_quantity",'lp_asset',OLD."lp_asset") FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_pools_insert AFTER INSERT ON pools +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'pools',json_object('asset_a_id',NEW."asset_a_id",'asset_b_id',NEW."asset_b_id"),NULL FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_pools_update AFTER UPDATE ON pools +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) AND (OLD."asset_a_id" IS NOT NEW."asset_a_id" OR OLD."asset_b_id" IS NOT NEW."asset_b_id" OR OLD."lp_asset" IS NOT NEW."lp_asset" OR OLD."pair" IS NOT NEW."pair" OR OLD."reserve_a" IS NOT NEW."reserve_a" OR OLD."reserve_b" IS NOT NEW."reserve_b" OR OLD."lp_supply" IS NOT NEW."lp_supply" OR OLD."price" IS NOT NEW."price" OR OLD."status" IS NOT NEW."status" OR OLD."block_index" IS NOT NEW."block_index" OR OLD."updated_block_index" IS NOT NEW."updated_block_index") +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'pools',json_object('asset_a_id',OLD."asset_a_id",'asset_b_id',OLD."asset_b_id"),json_object('asset_a_id',OLD."asset_a_id",'asset_b_id',OLD."asset_b_id",'lp_asset',OLD."lp_asset",'pair',OLD."pair",'reserve_a',OLD."reserve_a",'reserve_b',OLD."reserve_b",'lp_supply',OLD."lp_supply",'price',OLD."price",'status',OLD."status",'block_index',OLD."block_index",'updated_block_index',OLD."updated_block_index") FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_pools_delete AFTER DELETE ON pools +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'pools',json_object('asset_a_id',OLD."asset_a_id",'asset_b_id',OLD."asset_b_id"),json_object('asset_a_id',OLD."asset_a_id",'asset_b_id',OLD."asset_b_id",'lp_asset',OLD."lp_asset",'pair',OLD."pair",'reserve_a',OLD."reserve_a",'reserve_b',OLD."reserve_b",'lp_supply',OLD."lp_supply",'price',OLD."price",'status',OLD."status",'block_index',OLD."block_index",'updated_block_index',OLD."updated_block_index") FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_bets_insert AFTER INSERT ON bets +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'bets',json_object('tx_index',NEW."tx_index"),NULL FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_bets_update AFTER UPDATE ON bets +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) AND (OLD."tx_index" IS NOT NEW."tx_index" OR OLD."tx_hash" IS NOT NEW."tx_hash" OR OLD."block_index" IS NOT NEW."block_index" OR OLD."block_time" IS NOT NEW."block_time" OR OLD."source_id" IS NOT NEW."source_id" OR OLD."feed_address_id" IS NOT NEW."feed_address_id" OR OLD."bet_type" IS NOT NEW."bet_type" OR OLD."deadline" IS NOT NEW."deadline" OR OLD."wager_quantity" IS NOT NEW."wager_quantity" OR OLD."wager_remaining" IS NOT NEW."wager_remaining" OR OLD."counterwager_quantity" IS NOT NEW."counterwager_quantity" OR OLD."counterwager_remaining" IS NOT NEW."counterwager_remaining" OR OLD."target_value" IS NOT NEW."target_value" OR OLD."leverage" IS NOT NEW."leverage" OR OLD."expiration" IS NOT NEW."expiration" OR OLD."expire_index" IS NOT NEW."expire_index" OR OLD."fee_fraction_int" IS NOT NEW."fee_fraction_int" OR OLD."status" IS NOT NEW."status") +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'bets',json_object('tx_index',OLD."tx_index"),json_object('tx_index',OLD."tx_index",'tx_hash',CASE WHEN OLD."tx_hash" IS NULL THEN NULL ELSE hex(OLD."tx_hash") END,'block_index',OLD."block_index",'block_time',OLD."block_time",'source_id',OLD."source_id",'feed_address_id',OLD."feed_address_id",'bet_type',OLD."bet_type",'deadline',OLD."deadline",'wager_quantity',OLD."wager_quantity",'wager_remaining',OLD."wager_remaining",'counterwager_quantity',OLD."counterwager_quantity",'counterwager_remaining',OLD."counterwager_remaining",'target_value',OLD."target_value",'leverage',OLD."leverage",'expiration',OLD."expiration",'expire_index',OLD."expire_index",'fee_fraction_int',OLD."fee_fraction_int",'status',OLD."status") FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_bets_delete AFTER DELETE ON bets +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'bets',json_object('tx_index',OLD."tx_index"),json_object('tx_index',OLD."tx_index",'tx_hash',CASE WHEN OLD."tx_hash" IS NULL THEN NULL ELSE hex(OLD."tx_hash") END,'block_index',OLD."block_index",'block_time',OLD."block_time",'source_id',OLD."source_id",'feed_address_id',OLD."feed_address_id",'bet_type',OLD."bet_type",'deadline',OLD."deadline",'wager_quantity',OLD."wager_quantity",'wager_remaining',OLD."wager_remaining",'counterwager_quantity',OLD."counterwager_quantity",'counterwager_remaining',OLD."counterwager_remaining",'target_value',OLD."target_value",'leverage',OLD."leverage",'expiration',OLD."expiration",'expire_index',OLD."expire_index",'fee_fraction_int',OLD."fee_fraction_int",'status',OLD."status") FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_bet_matches_insert AFTER INSERT ON bet_matches +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'bet_matches',json_object('tx0_index',NEW."tx0_index",'tx1_index',NEW."tx1_index"),NULL FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_bet_matches_update AFTER UPDATE ON bet_matches +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) AND (OLD."tx0_index" IS NOT NEW."tx0_index" OR OLD."tx1_index" IS NOT NEW."tx1_index" OR OLD."tx0_hash" IS NOT NEW."tx0_hash" OR OLD."tx1_hash" IS NOT NEW."tx1_hash" OR OLD."tx0_address_id" IS NOT NEW."tx0_address_id" OR OLD."tx1_address_id" IS NOT NEW."tx1_address_id" OR OLD."feed_address_id" IS NOT NEW."feed_address_id" OR OLD."forward_quantity" IS NOT NEW."forward_quantity" OR OLD."backward_quantity" IS NOT NEW."backward_quantity" OR OLD."deadline" IS NOT NEW."deadline" OR OLD."target_value" IS NOT NEW."target_value" OR OLD."leverage" IS NOT NEW."leverage" OR OLD."initial_value" IS NOT NEW."initial_value" OR OLD."block_index" IS NOT NEW."block_index" OR OLD."block_time" IS NOT NEW."block_time" OR OLD."status" IS NOT NEW."status" OR OLD."tx0_bet_type" IS NOT NEW."tx0_bet_type" OR OLD."tx1_bet_type" IS NOT NEW."tx1_bet_type" OR OLD."fee_fraction_int" IS NOT NEW."fee_fraction_int" OR OLD."match_expire_index" IS NOT NEW."match_expire_index") +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'bet_matches',json_object('tx0_index',OLD."tx0_index",'tx1_index',OLD."tx1_index"),json_object('tx0_index',OLD."tx0_index",'tx1_index',OLD."tx1_index",'tx0_hash',CASE WHEN OLD."tx0_hash" IS NULL THEN NULL ELSE hex(OLD."tx0_hash") END,'tx1_hash',CASE WHEN OLD."tx1_hash" IS NULL THEN NULL ELSE hex(OLD."tx1_hash") END,'tx0_address_id',OLD."tx0_address_id",'tx1_address_id',OLD."tx1_address_id",'feed_address_id',OLD."feed_address_id",'forward_quantity',OLD."forward_quantity",'backward_quantity',OLD."backward_quantity",'deadline',OLD."deadline",'target_value',OLD."target_value",'leverage',OLD."leverage",'initial_value',OLD."initial_value",'block_index',OLD."block_index",'block_time',OLD."block_time",'status',OLD."status",'tx0_bet_type',OLD."tx0_bet_type",'tx1_bet_type',OLD."tx1_bet_type",'fee_fraction_int',OLD."fee_fraction_int",'match_expire_index',OLD."match_expire_index") FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_bet_matches_delete AFTER DELETE ON bet_matches +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'bet_matches',json_object('tx0_index',OLD."tx0_index",'tx1_index',OLD."tx1_index"),json_object('tx0_index',OLD."tx0_index",'tx1_index',OLD."tx1_index",'tx0_hash',CASE WHEN OLD."tx0_hash" IS NULL THEN NULL ELSE hex(OLD."tx0_hash") END,'tx1_hash',CASE WHEN OLD."tx1_hash" IS NULL THEN NULL ELSE hex(OLD."tx1_hash") END,'tx0_address_id',OLD."tx0_address_id",'tx1_address_id',OLD."tx1_address_id",'feed_address_id',OLD."feed_address_id",'forward_quantity',OLD."forward_quantity",'backward_quantity',OLD."backward_quantity",'deadline',OLD."deadline",'target_value',OLD."target_value",'leverage',OLD."leverage",'initial_value',OLD."initial_value",'block_index',OLD."block_index",'block_time',OLD."block_time",'status',OLD."status",'tx0_bet_type',OLD."tx0_bet_type",'tx1_bet_type',OLD."tx1_bet_type",'fee_fraction_int',OLD."fee_fraction_int",'match_expire_index',OLD."match_expire_index") FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_rps_insert AFTER INSERT ON rps +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'rps',json_object('tx_index',NEW."tx_index"),NULL FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_rps_update AFTER UPDATE ON rps +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) AND (OLD."tx_index" IS NOT NEW."tx_index" OR OLD."tx_hash" IS NOT NEW."tx_hash" OR OLD."block_index" IS NOT NEW."block_index" OR OLD."block_time" IS NOT NEW."block_time" OR OLD."source_id" IS NOT NEW."source_id" OR OLD."possible_moves" IS NOT NEW."possible_moves" OR OLD."wager" IS NOT NEW."wager" OR OLD."move_random_hash" IS NOT NEW."move_random_hash" OR OLD."expiration" IS NOT NEW."expiration" OR OLD."expire_index" IS NOT NEW."expire_index" OR OLD."status" IS NOT NEW."status") +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'rps',json_object('tx_index',OLD."tx_index"),json_object('tx_index',OLD."tx_index",'tx_hash',CASE WHEN OLD."tx_hash" IS NULL THEN NULL ELSE hex(OLD."tx_hash") END,'block_index',OLD."block_index",'block_time',OLD."block_time",'source_id',OLD."source_id",'possible_moves',OLD."possible_moves",'wager',OLD."wager",'move_random_hash',CASE WHEN OLD."move_random_hash" IS NULL THEN NULL ELSE hex(OLD."move_random_hash") END,'expiration',OLD."expiration",'expire_index',OLD."expire_index",'status',OLD."status") FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_rps_delete AFTER DELETE ON rps +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'rps',json_object('tx_index',OLD."tx_index"),json_object('tx_index',OLD."tx_index",'tx_hash',CASE WHEN OLD."tx_hash" IS NULL THEN NULL ELSE hex(OLD."tx_hash") END,'block_index',OLD."block_index",'block_time',OLD."block_time",'source_id',OLD."source_id",'possible_moves',OLD."possible_moves",'wager',OLD."wager",'move_random_hash',CASE WHEN OLD."move_random_hash" IS NULL THEN NULL ELSE hex(OLD."move_random_hash") END,'expiration',OLD."expiration",'expire_index',OLD."expire_index",'status',OLD."status") FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_rps_matches_insert AFTER INSERT ON rps_matches +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'rps_matches',json_object('tx0_index',NEW."tx0_index",'tx1_index',NEW."tx1_index"),NULL FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_rps_matches_update AFTER UPDATE ON rps_matches +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) AND (OLD."tx0_index" IS NOT NEW."tx0_index" OR OLD."tx1_index" IS NOT NEW."tx1_index" OR OLD."tx0_hash" IS NOT NEW."tx0_hash" OR OLD."tx1_hash" IS NOT NEW."tx1_hash" OR OLD."tx0_address_id" IS NOT NEW."tx0_address_id" OR OLD."tx1_address_id" IS NOT NEW."tx1_address_id" OR OLD."possible_moves" IS NOT NEW."possible_moves" OR OLD."wager" IS NOT NEW."wager" OR OLD."block_index" IS NOT NEW."block_index" OR OLD."block_time" IS NOT NEW."block_time" OR OLD."status" IS NOT NEW."status") +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'rps_matches',json_object('tx0_index',OLD."tx0_index",'tx1_index',OLD."tx1_index"),json_object('tx0_index',OLD."tx0_index",'tx1_index',OLD."tx1_index",'tx0_hash',CASE WHEN OLD."tx0_hash" IS NULL THEN NULL ELSE hex(OLD."tx0_hash") END,'tx1_hash',CASE WHEN OLD."tx1_hash" IS NULL THEN NULL ELSE hex(OLD."tx1_hash") END,'tx0_address_id',OLD."tx0_address_id",'tx1_address_id',OLD."tx1_address_id",'possible_moves',OLD."possible_moves",'wager',OLD."wager",'block_index',OLD."block_index",'block_time',OLD."block_time",'status',OLD."status") FROM reorg_undo_context WHERE singleton=1; +END; + +CREATE TRIGGER reorg_undo_rps_matches_delete AFTER DELETE ON rps_matches +WHEN EXISTS(SELECT 1 FROM reorg_undo_context WHERE singleton=1) +BEGIN + INSERT INTO reorg_undo(block_index,table_name,row_key,before_row) + SELECT block_index,'rps_matches',json_object('tx0_index',OLD."tx0_index",'tx1_index',OLD."tx1_index"),json_object('tx0_index',OLD."tx0_index",'tx1_index',OLD."tx1_index",'tx0_hash',CASE WHEN OLD."tx0_hash" IS NULL THEN NULL ELSE hex(OLD."tx0_hash") END,'tx1_hash',CASE WHEN OLD."tx1_hash" IS NULL THEN NULL ELSE hex(OLD."tx1_hash") END,'tx0_address_id',OLD."tx0_address_id",'tx1_address_id',OLD."tx1_address_id",'possible_moves',OLD."possible_moves",'wager',OLD."wager",'block_index',OLD."block_index",'block_time',OLD."block_time",'status',OLD."status") FROM reorg_undo_context WHERE singleton=1; +END; diff --git a/apps/api/src/indexer/events/context.ts b/apps/api/src/indexer/events/context.ts index e92d5a48..7a10f584 100644 --- a/apps/api/src/indexer/events/context.ts +++ b/apps/api/src/indexer/events/context.ts @@ -6,7 +6,7 @@ * one file per message type, each self-contained, routed by dispatch.ts. */ // A statement is a thunk so it can be bound against the live DB inside the batcher. -export type Stmt = (db: D1Database) => D1PreparedStatement; +export type Stmt = ((db: D1Database) => D1PreparedStatement) & { blockIndex?: number }; // One Counterparty event from the event stream. export interface Ev { event_index: number; diff --git a/apps/api/src/indexer/events/dispatch.ts b/apps/api/src/indexer/events/dispatch.ts index 6ca77901..a4939ab6 100644 --- a/apps/api/src/indexer/events/dispatch.ts +++ b/apps/api/src/indexer/events/dispatch.ts @@ -45,8 +45,10 @@ export function dispatch(ev: Ev, ctx: Ctx): void { // The chronological stream states block_time once per block (NEW_BLOCK); every later event in the // block inherits it. Divisibility comes from the chunk's local map, never from verbose enrichment. if (ev.event === "NEW_BLOCK" && p.block_time != null) ctx.blockTime = p.block_time; + const start = ctx.stmts.length; const h = HANDLERS[ev.event]; if (h) h({ ev, p, b, bt: p.block_time ?? ctx.blockTime, div: assetDivisible(ctx, p.asset) }, ctx); + for (const stmt of ctx.stmts.slice(start)) stmt.blockIndex = b; } // Every event name we have a handler for — used to build the stream fetch filter (single source of truth). export const HANDLED_EVENTS = Object.keys(HANDLERS); diff --git a/apps/api/src/indexer/reorg-undo.ts b/apps/api/src/indexer/reorg-undo.ts new file mode 100644 index 00000000..d6c17fd6 --- /dev/null +++ b/apps/api/src/indexer/reorg-undo.ts @@ -0,0 +1,126 @@ +/** Exact, bounded before-images. A batch's context and writes commit together; + * concurrent readers/writers never observe an active journal context. */ +export async function initializeUndo(db: D1Database, height: number): Promise { + await db.prepare("INSERT OR IGNORE INTO reorg_undo_state(singleton,floor) VALUES(1,?)").bind(height).run(); +} + +export async function pruneUndo(db: D1Database, height: number): Promise { + const floor = Math.max(0, height - 24); + await db.batch([ + db.prepare("DELETE FROM reorg_undo WHERE block_index<=?").bind(floor), + db.prepare("UPDATE reorg_undo_state SET floor=max(floor,?) WHERE singleton=1").bind(floor), + ]); +} + +export async function writeWithUndo( + db: D1Database, + writes: { statement: D1PreparedStatement; block?: number }[], +): Promise { + // At most two statements per write plus context cleanup, below D1's batch bound. + for (let offset = 0; offset < writes.length; offset += 40) { + const statements: D1PreparedStatement[] = [db.prepare("DELETE FROM reorg_undo_context")]; + let active: number | undefined; + for (const write of writes.slice(offset, offset + 40)) { + if (write.block !== active) { + statements.push( + write.block === undefined + ? db.prepare("DELETE FROM reorg_undo_context") + : db + .prepare( + "INSERT INTO reorg_undo_context(singleton,block_index) VALUES(1,?) ON CONFLICT(singleton) DO UPDATE SET block_index=excluded.block_index", + ) + .bind(write.block), + ); + active = write.block; + } + statements.push(write.statement); + } + statements.push(db.prepare("DELETE FROM reorg_undo_context")); + await db.batch(statements); + } +} + +const TABLES = new Set([ + "assets", + "orders", + "order_matches", + "dispensers", + "fairminters", + "pools", + "bets", + "bet_matches", + "rps", + "rps_matches", +]); +interface Column { + name: string; + type: string; + pk: number; +} +interface Undo { + seq: number; + table_name: string; + row_key: string; + before_row: string | null; +} +type StoredRow = Record; + +/** The sync entry point initializes the floor before checking for reorgs, + * so pre-deployment history fails closed. + * Each inverse and deletion of its journal row share one transaction, making + * interrupted rollback resumable without double undo. */ +export async function restoreUndo(db: D1Database, height: number): Promise { + const state = await db.prepare("SELECT floor FROM reorg_undo_state WHERE singleton=1").first<{ floor: number }>(); + if (!state) throw new Error("Undo history is not initialized; rebuild required"); + if (height < state.floor) throw new Error("Reorg predates exact undo history; rebuild required"); + const schemas = new Map(); + for (;;) { + const rows = await db + .prepare( + "SELECT seq,table_name,row_key,before_row FROM reorg_undo WHERE block_index>? ORDER BY seq DESC LIMIT 40", + ) + .bind(height) + .all(); + if (!rows.results.length) return true; + const statements = [db.prepare("DELETE FROM reorg_undo_context")]; + for (const row of rows.results) { + if (!TABLES.has(row.table_name)) throw new Error("Unknown undo table"); + let columns = schemas.get(row.table_name); + if (!columns) { + columns = (await db.prepare(`PRAGMA table_info(${row.table_name})`).all()).results; + schemas.set(row.table_name, columns); + } + const keys = columns.filter((column) => column.pk).sort((a, b) => a.pk - b.pk); + const decode = (column: Column, value: string | number | null) => { + if (column.type.toUpperCase() !== "BLOB" || value === null) return value; + if (typeof value !== "string" || !/^(?:[0-9a-f]{2})*$/i.test(value)) throw new Error("Invalid undo blob"); + return Uint8Array.from(value.match(/../g) ?? [], (byte) => parseInt(byte, 16)); + }; + if (row.before_row === null) { + const key = JSON.parse(row.row_key) as StoredRow; + statements.push( + db + .prepare( + `DELETE FROM ${row.table_name} WHERE ${keys.map((column) => `"${column.name}" IS ?`).join(" AND ")}`, + ) + .bind(...keys.map((column) => decode(column, key[column.name]))), + ); + } else { + const before = JSON.parse(row.before_row) as StoredRow; + const retained = columns.filter((column) => Object.hasOwn(before, column.name)); + const names = retained.map((column) => `"${column.name}"`); + const mutable = retained.filter((column) => !column.pk); + statements.push( + db + .prepare( + `INSERT INTO ${row.table_name}(${names.join(",")}) VALUES(${names.map(() => "?").join(",")}) + ON CONFLICT(${keys.map((column) => `"${column.name}"`).join(",")}) DO UPDATE SET ${mutable.map((column) => `"${column.name}"=excluded."${column.name}"`).join(",")}`, + ) + .bind(...retained.map((column) => decode(column, before[column.name]))), + ); + } + statements.push(db.prepare("DELETE FROM reorg_undo WHERE seq=?").bind(row.seq)); + } + await db.batch(statements); + } +} diff --git a/apps/api/src/indexer/sync.ts b/apps/api/src/indexer/sync.ts index 96f959f2..5cad9d15 100644 --- a/apps/api/src/indexer/sync.ts +++ b/apps/api/src/indexer/sync.ts @@ -1,3 +1,4 @@ +import { initializeUndo, pruneUndo, restoreUndo, writeWithUndo } from "#api/indexer/reorg-undo"; /** * Counterparty mirror engine — chronological event replay into normalized D1. * @@ -64,7 +65,10 @@ function setCoreStateStmt(db: D1Database, key: string, value: string): D1Prepare // without advancing the cursor; the chunk re-runs and idempotency re-applies it exactly. Never drops. async function batchAll(db: D1Database, stmts: Stmt[]): Promise { for (let i = 0; i < stmts.length; i += DB_BATCH) { - await db.batch(stmts.slice(i, i + DB_BATCH).map((f) => f(db))); + await writeWithUndo( + db, + stmts.slice(i, i + DB_BATCH).map((f) => ({ statement: f(db), block: f.blockIndex })), + ); } } @@ -208,6 +212,7 @@ export async function syncCoreEvents( let lastBlock = Number.parseInt(blockValue ?? "0", 10); if (!Number.isSafeInteger(lastIndex) || lastIndex < -1) throw new Error("replay cursor is invalid"); + await initializeUndo(env.CORE_DB, lastBlock); const rollbackValue = await getCoreState(env.CORE_DB, "rollback_to"); if (rollbackValue !== null) { const rollbackTo = Number(rollbackValue); @@ -251,8 +256,7 @@ export async function syncCoreEvents( // Persist the independently observed source height before replay. Even when a later event handler fails, // readiness must report the lag instead of comparing the local mirror to itself and claiming "synced". await setCoreStateStmt(env.CORE_DB, "source_tip_block", String(sourceTipBlock)).run(); - const followingWindow = tip - lastIndex < 5 * CHUNK; - if (followingWindow && lastBlock > 0) { + if (lastBlock > 0) { // Raw page writes can precede the durable cursor. Verify those headers // too, or a crash followed by a fork could retain orphan INSERT OR IGNORE rows. const rawTip = await env.CORE_DB.prepare( @@ -264,9 +268,12 @@ export async function syncCoreEvents( .bind(verifyBlock) .first<{ hash: string }>() .then((row) => row?.hash || null), - blockHash(env.COUNTERPARTY_API_BASE, verifyBlock), + sourceTipBlock < verifyBlock + ? Promise.resolve("past-source-tip") + : blockHash(env.COUNTERPARTY_API_BASE, verifyBlock), ]); - if (storedHash && actualHash && storedHash !== actualHash) { + if (!storedHash || !actualHash) throw new Error("Cannot verify applied block hash"); + if (storedHash !== actualHash) { let rollbackTo = verifyBlock - 1; let found = false; for (; rollbackTo >= Math.max(0, verifyBlock - 24); rollbackTo--) { @@ -330,7 +337,43 @@ export async function syncCoreEvents( assetDivisibility: await chunkAssetDivisibility(env.CORE_DB, slice), blockTime, }; - for (const event of slice) dispatch(event, ctx); + // Validate the actual event header, not just a separately cached height lookup. + let previousHeight = lastBlock; + let previousHash = + lastBlock > 0 + ? await env.CORE_DB.prepare("SELECT lower(hex(block_hash)) hash FROM blocks WHERE block_index=?") + .bind(lastBlock) + .first<{ hash: string }>() + .then((row) => row?.hash) + : undefined; + for (const event of slice) { + if (event.event === "NEW_BLOCK") { + const height = event.block_index; + const hash = event.params.block_hash; + if ( + typeof hash !== "string" || + !/^[0-9a-f]{64}$/i.test(hash) || + hash !== (await blockHash(env.COUNTERPARTY_API_BASE, height)) + ) + throw new Error(`Unverified event block ${height}`); + if (previousHash && (height !== previousHeight + 1 || event.params.previous_block_hash !== previousHash)) + throw new Error(`Event block ${height} does not extend the applied chain`); + previousHeight = height; + previousHash = hash; + } + dispatch(event, ctx); + } + // A fork during page retrieval must not overwrite the checkpoint with + // a child from the new branch and hide an orphaned parent forever. + if ( + lastBlock > 0 && + (await blockHash(env.COUNTERPARTY_API_BASE, lastBlock)) !== + (await env.CORE_DB.prepare("SELECT lower(hex(block_hash)) hash FROM blocks WHERE block_index=?") + .bind(lastBlock) + .first<{ hash: string }>() + .then((row) => row?.hash)) + ) + throw new Error("Applied chain changed while fetching events"); await batchAll(env.CORE_DB, [...dictionaryStatements(ctx.identities), ...ctx.stmts]); const dispenseTxs = slice .filter((event) => event.event === "DISPENSE") @@ -353,6 +396,7 @@ export async function syncCoreEvents( } } + await pruneUndo(env.CORE_DB, lastBlock); const caughtUp = lastIndex >= tip; if (caughtUp) { // Gated, not every tick. The prune retains the newest snapshot per @@ -442,6 +486,7 @@ async function deleteAbove(db: D1Database, table: string, block: number): Promis /** Reverse committed orphan balance deltas before deleting their provenance. */ export async function rollbackCoreDatabase(db: D1Database, rollbackTo: number, rollbackIndex: number): Promise { + await restoreUndo(db, rollbackTo); const restores = await balanceRollbackStatements(db, rollbackTo, rollbackIndex); for (let offset = 0; offset < restores.length; offset += DB_BATCH) { await db.batch(restores.slice(offset, offset + DB_BATCH)); @@ -483,12 +528,6 @@ export async function rollbackCoreDatabase(db: D1Database, rollbackTo: number, r AND NOT EXISTS (SELECT 1 FROM trades WHERE trades.venue=trade_legs.venue AND trades.ref=trade_legs.trade_ref)`, ) .run(); - await db - .prepare(`UPDATE orders SET status='open',closed_block_index=NULL WHERE closed_block_index>?`) - .bind(rollbackTo) - .run(); - await db.prepare(`UPDATE dispensers SET closed_block_index=NULL WHERE closed_block_index>?`).bind(rollbackTo).run(); - await db.prepare("DELETE FROM balance_snapshots WHERE block_index>?").bind(rollbackTo).run(); // Orphan ledger stays until all restores commit, so a crashed rollback can retry. await deleteAbove(db, "ledger_events", rollbackTo); diff --git a/apps/api/src/integrations/counterparty.ts b/apps/api/src/integrations/counterparty.ts index a5c81612..88b44ed8 100644 --- a/apps/api/src/integrations/counterparty.ts +++ b/apps/api/src/integrations/counterparty.ts @@ -1,3 +1,4 @@ +import { freshCounterpartyUrl } from "#api/integrations/fresh-read"; /** Counterparty HTTP client — fetch + decode (precision-safe) with good-citizen backoff. */ import { parseCounterpartyJson } from "#api/indexer/codec"; import { discard } from "#api/lib/net"; @@ -42,7 +43,9 @@ export async function counterpartyJson( for (let attempt = 0; ; attempt++) { const remaining = deadline - Date.now(); if (remaining <= 0) throw new Error(`Counterparty ${path} request deadline exceeded`); - const response = await fetch(`${api}${path}`, { signal: AbortSignal.timeout(Math.min(timeoutMs, remaining)) }); + const response = await fetch(freshCounterpartyUrl(`${api}${path}`), { + signal: AbortSignal.timeout(Math.min(timeoutMs, remaining)), + }); if (response.ok) { try { return parseCounterpartyResponse(await response.text()); diff --git a/apps/api/src/integrations/fresh-read.ts b/apps/api/src/integrations/fresh-read.ts new file mode 100644 index 00000000..bbc30a84 --- /dev/null +++ b/apps/api/src/integrations/fresh-read.ts @@ -0,0 +1,11 @@ +/** Core caches by URL across rollbacks. Its scalar arguments and verbose + * enrichment use the first value; the second value only changes the cache key. + * Keep pagination, limits and enrichment identical on every attempt. */ +export function freshCounterpartyUrl(input: string): string { + const url = new URL(input); + const verbose = url.searchParams.get("verbose") ?? "false"; + url.searchParams.delete("verbose"); + url.searchParams.append("verbose", verbose); + url.searchParams.append("verbose", crypto.randomUUID()); + return url.toString(); +} diff --git a/apps/api/tests/core-foundation-writes.test.ts b/apps/api/tests/core-foundation-writes.test.ts index 74c6800e..a65648b5 100644 --- a/apps/api/tests/core-foundation-writes.test.ts +++ b/apps/api/tests/core-foundation-writes.test.ts @@ -1,3 +1,5 @@ +import { counterpartyJson } from "#api/integrations/counterparty"; +import { restoreUndo, writeWithUndo } from "#api/indexer/reorg-undo"; import assert from "node:assert/strict"; import { readdirSync, readFileSync } from "node:fs"; import { DatabaseSync } from "node:sqlite"; @@ -108,6 +110,7 @@ test("raw balance quantities reject rounding, fractions, negatives and malformed test("audited repairs fix positive balances and snapshots atomically and reject stale plans", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const ctx = context(); dispatch(event("CREDIT", { address: "alice", asset: "RARE", quantity: "142" }, 1), ctx); await executeCore(database, ctx); @@ -152,6 +155,7 @@ test("audited repairs fix positive balances and snapshots atomically and reject test("snapshot failure cannot commit a balance high-water and retries are exact", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const ctx = context(); dispatch(event("CREDIT", { address: "alice", asset: "RARE", quantity: "10" }, 1), ctx); await executeCore(database, ctx); @@ -176,6 +180,7 @@ test("snapshot failure cannot commit a balance high-water and retries are exact" test("multiple blocks retain intermediate checkpoints and a snapshotless imported baseline", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const seed = context(); dispatch(event("CREDIT", { address: "alice", asset: "RARE", quantity: "100", block_index: 90 }, 1), seed); await executeCore(database, seed); @@ -207,6 +212,7 @@ test("multiple blocks retain intermediate checkpoints and a snapshotless importe test("rollback restores positive, zero and UTXO balances without snapshots or unapplied ledger deltas", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const utxo = `${"ab".repeat(32)}:0`; const seed = context(); dispatch(event("CREDIT", { address: "positive", asset: "RARE", quantity: "100" }, 1), seed); @@ -237,6 +243,7 @@ test("rollback restores positive, zero and UTXO balances without snapshots or un test("a rollback interrupted between batches resumes without reversing a balance twice", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const ctx = context(); for (let index = 0; index < 100; index++) { dispatch( @@ -269,6 +276,7 @@ test("a rollback interrupted between batches resumes without reversing a balance test("underflow aborts the entire planned balance slice without clamping", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const ctx = context(); dispatch(event("CREDIT", { address: "good", asset: "RARE", quantity: "5" }, 1), ctx); dispatch(event("DEBIT", { address: "bad", asset: "RARE", quantity: "1" }, 2), ctx); @@ -286,13 +294,16 @@ test("underflow aborts the entire planned balance slice without clamping", async test("a stale state hash cannot trigger a false rollback after a partial page", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const hash = "a1".repeat(32); database.exec(`INSERT INTO blocks(block_index,block_hash) VALUES(101,X'${hash}'); INSERT INTO core_state(key,value) VALUES ('last_event_index','10'),('last_block_index','101'), ('last_block_hash','${"a0".repeat(32)}');`); const originalFetch = globalThis.fetch; globalThis.fetch = async (input) => { - const url = String(input); + const parsed = new URL(String(input)); + parsed.searchParams.delete("verbose"); + const url = parsed.toString(); if (url.endsWith("/events?limit=1")) return new Response(JSON.stringify({ result: [{ event_index: 10 }] })); if (url.endsWith("/blocks/last")) return new Response(JSON.stringify({ result: { block_index: 101 } })); if (url.endsWith("/blocks/101")) return new Response(JSON.stringify({ result: { block_hash: hash } })); @@ -310,6 +321,7 @@ test("a stale state hash cannot trigger a false rollback after a partial page", test("rollback queries seek block ranges and exact balance identities", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const ctx = context(); dispatch(event("CREDIT", { address: "alice", asset: "RARE", quantity: "10", block_index: 101 }, 10), ctx); await executeCore(database, ctx); @@ -336,6 +348,7 @@ test("rollback queries seek block ranges and exact balance identities", async () test("a fork after raw writes but before the cursor commit discards the orphan prefix", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const seed = context(); dispatch(event("CREDIT", { address: "alice", asset: "RARE", quantity: "100", block_index: 99 }, 1), seed); await executeCore(database, seed); @@ -349,7 +362,9 @@ test("a fork after raw writes but before the cursor commit discards the orphan p INSERT INTO core_state(key,value) VALUES('last_block_index','100'),('last_event_index','9'),('last_block_hash','${hash}');`); const originalFetch = globalThis.fetch; globalThis.fetch = async (input) => { - const url = String(input); + const parsed = new URL(String(input)); + parsed.searchParams.delete("verbose"); + const url = parsed.toString(); if (url.endsWith("/events?limit=1")) return new Response(JSON.stringify({ result: [{ event_index: 9 }] })); if (url.endsWith("/blocks/last")) return new Response(JSON.stringify({ result: { block_index: 101 } })); if (url.endsWith("/blocks/101")) return new Response(JSON.stringify({ result: { block_hash: "cc".repeat(32) } })); @@ -371,10 +386,16 @@ test("a fork after raw writes but before the cursor commit discards the orphan p test("a missing event is never skipped even if a later page contains valid events", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); database.exec("INSERT INTO core_state(key,value) VALUES ('last_event_index','0'),('last_block_index','100')"); + database.exec(`INSERT OR IGNORE INTO blocks(block_index,block_hash) VALUES(100,X'${"a0".repeat(32)}');`); const originalFetch = globalThis.fetch; globalThis.fetch = async (input) => { - const url = String(input); + const parsed = new URL(String(input)); + parsed.searchParams.delete("verbose"); + const url = parsed.toString(); + if (url.endsWith("/blocks/100")) return Response.json({ result: { block_hash: "a0".repeat(32) } }); + if (url.endsWith("/blocks/101")) return Response.json({ result: { block_hash: "a1".repeat(32) } }); if (url.endsWith("/events?limit=1")) return new Response(JSON.stringify({ result: [{ event_index: 2 }] })); if (url.endsWith("/blocks/last")) return new Response(JSON.stringify({ result: { block_index: 100 } })); if (url.endsWith("/blocks/100")) return new Response(JSON.stringify({ result: {} })); @@ -397,6 +418,7 @@ test("a missing event is never skipped even if a later page contains valid event test("rollback refuses missing provenance before deleting any branch data", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const ctx = context(); dispatch(event("CREDIT", { address: "alice", asset: "RARE", quantity: "10", block_index: 101 }, 1), ctx); await executeCore(database, ctx); @@ -416,13 +438,17 @@ test("rollback refuses missing provenance before deleting any branch data", asyn test("pending rollback survives a process exit after branch deletion before the cursor commit", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const hash = "aa".repeat(32); database.exec(`INSERT INTO blocks(block_index,block_hash) VALUES(100,X'${hash}'); INSERT INTO core_state(key,value) VALUES ('last_event_index','10'),('last_block_index','101'), - ('rollback_to','100'),('rollback_event_index','9');`); + ('rollback_to','100'),('rollback_event_index','9'); + INSERT OR REPLACE INTO reorg_undo_state VALUES(1,100);`); const originalFetch = globalThis.fetch; globalThis.fetch = async (input) => { - const url = String(input); + const parsed = new URL(String(input)); + parsed.searchParams.delete("verbose"); + const url = parsed.toString(); if (url.endsWith("/events?limit=1")) return new Response(JSON.stringify({ result: [{ event_index: 9 }] })); if (url.endsWith("/blocks/last")) return new Response(JSON.stringify({ result: { block_index: 100 } })); if (url.endsWith("/blocks/100")) return new Response(JSON.stringify({ result: { block_hash: hash } })); @@ -441,6 +467,7 @@ test("pending rollback survives a process exit after branch deletion before the test("compact foundational writes preserve identities and converge on replay", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const hash = "ab".repeat(32); const previousHash = "99".repeat(32); const utxoHash = "cd".repeat(32); @@ -580,6 +607,7 @@ test("compact foundational writes preserve identities and converge on replay", a test("compact balance catch-up applies only events above an imported row high-water", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); database.exec(` INSERT INTO address_dictionary(address) VALUES ('alice'); INSERT INTO asset_dictionary(asset) VALUES ('RARE'); @@ -604,6 +632,7 @@ test("compact balance catch-up applies only events above an imported row high-wa test("compact asset creation, issuances, and MPMA sends preserve canonical identities", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const hash = "ef".repeat(32); const asset = "A123"; const ctx = context(); @@ -746,6 +775,7 @@ test("compact asset creation, issuances, and MPMA sends preserve canonical ident test("compact orders preserve match pairs and lifecycle updates", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const tx0 = "21".repeat(32); const tx1 = "43".repeat(32); const payHash = "65".repeat(32); @@ -885,6 +915,7 @@ test("compact orders preserve match pairs and lifecycle updates", async () => { test("compact transaction-level protocol records converge on replay", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const ctx = context(); const hash = (byte: string) => byte.repeat(32); @@ -996,6 +1027,7 @@ test("compact transaction-level protocol records converge on replay", async () = test("compact dispenser lifecycle preserves dispenser transaction relationships", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const dispenserHash = "71".repeat(32); const dispenseHash = "72".repeat(32); const refillHash = "73".repeat(32); @@ -1116,6 +1148,7 @@ test("compact dispenser lifecycle preserves dispenser transaction relationships" test("compact fairminter lifecycle preserves campaign and mint transaction identities", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const fairminterHash = "81".repeat(32); const fairmintHash = "82".repeat(32); const ctx = context(); @@ -1225,6 +1258,7 @@ test("compact fairminter lifecycle preserves campaign and mint transaction ident test("compact pool lifecycle preserves pair, swap, and liquidity identities", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const depositHash = "91".repeat(32); const matchHash = "92".repeat(32); const withdrawalHash = "93".repeat(32); @@ -1393,6 +1427,7 @@ test("compact pool swaps keep every fill when one routed order sweeps a pool twi // same pool. The fills share tx_hash/tx_index and differ only by event_index and quantities. const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const orderHash = "94".repeat(32); const ctx = context(); @@ -1478,6 +1513,7 @@ test("compact pool swaps keep every fill when one routed order sweeps a pool twi test("compact bet and RPS state machines preserve composite match identities", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const bet0Hash = "a1".repeat(32); const bet1Hash = "a2".repeat(32); const betMatchId = `${bet0Hash}_${bet1Hash}`; @@ -1643,15 +1679,23 @@ test("compact bet and RPS state machines preserve composite match identities", a test("replay advances its durable cursor", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); database.exec(` INSERT INTO core_state(key,value) VALUES ('last_event_index','0'), ('last_block_index','100'); `); const txHash = "d1".repeat(32); + database.exec( + `INSERT OR IGNORE INTO blocks(block_index,block_hash) VALUES(100,X'${"a0".repeat(32)}'),(101,X'${"a1".repeat(32)}');`, + ); const originalFetch = globalThis.fetch; globalThis.fetch = async (input) => { - const url = String(input); + const parsed = new URL(String(input)); + parsed.searchParams.delete("verbose"); + const url = parsed.toString(); + if (url.endsWith("/blocks/100")) return Response.json({ result: { block_hash: "a0".repeat(32) } }); + if (url.endsWith("/blocks/101")) return Response.json({ result: { block_hash: "a1".repeat(32) } }); if (url.endsWith("/events?limit=1")) { // The event row is authoritative even if a node's result_count uses // different indexing semantics. @@ -1722,6 +1766,7 @@ test("replay advances its durable cursor", async () => { test("replay checkpoints a dense fetched page in durable slices", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); database.exec(` INSERT INTO core_state(key,value) VALUES ('last_event_index','0'), @@ -1734,9 +1779,14 @@ test("replay checkpoints a dense fetched page in durable slices", async () => { params: {}, })); let batches = 0; + database.exec(`INSERT OR IGNORE INTO blocks(block_index,block_hash) VALUES(100,X'${"a0".repeat(32)}');`); const originalFetch = globalThis.fetch; globalThis.fetch = async (input) => { - const url = String(input); + const parsed = new URL(String(input)); + parsed.searchParams.delete("verbose"); + const url = parsed.toString(); + if (url.endsWith("/blocks/100")) return Response.json({ result: { block_hash: "a0".repeat(32) } }); + if (url.endsWith("/blocks/101")) return Response.json({ result: { block_hash: "a1".repeat(32) } }); if (url.endsWith("/events?limit=1")) { return new Response(JSON.stringify({ result_count: 251, result: [{ event_index: 251 }] })); } @@ -1764,6 +1814,7 @@ test("replay checkpoints a dense fetched page in durable slices", async () => { test("replay clamps the events page to the pending window and shrinks it when a page is unreadable", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); database.exec(` INSERT INTO core_state(key,value) VALUES ('last_event_index','0'), @@ -1777,9 +1828,16 @@ test("replay clamps the events page to the pending window and shrinks it when a tx_hash: hash, params: { tx_index: index, tx_hash: hash, block_index: 100 + index, source: "alice" }, }); + database.exec( + `INSERT OR IGNORE INTO blocks(block_index,block_hash) VALUES(100,X'${"a0".repeat(32)}'),(101,X'${"a1".repeat(32)}');`, + ); const originalFetch = globalThis.fetch; globalThis.fetch = async (input) => { - const url = String(input); + const parsed = new URL(String(input)); + parsed.searchParams.delete("verbose"); + const url = parsed.toString(); + if (url.endsWith("/blocks/100")) return Response.json({ result: { block_hash: "a0".repeat(32) } }); + if (url.endsWith("/blocks/101")) return Response.json({ result: { block_hash: "a1".repeat(32) } }); if (url.endsWith("/events?limit=1")) { return new Response(JSON.stringify({ result_count: 3 }), { status: 200 }); } @@ -1825,6 +1883,7 @@ test("replay clamps the events page to the pending window and shrinks it when a test("compact replay rolls back a mismatched checkpoint before accepting the replacement branch", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); database.exec(` INSERT INTO blocks(block_index,block_hash) VALUES(100,X'${"a0".repeat(32)}'),(101,X'${"a1".repeat(32)}'); INSERT INTO transactions(tx_index,tx_hash,block_index) VALUES(1,X'${"b1".repeat(32)}',101); @@ -1833,7 +1892,9 @@ test("compact replay rolls back a mismatched checkpoint before accepting the rep `); const originalFetch = globalThis.fetch; globalThis.fetch = async (input) => { - const url = String(input); + const parsed = new URL(String(input)); + parsed.searchParams.delete("verbose"); + const url = parsed.toString(); if (url.endsWith("/events?limit=1")) return new Response(JSON.stringify({ result_count: 10 })); if (url.endsWith("/blocks/last")) return new Response(JSON.stringify({ result: { block_index: 101 } })); if (url.endsWith("/blocks/101")) return new Response(JSON.stringify({ result: { block_hash: "replacement-101" } })); @@ -1901,6 +1962,7 @@ test("compact caught-up maintenance prunes superseded snapshots", async () => { test("compact rollback removes orphan rows and restores balance quantity and high-water", async () => { const database = new DatabaseSync(":memory:"); database.exec(CORE_DDL); + database.exec("INSERT INTO reorg_undo_state VALUES(1,0)"); const addressId = Number( database.prepare(`INSERT INTO address_dictionary(address) VALUES ('alice') RETURNING address_id`).get()?.address_id, ); @@ -1928,14 +1990,25 @@ test("compact rollback removes orphan rows and restores balance quantity and hig INSERT INTO ledger_events(event_index,direction,block_index,address_id,asset_id,quantity) VALUES (10,1,102,${addressId},${assetId},'10'); INSERT INTO orders(tx_index,tx_hash,block_index,status,closed_block_index) - VALUES (1,X'${"e1".repeat(32)}',100,'filled',102); + VALUES (1,X'${"e1".repeat(32)}',100,'open',NULL); + UPDATE orders SET give_remaining='80',get_remaining='160'; INSERT INTO dispensers(tx_index,tx_hash,block_index,source_id,asset_id,status,closed_block_index) - VALUES (3,X'${"e3".repeat(32)}',100,${addressId},${assetId},11,102); + VALUES (3,X'${"e3".repeat(32)}',100,${addressId},${assetId},0,NULL); INSERT INTO core_state(key,value) VALUES ('last_block_index','102'); `); - await rollbackCoreDatabase(d1(database), 101, 5); + const db = d1(database); + await writeWithUndo(db, [ + { + block: 102, + statement: db.prepare( + "UPDATE orders SET status='filled',closed_block_index=102,give_remaining='0',get_remaining='0'", + ), + }, + { block: 102, statement: db.prepare("UPDATE dispensers SET status=11,closed_block_index=102") }, + ]); + await rollbackCoreDatabase(db, 101, 5); const balance = database .prepare(`SELECT quantity,quantity_normalized,updated_block_index,updated_event_index FROM balances`) @@ -1948,12 +2021,17 @@ test("compact rollback removes orphan rows and restores balance quantity and hig assert.equal(database.prepare(`SELECT COUNT(*) count FROM transactions`).get()?.count, 1); assert.equal(database.prepare(`SELECT COUNT(*) count FROM ledger_events`).get()?.count, 0); assert.deepEqual( - { ...(database.prepare(`SELECT status,closed_block_index FROM orders`).get() as Record) }, - { status: "open", closed_block_index: null }, + { + ...(database.prepare(`SELECT status,closed_block_index,give_remaining,get_remaining FROM orders`).get() as Record< + string, + unknown + >), + }, + { status: "open", closed_block_index: null, give_remaining: "80", get_remaining: "160" }, ); assert.deepEqual( { ...(database.prepare(`SELECT status,closed_block_index FROM dispensers`).get() as Record) }, - { status: 11, closed_block_index: null }, + { status: 0, closed_block_index: null }, ); const state = Object.fromEntries( (database.prepare(`SELECT key,value FROM core_state`).all() as { key: string; value: string }[]).map((row) => [ @@ -1963,3 +2041,117 @@ test("compact rollback removes orphan rows and restores balance quantity and hig ); assert.equal(state.last_block_index, "101"); }); + +test("fixed event pages stay fresh across forks in a URL-retaining cache", async () => { + const original = globalThis.fetch; + const cache = new Map(); + let branch = "a"; + globalThis.fetch = async (input) => { + const url = new URL(String(input)); + assert.equal(url.searchParams.get("cursor"), "200"); + assert.equal(url.searchParams.get("verbose"), "true"); + if (!cache.has(url.href)) cache.set(url.href, JSON.stringify({ result: branch })); + return new Response(cache.get(url.href)); + }; + try { + assert.equal( + (await counterpartyJson<{ result: string }>("https://core.test", "/events?cursor=200&limit=100&verbose=true")) + .result, + "a", + ); + branch = "b"; + assert.equal( + (await counterpartyJson<{ result: string }>("https://core.test", "/events?cursor=200&limit=100&verbose=true")) + .result, + "b", + ); + } finally { + globalThis.fetch = original; + } +}); + +test("a stale parent height probe cannot splice a new event header onto the old chain", async () => { + const database = new DatabaseSync(":memory:"); + database.exec(CORE_DDL); + database.exec(`INSERT INTO blocks(block_index,block_hash) VALUES(100,X'${"aa".repeat(32)}'); + INSERT INTO core_state VALUES('last_block_index','100'),('last_event_index','0');`); + const original = globalThis.fetch; + globalThis.fetch = async (input) => { + const url = new URL(String(input)); + if (url.pathname === "/blocks/last") return Response.json({ result: { block_index: 101 } }); + if (url.pathname === "/blocks/100") return Response.json({ result: { block_hash: "aa".repeat(32) } }); + if (url.pathname === "/blocks/101") return Response.json({ result: { block_hash: "cc".repeat(32) } }); + if (!url.searchParams.has("cursor")) return Response.json({ result: [{ event_index: 1 }] }); + return Response.json({ + result: [ + { + event_index: 1, + event: "NEW_BLOCK", + block_index: 101, + params: { block_hash: "cc".repeat(32), previous_block_hash: "bb".repeat(32) }, + }, + ], + }); + }; + try { + await assert.rejects( + syncCoreEvents({ CORE_DB: d1(database), COUNTERPARTY_API_BASE: "https://core.test" }), + /does not extend/, + ); + assert.equal(database.prepare("SELECT MAX(block_index) n FROM blocks").get()?.n, 100); + assert.equal(database.prepare("SELECT value FROM core_state WHERE key='last_event_index'").get()?.value, "0"); + } finally { + globalThis.fetch = original; + database.close(); + } +}); + +test("catch-up verifies old checkpoints beyond the former 5000-event window and rejects unknown hashes", async () => { + const database = new DatabaseSync(":memory:"); + database.exec(CORE_DDL); + database.exec(`INSERT INTO blocks(block_index,block_hash) VALUES(100,X'${"aa".repeat(32)}'); + INSERT INTO core_state VALUES('last_block_index','100'),('last_event_index','0');`); + const original = globalThis.fetch; + globalThis.fetch = async (input) => { + const url = new URL(String(input)); + if (url.pathname === "/events") return Response.json({ result: [{ event_index: 6000 }] }); + if (url.pathname === "/blocks/last") return Response.json({ result: { block_index: 200 } }); + return Response.json({ result: {} }); + }; + try { + await assert.rejects( + syncCoreEvents({ CORE_DB: d1(database), COUNTERPARTY_API_BASE: "https://core.test" }), + /Cannot verify/, + ); + } finally { + globalThis.fetch = original; + database.close(); + } +}); + +test("mutable undo rolls back atomically on failure and rejects pre-deployment history", async () => { + const database = new DatabaseSync(":memory:"); + database.exec(CORE_DDL); + database.exec( + "INSERT INTO reorg_undo_state VALUES(1,100); INSERT INTO orders(tx_index,tx_hash,block_index,status,give_remaining) VALUES(1,zeroblob(32),100,'open','75')", + ); + const db = d1(database); + await writeWithUndo(db, [ + { block: 101, statement: db.prepare("UPDATE orders SET give_remaining='0',status='filled'") }, + ]); + await assert.rejects( + restoreUndo( + d1(database, () => { + throw new Error("crash"); + }), + 100, + ), + /crash/, + ); + assert.equal(database.prepare("SELECT give_remaining FROM orders").get()?.give_remaining, "0"); + await restoreUndo(db, 100); + await restoreUndo(db, 100); + assert.equal(database.prepare("SELECT give_remaining FROM orders").get()?.give_remaining, "75"); + await assert.rejects(restoreUndo(db, 99), /predates exact undo/); + database.close(); +}); diff --git a/apps/api/tests/node-test.d.ts b/apps/api/tests/node-test.d.ts index e25aa4e3..320299d1 100644 --- a/apps/api/tests/node-test.d.ts +++ b/apps/api/tests/node-test.d.ts @@ -19,6 +19,7 @@ declare module "node:assert/strict" { strictEqual(actual: unknown, expected: unknown, message?: string): void; ok(value: unknown, message?: string): void; match(value: string, regexp: RegExp, message?: string): void; + rejects(promise: Promise, expected?: RegExp): Promise; throws(fn: () => unknown, expected?: RegExp, message?: string): void; } const assert: Assert; diff --git a/apps/api/tests/replay-verbose-parity.test.ts b/apps/api/tests/replay-verbose-parity.test.ts index 273d96ea..9be481fa 100644 --- a/apps/api/tests/replay-verbose-parity.test.ts +++ b/apps/api/tests/replay-verbose-parity.test.ts @@ -371,7 +371,10 @@ async function replay(pages: FixtureEvent[][]): Promise { const tipIndex = pages.flat().length - 1; const originalFetch = globalThis.fetch; globalThis.fetch = async (input) => { - const url = String(input); + const parsed = new URL(String(input)); + parsed.searchParams.delete("verbose"); + const url = parsed.toString(); + if (url.endsWith("/blocks/101")) return Response.json({ result: { block_hash: "aa".repeat(32) } }); if (url.endsWith("/events?limit=1")) return new Response(JSON.stringify({ result_count: tipIndex + 1 })); if (url.endsWith("/blocks/last")) return new Response(JSON.stringify({ result: { block_index: 102 } })); if (url.endsWith("/blocks/102")) return new Response(JSON.stringify({ result: { block_hash: "ab".repeat(32) } })); diff --git a/docs/reorg-recovery.md b/docs/reorg-recovery.md new file mode 100644 index 00000000..63c4c5cc --- /dev/null +++ b/docs/reorg-recovery.md @@ -0,0 +1,30 @@ +# Reorg recovery + +Counterparty Core caches some block URLs across rollbacks. Every indexer read now +uses a fresh URL, including fixed event cursors, continuation pages and retries. +Core uses the first scalar `verbose` value; a second UUID value changes the URL +cache key without changing pagination or enrichment. The configured provider is +still trusted: this is not an independent Bitcoin consensus check. + +Replay verifies the last locally written block during catch-up as well as following. +An unavailable hash stops the pass. Each NEW_BLOCK must match a fresh header and +extend the locally applied hash; the parent is checked again after fetching a page. +A fork cannot be hidden by advancing to a child on a different branch. + +Migration `0100_reorg_undo.sql` retains exact before-images of mutable assets, +orders/matches, dispensers, fairminters, pools, bets/matches and RPS/matches. The +journal context and each batch of event writes share a D1 transaction. Rollback +applies inverses in reverse order, atomically removing each consumed journal row, +so interruption is resumable. Existing balance-delta rollback remains in place. +Remaining quantities and dispenser status are restored, not guessed from closure. + +Apply migration 0100 before deploying. The undo floor starts at the deployment +checkpoint and retains 24 blocks. Older forks stop for a verified rebuild; the +migration cannot invent before-images for previously indexed data. Pause replay +for administrative imports/rebuilds, then establish a new verified undo baseline. +Keep trigger column lists current when changing a journaled table's schema. + +Tests exercise a retained URL cache, stale parent probes, catch-up verification, +unknown hashes, exact partial-fill rollback, interrupted undo and missing history. +This change does not certify historical imported state or all asynchronous derived +projections; those retain their existing repair queues and maintenance behavior.