From c0c76ea57b38aa434dadebb26b3473a6e5c9d29b Mon Sep 17 00:00:00 2001
From: nekwasar <nekwasar@users.noreply.github.com>
Date: Sun, 26 Jul 2026 21:37:05 +0000
Subject: [PATCH] feat(network): add automated transaction resubmission with
 gas escalation (#661)

---
 src/network/nonce_tracker.py | 208 +++++++++++++++++++++++++++++++++++
 tests/test_nonce_tracker.py  | 198 +++++++++++++++++++++++++++++++++
 2 files changed, 406 insertions(+)

diff --git a/src/network/nonce_tracker.py b/src/network/nonce_tracker.py
index ecbd1e74..55219e56 100644
--- a/src/network/nonce_tracker.py
+++ b/src/network/nonce_tracker.py
@@ -847,6 +847,212 @@ def reconcile(self, address: str) -> ReconciliationResult:
 nonce_window = NonceWindow()
 
 
+# =========================================================================
+# Transaction Resubmission Manager with Gas Escalation
+# =========================================================================
+
+
+@dataclass
+class PendingSubmission:
+    """Tracks a submitted transaction awaiting confirmation.
+
+    Attributes
+    ----------
+    tx_id:
+        Unique transaction identifier (e.g. hash).
+    base_fee:
+        The base fee in stroops for this submission attempt.
+    submitted_at:
+        Monotonic timestamp of submission.
+    attempts:
+        Number of resubmission attempts so far.
+    """
+
+    tx_id: str
+    base_fee: int
+    submitted_at: float = field(default_factory=time.monotonic)
+    attempts: int = 0
+
+
+class TransactionResubmitter:
+    """Automated transaction resubmission manager with fee escalation.
+
+    Monitors pending transactions and resubmits them with escalated fees
+    when they remain unconfirmed beyond a configurable timeout.  Fee
+    escalation follows a multiplicative strategy: each attempt increases
+    the base fee by *escalation_factor* until *max_fee* is reached.
+
+    The resubmitter does **not** dispatch transactions itself — instead it
+    calls a user-provided *resubmit_fn* callback that is responsible for
+    building and broadcasting the replacement transaction with the new fee.
+
+    Parameters
+    ----------
+    initial_fee:
+        Starting base fee in stroops (default: 100 = 0.00001 XLM).
+    escalation_factor:
+        Multiplier applied per attempt (default: 1.5 → 50% increase).
+    max_fee:
+        Ceiling in stroops beyond which fee will not escalate (default: 10000).
+    resubmit_timeout:
+        Seconds since submission before a pending tx is escalated (default: 30).
+    resubmit_fn:
+        Async callback ``(tx_id: str, new_fee: int) -> bool`` that performs
+        the actual resubmission.  Must return ``True`` on success.
+
+    Example usage::
+
+        async def resubmit(tx_id: str, fee: int) -> bool:
+            tx = build_replacement(tx_id, fee)
+            return await broadcast(tx)
+
+        submitter = TransactionResubmitter(resubmit_fn=resubmit)
+        submitter.track("tx-hash-123", base_fee=100)
+
+        # Called periodically:
+        await submitter.escalate_pending()
+    """
+
+    def __init__(
+        self,
+        initial_fee: int = 100,
+        escalation_factor: float = 1.5,
+        max_fee: int = 10_000,
+        resubmit_timeout: float = 30.0,
+        resubmit_fn: Optional[Callable[[str, int], bool]] = None,
+    ) -> None:
+        if initial_fee < 100:
+            raise ValueError("initial_fee must be at least 100 stroops")
+        if escalation_factor <= 1.0:
+            raise ValueError("escalation_factor must be > 1.0")
+        if max_fee < initial_fee:
+            raise ValueError("max_fee must be >= initial_fee")
+        if resubmit_timeout <= 0:
+            raise ValueError("resubmit_timeout must be > 0")
+
+        self._initial_fee = initial_fee
+        self._escalation_factor = escalation_factor
+        self._max_fee = max_fee
+        self._resubmit_timeout = resubmit_timeout
+        self._resubmit_fn = resubmit_fn
+
+        self._lock = threading.Lock()
+        self._pending: Dict[str, PendingSubmission] = {}
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    def track(self, tx_id: str, base_fee: Optional[int] = None) -> None:
+        """Register *tx_id* for automated escalation monitoring.
+
+        Parameters
+        ----------
+        tx_id:
+            Unique transaction identifier.
+        base_fee:
+            Initial base fee in stroops. Defaults to *initial_fee*.
+        """
+        with self._lock:
+            if tx_id in self._pending:
+                logger.debug("[TxResubmitter] %s already tracked — skipping", tx_id)
+                return
+            self._pending[tx_id] = PendingSubmission(
+                tx_id=tx_id,
+                base_fee=base_fee if base_fee is not None else self._initial_fee,
+            )
+            logger.info("[TxResubmitter] Tracking %s (fee=%d)", tx_id, base_fee or self._initial_fee)
+
+    def untrack(self, tx_id: str) -> None:
+        """Remove *tx_id* from escalation monitoring (e.g. on confirmation).
+
+        Parameters
+        ----------
+        tx_id:
+            Transaction identifier to stop tracking.
+        """
+        with self._lock:
+            self._pending.pop(tx_id, None)
+            logger.info("[TxResubmitter] Stopped tracking %s", tx_id)
+
+    def set_resubmit_fn(self, fn: Callable[[str, int], bool]) -> None:
+        """Set or replace the resubmission callback."""
+        self._resubmit_fn = fn
+
+    async def escalate_pending(self) -> List[str]:
+        """Check all pending transactions and escalate those past the timeout.
+
+        For each pending transaction whose age exceeds *resubmit_timeout*,
+        computes the escalated fee and calls *resubmit_fn*.
+
+        Returns
+        -------
+        List[str]
+            Transaction IDs that were escalated (or attempted).
+        """
+        now = time.monotonic()
+        candidates: List[PendingSubmission] = []
+
+        with self._lock:
+            for tx_id, sub in list(self._pending.items()):
+                age = now - sub.submitted_at
+                if age >= self._resubmit_timeout:
+                    candidates.append(sub)
+
+        escalated: List[str] = []
+        for sub in candidates:
+            new_fee = self._compute_escalated_fee(sub)
+            sub.attempts += 1
+            sub.submitted_at = time.monotonic()
+
+            try:
+                if self._resubmit_fn is not None:
+                    success = self._resubmit_fn(sub.tx_id, new_fee)
+                    if success:
+                        sub.base_fee = new_fee
+                        escalated.append(sub.tx_id)
+                        logger.info(
+                            "[TxResubmitter] Resubmitted %s (attempt=%d, fee=%d)",
+                            sub.tx_id, sub.attempts, new_fee,
+                        )
+                    else:
+                        logger.warning(
+                            "[TxResubmitter] Resubmit callback returned False for %s",
+                            sub.tx_id,
+                        )
+            except Exception as exc:
+                logger.error(
+                    "[TxResubmitter] Resubmit failed for %s: %s",
+                    sub.tx_id, exc,
+                )
+
+        return escalated
+
+    def get_pending_count(self) -> int:
+        """Return the number of currently tracked pending transactions."""
+        with self._lock:
+            return len(self._pending)
+
+    def get_pending_ids(self) -> List[str]:
+        """Return a snapshot of all tracked transaction IDs."""
+        with self._lock:
+            return list(self._pending.keys())
+
+    # ------------------------------------------------------------------
+    # Internal helpers
+    # ------------------------------------------------------------------
+
+    def _compute_escalated_fee(self, sub: PendingSubmission) -> int:
+        """Calculate the next fee for *sub* using multiplicative escalation.
+
+        The formula is::
+
+            new_fee = min(current_fee * escalation_factor, max_fee)
+        """
+        raw = int(sub.base_fee * self._escalation_factor)
+        return min(raw, self._max_fee)
+
+
 class RPCNodeFailoverSupervisor:
     """Proactive RPC node failover supervisor that monitors node connectivity.
 
@@ -1001,6 +1207,8 @@ def _run_monitor(self) -> None:
     "NonceWindow",
     "NonceGapDetector",
     "NonceRecoveryEngine",
+    "TransactionResubmitter",
+    "PendingSubmission",
     "GapReport",
     "ReconciliationResult",
     "nonce_tracker",
diff --git a/tests/test_nonce_tracker.py b/tests/test_nonce_tracker.py
index a2d06211..b494db94 100644
--- a/tests/test_nonce_tracker.py
+++ b/tests/test_nonce_tracker.py
@@ -16,6 +16,8 @@
     NonceTracker,
     NonceWindow,
     ReconciliationResult,
+    TransactionResubmitter,
+    PendingSubmission,
 )
 
 
@@ -927,3 +929,199 @@ def test_gap_report_stale_equals_current_is_not_a_gap() -> None:
     )
     assert not report.has_gaps
 
+
+# =========================================================================
+# Gas Escalation Resubmission Tests
+# =========================================================================
+
+
+def test_gas_escalation_resubmission_construction_defaults() -> None:
+    """TransactionResubmitter constructs with default parameters."""
+    submitter = TransactionResubmitter()
+    assert submitter._initial_fee == 100
+    assert submitter._escalation_factor == 1.5
+    assert submitter._max_fee == 10_000
+    assert submitter._resubmit_timeout == 30.0
+    assert submitter.get_pending_count() == 0
+
+
+def test_gas_escalation_resubmission_custom_params() -> None:
+    """TransactionResubmitter accepts custom parameters."""
+    submitter = TransactionResubmitter(
+        initial_fee=200,
+        escalation_factor=2.0,
+        max_fee=5_000,
+        resubmit_timeout=10.0,
+    )
+    assert submitter._initial_fee == 200
+    assert submitter._escalation_factor == 2.0
+    assert submitter._max_fee == 5_000
+    assert submitter._resubmit_timeout == 10.0
+
+
+def test_gas_escalation_resubmission_rejects_invalid_params() -> None:
+    """Invalid constructor parameters raise ValueError."""
+    with pytest.raises(ValueError, match="initial_fee"):
+        TransactionResubmitter(initial_fee=50)
+    with pytest.raises(ValueError, match="escalation_factor"):
+        TransactionResubmitter(escalation_factor=1.0)
+    with pytest.raises(ValueError, match="max_fee"):
+        TransactionResubmitter(initial_fee=500, max_fee=300)
+    with pytest.raises(ValueError, match="resubmit_timeout"):
+        TransactionResubmitter(resubmit_timeout=0)
+
+
+def test_gas_escalation_resubmission_track_and_untrack() -> None:
+    """Track adds and untrack removes from pending set."""
+    submitter = TransactionResubmitter()
+    submitter.track("tx-1")
+    assert submitter.get_pending_count() == 1
+    assert "tx-1" in submitter.get_pending_ids()
+
+    submitter.track("tx-2", base_fee=500)
+    assert submitter.get_pending_count() == 2
+
+    submitter.untrack("tx-1")
+    assert submitter.get_pending_count() == 1
+    assert "tx-1" not in submitter.get_pending_ids()
+
+
+def test_gas_escalation_resubmission_skips_duplicate() -> None:
+    """Tracking the same tx_id twice does not create duplicates."""
+    submitter = TransactionResubmitter()
+    submitter.track("tx-1")
+    submitter.track("tx-1")
+    assert submitter.get_pending_count() == 1
+
+
+def test_gas_escalation_resubmission_computes_escalated_fee() -> None:
+    """_compute_escalated_fee multiplies base fee by factor, capped at max."""
+    submitter = TransactionResubmitter(initial_fee=100, escalation_factor=2.0, max_fee=1000)
+    from network.nonce_tracker import PendingSubmission
+
+    sub = PendingSubmission(tx_id="tx-1", base_fee=100)
+    assert submitter._compute_escalated_fee(sub) == 200
+
+    sub.base_fee = 600
+    assert submitter._compute_escalated_fee(sub) == 1000  # capped at max_fee
+
+    sub.base_fee = 1000
+    assert submitter._compute_escalated_fee(sub) == 1000  # already at cap
+
+
+def test_gas_escalation_resubmission_escalate_pending_timeout_not_reached() -> None:
+    """No escalation when txs have not yet exceeded the timeout."""
+    submitter = TransactionResubmitter(resubmit_timeout=60.0)
+    submitter.track("tx-1", base_fee=100)
+
+    import asyncio
+    escalated = asyncio.run(submitter.escalate_pending())
+    assert escalated == []
+
+
+def test_gas_escalation_resubmission_escalate_pending_calls_callback() -> None:
+    """Callback is invoked for pending tx exceeding timeout."""
+    callback_log: list = []
+
+    def fake_resubmit(tx_id: str, new_fee: int) -> bool:
+        callback_log.append((tx_id, new_fee))
+        return True
+
+    submitter = TransactionResubmitter(
+        resubmit_timeout=0.001,
+        initial_fee=100,
+        escalation_factor=2.0,
+        max_fee=10_000,
+        resubmit_fn=fake_resubmit,
+    )
+
+    # Manually set submitted_at far in the past to trigger escalation
+    from network.nonce_tracker import PendingSubmission
+    import time
+
+    submitter._pending["tx-1"] = PendingSubmission(
+        tx_id="tx-1", base_fee=100, submitted_at=time.monotonic() - 60,
+    )
+
+    import asyncio
+    escalated = asyncio.run(submitter.escalate_pending())
+    assert "tx-1" in escalated
+    assert len(callback_log) == 1
+    assert callback_log[0] == ("tx-1", 200)  # 100 * 2.0
+
+
+def test_gas_escalation_resubmission_escalate_multiple() -> None:
+    """Multiple pending txs are escalated correctly."""
+    callback_log: list = []
+
+    def fake_resubmit(tx_id: str, new_fee: int) -> bool:
+        callback_log.append((tx_id, new_fee))
+        return True
+
+    submitter = TransactionResubmitter(
+        resubmit_timeout=0.001,
+        escalation_factor=1.5,
+        max_fee=10_000,
+        resubmit_fn=fake_resubmit,
+    )
+
+    import time
+    from network.nonce_tracker import PendingSubmission
+
+    for i in range(3):
+        submitter._pending[f"tx-{i}"] = PendingSubmission(
+            tx_id=f"tx-{i}", base_fee=100, submitted_at=time.monotonic() - 60,
+        )
+
+    import asyncio
+    escalated = asyncio.run(submitter.escalate_pending())
+    assert len(escalated) == 3
+    assert len(callback_log) == 3
+    for tx_id, new_fee in callback_log:
+        assert new_fee == 150  # 100 * 1.5
+
+
+def test_gas_escalation_resubmission_callback_failure_does_not_crash() -> None:
+    """A failing callback is caught and does not prevent other escalations."""
+    call_count = 0
+
+    def fake_resubmit(tx_id: str, new_fee: int) -> bool:
+        nonlocal call_count
+        call_count += 1
+        if call_count == 1:
+            raise RuntimeError("Network error")
+        return True
+
+    submitter = TransactionResubmitter(
+        resubmit_timeout=0.001,
+        resubmit_fn=fake_resubmit,
+    )
+
+    import time
+    from network.nonce_tracker import PendingSubmission
+
+    submitter._pending["tx-1"] = PendingSubmission(
+        tx_id="tx-1", base_fee=100, submitted_at=time.monotonic() - 60,
+    )
+    submitter._pending["tx-2"] = PendingSubmission(
+        tx_id="tx-2", base_fee=100, submitted_at=time.monotonic() - 60,
+    )
+
+    import asyncio
+    escalated = asyncio.run(submitter.escalate_pending())
+    assert len(escalated) == 1
+    assert "tx-2" in escalated
+    assert call_count == 2
+
+
+def test_gas_escalation_resubmission_set_resubmit_fn() -> None:
+    """set_resubmit_fn replaces the callback after construction."""
+    submitter = TransactionResubmitter()
+    assert submitter._resubmit_fn is None
+
+    def fn(tx_id: str, fee: int) -> bool:
+        return True
+
+    submitter.set_resubmit_fn(fn)
+    assert submitter._resubmit_fn is fn
+