382 lines
9.9 KiB
Markdown
382 lines
9.9 KiB
Markdown
# Job 核心系統規劃
|
||
|
||
## 目標
|
||
|
||
建立一套通用 job system，讓任何長任務、流程任務、定時任務未來都能共用。Job 不只是背景任務，而是「有模板、有設定、有狀態、有進度、有取消能力、有重跑策略、有排程能力」的工作單元。
|
||
|
||
## 核心設計
|
||
|
||
採用：
|
||
|
||
```text
|
||
Mongo = job/template/run/history 的真相來源
|
||
Redis = queue、distributed lock、schedule tick、短期 lease
|
||
```
|
||
|
||
```mermaid
|
||
flowchart LR
|
||
Api[GoAPI] --> Template[JobTemplate]
|
||
Api --> JobRun[JobRunMongo]
|
||
JobRun --> RedisQueue[RedisQueue]
|
||
Scheduler[SchedulerTick] --> RedisQueue
|
||
Worker[Worker] --> RedisQueue
|
||
Worker --> JobRun
|
||
Worker --> Step[JobStep]
|
||
```
|
||
|
||
## 核心概念
|
||
|
||
### JobTemplate
|
||
|
||
Template 定義「這種 job 要怎麼做」。例如：
|
||
|
||
```text
|
||
demo_long_task
|
||
external_worker_task
|
||
scheduled_report
|
||
multi_step_pipeline
|
||
```
|
||
|
||
Template 要回答：
|
||
|
||
- 這個 job 的輸入 payload schema 是什麼
|
||
- 有哪些 steps
|
||
- 最終狀態是什麼
|
||
- 可不可以重複執行
|
||
- 是否允許同 account / 同 target 同時跑
|
||
- retry policy 是什麼
|
||
- timeout 是多少
|
||
- 是否可被排程
|
||
- 是否支援取消，以及取消時 worker 要如何收斂
|
||
|
||
### JobRun
|
||
|
||
JobRun 是每一次執行實例。它引用 template，保存當次 payload、狀態、進度、結果、錯誤與執行歷史。
|
||
|
||
### JobSchedule
|
||
|
||
JobSchedule 是「何時建立 JobRun」。支援：
|
||
|
||
- cron
|
||
- enabled / disabled
|
||
- timezone
|
||
- payload template
|
||
- target scope，例如 user/account/system
|
||
- nextRunAt / lastRunAt
|
||
|
||
### JobFlow
|
||
|
||
Flow 是多步驟流程。第一版不用做完整 DAG，先支援線性 steps：
|
||
|
||
```text
|
||
multi_step_pipeline:
|
||
1. prepare
|
||
2. execute
|
||
3. finalize
|
||
```
|
||
|
||
之後再擴成 DAG 或 conditional branch。
|
||
|
||
## Mongo Collections
|
||
|
||
### `job_templates`
|
||
|
||
```json
|
||
{
|
||
"_id": "...",
|
||
"type": "demo_long_task",
|
||
"version": 1,
|
||
"name": "示範長任務",
|
||
"description": "展示 job template、進度、取消、重跑與排程能力",
|
||
"enabled": true,
|
||
"repeatable": true,
|
||
"concurrencyPolicy": "reject_same_scope",
|
||
"dedupeKeys": ["scope_id", "target"],
|
||
"timeoutSeconds": 600,
|
||
"cancelPolicy": {
|
||
"supported": true,
|
||
"mode": "cooperative",
|
||
"graceSeconds": 30
|
||
},
|
||
"retryPolicy": {
|
||
"maxAttempts": 2,
|
||
"backoffSeconds": [30, 120]
|
||
},
|
||
"steps": [
|
||
{ "id": "prepare", "name": "準備資料", "workerType": "go", "timeoutSeconds": 60, "cancelable": true },
|
||
{ "id": "execute", "name": "執行任務", "workerType": "go", "timeoutSeconds": 300, "cancelable": true },
|
||
{ "id": "finalize", "name": "整理結果", "workerType": "go", "timeoutSeconds": 30, "cancelable": false }
|
||
],
|
||
"createAt": 0,
|
||
"updateAt": 0
|
||
}
|
||
```
|
||
|
||
### `job_runs`
|
||
|
||
```json
|
||
{
|
||
"_id": "...",
|
||
"templateType": "demo_long_task",
|
||
"templateVersion": 1,
|
||
"scope": "user",
|
||
"scopeId": "user_123",
|
||
"status": "pending",
|
||
"phase": "prepare",
|
||
"payload": {},
|
||
"progress": {
|
||
"summary": "等待 worker 執行",
|
||
"percentage": 20,
|
||
"steps": []
|
||
},
|
||
"result": null,
|
||
"error": null,
|
||
"attempt": 0,
|
||
"maxAttempts": 2,
|
||
"lockedBy": null,
|
||
"lockedUntil": null,
|
||
"cancelRequestedAt": null,
|
||
"cancelReason": null,
|
||
"scheduledAt": null,
|
||
"startedAt": null,
|
||
"completedAt": null,
|
||
"createAt": 0,
|
||
"updateAt": 0
|
||
}
|
||
```
|
||
|
||
### `job_schedules`
|
||
|
||
```json
|
||
{
|
||
"_id": "...",
|
||
"templateType": "demo_long_task",
|
||
"scope": "user",
|
||
"scopeId": "user_123",
|
||
"enabled": true,
|
||
"cron": "0 9 * * *",
|
||
"timezone": "Asia/Taipei",
|
||
"payloadTemplate": {},
|
||
"lastRunAt": null,
|
||
"nextRunAt": 0,
|
||
"createAt": 0,
|
||
"updateAt": 0
|
||
}
|
||
```
|
||
|
||
### `job_events`
|
||
|
||
用來觀察與 audit：
|
||
|
||
```json
|
||
{
|
||
"_id": "...",
|
||
"jobId": "...",
|
||
"type": "status_changed",
|
||
"from": "pending",
|
||
"to": "running",
|
||
"message": "worker claimed job",
|
||
"metadata": {},
|
||
"createAt": 0
|
||
}
|
||
```
|
||
|
||
## Status Model
|
||
|
||
```text
|
||
pending = 已建立，等待 queue
|
||
queued = 已推進 Redis queue
|
||
running = worker 執行中
|
||
waiting_worker = 等外部 worker 回寫
|
||
cancel_requested = 使用者已要求取消，等待 worker cooperative stop
|
||
succeeded = 成功完成
|
||
failed = 最終失敗
|
||
cancelled = 使用者取消
|
||
expired = lock/timeout 過期後無法恢復
|
||
```
|
||
|
||
Step status：
|
||
|
||
```text
|
||
pending | running | succeeded | failed | skipped | cancelled
|
||
```
|
||
|
||
## 取消語意
|
||
|
||
取消是第一版必做能力，採 cooperative cancellation：
|
||
|
||
```mermaid
|
||
flowchart LR
|
||
User[User] --> ApiCancel[CancelAPI]
|
||
ApiCancel --> Run[JobRun cancel_requested]
|
||
Run --> RedisCancel[RedisCancelSignal]
|
||
Worker[Worker] -->|"poll cancel flag"| Run
|
||
Worker --> Stop[StopCurrentStep]
|
||
Stop --> Final[JobRun cancelled]
|
||
```
|
||
|
||
規則：
|
||
|
||
- `pending` / `queued`：取消後直接變 `cancelled`，並盡量從 Redis queue 移除；若無法移除，worker claim 時必須檢查狀態並跳過。
|
||
- `running`：狀態改為 `cancel_requested`，寫入 `cancelRequestedAt` / `cancelReason`，worker 必須在 step 間或長任務 checkpoint 檢查取消旗標。
|
||
- `waiting_worker`：狀態改為 `cancel_requested`，同時寫 Redis cancel signal；外部 worker 回寫前要檢查 job 狀態。
|
||
- `succeeded` / `failed` / `cancelled` / `expired`：不可取消，回傳 ResourceInvalidState。
|
||
- worker 收到取消後呼叫 `AcknowledgeCancel(jobId, workerId)`，釋放 lock，寫入 `job_events`，狀態變 `cancelled`。
|
||
- 若 `cancel_requested` 超過 template 的 `cancelPolicy.graceSeconds`，scheduler/reaper 可標記為 `cancelled` 或 `expired`，第一版建議標記 `cancelled` 並記錄 timeout event。
|
||
|
||
## 狀態與 Lock 安全規則
|
||
|
||
第一版已把最容易出問題的 race condition 收斂在 repository / usecase：
|
||
|
||
- `ClaimNext` 只能從 `pending` / `queued` conditional update 成 `running`。如果 API 同時取消，Mongo update 會被拒絕。
|
||
- `RequestCancel` 只能從 cancellable 狀態 conditional update；`pending` / `queued` 直接變 `cancelled`，`running` / `waiting_worker` 變 `cancel_requested`。
|
||
- `CompleteRun` / `FailRun` / `UpdateProgress` 必須帶 `workerID`，並且只能更新 `lockedBy == workerID` 的 job。
|
||
- Redis `jobs:lock:<jobId>` 的 value 是 `workerID`；`ReleaseLock` / `RefreshLock` 使用 owner check，避免舊 worker 誤刪新 worker 的 lock。
|
||
- Worker 長任務要定期 heartbeat，呼叫 `RefreshRunLock(jobId, workerID, ttlSeconds)`。自訂 step handler 可用 `StepContext.Heartbeat`。
|
||
|
||
之後新增狀態轉移時，不要直接使用裸 `Update`；若是生命週期狀態，應新增明確的 guarded repository 方法或使用現有 conditional update。
|
||
|
||
## Redis Keys
|
||
|
||
```text
|
||
jobs:queue:<workerType> # list 或 stream，worker 消費
|
||
jobs:lock:<jobId> # lease lock
|
||
jobs:scheduler:lock # scheduler singleton lock
|
||
jobs:dedupe:<template>:<hash> # 防止同 scope 重複跑
|
||
jobs:cancel:<jobId> # cancel signal，worker checkpoint 讀取
|
||
```
|
||
|
||
第一版建議用 Redis List 即可，之後需要 ack/replay 再升級 Redis Streams。
|
||
|
||
## API 規劃
|
||
|
||
```text
|
||
GET /api/v1/job/templates
|
||
GET /api/v1/job/templates/:type
|
||
PUT /api/v1/job/templates/:type
|
||
|
||
POST /api/v1/jobs
|
||
GET /api/v1/jobs/:id
|
||
GET /api/v1/jobs?page=1&pageSize=20
|
||
POST /api/v1/jobs/:id/cancel
|
||
POST /api/v1/jobs/:id/retry
|
||
|
||
GET /api/v1/job/schedules?page=1&pageSize=20
|
||
POST /api/v1/job/schedules
|
||
PUT /api/v1/job/schedules/:id
|
||
POST /api/v1/job/schedules/:id/enable
|
||
POST /api/v1/job/schedules/:id/disable
|
||
```
|
||
|
||
所有列表回應使用目前標準：
|
||
|
||
```json
|
||
{
|
||
"code": 102000,
|
||
"message": "SUCCESS",
|
||
"data": {
|
||
"pagination": {
|
||
"total": 42,
|
||
"page": 1,
|
||
"pageSize": 10,
|
||
"totalPages": 5
|
||
},
|
||
"list": []
|
||
}
|
||
}
|
||
```
|
||
|
||
## 分層規劃
|
||
|
||
```text
|
||
internal/model/job/
|
||
domain/entity/template.go
|
||
domain/entity/run.go
|
||
domain/entity/schedule.go
|
||
domain/entity/event.go
|
||
domain/enum/status.go
|
||
domain/repository/*.go
|
||
domain/usecase/*.go
|
||
repository/mongo_*.go
|
||
repository/redis_queue.go
|
||
usecase/template_usecase.go
|
||
usecase/run_usecase.go
|
||
usecase/schedule_usecase.go
|
||
usecase/progress_usecase.go
|
||
usecase/worker_usecase.go
|
||
|
||
internal/logic/job/
|
||
create_job_logic.go
|
||
get_job_logic.go
|
||
list_jobs_logic.go
|
||
cancel_job_logic.go
|
||
retry_job_logic.go
|
||
template_logic.go
|
||
schedule_logic.go
|
||
|
||
internal/worker/job/
|
||
runner.go
|
||
scheduler.go
|
||
dispatcher.go
|
||
```
|
||
|
||
## Template 執行規則
|
||
|
||
### 可不可以重複做
|
||
|
||
由 `repeatable` 與 `concurrencyPolicy` 控制：
|
||
|
||
```text
|
||
repeatable=false
|
||
同 dedupe key 完成過就不再建立
|
||
|
||
repeatable=true + reject_same_scope
|
||
已有 running/pending job 時拒絕建立
|
||
|
||
repeatable=true + allow_parallel
|
||
允許平行跑
|
||
|
||
repeatable=true + replace_existing
|
||
取消舊 job，建立新 job
|
||
```
|
||
|
||
### 最終狀態
|
||
|
||
Template 可定義成功條件：
|
||
|
||
```text
|
||
successWhen = all_steps_succeeded
|
||
```
|
||
|
||
第一版只支援 `all_steps_succeeded`。之後再加 `any_step_succeeded` 或 conditional flow。
|
||
|
||
### 取消能力
|
||
|
||
Template 用 `cancelPolicy` 控制取消：
|
||
|
||
```text
|
||
supported=false
|
||
API 不允許取消此 job
|
||
|
||
mode=cooperative
|
||
worker checkpoint 檢查取消旗標後收斂
|
||
|
||
graceSeconds=30
|
||
cancel_requested 超過 grace 後由 reaper 收斂
|
||
```
|
||
|
||
Step 用 `cancelable` 控制目前步驟是否可立即停止。若目前 step 不可取消，worker 需要在 step 結束後停止後續 steps，最後狀態仍為 `cancelled`。
|
||
|
||
## 第一版建議實作順序
|
||
|
||
1. 建 `model/job` 的 enum/entity/repository interface。
|
||
2. 實作 Mongo repositories：template/run/schedule/event。
|
||
3. 實作 Redis queue/lock repository。
|
||
4. 實作 `CreateRun`：讀 template、檢查 repeat/concurrency、建立 run、push queue。
|
||
5. 實作 `ClaimNext`：worker 從 Redis 取 job，Mongo 設 lock/status。
|
||
6. 實作 `RequestCancel`：狀態轉 `cancel_requested` 或直接 `cancelled`，寫 Redis cancel signal 與 event。
|
||
7. 實作 `AcknowledgeCancel`：worker 收斂後釋放 lock，狀態轉 `cancelled`。
|
||
8. 實作 `UpdateProgress`、`Complete`、`Fail`、`Retry`。
|
||
9. 實作 schedule tick：掃 `job_schedules.nextRunAt <= now`，建立 JobRun。
|
||
10. 實作 API 與文件。
|