feat: worker lane split, dirty flag optimization, and lease-based sync - #45
Merged
Conversation
Port sync infrastructure overhaul from prod: Architecture: - Replace QStash queue with lease-based connection_sync_state control plane - Add worker lane split (realtime vs reconcile claim modes) - Add dedicated webhook ingress service (webhooks_index.py) - Extract app_factory.py from monolithic index.py - Add Railway sync worker entrypoint (sync_worker.py) New modules: - sync_dispatcher.py: shared dispatch layer for marking streams dirty - sync_state_store.py: lease claim/heartbeat/complete/fail helpers - stream_worker.py: Railway worker loop for lease-based sync - failure_policy.py: auto-quarantine for permanently broken connections - google_webhook_security.py: HMAC token signing for Calendar webhooks - google_retry.py: Google API retry helpers Database migrations: - connection_sync_state table with RPCs (mark dirty, claim, heartbeat, complete, fail) - Skip redundant dirty-mark writes (dedup Pub/Sub bursts) - Per-stream next_reconcile_at replacing broad sweep - Claim mode split (dirty_only/reconcile_only/any) with jitter - Autovacuum tuning for high-churn sync state table Router changes: - Webhooks: ingress dedup cache, Pub/Sub JWT verification, dirty marking - Workers: lease-based execution wrapper, mode dispatch - Cron: watch-health detection, stream-aware setup-missing-watches - Calendar/Email: manual sync via mark_stream_dirty (priority 200) - Auth service: initial sync uses dirty marking instead of QStash
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds a lease-based control plane and worker for per-connection syncs, replaces inline/queue syncs with dirty-mark dispatch, centralizes FastAPI app creation and webhook ingress, introduces Google retry and calendar-token helpers, many DB migrations for connection sync state, and broad test coverage updates. Changes
Sequence Diagram(s)sequenceDiagram
participant Webhook
participant ControlPlane as Control Plane\n(Supabase RPC)
participant Worker
participant SyncService as Sync Service\n(Gmail/Calendar)
participant Database as Database\n(Ext Connections)
rect rgba(100,150,200,0.5)
Note over Webhook,ControlPlane: Ingress → Dirty-mark
Webhook->>ControlPlane: mark_connection_sync_dirty(connection_id, sync_kind, metadata)
ControlPlane->>ControlPlane: upsert connection_sync_state (dirty=true, dirty_generation++)
ControlPlane-->>Webhook: true
end
rect rgba(150,100,150,0.5)
Note over Worker,ControlPlane: Claim loop
Worker->>ControlPlane: claim_connection_sync_lease(worker_id, lease_seconds, claim_mode)
ControlPlane->>ControlPlane: SELECT FOR UPDATE SKIP LOCKED → set lease_owner/expires
ControlPlane-->>Worker: claim payload {connection_id, provider, sync_kind, cursor}
end
rect rgba(150,200,100,0.5)
Note over Worker,SyncService: Lease-protected execution
Worker->>Worker: start_connection_sync_lease_heartbeat()
Worker->>SyncService: dispatch sync for connection (cursor, metadata)
SyncService->>Database: read credentials & provider data
SyncService->>Database: upsert results
SyncService-->>Worker: {status: "ok", latest_cursor}
Worker->>Worker: stop_connection_sync_lease_heartbeat()
end
rect rgba(200,150,100,0.5)
Note over Worker,ControlPlane: Completion / Quarantine
alt success
Worker->>ControlPlane: complete_connection_sync_lease(last_synced_cursor, latest_cursor)
ControlPlane->>ControlPlane: merge cursors, clear lease, set next_reconcile_at
else permanent failure
Worker->>ControlPlane: fail_connection_sync_lease(error, retry_seconds)
ControlPlane->>ControlPlane: increment retry_count, set next_retry_at
Worker->>Database: deactivate ext_connection & push_subscriptions (quarantine)
end
end
Estimated code review effort🎯 5 (Critical) | ⏱️ ~110 minutes Poem
✨ Finishing Touches🧪 Generate unit tests (beta)
|
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 13
Note
Due to the large number of review comments, Critical, Major severity comments were prioritized as inline comments.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (4)
core-api/tests/unit/test_notify_attendees_google.py (2)
108-118:⚠️ Potential issue | 🔴 CriticalTest expects tuple
(ok, _tz)but_update_google_eventreturns a single boolean.The function signature at line 196 of
core-api/api/services/calendar/update_event.pyexplicitly declares-> boolas its return type. All return statements in the function return onlyTrueorFalse, not a tuple. The test's tuple unpacking will fail at runtime withTypeError: cannot unpack non-iterable bool object.🐛 Proposed fix to match actual function signature
- ok, _tz = _update_google_event( + ok = _update_google_event( user_id='user-1', user_jwt='jwt', external_id='ext-1', event_data=event_data, connection_id=None, scope='instance', recurring_event_id=None, cutoff_start=None, )🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/tests/unit/test_notify_attendees_google.py` around lines 108 - 118, The test is unpacking a tuple but _update_google_event (in update_event.py) returns a single bool; update the test in test_notify_attendees_google.py to stop tuple-unpacking: call _update_google_event(...) into a single variable (e.g., ok) and assert that ok is True (remove the second _tz variable), so the test matches the function signature and return type.
79-83:⚠️ Potential issue | 🔴 CriticalTest unpacking expects 4 values but
_create_google_eventreturns only 3.The function returns
(event_id, meeting_link, error), not(event_id, meeting_link, error, timezone). Remove_tzfrom the unpacking:Fix
- ev_id, meeting_link, err, _tz = _create_google_event('user-1', 'jwt', event_data, None) + ev_id, meeting_link, err = _create_google_event('user-1', 'jwt', event_data, None)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/tests/unit/test_notify_attendees_google.py` around lines 79 - 83, The test unpacks four values but _create_google_event returns only three; update the unpacking in the test to expect three values (event_id, meeting_link, err) by removing the unused _tz variable and adjust any subsequent references to _tz (none here), ensuring the call to _create_google_event and assertions (e.g., checking recorder['insert']['sendUpdates'] and recorder['insert']['body']) remain unchanged; search for other tests using _create_google_event to apply the same 3-value unpack fix if present.core-api/api/routers/email.py (1)
983-994:⚠️ Potential issue | 🟠 MajorOnly load the fields this endpoint actually needs.
This path now just marks streams dirty, but it still selects
access_token/refresh_tokenand runsdecrypt_ext_connection_tokens()on every connection. That unnecessarily exposes secrets in-process and can fail the whole/syncrequest on a bad token blob even though dirty-marking only needsidandprovider.✂️ Narrow the query to the fields actually used
connections_result = auth_supabase.table('ext_connections')\ - .select('id, provider, provider_email, access_token, refresh_token, token_expires_at, metadata, is_primary')\ + .select('id, provider')\ .eq('user_id', user_id)\ .eq('is_active', True)\ .in_('provider', ['google', 'microsoft'])\ .execute() - connections = [decrypt_ext_connection_tokens(c) for c in (connections_result.data or [])] + connections = connections_result.data or []🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/routers/email.py` around lines 983 - 994, The query in the email sync path retrieves sensitive token fields and calls decrypt_ext_connection_tokens() even though this endpoint only needs to mark streams dirty; update the select on the Supabase query produced by get_authenticated_supabase_client(user_jwt) to only request the minimal fields (e.g., 'id' and 'provider'), remove the decryption step (do not call decrypt_ext_connection_tokens on connections_result.data), and update the downstream logic that builds google_connections and microsoft_connections to operate on the lightweight connection objects (use c.get('provider') and c.get('id') only) so no access_token/refresh_token handling or decryption is performed in this path.core-api/api/routers/webhooks.py (1)
471-490:⚠️ Potential issue | 🟠 MajorLet transient Microsoft dirty-mark failures reach the 503 path.
_mark_microsoft_notification_dirty()catches infrastructure errors and turns them into{"success": False}, andmicrosoft_webhook_notification()ignores those results and still returns 202. That means Supabase/marking outages are acknowledged instead of retried, even though this endpoint is supposed to return 503 on transient failures. Keep validation misses as data returns, but let unexpected exceptions bubble so the outer handler can NACK.🛠️ Suggested fix
- except Exception as e: - logger.warning(f"⚠️ [Microsoft] Failed to mark notification dirty: {e}") - return {"success": False, "error": str(e)} + except Exception: + logger.exception("⚠️ [Microsoft] Failed to mark notification dirty") + raiseAlso applies to: 564-566
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/routers/webhooks.py` around lines 471 - 490, The webhook currently swallows infrastructure errors inside _mark_microsoft_notification_dirty so microsoft_webhook_notification always returns 202; change _mark_microsoft_notification_dirty to not convert unexpected infra exceptions into {"success": False} but instead re-raise those exceptions (or raise a specific transient error) while still returning {"success": False} for known validation/data issues, and update microsoft_webhook_notification to detect such raised exceptions (or catch the specific transient error) so the outer try/except can return 503; apply the same change for the analogous block handling notifications at the other occurrence referenced in the comment.
🟡 Minor comments (8)
core-api/tests/unit/test_share_link_access.py-37-38 (1)
37-38:⚠️ Potential issue | 🟡 Minor
ilikefake does not emulate SQL ILIKE wildcard behavior.The fake allows
ilikefilters but uses equality-style matching instead of pattern matching. Production code uses wildcard patterns like.ilike("email", f"{query}%")which will not be correctly simulated by current implementation—tests may pass with incorrect results.Suggested patch
+import re + class FakeSupabaseQuery: @@ def _matches(self, row: Dict[str, Any]) -> bool: for op, field, value in self._filters: row_value = row.get(field) if op == "eq" and row_value != value: return False if op == "ilike": - if row_value is None or str(row_value).lower() != str(value).lower(): + if row_value is None: + return False + pattern = "^" + re.escape(str(value)).replace("%", ".*").replace("_", ".") + "$" + if re.match(pattern, str(row_value), re.IGNORECASE) is None: return False return True🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/tests/unit/test_share_link_access.py` around lines 37 - 38, The test fake currently stores ("ilike", field, value) but matches it by equality; change the fake's filter evaluation to emulate SQL ILIKE wildcard semantics: when encountering a filter tuple with operator "ilike" (from self._filters), convert the SQL pattern (value) where % -> .* and _ -> . into a regex (properly escaping other chars), then perform a case-insensitive regex match against the record[field] (use re.IGNORECASE) so patterns like f"{query}%" behave like SQL ILIKE; update the code path that iterates self._filters and evaluates records (the function that consumes self._filters) to use this regex-based matching for "ilike" entries.core-api/tests/unit/test_token_encryption.py-61-80 (1)
61-80:⚠️ Potential issue | 🟡 MinorUse obviously fake plaintext tokens here.
These values only need to be non-Fernet strings, but the
ya29...shape looks close enough to a real Google access token to keep tripping static analysis on Line 70. Replacing them with clearly synthetic placeholders preserves the test intent without secret-scanner noise.🧹 Example cleanup
- original = "ya29.a0AfH6SMB_super_secret_access_token" + original = "plain-non-fernet-access-token" @@ - token = "ya29.a0AfH6SMB_test_token" + token = "plain-non-fernet-token" @@ - plaintext = "ya29.a0AfH6SMB_plaintext_token" + plaintext = "plain-migration-token" @@ - assert decrypt_token("ya29.plain_token") == "ya29.plain_token" + assert decrypt_token("plain-token") == "plain-token"Also applies to: 236-242
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/tests/unit/test_token_encryption.py` around lines 61 - 80, Replace realistic-looking Google token strings used in the tests with clearly synthetic placeholders: update the plaintexts in the tests that call encrypt_token and decrypt_token (notably the variables in test_produces_unique_ciphertext, test_decrypt_plaintext_passthrough, and the earlier test using original = "ya29...") to non-sensitive, obviously fake values (e.g., "fake_token_1", "fake_token_2" or "plaintext_token_placeholder") so the tests still verify non-Fernet passthrough and unique ciphertext behavior while avoiding secret-scanner false positives.core-api/setup_pubsub_subscription.py-85-89 (1)
85-89:⚠️ Potential issue | 🟡 MinorAdd validation for empty
WEBHOOK_BASE_URL.If
WEBHOOK_BASE_URLis unset or empty,push_endpointbecomes/api/webhooks/gmail, which is invalid and would silently configure a broken push subscription.Proposed fix: fail fast on missing WEBHOOK_BASE_URL
webhook_base_url = os.getenv( "WEBHOOK_BASE_URL", "", # Set WEBHOOK_BASE_URL in your environment ).rstrip("/") + + if not webhook_base_url: + print("❌ WEBHOOK_BASE_URL is required but not set") + print(" Example: WEBHOOK_BASE_URL=https://core-webhooks-production.up.railway.app") + sys.exit(1) parts = pubsub_topic.split("/")Also applies to: 149-150
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/setup_pubsub_subscription.py` around lines 85 - 89, Validate that webhook_base_url is non-empty after reading os.getenv and rstrip("/") and fail fast if it's empty: when constructing push_endpoint (using webhook_base_url and "/api/webhooks/gmail") check webhook_base_url and raise a clear exception or exit with an error log so you don't silently configure a broken subscription; update the same validation where webhook_base_url is used later (the push_endpoint creation around the code referencing lines ~149-150) to reuse the same check or helper to ensure consistency.core-api/api/routers/email.py-1034-1043 (1)
1034-1043:⚠️ Potential issue | 🟡 MinorAdvertise the new
202contract in the route metadata.The handler now returns
JSONResponse(status_code=202, ...), but the route decorator still declaresstatus_code=200. Runtime is fine, but the OpenAPI schema / generated clients will keep modeling this as a synchronous200endpoint.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/routers/email.py` around lines 1034 - 1043, The route decorator still declares status_code=200 while the handler returns JSONResponse(status_code=202); update the route decorator for this endpoint (the function with the route decorator above the handler that returns JSONResponse(... status_code=202)) to declare status_code=202 and adjust its OpenAPI responses/response_model metadata accordingly so the generated schema/client reflect the asynchronous 202 contract (add a 202 response entry and description if the decorator uses responses= or response_model=).core-api/api/app_factory.py-144-150 (1)
144-150:⚠️ Potential issue | 🟡 MinorReplace deprecated
datetime.utcnow()with timezone-aware alternative.
datetime.utcnow()is deprecated in Python 3.12+ and returns a naive datetime. Usedatetime.now(timezone.utc)for a timezone-aware UTC timestamp.🔧 Proposed fix
+from datetime import datetime, timezone + `@app.get`("/api/health", response_model=HealthResponse) async def health_check(): return { "status": "healthy", "service": service_name, - "timestamp": datetime.utcnow().isoformat() + "Z", + "timestamp": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/app_factory.py` around lines 144 - 150, In the health_check endpoint (function health_check) replace the naive datetime usage datetime.utcnow().isoformat() + "Z" with a timezone-aware call such as datetime.now(timezone.utc).isoformat() (or datetime.now(tz=timezone.utc).isoformat()) and update imports to include timezone from datetime so the returned timestamp is UTC-aware rather than naive.core-api/supabase/migrations/20260402000001_skip_redundant_dirty_mark_writes.sql-90-107 (1)
90-107:⚠️ Potential issue | 🟡 MinorKeep
last_provider_event_atmonotonic in the upsert path.
last_provider_event_at = COALESCE(p_provider_event_at, v_now)can move this field backwards when provider events arrive out of order. Alast_*timestamp should only advance; otherwise downstream health/order checks can observe older state than the row already had.🛠️ Suggested fix
- last_provider_event_at = COALESCE(p_provider_event_at, v_now), + last_provider_event_at = GREATEST( + COALESCE(public.connection_sync_state.last_provider_event_at, '-infinity'::timestamptz), + COALESCE(p_provider_event_at, v_now) + ),🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/supabase/migrations/20260402000001_skip_redundant_dirty_mark_writes.sql` around lines 90 - 107, The upsert currently sets last_provider_event_at = COALESCE(p_provider_event_at, v_now), which can move the timestamp backwards; change the assignment in the ON CONFLICT DO UPDATE block to keep last_provider_event_at monotonic by taking the maximum of the existing value and the incoming value, e.g. set last_provider_event_at = GREATEST(public.connection_sync_state.last_provider_event_at, COALESCE(p_provider_event_at, v_now)), so the stored last_provider_event_at only ever advances (handles NULLs via COALESCE).core-api/tests/unit/test_stream_worker.py-346-355 (1)
346-355:⚠️ Potential issue | 🟡 MinorPin
SYNC_WORKER_POLL_SECONDSin this test.
run_forever()reads the env var, so a CI or local override can change the sleep duration and makeassert_called_once_with(5.0)flaky. Set the env inside the test before invoking the loop.🧪 Suggested fix
-def test_run_forever_recovers_from_run_once_exception(): +def test_run_forever_recovers_from_run_once_exception(monkeypatch): from api.services.syncs import stream_worker + monkeypatch.setenv("SYNC_WORKER_POLL_SECONDS", "5") with patch.object(stream_worker, "build_worker_id", return_value="worker-1"): with patch.object(stream_worker, "run_once", side_effect=[RuntimeError("boom"), KeyboardInterrupt]): with patch.object(stream_worker.time, "sleep") as sleep_mock: with pytest.raises(KeyboardInterrupt): stream_worker.run_forever()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/tests/unit/test_stream_worker.py` around lines 346 - 355, Pin the SYNC_WORKER_POLL_SECONDS env var before invoking stream_worker.run_forever in test_run_forever_recovers_from_run_once_exception so the sleep assertion is not flaky; set the env to "5" (e.g. using monkeypatch.setenv("SYNC_WORKER_POLL_SECONDS", "5") or os.environ["SYNC_WORKER_POLL_SECONDS"]="5") prior to calling stream_worker.run_forever and restore/cleanup afterwards; keep the existing patches for stream_worker.build_worker_id, stream_worker.run_once, and stream_worker.time.sleep.core-api/api/services/syncs/watch_manager.py-36-42 (1)
36-42:⚠️ Potential issue | 🟡 MinorGuard the webhook-base fallback against
None.The fallback path in both helpers uses
getattr(settings, "webhook_base_url", "").rstrip("/"), which can raiseAttributeErrorifwebhook_base_urlisNone. While the field is typed asstr = ""in the Settings class, thegetattr()default only protects against missing attributes—not against an existing attribute with aNonevalue. In test stubs or lightweight implementations that bypass Pydantic validation, this edge case could surface. Addingor ""safely handles bothNoneand empty-string cases.Suggested fix
def _get_gmail_webhook_url() -> str: """Resolve the Gmail webhook URL while tolerating lightweight test stubs.""" explicit = getattr(settings, "gmail_webhook_url", None) if explicit: return explicit - base_url = getattr(settings, "webhook_base_url", "").rstrip("/") + base_url = (getattr(settings, "webhook_base_url", "") or "").rstrip("/") return f"{base_url}/api/webhooks/gmail" def _get_calendar_webhook_url() -> str: """Resolve the Calendar webhook URL while tolerating lightweight test stubs.""" explicit = getattr(settings, "calendar_webhook_url", None) if explicit: return explicit - base_url = getattr(settings, "webhook_base_url", "").rstrip("/") + base_url = (getattr(settings, "webhook_base_url", "") or "").rstrip("/") return f"{base_url}/api/webhooks/calendar"🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/services/syncs/watch_manager.py` around lines 36 - 42, _get_gmail_webhook_url currently calls getattr(settings, "webhook_base_url", "").rstrip("/") which will raise if webhook_base_url exists but is None; change the assignment to use a safe coalesce like (getattr(settings, "webhook_base_url", None) or "").rstrip("/") so None becomes an empty string before rstrip, and apply the same defensive pattern to the companion helper(s) that build webhook URLs in this module.
🧹 Nitpick comments (8)
core-api/tests/unit/test_google_webhook_security.py (1)
13-43: Good coverage of primary paths.Tests verify both the enabled-secret round-trip (including mismatch detection) and the disabled-secret bypass behavior.
Optional: Consider adding edge case for empty-string token when secret is configured
def test_calendar_channel_token_rejects_empty_string_when_secret_configured(monkeypatch): from lib.google_webhook_security import verify_google_calendar_channel_token monkeypatch.setattr( "api.config.settings.google_calendar_webhook_secret", "test-secret", ) # Empty string should be rejected when a secret is configured assert verify_google_calendar_channel_token("conn-1", "channel-1", "") is False🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/tests/unit/test_google_webhook_security.py` around lines 13 - 43, Add a unit test to cover the edge case where a secret is configured but the provided token is an empty string: in core-api/tests/unit/test_google_webhook_security.py create a new test (e.g., test_calendar_channel_token_rejects_empty_string_when_secret_configured) that monkeypatches api.config.settings.google_calendar_webhook_secret to a non-empty value, imports verify_google_calendar_channel_token, and asserts verify_google_calendar_channel_token("conn-1", "channel-1", "") is False; this uses the existing verify_google_calendar_channel_token symbol to ensure empty-string tokens are rejected when a secret is set.core-api/webhooks_index.py (1)
3-8: Prefer avoiding import-timesys.pathmutation in the webhook entrypoint.This pattern can create fragile import behavior across environments. If deployment runs module-style (
python -m/ proper working dir), you can remove it and keep imports deterministic.♻️ Proposed cleanup
-import os -import sys - -sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) - from api.app_factory import create_webhooks_app🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/webhooks_index.py` around lines 3 - 8, Remove the import-time sys.path mutation in webhooks_index.py: delete the sys.path.insert(...) line and rely on package-style imports so create_webhooks_app from api.app_factory is resolved via normal Python package/module resolution; ensure the package is installed or the project is executed with the correct working directory or via python -m so the import of create_webhooks_app succeeds without modifying sys.path at import time.core-api/sync_worker.py (1)
9-17: Scope logging configuration to runtime entrypoint.Moving
basicConfigunder amain()path avoids import-time global logging side effects.♻️ Proposed refactor
-logging.basicConfig( - level=logging.INFO, - format="%(asctime)s %(levelname)s %(name)s %(message)s", - stream=sys.stdout, -) - - -if __name__ == "__main__": - run_forever() +def main() -> None: + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s %(message)s", + stream=sys.stdout, + ) + run_forever() + + +if __name__ == "__main__": + main()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/sync_worker.py` around lines 9 - 17, The logging.basicConfig call is executed at import time causing global side effects; move it into a dedicated main() entrypoint so configuration happens at runtime. Create a main() function that calls logging.basicConfig(...) then run_forever(), and update the if __name__ == "__main__": block to call main(); reference logging.basicConfig and run_forever to locate the code to change.core-api/tests/unit/test_sync_dispatcher.py (1)
60-64: Consider a simpler approach for the throwing mock.The generator-throw pattern
(_ for _ in ()).throw(AssertionError(...))is unconventional and flagged by SonarCloud. A clearer alternative would be a function that raises directly.♻️ Simpler throwing mock
- queue_client = SimpleNamespace( - enqueue_sync_for_connection=lambda *_args, **_kwargs: (_ for _ in ()).throw( - AssertionError("enqueue should not run when mark dirty fails") - ) - ) + def _fail_if_called(*_args, **_kwargs): + raise AssertionError("enqueue should not run when mark dirty fails") + + queue_client = SimpleNamespace(enqueue_sync_for_connection=_fail_if_called)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/tests/unit/test_sync_dispatcher.py` around lines 60 - 64, Replace the exotic generator-throw mock for queue_client.enqueue_sync_for_connection with a simple callable that raises directly: locate the queue_client SimpleNamespace in the test and change the enqueue_sync_for_connection implementation to a normal function (or lambda) that raises AssertionError("enqueue should not run when mark dirty fails") when invoked so the test intent is clearer and SonarCloud warning is resolved.core-api/tests/unit/test_sync_state_store.py (1)
105-116: Consider making expected values explicit in the parametrize decorator.The conditional logic for calculating
expectedis harder to follow than explicit test cases. This could be clearer with direct(value, expected)tuples.♻️ Clearer parametrization
-@pytest.mark.parametrize("value", ["", "0", "-1", "abc", str(8 * 24 * 3600)]) -def test_get_reconcile_interval_seconds_clamps_invalid_values(monkeypatch, value): +@pytest.mark.parametrize("value,expected", [ + ("", 21600), # Empty string -> default + ("0", 300), # Zero -> minimum + ("-1", 300), # Negative -> minimum + ("abc", 21600), # Non-numeric -> default + (str(8 * 24 * 3600), 7 * 24 * 3600), # Too large -> maximum (7 days) +]) +def test_get_reconcile_interval_seconds_clamps_invalid_values(monkeypatch, value, expected): from api.services.syncs.sync_state_store import get_reconcile_interval_seconds monkeypatch.setenv("RECONCILE_INTERVAL_SECONDS", value) - expected = 300 if value == "0" or value == "-1" else 21600 - if value == str(8 * 24 * 3600): - expected = 7 * 24 * 3600 - assert get_reconcile_interval_seconds() == expected🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/tests/unit/test_sync_state_store.py` around lines 105 - 116, Refactor the test_get_reconcile_interval_seconds_clamps_invalid_values to use explicit (value, expected) tuples in the pytest.mark.parametrize decorator instead of computing `expected` inside the test; update the test signature to accept both parameters and assert get_reconcile_interval_seconds() equals the provided expected; this targets the test for the get_reconcile_interval_seconds function and removes the conditional logic currently inside the test for clarity and future maintainability.core-api/api/app_factory.py (1)
51-65: Theasynckeyword adds unnecessary overhead here.The
global_exception_handlerperforms noawaitoperations. While FastAPI handles this, removingasyncwould be slightly more efficient.♻️ Optional: Remove async keyword
- async def global_exception_handler(request: Request, exc: Exception): + def global_exception_handler(request: Request, exc: Exception):🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/app_factory.py` around lines 51 - 65, The global_exception_handler is declared async but contains no awaitable operations; change it to a synchronous function by removing the async keyword from the global_exception_handler definition and keep its body and return of JSONResponse unchanged, then register it via app.add_exception_handler(Exception, global_exception_handler) as currently done (ensure the function signature remains global_exception_handler(request: Request, exc: Exception) and that Sentry capture and logging calls remain synchronous).core-api/tests/unit/test_workers_mode_dispatch.py (1)
381-412: Consider usingpytest-mockorExitStackto reduce nesting depth.The 6-level nested
with patch.objectblocks make this test harder to read and maintain. This is a common pattern across multiple tests in this file.♻️ Example using ExitStack to flatten patches
from contextlib import ExitStack def test_run_with_stream_lease_quarantines_when_retry_cap_is_hit(monkeypatch): from api.routers import workers monkeypatch.setenv("MAX_FAILURE_RETRY_COUNT", "3") with ExitStack() as stack: stack.enter_context(patch.object(workers, "get_service_role_client", return_value=MagicMock())) stack.enter_context(patch.object( workers, "claim_connection_sync_lease", return_value={"connection_id": "conn-1", "provider": "microsoft"}, )) stack.enter_context(patch.object( workers, "get_connection_sync_state", return_value={"retry_count": 2}, )) complete_mock = stack.enter_context(patch("api.services.syncs.failure_policy.complete_connection_sync_lease")) stack.enter_context(patch.object(workers, "fail_connection_sync_lease")) deactivate_mock = stack.enter_context(patch( "api.services.syncs.failure_policy.deactivate_connection_with_subscriptions" )) result = workers._run_with_stream_lease( "conn-1", workers.SYNC_KIND_EMAIL, lambda: {"status": "error", "message": "boom"}, ) assert result["status"] == "quarantined" complete_mock.assert_called_once() deactivate_mock.assert_called_once()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/tests/unit/test_workers_mode_dispatch.py` around lines 381 - 412, The test test_run_with_stream_lease_quarantines_when_retry_cap_is_hit has deeply nested patch.contexts; replace the 6-level nested patch.object/patch calls that target workers.get_service_role_client, workers.claim_connection_sync_lease, workers.get_connection_sync_state, api.services.syncs.failure_policy.complete_connection_sync_lease, workers.fail_connection_sync_lease, and api.services.syncs.failure_policy.deactivate_connection_with_subscriptions with a flattened approach (use contextlib.ExitStack to enter all patches and capture returned mock objects into variables, or use pytest-mock fixtures like mocker.patch to create the mocks) so the call to workers._run_with_stream_lease remains the same and assertions use the captured complete_mock, fail_mock, and deactivate_mock. Ensure behavior and return_value setups are preserved (e.g., MAX_FAILURE_RETRY_COUNT env, claim_connection_sync_lease -> {"connection_id":"conn-1","provider":"microsoft"}, get_connection_sync_state -> {"retry_count":2}) and that complete_mock.assert_called_once(), fail_mock.assert_not_called(), and deactivate_mock.assert_called_once() remain.core-api/api/services/syncs/sync_state_store.py (1)
193-244: Extract the heartbeat tick to clear the Sonar blocker.Line 193 is doing loop control, RPC I/O, lease-loss handling, and logging in one closure. Pulling the body into a small helper should drop the cognitive complexity below the threshold and make the failure path easier to unit-test.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/services/syncs/sync_state_store.py` around lines 193 - 244, The closure _heartbeat_loop in start_connection_sync_lease_heartbeat does loop control, RPC I/O, lease-loss handling and logging; extract the inner tick (the body of the while loop that calls heartbeat_connection_sync_lease and handles the alive/exception paths) into a small helper function (e.g., _heartbeat_tick or heartbeat_connection_sync_lease_tick) that accepts service_supabase, connection_id, sync_kind, worker_id, lease_seconds and logger (and optionally stop_event) and returns a boolean/enum indicating whether to continue or stop; then simplify _heartbeat_loop to just call that helper on each interval and act on its return value, keeping start_connection_sync_lease_heartbeat otherwise unchanged.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@core-api/api/app_factory.py`:
- Around line 40-47: The Sentry initialization currently enables
send_default_pii=True which may leak user PII; update the sentry_sdk.init call
in app_factory (the call that uses settings.sentry_dsn, settings.api_env and
_sentry_filter_noise) to disable PII by default (set send_default_pii=False) or
wire it to a new config flag (e.g., settings.sentry_send_default_pii) so the
behavior is explicit and configurable, and ensure any change is accompanied by
documentation/notes about retaining PII only when explicitly enabled; keep the
_SENTRY_INITIALIZED flag and existing before_send=_sentry_filter_noise
unchanged.
In `@core-api/api/routers/calendar.py`:
- Around line 466-470: The current branch treats streams_marked <= 0 as a
dispatch failure and returns 503, but streams_marked == 0 can mean streams were
already queued/leased; change the logic to only raise the 503 for a genuine
failure (e.g., streams_marked < 0 or an explicit error indicator from the
dispatcher) and treat streams_marked == 0 as "already scheduled" returning HTTP
202. Update the code around the streams_marked check in the calendar sync
handler (the block using the streams_marked variable and the HTTPException) so
that streams_marked == 0 returns a 202 Accepted, while only truly negative/error
return values still produce the 503.
In `@core-api/api/routers/cron.py`:
- Around line 826-865: The code currently selects all active watches and then
applies batch_size later, causing full-fleet reads; change the logic so the
database queries enforce the oldest-first cap: modify the "push_subscriptions"
query (watch_result) to order by created_at ascending (or the appropriate
timestamp) and include a limit(batch_size) so only the candidate window is
returned, then compute connection_ids from that limited active_watches set and
pass only those IDs into the "connection_sync_state" query (state_result) and
its .in_("connection_id", ...) call; ensure WATCH_PROVIDER_TO_SYNC_KIND is still
used for the sync_kind filter so you only fetch sync state for the capped
candidate window.
In `@core-api/api/routers/webhooks.py`:
- Around line 29-34: The Pub/Sub JWT cache (_PUBSUB_JWT_CACHE) is unbounded and
can grow on bursts; change it to be a bounded LRU-like cache similar to the
Gmail dedup logic: replace the raw dict with an OrderedDict (or wrap with an
eviction policy) and add a max size constant (e.g.,
_PUBSUB_JWT_CACHE_MAX_ENTRIES), ensure all accesses/updates use
_PUBSUB_JWT_CACHE_LOCK, on insert move keys to the end and, while len > MAX,
popitem(last=False) to evict oldest; keep TTL expiration logic when reading but
also enforce size-based eviction on writes to avoid O(n) sweeps and unbounded
memory growth.
In `@core-api/api/routers/workers.py`:
- Around line 437-449: The current lease completion treats result.status ==
"skipped" as success; instead, update the _run_with_stream_lease handling so
that only explicit success (e.g., status == "ok") leads to
complete_connection_sync_lease, and any permanent-failure signals emitted by
_sync_single_gmail or _sync_single_calendar (currently using status="skipped")
are routed to the error path using _fail_with_backoff (or the existing
fail/quarantine flow) so they accumulate retries/quarantine appropriately;
locate the status check in _run_with_stream_lease and change the conditional to
call _fail_with_backoff (or the fail/quarantine helper) for non-success statuses
(including skipped when used for auth/API failure) while preserving true no-op
skipped behavior where upstream functions emit a distinct marker for no-op.
In `@core-api/api/services/auth.py`:
- Around line 135-140: The log lines in add_email_account that call logger.info
and logger.warning are emitting full provider_email; replace provider_email with
a redacted form (e.g., call the project’s email-redaction helper such as
redact_email(provider_email) or mask the local part) so raw user addresses are
not logged; update both the info/warning at the shown block and the similar
occurrences around lines 249-254 (same logger calls) to use the redacted value
instead of provider_email.
- Around line 107-141: The current logic treats a False return from
mark_stream_dirty as a non-fatal warning and continues, which can leave a new
account without an initial sync; update the code around the
gmail_marked/calendar_marked checks to treat any False as a retryable failure by
propagating an error (e.g., raise a specific retryable exception or return an
error status) instead of logging and returning success—ensure the change uses
the existing mark_stream_dirty symbol and includes provider_email in the error
message for context; apply the same fix to the analogous block referenced at
lines 218-255 so both initial-sync scheduling failures are surfaced to the
caller for retry.
In `@core-api/api/services/syncs/stream_worker.py`:
- Around line 229-269: The current flow stops the heartbeat via
stop_connection_sync_lease_heartbeat() before finalizing the lease, which can
leave the claim stranded if fail_connection_sync_lease() or
complete_connection_sync_lease() block or raise; change the logic to finalize
the lease first (call maybe_quarantine_failed_connection() and on error path
call fail_connection_sync_lease(), on success extract result_cursor via
worker_router._extract_result_cursor() and call
complete_connection_sync_lease()), and then in a finally block call
stop_connection_sync_lease_heartbeat(heartbeat_stop, heartbeat_thread) if
heartbeat_stop and heartbeat_thread are not None; ensure all return paths still
return the original result or quarantine_result as before but always stop the
heartbeat in finally.
In `@core-api/api/services/syncs/watch_manager.py`:
- Around line 452-461: When channel_id or resource_id is missing in the
Calendar-watch failure path, mark the DB watch row inactive (e.g., set is_active
= False and/or set a quarantine flag or error_reason) and persist the change
before returning the failure to prevent future renew attempts from repeatedly
picking the malformed record; update the logger to reflect the row was
deactivated/quarantined. Apply the same fix to the duplicate block later in the
file (the other branch that returns the same failure around the 788-798 area).
Use the existing symbols channel_id, resource_id, is_active, user_id and the
watch row object/ORM save method in watch_manager.py to locate and update the
code paths, and ensure the DB write succeeds (log or raise on DB errors).
In `@core-api/api/services/webhooks/gmail_webhook.py`:
- Around line 132-234: reconcile_gmail_connection returns failure dicts with
"status" and "message" only, but _sync_single_gmail expects an "error" key;
update reconcile_gmail_connection so every non-success return includes an
"error" field (e.g., same value as "message") — specifically the returns where
it returns {"status":"error","message":...}, {"status":"skipped","message":...},
and the fallback-recovery branches where it returns those dicts — so callers
like _sync_single_gmail can read result.get("error") and preserve the provider
error text.
In `@core-api/tests/unit/test_token_encryption.py`:
- Around line 17-24: Remove the module-level os.environ.setdefault calls and
instead create a pytest fixture (e.g., supabase_env) or use the built-in
monkeypatch fixture inside the tests in test_token_encryption.py to set
SUPABASE_URL and SUPABASE_ANON_KEY for each test; update tests that rely on
those env vars to use the fixture or call monkeypatch.setenv before importing or
invoking the code that reads the vars so the overrides are scoped per-test and
do not mutate process-wide state during collection.
- Around line 323-326: Replace the incorrect patch of
api.services.google_auth._refresh_creds with a patch of
google.auth.transport.requests.Request so the test uses the mocked
Credentials.refresh flow; specifically, in the test around
_refresh_and_save_token where get_service_role_client and Credentials are
patched, remove the with patch('api.services.google_auth._refresh_creds') and
instead add with patch('google.auth.transport.requests.Request') so
credentials.refresh(Request()) in _refresh_and_save_token uses the patched
Request and no real HTTP call occurs, allowing the mocked credentials to
exercise the token encryption and DB write assertions.
In `@core-api/tests/unit/test_workspace_default_apps.py`:
- Line 8: The test expects "tasks" as a default app but production code's
VALID_APP_TYPES (in api/services/workspaces/apps.py) is missing "tasks"; update
the VALID_APP_TYPES tuple/sequence to include the string "tasks" so that
VALID_APP_TYPES contains ["chat","email","calendar","projects","files","tasks"]
(or equivalent order) ensuring functions that reference VALID_APP_TYPES validate
"tasks" correctly.
---
Outside diff comments:
In `@core-api/api/routers/email.py`:
- Around line 983-994: The query in the email sync path retrieves sensitive
token fields and calls decrypt_ext_connection_tokens() even though this endpoint
only needs to mark streams dirty; update the select on the Supabase query
produced by get_authenticated_supabase_client(user_jwt) to only request the
minimal fields (e.g., 'id' and 'provider'), remove the decryption step (do not
call decrypt_ext_connection_tokens on connections_result.data), and update the
downstream logic that builds google_connections and microsoft_connections to
operate on the lightweight connection objects (use c.get('provider') and
c.get('id') only) so no access_token/refresh_token handling or decryption is
performed in this path.
In `@core-api/api/routers/webhooks.py`:
- Around line 471-490: The webhook currently swallows infrastructure errors
inside _mark_microsoft_notification_dirty so microsoft_webhook_notification
always returns 202; change _mark_microsoft_notification_dirty to not convert
unexpected infra exceptions into {"success": False} but instead re-raise those
exceptions (or raise a specific transient error) while still returning
{"success": False} for known validation/data issues, and update
microsoft_webhook_notification to detect such raised exceptions (or catch the
specific transient error) so the outer try/except can return 503; apply the same
change for the analogous block handling notifications at the other occurrence
referenced in the comment.
In `@core-api/tests/unit/test_notify_attendees_google.py`:
- Around line 108-118: The test is unpacking a tuple but _update_google_event
(in update_event.py) returns a single bool; update the test in
test_notify_attendees_google.py to stop tuple-unpacking: call
_update_google_event(...) into a single variable (e.g., ok) and assert that ok
is True (remove the second _tz variable), so the test matches the function
signature and return type.
- Around line 79-83: The test unpacks four values but _create_google_event
returns only three; update the unpacking in the test to expect three values
(event_id, meeting_link, err) by removing the unused _tz variable and adjust any
subsequent references to _tz (none here), ensuring the call to
_create_google_event and assertions (e.g., checking
recorder['insert']['sendUpdates'] and recorder['insert']['body']) remain
unchanged; search for other tests using _create_google_event to apply the same
3-value unpack fix if present.
---
Minor comments:
In `@core-api/api/app_factory.py`:
- Around line 144-150: In the health_check endpoint (function health_check)
replace the naive datetime usage datetime.utcnow().isoformat() + "Z" with a
timezone-aware call such as datetime.now(timezone.utc).isoformat() (or
datetime.now(tz=timezone.utc).isoformat()) and update imports to include
timezone from datetime so the returned timestamp is UTC-aware rather than naive.
In `@core-api/api/routers/email.py`:
- Around line 1034-1043: The route decorator still declares status_code=200
while the handler returns JSONResponse(status_code=202); update the route
decorator for this endpoint (the function with the route decorator above the
handler that returns JSONResponse(... status_code=202)) to declare
status_code=202 and adjust its OpenAPI responses/response_model metadata
accordingly so the generated schema/client reflect the asynchronous 202 contract
(add a 202 response entry and description if the decorator uses responses= or
response_model=).
In `@core-api/api/services/syncs/watch_manager.py`:
- Around line 36-42: _get_gmail_webhook_url currently calls getattr(settings,
"webhook_base_url", "").rstrip("/") which will raise if webhook_base_url exists
but is None; change the assignment to use a safe coalesce like
(getattr(settings, "webhook_base_url", None) or "").rstrip("/") so None becomes
an empty string before rstrip, and apply the same defensive pattern to the
companion helper(s) that build webhook URLs in this module.
In `@core-api/setup_pubsub_subscription.py`:
- Around line 85-89: Validate that webhook_base_url is non-empty after reading
os.getenv and rstrip("/") and fail fast if it's empty: when constructing
push_endpoint (using webhook_base_url and "/api/webhooks/gmail") check
webhook_base_url and raise a clear exception or exit with an error log so you
don't silently configure a broken subscription; update the same validation where
webhook_base_url is used later (the push_endpoint creation around the code
referencing lines ~149-150) to reuse the same check or helper to ensure
consistency.
In
`@core-api/supabase/migrations/20260402000001_skip_redundant_dirty_mark_writes.sql`:
- Around line 90-107: The upsert currently sets last_provider_event_at =
COALESCE(p_provider_event_at, v_now), which can move the timestamp backwards;
change the assignment in the ON CONFLICT DO UPDATE block to keep
last_provider_event_at monotonic by taking the maximum of the existing value and
the incoming value, e.g. set last_provider_event_at =
GREATEST(public.connection_sync_state.last_provider_event_at,
COALESCE(p_provider_event_at, v_now)), so the stored last_provider_event_at only
ever advances (handles NULLs via COALESCE).
In `@core-api/tests/unit/test_share_link_access.py`:
- Around line 37-38: The test fake currently stores ("ilike", field, value) but
matches it by equality; change the fake's filter evaluation to emulate SQL ILIKE
wildcard semantics: when encountering a filter tuple with operator "ilike" (from
self._filters), convert the SQL pattern (value) where % -> .* and _ -> . into a
regex (properly escaping other chars), then perform a case-insensitive regex
match against the record[field] (use re.IGNORECASE) so patterns like f"{query}%"
behave like SQL ILIKE; update the code path that iterates self._filters and
evaluates records (the function that consumes self._filters) to use this
regex-based matching for "ilike" entries.
In `@core-api/tests/unit/test_stream_worker.py`:
- Around line 346-355: Pin the SYNC_WORKER_POLL_SECONDS env var before invoking
stream_worker.run_forever in test_run_forever_recovers_from_run_once_exception
so the sleep assertion is not flaky; set the env to "5" (e.g. using
monkeypatch.setenv("SYNC_WORKER_POLL_SECONDS", "5") or
os.environ["SYNC_WORKER_POLL_SECONDS"]="5") prior to calling
stream_worker.run_forever and restore/cleanup afterwards; keep the existing
patches for stream_worker.build_worker_id, stream_worker.run_once, and
stream_worker.time.sleep.
In `@core-api/tests/unit/test_token_encryption.py`:
- Around line 61-80: Replace realistic-looking Google token strings used in the
tests with clearly synthetic placeholders: update the plaintexts in the tests
that call encrypt_token and decrypt_token (notably the variables in
test_produces_unique_ciphertext, test_decrypt_plaintext_passthrough, and the
earlier test using original = "ya29...") to non-sensitive, obviously fake values
(e.g., "fake_token_1", "fake_token_2" or "plaintext_token_placeholder") so the
tests still verify non-Fernet passthrough and unique ciphertext behavior while
avoiding secret-scanner false positives.
---
Nitpick comments:
In `@core-api/api/app_factory.py`:
- Around line 51-65: The global_exception_handler is declared async but contains
no awaitable operations; change it to a synchronous function by removing the
async keyword from the global_exception_handler definition and keep its body and
return of JSONResponse unchanged, then register it via
app.add_exception_handler(Exception, global_exception_handler) as currently done
(ensure the function signature remains global_exception_handler(request:
Request, exc: Exception) and that Sentry capture and logging calls remain
synchronous).
In `@core-api/api/services/syncs/sync_state_store.py`:
- Around line 193-244: The closure _heartbeat_loop in
start_connection_sync_lease_heartbeat does loop control, RPC I/O, lease-loss
handling and logging; extract the inner tick (the body of the while loop that
calls heartbeat_connection_sync_lease and handles the alive/exception paths)
into a small helper function (e.g., _heartbeat_tick or
heartbeat_connection_sync_lease_tick) that accepts service_supabase,
connection_id, sync_kind, worker_id, lease_seconds and logger (and optionally
stop_event) and returns a boolean/enum indicating whether to continue or stop;
then simplify _heartbeat_loop to just call that helper on each interval and act
on its return value, keeping start_connection_sync_lease_heartbeat otherwise
unchanged.
In `@core-api/sync_worker.py`:
- Around line 9-17: The logging.basicConfig call is executed at import time
causing global side effects; move it into a dedicated main() entrypoint so
configuration happens at runtime. Create a main() function that calls
logging.basicConfig(...) then run_forever(), and update the if __name__ ==
"__main__": block to call main(); reference logging.basicConfig and run_forever
to locate the code to change.
In `@core-api/tests/unit/test_google_webhook_security.py`:
- Around line 13-43: Add a unit test to cover the edge case where a secret is
configured but the provided token is an empty string: in
core-api/tests/unit/test_google_webhook_security.py create a new test (e.g.,
test_calendar_channel_token_rejects_empty_string_when_secret_configured) that
monkeypatches api.config.settings.google_calendar_webhook_secret to a non-empty
value, imports verify_google_calendar_channel_token, and asserts
verify_google_calendar_channel_token("conn-1", "channel-1", "") is False; this
uses the existing verify_google_calendar_channel_token symbol to ensure
empty-string tokens are rejected when a secret is set.
In `@core-api/tests/unit/test_sync_dispatcher.py`:
- Around line 60-64: Replace the exotic generator-throw mock for
queue_client.enqueue_sync_for_connection with a simple callable that raises
directly: locate the queue_client SimpleNamespace in the test and change the
enqueue_sync_for_connection implementation to a normal function (or lambda) that
raises AssertionError("enqueue should not run when mark dirty fails") when
invoked so the test intent is clearer and SonarCloud warning is resolved.
In `@core-api/tests/unit/test_sync_state_store.py`:
- Around line 105-116: Refactor the
test_get_reconcile_interval_seconds_clamps_invalid_values to use explicit
(value, expected) tuples in the pytest.mark.parametrize decorator instead of
computing `expected` inside the test; update the test signature to accept both
parameters and assert get_reconcile_interval_seconds() equals the provided
expected; this targets the test for the get_reconcile_interval_seconds function
and removes the conditional logic currently inside the test for clarity and
future maintainability.
In `@core-api/tests/unit/test_workers_mode_dispatch.py`:
- Around line 381-412: The test
test_run_with_stream_lease_quarantines_when_retry_cap_is_hit has deeply nested
patch.contexts; replace the 6-level nested patch.object/patch calls that target
workers.get_service_role_client, workers.claim_connection_sync_lease,
workers.get_connection_sync_state,
api.services.syncs.failure_policy.complete_connection_sync_lease,
workers.fail_connection_sync_lease, and
api.services.syncs.failure_policy.deactivate_connection_with_subscriptions with
a flattened approach (use contextlib.ExitStack to enter all patches and capture
returned mock objects into variables, or use pytest-mock fixtures like
mocker.patch to create the mocks) so the call to workers._run_with_stream_lease
remains the same and assertions use the captured complete_mock, fail_mock, and
deactivate_mock. Ensure behavior and return_value setups are preserved (e.g.,
MAX_FAILURE_RETRY_COUNT env, claim_connection_sync_lease ->
{"connection_id":"conn-1","provider":"microsoft"}, get_connection_sync_state ->
{"retry_count":2}) and that complete_mock.assert_called_once(),
fail_mock.assert_not_called(), and deactivate_mock.assert_called_once() remain.
In `@core-api/webhooks_index.py`:
- Around line 3-8: Remove the import-time sys.path mutation in
webhooks_index.py: delete the sys.path.insert(...) line and rely on
package-style imports so create_webhooks_app from api.app_factory is resolved
via normal Python package/module resolution; ensure the package is installed or
the project is executed with the correct working directory or via python -m so
the import of create_webhooks_app succeeds without modifying sys.path at import
time.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 277ead35-399a-47cb-8001-6e2a11a3d5e8
📒 Files selected for processing (57)
core-api/api/app_factory.pycore-api/api/config.pycore-api/api/routers/calendar.pycore-api/api/routers/cron.pycore-api/api/routers/email.pycore-api/api/routers/webhooks.pycore-api/api/routers/workers.pycore-api/api/services/auth.pycore-api/api/services/syncs/failure_policy.pycore-api/api/services/syncs/google_error_utils.pycore-api/api/services/syncs/google_services.pycore-api/api/services/syncs/stream_worker.pycore-api/api/services/syncs/sync_dispatcher.pycore-api/api/services/syncs/sync_gmail.pycore-api/api/services/syncs/sync_gmail_cron.pycore-api/api/services/syncs/sync_google_calendar.pycore-api/api/services/syncs/sync_google_calendar_cron.pycore-api/api/services/syncs/sync_state_store.pycore-api/api/services/syncs/watch_manager.pycore-api/api/services/webhooks/__init__.pycore-api/api/services/webhooks/calendar_webhook.pycore-api/api/services/webhooks/gmail_webhook.pycore-api/index.pycore-api/lib/google_retry.pycore-api/lib/google_webhook_security.pycore-api/setup_pubsub_subscription.pycore-api/supabase/migrations/20260401000001_enforce_single_active_push_subscription.sqlcore-api/supabase/migrations/20260401000002_connection_sync_state.sqlcore-api/supabase/migrations/20260402000001_skip_redundant_dirty_mark_writes.sqlcore-api/supabase/migrations/20260402000002_next_reconcile_at.sqlcore-api/supabase/migrations/20260403000001_reconcile_claim_mode_and_jitter.sqlcore-api/supabase/migrations/20260403000002_connection_sync_state_autovacuum.sqlcore-api/sync_worker.pycore-api/tests/unit/test_auth_initial_sync_queue.pycore-api/tests/unit/test_calendar_sync_dirty_only.pycore-api/tests/unit/test_cron_batch_mode_toggle.pycore-api/tests/unit/test_cron_google_oauth_failures.pycore-api/tests/unit/test_cron_watch_health.pycore-api/tests/unit/test_email_sync_queue_only.pycore-api/tests/unit/test_failure_policy.pycore-api/tests/unit/test_google_webhook_security.pycore-api/tests/unit/test_image_proxy.pycore-api/tests/unit/test_notify_attendees_google.pycore-api/tests/unit/test_share_link_access.pycore-api/tests/unit/test_stream_worker.pycore-api/tests/unit/test_sync_dispatcher.pycore-api/tests/unit/test_sync_google_calendar_cron_tokens.pycore-api/tests/unit/test_sync_state_store.pycore-api/tests/unit/test_token_encryption.pycore-api/tests/unit/test_token_encryption_regressions.pycore-api/tests/unit/test_user_profile_helpers.pycore-api/tests/unit/test_watch_manager_permanent_failures.pycore-api/tests/unit/test_webhook_dirty_marking.pycore-api/tests/unit/test_workers_mode_dispatch.pycore-api/tests/unit/test_workspace_default_apps.pycore-api/tests/unit/test_workspace_invitations_service.pycore-api/webhooks_index.py
Comment on lines
+40
to
+47
| sentry_sdk.init( | ||
| dsn=settings.sentry_dsn, | ||
| environment=settings.api_env, | ||
| traces_sample_rate=0.05, | ||
| send_default_pii=True, | ||
| before_send=_sentry_filter_noise, | ||
| ) | ||
| _SENTRY_INITIALIZED = True |
There was a problem hiding this comment.
Reconsider send_default_pii=True for privacy compliance.
Enabling send_default_pii causes Sentry to automatically capture user identifiers, IP addresses, and potentially email addresses from request contexts. This may conflict with GDPR/CCPA requirements and should be explicitly justified or disabled.
🛡️ Suggested fix
sentry_sdk.init(
dsn=settings.sentry_dsn,
environment=settings.api_env,
traces_sample_rate=0.05,
- send_default_pii=True,
+ send_default_pii=False,
before_send=_sentry_filter_noise,
)If PII capture is intentional for debugging, consider documenting this decision and ensuring appropriate data retention policies are in place.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| sentry_sdk.init( | |
| dsn=settings.sentry_dsn, | |
| environment=settings.api_env, | |
| traces_sample_rate=0.05, | |
| send_default_pii=True, | |
| before_send=_sentry_filter_noise, | |
| ) | |
| _SENTRY_INITIALIZED = True | |
| sentry_sdk.init( | |
| dsn=settings.sentry_dsn, | |
| environment=settings.api_env, | |
| traces_sample_rate=0.05, | |
| send_default_pii=False, | |
| before_send=_sentry_filter_noise, | |
| ) | |
| _SENTRY_INITIALIZED = True |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@core-api/api/app_factory.py` around lines 40 - 47, The Sentry initialization
currently enables send_default_pii=True which may leak user PII; update the
sentry_sdk.init call in app_factory (the call that uses settings.sentry_dsn,
settings.api_env and _sentry_filter_noise) to disable PII by default (set
send_default_pii=False) or wire it to a new config flag (e.g.,
settings.sentry_send_default_pii) so the behavior is explicit and configurable,
and ensure any change is accompanied by documentation/notes about retaining PII
only when explicitly enabled; keep the _SENTRY_INITIALIZED flag and existing
before_send=_sentry_filter_noise unchanged.
Comment on lines
+466
to
+470
| if streams_marked <= 0: | ||
| raise HTTPException( | ||
| status_code=status.HTTP_503_SERVICE_UNAVAILABLE, | ||
| detail="Calendar sync is temporarily unavailable. Please try again shortly.", | ||
| ) |
There was a problem hiding this comment.
Don't turn an already-queued sync into a 503.
With the new dirty-write dedupe, streams_marked == 0 can happen when every calendar stream is already dirty or leased. In that case this branch reports "temporarily unavailable" even though the sync is already queued/running. Please distinguish "already scheduled" from real dispatch failure and keep the former as 202.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@core-api/api/routers/calendar.py` around lines 466 - 470, The current branch
treats streams_marked <= 0 as a dispatch failure and returns 503, but
streams_marked == 0 can mean streams were already queued/leased; change the
logic to only raise the 503 for a genuine failure (e.g., streams_marked < 0 or
an explicit error indicator from the dispatcher) and treat streams_marked == 0
as "already scheduled" returning HTTP 202. Update the code around the
streams_marked check in the calendar sync handler (the block using the
streams_marked variable and the HTTPException) so that streams_marked == 0
returns a 202 Accepted, while only truly negative/error return values still
produce the 503.
Comment on lines
+826
to
+865
| watch_result = service_supabase.table("push_subscriptions")\ | ||
| .select( | ||
| "id, ext_connection_id, provider, created_at, expiration, " | ||
| "notification_count, last_notification_at" | ||
| )\ | ||
| .in_("provider", list(WATCH_PROVIDER_TO_SYNC_KIND.keys()))\ | ||
| .eq("is_active", True)\ | ||
| .execute() | ||
|
|
||
| active_watches = watch_result.data or [] | ||
| if not active_watches: | ||
| duration = (datetime.now(timezone.utc) - start_time).total_seconds() | ||
| capture_checkin(monitor_slug="watch-health", check_in_id=check_in_id, status=MonitorStatus.OK) | ||
| return { | ||
| "status": "completed", | ||
| "message": "No active Google watches", | ||
| "duration_seconds": duration, | ||
| "checked": 0, | ||
| "queued": 0, | ||
| "errors": 0, | ||
| } | ||
|
|
||
| connection_ids = sorted( | ||
| { | ||
| str(watch["ext_connection_id"]) | ||
| for watch in active_watches | ||
| if watch.get("ext_connection_id") | ||
| } | ||
| ) | ||
|
|
||
| state_by_key: Dict[tuple[str, str], Dict[str, Any]] = {} | ||
| if connection_ids: | ||
| state_result = service_supabase.table("connection_sync_state")\ | ||
| .select( | ||
| "connection_id, provider, sync_kind, dirty, retry_count, " | ||
| "next_retry_at, lease_expires_at, last_sync_finished_at" | ||
| )\ | ||
| .in_("connection_id", connection_ids)\ | ||
| .in_("sync_kind", list(WATCH_PROVIDER_TO_SYNC_KIND.values()))\ | ||
| .execute() |
There was a problem hiding this comment.
watch-health still reads the whole fleet before applying the cap.
batch_size is enforced only after loading every active Google watch and every matching sync-state row. On a large tenant, a single recovery run still scales with the entire fleet even if it only queues 25 marks. Push the oldest-first limit/pagination into the queries and fetch sync state only for that candidate window.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@core-api/api/routers/cron.py` around lines 826 - 865, The code currently
selects all active watches and then applies batch_size later, causing full-fleet
reads; change the logic so the database queries enforce the oldest-first cap:
modify the "push_subscriptions" query (watch_result) to order by created_at
ascending (or the appropriate timestamp) and include a limit(batch_size) so only
the candidate window is returned, then compute connection_ids from that limited
active_watches set and pass only those IDs into the "connection_sync_state"
query (state_result) and its .in_("connection_id", ...) call; ensure
WATCH_PROVIDER_TO_SYNC_KIND is still used for the sync_kind filter so you only
fetch sync state for the capped candidate window.
Comment on lines
+29
to
+34
| _PUBSUB_JWT_CACHE: Dict[str, tuple[dict, float]] = {} | ||
| _PUBSUB_JWT_CACHE_LOCK = threading.Lock() | ||
| _GMAIL_INGRESS_DEDUP_TTL_SECONDS = 120.0 | ||
| _GMAIL_INGRESS_DEDUP_MAX_ENTRIES = 20_000 | ||
| _GMAIL_INGRESS_DEDUP_CACHE: "OrderedDict[str, float]" = OrderedDict() | ||
| _GMAIL_INGRESS_DEDUP_LOCK = threading.Lock() |
There was a problem hiding this comment.
Bound the Pub/Sub JWT cache before this ingress sees real traffic.
_PUBSUB_JWT_CACHE is keyed by raw tokens and only prunes expired entries. A burst of unique tokens can therefore grow memory until expiry and make every insert do an O(n) sweep. Give it the same kind of size cap/LRU eviction you already added for Gmail dedup.
🛠️ Suggested fix
_PUBSUB_JWT_CACHE: Dict[str, tuple[dict, float]] = {}
_PUBSUB_JWT_CACHE_LOCK = threading.Lock()
+_PUBSUB_JWT_CACHE_MAX_ENTRIES = 4_096
@@
with _PUBSUB_JWT_CACHE_LOCK:
_PUBSUB_JWT_CACHE[encoded_token] = (claims, expiry)
expired_tokens = [
token
for token, (_, token_expiry) in _PUBSUB_JWT_CACHE.items()
if token_expiry <= now
]
for token in expired_tokens:
_PUBSUB_JWT_CACHE.pop(token, None)
+ while len(_PUBSUB_JWT_CACHE) > _PUBSUB_JWT_CACHE_MAX_ENTRIES:
+ _PUBSUB_JWT_CACHE.pop(next(iter(_PUBSUB_JWT_CACHE)))Also applies to: 117-149
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@core-api/api/routers/webhooks.py` around lines 29 - 34, The Pub/Sub JWT cache
(_PUBSUB_JWT_CACHE) is unbounded and can grow on bursts; change it to be a
bounded LRU-like cache similar to the Gmail dedup logic: replace the raw dict
with an OrderedDict (or wrap with an eviction policy) and add a max size
constant (e.g., _PUBSUB_JWT_CACHE_MAX_ENTRIES), ensure all accesses/updates use
_PUBSUB_JWT_CACHE_LOCK, on insert move keys to the end and, while len > MAX,
popitem(last=False) to evict oldest; keep TTL expiration logic when reading but
also enforce size-based eviction on writes to avoid O(n) sweeps and unbounded
memory growth.
Comment on lines
+17
to
+24
| # Prevent module import failures from lib.supabase_client singleton init. | ||
| os.environ.setdefault("SUPABASE_URL", "https://test.supabase.co") | ||
| os.environ.setdefault( | ||
| "SUPABASE_ANON_KEY", | ||
| "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9." | ||
| "eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6InRlc3QiLCJyb2xlIjoiYW5vbiJ9." | ||
| "testsignature", | ||
| ) |
There was a problem hiding this comment.
Scope the SUPABASE_* overrides to each test.
Setting these vars at module import time mutates process-wide state during collection. If this file imports before another config test, those tests will no longer observe missing SUPABASE_* env vars and can pass or fail based on order alone.
🧪 Suggested fixture
-# Prevent module import failures from lib.supabase_client singleton init.
-os.environ.setdefault("SUPABASE_URL", "https://test.supabase.co")
-os.environ.setdefault(
- "SUPABASE_ANON_KEY",
- "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."
- "eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6InRlc3QiLCJyb2xlIjoiYW5vbiJ9."
- "testsignature",
-)
+@pytest.fixture(autouse=True)
+def _supabase_env(monkeypatch):
+ monkeypatch.setenv("SUPABASE_URL", "https://test.supabase.co")
+ monkeypatch.setenv(
+ "SUPABASE_ANON_KEY",
+ "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."
+ "eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6InRlc3QiLCJyb2xlIjoiYW5vbiJ9."
+ "testsignature",
+ )🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@core-api/tests/unit/test_token_encryption.py` around lines 17 - 24, Remove
the module-level os.environ.setdefault calls and instead create a pytest fixture
(e.g., supabase_env) or use the built-in monkeypatch fixture inside the tests in
test_token_encryption.py to set SUPABASE_URL and SUPABASE_ANON_KEY for each
test; update tests that rely on those env vars to use the fixture or call
monkeypatch.setenv before importing or invoking the code that reads the vars so
the overrides are scoped per-test and do not mutate process-wide state during
collection.
Comment on lines
+323
to
+326
| with patch('api.services.google_auth.get_service_role_client', return_value=mock_supabase): | ||
| with patch('api.services.google_auth.Credentials', return_value=mock_credentials): | ||
| with patch('api.services.google_auth._refresh_creds'): | ||
| from api.services.google_auth import _refresh_and_save_token |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
google_auth_file="$(fd 'google_auth\.py$' | head -n1)"
echo "== Inspecting ${google_auth_file} =="
rg -n -C4 --type=py '\bdef _refresh_and_save_token\b|\b_refresh_creds\b|Credentials\(|\.refresh\s*\(' "$google_auth_file"
sed -n '1,260p' "$google_auth_file"Repository: 10xapp/core-oss
Length of output: 10231
🏁 Script executed:
cat -n core-api/tests/unit/test_token_encryption.py | sed -n '315,335p'Repository: 10xapp/core-oss
Length of output: 1167
🏁 Script executed:
cat -n core-api/tests/unit/test_token_encryption.py | sed -n '280,330p'Repository: 10xapp/core-oss
Length of output: 2556
🏁 Script executed:
cat -n core-api/tests/unit/test_token_encryption.py | sed -n '1,50p'Repository: 10xapp/core-oss
Length of output: 1902
🏁 Script executed:
rg -n 'def _refresh_creds|_refresh_creds\(' core-api/api/services/google_auth.pyRepository: 10xapp/core-oss
Length of output: 41
🏁 Script executed:
rg -n 'from google.auth.transport.requests import Request' core-api/api/services/google_auth.pyRepository: 10xapp/core-oss
Length of output: 112
Patch the non-existent _refresh_creds function that causes the test to fail.
Line 325 patches api.services.google_auth._refresh_creds, which does not exist in the module. The actual code invokes credentials.refresh(Request()) to perform the token refresh. Replace this patch with:
with patch('google.auth.transport.requests.Request'):This prevents the actual HTTP request during testing while allowing the mocked credentials object to proceed through the refresh flow and validate that tokens are encrypted before database writes.
🧰 Tools
🪛 GitHub Actions: CI
[error] 325-325: TestWritePathEncryption.test_google_auth_refresh_encrypts_before_update failed: AttributeError: api.services.google_auth does not have the attribute '_refresh_creds' (patch target missing)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@core-api/tests/unit/test_token_encryption.py` around lines 323 - 326, Replace
the incorrect patch of api.services.google_auth._refresh_creds with a patch of
google.auth.transport.requests.Request so the test uses the mocked
Credentials.refresh flow; specifically, in the test around
_refresh_and_save_token where get_service_role_client and Credentials are
patched, remove the with patch('api.services.google_auth._refresh_creds') and
instead add with patch('google.auth.transport.requests.Request') so
credentials.refresh(Request()) in _refresh_and_save_token uses the patched
Request and no real HTTP call occurs, allowing the mocked credentials to
exercise the token encryption and DB write assertions.
| from api.services.workspaces.crud import create_workspace | ||
|
|
||
| EXPECTED_DEFAULT_APPS = ["chat", "email", "calendar", "projects", "files"] | ||
| EXPECTED_DEFAULT_APPS = ["chat", "email", "calendar", "projects", "files", "tasks"] |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check the current definition of VALID_APP_TYPES
rg -n "VALID_APP_TYPES" --type py -C 3Repository: 10xapp/core-oss
Length of output: 2406
🏁 Script executed:
#!/bin/bash
# Get the complete VALID_APP_TYPES definition
sed -n '14,25p' core-api/api/services/workspaces/apps.pyRepository: 10xapp/core-oss
Length of output: 199
Add "tasks" to VALID_APP_TYPES in api/services/workspaces/apps.py.
The test expects "tasks" as a valid default app, but it is missing from the VALID_APP_TYPES tuple in the production code. The assertion at line 110 will fail because "tasks" is not in VALID_APP_TYPES.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@core-api/tests/unit/test_workspace_default_apps.py` at line 8, The test
expects "tasks" as a default app but production code's VALID_APP_TYPES (in
api/services/workspaces/apps.py) is missing "tasks"; update the VALID_APP_TYPES
tuple/sequence to include the string "tasks" so that VALID_APP_TYPES contains
["chat","email","calendar","projects","files","tasks"] (or equivalent order)
ensuring functions that reference VALID_APP_TYPES validate "tasks" correctly.
- test_notify_attendees_google: unpack 3 values from _create_google_event (core-oss doesn't return resolved_tz), single bool from _update_google_event - test_workspace_default_apps: remove 'tasks' from expected default apps (todos/tasks not in core-oss) - google_auth: port _refresh_creds and concurrent refresh race guard from prod
There was a problem hiding this comment.
🧹 Nitpick comments (1)
core-api/api/services/google_auth.py (1)
358-362: Consider usingcredentials.expiryfrom the refreshed credentials.After refresh,
credentials.expirycontains the actual expiration time from Google. The sibling function ingoogle_services.py(lines 115-121) uses this value with a fallback, which is more accurate than always assuming 1 hour.♻️ Suggested improvement
# Perform the refresh (with retry for transient transport errors) _refresh_creds(credentials, Request(), context=connection_id[:8] if connection_id else '') - # Calculate new expiry time - new_expires_at = datetime.now(timezone.utc) + timedelta(seconds=DEFAULT_TOKEN_LIFETIME_SECONDS) + # Use actual expiry from Google credentials (make timezone-aware if needed) + if credentials.expiry: + if credentials.expiry.tzinfo is None: + new_expires_at = credentials.expiry.replace(tzinfo=timezone.utc) + else: + new_expires_at = credentials.expiry + else: + # Fallback to 1 hour if expiry not provided + new_expires_at = datetime.now(timezone.utc) + timedelta(seconds=DEFAULT_TOKEN_LIFETIME_SECONDS)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/services/google_auth.py` around lines 358 - 362, The code is setting new_expires_at using DEFAULT_TOKEN_LIFETIME_SECONDS instead of the actual refreshed token expiry; after calling _refresh_creds(credentials, Request(), context=...), read credentials.expiry and use it as the new_expires_at (with a fallback to datetime.now(timezone.utc) + timedelta(seconds=DEFAULT_TOKEN_LIFETIME_SECONDS) if credentials.expiry is None) so that the expiry reflects the value returned by Google (mirroring the approach used in the sibling function in google_services.py).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@core-api/api/services/google_auth.py`:
- Around line 358-362: The code is setting new_expires_at using
DEFAULT_TOKEN_LIFETIME_SECONDS instead of the actual refreshed token expiry;
after calling _refresh_creds(credentials, Request(), context=...), read
credentials.expiry and use it as the new_expires_at (with a fallback to
datetime.now(timezone.utc) + timedelta(seconds=DEFAULT_TOKEN_LIFETIME_SECONDS)
if credentials.expiry is None) so that the expiry reflects the value returned by
Google (mirroring the approach used in the sibling function in
google_services.py).
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 3bcabb58-434f-4219-8535-b75ae1b3675b
📒 Files selected for processing (1)
core-api/api/services/google_auth.py
CodeQL fix: - Redact provider_email in all auth.py log lines (ja***@gmail.com) Code review fixes: - calendar/email routers: streams_marked==0 with connections returns 202 (already scheduled), not 503 - webhooks: bound Pub/Sub JWT cache with 4096 entry LRU eviction - workers: route 'skipped' status through error/fail path instead of completing the lease - stream_worker: move heartbeat stop to finally block after lease finalization to prevent stranded claims - watch_manager: deactivate malformed Calendar watches (missing channel_id/resource_id) instead of leaving is_active=true permanently - gmail_webhook: add 'error' field to all reconcile failure return dicts for worker result parsing compatibility - auth: raise RuntimeError on mark_stream_dirty failure so callers see the initial sync scheduling failure Tests updated to match new behavior.
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
core-api/api/routers/email.py (1)
983-994:⚠️ Potential issue | 🟠 MajorDon't decrypt connection tokens in
/api/email/sync.This handler only uses
idandprovider, but it now selectsaccess_token/refresh_tokenand decrypts every active connection first. That widens plaintext secret exposure on a user-facing path and can make scheduling fail even when the dirty-mark itself would have succeeded.🔐 Suggested fix
auth_supabase = get_authenticated_supabase_client(user_jwt) service_supabase = get_service_role_client() connections_result = auth_supabase.table('ext_connections')\ - .select('id, provider, provider_email, access_token, refresh_token, token_expires_at, metadata, is_primary')\ + .select('id, provider')\ .eq('user_id', user_id)\ .eq('is_active', True)\ .in_('provider', ['google', 'microsoft'])\ .execute() - connections = [decrypt_ext_connection_tokens(c) for c in (connections_result.data or [])] + connections = connections_result.data or [] google_connections = [c for c in connections if c.get('provider') == 'google'] microsoft_connections = [c for c in connections if c.get('provider') == 'microsoft']🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/routers/email.py` around lines 983 - 994, The handler is unnecessarily selecting and decrypting tokens; update the query and post-processing so only non-secret fields are fetched and no decryption occurs: remove access_token, refresh_token, token_expires_at from the .select(...) call on auth_supabase.table('ext_connections'), stop calling decrypt_ext_connection_tokens on the results (use connections_result.data or its safe subset directly), and keep the existing filtering into google_connections and microsoft_connections based only on the provider field (and other non-secret fields like id/provider_email/is_primary/metadata if needed).core-api/api/routers/webhooks.py (1)
475-487:⚠️ Potential issue | 🟠 MajorDon't ACK retryable Microsoft scheduling failures.
_mark_microsoft_notification_dirty()converts infrastructure problems into{"success": False}, butmicrosoft_webhook_notification()ignoresresultsand still returns202. A transient Supabase/dirty-mark failure therefore drops the notification permanently because Graph never sees a non-2xx response.Also applies to: 497-570
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/routers/webhooks.py` around lines 475 - 487, The handler currently awaits _mark_microsoft_notification_dirty(notification) for each item but ignores its return values and always responds 202, which silences transient failures; instead, collect each result from _mark_microsoft_notification_dirty in microsoft_webhook_notification and if any result indicates failure (e.g., result.get("success") is False or an error flag), return a non-2xx (5xx) response or raise an HTTP error so Microsoft will retry the webhook; apply the same logic to the other Microsoft webhook blocks referenced (lines ~497-570) so any infrastructure/dirty-mark failure surfaces as a non-ACK to trigger retries.
♻️ Duplicate comments (3)
core-api/api/routers/webhooks.py (1)
227-231:⚠️ Potential issue | 🟠 MajorRedact Gmail addresses in these new ingress logs.
These debug lines log raw
emailAddressvalues on the normal, duplicate, and no-match paths. That reintroduces user identifiers into logs on one of the hottest endpoints in the system.Also applies to: 245-250, 263-264
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/routers/webhooks.py` around lines 227 - 231, The debug logs currently include raw payload.get("emailAddress") values; update the logger.debug calls that print payload.get("emailAddress") (the three logger.debugs in webhooks.py that handle decoded Gmail payload / normal/duplicate/no-match paths) to redact or anonymize the address before logging (e.g., mask local-part or log a stable hash/obfuscated string). Locate the logger.debug invocations that reference payload.get("emailAddress") and replace the logged value with a redaction helper (e.g., redact_email(payload.get("emailAddress")) or hash_email(...)) so the rest of the debug message stays the same but no raw email addresses are written to logs.core-api/api/services/syncs/stream_worker.py (1)
186-227:⚠️ Potential issue | 🟠 MajorThe exception path still stops heartbeat too early.
This branch calls
stop_connection_sync_lease_heartbeat()beforemaybe_quarantine_failed_connection()/fail_connection_sync_lease(). If either of those operations stalls or raises, the claim can stay leased until expiry.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/services/syncs/stream_worker.py` around lines 186 - 227, The code currently stops the heartbeat too early; move the call to stop_connection_sync_lease_heartbeat(heartbeat_stop, heartbeat_thread) so it runs only after attempting maybe_quarantine_failed_connection(...) and fail_connection_sync_lease(...), or guard it so the heartbeat is kept active until those operations complete (use the same heartbeat_stop/heartbeat_thread symbols). Ensure you still stop the heartbeat in all exit paths (including after exceptions from quarantine/fail) to avoid leaks, and reference the functions stop_connection_sync_lease_heartbeat, maybe_quarantine_failed_connection, and fail_connection_sync_lease and the variables heartbeat_stop/heartbeat_thread when making the change.core-api/api/services/auth.py (1)
146-151:⚠️ Potential issue | 🟠 MajorThese
RuntimeErrors still never surface to the caller.
complete_oauth_flow()andadd_email_account()catch these exceptions and only log them, so a brand-new connection can still return success with no initial sync actually scheduled. Please either let the exception propagate or persist a retryable setup failure instead of swallowing it.Also applies to: 262-267
🧹 Nitpick comments (1)
core-api/tests/unit/test_watch_manager_permanent_failures.py (1)
226-242: Consider stricter assertion for single update call.The test correctly validates that malformed watch rows (missing
channel_id/resource_id) are deactivated withsuccess=False. The implementation performs exactly oneupdatecall in the malformed path.Suggested stricter assertion
# Malformed rows are now deactivated to prevent permanent self-recovery failure - query.update.assert_called() + query.update.assert_called_once()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/tests/unit/test_watch_manager_permanent_failures.py` around lines 226 - 242, The test should assert that exactly one deactivation update occurs: change the loose assertion query.update.assert_called() in test_stop_calendar_watch_missing_identifiers_deactivates_malformed_row to a stricter assertion that ensures a single call (e.g., use query.update.assert_called_once() or assert_called_once_with(expected args)) so the malformed-path in stop_calendar_watch triggers exactly one deactivation; locate the assertion in test_stop_calendar_watch_missing_identifiers_deactivates_malformed_row and update it to the single-call check.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@core-api/api/routers/calendar.py`:
- Around line 466-473: The branch that raises HTTPException when there are no
connections should return a 4xx client error instead of 503; replace
status.HTTP_503_SERVICE_UNAVAILABLE with an appropriate client error (e.g.
status.HTTP_400_BAD_REQUEST or status.HTTP_404_NOT_FOUND) in the HTTPException
raised in the block that checks `if not connections` (near `streams_marked`,
`connections`, `logger.info`) so callers get a client-side response indicating
they need to connect a calendar.
In `@core-api/api/routers/workers.py`:
- Around line 427-448: The heartbeat is being stopped before finalizing the
lease, which can leave the claim active if finalization (either
_fail_with_backoff or complete_connection_sync_lease) blocks or raises; change
the control flow in the worker run: after calling processor() handle exceptions
and non-success results by first calling _fail_with_backoff(...) or
complete_connection_sync_lease(...) as appropriate (use
_extract_result_cursor(result) for the cursor when completing), and only after
those finalization calls return or error should you call
stop_connection_sync_lease_heartbeat(heartbeat_stop, heartbeat_thread); ensure
exceptions from finalization are caught/logged but heartbeat is stopped in a
finally path.
In `@core-api/api/services/auth.py`:
- Around line 23-28: The _redact_email helper can raise on None or malformed
addresses; update it to first guard against falsy/non-string inputs and inputs
without a valid local part: if email is falsy or doesn't contain "@", return
"***"; otherwise split with rsplit("@",1), and if local is empty return
f"***@{domain}", if local length is 1 return f"{local[0]}***@{domain}", and if
length >1 preserve first two chars then mask (as current logic does) so calling
remove_email_account with None or "@example.com" won't raise.
---
Outside diff comments:
In `@core-api/api/routers/email.py`:
- Around line 983-994: The handler is unnecessarily selecting and decrypting
tokens; update the query and post-processing so only non-secret fields are
fetched and no decryption occurs: remove access_token, refresh_token,
token_expires_at from the .select(...) call on
auth_supabase.table('ext_connections'), stop calling
decrypt_ext_connection_tokens on the results (use connections_result.data or its
safe subset directly), and keep the existing filtering into google_connections
and microsoft_connections based only on the provider field (and other non-secret
fields like id/provider_email/is_primary/metadata if needed).
In `@core-api/api/routers/webhooks.py`:
- Around line 475-487: The handler currently awaits
_mark_microsoft_notification_dirty(notification) for each item but ignores its
return values and always responds 202, which silences transient failures;
instead, collect each result from _mark_microsoft_notification_dirty in
microsoft_webhook_notification and if any result indicates failure (e.g.,
result.get("success") is False or an error flag), return a non-2xx (5xx)
response or raise an HTTP error so Microsoft will retry the webhook; apply the
same logic to the other Microsoft webhook blocks referenced (lines ~497-570) so
any infrastructure/dirty-mark failure surfaces as a non-ACK to trigger retries.
---
Duplicate comments:
In `@core-api/api/routers/webhooks.py`:
- Around line 227-231: The debug logs currently include raw
payload.get("emailAddress") values; update the logger.debug calls that print
payload.get("emailAddress") (the three logger.debugs in webhooks.py that handle
decoded Gmail payload / normal/duplicate/no-match paths) to redact or anonymize
the address before logging (e.g., mask local-part or log a stable
hash/obfuscated string). Locate the logger.debug invocations that reference
payload.get("emailAddress") and replace the logged value with a redaction helper
(e.g., redact_email(payload.get("emailAddress")) or hash_email(...)) so the rest
of the debug message stays the same but no raw email addresses are written to
logs.
In `@core-api/api/services/syncs/stream_worker.py`:
- Around line 186-227: The code currently stops the heartbeat too early; move
the call to stop_connection_sync_lease_heartbeat(heartbeat_stop,
heartbeat_thread) so it runs only after attempting
maybe_quarantine_failed_connection(...) and fail_connection_sync_lease(...), or
guard it so the heartbeat is kept active until those operations complete (use
the same heartbeat_stop/heartbeat_thread symbols). Ensure you still stop the
heartbeat in all exit paths (including after exceptions from quarantine/fail) to
avoid leaks, and reference the functions stop_connection_sync_lease_heartbeat,
maybe_quarantine_failed_connection, and fail_connection_sync_lease and the
variables heartbeat_stop/heartbeat_thread when making the change.
---
Nitpick comments:
In `@core-api/tests/unit/test_watch_manager_permanent_failures.py`:
- Around line 226-242: The test should assert that exactly one deactivation
update occurs: change the loose assertion query.update.assert_called() in
test_stop_calendar_watch_missing_identifiers_deactivates_malformed_row to a
stricter assertion that ensures a single call (e.g., use
query.update.assert_called_once() or assert_called_once_with(expected args)) so
the malformed-path in stop_calendar_watch triggers exactly one deactivation;
locate the assertion in
test_stop_calendar_watch_missing_identifiers_deactivates_malformed_row and
update it to the single-call check.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: af46e242-7325-4998-9cea-cb0a43e7b8b4
📒 Files selected for processing (11)
core-api/api/routers/calendar.pycore-api/api/routers/email.pycore-api/api/routers/webhooks.pycore-api/api/routers/workers.pycore-api/api/services/auth.pycore-api/api/services/syncs/stream_worker.pycore-api/api/services/syncs/watch_manager.pycore-api/api/services/webhooks/gmail_webhook.pycore-api/tests/unit/test_calendar_sync_dirty_only.pycore-api/tests/unit/test_email_sync_queue_only.pycore-api/tests/unit/test_watch_manager_permanent_failures.py
🚧 Files skipped from review as they are similar to previous changes (3)
- core-api/tests/unit/test_calendar_sync_dirty_only.py
- core-api/tests/unit/test_email_sync_queue_only.py
- core-api/api/services/syncs/watch_manager.py
Comment on lines
+466
to
+473
| if streams_marked <= 0 and connections: | ||
| # streams_marked==0 with connections means already dirty/leased — still 202 | ||
| logger.info(f"ℹ️ All {len(connections)} calendar streams already scheduled for user {user_id[:8]}...") | ||
| elif not connections: | ||
| raise HTTPException( | ||
| status_code=status.HTTP_503_SERVICE_UNAVAILABLE, | ||
| detail="No active calendar connections found.", | ||
| ) |
There was a problem hiding this comment.
Return a 4xx when no calendar account is connected.
"No active calendar connections found" is a user/account state problem, not a transient backend outage. Returning 503 invites pointless retries instead of telling the caller they need to connect a calendar first.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@core-api/api/routers/calendar.py` around lines 466 - 473, The branch that
raises HTTPException when there are no connections should return a 4xx client
error instead of 503; replace status.HTTP_503_SERVICE_UNAVAILABLE with an
appropriate client error (e.g. status.HTTP_400_BAD_REQUEST or
status.HTTP_404_NOT_FOUND) in the HTTPException raised in the block that checks
`if not connections` (near `streams_marked`, `connections`, `logger.info`) so
callers get a client-side response indicating they need to connect a calendar.
Comment on lines
+427
to
+448
| try: | ||
| result = processor() | ||
| except Exception as exc: | ||
| stop_connection_sync_lease_heartbeat(heartbeat_stop, heartbeat_thread) | ||
| failure_result = _fail_with_backoff(str(exc)) | ||
| logger.error(f"[Worker] {sync_kind} leased run failed for {connection_id[:8]}...: {exc}") | ||
| return failure_result | ||
|
|
||
| stop_connection_sync_lease_heartbeat(heartbeat_stop, heartbeat_thread) | ||
|
|
||
| if result.get("status") in ("error", "skipped"): | ||
| return _fail_with_backoff(str(result.get("message") or result.get("error") or "sync failed")) | ||
|
|
||
| result_cursor = _extract_result_cursor(result) | ||
| complete_connection_sync_lease( | ||
| service_supabase, | ||
| connection_id, | ||
| sync_kind, | ||
| worker_id, | ||
| last_synced_cursor=result_cursor, | ||
| latest_seen_cursor=result_cursor, | ||
| ) |
There was a problem hiding this comment.
Finalize the lease before stopping the heartbeat.
The heartbeat is stopped before _fail_with_backoff() and before complete_connection_sync_lease(). If either finalization call blocks or raises, the claim can remain leased until timeout, delaying retries or letting duplicate work slip in.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@core-api/api/routers/workers.py` around lines 427 - 448, The heartbeat is
being stopped before finalizing the lease, which can leave the claim active if
finalization (either _fail_with_backoff or complete_connection_sync_lease)
blocks or raises; change the control flow in the worker run: after calling
processor() handle exceptions and non-success results by first calling
_fail_with_backoff(...) or complete_connection_sync_lease(...) as appropriate
(use _extract_result_cursor(result) for the cursor when completing), and only
after those finalization calls return or error should you call
stop_connection_sync_lease_heartbeat(heartbeat_stop, heartbeat_thread); ensure
exceptions from finalization are caught/logged but heartbeat is stopped in a
finally path.
Comment on lines
+23
to
+28
| def _redact_email(email: str) -> str: | ||
| """Partially redact an email for safe logging: ja***@gmail.com""" | ||
| if "@" not in email: | ||
| return "***" | ||
| local, domain = email.rsplit("@", 1) | ||
| return f"{local[:2]}***@{domain}" if len(local) > 2 else f"{local[0]}***@{domain}" |
There was a problem hiding this comment.
Make _redact_email() tolerate missing or malformed addresses.
remove_email_account() passes account.get("provider_email") into this helper. On legacy/null rows, _redact_email(None) or _redact_email("@example.com") raises after the delete has already succeeded, so the API can return 500 for a successful removal.
🛠️ Suggested fix
-def _redact_email(email: str) -> str:
+def _redact_email(email: Optional[str]) -> str:
"""Partially redact an email for safe logging: ja***@gmail.com"""
- if "@" not in email:
+ if not email or "@" not in email:
return "***"
local, domain = email.rsplit("@", 1)
- return f"{local[:2]}***@{domain}" if len(local) > 2 else f"{local[0]}***@{domain}"
+ if not local or not domain:
+ return "***"
+ prefix = local[:2] if len(local) > 2 else local[:1]
+ return f"{prefix}***@{domain}"🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@core-api/api/services/auth.py` around lines 23 - 28, The _redact_email helper
can raise on None or malformed addresses; update it to first guard against
falsy/non-string inputs and inputs without a valid local part: if email is falsy
or doesn't contain "@", return "***"; otherwise split with rsplit("@",1), and if
local is empty return f"***@{domain}", if local length is 1 return
f"{local[0]}***@{domain}", and if length >1 preserve first two chars then mask
(as current logic does) so calling remove_email_account with None or
"@example.com" won't raise.
…irty-mark failures Revert three changes that diverged from prod: - calendar/email: streams_marked==0 back to 503 (not 202) - workers/stream_worker: only route 'error' through fail path, not 'skipped' - auth: mark_stream_dirty failure logs warning only, does not raise
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
core-api/api/routers/email.py (1)
983-994:⚠️ Potential issue | 🟠 MajorDon't decrypt provider tokens just to dirty-mark streams.
This endpoint only uses
idandprovider, but Lines 985-992 fetch and decryptaccess_token/refresh_tokenfor every active connection. That widens secret exposure on the request path and adds needless CPU work.Suggested fix
- connections_result = auth_supabase.table('ext_connections')\ - .select('id, provider, provider_email, access_token, refresh_token, token_expires_at, metadata, is_primary')\ + connections_result = auth_supabase.table('ext_connections')\ + .select('id, provider')\ .eq('user_id', user_id)\ .eq('is_active', True)\ .in_('provider', ['google', 'microsoft'])\ .execute() - - connections = [decrypt_ext_connection_tokens(c) for c in (connections_result.data or [])] + + connections = connections_result.data or []🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/routers/email.py` around lines 983 - 994, The code is decrypting provider tokens unnecessarily; change the query on ext_connections (via get_authenticated_supabase_client / connections_result) to only select the fields needed (id and provider) and remove the decrypt_ext_connection_tokens call—use connections_result.data (or its safe default) directly to build google_connections and microsoft_connections by filtering on the provider field so access_token/refresh_token are never fetched or decrypted.
♻️ Duplicate comments (6)
core-api/api/services/syncs/stream_worker.py (1)
186-227:⚠️ Potential issue | 🟠 MajorKeep the heartbeat alive until exception finalization finishes.
Lines 187-188 stop the heartbeat before the quarantine/fail RPCs run. If either of those calls blocks or raises, the lease is stranded until expiry and retries get delayed.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/services/syncs/stream_worker.py` around lines 186 - 227, The heartbeat is being stopped too early — move the stop_connection_sync_lease_heartbeat(heartbeat_stop, heartbeat_thread) call so the lease heartbeat remains active while maybe_quarantine_failed_connection(...) and fail_connection_sync_lease(...) run; i.e., only call stop_connection_sync_lease_heartbeat after the quarantine/fail RPCs and the final logger.exception/return have completed, ensuring you still check heartbeat_stop and heartbeat_thread for None and wrap the stop call in its own try/except to avoid masking errors from quarantine/fail.core-api/api/services/auth.py (2)
115-149:⚠️ Potential issue | 🟠 MajorDon't report success when initial sync was never scheduled.
If these dirty-mark calls come back false, the OAuth/account-add flow still succeeds and the new connection can stay empty until some unrelated later trigger. This needs a retryable failure or a compensating fallback when no stream was scheduled.
Also applies to: 226-263
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/services/auth.py` around lines 115 - 149, The review points out that when mark_stream_dirty(...) for Gmail or Calendar returns False the flow still reports success; change the post-mark behavior in the auth flow to treat a partial/complete failure to schedule initial sync as an actual error: if either gmail_marked or calendar_marked is False, raise a retryable exception (e.g., a specific SyncSchedulingError) or invoke a compensating fallback (e.g., enqueue a retry job) instead of only logging a warning; update the code around the calls to mark_stream_dirty, the conditional that currently uses logger.info/logger.warning and the return path, and apply the same fix to the other identical block that currently handles initial sync (the duplicate at the other range) so new connections aren’t left unsynced.
23-28:⚠️ Potential issue | 🟠 MajorMake
_redact_email()null/malformed-safe.
remove_email_account()passesaccount.get("provider_email"), so Line 25 still throws onNone, and Line 28 throws on addresses like"@example.com". That can turn a successful delete into a 500 during logging.Suggested fix
-def _redact_email(email: str) -> str: +def _redact_email(email: Optional[str]) -> str: """Partially redact an email for safe logging: ja***@gmail.com""" - if "@" not in email: + if not email or "@" not in email: return "***" local, domain = email.rsplit("@", 1) - return f"{local[:2]}***@{domain}" if len(local) > 2 else f"{local[0]}***@{domain}" + if not domain: + return "***" + if not local: + return f"***@{domain}" + prefix = local[:2] if len(local) > 2 else local[:1] + return f"{prefix}***@{domain}"🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/services/auth.py` around lines 23 - 28, The _redact_email function needs to be null- and malformed-safe: update _redact_email to immediately return "***" if email is None or not a str, or if "@" is missing or rsplit("@",1) yields an empty domain/local; then safely compute local and domain, handling local lengths 0,1,2+ without indexing into an empty string (e.g., return "***" for empty local, f"{local[0]}***@{domain}" for length==1, f"{local[:2]}***@{domain}" for length>2, and f"{local}***@{domain}" for length==2). Ensure no operations assume non-empty parts so logging cannot raise on None or malformed inputs in remove_email_account.core-api/api/routers/workers.py (2)
437-449:⚠️ Potential issue | 🟠 MajorPermanent provider failures are still being completed as success.
The Google sync helpers return
status="skipped"for permanent auth/API failures, but_run_with_stream_lease()only fails leases onstatus == "error". That clears dirty state instead of accumulating retries/quarantine for broken connections.Also applies to: 484-487, 511-514, 558-561
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/routers/workers.py` around lines 437 - 449, The code currently treats only result.get("status") == "error" as a failure, so permanent provider failures returned as status == "skipped" get treated as success and clear dirty state; update the failure detection in _run_with_stream_lease (and the other identical result-handling blocks) to treat both "error" and "skipped" as failures for permanent auth/API errors — i.e., change the conditional that calls _fail_with_backoff to check for status in ("error", "skipped") (or detect a permanent failure flag in the result) and only call complete_connection_sync_lease when status indicates a genuine success; keep using _extract_result_cursor and _fail_with_backoff/complete_connection_sync_lease but ensure skipped permanent failures do not call complete_connection_sync_lease so retries/quarantine accumulate.
427-449:⚠️ Potential issue | 🟠 MajorFinalize the lease before stopping its heartbeat.
Lines 430 and 435 tear down the heartbeat before
_fail_with_backoff()/complete_connection_sync_lease(). If finalization fails, the claim can sit leased until timeout.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/routers/workers.py` around lines 427 - 449, The heartbeat is being stopped before the lease is finalized—move the stop_connection_sync_lease_heartbeat(heartbeat_stop, heartbeat_thread) call so the heartbeat remains active until after lease finalization: in the exception path, call _fail_with_backoff(...) and then perform any lease-finalization (e.g., complete_connection_sync_lease or other cleanup) while the heartbeat is still running, and only after that call stop_connection_sync_lease_heartbeat; likewise, in the success path, call complete_connection_sync_lease(...) first and then stop_connection_sync_lease_heartbeat; keep references to processor, stop_connection_sync_lease_heartbeat, complete_connection_sync_lease, _fail_with_backoff, heartbeat_stop and heartbeat_thread to locate and reorder the calls.core-api/api/routers/calendar.py (1)
431-470:⚠️ Potential issue | 🟠 Major
streams_marked == 0is not always a backend outage.This branch currently turns three different outcomes into the same
503: no connected calendar account, stream already dirty/leased, and real dispatch failure. That misleads clients into retrying when sync is already queued/running or when the user just needs to connect an account first.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/routers/calendar.py` around lines 431 - 470, The code conflates three outcomes into a 503; change the loop to distinguish successes, skips, and dispatch failures by modifying mark_stream_dirty (or its caller contract) to return/raise distinct signals (e.g., SUCCESS, SKIPPED, FAILURE or raise on failure) and track counts: success_count, skipped_count, failure_count while iterating over connections_result.data using mark_stream_dirty(service_supabase, cid, provider, SYNC_KIND_CALENDAR, priority=MANUAL_SYNC_PRIORITY, metadata=...). After the loop: if no connections -> return a 400/appropriate client error indicating the user has no connected calendar; if success_count>0 -> return 200 with the number marked; if success_count==0 and skipped_count>0 -> return 200 indicating sync already queued/in-progress; only raise the 503 when failure_count>0 and success_count==0 (real dispatch failure). Ensure references to connections_result, mark_stream_dirty, streams_marked/SYNC_KIND_CALENDAR/MANUAL_SYNC_PRIORITY are updated accordingly.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@core-api/api/routers/email.py`:
- Around line 1024-1031: The current handling treats any falsey streams_marked
from mark_stream_dirty() as a dispatch failure and raises HTTP 503; instead,
distinguish three cases: if streams_marked < 0 (or whatever sentinel your
mark_stream_dirty() uses to indicate an actual dispatch error) keep logging an
error and raise HTTP 503; if streams_marked == 0 treat this as "already
scheduled/running" and return HTTP 202 Accepted (with a debug/info log that
includes the masked user_id); otherwise (streams_marked > 0) proceed as success.
Update the branch around mark_stream_dirty()/streams_marked, adjust
logger.error/logger.info messages to include user_id[:8], and replace the
unconditional HTTPException with the appropriate 202 response for the
already-queued case.
---
Outside diff comments:
In `@core-api/api/routers/email.py`:
- Around line 983-994: The code is decrypting provider tokens unnecessarily;
change the query on ext_connections (via get_authenticated_supabase_client /
connections_result) to only select the fields needed (id and provider) and
remove the decrypt_ext_connection_tokens call—use connections_result.data (or
its safe default) directly to build google_connections and microsoft_connections
by filtering on the provider field so access_token/refresh_token are never
fetched or decrypted.
---
Duplicate comments:
In `@core-api/api/routers/calendar.py`:
- Around line 431-470: The code conflates three outcomes into a 503; change the
loop to distinguish successes, skips, and dispatch failures by modifying
mark_stream_dirty (or its caller contract) to return/raise distinct signals
(e.g., SUCCESS, SKIPPED, FAILURE or raise on failure) and track counts:
success_count, skipped_count, failure_count while iterating over
connections_result.data using mark_stream_dirty(service_supabase, cid, provider,
SYNC_KIND_CALENDAR, priority=MANUAL_SYNC_PRIORITY, metadata=...). After the
loop: if no connections -> return a 400/appropriate client error indicating the
user has no connected calendar; if success_count>0 -> return 200 with the number
marked; if success_count==0 and skipped_count>0 -> return 200 indicating sync
already queued/in-progress; only raise the 503 when failure_count>0 and
success_count==0 (real dispatch failure). Ensure references to
connections_result, mark_stream_dirty,
streams_marked/SYNC_KIND_CALENDAR/MANUAL_SYNC_PRIORITY are updated accordingly.
In `@core-api/api/routers/workers.py`:
- Around line 437-449: The code currently treats only result.get("status") ==
"error" as a failure, so permanent provider failures returned as status ==
"skipped" get treated as success and clear dirty state; update the failure
detection in _run_with_stream_lease (and the other identical result-handling
blocks) to treat both "error" and "skipped" as failures for permanent auth/API
errors — i.e., change the conditional that calls _fail_with_backoff to check for
status in ("error", "skipped") (or detect a permanent failure flag in the
result) and only call complete_connection_sync_lease when status indicates a
genuine success; keep using _extract_result_cursor and
_fail_with_backoff/complete_connection_sync_lease but ensure skipped permanent
failures do not call complete_connection_sync_lease so retries/quarantine
accumulate.
- Around line 427-449: The heartbeat is being stopped before the lease is
finalized—move the stop_connection_sync_lease_heartbeat(heartbeat_stop,
heartbeat_thread) call so the heartbeat remains active until after lease
finalization: in the exception path, call _fail_with_backoff(...) and then
perform any lease-finalization (e.g., complete_connection_sync_lease or other
cleanup) while the heartbeat is still running, and only after that call
stop_connection_sync_lease_heartbeat; likewise, in the success path, call
complete_connection_sync_lease(...) first and then
stop_connection_sync_lease_heartbeat; keep references to processor,
stop_connection_sync_lease_heartbeat, complete_connection_sync_lease,
_fail_with_backoff, heartbeat_stop and heartbeat_thread to locate and reorder
the calls.
In `@core-api/api/services/auth.py`:
- Around line 115-149: The review points out that when mark_stream_dirty(...)
for Gmail or Calendar returns False the flow still reports success; change the
post-mark behavior in the auth flow to treat a partial/complete failure to
schedule initial sync as an actual error: if either gmail_marked or
calendar_marked is False, raise a retryable exception (e.g., a specific
SyncSchedulingError) or invoke a compensating fallback (e.g., enqueue a retry
job) instead of only logging a warning; update the code around the calls to
mark_stream_dirty, the conditional that currently uses
logger.info/logger.warning and the return path, and apply the same fix to the
other identical block that currently handles initial sync (the duplicate at the
other range) so new connections aren’t left unsynced.
- Around line 23-28: The _redact_email function needs to be null- and
malformed-safe: update _redact_email to immediately return "***" if email is
None or not a str, or if "@" is missing or rsplit("@",1) yields an empty
domain/local; then safely compute local and domain, handling local lengths
0,1,2+ without indexing into an empty string (e.g., return "***" for empty
local, f"{local[0]}***@{domain}" for length==1, f"{local[:2]}***@{domain}" for
length>2, and f"{local}***@{domain}" for length==2). Ensure no operations assume
non-empty parts so logging cannot raise on None or malformed inputs in
remove_email_account.
In `@core-api/api/services/syncs/stream_worker.py`:
- Around line 186-227: The heartbeat is being stopped too early — move the
stop_connection_sync_lease_heartbeat(heartbeat_stop, heartbeat_thread) call so
the lease heartbeat remains active while maybe_quarantine_failed_connection(...)
and fail_connection_sync_lease(...) run; i.e., only call
stop_connection_sync_lease_heartbeat after the quarantine/fail RPCs and the
final logger.exception/return have completed, ensuring you still check
heartbeat_stop and heartbeat_thread for None and wrap the stop call in its own
try/except to avoid masking errors from quarantine/fail.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 8cd4c46a-3dbe-4775-8d2c-86fa0beb4f07
📒 Files selected for processing (7)
core-api/api/routers/calendar.pycore-api/api/routers/email.pycore-api/api/routers/workers.pycore-api/api/services/auth.pycore-api/api/services/syncs/stream_worker.pycore-api/tests/unit/test_calendar_sync_dirty_only.pycore-api/tests/unit/test_email_sync_queue_only.py
✅ Files skipped from review due to trivial changes (2)
- core-api/tests/unit/test_calendar_sync_dirty_only.py
- core-api/tests/unit/test_email_sync_queue_only.py
Comment on lines
+1024
to
+1031
| if streams_marked <= 0: | ||
| logger.error( | ||
| f"❌ Failed to mark any email streams dirty for user {user_id[:8]}..." | ||
| ) | ||
| raise HTTPException( | ||
| status_code=status.HTTP_503_SERVICE_UNAVAILABLE, | ||
| detail="Email sync is temporarily unavailable. Please try again shortly.", | ||
| ) |
There was a problem hiding this comment.
Treat “already scheduled” differently from dispatch failure.
A falsey mark_stream_dirty() here can mean the stream is already dirty or leased, so Line 1024 currently turns an in-flight sync into a 503. Reserve 503 for real dispatch failures and return 202 when the work is already queued/running.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@core-api/api/routers/email.py` around lines 1024 - 1031, The current handling
treats any falsey streams_marked from mark_stream_dirty() as a dispatch failure
and raises HTTP 503; instead, distinguish three cases: if streams_marked < 0 (or
whatever sentinel your mark_stream_dirty() uses to indicate an actual dispatch
error) keep logging an error and raise HTTP 503; if streams_marked == 0 treat
this as "already scheduled/running" and return HTTP 202 Accepted (with a
debug/info log that includes the masked user_id); otherwise (streams_marked > 0)
proceed as success. Update the branch around mark_stream_dirty()/streams_marked,
adjust logger.error/logger.info messages to include user_id[:8], and replace the
unconditional HTTPException with the appropriate 202 response for the
already-queued case.
There was a problem hiding this comment.
♻️ Duplicate comments (3)
core-api/api/services/auth.py (3)
23-28:⚠️ Potential issue | 🟠 MajorMake
_redact_email()tolerate null and malformed addresses.
remove_email_account()passesaccount.get("provider_email")here, so legacy/null rows can turn a successful delete into a 500 (NonehitsTypeError,"@example.com"hitsIndexError).Proposed fix
-def _redact_email(email: str) -> str: +def _redact_email(email: Optional[str]) -> str: """Partially redact an email for safe logging: ja***@gmail.com""" - if "@" not in email: + if not email or "@" not in email: return "***" local, domain = email.rsplit("@", 1) - return f"{local[:2]}***@{domain}" if len(local) > 2 else f"{local[0]}***@{domain}" + if not local: + return f"***@{domain}" if domain else "***" + prefix = local[:2] if len(local) > 1 else local[:1] + return f"{prefix}***@{domain}"🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/services/auth.py` around lines 23 - 28, The _redact_email function currently assumes a valid non-empty email and can raise TypeError/IndexError when passed None or malformed addresses; update _redact_email to defensively handle None, non-string inputs, and addresses missing a local part by first validating the input (e.g., if not isinstance(email, str) or "@" not in email or local part is empty) and returning a safe default like "***"; otherwise perform the existing redaction logic (referencing _redact_email and call sites such as remove_email_account) so legacy/null provider_email values no longer cause exceptions.
256-263:⚠️ Potential issue | 🟠 MajorMicrosoft initial-sync failures are still fully suppressed.
When dirty-marking fails, this helper now just falls through, and the warning is commented out. That makes lost initial-sync scheduling invisible for new Microsoft accounts.
Proposed fix
if email_marked and calendar_marked: - # logger.info(f"✅ [Microsoft] Initial sync scheduled for {_redact_email(provider_email)}") + logger.info(f"✅ [Microsoft] Initial sync scheduled for {_redact_email(provider_email)}") return - # logger.warning( - # f"⚠️ [Microsoft] Initial sync dirty-mark partial/failed for {_redact_email(provider_email)}. " - # f"email_marked={email_marked}, calendar_marked={calendar_marked}" - # ) + raise RuntimeError( + f"[Microsoft] Initial sync scheduling failed for {_redact_email(provider_email)} " + f"(email_marked={email_marked}, calendar_marked={calendar_marked})" + )🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/services/auth.py` around lines 256 - 263, The helper currently suppresses both success and failure logs by commenting them out, making initial-sync failures invisible; restore logging so successful initial-sync scheduling still logs (uncomment the logger.info that references _redact_email(provider_email)) and, importantly, uncomment and emit the logger.warning when email_marked or calendar_marked are not both true, including the redacted provider_email and the boolean flags email_marked and calendar_marked in the message so failed/partial dirty-mark attempts are visible; locate the block using the symbols provider_email, _redact_email, email_marked, and calendar_marked and re-enable those log statements at the appropriate levels.
142-149:⚠️ Potential issue | 🟠 MajorDon't treat Google initial-sync scheduling failure as success.
If either
mark_stream_dirty()call returnsFalse, this helper only warns and returns. With the inline fallback removed, that can leave a brand-new connection unsynced indefinitely.Proposed fix
if gmail_marked and calendar_marked: logger.info(f"✅ [Google] Initial sync scheduled for {_redact_email(provider_email)}") return - logger.warning( - f"⚠️ [Google] Initial sync dirty-mark partial/failed for {_redact_email(provider_email)}. " - f"gmail_marked={gmail_marked}, calendar_marked={calendar_marked}" + raise RuntimeError( + f"[Google] Initial sync scheduling failed for {_redact_email(provider_email)} " + f"(gmail_marked={gmail_marked}, calendar_marked={calendar_marked})" )🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@core-api/api/services/auth.py` around lines 142 - 149, The current helper treats partial failures (gmail_marked or calendar_marked being False) as non-fatal by logging a warning and returning, which can leave new Google connections unsynced; instead detect when either mark_stream_dirty() call failed and surface an error so the caller can retry/handle it: replace the warning-only path for when gmail_marked or calendar_marked is False with an error-level log (include _redact_email(provider_email) and the boolean flags) and then raise an exception (e.g., RuntimeError or a domain-specific exception) so the failure is not silently ignored and will trigger retry/cleanup by upstream code.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Duplicate comments:
In `@core-api/api/services/auth.py`:
- Around line 23-28: The _redact_email function currently assumes a valid
non-empty email and can raise TypeError/IndexError when passed None or malformed
addresses; update _redact_email to defensively handle None, non-string inputs,
and addresses missing a local part by first validating the input (e.g., if not
isinstance(email, str) or "@" not in email or local part is empty) and returning
a safe default like "***"; otherwise perform the existing redaction logic
(referencing _redact_email and call sites such as remove_email_account) so
legacy/null provider_email values no longer cause exceptions.
- Around line 256-263: The helper currently suppresses both success and failure
logs by commenting them out, making initial-sync failures invisible; restore
logging so successful initial-sync scheduling still logs (uncomment the
logger.info that references _redact_email(provider_email)) and, importantly,
uncomment and emit the logger.warning when email_marked or calendar_marked are
not both true, including the redacted provider_email and the boolean flags
email_marked and calendar_marked in the message so failed/partial dirty-mark
attempts are visible; locate the block using the symbols provider_email,
_redact_email, email_marked, and calendar_marked and re-enable those log
statements at the appropriate levels.
- Around line 142-149: The current helper treats partial failures (gmail_marked
or calendar_marked being False) as non-fatal by logging a warning and returning,
which can leave new Google connections unsynced; instead detect when either
mark_stream_dirty() call failed and surface an error so the caller can
retry/handle it: replace the warning-only path for when gmail_marked or
calendar_marked is False with an error-level log (include
_redact_email(provider_email) and the boolean flags) and then raise an exception
(e.g., RuntimeError or a domain-specific exception) so the failure is not
silently ignored and will trigger retry/cleanup by upstream code.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 473be3b1-bc8f-4e2f-98ea-caefca9436c0
📒 Files selected for processing (1)
core-api/api/services/auth.py
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
connection_sync_statecontrol plane for sync executionrealtimevsreconcileclaim modes) for Railway deploymentapp_factory.pyfrom monolithicindex.pynext_reconcile_atreplacing broad sweep reconciliationfailure_policy.pyNew files
api/app_factory.py— shared FastAPI app factoryapi/services/syncs/sync_dispatcher.py— dispatch layer for dirty markingapi/services/syncs/sync_state_store.py— lease claim/heartbeat/complete/failapi/services/syncs/stream_worker.py— Railway worker loopapi/services/syncs/failure_policy.py— quarantine logiclib/google_webhook_security.py— HMAC token signinglib/google_retry.py— Google API retry helperssync_worker.py/webhooks_index.py— Railway entrypointsconnection_sync_statetable + RPCsTest plan
test_notify_attendees_google) — unrelated calendar feature driftSummary by CodeRabbit
New Features
Bug Fixes
Refactor