PostgreSQL persistence workflow practice: build a task orchestration system in 30 minutes

-- 工作流实例表
CREATE TABLE workflows (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    workflow_type VARCHAR(100) NOT NULL,
    status VARCHAR(20) NOT NULL DEFAULT 'pending',
    input_data JSONB,
    current_step INTEGER DEFAULT 0,
    max_retries INTEGER DEFAULT 3,
    error_message TEXT,
    created_at TIMESTAMPTZ DEFAULT now(),
    updated_at TIMESTAMPTZ DEFAULT now()
);

-- 步骤执行记录（检查点）
CREATE TABLE workflow_steps (
    id SERIAL PRIMARY KEY,
    workflow_id UUID REFERENCES workflows(id),
    step_number INTEGER NOT NULL,
    step_name VARCHAR(200),
    status VARCHAR(20) NOT NULL DEFAULT 'pending',
    output_data JSONB,
    started_at TIMESTAMPTZ,
    completed_at TIMESTAMPTZ,
    error_message TEXT,
    UNIQUE(workflow_id, step_number)
);

-- 索引：加速 Worker 轮询
CREATE INDEX idx_workflows_status ON workflows(status, created_at);
CREATE INDEX idx_workflow_steps_lookup ON workflow_steps(workflow_id, step_number);

import psycopg2
import json

def dequeue_workflow(conn):
    """从 Postgres 表中取出一条待执行的工作流"""
    with conn.cursor() as cur:
        cur.execute("""
            SELECT id, workflow_type, input_data, current_step
            FROM workflows
            WHERE status = 'pending'
            ORDER BY created_at
            LIMIT 1
            FOR UPDATE SKIP LOCKED
        """)
        row = cur.fetchone()
        if row:
            cur.execute(
                "UPDATE workflows SET status = 'running', updated_at = now() WHERE id = %s",
                (row[0],)
            )
            conn.commit()
            return {
                "id": row[0],
                "type": row[1],
                "input": row[2],
                "step": row[3]
            }
    conn.rollback()
    return None

def execute_step(conn, workflow_id, step_number, step_fn, step_name=""):
    """执行一个步骤，自动处理跳过已完成的步骤和故障恢复"""

# 先检查该步骤是否已完成（故障恢复的关键）
    with conn.cursor() as cur:
        cur.execute(
            """SELECT status, output_data FROM workflow_steps
               WHERE workflow_id = %s AND step_number = %s""",
            (workflow_id, step_number)
        )
        existing = cur.fetchone()
        if existing and existing[0] == 'completed':
            return existing[1]  # 已完成的步骤直接返回，零开销跳过

# 创建步骤记录（ON CONFLICT DO NOTHING 防止并发重复）
    with conn.cursor() as cur:
        cur.execute(
            """INSERT INTO workflow_steps (workflow_id, step_number, step_name, status, started_at)
               VALUES (%s, %s, %s, 'running', now())
               ON CONFLICT (workflow_id, step_number) DO NOTHING""",
            (workflow_id, step_number, step_name)
        )
        conn.commit()

    try:
        result = step_fn()
        with conn.cursor() as cur:
            cur.execute(
                """UPDATE workflow_steps
                   SET status = 'completed', output_data = %s, completed_at = now()
                   WHERE workflow_id = %s AND step_number = %s""",
                (json.dumps(result), workflow_id, step_number)
            )
            cur.execute(
                "UPDATE workflows SET current_step = %s, updated_at = now() WHERE id = %s",
                (step_number, workflow_id)
            )
        conn.commit()
        return result
    except Exception as e:
        with conn.cursor() as cur:
            cur.execute(
                """UPDATE workflow_steps
                   SET status = 'failed', error_message = %s, completed_at = now()
                   WHERE workflow_id = %s AND step_number = %s""",
                (str(e), workflow_id, step_number)
            )
        conn.commit()
        raise

-- 查看 24 小时内各类型工作流的成功率
SELECT
    workflow_type,
    COUNT(*) AS total,
    COUNT(*) FILTER (WHERE status = 'completed') AS completed,
    ROUND(100.0 * COUNT(*) FILTER (WHERE status = 'completed') / COUNT(*), 1) AS success_pct
FROM workflows
WHERE created_at > now() - INTERVAL '24 hours'
GROUP BY workflow_type;

-- 找出卡住超过 5 分钟的工作流（可能是 Worker 崩溃未恢复）
SELECT id, workflow_type, current_step, updated_at
FROM workflows
WHERE status = 'running'
  AND updated_at < now() - INTERVAL '5 minutes';

-- 分析各步骤平均执行时长，找出瓶颈步骤
SELECT
    w.workflow_type,
    ws.step_name,
    ROUND(AVG(EXTRACT(EPOCH FROM (ws.completed_at - ws.started_at)))::numeric, 2) AS avg_seconds
FROM workflow_steps ws
JOIN workflows w ON w.id = ws.workflow_id
WHERE ws.status = 'completed'
GROUP BY w.workflow_type, ws.step_name
ORDER BY avg_seconds DESC;