Fix Betfair duplicate cancel event race condition

- Add FifoCache to track terminal orders and prevent duplicate cancels
- Guard both cancel and lapse paths against duplicate terminal events
- Sync terminal orders cache on reconnect for closed orders
- Add test for duplicate cancel event prevention

Author: cjdsellers

RELEASES.md

@@ -52,6 +52,7 @@ Released on TBD (UTC).
 - Fixed `request_order_book_snapshot` and add Bybit support (#3416), thanks @dxwil
 - Fixed Redis cache buffer flushing during idle periods (#3426), thanks for reporting @santivazq
 - Fixed Betfair dropped fills from premature cache update
+- Fixed Betfair duplicate cancel event race condition
 - Fixed Binance Spot WebSocket subscription acknowledgment parsing (#3382), thanks @Johnkhk
 - Fixed Binance Futures instrument parsing for margin requirements (#3420), thanks @linimin
 - Fixed Binance algo order quantity `AttributeError` on _mem access

nautilus_trader/adapters/betfair/execution.py

@@ -71,6 +71,7 @@
 from nautilus_trader.core.datetime import ensure_pydatetime_utc
 from nautilus_trader.core.datetime import millis_to_nanos
 from nautilus_trader.core.datetime import secs_to_nanos
+from nautilus_trader.core.nautilus_pyo3 import FifoCache
 from nautilus_trader.core.uuid import UUID4
 from nautilus_trader.execution.messages import CancelAllOrders
 from nautilus_trader.execution.messages import CancelOrder
@@ -188,6 +189,10 @@ def __init__(
         # Stores published executions per order to avoid duplicates and support reconciliation
         self._published_executions: dict[ClientOrderId, list[TradeId]] = defaultdict(list)
 
+        # Tracks orders for which a terminal event (cancel/expire) has been generated
+        # to prevent duplicate events from race conditions with multiple event sources
+        self._terminal_orders: FifoCache = FifoCache()
+
     @property
     def instrument_provider(self) -> BetfairInstrumentProvider:
         """
@@ -284,6 +289,9 @@ def _sync_fill_caches_from_orders(self) -> None:
         synced_count = 0
 
         for order in orders:
+            if order.is_closed:
+                self._terminal_orders.add(order.client_order_id.value)
+
             if order.filled_qty > 0:
                 self._filled_qty_cache[order.client_order_id] = order.filled_qty
 
@@ -300,6 +308,14 @@ def _sync_fill_caches_from_orders(self) -> None:
                 LogColor.BLUE,
             )
 
+    def _try_mark_terminal_order(self, client_order_id: ClientOrderId) -> bool:
+        key = client_order_id.value
+        if key in self._terminal_orders:
+            return False
+
+        self._terminal_orders.add(key)
+        return True
+
     # -- ACCOUNT HANDLERS -------------------------------------------------------------------------
 
     async def _update_account_state(self) -> None:
@@ -1244,6 +1260,11 @@ def _handle_stream_execution_complete_order_update(
             )
             # If this is the result of a ModifyOrder, we don't want to emit a cancel
             if key not in self._pending_update_order_client_ids:
+                # Guard against duplicate terminal events from race conditions
+                if not self._try_mark_terminal_order(client_order_id):
+                    self._log.debug(f"Skipping duplicate cancel for {client_order_id!r}")
+                    return
+
                 # The remainder of this order has been canceled
                 canceled_ts = self._get_canceled_timestamp(unmatched_order)
                 self.generate_order_canceled(
@@ -1273,11 +1294,13 @@ def _handle_stream_execution_complete_order_update(
                 self._log.error("Cannot handle cancel: {order.client_order_id!r} not found")
                 return
 
-            # Check if order is still open before generating a cancel.
-            # Note: A race condition exists where a closing event might still be en route
-            # to the execution engine. Running with this for now to avoid the complexity
-            # of another hot cache to deal with the lapsed bet sequencing.
+            # Check if order is still open before generating a cancel
             if order.is_open:
+                # Guard against duplicate terminal events from race conditions
+                if not self._try_mark_terminal_order(client_order_id):
+                    self._log.debug(f"Skipping duplicate lapse cancel for {client_order_id!r}")
+                    return
+
                 canceled_ts = self._get_canceled_timestamp(unmatched_order)
                 self.generate_order_canceled(
                     strategy_id=order.strategy_id,

tests/integration_tests/adapters/betfair/test_betfair_execution.py

@@ -698,6 +698,31 @@ async def test_duplicate_trade_id(exec_client, setup_order_state, fill_events, c
     assert fill3.trade_id.value == "75076f6b172799e168869d64df86b4d2717d"
 
 
+@pytest.mark.asyncio
+async def test_duplicate_cancel_events_prevented(exec_client, setup_order_state, cancel_events):
+    """
+    Test that duplicate cancel events from the stream don't cause state transition
+    errors.
+
+    This prevents InvalidStateTrigger: CANCELED -> CANCELED errors when the same
+    cancel message arrives multiple times (e.g., from stream replay or reconnect).
+
+    """
+    # Arrange
+    order_change_message = BetfairStreaming.ocm_CANCEL()
+    await setup_order_state(order_change_message=order_change_message)
+
+    # Act - Send the same cancel message twice (simulating reconnect/replay)
+    exec_client.handle_order_stream_update(order_change_message)
+    await asyncio.sleep(0)
+    exec_client.handle_order_stream_update(order_change_message)
+    await asyncio.sleep(0)
+
+    # Assert - Only one cancel event should be generated
+    assert len(cancel_events) == 1
+    assert isinstance(cancel_events[0], OrderCanceled)
+
+
 @pytest.mark.parametrize(
     ("side", "price", "quantity", "free"),
     [