Total memory allocated: 0.444GB
Peak memory usage: 0.735GB
Top allocations by function:
do_execute              0.163GB   ← SQLAlchemy executor
_populate_full          …说起来，← ORM row population
fetchall                …/ pre

证据解读 / h4

li **do_execute** 占用 **0.163 GB** → 大块来自 **SQLAlchemy 执行查询**。li **_populate_full** 次数激增 → **一次性加载海量行** 为 ORM 对象。li **Peak .7 GB** 单次峰值远超合理范围。/ ul

This confirms that heartbeat worker’s memory explosion originates from a massive DB query that materializes thousands of ORM objects in one go.

bash grep -rn “task_status_sync_job” --include=”*.py” text app/scheduler/task_status_sync_job.py The task is scheduled by Celery beat with `soft_time_limit=` seconds. ### 深入心跳同步逻辑 Core logic lives in `domain/ability/task_status_sync_ability.py`,which performs two queries: 1️⃣ `get_pending_and_running_tasks` – fetches all **PENDING** and **RUNNING** tasks. 2️⃣ `get_success_tasks_with_incomplete_progress` – fetches all **SUCCESS** tasks whose `progress!= ，`. Both results are n synchronized back to Celery. ### 发现问题 In `infrastructure/gatewayimpl/task_gateway_impl.py` two methods **reuse heavy business‑side query APIs**: * SELECT * . * `.all` loads every row at once. * No time window → full‑table scan that grows over time. * Filtering happens in Python after loading all rows.

. 根因分析

LI . 修复方案

List的观点是。... def find_success_with_incomplete_progress -> List: ... * Returns only fields required for heartbeat . No `result`. --- ### 修复三：为 PENDING/RUNNING 加时间窗口 + ASC 排序 python def find_pending_running_for_heartbeat: cutoff = datetime.utcnow - timedelta # 最近7天的活跃任务 return .filter,TaskQueueDO.created_time>= cutoff) .order_by) .limit .with_entities(TaskQueueDO.task_id,TaskQueueDO.status,TaskQueueDO.progress) .all) * 老化任务被视为僵尸，不再参与同步。--- ### 修复四：为 SUCCESS 异常下推过滤 + DESC 排序 python def find_success_with_incomplete_progress: cutoff = datetime.utcnow - timedelta # 最近30天的成功异常 return .filter(TaskQueueDO.status == "SUCCESS",TaskQueueDO.progress!= EXPECTED_PROGRESS,TaskQueueDO.last_updated_time>= cutoff) .order_by) .limit .with_entities(TaskQueueDO.task_id。TaskQueueDO.status,TaskQueueDO.progress) .all) * 完全在 SQL 层完成过滤，避免 Python 中的大数组遍历。--- ### 修复五：在能力层使用轻量查询 python # domain/ability/task_status_sync_ability.py def get_pending_and_running_tasks: return self.task_gateway.find_pending_running_for_heartbeat def get_success_tasks_with_incomplete_progress: return self.task_gateway.find_success_with_incomplete_progress * 外部接口保持兼容，仅内部实现换成轻量方法。--- ### 修复六：批量缓存 Celery 状态避免 N+1 调用 python states = AsyncResult.bulk_get_state # 假设实现了 bulk 接口 for task,state in zip: …* 大幅降低网络往返和对象创建次数。其实，--- ### 修复七：为 task_queue 添加必要索引 xml * 索引创建在停机窗口完成。不会影响线上流量，

. Liquibase 旧迁移幂等性问题 #### 新问题 上线后容器重启时报错： Unexpected error running Liquibase: Migration failedERROR: relation "idx_eval_dataset_file_dataset_id" already exists #### 原因 旧 changeset 使用了裸 `CREATE INDEX …`，当数据库已手动创建该索引而 Liquibase 的 changelog 表缺失对应记录时会 执行导致冲突。#### 修复方式 将所有非幂等 DDL 改为带 IF NOT EXISTS，并在必要时执行： bash liquibase clearCheckSums # 重置 checksum。使修改生效 #### 教训 所有迁移脚本必须是幂等的——特别是 CREATE INDEX、CREATE TABLE 等操作。按理说，