feat(dash-spv): filter re-match across reorg and safe auto-rebroadcast (#170)

* feat(key-wallet-manager): add `WalletEvent::TxRepeatedlyOrphaned` variant

New variant emitted by the SPV mempool manager when an outgoing
transaction demoted by reorg has exhausted its rebroadcast retry
budget. UI consumers surface this so the user can decide to abandon,
re-sign, or wait.

The `wallet_id` field is `Option<WalletId>` because the rebroadcast
loop may emit this for a txid that is no longer attributable to any
managed wallet. `WalletEvent::wallet_id()` is widened to
`Option<WalletId>` to accommodate.

* feat(dash-spv): drive wallet rewind and per-wallet effective floor on `ChainReorg`

`FiltersManager` now calls `WalletInterface::rewind_to_height` when it
observes a `ChainReorg` event, then resets its filter scan cursor to
the effective per-wallet floor (`min(fork_height, wallet.synced_height)`)
instead of always to `fork_height`. A wallet whose ingest tip is behind
the fork never saw the orphaned chain, so the filter pipeline does not
need to re-match the segment between its tip and the fork.

If the wallet refuses the rewind (e.g. chainlock floor, defense in depth since the SPV cascade already enforces this upstream), log and fall back to `fork_height` for the floor without touching wallet state.

Renames `reset_for_reorg`'s parameter to `effective_floor` to make the
contract explicit.

* feat(dash-spv): auto-rebroadcast reorg-demoted transactions with safety checks

`MempoolManager` now subscribes to `WalletEvent` broadcasts and drives
an auto-rebroadcast pipeline for transactions demoted by a chain reorg:

- `WalletEvent::Reorg` enqueues each `demoted_txid` after a
  chainlock-floor defense check (the wallet already refuses to rewind
  below its chainlock floor upstream, this branch is belt-and-braces).
- Block-height `nLockTime` is compared against the current chain tip;
  unsatisfied locktimes defer the broadcast. Timestamp locktimes are
  compared best-effort against wall clock, never against an MTP cache
  the SPV side does not maintain.
- An input-conflict check evicts pending entries whose inputs collide
  with any transaction confirming on the new chain (delivered via
  `WalletEvent::BlockProcessed`).
- Each attempt advances an exponential backoff capped at one hour.
  After `MAX_REORG_REBROADCAST_ATTEMPTS` (10) failures the entry is
  dropped and a `WalletEvent::TxRepeatedlyOrphaned` is emitted so
  the UI can surface it for user intervention.

The manager owns a forwarder `broadcast::channel` so it can both subscribe to wallet events (replaceable via `set_wallet_event_subscription`) and emit `TxRepeatedlyOrphaned` without changing the trait surface.

The constructor gains a `chainlock_height: Arc<AtomicU32>` parameter threaded through from `lifecycle.rs`. `MempoolManager` also picks up a `current_tip_height` field updated via `SyncManager::update_target_height`.

Tests cover the happy path, input-conflict eviction, locktime defer
followed by tip advance, and the retry-cap orphan event.

* chore(dash-spv-ffi): mark new `TxRepeatedlyOrphaned` variant as not-yet-bridged

Add the missing match arm in `FFIWalletEventCallbacks::dispatch`. The
variant is left without an FFI surface for now (matching the existing
`Reorg` arm) so the C ABI stays unchanged. Wiring a dedicated callback
is deferred to a follow-up that surfaces orphaned-tx state to the UI.

* chore: pr cleanup: fix FQ paths and remove what-comments

* test(dash-spv): cover rebroadcast backoff, chainlock floor, and channel-driven paths

Addresses four `manki-review` testing-coverage findings on PR #170:

- Extend `test_reorg_demoted_tx_rebroadcast_happy_path` with a second immediate `drive_reorg_rebroadcast` call to assert the exponential backoff suppresses a duplicate broadcast and does not bump `attempts`.
- Add `test_chainlock_floor_prevents_enqueue` exercising the chainlock-floor defense in `enqueue_demoted_for_rebroadcast` when `chainlock_height > current_tip_height`.
- Add `test_input_conflict_via_channel_evicts_rebroadcast` so the `WalletEvent::BlockProcessed` arm of `drain_wallet_events` is exercised through the channel, not by calling `evict_conflicting_rebroadcasts` directly.
- Extend `test_block_processed_removes_confirmed_txids` to seed `pending_rebroadcast` via `enqueue_demoted_for_rebroadcast` and assert the confirmed txid is removed from it, locking in the `BlockProcessed` handler's `pending_rebroadcast.remove(txid)` line.

Promotes `enqueue_demoted_for_rebroadcast` to `pub(super)` so sibling-module tests can drive the rebroadcast queue through the public API.

* ci: trigger workflows on feat/filter-rematch-and-rebroadcast

* test(dash-spv): narrow `enqueue_demoted_for_rebroadcast` visibility and clarify chainlock-floor test

Keep `enqueue_demoted_for_rebroadcast` private and expose a `#[cfg(test)] pub(super)` wrapper so sibling-module tests in `sync_manager.rs` can seed `pending_rebroadcast` without widening production visibility.

Drop the unused `wallet_event_rx` / `set_wallet_event_subscription` setup from `test_chainlock_floor_prevents_enqueue` (the test exercises the direct call, not the channel path) and document the ordering invariant the `transactions` postcondition relies on (the guard must short-circuit before the wallet fetch/insert path).

Author: xdustinface

dash-spv-ffi/src/callbacks.rs

@@ -1215,6 +1215,14 @@ impl FFIWalletEventCallbacks {
                 // conflicted txid lists and the post-rewind balance.
                 // Until then this variant has no surface on the C ABI.
             }
+            WalletEvent::TxRepeatedlyOrphaned {
+                ..
+            } => {
+                // TODO(issue #146): wire a dedicated FFI callback so
+                // durable consumers can surface "this transaction has
+                // been orphaned too many times" to the UI. Until then
+                // this variant has no surface on the C ABI.
+            }
             WalletEvent::ChainLockProcessed {
                 wallet_id,
                 chain_lock,

dash-spv/src/client/lifecycle.rs

@@ -163,12 +163,16 @@ impl<W: WalletInterface, N: NetworkManager, S: StorageManager> DashSpvClient<W,
         // Build mempool manager if tracking is enabled
         if config.enable_mempool_tracking {
             let initial_revision = wallet.read().await.monitor_revision();
-            managers.mempool = Some(MempoolManager::new(
+            let wallet_event_rx = wallet.read().await.subscribe_events();
+            let mut mempool = MempoolManager::new(
                 wallet.clone(),
                 config.mempool_strategy,
                 config.max_mempool_transactions,
                 initial_revision,
-            ));
+                chainlock_height.clone(),
+            );
+            mempool.set_wallet_event_subscription(wallet_event_rx);
+            managers.mempool = Some(mempool);
         }
 
         // `reorg_generation` is held by each manager (via clones threaded in

dash-spv/src/sync/filters/manager.rs

@@ -149,16 +149,46 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
         self.filter_pipeline = FiltersPipeline::new();
     }
 
-    /// Reset all sync state to begin again at `fork_height` after a reorg
-    /// cascade. Clears in-flight batches and rewinds the per-batch
+    /// Drive the per-wallet rewind and compute the effective floor for
+    /// the filter re-match. Returns the lowest per-wallet `synced_height`
+    /// after the rewind clamps each wallet to `min(fork_height, current)`.
+    ///
+    /// The chainlock-floor invariant is enforced upstream by the reorg
+    /// cascade, so a `BelowChainLockFloor` error here is defense in
+    /// depth: log and fall back to `fork_height` for the floor without
+    /// touching wallet state.
+    pub(super) async fn rewind_wallet_for_reorg(&self, fork_height: u32) -> u32 {
+        let mut wallet = self.wallet.write().await;
+        match wallet.rewind_to_height(fork_height).await {
+            Ok(_) => wallet.synced_height().min(fork_height),
+            Err(e) => {
+                tracing::warn!(
+                    "FiltersManager: wallet rewind to {} rejected: {}; using fork_height as floor",
+                    fork_height,
+                    e
+                );
+                fork_height
+            }
+        }
+    }
+
+    /// Reset all sync state to begin again at `effective_floor` after a
+    /// reorg cascade. Clears in-flight batches and rewinds the per-batch
     /// processing cursors so the next `FilterHeadersStored` event drives a
-    /// fresh download starting at the truncated ancestor.
-    pub(super) fn reset_for_reorg(&mut self, fork_height: u32) {
+    /// fresh download starting at the effective floor.
+    ///
+    /// `effective_floor` is the minimum of `fork_height` and the lowest
+    /// per-wallet `synced_height` across the wallet set, computed after
+    /// the wallet rewind. A wallet whose ingest tip is behind the fork
+    /// point never saw the orphaned chain, so the filter pipeline does
+    /// not need to re-match the segment between that wallet's tip and
+    /// the fork.
+    pub(super) fn reset_for_reorg(&mut self, effective_floor: u32) {
         self.reset_for_rescan();
-        self.next_batch_to_store = fork_height;
-        self.processing_height = fork_height;
-        self.progress.update_committed_height(fork_height);
-        self.progress.update_stored_height(fork_height);
+        self.next_batch_to_store = effective_floor;
+        self.processing_height = effective_floor;
+        self.progress.update_committed_height(effective_floor);
+        self.progress.update_stored_height(effective_floor);
     }
 
     async fn load_filters(
@@ -2405,4 +2435,35 @@ mod tests {
             events
         );
     }
+
+    #[tokio::test]
+    async fn test_rewind_wallet_for_reorg_uses_min_of_wallet_and_fork() {
+        let manager = create_test_manager().await;
+        manager.wallet.write().await.update_wallet_synced_height(&MOCK_WALLET_ID, 40);
+
+        let effective_floor = manager.rewind_wallet_for_reorg(100).await;
+
+        assert_eq!(
+            effective_floor, 40,
+            "wallet behind fork_height should drag floor down to its synced_height"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_rewind_wallet_for_reorg_clamps_to_fork_height() {
+        let manager = create_test_manager().await;
+        manager.wallet.write().await.update_wallet_synced_height(&MOCK_WALLET_ID, 200);
+
+        let effective_floor = manager.rewind_wallet_for_reorg(100).await;
+
+        assert_eq!(
+            effective_floor, 100,
+            "wallet ahead of fork_height should be clamped and floor equals fork_height"
+        );
+        assert_eq!(
+            manager.wallet.read().await.wallet_synced_height(&MOCK_WALLET_ID),
+            100,
+            "rewind must clamp wallet's synced_height to fork_height"
+        );
+    }
 }

dash-spv/src/sync/filters/sync_manager.rs

@@ -190,10 +190,15 @@ impl<
                 ..
             } => {
                 tracing::info!(
-                    "FiltersManager: cascading ChainReorg, resetting state at {}",
+                    "FiltersManager: cascading ChainReorg, rewinding wallet at {}",
                     fork_height
                 );
-                self.reset_for_reorg(*fork_height);
+                let effective_floor = self.rewind_wallet_for_reorg(*fork_height).await;
+                tracing::info!(
+                    "FiltersManager: re-matching filters from effective floor {}",
+                    effective_floor
+                );
+                self.reset_for_reorg(effective_floor);
                 self.set_state(SyncState::WaitForEvents);
                 return Ok(vec![]);
             }