fix: avoid blocking account sync lock contention

Author: tianhe

backend_api_python/app/services/live_trading/account_positions.py

@@ -31,13 +31,13 @@ def sync_account_positions(
     exchange_id: str,
     market_type: str,
     legs: List[AccountLegSnapshot],
-) -> None:
+) -> bool:
     """
     Replace L1 rows for (credential_id, market_type) with the latest exchange snapshot.
     """
     cred = int(credential_id or 0)
     if cred <= 0:
-        return
+        return False
     mt = str(market_type or "swap").strip().lower()
     if mt in ("futures", "future", "perp", "perpetual"):
         mt = "swap"
@@ -48,9 +48,18 @@ def sync_account_positions(
     with get_db_connection() as db:
         cur = db.cursor()
         cur.execute(
-            "SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))",
+            "SELECT pg_try_advisory_xact_lock(hashtextextended(%s, 0)) AS acquired",
             (f"account-position-sync:{uid}:{cred}:{mt}",),
         )
+        lock_row = cur.fetchone() or {}
+        acquired = bool(
+            lock_row.get("acquired")
+            if isinstance(lock_row, dict)
+            else lock_row[0]
+        )
+        if not acquired:
+            cur.close()
+            return False
         for leg in legs or []:
             sym = normalize_strategy_symbol(leg.symbol) or str(leg.symbol or "").strip()
             side = str(leg.side or "").strip().lower()
@@ -118,6 +127,7 @@ def sync_account_positions(
             )
         db.commit()
         cur.close()
+    return True
 
 
 def account_legs_from_exchange_maps(

backend_api_python/tests/test_position_ownership.py

@@ -19,13 +19,17 @@
 
 
 class _RecordingCursor:
-    def __init__(self):
+    def __init__(self, *, lock_acquired=True):
         self.calls = []
         self.rowcount = 1
+        self.lock_acquired = lock_acquired
 
     def execute(self, sql, params=None):
         self.calls.append((sql, params))
 
+    def fetchone(self):
+        return {"acquired": self.lock_acquired}
+
     def close(self):
         return None
 
@@ -58,20 +62,39 @@ def test_account_position_sync_serializes_each_credential_bucket(monkeypatch):
     db = _RecordingDb(cursor)
     monkeypatch.setattr(account_positions, "get_db_connection", lambda: _RecordingContext(db))
 
-    account_positions.sync_account_positions(
+    synced = account_positions.sync_account_positions(
         user_id=7,
         credential_id=33,
         exchange_id="gate",
         market_type="swap",
         legs=[],
     )
 
-    assert "pg_advisory_xact_lock" in cursor.calls[0][0]
+    assert synced is True
+    assert "pg_try_advisory_xact_lock" in cursor.calls[0][0]
     assert cursor.calls[0][1] == ("account-position-sync:7:33:swap",)
     assert "DELETE FROM qd_account_positions" in cursor.calls[1][0]
     assert db.committed is True
 
 
+def test_account_position_sync_skips_instead_of_waiting_for_busy_bucket(monkeypatch):
+    cursor = _RecordingCursor(lock_acquired=False)
+    db = _RecordingDb(cursor)
+    monkeypatch.setattr(account_positions, "get_db_connection", lambda: _RecordingContext(db))
+
+    synced = account_positions.sync_account_positions(
+        user_id=7,
+        credential_id=33,
+        exchange_id="gate",
+        market_type="swap",
+        legs=[],
+    )
+
+    assert synced is False
+    assert len(cursor.calls) == 1
+    assert db.committed is False
+
+
 def test_fresh_snapshot_clears_only_older_recorded_drift(monkeypatch):
     cursor = _RecordingCursor()
     db = _RecordingDb(cursor)