Merge #2167: fix(bitcoind_rpc): emit invalidated heights when start_height is above agreement point

76b6530 fix(bitcoind_rpc): emit invalidated heights when start_height is above agreement point (志宇)

Pull request description:

  ### Description

  When a reorg drops the agreement point below `start_height`, the emitter would skip directly to `start_height`, producing a checkpoint that could not connect with the caller's local chain (`CannotConnectError`).

  This affects callers that create a new `Emitter` on each sync with `start_height = tip_height` — a common pattern.

  **Fix:** Override `start_height` to the agreement height when a reorg is detected (agreement point below both `start_height` and `last_cp`), so the emitter revisits the invalidated block heights.

  Context: https://discord.com/channels/753336465005608961/753367451319926827/1489544612094808235

  cc @stevenroose

  ### Notes to the reviewers

  The key change is in `poll()` in `crates/bitcoind_rpc/src/lib.rs`: when `AgreementFound` is handled and the agreement point is below both `start_height` and `last_cp.height()`, we lower `start_height` to the agreement height. This ensures the emitter emits the invalidated blocks instead of skipping over them.

  ### Changelog notice

  ```md
  Fixed:
  - `Emitter` producing un-connectable checkpoints when `start_height` is above the agreement point after a reorg.
  ```

  ### Checklists

  #### All Submissions:

  * [x] I followed the [contribution guidelines](https://github.com/bitcoindevkit/bdk/blob/master/CONTRIBUTING.md)

  #### Bugfixes:

  * [ ] This pull request breaks the existing API
  * [x] I've added tests to reproduce the issue which are now passing
  * [x] I'm linking the issue being fixed by this PR

ACKs for top commit:
  luisschwab:
    ACK 76b6530

Tree-SHA512: ec9c382e8619b56284ee8ac934e5934dab3b067c0f33f79f4aa3b1a8877750093a6aeff939fe51add73dd1f71d4ce786af57a905730c150b3c112ac62ff210a8

Author: evanlinjin

crates/bitcoind_rpc/src/lib.rs

@@ -282,7 +282,7 @@ where
     let client = &*emitter.client;
 
     if let Some(last_res) = &emitter.last_block {
-        let next_hash = if last_res.height < emitter.start_height as _ {
+        let next_hash = if last_res.height + 1 < emitter.start_height as _ {
             // enforce start height
             let next_hash = client.get_block_hash(emitter.start_height as _)?;
             // make sure last emission is still in best chain
@@ -362,6 +362,13 @@ where
                 continue;
             }
             PollResponse::AgreementFound(res, cp) => {
+                // When a reorg happens, the agreement point drops below `last_cp`. We
+                // override `start_height` so the emitter revisits the invalidated heights.
+                if (res.height as u32) < emitter.start_height
+                    && (res.height as u32) < emitter.last_cp.height()
+                {
+                    emitter.start_height = res.height as _;
+                }
                 // get rid of evicted blocks
                 emitter.last_cp = cp;
                 emitter.last_block = Some(res);

crates/bitcoind_rpc/tests/test_emitter.rs

@@ -487,7 +487,7 @@ fn mempool_avoids_re_emission() -> anyhow::Result<()> {
 /// If blockchain re-org includes the start height, emit new start height block
 ///
 /// 1. mine 101 blocks
-/// 2. emit blocks 99a, 100a
+/// 2. emit blocks 98a, 99a, 100a
 /// 3. invalidate blocks 99a, 100a, 101a
 /// 4. mine new blocks 99b, 100b, 101b
 /// 5. emit block 99b
@@ -505,48 +505,45 @@ fn no_agreement_point() -> anyhow::Result<()> {
     let mut emitter = Emitter::new(
         &client,
         CheckPoint::new(0, env.genesis_hash()?),
-        (PREMINE_COUNT - 2) as u32,
+        (PREMINE_COUNT - 3) as u32,
         NO_EXPECTED_MEMPOOL_TXS,
     );
 
     // mine 101 blocks
     env.mine_blocks(PREMINE_COUNT, None)?;
 
-    // emit block 99a
-    let block_header_99a = emitter
-        .next_block()?
-        .expect("block 99a header")
-        .block
-        .header;
-    let block_hash_99a = block_header_99a.block_hash();
-    let block_hash_98a = block_header_99a.prev_blockhash;
-
-    // emit block 100a
-    let block_header_100a = emitter.next_block()?.expect("block 100a header").block;
-    let block_hash_100a = block_header_100a.block_hash();
+    // emit blocks: 98a, 99a, 100a
+    let block_98a = emitter.next_block()?.expect("block 98a");
+    let block_99a = emitter.next_block()?.expect("block 99a");
+    let block_100a = emitter.next_block()?.expect("block 100a");
+    assert_eq!(block_98a.block_height(), 98);
+    assert_eq!(block_99a.block_height(), 99);
+    assert_eq!(block_100a.block_height(), 100);
 
     // get hash for block 101a
-    let block_hash_101a = env.rpc_client().get_block_hash(101)?.block_hash()?;
+    let blockhash_101a = env.rpc_client().get_block_hash(101)?.block_hash()?;
 
     // invalidate blocks 99a, 100a, 101a
-    env.rpc_client().invalidate_block(block_hash_99a)?;
-    env.rpc_client().invalidate_block(block_hash_100a)?;
-    env.rpc_client().invalidate_block(block_hash_101a)?;
+    env.rpc_client().invalidate_block(blockhash_101a)?;
+    env.rpc_client().invalidate_block(block_100a.block_hash())?;
+    env.rpc_client().invalidate_block(block_99a.block_hash())?;
 
     // mine new blocks 99b, 100b, 101b
     env.mine_blocks(3, None)?;
 
     // emit block header 99b
-    let block_header_99b = emitter
-        .next_block()?
-        .expect("block 99b header")
-        .block
-        .header;
-    let block_hash_99b = block_header_99b.block_hash();
-    let block_hash_98b = block_header_99b.prev_blockhash;
+    let block_99b = emitter.next_block()?.expect("block 99b");
+    assert_eq!(block_99b.block_height(), 99);
 
-    assert_ne!(block_hash_99a, block_hash_99b);
-    assert_eq!(block_hash_98a, block_hash_98b);
+    assert_ne!(block_99a.block_hash(), block_99b.block_hash());
+    assert_eq!(
+        block_98a.block_hash(),
+        block_99a.block.header.prev_blockhash
+    );
+    assert_eq!(
+        block_98a.block_hash(),
+        block_99b.block.header.prev_blockhash
+    );
 
     Ok(())
 }
@@ -661,6 +658,47 @@ fn test_expect_tx_evicted() -> anyhow::Result<()> {
     Ok(())
 }
 
+/// Creating a new [`Emitter`] after a reorg with `start_height` at the tip should still
+/// produce a connectable checkpoint. When blocks are invalidated, the emitted checkpoint must
+/// include the invalidation height so the update can connect with the original chain.
+#[test]
+fn test_sync_with_new_emitter_after_reorg() -> anyhow::Result<()> {
+    let env = TestEnv::new()?;
+    let (mut local_chain, _) = LocalChain::from_genesis(env.genesis_hash()?);
+    let client = ClientExt::get_rpc_client(&env)?;
+
+    env.mine_blocks(110, None)?;
+
+    let mut emitter = Emitter::new(&client, local_chain.tip(), 0, NO_EXPECTED_MEMPOOL_TXS);
+    while let Some(emission) = emitter.next_block()? {
+        let _ = local_chain.apply_update(emission.checkpoint)?;
+    }
+
+    let pre_reorg_tip = local_chain.tip();
+    let tip_height = pre_reorg_tip.height();
+
+    env.reorg(6)?;
+
+    // New emitter with start_height = tip height (common caller pattern).
+    let mut emitter = Emitter::new(
+        &client,
+        local_chain.tip(),
+        tip_height,
+        NO_EXPECTED_MEMPOOL_TXS,
+    );
+
+    while let Some(emission) = emitter.next_block()? {
+        let _ = local_chain
+            .apply_update(emission.checkpoint)
+            .expect("emission checkpoint must connect with local chain");
+    }
+
+    assert_eq!(local_chain.tip().height(), tip_height);
+    assert_ne!(local_chain.tip().hash(), pre_reorg_tip.hash());
+
+    Ok(())
+}
+
 #[test]
 fn detect_new_mempool_txs() -> anyhow::Result<()> {
     let env = TestEnv::new()?;