mirror of
https://github.com/eigent-ai/eigent.git
synced 2026-10-19 17:03:28 +00:00
fix: let new Sessions start while Cloud history restores (#2028)
Some checks are pending
CodeQL Advanced / Analyze (actions) (push) Waiting to run
CodeQL Advanced / Analyze (javascript-typescript) (push) Waiting to run
CodeQL Advanced / Analyze (python) (push) Waiting to run
Pre-commit / pre-commit (push) Waiting to run
Test / Run Web + Local Brain Smoke (push) Waiting to run
Test / Run Frontend Guardrails (push) Waiting to run
Test / Run Browser Preview Electron Test (push) Waiting to run
Test / Run Python Tests (push) Waiting to run
Some checks are pending
CodeQL Advanced / Analyze (actions) (push) Waiting to run
CodeQL Advanced / Analyze (javascript-typescript) (push) Waiting to run
CodeQL Advanced / Analyze (python) (push) Waiting to run
Pre-commit / pre-commit (push) Waiting to run
Test / Run Web + Local Brain Smoke (push) Waiting to run
Test / Run Frontend Guardrails (push) Waiting to run
Test / Run Browser Preview Electron Test (push) Waiting to run
Test / Run Python Tests (push) Waiting to run
This commit is contained in:
parent
9c76fba833
commit
3a709a5230
5 changed files with 104 additions and 5 deletions
|
|
@ -654,7 +654,9 @@ async def get_workspace_session_model(
|
|||
"space_id": space_id,
|
||||
"project_id": project_id,
|
||||
"accepted": accepted,
|
||||
"restore_pending": is_default_cloud_history_bootstrap_pending(),
|
||||
"restore_pending": is_default_cloud_history_bootstrap_pending(
|
||||
project_id
|
||||
),
|
||||
}
|
||||
except Exception as exc:
|
||||
raise HTTPException(
|
||||
|
|
|
|||
|
|
@ -673,6 +673,10 @@ class CloudSyncWorker:
|
|||
self._bootstrap_lock = asyncio.Lock()
|
||||
self._bootstrap_attempt_count = 0
|
||||
self._bootstrap_next_attempt_at = 0.0
|
||||
# Projects that have Cloud history, once the list has been read,
|
||||
# and those whose history this restore has already imported.
|
||||
self._bootstrap_listed: frozenset[str] | None = None
|
||||
self._bootstrap_restored: set[str] = set()
|
||||
self._auth_paused_configuration: CloudSyncConfiguration | None = None
|
||||
self._auth_pause_code: str | None = None
|
||||
self._auth_retry_at = 0.0
|
||||
|
|
@ -694,9 +698,28 @@ class CloudSyncWorker:
|
|||
|
||||
return self._bootstrap_pending
|
||||
|
||||
def bootstrap_pending_for(self, project_id: str) -> bool:
|
||||
"""Whether Cloud history for one project may still be restoring.
|
||||
|
||||
Until the project list is read, any project may have history. After
|
||||
that, only listed projects that are not imported yet are pending; a
|
||||
project without Cloud history has nothing to wait for.
|
||||
"""
|
||||
|
||||
if not self._bootstrap_pending:
|
||||
return False
|
||||
listed = self._bootstrap_listed
|
||||
if listed is None:
|
||||
return True
|
||||
return (
|
||||
project_id in listed and project_id not in self._bootstrap_restored
|
||||
)
|
||||
|
||||
def configure(self, configuration: CloudSyncConfiguration) -> None:
|
||||
if configuration != self._configuration:
|
||||
self._bootstrap_pending = True
|
||||
self._bootstrap_listed = None
|
||||
self._bootstrap_restored = set()
|
||||
self._bootstrap_attempt_count = 0
|
||||
self._bootstrap_next_attempt_at = 0.0
|
||||
self._auth_paused_configuration = None
|
||||
|
|
@ -1042,6 +1065,11 @@ class CloudSyncWorker:
|
|||
raise RunEventSyncProtocolError(
|
||||
"invalid Run sync project descriptor"
|
||||
)
|
||||
if self._configuration == configuration:
|
||||
self._bootstrap_listed = frozenset(
|
||||
str(item["project_id"]) for item in project_items
|
||||
)
|
||||
for item in project_items:
|
||||
project_id = str(item["project_id"])
|
||||
# Snapshot and event paging may race a new ingest. Repeat until the
|
||||
# snapshot watermark matches the locally imported cursor.
|
||||
|
|
@ -1114,6 +1142,8 @@ class CloudSyncWorker:
|
|||
if target_cursor != int(snapshot.get("current_cursor", 0)):
|
||||
continue
|
||||
raise
|
||||
if self._configuration == configuration:
|
||||
self._bootstrap_restored.add(project_id)
|
||||
break
|
||||
else:
|
||||
raise RunEventSyncProtocolError(
|
||||
|
|
|
|||
|
|
@ -132,10 +132,21 @@ def notify_default_cloud_sync_worker() -> None:
|
|||
loop.call_soon_threadsafe(_default_command_worker.notify)
|
||||
|
||||
|
||||
def is_default_cloud_history_bootstrap_pending() -> bool:
|
||||
"""Report whether Cloud history is still restoring in the background."""
|
||||
def is_default_cloud_history_bootstrap_pending(
|
||||
project_id: str | None = None,
|
||||
) -> bool:
|
||||
"""Report whether Cloud history is still restoring in the background.
|
||||
|
||||
return bool(_default_worker and _default_worker.bootstrap_pending)
|
||||
With a project, report only whether that project's history may still be
|
||||
on its way, so a new Session need not wait for the whole account.
|
||||
"""
|
||||
|
||||
worker = _default_worker
|
||||
if worker is None:
|
||||
return False
|
||||
if project_id is None:
|
||||
return worker.bootstrap_pending
|
||||
return worker.bootstrap_pending_for(project_id)
|
||||
|
||||
|
||||
async def bootstrap_default_cloud_history() -> None:
|
||||
|
|
|
|||
|
|
@ -111,7 +111,9 @@ def test_session_model_recovery_uses_local_control_scope_and_preserves_unknown_r
|
|||
run_id="unaccepted", project_id="session-1", status="pending"
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
runtime, "is_default_cloud_history_bootstrap_pending", lambda: True
|
||||
runtime,
|
||||
"is_default_cloud_history_bootstrap_pending",
|
||||
lambda project_id=None: project_id == "session-1",
|
||||
)
|
||||
before = journal._connection.total_changes
|
||||
response = client.get(path, params=params, headers=_headers())
|
||||
|
|
|
|||
|
|
@ -925,6 +925,60 @@ async def test_worker_bootstraps_missing_history_without_upload_echo(journal):
|
|||
await worker.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_restore_keeps_only_projects_with_cloud_history_pending(journal):
|
||||
snapshot_requested = asyncio.Event()
|
||||
release_snapshot = asyncio.Event()
|
||||
|
||||
class SlowSnapshotTransport(FakeTransport):
|
||||
async def project_snapshot(self, configuration, project_id):
|
||||
snapshot_requested.set()
|
||||
await release_snapshot.wait()
|
||||
return await super().project_snapshot(configuration, project_id)
|
||||
|
||||
transport = SlowSnapshotTransport()
|
||||
transport.projects = [
|
||||
{
|
||||
"project_id": "project-cloud",
|
||||
"current_cursor": 0,
|
||||
"updated_at": "2026-08-06T00:00:02+00:00",
|
||||
}
|
||||
]
|
||||
transport.snapshots["project-cloud"] = {
|
||||
"project_id": "project-cloud",
|
||||
"current_cursor": 0,
|
||||
"runs": [],
|
||||
"recent_events": [],
|
||||
"events_truncated": False,
|
||||
}
|
||||
worker = _worker(journal, transport)
|
||||
|
||||
# Before the project list is read, any project may have history.
|
||||
assert worker.bootstrap_pending_for("project-new") is True
|
||||
|
||||
restore = asyncio.create_task(worker.bootstrap_once())
|
||||
await asyncio.wait_for(snapshot_requested.wait(), timeout=5)
|
||||
assert worker.bootstrap_pending
|
||||
assert worker.bootstrap_pending_for("project-cloud") is True
|
||||
assert worker.bootstrap_pending_for("project-new") is False
|
||||
|
||||
release_snapshot.set()
|
||||
await asyncio.wait_for(restore, timeout=5)
|
||||
assert not worker.bootstrap_pending
|
||||
assert worker.bootstrap_pending_for("project-cloud") is False
|
||||
|
||||
# New credentials start a new restore, so every project waits again.
|
||||
worker.configure(
|
||||
CloudSyncConfiguration(
|
||||
endpoint_url="https://example.test/api/v1/sync/events:ingest",
|
||||
authorization="Bearer rotated",
|
||||
desktop_instance_id="desk-1",
|
||||
)
|
||||
)
|
||||
assert worker.bootstrap_pending_for("project-new") is True
|
||||
await worker.close()
|
||||
|
||||
|
||||
def test_artifact_cloud_projection_removes_machine_local_paths():
|
||||
from app.run_sync.cloud_sync import _cloud_event_payload
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue