The factory now picks between two backends based on
settings.storage_backend ("s3" default, "local" for dev / single-node /
air-gapped deployments). The new factory helper build_storage_config()
takes one of the 4 PURPOSE_BUCKETS ("workspace" | "version" |
"run_log" | "trash") and returns the kwargs for create_storage(...).
s3 mode: AsyncStorageBackend over an S3-compatible service
(S3_WORKSPACE_BUCKET etc. as separate buckets).
local mode: AsyncStorageBackend over on-disk files; the 4 buckets
become subdirectories of LOCAL_STORAGE_BASE_DIR (default
"/data"), so the same 4-bucket layout works in both modes.
Concretely:
- common/config.py: add storage_backend (default "s3") +
local_storage_base_dir (default "/data").
- common/storage/factory.py: add PURPOSE_BUCKETS constant +
build_storage_config(bucket_name) helper.
- backend/main.py + backend/storage_api.py: lifespan collapses the
4-instance construction into one dict comprehension:
app.state.object_stores = {
name: create_storage(build_storage_config(name))
for name in PURPOSE_BUCKETS
}
(was 4x ~10-line dicts, one per bucket).
- runtime/mount.py: when STORAGE_BACKEND=local, skip the rclone mount
entirely (the shared docker volume at LOCAL_STORAGE_BASE_DIR is the
store; runtime reads directly).
- docker-compose.yml: mount the shared local-storage volume at /data
in both backend and runtime containers.
- .env.example: document STORAGE_BACKEND + LOCAL_STORAGE_BASE_DIR.
Dependencies added to support both backends:
- aiofiles>=25.1.0 (local async I/O) to backend + common + runtime.
- aioboto3>=15.5.0 (async S3) to common.
- uv.lock regenerated.
After this commit, both modes deploy end-to-end. The s3 mode is the
production default; local mode is opt-in via STORAGE_BACKEND=local.
schedule:
- New _execution_loop runs alongside _database_event_loop. It claims
job.node.execute rows, sets a 30-min lease on available_at, then
dispatches each as asyncio.create_task under a Semaphore(N).
Polling loop is back to sub-millisecond turnaround for
schedule.run.requested and job.node.finished. Long notebook
execution no longer blocks DAG advance events.
- _process_pending_events filters by event_type IN
('schedule.run.requested', 'job.node.finished'); the executor
loop owns job.node.execute exclusively.
- _process_outbox_event builds a plain dict envelope before
handler dispatch; the previous ORM-row handoff risked
DetachedInstanceError once the outer session closed.
- _sync_once uses get_job + reschedule_job for existing job ids
instead of add_job(replace_existing=True). Each cron schedule
no longer removed-and-readded every 5s.
- service.py threads settings.schedule_execution_concurrency into
the orchestrator (default 4).
common:
- create_async_engine gets explicit pool_size=10, max_overflow=20,
pool_recycle=1800. No more relying on SQLAlchemy defaults.
- New schedule_execution_concurrency setting.
runtime:
- scan_workspaces: add missing 'import os' (NameError on startup)
and switch to asyncio.gather bounded by Semaphore(4) so N
workspaces start in parallel instead of sequentially.
Co-Authored-By: Claude <noreply@anthropic.com>
tao.chen