test: wallet, coverage for crash on dup block disconnection during un…

…clean shutdown

Co-authored-by: furszy <matiasfurszyfer@protonmail.com>

test/functional/wallet_reorgsrestore.py

@@ -31,6 +31,41 @@ def set_test_params(self):
     def skip_test_if_missing_module(self):
         self.skip_if_no_wallet()
 
+    def test_reorg_handling_during_unclean_shutdown(self):
+        self.log.info("Test crash due to a duplicate block disconnection event during an unclean shutdown")
+        node = self.nodes[0]
+        # Receive coinbase reward on a new wallet
+        node.createwallet(wallet_name="reorg_crash", load_on_startup=True)
+        wallet = node.get_wallet_rpc("reorg_crash")
+        self.generatetoaddress(node, 1, wallet.getnewaddress(), sync_fun=self.no_op)
+
+        # Restart to ensure node and wallet are flushed
+        self.restart_node(0)
+        wallet = node.get_wallet_rpc("reorg_crash")
+        assert_equal(wallet.getwalletinfo()['immature_balance'], 25)
+
+        # Disconnect tip
+        tip = wallet.getbestblockhash()
+        wallet.invalidateblock(tip)
+        wallet.syncwithvalidationinterfacequeue()
+
+        # Tip was disconnected, ensure coinbase has been abandoned
+        assert_equal(wallet.getwalletinfo()['immature_balance'], 0)
+        coinbase_tx_id = wallet.getblock(tip, verbose=1)["tx"][0]
+        assert_equal(wallet.gettransaction(coinbase_tx_id)['details'][0]['abandoned'], True)
+
+        # Abort process abruptly to force an unclean shutdown (no flush)
+        node.process.kill()
+
+        # Start wallet again. Ensure the tx is still abandoned and the wallet has no balance.
+        self.start_node(0)
+        wallet = node.get_wallet_rpc("reorg_crash")
+        assert_equal(wallet.getbestblockhash(), tip)
+        assert_equal(wallet.getwalletinfo()['immature_balance'], 0)
+        wallet.invalidateblock(tip)
+        # Ensure abandoned state persists after restart
+        assert_equal(wallet.gettransaction(coinbase_tx_id)['details'][0]['abandoned'], True)
+
     def run_test(self):
         # Send a tx from which to conflict outputs later
         txid_conflict_from = self.nodes[0].sendtoaddress(self.nodes[0].getnewaddress(), Decimal("10"))
@@ -100,5 +135,8 @@ def run_test(self):
         assert_equal(conflicted_after_reorg["confirmations"], 1)
         assert conflicting["blockhash"] != conflicted_after_reorg["blockhash"]
 
+        # Verify reorg behavior during an unclean shutdown
+        self.test_reorg_handling_during_unclean_shutdown()
+
 if __name__ == '__main__':
     ReorgsRestoreTest(__file__).main()