hotfix: stabilize sync jobs with reduced concurrency and enhanced deadlock handling

[orhanrauf]

• logs
• less workers: 100 -> 20
• more retries on deadlock
• deduplication async lock

---

Summary by cubic

Stabilizes sync runs by reducing concurrency and hardening deadlock handling. Adds an async dedup lock, smarter DB retries, and moves noisy logs to debug with timing across the pipeline.

• Bug Fixes

  • Prevent duplicate inserts with an async lock in entity dedup to avoid race conditions and deadlocks.
  • Retry PostgreSQL deadlocks with 5 attempts, exponential backoff + jitter, and clearer diagnostics.

• Performance

  • Lower max workers and thread pool from 100 to 20; align DB pool sizing to prevent connection exhaustion.
  • Add stage-by-stage debug timing and batch IDs across chunking, Mistral conversion, OpenAI embedding, Qdrant upserts/deletes, and orchestration.
  • Shift high-volume logs to debug; include throughput metrics and semaphore state in Qdrant operations.

^{Written for commit d0d3f12. Summary will update automatically on new commits.}

[cubic-dev-ai]

2 issues found across 9 files

Prompt for AI agents (all 2 issues)

Understand the root cause of the following 2 issues and fix them.


<file name="backend/airweave/platform/converters/mistral_converter.py">

<violation number="1" location="backend/airweave/platform/converters/mistral_converter.py:123">
Rule violated: **Check for Cursor Rules Drift**

Cursor backend rules already require duration measurements to use time.monotonic(), but this new timing instrumentation in convert_batch relies on time.time(). Please switch this and the other stage timings in this block to time.monotonic() so the implementation matches the documented backend rule.</violation>
</file>

<file name="backend/airweave/platform/sync/orchestrator.py">

<violation number="1" location="backend/airweave/platform/sync/orchestrator.py:70">
Rule violated: **Check for Cursor Rules Drift**

Cleanup-phase logging was downgraded to DEBUG, but the sync-architecture Cursor rule still instructs INFO-level logging for these 🧹 messages. Please update the rule to reflect the new logging level so Cursor guidance stays accurate.</violation>
</file>

_{React with 👍 or 👎 to teach cubic. Mention @cubic-dev-ai to give feedback, ask questions, or re-run the review.}

[cubic-dev-ai]

Rule violated: Check for Cursor Rules Drift

Cursor backend rules already require duration measurements to use time.monotonic(), but this new timing instrumentation in convert_batch relies on time.time(). Please switch this and the other stage timings in this block to time.monotonic() so the implementation matches the documented backend rule.

Prompt for AI agents

Address the following comment on backend/airweave/platform/converters/mistral_converter.py at line 123:

<comment>Cursor backend rules already require duration measurements to use time.monotonic(), but this new timing instrumentation in convert_batch relies on time.time(). Please switch this and the other stage timings in this block to time.monotonic() so the implementation matches the documented backend rule.</comment>

<file context>
@@ -118,48 +118,104 @@ async def convert_batch(self, file_paths: List[str]) -&gt; Dict[str, str]:
         &quot;&quot;&quot;
+        import time
+
+        batch_start = time.time()
+        logger.debug(f&quot;🔶 Mistral Converter: Starting batch conversion of {len(file_paths)} files&quot;)
+
</file context>

[cubic-dev-ai]

Rule violated: Check for Cursor Rules Drift

Cleanup-phase logging was downgraded to DEBUG, but the sync-architecture Cursor rule still instructs INFO-level logging for these 🧹 messages. Please update the rule to reflect the new logging level so Cursor guidance stays accurate.

Prompt for AI agents

Address the following comment on backend/airweave/platform/sync/orchestrator.py at line 70:

<comment>Cleanup-phase logging was downgraded to DEBUG, but the sync-architecture Cursor rule still instructs INFO-level logging for these 🧹 messages. Please update the rule to reflect the new logging level so Cursor guidance stays accurate.</comment>

<file context>
@@ -65,33 +65,41 @@ async def run(self) -&gt; schemas.Sync:
+            self.sync_context.logger.debug(&quot;🚀 PHASE 1: Starting sync initialization...&quot;)
             await self._start_sync()
-            self.sync_context.logger.info(f&quot;✅ PHASE 1 complete ({time.time() - phase_start:.2f}s)&quot;)
+            self.sync_context.logger.debug(
+                f&quot;✅ PHASE 1 complete ({time.time() - phase_start:.2f}s)&quot;
+            )
</file context>