Files
model-platform/DEVELOP.md
T
tao.chen 309b657d35 docs: align with new storage architecture (s3 + local + server-proxied PUT)
All operator- and developer-facing docs updated to reflect:

  - The unified AsyncStorageBackend abstraction (s3 + local backends).
  - The STORAGE_BACKEND toggle ("s3" default, "local" for dev /
    single-node / air-gapped deployments).
  - The 4-purpose-bucket layout (workspace / version / run_log / trash)
    in both modes — 4 separate S3 buckets in s3 mode, 4 subdirectories
    of LOCAL_STORAGE_BASE_DIR in local mode.
  - The S3_* env var naming (was RUSTFS_*).
  - The server-proxied upload flow (was browser-direct presign-PUT):
    POST /internal/v1/uploads → PUT /internal/v1/uploads/{id} with
    raw bytes → server calls backend.put().
  - The factory helpers workspaces_root() (runtime's view of the
    workspace bucket on disk) and rclone_remote_spec() (s3-mode mount
    source).
  - The "two settings describing the same thing" cleanup: the deleted
    settings.workspace_root, settings.workspaces_root, and
    settings.remote_bucket fields.

Files touched:
  - API.md (§5 data-resource upload flow, §9 storage control plane,
    §10 readiness example)
  - ARCHITECTURE.md (storage layer diagram)
  - CLAUDE.md (architecture description + volume-preservation note)
  - DEVELOP.md (settings list, Storage section, "Wire a new bucket"
    how-to, dev-export example, troubleshooting network hint)
  - README.md (architecture diagram, container table, quick-start
    credentials note, tear-down note, Storage layout section)
  - REFACTOR_NOTES.md (final container list with s3 explanation)
  - backend/README.md (storage backend description)
  - migrations/data/README.md (step 11/12 record mentioning object
    storage)

A handful of historical "RustFS" mentions are intentionally retained
where they name a specific S3-compatible product (e.g. as an example
in REFACTOR_NOTES.md's container list) or document the pre-2026
abstraction name (DEVELOP.md Storage section).
2026-08-05 13:13:20 +08:00

17 KiB

DEVELOP.md — Developer Guide

This guide is for engineers working on the model platform codebase. For high-level design see ARCHITECTURE.md; for the current state of in-flight refactors see HANDOVER.md.

Code layout

common/                Pure-Python shared library
  config.py            Settings (pydantic-settings, lru_cache singleton)
  db/                  SQLAlchemy 2.0 async engine, session_scope, Base
  db/models/           26 tables in 9 domain files (zero FK, zero relationship)
  scheduler/           build_sqlalchemy_jobstore (delayed import)
  storage/             AsyncStorageBackend abstraction (s3 + local impls) + Pydantic schemas
  eventing.py          add_outbox_event / utcnow / event_time
  service_app.py       /health/ready TCP probe, /api/v1/health
  schemas.py           StrictModel base
  utils.py             get_free_port, start_process

backend/               Public FastAPI service + internal /internal/v1/* sub-app
  main.py              lifespan + route registration
  jupyter.py           /api/v1/auth/jupyter — the ONLY auth entry
  scripts.py           CRUD for scripts/notebooks (object storage via AsyncStorageBackend)
  schedules.py         DAG CRUD: schedules, nodes, edges
  schedule_runs.py     Trigger / list / get runs
  schedule_schemas.py  Pydantic request/response models
  admin.py             Admin endpoints
  resources.py         Misc data resources
  storage_api.py       /internal/v1/* (sub-app merged into main)
  storage_client.py    Stub (HTTP client removed post-migration; rewrite pending)
  schedule_client.py   Placeholder module (was the HTTP-push executor client)
  runtime_client.py    Self-contained httpx wrapper for the runtime
  jupyter.py           auth_request handler
  dependencies.py      request_context, database_session

schedule/              Schedule Executor (DAG worker)
  context.py           Constants + naive_utc
  scheduler.py         CronScheduler (APScheduler + 5s sync loop)
  orchestrator.py      DispatchOrchestrator (Outbox poll + DAG advance)
  worker.py            NodeExecutor (notebook / python execution)
  service.py           SchedulerService facade (composes the three)
  main.py              Lifespan + FastAPI app
  storage_client.py    Stub (SchedulerStorageClient rewrite pending — use AsyncStorageBackend directly)
  execution.py         execute_artifact (notebook + python paths)
  notebook_runner.py   Subprocess entry point (nbclient)

runtime/               Jupyter Runtime
  main.py              FastAPI entry: jupyter action endpoints
  process.py           Per-workspace subprocess pool + asyncio locks
  mount.py             rclone FUSE mount lifecycle

frontend/              React Router SPA (vite build → nginx)
  app/                 features/ routes/ services/ components/

migrations/            Alembic schema versions
docker-compose.yml     4 services
default.conf           Nginx template
scripts/nginx-entrypoint.sh
.env.example

Configuration system

All env vars go through one place: common/src/common/config.py.

from common.config import settings

settings.database_url          # str
settings.storage_backend       # str: "s3" (default) or "local"
settings.local_storage_base_dir  # str: root dir for storage data (default "/data"); see "Storage" below for per-mode derivation
settings.s3_endpoint            # str (full URL, e.g. "http://s3:9000"; s3 mode only)
settings.s3_access_key          # str (s3 mode only)
settings.s3_secret_key          # str (s3 mode only)
settings.s3_workspace_bucket    # str (s3 mode only)
settings.s3_version_bucket      # str (s3 mode only)
settings.s3_run_log_bucket      # str (s3 mode only)
settings.s3_trash_bucket        # str (s3 mode only)
settings.s3_trash_retention_days  # int (s3 mode only)
settings.jwt_secret             # HS256 secret for the auth_request handler
settings.backend_api_url       # schedule → backend HTTP base
settings.runtime_api_url       # backend → runtime HTTP base
settings.public_base_url       # runtime public base URL
settings.service_name          # surfaced in /health
settings.readiness_targets     # CSV host:port list for /health/ready

Settings reads from process env first, then from a .env file at CWD if present. pydantic-settings auto-loads. case_sensitive=False so DATABASE_URL / database_url both work.

Adding a new env var

  1. Add the field to Settings:
    new_var: str = Field(default="x", description="...")
    
  2. Add the line to .env.example with a comment.
  3. Use settings.new_var at the call site.

Do not call os.environ["NEW_VAR"] or os.getenv("NEW_VAR") in application code. The grep below should return zero hits:

grep -rnE 'os\.(environ\[?["\x27][A-Z_]+|getenv\(["\x27][A-Z_]+)' --include="*.py" \
  backend/src/ schedule/src/ runtime/src/ common/src/

Conventions

Database / SQLAlchemy

  • No foreign keys, no relationship — every join is explicit.
  • Every table has is_deleted TINYINT(1) NOT NULL DEFAULT 0 and deleted_at DATETIME(3) NULL; queries must filter is_deleted == 0 (or deleted_at.is_(None)) to avoid logical-deleted rows.
  • Domain files under common/src/common/db/models/ are split by bounded context: audit / events / experiments / identity / runtime / schedules / scripts / storage / workspaces.
  • All models are class X(Base) SQLAlchemy 2.0 declarative-mapped.
  • The full schema is in migrations/versions/. Apply with:
    uv run --frozen --package backend alembic upgrade head
    

Storage

  • All object bytes go through common.storage.AsyncStorageBackend, created by create_storage(config) from common.storage.factory.
  • Two backends are registered: local (filesystem, local mode) and s3 (S3-compatible service, s3 mode). Selection is per-deployment via settings.storage_backend ("s3" default, "local" for dev / single-node / air-gapped).
  • The factory helper build_storage_config(bucket_name) returns the right create_storage kwargs for each of the 4 purpose buckets (workspace, version, run_log, trash). Use it in lifespan code; route handlers don't see the difference.
  • Bucket resolution from usage_type is in one place (backend/storage_api.py:resolve_bucket); route handlers only know about app.state.object_stores[bucket_name].
  • The runtime's view of the workspace bucket on disk is exposed by common.storage.workspaces_root():
    • s3 mode: ${settings.local_storage_base_dir}/workspaces (default /data/workspaces, the rclone FUSE mount target).
    • local mode: ${settings.local_storage_base_dir}/workspace (default /data/workspace, a subdir of the shared local-storage volume). settings.local_storage_base_dir is the only path setting; the helper handles the per-mode suffix. Don't read settings.workspaces_root or any other path setting directly in runtime code — use this helper.
  • The pre-2026 abstraction (RustFSObjectStore / common.storage.client / StorageClient HTTP wrapper) is gone. Don't reintroduce it.

Auth

  • Browser → /jupyter/{workspace_id}/... → Nginx auth_requestGET /api/v1/auth/jupyter (Backend).
  • The handler:
    1. Parses cookie / Bearer JWT (HS256 + settings.jwt_secret).
    2. Verifies WorkspaceMembers for the workspace.
    3. Verifies Scripts.is_locked for the requested notebook path (owner / unlocked → allow; otherwise 403).
    4. Calls RuntimeClient.get_workspace / start_workspace.
    5. Returns x-upstream-addr + x-jupyter-internal-token response headers. Browser never holds the runtime token.
  • Nginx captures the headers via auth_request_set and proxies to the upstream sub-process with Authorization: token $jupyter_token.

Permission gates

  • For write operations on a script/notebook, call require_script_modify_access(script, user_id=..., is_admin=...) from backend/scripts.py. It enforces:
    • admin or owner → allow
    • non-owner, is_locked == 0 → allow
    • non-owner, is_locked == 1 → 403
  • Read endpoints (list_scripts, get_script) intentionally do not check is_locked — workspace members can see the script list.

Outbox events

  • The platform's only async-messaging fabric is the MySQL OutboxEvents table. Producers (Backend) write rows in the same transaction as the business state. Consumers (Schedule Executor) poll every 250 ms and update event_status to published or failed (with retry).
  • Use common.eventing.add_outbox_event to write.
  • event_type values currently in use:
    • schedule.run.requested — produced by schedule_runs.py / cron post-back
    • job.node.execute — produced by orchestrator when a node is ready
    • job.node.finished — produced by worker after node execution
  • consumer_inbox provides exactly-once delivery per (consumer_name, event_id) (with process_status: processing → succeeded lifecycle).

Async / sync signatures

  • SchedulerService.start() and SchedulerService.close() are async def (so the FastAPI lifespan can await them).
  • CronScheduler.start(), DispatchOrchestrator.start() are def (sync) — they only create_task(...) and return. Don't await them.
  • cron.start(), orchestrator.start(), worker.handle_node_execute are wired together in SchedulerService.__init__; their lifetime is owned by the facade.

Local development

One-time setup

# Python workspace (monorepo via uv workspaces)
uv sync --all-packages

# Frontend deps
cd frontend && pnpm install && cd ..

Per-service dev

# Backend (terminal 1)
export DATABASE_URL="mysql+asyncmy://model_platform:model_platform@127.0.0.1:3306/model_platform?charset=utf8mb4"
export STORAGE_BACKEND=s3
export S3_ACCESS_KEY=modelplatform
export S3_SECRET_KEY=modelplatformsecret
export S3_ENDPOINT=http://127.0.0.1:9000
# Or for local mode:
# export STORAGE_BACKEND=local
# export LOCAL_STORAGE_BASE_DIR=/data
uv run --frozen --package backend uvicorn backend.main:app --host 0.0.0.0 --port 8000 --reload

# Schedule Executor (terminal 2)
uv run --frozen --package schedule uvicorn schedule.main:app --host 0.0.0.0 --port 8001 --reload

# Runtime (terminal 3 — needs SYS_ADMIN, FUSE, devmode)
uv run --frozen --package runtime python -m runtime.main

Frontend dev

cd frontend
pnpm dev          # http://localhost:5173, proxies /api to backend
pnpm typecheck
pnpm build

Static checks

# Python compile
uv run --frozen --package backend python -m compileall -q backend/src common/src
uv run --frozen --package schedule python -m compileall -q schedule/src
uv run --frozen --package runtime python -m compileall -q runtime/src

# Type check (frontend)
cd frontend && pnpm typecheck && cd ..

# Docker compose config
docker compose config --quiet

Smoke test

# Run the full ORM import + Settings smoke test
PYTHONPATH="backend/src:common/src" uv run --frozen --package backend python -c "
from backend.main import app
from common.config import settings
print('backend:', len(app.routes), 'routes')
print('settings ok:', settings.s3_endpoint)
"

Common tasks

Add a new DAG endpoint

  1. Add the route handler in backend/schedules.py (DAG template) or backend/schedule_runs.py (run lifecycle).
  2. Validate request via backend/schedule_schemas.py.
  3. For mutations on nodes/edges/versions: route through create_script_record / get_script_row and apply require_script_modify_access if it touches a script.
  4. If it produces an outbox event, use add_outbox_event(session, event_type="...", producer="...", ...).

Add a new env var

See "Adding a new env var" above.

Add a new MySQL table

  1. Add a model class in common/src/common/db/models/<domain>.py. Include is_deleted TINYINT(1) NOT NULL DEFAULT 0 and deleted_at DATETIME(3) NULL.
  2. Export it from common/src/common/db/models/__init__.py.
  3. Generate the migration:
    uv run --frozen --package backend alembic revision --autogenerate -m "add <feature>"
    
  4. Review the generated migrations/versions/*.py — Alembic may miss comments / server defaults. Manually fix the migration.
  5. Apply locally:
    uv run --frozen --package backend alembic upgrade head
    

Wire a new storage bucket

The current 4 buckets are wired in backend/storage_api.py:resolve_bucket:

BUCKET_FOR_USAGE: dict[str, str] = {
    "working_copy":     settings.s3_workspace_bucket,
    "public_script":    settings.s3_workspace_bucket,
    "data_resource":    settings.s3_workspace_bucket,
    "snapshot":         settings.s3_workspace_bucket,
    "version_artifact": settings.s3_version_bucket,
    "run_log":          settings.s3_run_log_bucket,
    "run_result":       settings.s3_run_log_bucket,
}

The constant PURPOSE_BUCKETS = ("workspace", "version", "run_log", "trash") in common.storage.factory enumerates the four backends built in the backend lifespan. To add a fifth bucket:

  1. Add the env var to Settings (s3 mode only):
    s3_<feature>_bucket: str = Field(default="<feature>", description="...")
    
  2. Add to .env.example with a one-line comment.
  3. Append "<feature>" to the PURPOSE_BUCKETS tuple in common/storage/factory.py. build_storage_config("<feature>") will then automatically read settings.s3_<feature>_bucket (s3 mode) or use <local_storage_base_dir>/<feature> (local mode).
  4. Extend the Literal in common/storage/schemas.py (in CreateUploadRequest.usage_type, ServerObjectRequest.usage_type) to include the new value.
  5. Add an entry in BUCKET_FOR_USAGE mapping the new usage_type to the new bucket env var.
  6. Pre-create the bucket (s3 mode) or subdirectory (local mode) in the deployment. The backend no longer auto-creates buckets.

A workspace's artifact_bucket column (when non-null) overrides the default for that workspace, regardless of usage_type.

Add a new schedule node type

schedule/execution.py dispatches on script_type in execute_artifact. Add a new branch + a new _<type> function. worker.py does not need to change — the dispatch happens inside execute_artifact.

Tests

There is no formal test suite yet (see HANDOVER §Pending Tasks P1). A reasonable first test surface:

  • require_script_modify_access (admin / owner / non-owner-unlock / non-owner-lock): pure-function unit test, no DB.
  • validate_dag (cycle detection + orphan detection) in backend/schedules.py.
  • execute_artifact end-to-end with mocked content_hash and a real tempfile.TemporaryDirectory.

Test convention: pytest with pytest-asyncio for async def handlers. Use SQLite in-memory (or a MySQL test container) for DB integration. Use moto for S3.

Troubleshooting

"the greenlet library is required"

SQLAlchemy 2.0 needs greenlet for engine.dispose() in async contexts. Add greenlet>=3.0.0 to common/pyproject.toml and uv sync --all-packages. (Already present in this repo.)

"Can't connect to MySQL server"

Either MySQL isn't running, or the network namespace doesn't allow mysql:3306 resolution. Inside the Docker network, services reach each other by service name (mysql, backend, runtime, schedule, s3).

Jupyter routing 401s

Inspect docker compose logs backendjupyter.py:check_notebook_is_locked or load_active_membership will return an explicit reason. Then check the JWT (use JWT_SECRET from .env).

Schedule run never advances

schedule_runs.run_status is stuck at queued. Two likely causes:

  • outbox_events is empty (Backend's add_outbox_event failed — check add_outbox_event in schedule_runs.py).
  • The orchestrator's polling loop is dead. Check docker compose logs schedule and look for "database event loop failed" exceptions.

"AttributeError: 'SchedulerService' object has no attribute 'worker'"

worker must be constructed before orchestrator in SchedulerService.__init__, because orchestrator's dispatch table captures self.worker.handle_node_execute at construction time. See service.py — the order is load-bearing.

Style

  • Type hints everywhere (this repo uses from __future__ import annotations).
  • 4-space indent, double quotes, no trailing whitespace.
  • Comments are technical (explain why, not what).
  • Module docstrings document non-obvious invariants. Don't add docstrings to functions whose behavior is self-evident from the name.
  • 4 levels of indentation = "this function is doing too much; split it". (Project convention; see e.g. execute_artifact.)

See also

  • ARCHITECTURE.md — design diagrams
  • HANDOVER.md — current refactor state and pending work
  • CLAUDE.md — agent-facing conventions for the repo
  • models / __init__.py — exhaustive list of all 26 tables
  • common/config.py — all env vars in one place