fix(utxo): mempool atomicity BUG-1 + BUG-4 (PR 6146 review)

node/utxo_db.py

@@ -714,8 +714,31 @@ def abort() -> bool:
                 ),
             )
 
+        # -- BUG-4: evict stale mempool txs referencing spent inputs ----
+        # Must run BEFORE COMMIT when the caller holds an external
+        # connection (manage_tx=False), because SQLite only allows
+        # one writer at a time. When we own the transaction
+        # (manage_tx=True), we commit first, then evict on a fresh
+        # connection so a failure cannot roll back the durable spend.
+            _spent_ids = list(set(input_box_ids + list(data_inputs)))
+            if _spent_ids:
+                if not manage_tx:
+                    # External-connection path: evict inside the caller's
+                    # transaction so the DELETEs share the same write lock.
+                    try:
+                        self._evict_stale_data_input_txs(_spent_ids, conn=conn)
+                    except Exception:
+                        pass  # best-effort; outer caller will commit the spend
             if manage_tx:
                 conn.execute("COMMIT")
+                # Own-transaction path: spend is committed. Evict on a
+                # separate connection - a failure here does not affect
+                # the committed transaction.
+                if _spent_ids:
+                    try:
+                        self._evict_stale_data_input_txs(_spent_ids)
+                    except Exception:
+                        pass  # best-effort; already committed
             return True
 
         except Exception:
@@ -729,6 +752,7 @@ def abort() -> bool:
             if own:
                 conn.close()
 
+
     # -- state root ----------------------------------------------------------
 
     def compute_state_root(self) -> str:
@@ -1025,19 +1049,111 @@ def mempool_add(self, tx: dict) -> bool:
             conn.close()
 
     def mempool_remove(self, tx_id: str):
-        """Remove a transaction from the mempool."""
+        """Remove a transaction from the mempool.
+
+        Uses BEGIN IMMEDIATE to ensure atomicity of the two DELETE
+        operations. Without it, a crash between deletes can leave
+        orphan utxo_mempool_inputs rows, causing a persistent UTXO
+        lock / mempool DoS (BUG-1).
+        """
         conn = self._conn()
         try:
+            conn.execute("BEGIN IMMEDIATE")
             conn.execute(
                 "DELETE FROM utxo_mempool_inputs WHERE tx_id = ?", (tx_id,)
             )
             conn.execute(
                 "DELETE FROM utxo_mempool WHERE tx_id = ?", (tx_id,)
             )
             conn.commit()
+        except Exception:
+            try:
+                conn.execute("ROLLBACK")
+            except Exception:
+                pass
+            raise
         finally:
             conn.close()
 
+    def _evict_stale_data_input_txs(self, spent_box_ids: List[str],
+                              conn: Optional[sqlite3.Connection] = None) -> int:
+        """Remove mempool txs whose inputs or data_inputs include any of spent_box_ids.
+
+        BUG-4 fix: apply_transaction() now proactively evicts mempool
+        transactions that became invalid because a box they depend on
+        (as a regular input or data_input) was just spent. Without this,
+        stale txs hold their normal inputs reserved in utxo_mempool_inputs
+        until candidate selection catches them — an availability gap.
+
+        Search strategy:
+        1. Check utxo_mempool_inputs for txs claiming any spent box as a
+           regular input.
+        2. Scan utxo_mempool.tx_data_json for txs whose data_inputs
+           reference any spent box (since data_inputs are not recorded
+           in utxo_mempool_inputs — they are read-only references).
+        """
+        if not spent_box_ids:
+            return 0
+        own_conn = conn is None
+        if own_conn:
+            conn = self._conn()
+        try:
+            spent_set = set(spent_box_ids)
+            stale_tx_ids = set()
+
+            # 1. Txs claiming spent boxes as regular inputs
+            placeholders = ",".join("?" for _ in spent_box_ids)
+            rows = conn.execute(
+                f"SELECT DISTINCT tx_id FROM utxo_mempool_inputs "
+                f"WHERE box_id IN ({placeholders})",
+                spent_box_ids,
+            ).fetchall()
+            for row in rows:
+                stale_tx_ids.add(row["tx_id"])
+
+            # 2. Txs referencing spent boxes as data_inputs
+            #    (not stored in utxo_mempool_inputs, so parse tx_data_json)
+            all_mempool = conn.execute(
+                "SELECT tx_id, tx_data_json FROM utxo_mempool"
+            ).fetchall()
+            for mp_row in all_mempool:
+                if mp_row["tx_id"] in stale_tx_ids:
+                    continue  # already flagged
+                try:
+                    tx_data = json.loads(mp_row["tx_data_json"])
+                    di = tx_data.get("data_inputs", [])
+                    if di and spent_set & set(di):
+                        stale_tx_ids.add(mp_row["tx_id"])
+                except (json.JSONDecodeError, TypeError):
+                    continue
+
+            if not stale_tx_ids:
+                return 0
+
+            tx_ids = list(stale_tx_ids)
+            tx_placeholders = ",".join("?" for _ in tx_ids)
+            conn.execute(
+                f"DELETE FROM utxo_mempool_inputs WHERE tx_id IN ({tx_placeholders})",
+                tx_ids,
+            )
+            conn.execute(
+                f"DELETE FROM utxo_mempool WHERE tx_id IN ({tx_placeholders})",
+                tx_ids,
+            )
+            if own_conn:
+                conn.commit()
+            return len(tx_ids)
+        except Exception:
+            if own_conn:
+                try:
+                    conn.execute("ROLLBACK")
+                except Exception:
+                    pass
+            return 0
+        finally:
+            if own_conn:
+                conn.close()
+
     def mempool_get_block_candidates(self, max_count: int = 100) -> List[dict]:
         """Get highest-fee transactions from mempool for block inclusion."""
         self.mempool_clear_expired()

tests/test_bug4_external_conn.py

@@ -0,0 +1,189 @@
+"""Regression test: BUG-4 stale eviction must work via external-connection path.
+
+The utxo_endpoints.py transfer endpoint opens BEGIN IMMEDIATE on its own
+connection, then calls apply_transaction(conn=outer_conn). The eviction must
+run on that same connection inside the transaction.
+"""
+import json, time, sqlite3, pytest
+from node.utxo_db import UtxoDB
+
+
+@pytest.fixture
+def db(tmp_path):
+    db_path = str(tmp_path / "test_utxo.db")
+    instance = UtxoDB(db_path)
+    instance.init_tables()
+    return instance
+
+
+def _mint(db, addr, value, height=1):
+    """Mint a box and return its box_id."""
+    result = db.apply_transaction({
+        "tx_type": "mining_reward",
+        "inputs": [],
+        "outputs": [{"address": addr, "value_nrtc": value}],
+        "fee_nrtc": 0,
+        "data_inputs": [],
+        "_allow_minting": True,
+    }, block_height=height)
+    assert result, "minting failed"
+    conn = db._conn()
+    try:
+        row = conn.execute(
+            "SELECT box_id FROM utxo_boxes WHERE owner_address=? AND spent_at IS NULL ORDER BY box_id ASC LIMIT 1",
+            (addr,),
+        ).fetchone()
+    finally:
+        conn.close()
+    assert row, f"no box for {addr}"
+    return row["box_id"]
+
+
+def _add_mempool_tx_with_data_input(db, tx_id, input_box_ids, data_input_box_ids, fee=100):
+    """Directly insert a mempool tx with data_inputs in tx_data_json."""
+    now = int(time.time())
+    tx_data = {
+        "tx_id": tx_id,
+        "data_inputs": data_input_box_ids,
+    }
+    conn = db._conn()
+    try:
+        conn.execute("BEGIN IMMEDIATE")
+        conn.execute(
+            "INSERT INTO utxo_mempool (tx_id, tx_data_json, fee_nrtc, expires_at, submitted_at) VALUES (?,?,?,?,?)",
+            (tx_id, json.dumps(tx_data), fee, now + 3600, now),
+        )
+        for bid in input_box_ids:
+            conn.execute(
+                "INSERT INTO utxo_mempool_inputs (box_id, tx_id) VALUES (?,?)",
+                (bid, tx_id),
+            )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+class TestExternalConnEvictionBug4:
+    """BUG-4 fix: stale eviction must work when apply_transaction() is called
+    with an external connection (the utxo_endpoints.py transfer pattern)."""
+
+    def test_stale_data_input_tx_evicted_via_external_conn(self, db):
+        """Stale mempool tx referencing a spent box via data_input must be
+        evicted even when the spend happens via an external connection
+        (BEGIN IMMEDIATE on outer, apply_transaction(conn=outer), COMMIT)."""
+        # 1. Mint two boxes
+        box_a = _mint(db, "alice", 10000)
+        box_b = _mint(db, "bob", 20000, height=2)
+
+        # 2. Add mempool tx that uses box_a as data_input
+        _add_mempool_tx_with_data_input(
+            db, "tx_stale_di", [box_b], [box_a], fee=50
+        )
+
+        # Verify tx is in mempool
+        conn = db._conn()
+        try:
+            row = conn.execute(
+                "SELECT tx_id FROM utxo_mempool WHERE tx_id=?", ("tx_stale_di",)
+            ).fetchone()
+        finally:
+            conn.close()
+        assert row is not None, "stale tx should be in mempool"
+
+        # 3. Spend box_a via EXTERNAL CONNECTION path
+        outer = sqlite3.connect(db.db_path)
+        outer.row_factory = sqlite3.Row
+        outer.execute("BEGIN IMMEDIATE")
+
+        ok = db.apply_transaction({
+            "tx_type": "transfer",
+            "inputs": [{"box_id": box_a, "spending_proof": "p2"}],
+            "outputs": [{"address": "carol", "value_nrtc": 9900}],
+            "fee_nrtc": 100,
+            "data_inputs": [],
+        }, block_height=3, conn=outer)
+
+        outer.execute("COMMIT")
+        outer.close()
+        assert ok, "apply_transaction via external conn failed"
+
+        # 4. Verify stale tx was evicted
+        conn = db._conn()
+        try:
+            mp = conn.execute(
+                "SELECT tx_id FROM utxo_mempool WHERE tx_id=?", ("tx_stale_di",)
+            ).fetchone()
+            inp = conn.execute(
+                "SELECT tx_id FROM utxo_mempool_inputs WHERE tx_id=?", ("tx_stale_di",)
+            ).fetchone()
+        finally:
+            conn.close()
+        assert mp is None, "BUG-4: stale tx should be evicted via external-conn path"
+        assert inp is None, "BUG-4: stale input rows should be cleaned up"
+
+    def test_stale_regular_input_tx_evicted_via_external_conn(self, db):
+        """Stale mempool tx claiming a spent box as regular input must be
+        evicted via the external-connection path."""
+        box_a = _mint(db, "alice", 10000)
+
+        # Add mempool tx claiming box_a
+        _add_mempool_tx_with_data_input(
+            db, "tx_stale_reg", [box_a], [], fee=50
+        )
+
+        # Spend box_a via external connection
+        outer = sqlite3.connect(db.db_path)
+        outer.row_factory = sqlite3.Row
+        outer.execute("BEGIN IMMEDIATE")
+
+        ok = db.apply_transaction({
+            "tx_type": "transfer",
+            "inputs": [{"box_id": box_a, "spending_proof": "p2"}],
+            "outputs": [{"address": "carol", "value_nrtc": 9900}],
+            "fee_nrtc": 100,
+            "data_inputs": [],
+        }, block_height=2, conn=outer)
+
+        outer.execute("COMMIT")
+        outer.close()
+        assert ok
+
+        # Verify eviction
+        conn = db._conn()
+        try:
+            mp = conn.execute(
+                "SELECT tx_id FROM utxo_mempool WHERE tx_id=?", ("tx_stale_reg",)
+            ).fetchone()
+        finally:
+            conn.close()
+        assert mp is None, "stale tx with regular input should be evicted"
+
+    def test_own_conn_eviction_still_works(self, db):
+        """Regression guard: own-connection (manage_tx=True) path still
+        evicts stale txs after the fix."""
+        box_a = _mint(db, "alice", 10000)
+        box_b = _mint(db, "bob", 20000, height=2)
+
+        _add_mempool_tx_with_data_input(
+            db, "tx_stale_own", [box_b], [box_a], fee=50
+        )
+
+        # Own-connection path (no conn=... argument)
+        ok = db.apply_transaction({
+            "tx_type": "transfer",
+            "inputs": [{"box_id": box_a, "spending_proof": "p2"}],
+            "outputs": [{"address": "carol", "value_nrtc": 9900}],
+            "fee_nrtc": 100,
+            "data_inputs": [],
+        }, block_height=3)
+        assert ok
+
+        # Verify eviction
+        conn = db._conn()
+        try:
+            mp = conn.execute(
+                "SELECT tx_id FROM utxo_mempool WHERE tx_id=?", ("tx_stale_own",)
+            ).fetchone()
+        finally:
+            conn.close()
+        assert mp is None, "own-connection eviction should still work"

tests/test_p2p_blocks.py

@@ -0,0 +1 @@
+import pytest