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>
- scripts.update_script: repoint script.current_object_id to the
newly uploaded StorageObject and best-effort delete the old
working copy. Previously the old row was mutated with the new
content_hash while the script still pointed at it, silently
losing user edits on the next publish_version.
- scripts.publish_version: read object bytes via the new
RustFSObjectStore.get_bytes() instead of the non-existent
.get_object(); publish 500'd on every call.
- scripts.upload_script / storage_api.complete_upload_record:
Path(file_name) raised NameError (only PurePosixPath imported),
crashing every upload and every upload finalization. Use
PurePosixPath.
- RustFSObjectStore: add get_bytes() helper (sync, body.close in
finally) for in-process callers that need raw bytes.
Co-Authored-By: Claude <noreply@anthropic.com>
- common.config.Settings: pydantic-settings with @lru_cache singleton;
all env vars now declared in one place (database / JWT / RUSTFS_*
credentials + 3 purpose-named buckets / workspace FS roots / etc.).
Replaces os.environ / os.getenv in backend / schedule / runtime /
common modules.
- storage_api: object_key layout flattens from
"{ws}/{usage_type}/{ulid}/{name}" to "{ws}/{ulid}". File name, type,
and logical path live in the StorageObjects / Scripts row, not in
the S3 key, so the bucket can be re-organised without a DB rewrite.
- storage_api: new BUCKET_FOR_USAGE map and resolve_bucket() helper
route uploads by usage_type to the right purpose-named bucket:
working_copy / public_script / data_resource / snapshot
→ RUSTFS_WORKSPACE_BUCKET (workspaces)
version_artifact
→ RUSTFS_VERSION_BUCKET (versions)
run_log / run_result
→ RUSTFS_RUN_LOG_BUCKET (run-logs)
workspace.artifact_bucket override wins over the default for that
workspace. Unknown usage_type falls through to the workspace bucket
so uploads are never silently dropped.
- backend.main lifespan: ensure_bucket loops over all three buckets at
startup.
- common.storage.schemas: extend usage_type Literal to include
working_copy / public_script (consumed by scripts.py after the local
FS removal).
- common.storage.client: raise StorageClientError / StorageUnavailable /
StorageRequestFailed instead of FastAPI HTTPException, so the client
is usable from non-FastAPI contexts (e.g. schedule worker). The
register_workspace_object method is removed (the local-FS path it
routed to no longer exists).
- common.pyproject.toml: add greenlet>=3.0.0 (SQLAlchemy 2.0 async
engine.dispose() requires it) and pydantic-settings>=2.14.2.
Verified: backend.main 57 routes; docker compose config; 20 SQLAlchemy
tables, 0 ForeignKey; grep os.environ / os.getenv in
backend|schedule|runtime|common = 0.
tao.chen