tests: toyserver: add tests, implement mempool replacement

diff --git a/tests/toyserver/test_toyserver.py b/tests/toyserver/test_toyserver.py
new file mode 100644
index 0000000..5d23e6e
--- /dev/null
+++ b/tests/toyserver/test_toyserver.py
@@ -0,0 +1,164 @@
+import electrum_ecc as ecc
+
+from electrum.address_synchronizer import TX_HEIGHT_UNCONFIRMED
+from electrum.bitcoin import COIN, construct_script, opcodes
+from electrum.fee_policy import FixedFeePolicy
+from electrum.simple_config import SimpleConfig
+from electrum.transaction import PartialTxInput, PartialTxOutput, TxOutput
+from electrum.wallet import Abstract_Wallet
+
+from .. import ElectrumTestCase
+from .. import restore_wallet_from_text__for_unittest
+from .toyserver import ToyServer, topologically_sort_subgraph, TxConflicts, TxConflictsBlockchain
+
+
+class TestToyServer(ElectrumTestCase):
+    REGTEST = True
+
+    async def asyncSetUp(self):
+        await super().asyncSetUp()
+        self.config = SimpleConfig({'electrum_path': self.electrum_path})
+        self.toyserver = ToyServer()
+        await self.toyserver.start()
+        assert self.toyserver.cur_height == 0
+        for _ in range(10):  # mine some blocks
+            await self.toyserver.mine_block()
+        await self.toyserver.set_up_faucet(config=self.config)
+
+    async def asyncTearDown(self):
+        await self.toyserver.stop()
+        await super().asyncTearDown()
+
+    async def test_topological_sort(self):
+        graph = {
+            "A": ["B", "C"],
+            "B": ["X"],
+            "C": ["D1"],
+            "D1": ["D2"],
+            "D2": ["D3"],
+            "D3": ["D4"],
+            "D4": ["X"],
+            "X": ["Y"],
+            "Y": [],
+        }
+        get_direct_children = lambda x: graph[x]
+        # note: there are multiple valid orderings, not just the one we assert
+        self.assertEqual(
+            ["A", "C", "D1", "D2", "D3", "D4", "B", "X", "Y"],
+            topologically_sort_subgraph(["A"], get_direct_children=get_direct_children))
+        self.assertEqual(
+            ["A", "C", "D1", "D2", "D3", "D4", "B", "X", "Y"],
+            topologically_sort_subgraph(list(graph), get_direct_children=get_direct_children))
+
+        graph = {
+            "B": ["X"],
+            "C": ["D1"],
+            "D1": ["D2"],
+            "D2": ["D3"],
+            "D3": ["D4"],
+            "D4": ["X"],
+            "X": ["Y"],
+            "Y": [],
+        }
+        self.assertEqual(
+            ["C", "D1", "D2", "D3", "D4", "B", "X", "Y"],
+            topologically_sort_subgraph(["B", "C"], get_direct_children=get_direct_children))
+
+
+    async def test_basic_mempool_and_mining_txs(self):
+        server_height = self.toyserver.cur_height
+        self.assertEqual(self.toyserver.get_mempool_txids(), set())
+        secret_key = 0
+        for cycle in range(2):
+            mempool_txids = set()
+            # populate mempool
+            for _ in range(5):
+                secret_key += 1
+                spk = construct_script([secret_key * ecc.GENERATOR.get_public_key_bytes(compressed=True), opcodes.OP_CHECKSIG])
+                txout = TxOutput(scriptpubkey=spk, value=1 * COIN)
+                tx = await self.toyserver.ask_faucet([txout])
+                mempool_txids.add(tx.txid())
+                self.assertEqual(mempool_txids, self.toyserver.get_mempool_txids())
+                self.assertEqual(None, self.toyserver.block_height_from_txid(tx.txid()))
+            # mine a block
+            await self.toyserver.mine_block()
+            server_height += 1
+            self.assertEqual(server_height, self.toyserver.cur_height)
+            self.assertEqual(set(), self.toyserver.get_mempool_txids())
+            for txid in mempool_txids:  # old mempool
+                self.assertEqual(server_height, self.toyserver.block_height_from_txid(txid))
+
+    async def test_mempool_replacement(self):
+        self.assertEqual(self.toyserver.get_mempool_txids(), set())
+
+        w = restore_wallet_from_text__for_unittest("9dk", path=None, config=self.config, gap_limit=4)['wallet']  # type: Abstract_Wallet
+        # fund w
+        w_addr0 = w.get_receiving_addresses()[0]
+        funding_tx = await self.toyserver.ask_faucet([TxOutput.from_address_and_value(w_addr0, 20 * COIN)])
+        w.adb.add_transaction(funding_tx)
+        self.assertEqual(self.toyserver.get_mempool_txids(), {funding_tx.txid()})
+        # wallet sends all to itself on fresh address: tx1
+        w_addr1 = w.get_receiving_addresses()[1]
+        tx1 = w.make_unsigned_transaction(
+            outputs=[PartialTxOutput.from_address_and_value(w_addr1, "!")], fee_policy=FixedFeePolicy(5000))
+        w.sign_transaction(tx1, password=None)
+        w.adb.add_transaction(tx1)
+        self.assertEqual(self.toyserver._get_transitive_conflict_txids(tx1), set())
+        await self.toyserver.mempool_add_tx(tx1)
+        self.assertEqual(self.toyserver.get_mempool_txids(), {funding_tx.txid(), tx1.txid()})
+        self.assertEqual(self.toyserver._get_transitive_conflict_txids(tx1, include_self=False), set())
+        self.assertEqual(self.toyserver._get_transitive_conflict_txids(tx1, include_self=True), {tx1.txid()})
+        # wallet sends all to itself on fresh address: tx2
+        w_addr2 = w.get_receiving_addresses()[2]
+        tx2 = w.make_unsigned_transaction(
+            outputs=[PartialTxOutput.from_address_and_value(w_addr2, "!")], fee_policy=FixedFeePolicy(5000))
+        w.sign_transaction(tx2, password=None)
+        w.adb.add_transaction(tx2)
+        self.assertEqual(self.toyserver._get_transitive_conflict_txids(tx2), set())
+        await self.toyserver.mempool_add_tx(tx2)
+        self.assertEqual(self.toyserver.get_mempool_txids(), {funding_tx.txid(), tx1.txid(), tx2.txid()})
+
+        self.assertEqual(len(w.adb.get_history(w.get_addresses())), 3)
+
+        # -- wallet wants to double-spend tx1 (also invalidating tx2)
+        # first, wallet tries tx1b, but uses too low fee
+        w.adb.remove_transaction(tx1.txid())
+        self.assertEqual(len(w.adb.get_history(w.get_addresses())), 1)
+        w_addr3 = w.get_receiving_addresses()[3]
+        tx1b = w.make_unsigned_transaction(
+            outputs=[PartialTxOutput.from_address_and_value(w_addr3, "!")], fee_policy=FixedFeePolicy(5000))
+        w.sign_transaction(tx1b, password=None)
+        w.adb.add_transaction(tx1b)
+        self.assertEqual(len(w.adb.get_history(w.get_addresses())), 2)
+        self.assertEqual(self.toyserver._get_transitive_conflict_txids(tx1b), {tx1.txid(), tx2.txid()})
+        with self.assertRaises(TxConflicts):
+            await self.toyserver.mempool_add_tx(tx1b)
+        self.assertEqual(self.toyserver.get_mempool_txids(), {funding_tx.txid(), tx1.txid(), tx2.txid()})
+
+        # second, wallet tries tx1c, which pays high enough fees for replacement
+        w.adb.remove_transaction(tx1b.txid())
+        tx1c = w.make_unsigned_transaction(
+            outputs=[PartialTxOutput.from_address_and_value(w_addr3, "!")], fee_policy=FixedFeePolicy(10_001))
+        w.sign_transaction(tx1c, password=None)
+        w.adb.add_transaction(tx1c)
+        self.assertEqual(self.toyserver._get_transitive_conflict_txids(tx1c), {tx1.txid(), tx2.txid()})
+        await self.toyserver.mempool_add_tx(tx1c)
+        self.assertEqual(self.toyserver.get_mempool_txids(), {funding_tx.txid(), tx1c.txid()})
+
+        # mine a block
+        await self.toyserver.mine_block()
+        self.assertEqual(self.toyserver.get_mempool_txids(), set())
+        self.assertEqual(self.toyserver.cur_height, self.toyserver.block_height_from_txid(funding_tx.txid()))
+        self.assertEqual(self.toyserver.cur_height, self.toyserver.block_height_from_txid(tx1c.txid()))
+
+        # -- wallet wants to double-spend tx1c - but it is already mined!
+        w.adb.remove_transaction(tx1c.txid())
+        tx1d = w.make_unsigned_transaction(
+            outputs=[PartialTxOutput.from_address_and_value(w_addr3, "!")], fee_policy=FixedFeePolicy(25_000))
+        w.sign_transaction(tx1d, password=None)
+        w.adb.add_transaction(tx1d)
+        self.assertEqual(len(w.adb.get_history(w.get_addresses())), 2)
+        self.assertEqual(self.toyserver._get_transitive_conflict_txids(tx1d), {tx1c.txid()})
+        with self.assertRaises(TxConflictsBlockchain):
+            await self.toyserver.mempool_add_tx(tx1d)
+        self.assertEqual(self.toyserver.get_mempool_txids(), set())
diff --git a/tests/toyserver/toyserver.py b/tests/toyserver/toyserver.py
index 71281c4..4edf031 100644
--- a/tests/toyserver/toyserver.py
+++ b/tests/toyserver/toyserver.py
@@ -33,23 +33,29 @@ REGTEST_GENESIS_HEADER = bfh("01000000000000000000000000000000000000000000000000
 T = TypeVar("T")
 
 def topologically_sort_subgraph(
-    start_node: T,
+    start_nodes: Iterable[T],
     *,
     get_direct_children: Callable[[T], Iterable[T]],
 ) -> Sequence[T]:
-    children = []
-    seen = set()
+    # based on pseudo-code in https://en.wikipedia.org/wiki/Topological_sorting#Depth-first_search
+    res = OrderedSet()  # "permanent mark"
+    seen = set()        # "temporary mark"
 
     def recurse(node: str):
+        if node in res:
+            return
+        if node in seen:
+            raise Exception("cycle detected")
+        seen.add(node)
         direct_children = get_direct_children(node)
         for child in direct_children:
-            if child not in seen:
-                seen.add(child)
-                recurse(child)
-        children.append(node)
+            recurse(child)
+        res.add(node)
 
-    recurse(start_node)
-    return children[::-1]
+    for start_node in start_nodes:
+        if start_node not in res:
+            recurse(start_node)
+    return list(res)[::-1]
 
 
 
@@ -63,13 +69,20 @@ class FakeBlock:
             object.__setattr__(self, 'txids', OrderedSet())
 
 
-class ToyServer:
+class TxConflicts(Exception): pass
+class TxConflictsMempool(TxConflicts): pass
+class TxConflictsBlockchain(TxConflicts): pass
+
+
+class ToyServer(Logger):
     """Electrum Server backend"""
 
     asyncio_server: asyncio.base_events.Server
     server_port: int
+    min_relay_feerate = 2000  # in satoshi per kvbyte
 
     def __init__(self):
+        Logger.__init__(self)
         self.sessions = set()  # type: Set[ToyServerSession]
         self._blocks = [FakeBlock(header=REGTEST_GENESIS_HEADER)]  # type: list[FakeBlock]
 
@@ -165,6 +178,14 @@ class ToyServer:
     def _add_tx(self, tx: Transaction) -> set[str]:
         txid = tx.txid()
         assert txid
+        if txid in self.txs:  # already added
+            funded_sh, spent_sh = self._get_funded_and_spent_scripthashes(tx)
+            return funded_sh | spent_sh
+        # we forbid conflicting txs. for mempool replacement, the caller must already have rm-ed the conflicts.
+        conflict_txids = self._get_transitive_conflict_txids(tx)
+        assert not conflict_txids, "tx conflict"
+        self.logger.debug(f"_add_tx: {txid}")
+        # update txid->tx map
         self.txs[txid] = bfh(str(tx))
         # fund UTXOs
         for txout_idx, txout in enumerate(tx.outputs()):
@@ -176,13 +197,13 @@ class ToyServer:
                 continue
             double_spender_txid = self.txo_to_spender_txid.get(txin.prevout, ...)
             if double_spender_txid is ...:
-                raise RPCError(DAEMON_ERROR, f"cannot spend non-existent UTXO: {txin.prevout}")
+                raise Exception(f"cannot spend non-existent UTXO: {txin.prevout}")
             elif double_spender_txid is None:  # UTXO exists and is unspent
                 self.txo_to_spender_txid[txin.prevout] = txid
-            elif double_spender_txid == txid:  # already marked?
-                pass                           # (duplicate calls, e.g. when added to mempool, and when mined)
-            else:  # conflict
-                raise RPCError(DAEMON_ERROR, f"cannot double-spend UTXO: {txin.prevout}. conflict: {txid} vs {double_spender_txid}")
+            elif double_spender_txid == txid:
+                raise Exception("TXO already marked as spent by same txid?")
+            else:
+                raise Exception(f"cannot double-spend UTXO: {txin.prevout}. conflict: {txid} vs {double_spender_txid}")
         # update touched scripthashes
         funded_sh, spent_sh = self._get_funded_and_spent_scripthashes(tx)
         for sh in funded_sh:
@@ -194,6 +215,7 @@ class ToyServer:
     def _remove_tx_that_has_no_children(self, tx: Transaction) -> set[str]:
         txid = tx.txid()
         assert txid
+        self.logger.debug(f"_remove_tx_that_has_no_children: {txid}")
         assert self.block_height_from_txid(txid) is None, "tx already mined"
         self.txs.pop(txid)
         # un-fund UTXOs
@@ -218,6 +240,7 @@ class ToyServer:
     def _remove_tx_and_all_children(self, tx: Transaction) -> set[str]:
         txid = tx.txid()
         assert txid
+        assert txid in self.txs, "unknown tx"
         children = self._get_transitive_children_txids(txid)
         touched_sh = set()
         for txid in children:
@@ -238,9 +261,9 @@ class ToyServer:
         """Returns all (grand-)children, including orig tx.
         Topologically sorted, children first.
         """
-        return topologically_sort_subgraph(txid, get_direct_children=self._get_direct_children_txids)[::-1]
+        return topologically_sort_subgraph([txid], get_direct_children=self._get_direct_children_txids)[::-1]
 
-    def _get_direct_conflicts(self, tx: Transaction) -> Sequence[str]:
+    def _get_direct_conflict_txids(self, tx: Transaction, *, include_self: bool = True) -> Sequence[str]:
         txid = tx.txid()
         assert txid
         res = []
@@ -248,17 +271,52 @@ class ToyServer:
             if txin.is_coinbase_input():
                 continue
             if double_spender_txid := self.txo_to_spender_txid.get(txin.prevout, None):
-                res.append(double_spender_txid)
+                if double_spender_txid != txid or include_self:
+                    res.append(double_spender_txid)
         return res
 
-    def _get_transitive_conflicts(self, tx: Transaction) -> Iterable[str]:
+    def _get_transitive_conflict_txids(self, tx: Transaction, *, include_self: bool = True) -> Iterable[str]:
         res = set()
-        for direct_conflict in self._get_direct_conflicts(tx):
-            res |= self._get_transitive_children_txids(direct_conflict)
+        for direct_conflict in self._get_direct_conflict_txids(tx, include_self=include_self):
+            res |= set(self._get_transitive_children_txids(direct_conflict))
         return res
 
-    async def mempool_add_tx(self, tx: Transaction) -> None:
-        touched_sh = self._add_tx(tx)
+    def _txs_from_txids(self, txids: Iterable[str]) -> Iterable[Transaction]:
+        return [Transaction(self.txs[txid]) for txid in txids]
+
+    def _get_fee_sat_paid_by_tx(self, tx: Transaction) -> int:
+        input_sum = 0
+        for txin in tx.inputs():
+            parent_tx_raw = self.txs[txin.prevout.txid.hex()]  # parent must not be missing!
+            parent_tx = Transaction(parent_tx_raw)
+            ptxout = parent_tx.outputs()[txin.prevout.out_idx]
+            input_sum += ptxout.value
+        return input_sum - tx.output_value()
+
+    async def mempool_add_tx(self, newtx: Transaction) -> None:
+        touched_sh = set()
+        conflict_txids = self._get_transitive_conflict_txids(newtx, include_self=False)
+        conflict_txs = self._txs_from_txids(conflict_txids)
+        if conflict_txids:
+            if any(self.block_height_from_txid(txid) is not None for txid in conflict_txids):
+                raise TxConflictsBlockchain()
+            conflict_wu = sum(tx.estimated_weight() for tx in conflict_txs)
+            conflict_fee = sum(self._get_fee_sat_paid_by_tx(tx) for tx in conflict_txs)
+            conflict_sat_per_kvbyte = 4000 * conflict_fee // conflict_wu
+            repl_fee = self._get_fee_sat_paid_by_tx(newtx)
+            repl_sat_per_kvbyte = 4000 * repl_fee // newtx.estimated_weight()
+            # our mempool replacement policy is simple but still similar to bitcoin core:
+            if not (
+                repl_fee > conflict_fee
+                and repl_sat_per_kvbyte >= conflict_sat_per_kvbyte + self.min_relay_feerate
+            ):
+                raise TxConflictsMempool(f"mempool conflict. {len(conflict_txs)=}. {repl_fee=}, {conflict_fee=}. {repl_sat_per_kvbyte=}, {conflict_sat_per_kvbyte=}")
+            # rm conflicts
+            for tx in conflict_txs:
+                if tx.txid() in self.txs:  # might already be removed in an earlier loop iter
+                    touched_sh = self._remove_tx_and_all_children(tx)
+        # no more conflicts. add new tx.
+        touched_sh |= self._add_tx(newtx)
         # notify clients
         for session in self.sessions:
             await session.server_send_notifications(touched_sh=touched_sh)
@@ -285,6 +343,7 @@ class ToyServer:
         coinbase_tx = Transaction(None)
         coinbase_tx._inputs = [TxInput(prevout=TxOutpoint(txid=bytes(32), out_idx=0xffffffff))]
         coinbase_tx._outputs = list(coinbase_outputs or []) + [TxOutput(scriptpubkey=bfh("6a04deadbeef"), value=0)]
+        coinbase_tx._locktime = self.cur_height  # to prevent duplicate txids (our fake coinbase txs are low-entropy)
         txs = OrderedSet()  # type: OrderedSet[Transaction]
         txs.add(coinbase_tx)
         txs |= OrderedSet(extra_txs)
@@ -417,9 +476,9 @@ class ToyServerSession(aiorpcx.RPCSession, Logger):
 
     async def _handle_mempool_get_info(self):
         return {
-            "mempoolminfee": 0.00001000,
-            "minrelaytxfee": 0.00001000,
-            "incrementalrelayfee": 0.00001000,
+            "mempoolminfee": self.svr.min_relay_feerate / COIN,
+            "minrelaytxfee": self.svr.min_relay_feerate / COIN,
+            "incrementalrelayfee": self.svr.min_relay_feerate / COIN,
         }
 
     def _get_headersub_result(self):
@@ -475,7 +534,10 @@ class ToyServerSession(aiorpcx.RPCSession, Logger):
     async def _handle_transaction_broadcast(self, raw_tx: str) -> str:
         tx = Transaction(raw_tx)
         txid = tx.txid()
-        await self.svr.mempool_add_tx(tx)  # TODO don't await, just queue up? this sends notifs before response, lol
+        try:
+            await self.svr.mempool_add_tx(tx)  # TODO don't await, just queue up? this sends notifs before response, lol
+        except TxConflicts as e:
+            raise RPCError(DAEMON_ERROR, str(e)) from e
         return txid
 
     async def _handle_scripthash_subscribe(self, sh: str) -> Optional[str]: