Exclude withdrawn pools from the reserve snapshot

The snapshot introduced in e733bad enumerated every pool the token ever had, so a
pool drained months ago still voted on the reference. A withdrawal writes no new
pool_history_entry row, so the drained pool's last entry still shows full
pre-withdrawal reserves -- it looks like deep liquidity that no longer exists.

This was a regression from the snapshot change: the previous per-window fold only
learned pools that actually traded in the window, so long-dead pools never entered
the map. It is also the OLA failure the Stage 2 design called out, where a 10.9B-sat
pool was withdrawn 830s before a crash and a lingering ghost would have muted it.

Reuses the filter poolvisitor already applies: a pool is visible if it was never
withdrawn, or if its withdrawal transaction is at or after the query instant, so
historical queries still see pools that were live at the time they ask about.

Tests: 239 passing

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/db/cauldron/candlestick/mod.rs

@@ -289,6 +289,12 @@ ORDER BY phe.effective_timestamp ASC, phe.sequence ASC;
 /// Enumerates the token's pools and seeks each one's latest history row, rather than
 /// scanning every row the token ever produced — a busy token has hundreds of thousands,
 /// and this runs once per query.
+///
+/// Pools already withdrawn at `timestamp` are excluded. A withdrawal leaves no history
+/// row behind, so a drained pool's last entry still shows its full pre-withdrawal
+/// reserves; counting those would let a pool that no longer holds anything go on voting
+/// the reference. Pools withdrawn *after* `timestamp` are kept, since they were live then
+/// — the same rule `poolvisitor` applies.
 pub(crate) async fn fetch_reserve_snapshot(
     pool: &SqlitePool,
     token_blob: &[u8],
@@ -305,11 +311,18 @@ JOIN pool_history_entry AS phe ON phe.utxo = (
   ORDER BY prior.effective_timestamp DESC, prior.sequence DESC
   LIMIT 1
 )
-WHERE p.token_id = ?;
+WHERE p.token_id = ?
+  AND (p.withdrawn_in_utxo IS NULL OR (
+        SELECT t.effective_timestamp
+        FROM utxo_spending AS us
+        JOIN tx AS t ON us.txid = t.txid
+        WHERE us.spent_utxo_hash = p.withdrawn_in_utxo
+      ) >= ?);
 "#;
     let rows = sqlx::query(sql)
         .bind(timestamp)
         .bind(token_blob)
+        .bind(timestamp)
         .fetch_all(pool)
         .await?;
 

src/db/cauldron/candlestick/tests.rs

@@ -9,6 +9,7 @@ use crate::db::cauldron::{
     pool::{self, dummy_init_seq},
     tx::{self, insert_block_tx, insert_mempool_tx},
     utxo_funding::{self, insert_utxo_funding},
+    utxo_spending,
 };
 use crate::utiltest::mock_db_pool;
 use bitcoin_hashes::Hash;
@@ -38,12 +39,46 @@ fn dummy_cauldron(
 
 async fn setup_db(pool: sqlx::SqlitePool) {
     utxo_funding::create_table(&pool).await;
+    utxo_spending::create_table(&pool).await;
     tx::create_table(&pool).await;
     pool::create_table(&pool).await;
     ohlcv::create_table(&pool).await;
     dummy_init_seq();
 }
 
+/// Register a pool row so the reserve snapshot can find it, optionally marking it
+/// withdrawn by the transaction `withdrawn_by` at `withdrawn_at`.
+async fn register_pool(
+    conn: &mut sqlx::pool::PoolConnection<sqlx::Sqlite>,
+    pool_hash: &OutPointHash,
+    token: &TokenID,
+    withdrawal: Option<(Txid, i64)>,
+) {
+    let withdrawn_utxo = withdrawal.map(|(txid, ts)| {
+        let spent = OutPointHash::from_byte_array(*txid.as_byte_array());
+        (spent, txid, ts)
+    });
+
+    sqlx::query("INSERT OR REPLACE INTO pool (creation_utxo, owner_pkh, token_id, withdrawn_in_utxo) VALUES (?, ?, ?, ?)")
+        .bind(pool_hash.to_blob())
+        .bind(PubkeyHash::all_zeros().to_blob())
+        .bind(token.to_blob())
+        .bind(withdrawn_utxo.as_ref().map(|(spent, _, _)| spent.to_blob()))
+        .execute(&mut **conn)
+        .await
+        .unwrap();
+
+    if let Some((spent, txid, ts)) = withdrawn_utxo {
+        insert_mempool_tx(&mut **conn, &txid, ts as u64).await.unwrap();
+        sqlx::query("INSERT OR REPLACE INTO utxo_spending (spent_utxo_hash, txid) VALUES (?, ?)")
+            .bind(spent.to_blob())
+            .bind(txid.to_blob())
+            .execute(&mut **conn)
+            .await
+            .unwrap();
+    }
+}
+
 async fn insert_trade_at(
     conn: &mut sqlx::pool::PoolConnection<sqlx::Sqlite>,
     token: &TokenID,
@@ -218,6 +253,50 @@ async fn test_single_direction_multileg_closes_on_its_last_leg() {
     );
 }
 
+/// A pool that has been drained leaves its last history row showing full pre-withdrawal
+/// reserves, because a withdrawal writes no new row. Seeding the policy from those would
+/// let a pool holding nothing keep voting the reference — the OLA case, where a 10.9B-sat
+/// pool was withdrawn shortly before a crash and a lingering ghost would have muted it.
+#[tokio::test]
+async fn test_reserve_snapshot_drops_pools_withdrawn_before_the_instant() {
+    let db = mock_db_pool(setup_db).await;
+    let token = TokenID::from_byte_array([0xC6; 32]);
+    let token_blob = token.to_blob();
+
+    let mut conn = db.cauldron_w.acquire().await.unwrap();
+    insert_trade_at(&mut conn, &token, 0x41, 1000, 100_000, 2_000).await;
+
+    let pool_hash = OutPointHash::from_byte_array([0x41u8.wrapping_add(0x80); 32]);
+    register_pool(&mut conn, &pool_hash, &token, None).await;
+
+    let live = super::fetch_reserve_snapshot(&db.cauldron_r, &token_blob, 5000)
+        .await
+        .unwrap();
+    assert_eq!(live.len(), 1, "a live pool must be in the snapshot");
+
+    // Withdraw it at ts=2000 and ask again afterwards.
+    let withdrawal_tx = Txid::from_byte_array([0x42; 32]);
+    register_pool(&mut conn, &pool_hash, &token, Some((withdrawal_tx, 2000))).await;
+
+    let after = super::fetch_reserve_snapshot(&db.cauldron_r, &token_blob, 5000)
+        .await
+        .unwrap();
+    assert!(
+        after.is_empty(),
+        "a pool withdrawn at 2000 must not vote on the reference at 5000: {after:?}"
+    );
+
+    // But it was live at ts=1500, so a historical query must still see it.
+    let before = super::fetch_reserve_snapshot(&db.cauldron_r, &token_blob, 1500)
+        .await
+        .unwrap();
+    assert_eq!(
+        before.len(),
+        1,
+        "a pool withdrawn later was still live earlier and must remain visible"
+    );
+}
+
 /// Regression: the seed lookback was briefly capped at 24 hours, which silently dropped
 /// the carry-forward price for any token trading less often than daily and left a hole
 /// where the leading candles should be.

src/db/cauldron/ohlcv.rs

@@ -317,7 +317,7 @@ pub async fn get_active_candles(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::db::cauldron::{pool as cauldron_pool, tx, utxo_funding};
+    use crate::db::cauldron::{pool as cauldron_pool, tx, utxo_funding, utxo_spending};
     use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
     use std::sync::atomic::{AtomicU64, Ordering};
 
@@ -335,6 +335,7 @@ mod tests {
     async fn setup_db(pool: &SqlitePool) {
         tx::create_table(pool).await;
         utxo_funding::create_table(pool).await;
+        utxo_spending::create_table(pool).await;
         cauldron_pool::create_table(pool).await;
         create_table(pool).await; // ohlcv_1h + idx_phe_txid
     }

src/rpc/candlesticks/tests.rs

@@ -8,6 +8,7 @@ use crate::db::cauldron::{
     pool::{self, dummy_init_seq, insert_new_pool},
     tx::{self, insert_block_tx, insert_mempool_tx},
     utxo_funding::{self, insert_utxo_funding},
+    utxo_spending,
 };
 use crate::utiltest::mock_db_pool;
 use crate::OhlcvState;
@@ -64,6 +65,7 @@ fn dummy_cauldron(
 
 async fn setup_mock_db(pool: sqlx::SqlitePool) {
     utxo_funding::create_table(&pool).await;
+    utxo_spending::create_table(&pool).await;
     tx::create_table(&pool).await;
     pool::create_table(&pool).await;
     ohlcv::create_table(&pool).await;
@@ -201,6 +203,7 @@ async fn setup_mock_db(pool: sqlx::SqlitePool) {
 
 async fn setup_seed_db(pool: sqlx::SqlitePool) {
     utxo_funding::create_table(&pool).await;
+    utxo_spending::create_table(&pool).await;
     tx::create_table(&pool).await;
     pool::create_table(&pool).await;
     ohlcv::create_table(&pool).await;
@@ -379,6 +382,7 @@ async fn test_multiple_candlesticks_endpoint() {
 async fn test_single_swap_multiple_pools() {
     let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move {
         utxo_funding::create_table(&pool).await;
+        utxo_spending::create_table(&pool).await;
         tx::create_table(&pool).await;
         pool::create_table(&pool).await;
         ohlcv::create_table(&pool).await;
@@ -481,6 +485,7 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() {
     let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move {
         // --- boilerplate setup ---
         utxo_funding::create_table(&pool).await;
+        utxo_spending::create_table(&pool).await;
         tx::create_table(&pool).await;
         pool::create_table(&pool).await;
         ohlcv::create_table(&pool).await;