tao.chen/model-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: docs

Browse Source
This commit is contained in:
tao.chen
2026-08-25 14:17:05 +08:00
parent 76aa2dcb27
commit e805853b28
3 changed files with 440 additions and 259 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ARCHITECTURE.md
+4 -38
View File
@@ -1,39 +1,5 @@
# 简化系统架构 # Architecture
```text This file has been merged into [`DEVELOP.md`](./DEVELOP.md) — see
Browser [§Architecture](./DEVELOP.md#architecture) for the authoritative
| component diagram, capability map, container table, and storage layout.
v
Nginx Gateway (静态 React Router SPA + /api + /jupyter 代理)
|-----------------------------|
| /api/v1 | /jupyter/
v v
FastAPI Backend Shared Jupyter Server
| ^
| Runtime HTTP | Runtime 管理会话/票据
v |
Runtime Manager ---------------|
|
+------ MySQL（编辑租约、运行实例）
FastAPI Backend
| 1. 写 schedule_runs + outbox_events
| 2. 尝试 HTTP 立即推送
v
Schedule Executor（APScheduler）
|-- MySQL APSchedulerJobStore
|-- MySQL Outbox 轮询兜底
|-- DAG 节点执行与重试
|-- S3 日志/结果
+-- Backend 内部 Storage API
```
## 关键简化
1. 删除 Redis 服务、Redis Streams 和 Redis 文件锁。
2. 调度定义、运行记录、Outbox、Inbox、Cron JobStore 都由 MySQL 保存。
3. 立即运行采用 Backend -> Schedule Executor 内部 HTTP 推送；推送失败由 MySQL Outbox 轮询兜底。
4. Schedule Executor 自带 APScheduler，负责 Cron 触发和 DAG 执行。
5. 文件编辑锁改为 MySQL 租约，Runtime 单副本运行。
6. Jupyter 使用一个共享容器，工作区目录通过 Volume 挂载同步。
7. 前端改为 React Router SPA，并按 feature / route / service / component 分层。
DEVELOP.md
+423 -75
View File
@@ -1,8 +1,136 @@
# DEVELOP.md — Developer Guide # DEVELOP.md — Developer Guide
This guide is for engineers working on the model platform codebase. For This guide is for engineers working on the model platform codebase. It is the
high-level design see `ARCHITECTURE.md`; for the current state of in-flight authoritative source for architecture, code layout, configuration, conventions,
refactors see `HANDOVER.md`. and common tasks. `ARCHITECTURE.md` now only records the simplification
history; `HANDOVER.md` covers recent commits and pending work.
## Architecture
The platform is a self-hosted Jupyter model development environment: interactive
workspaces, DAG scheduling, object-store artifacts, and per-workspace runtimes
— all exposed through a single Nginx gateway.
### Capability map
| Capability | Where it lives |
|---|---|
| Workspace notebook editing, row-level lock | `backend/api/jupyter.py` + `scripts.is_locked` |
| Jupyter auth routing (browser never holds runtime token) | `nginx/default.conf` + `auth_request` + `backend/api/jupyter.py` |
| Object storage for notebook / script / version / run_log (s3 / local) | `common/storage/` + `backend/services/storage.py` |
| DAG scheduling: nodes, edges, cron, manual trigger, retry, snapshot | `backend/api/schedules/` + `backend/api/schedules/runs.py` + `schedule/` (5 layers) |
| DAG execution via MySQL Outbox (no Redis, no in-process queue) | `schedule/application/orchestrator.py` + `schedule/execution/worker.py` |
| Per-workspace Jupyter subprocess pool, asyncio lock | `runtime/process.py` |
| Runtime rclone FUSE mount of workspace bucket (s3 mode only) | `runtime/mount.py` |
| 18 MySQL tables, soft delete, zero FK, async SQLAlchemy 2.0 | `common/db/models/` |
### Component diagram
```
┌────────────────────┐
│ Browser (SPA) │
└─────────┬──────────┘
│ HTTPS / WS
┌─────────▼──────────┐
│ Nginx (only :80) │ ← templates/default.conf
│ /api/ /jupyter/ /storage/
└────┬───────┬──────┘
│ │
┌──────────────┘ └─────────────┐
▼ ▼
┌──────────────────┐ ┌──────────────────────┐
│ FastAPI Backend │ │ Runtime (Jupyter) │
│ + /internal/v1 │ control │ - rclone FUSE mount │
│ /objects │ token-Auth │ │
│ (storage) ├──────────────►│ - subprocess pool │
│ - DAG CRUD │ │ (per workspace) │
│ - script CRUD │ └──────────┬───────────┘
│ - auth_request │ │ FUSE / shared vol
│ - /api/v1/... │ ▼
└────┬──────┬──────┘ ┌──────────────────────┐
│ │ │ Object storage │
│ └──────── HTTP ───────►│ (s3: S3 service / │
▼ │ local: shared vol) │
┌────────────┐ │ 4 buckets per usage │
│ MySQL │◄───────── poll ─────│ │
│ - 18 tbls │ └──────────────────────┘
│ - outbox │
│ - jobstore │
└────┬───────┘
▲
│ outbox poll
┌────┴──────────────────────────┐
│ Schedule Executor │
│ - CronScheduler (APScheduler) │
│ - DispatchOrchestrator │
│ - NodeExecutor (worker) │
│ - SchedulerService (facade) │
└───────────────────────────────┘
```
### Services (docker-compose)
The architecture intentionally exposes only one host port (the gateway); all
other services are on the Docker internal network.
| Service | Image | Exposed | Purpose |
|---|---|---|---|
| `web` | `nginx:alpine` | host `:8888` → `:80` | SPA, `/api/` reverse proxy, `/jupyter/{ws}/` `auth_request` proxy, `/storage/` S3 passthrough (s3 mode only) |
| `backend` | `Dockerfile` | internal only | DAG CRUD, script CRUD, schedule trigger, `/api/v1/auth/jupyter`, `/internal/v1/objects` inter-service RPC (shared `INTERNAL_SERVICE_TOKEN`, see `§Auth`) |
| `runtime` | `Dockerfile` | internal only | Per-workspace Jupyter subprocess pool, rclone FUSE mount of `workspace` bucket (s3 mode) |
| `schedule` | `Dockerfile` | internal only | Cron tick + DAG execution (polls MySQL Outbox) |
The pre-2026 host-port mappings for backend/runtime (`8891:8000` / `8892:8000`)
were removed; no service is reachable from the host except Nginx anymore.
Backend ↔ schedule now talk via the `X-Internal-Service-Token` header on
`/internal/v1/*`. See `API.md §9`.
### Storage layout
The object store is selected at deploy time by `settings.storage_backend`
(`"s3"` default, `"local"` for dev / single-node / air-gapped). Either way
there are 4 purpose-named buckets, resolved in a single place
(`common/storage/factory.py:actual_bucket_name` + `USAGE_TYPE_TO_PURPOSE`).
| `usage_type` | Bucket | Env var | Default name (s3) |
|---|---|---|---|
| `working_copy`, `public_script`, `data_resource`, `snapshot` | `workspace` | `S3_WORKSPACE_BUCKET` | `workspace` |
| `version_artifact` | `version` | `S3_VERSION_BUCKET` | `version` |
| `run_log`, `run_result` | `run_log` | `S3_RUN_LOG_BUCKET` | `run-log` |
| (soft-delete target) | `trash` | `S3_TRASH_BUCKET` | `trash` |
`STORAGE_BACKEND=s3` → 4 separate S3 buckets.
`STORAGE_BACKEND=local` → 4 subdirectories under `LOCAL_STORAGE_BASE_DIR`
(default `/data`):
```
/data/
├── workspace/ # S3_WORKSPACE_BUCKET
├── version/ # S3_VERSION_BUCKET
├── run_log/ # S3_RUN_LOG_BUCKET
└── trash/ # S3_TRASH_BUCKET
```
A workspace's `Workspaces.artifact_bucket` column (when non-null) overrides
the default for that workspace, regardless of `usage_type` — useful for
isolating paid customers onto a dedicated bucket.
Object keys are a flat two-level path — `workspace_id` plus a server-issued
ULID — preserving the original file extension so Jupyter can pick its editor
from the suffix:
```
<bucket>/<workspace_id>/<ulid>{.<ext>}
```
File name, extension, MIME, and logical path all live on `StorageObjects` /
`Scripts` rows; reorganizing the bucket does not require rewriting the
database. Backend code never writes to the container local filesystem except
in `STORAGE_BACKEND=local` mode (where the shared `local-storage` volume IS
the canonical store). Schedule Executor stages node artifacts in
`tempfile.TemporaryDirectory()` (auto-cleanup). Only the `runtime` container
keeps a host volume — s3 mode needs it for rclone FUSE; local mode is a no-op
passthrough.
## Code layout ## Code layout
@@ -14,7 +142,7 @@ development" for the gotcha).
common/src/common/ Pure-Python shared library common/src/common/ Pure-Python shared library
config.py Settings (pydantic-settings, lru_cache singleton) config.py Settings (pydantic-settings, lru_cache singleton)
db/ SQLAlchemy 2.0 async engine, session_scope, Base db/ SQLAlchemy 2.0 async engine, session_scope, Base
db/models/ 26 tables in 9 domain files (zero FK, zero relationship) db/models/ 18 tables in 7 domain files (zero FK, zero relationship)
auth/ JWT / bcrypt / workspace membership helpers auth/ JWT / bcrypt / workspace membership helpers
scheduler/ APScheduler trigger helpers (delayed import) scheduler/ APScheduler trigger helpers (delayed import)
storage/ AsyncStorageBackend abstraction + Pydantic schemas storage/ AsyncStorageBackend abstraction + Pydantic schemas
@@ -70,8 +198,8 @@ schedule/src/schedule/ Schedule Executor (DAG worker)
service.py SchedulerService (composes the three) service.py SchedulerService (composes the three)
orchestrator.py DispatchOrchestrator (Outbox poll + DAG advance) orchestrator.py DispatchOrchestrator (Outbox poll + DAG advance)
execution/ DAG node execution execution/ DAG node execution
executor.py NodeExecutor (notebook / python dispatch) executor.py Dispatch helpers (script-type routing)
worker.py asyncio entry, schedule-spawned task boundary worker.py NodeExecutor + asyncio entry, schedule-spawned task boundary
runners/notebook.py nbclient subprocess path (6-line `notebook_runner` shim re-exports `main`) runners/notebook.py nbclient subprocess path (6-line `notebook_runner` shim re-exports `main`)
infrastructure/ External-system adapters infrastructure/ External-system adapters
storage/client.py SchedulerStorageClient — talks to backend /internal/v1/objects storage/client.py SchedulerStorageClient — talks to backend /internal/v1/objects
@@ -85,7 +213,7 @@ frontend/ React Router SPA (vite build → nginx)
app/ features/ routes/ services/ components/ app/ features/ routes/ services/ components/
migrations/ Alembic schema versions migrations/ Alembic schema versions
docker-compose.yml 4 services (gateway / backend / schedule / runtime) docker-compose.yml 5 services (migrate / web / backend / runtime / schedule)
default.conf Nginx template default.conf Nginx template
scripts/nginx-entrypoint.sh scripts/nginx-entrypoint.sh
.env.example All 26 config.py keys documented .env.example All 26 config.py keys documented
@@ -103,7 +231,7 @@ settings.database_url # str — SQLAlchemy async URL (mysql+asyncmy, ch
settings.jwt_secret # HS256 secret for the auth_request handler settings.jwt_secret # HS256 secret for the auth_request handler
settings.cookie_force_secure # bool — write Secure flag even on plain HTTP (TLS-terminating proxy) settings.cookie_force_secure # bool — write Secure flag even on plain HTTP (TLS-terminating proxy)
settings.service_name # surfaced in /health settings.service_name # surfaced in /health
settings.schedule_event_namespace # APScheduler JobStore namespace + Outbox scope prefix settings.schedule_event_namespace # Outbox event-type namespace prefix (NOT APScheduler JobStore)
settings.readiness_targets # CSV host:port list for /health/ready settings.readiness_targets # CSV host:port list for /health/ready
# HTTP clients (intra-cluster URLs) # HTTP clients (intra-cluster URLs)
@@ -135,6 +263,26 @@ settings.s3_trash_retention_days # int (s3 mode only)
settings.schedule_execution_concurrency # int — max concurrent notebook subprocesses settings.schedule_execution_concurrency # int — max concurrent notebook subprocesses
``` ```
### Encrypted env values
`Settings` runs a `model_validator` (`_decrypt_encrypted_fields`,
`common/src/common/config.py:168-180`) that scans every string field for the
prefix `ENC(...)` and decrypts the inner value with `APP_CONFIG_SECRET_KEY`
using Fernet. Use this for secrets that should not be stored in plain `.env`
files (e.g. third-party API tokens shipped via deployment config).
```bash
# .env
APP_CONFIG_SECRET_KEY=<base64 Fernet key — generate with `python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"`>
SOME_TOKEN=ENC(gAAAAABm...) # ciphertext produced by Fernet.encrypt(b"plaintext")
```
At process start, `SOME_TOKEN` resolves to the decrypted plaintext. If the
field is *not* encrypted (no `ENC(...)` prefix), it passes through unchanged,
so plain `.env` files keep working. The pre-2026 in-repo `encrypt_secret.py`
script was the CLI wrapper around the same Fernet key; if you have old
ciphertexts they round-trip with the new `APP_CONFIG_SECRET_KEY` value.
`Settings` reads from process env first, then from a `.env` file at CWD `Settings` reads from process env first, then from a `.env` file at CWD
if present. `pydantic-settings` auto-loads. `case_sensitive=False` so if present. `pydantic-settings` auto-loads. `case_sensitive=False` so
`DATABASE_URL` / `database_url` both work. The full list of 26 fields `DATABASE_URL` / `database_url` both work. The full list of 26 fields
@@ -167,8 +315,13 @@ grep -rnE 'os\.(environ\[?["\x27][A-Z_]+|getenv\(["\x27][A-Z_]+)' --include="*.p
`deleted_at DATETIME(3) NULL`; queries must filter `is_deleted == 0` `deleted_at DATETIME(3) NULL`; queries must filter `is_deleted == 0`
(or `deleted_at.is_(None)`) to avoid logical-deleted rows. (or `deleted_at.is_(None)`) to avoid logical-deleted rows.
- Domain files under `common/src/common/db/models/` are split by - Domain files under `common/src/common/db/models/` are split by
bounded context: `audit` / `events` / `experiments` / `identity` / bounded context: `events` (2 tables: `ConsumerInbox`,
`runtime` / `schedules` / `scripts` / `storage` / `workspaces`. `OutboxEvents`) / `identity` (4: `Users`, `Roles`, `RolePermissions`,
`Permissions`) / `schedules` (5: `Schedules`, `ScheduleRuns`,
`ScheduleNodes`, `ScheduleEdges`, `ScheduleNodeRuns`) / `scripts`
(2: `Scripts`, `Versions`) / `storage` (3: `StorageObjects`,
`UploadSessions`, `DataResources`) / `workspaces` (2: `Workspaces`,
`WorkspaceMembers`) — 18 tables across 6 model files.
- All models are `class X(Base)` SQLAlchemy 2.0 declarative-mapped. - All models are `class X(Base)` SQLAlchemy 2.0 declarative-mapped.
- The full schema is in `migrations/versions/`. Apply with: - The full schema is in `migrations/versions/`. Apply with:
```bash ```bash
@@ -187,16 +340,18 @@ grep -rnE 'os\.(environ\[?["\x27][A-Z_]+|getenv\(["\x27][A-Z_]+)' --include="*.p
right `create_storage` kwargs for each of the 4 purpose buckets right `create_storage` kwargs for each of the 4 purpose buckets
(`workspace`, `version`, `run_log`, `trash`). Use it in lifespan code; (`workspace`, `version`, `run_log`, `trash`). Use it in lifespan code;
route handlers don't see the difference. route handlers don't see the difference.
- Bucket resolution from `usage_type` is in **one place** - Bucket resolution from `usage_type` lives in **one place**
(`backend/storage_api.py:resolve_bucket`); route handlers only know (`common/src/common/storage/factory.py:actual_bucket_name` +
about `app.state.object_stores[bucket_name]`. `USAGE_TYPE_TO_PURPOSE`). The route handlers in
`backend/src/backend/api/storage.py` only know about
`app.state.object_stores[bucket_name]` and never call
`settings.s3_*_bucket` directly.
- The runtime's view of the workspace bucket on disk is exposed by - The runtime's view of the workspace bucket on disk is exposed by
`common.storage.workspaces_root()`: `common.storage.workspaces_root()`:
- `s3` mode: `${settings.local_storage_base_dir}/workspace` - Both modes resolve to `${settings.local_storage_base_dir}/workspace`
(default `/data/workspace`, the rclone FUSE mount target). (default `/data/workspace`); s3 mode uses it as the rclone FUSE
- `local` mode: `${settings.local_storage_base_dir}/workspace` mount target, local mode uses it as a subdir of the shared
(default `/data/workspace`, a subdir of the shared local-storage `local-storage` volume.
volume).
`settings.local_storage_base_dir` is the **only** path setting; the `settings.local_storage_base_dir` is the **only** path setting; the
helper handles the per-mode suffix. Don't read `settings.workspaces_root` helper handles the per-mode suffix. Don't read `settings.workspaces_root`
or any other path setting directly in runtime code — use this helper. or any other path setting directly in runtime code — use this helper.
@@ -269,21 +424,183 @@ grep -rnE 'os\.(environ\[?["\x27][A-Z_]+|getenv\(["\x27][A-Z_]+)' --include="*.p
`backend/src/backend/api/resources.py` 中按相同模式分别构造 owner-scoped `backend/src/backend/api/resources.py` 中按相同模式分别构造 owner-scoped
LIKE 前缀。 LIKE 前缀。
### Outbox events ### Eventing — Outbox + Inbox + business tables
- The platform's only async-messaging fabric is the MySQL The platform has **no** message broker. The only async fabric is three MySQL
`OutboxEvents` table. Producers (Backend) write rows in the same tables, each on a different side of the same boundary:
transaction as the business state. Consumers (Schedule Executor)
poll every 250 ms and update `event_status` to `published` or | Table | Side | Job |
`failed` (with retry). |---|---|---|
- Use `common.eventing.add_outbox_event` to write. | business tables (`schedule_runs`, `schedule_node_runs`, etc.) | producer's local commit | The state change that *triggered* the event |
- `event_type` values currently in use: | `outbox_events` | producer side | Durable handoff: "this business state change produced this event" |
- `schedule.run.requested` — produced by `schedule_runs.py` / cron post-back | `consumer_inbox` | consumer side | Idempotency: "this consumer has already processed this event" |
- `job.node.execute` — produced by orchestrator when a node is ready
- `job.node.finished` — produced by worker after node execution Together they implement the **Transactional Outbox** + **Consumer Inbox**
- `consumer_inbox` provides exactly-once delivery per patterns: atomic publish on the producer side, idempotent side effects on
`(consumer_name, event_id)` (with `process_status: processing → the consumer side. There is no Redis Stream, no Kafka, no in-process queue.
succeeded` lifecycle). If MySQL is down, both sides are down — and that's intentional.
#### Why two tables and not one
A naïve "publish event by inserting a row, then have the consumer mark it
done" design breaks under two failure modes:
1. **Producer crash after business commit, before publish.** The state
change is real but the event was never sent. Consumers never see it.
2. **Consumer crashes mid-processing.** The event was received but the
side effect may have happened or not. Re-delivery causes double
execution.
`outbox_events` fixes (1): writing to the business table and the outbox
happens in **one transaction**, so either both are durable or neither is.
A separate dispatcher (`schedule/application/orchestrator.py`) polls
`outbox_events` and pushes to consumers. The dispatcher crashing is
harmless — the next poll picks up where it left off.
`consumer_inbox` fixes (2): before applying a side effect, the consumer
inserts a row keyed by `(consumer_name, event_id)`. Re-delivery of the same
`event_id` hits the existing row and short-circuits.
#### Producer side — `outbox_events`
Schema lives in `common/db/models/events.py` (model `OutboxEvents`) and the
baseline migration `migrations/versions/e1f2a3b4c5d6_rebuild_baseline.py`.
Key fields:
- `event_id` (ULID, PK) — globally unique; `consumer_inbox` references this
as the dedup key.
- `aggregate_type` + `aggregate_id` — the producer's domain object
(e.g. `("schedule_run", "<run_id>")`).
- `event_type` — namespaced via `schedule_event_type(...)` so multiple
deployments sharing one MySQL don't cross-consume (see below).
- `schema_version` (default 1) — consumer can branch on this when the
payload shape changes.
- `payload_json` — the event body; opaque to the outbox.
- `event_status` (`pending` → `published` | `failed`) + `available_at` +
`retry_count` + `last_error` — dispatcher state machine. Indexed on
`(event_status, available_at, created_at)` because that's the poll
hot path.
- `idempotency_key` + index — upstream dedup at the producer (e.g. two
requests from the user that should produce one event, not two).
- `trace_id` — links to the request that produced the event for log
correlation.
Write path — the **only** entry point is `common.eventing.add_outbox_event`:
```python
await add_outbox_event(
session,
event_type=schedule_event_type("schedule.run.requested"),
producer="backend.api.schedules.runs",
trace_id=request.state.trace_id,
aggregate_type="schedule_run",
aggregate_id=run_id,
idempotency_key=f"schedule.run.requested:{run_id}",
payload={"run_id": run_id, "schedule_id": schedule_id},
)
```
`session` is the **same** SQLAlchemy session as the business-table write;
the outbox row goes in via `session.add(...)` and commits together with
the business row. Never call `session.commit()` between the business
write and the outbox write — that defeats the whole pattern.
**Event type namespacing.** Every event type MUST go through
`common.eventing.schedule_event_type(raw)` before being passed to
`add_outbox_event`. The helper prefixes `settings.schedule_event_namespace`
(default `model-platform-develop`) so that two deployments sharing one
MySQL — common during development — don't accidentally consume each
other's events. The constants `SCHEDULE_RUN_REQUESTED_EVENT`,
`NODE_EXECUTE_EVENT`, `NODE_FINISHED_EVENT` in
`schedule/application/orchestrator.py` and `schedule/execution/worker.py`
are already namespaced; never hard-code the raw string.
Current event types in use:
| Event | Produced by | Consumed by |
|---|---|---|
| `schedule.run.requested` | `backend/api/schedules/runs.py`, cron post-back | `schedule/application/orchestrator.py` (`schedule-orchestrator`) |
| `job.node.execute` | `schedule/application/orchestrator.py` when a node is ready | `schedule/execution/worker.py` (`schedule-results` writes back via the orchestrator) |
| `job.node.finished` | `schedule/execution/worker.py` after a node run finishes | `schedule/application/orchestrator.py` (`schedule-results`) |
The dispatcher polls every `0.25s` when there's pending work, dropping to
`1s` when idle — see the loop in
`schedule/application/orchestrator.py` (the `asyncio.sleep(0.25)` /
`asyncio.sleep(1)` branches).
#### Consumer side — `consumer_inbox`
Model `ConsumerInbox` in `common/db/models/events.py`. Composite PK
`(consumer_name, event_id)` so multiple consumers can independently
process the same `event_id`.
Lifecycle (lives in `schedule/application/orchestrator.py:_start_inbox` /
`_finish_inbox`):
```
┌──────────────┐
│ processing │ ← INSERT or re-claim on re-delivery
└──────┬───────┘
│
success │ failure
▼
┌─────────────┐
│ succeeded │ (terminal — no re-process)
└─────────────┘
on failure, error_message is set and the row is left in
`failed` for inspection; the dispatcher does NOT auto-retry
consumer failures (only producer-side publish failures).
```
The consumer **must** call `_start_inbox` (or equivalent) at the top of
every event handler. The function returns `(inbox_row, should_process)`;
if `should_process` is `False`, the event was already handled and the
handler returns immediately. On success the handler calls `_finish_inbox`
to flip `process_status` to `succeeded`.
Currently registered `consumer_name` values (see `orchestrator.py:594`,
`:924`):
- `schedule-orchestrator` — consumes `schedule.run.requested`
- `schedule-results` — consumes `job.node.finished`
A new consumer = a new `consumer_name` string. Two consumers sharing a
name will collide on the PK — pick a stable, descriptive name and treat
it as a contract.
#### Failure modes the design covers
| Scenario | What happens |
|---|---|
| Backend crashes after business commit, before dispatcher polls | Outbox row exists; next poll picks it up. |
| Dispatcher crashes after poll, before HTTP push to executor | `event_status` still `pending`; next poll retries. |
| Executor crashes mid-handler | Inbox row stays `processing`; on redelivery the handler re-enters `_start_inbox`, sees `succeeded`? — no, sees `processing` and re-runs. **This is currently a known soft spot** — the executor's `_finish_inbox` must run, and a crash before that means a re-run. Don't perform side effects before `_finish_inbox` succeeds. |
| Same `event_id` delivered twice (e.g. HTTP retry after success) | Inbox short-circuits; the second delivery is a no-op. |
| Multiple development deployments share one MySQL | `schedule_event_namespace` prefixes keep them isolated; each deployment only sees its own events. |
#### Adding a new event
1. Pick an `aggregate_type` / `aggregate_id` pair that identifies the
producing domain object.
2. Pick a `consumer_name` for each consumer. Stable, descriptive,
never reused for a different purpose.
3. Define the event type constant:
```python
# in the producing module
MY_NEW_EVENT = schedule_event_type("schedule.my_new_event")
```
4. Write it via `add_outbox_event` in the same transaction as the
business mutation.
5. In the consumer, start with `_start_inbox` (or follow the pattern
in `orchestrator.py`) before doing any side effects, and call
`_finish_inbox` on success.
6. Update the table above.
Do **not** invent a new messaging fabric (Redis Stream, Kafka, in-process
queue). The whole point of this design is that MySQL is the single
authority — adding a second one doubles the failure surface.
### Async / sync signatures ### Async / sync signatures
@@ -373,14 +690,17 @@ print('settings ok:', settings.s3_endpoint)
### Add a new DAG endpoint ### Add a new DAG endpoint
1. Add the route handler in `backend/schedules.py` (DAG template) or 1. Add the route handler in `backend/src/backend/api/schedules/schedules.py`
`backend/schedule_runs.py` (run lifecycle). (DAG template) or `backend/src/backend/api/schedules/runs.py` (run
2. Validate request via `backend/schedule_schemas.py`. lifecycle).
2. Validate request via Pydantic schemas in
`backend/src/backend/schemas/schedules.py`.
3. For mutations on nodes/edges/versions: route through 3. For mutations on nodes/edges/versions: route through
`create_script_record` / `get_script_row` and apply `create_script_record` / `get_script_row` and apply
`require_script_modify_access` if it touches a script. `require_script_modify_access` if it touches a script.
4. If it produces an outbox event, use 4. If it produces an outbox event, use
`add_outbox_event(session, event_type="...", producer="...", ...)`. `add_outbox_event(session, event_type=schedule_event_type("..."), producer="...", ...)`
in the same transaction as the business write.
### Add a new env var ### Add a new env var
@@ -405,22 +725,17 @@ See "Adding a new env var" above.
### Wire a new storage bucket ### Wire a new storage bucket
The current 4 buckets are wired in `backend/storage_api.py:resolve_bucket`: The current 4 buckets are wired in `common/src/common/storage/factory.py`.
`USAGE_TYPE_TO_PURPOSE` maps each `usage_type` (`working_copy`,
```python `public_script`, `data_resource`, `snapshot`, `version_artifact`,
BUCKET_FOR_USAGE: dict[str, str] = { `run_log`, `run_result`) to one of the 4 purpose buckets (`workspace`,
"working_copy": settings.s3_workspace_bucket, `version`, `run_log`, `trash`); `actual_bucket_name(purpose)` then
"public_script": settings.s3_workspace_bucket, returns the s3-style identifier from the corresponding
"data_resource": settings.s3_workspace_bucket, `settings.s3_<purpose>_bucket`. `BUCKET_FOR_USAGE` is a dict
"snapshot": settings.s3_workspace_bucket, comprehension built from these two.
"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")` The constant `PURPOSE_BUCKETS = ("workspace", "version", "run_log", "trash")`
in `common.storage.factory` enumerates the four backends built in the in `common/storage/factory.py` enumerates the four backends built in the
backend lifespan. To add a fifth bucket: backend lifespan. To add a fifth bucket:
1. Add the env var to `Settings` (s3 mode only): 1. Add the env var to `Settings` (s3 mode only):
@@ -435,36 +750,62 @@ backend lifespan. To add a fifth bucket:
4. Extend the `Literal` in `common/storage/schemas.py` (in 4. Extend the `Literal` in `common/storage/schemas.py` (in
`CreateUploadRequest.usage_type`, `ServerObjectRequest.usage_type`) `CreateUploadRequest.usage_type`, `ServerObjectRequest.usage_type`)
to include the new value. to include the new value.
5. Add an entry in `BUCKET_FOR_USAGE` mapping the new `usage_type` to 5. Add an entry in `USAGE_TYPE_TO_PURPOSE` mapping the new `usage_type`
the new bucket env var. to the new purpose; `BUCKET_FOR_USAGE` is regenerated automatically.
6. Pre-create the bucket (s3 mode) or subdirectory (local mode) in the 6. Pre-create the bucket (s3 mode) or subdirectory (local mode) in the
deployment. The backend no longer auto-creates buckets. deployment. The backend no longer auto-creates buckets.
A workspace's `artifact_bucket` column (when non-null) overrides the A workspace's `Workspaces.artifact_bucket` column (when non-null)
default for that workspace, regardless of `usage_type`. overrides the default for that workspace, regardless of `usage_type`.
### Add a new schedule node type ### Add a new schedule node type
`schedule/execution.py` dispatches on `script_type` in `schedule/src/schedule/execution/runners/notebook.py:execute_artifact`
`execute_artifact`. Add a new branch + a new `_<type>` function. dispatches on `script_type`. Add a new branch + a new `_<type>` function
`worker.py` does not need to change — the dispatch happens inside in that file. The NodeExecutor in `worker.py` does not need to change —
`execute_artifact`. the dispatch happens inside `execute_artifact`.
## Tests ## Tests
There is **no formal test suite yet** (see HANDOVER §Pending Tasks There is an in-tree test suite, mostly covering scripts / resources /
P1). A reasonable first test surface: DAG validation / upload state transitions. It is **not** the formal
release-gate suite the project still owes (see `HANDOVER.md` §8 for
known gaps: trash reaper, cross-backend migration, ENC round-trip).
- `require_script_modify_access` (admin / owner / non-owner-unlock / Current coverage:
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` | Area | Tests | Where |
handlers. Use SQLite in-memory (or a MySQL test container) for DB |---|---|---|
integration. Use moto for S3. | Scripts (CRUD, soft delete, parent path, same-name siblings, visibility) | 10 functions across `test_scripts.py`, `test_count_scripts.py`, `test_list_scripts_parent_path.py`, `test_storage_upload_status.py` | `backend/tests/` |
| Resources (visibility, ownership, idempotency) | ~5 in `test_resources.py` | `backend/tests/` |
| DAG validation (cycle / orphan detection) | `test_validate_dag.py` | `backend/tests/` |
| Audit log middleware | `test_audit_logging.py` | `backend/tests/` |
| Jupyter auth cache | `test_jupyter_auth_cache.py` | `backend/tests/` |
| Runtime client (directory listing, error mapping) | `test_runtime_client_directories.py` | `backend/tests/` |
| Schedule layer (worker, janitor, layering invariants) | 11 functions in `test_janitor.py`, `test_layering.py`, `test_worker.py` | `schedule/tests/` |
Run:
```bash
# Backend
uv run --package backend pytest backend/tests -q
# Schedule
uv run --package schedule pytest schedule/tests -q
```
Conventions:
- pytest + `pytest-asyncio` for `async def` handlers.
- MySQL is required for the ORM tests (not SQLite — CHAR(26) ULIDs and
`mysql.TINYINT(1)` quirks do not translate). Local docker-compose
MySQL is the typical target.
- For storage, `common.storage.factory` selects between s3 and local
via `settings.storage_backend`; tests that exercise both modes
monkeypatch that setting (see `common/tests/storage/test_factory.py`).
- For HTTP boundaries (httpx to backend / runtime), tests use `respx`
with `assert_all_called=False` so unused stubs don't fail the test
— see the engineering notes in `CLAUDE.md`.
## Troubleshooting ## Troubleshooting
@@ -501,7 +842,10 @@ check the JWT (use `JWT_SECRET` from `.env`).
`worker` must be constructed before `orchestrator` in `worker` must be constructed before `orchestrator` in
`SchedulerService.__init__`, because orchestrator's dispatch table `SchedulerService.__init__`, because orchestrator's dispatch table
captures `self.worker.handle_node_execute` at construction time. captures `self.worker.handle_node_execute` at construction time.
See `service.py` — the order is load-bearing. See `schedule/src/schedule/application/service.py` — the order is
load-bearing. (This was the symptom during the flat → layered schedule
refactor; if you see it today, the most likely cause is a partial
rebase that left an old import path.)
## Style ## Style
@@ -517,8 +861,12 @@ See `service.py` — the order is load-bearing.
## See also ## See also
- `ARCHITECTURE.md` — design diagrams - `ARCHITECTURE.md` — kept as a thin redirect; the authoritative
- `HANDOVER.md` — current refactor state and pending work architecture diagrams and capability map now live in
- `CLAUDE.md` — agent-facing conventions for the repo [§Architecture](#architecture) above.
- `models / __init__.py` — exhaustive list of all 26 tables - `HANDOVER.md` — current refactor state, recent commits, pending work.
- `common/config.py` — all env vars in one place - `CLAUDE.md` — agent-facing conventions for the repo.
- `common/src/common/db/models/__init__.py` — exhaustive list of all 18 tables.
- `common/src/common/config.py` — all env vars in one place.
- Per-package READMEs: `backend/README.md`, `common/README.md`,
`runtime/README.md`, `frontend/README.md`.
README.md
+12 -145
View File
@@ -9,97 +9,16 @@
## 它做什么 ## 它做什么
| 能力 | 位置 | 能力清单、组件图、容器表、存储布局、配置参考——**全部在 [`DEVELOP.md`](./DEVELOP.md)**：
|---|---|
| workspace 内 notebook 编辑，行级锁 | `backend/jupyter.py` + `scripts.is_locked` |
| Jupyter 鉴权路由（浏览器永远拿不到 runtime token） | `nginx/default.conf` + `auth_request` + `backend/jupyter.py` |
| notebook / script / version / run_log 的对象存储（s3 / local 二选一） | `common/storage/` + `backend/scripts.py` |
| DAG 调度：节点、边、cron、手动触发、重试、快照 | `backend/schedules.py` + `backend/schedule_runs.py` + `schedule/`（5 个模块） |
| DAG 执行走 MySQL Outbox（无 Redis，无进程内队列） | `schedule/orchestrator.py` + `schedule/worker.py` |
| 每个 workspace 一个 Jupyter 子进程池，配 asyncio 锁 | `runtime/process.py` |
| runtime 内 rclone FUSE 把 workspace 桶挂上来（s3 模式） | `runtime/mount.py` |
| 仅 MySQL 持久化（26 张表，软删除，无外键） | `common/db/models/` |
## 架构一览 - 系统整体架构（[§Architecture](./DEVELOP.md#architecture)）
- 18 张 MySQL 表的 Eventing 协作模式（[§Eventing](./DEVELOP.md#eventing--outbox--inbox--business-tables)）
- 代码目录布局（[§Code layout](./DEVELOP.md#code-layout)）
- 全部 26 个环境变量（[§Configuration system](./DEVELOP.md#configuration-system)）
- 写代码的约定（[§Conventions](./DEVELOP.md#conventions)）
- 加新表 / 新桶 / 新节点类型的步骤（[§Common tasks](./DEVELOP.md#common-tasks)）
``` `ARCHITECTURE.md` 现已并入 `DEVELOP.md`；`HANDOVER.md` 记录近期重构与待办事项。
┌────────────────────┐
│ Browser (SPA) │
└─────────┬──────────┘
│ HTTPS / WS
┌─────────▼──────────┐
│ Nginx (only :80) │ ← templates/default.conf
│ /api/ /jupyter/ /storage/
└────┬───────┬──────┘
│ │
┌──────────────┘ └─────────────┐
▼ ▼
┌──────────────────┐ ┌──────────────────────┐
│ FastAPI Backend │ │ Runtime (Jupyter) │
│ + /internal/v1 │ control │ - rclone FUSE mount │ (P0-1)
│ /objects │ token-Auth │ │
│ (storage) ├──────────────►│ - subprocess pool │
│ - DAG CRUD │ │ (per workspace) │
│ - script CRUD │ └──────────┬───────────┘
│ - auth_request │ │ FUSE / shared vol
│ - /api/v1/... │ ▼
└────┬──────┬──────┘ ┌──────────────────────┐
│ │ │ Object storage │
│ └──────── HTTP ───────►│ (s3: S3 service / │
▼ │ local: shared vol) │
┌────────────┐ │ 4 buckets per usage │
│ MySQL │◄───────── poll ─────│ │
│ - 26 tbls │ └──────────────────────┘
│ - outbox │
│ - jobstore │
└────┬───────┘
▲
│ outbox poll
┌────┴──────────────────────────┐
│ Schedule Executor │
│ - CronScheduler (APScheduler) │
│ - DispatchOrchestrator │
│ - NodeExecutor (worker) │
│ - SchedulerService (facade) │
└───────────────────────────────┘
```
对象存储通过 `STORAGE_BACKEND`（s3 | local）二选一。s3 模式下 4 个 purpose 命名桶
（`workspace` / `version` / `run-log` / `trash`）是独立的 S3 bucket；local 模式下
是 `LOCAL_STORAGE_BASE_DIR` 的子目录，通过 Docker volume `local-storage` 共享。
详见 `DEVELOP.md` §存储。
详细设计见 `ARCHITECTURE.md`。实现的偏离和近期重构记录在 `HANDOVER.md`。
## 目录结构
```text
frontend/ React Router SPA
backend/ FastAPI：公开 API + 内部存储 API
runtime/ Jupyter 子进程管理 + rclone FUSE
schedule/ DAG 调度器（5 模块：context/scheduler/
orchestrator/worker/service）
common/ 配置、SQLAlchemy 模型、存储 SDK、
outbox 事件、jobstore
migrations/ Alembic 基线 + 各特性 migration
nginx/ （仅概念 — 见下方「容器」一节）
scripts/ nginx-entrypoint.sh（模板渲染）
docker-compose.yml 4 服务 — web / backend / runtime / schedule
default.conf Nginx 模板（挂载，启动时渲染）
.env.example common.config.Settings 消费的所有环境变量
```
## 容器
| 服务 | 镜像 | 暴露 | 用途 |
|---|---|---|---|
| `web` | `nginx:alpine` | 宿主机 `:8888` → `:80` | SPA、`/api/` 反向代理、`/jupyter/{ws}/` auth_request 代理、`/storage/` S3 直通（仅 s3 模式） |
| `backend` | `Dockerfile` | 仅内网 | DAG CRUD、script CRUD、schedule 触发、`/api/v1/auth/jupyter`、`/internal/v1/objects` 服务间 RPC(共享 `INTERNAL_SERVICE_TOKEN` 鉴权,P0-1)|
| `runtime` | `Dockerfile` | 仅内网 | 每个 workspace 一个 Jupyter 子进程池、rclone FUSE 挂载 `workspace` 桶（s3 模式） |
| `schedule` | `Dockerfile` | 仅内网 | cron tick + DAG 执行（轮询 MySQL Outbox） |
架构**故意只暴露一个宿主机端口**(网关);其他服务都在 Docker 内网。
这一点在 `docker-compose.yml` 里强制执行 — backend / runtime / schedule 都没有 `ports:`。在 P0-1 之前,backend 与 runtime 曾短暂地把 `8891` / `8892` 映射到宿主机;此映射已被删除,改用 `INTERNAL_SERVICE_TOKEN` 头对 `/internal/v1/*` 做服务间鉴权,见 `API.md §9`。
## 快速启动 ## 快速启动
@@ -144,64 +63,12 @@ docker compose down
docker compose down -v docker compose down -v
``` ```
## 配置
所有环境变量在 `common/src/common/config.py` 里用 pydantic-settings 的 `Settings`
类一次性声明，外面套一层 `@lru_cache` 单例。新增环境变量：
1. 在 `common/src/common/config.py` 的 `Settings` 里加字段（带合理 default，使 dev 启动不需要设）
2. 在 `.env.example` 加一行带注释
3. 调用点用 `settings.<name>`，永远不要用 `os.environ["..."]`
完整环境变量列表和含义见 `DEVELOP.md`。
## 存储布局
4 个 purpose 命名桶。从 `StorageObjects.usage_type` 到桶的映射由
**单一入口**（`backend/storage_api.py:resolve_bucket`）决定：
| `usage_type` | 桶（环境变量） | 默认名 |
|---|---|---|
| `working_copy`、`public_script`、`data_resource`、`snapshot` | `S3_WORKSPACE_BUCKET` | `workspace` |
| `version_artifact` | `S3_VERSION_BUCKET` | `version` |
| `run_log`、`run_result` | `S3_RUN_LOG_BUCKET` | `run-log` |
| （软删除目标） | `S3_TRASH_BUCKET` | `trash` |
`STORAGE_BACKEND=s3` 模式下是 4 个独立 S3 桶。`STORAGE_BACKEND=local` 模式下
是 `LOCAL_STORAGE_BASE_DIR`（默认 `/data`）下的 4 个子目录：
```
/data/
├── workspace/ # S3_WORKSPACE_BUCKET
├── version/ # S3_VERSION_BUCKET
├── run_log/ # S3_RUN_LOG_BUCKET
└── trash/ # S3_TRASH_BUCKET
```
某个 workspace 的 `artifact_bucket` 列（非 NULL 时）覆盖该 workspace 的默认桶，
无视 `usage_type` — 适合把付费客户隔离到专属桶。
对象 key 是两层扁平路径 — `workspace_id` 加服务端签发的 `ulid`：
```
<workspace_bucket>/<workspace_id>/<ulid>{.<ext>}
```
文件名、扩展名、MIME、逻辑路径都放在 `StorageObjects` 和 `Scripts` 行里，不进
object key — 重新组织存储不需要重写数据库。
Backend 代码从不写容器本地文件系统（`STORAGE_BACKEND=local` 模式除外，那里共享
`local-storage` volume 就是规范存储）。Schedule Executor 在 `tempfile.TemporaryDirectory()`
里暂存节点工件（自动清理）。只有 `runtime` 容器保留宿主 volume — s3 模式下 rclone FUSE
挂载需要；local 模式下是 no-op 透传。
## 文档 ## 文档
- `README.md`（本文）— 快速导读 - [`DEVELOP.md`](./DEVELOP.md) — 权威文档：架构、代码布局、配置、约定、常用任务、测试、故障排查
- `ARCHITECTURE.md` — 设计图 + 简化历史 - [`HANDOVER.md`](./HANDOVER.md) — 近期 commit / 实现偏离 / 待办事项
- `HANDOVER.md` — 实现偏离、近期重构、待办事项 - [`API.md`](./API.md) — REST API 契约
- `DEVELOP.md` — 开发指南（环境变量、代码规约、常用操作） - [`CLAUDE.md`](./CLAUDE.md) — agent 面向的本仓库规约
- `CLAUDE.md` — agent 面向的本仓库规约
## 许可 ## 许可