修复恢复记录字段缺失导致历史任务列表为空

This commit is contained in:
magicwang1111
2026-10-28 11:41:02 +08:00
parent 9b41552baf
commit b9eb357571
4 changed files with 103 additions and 0 deletions
+63
View File
@@ -0,0 +1,63 @@
const assert = require('node:assert/strict');
const { spawnSync } = require('node:child_process');
const path = require('node:path');
const { chromium } = require(process.env.PLAYWRIGHT_MODULE || 'playwright');
const fixture = spawnSync(process.env.COMFY_TEST_PYTHON || 'python', ['-c', `
import importlib.util,json,sys,tempfile
from pathlib import Path
root=Path.cwd()
sys.path.insert(0,str(root.parents[1]))
from comfy_execution.jobs import get_all_jobs,normalize_history_item
spec=importlib.util.spec_from_file_location('history_store',root/'utils/history_store.py')
module=importlib.util.module_from_spec(spec); spec.loader.exec_module(module)
with tempfile.TemporaryDirectory() as directory:
store=module.HistoryStore(str(Path(directory)/'history.db'))
old={'prompt':[0,'recovered',{}, {'create_time':1000},[]], 'user_id':'a',
'api_recovery':{'task':{'state':'pending'}},'outputs':{},
'status':{'status_str':'error','completed':False,'messages':[
['execution_start',{'timestamp':1000}],
['execution_error',{'node_id':'7','node_type':'APIRecovery',
'exception_type':'WorkerInterrupted','exception_message':'Interrupted','timestamp':1100}]]}}
store.save('recovered',old,10)
store.save('previous',{'prompt':[1,'previous',{}, {'create_time':900},[]],
'user_id':'a','outputs':{},'status':{'status_str':'success','completed':True,'messages':[]}},10)
history=store.query(owner_id='a',for_jobs=True)
jobs,total=get_all_jobs([],[],history,limit=200,offset=0)
print(json.dumps({'page':{'jobs':jobs,'pagination':{'offset':0,'limit':200,'total':total,'has_more':False}},
'detail':normalize_history_item('recovered',store.query(prompt_id='recovered')['recovered'],include_outputs=True)}))
`], { cwd: path.resolve(__dirname, '..'), encoding: 'utf8' });
assert.equal(fixture.status, 0, fixture.stderr);
const payload = JSON.parse(fixture.stdout);
(async () => {
const browser = await chromium.launch({ headless: true, executablePath: process.env.PLAYWRIGHT_CHROMIUM_EXECUTABLE || undefined });
try {
const page = await browser.newPage();
let broken = true;
await page.route('**/api/jobs?*', route => {
const body = structuredClone(payload.page);
if (broken) delete body.jobs.find(job => job.id === 'recovered').execution_error.traceback;
return route.fulfill({ json: body });
});
await page.route('**/api/jobs/recovered', route => route.fulfill({ json: payload.detail }));
await page.goto(process.env.COMFY_TEST_URL || 'http://127.0.0.1:8200');
await page.waitForFunction(() => window.comfyAPI?.api?.api);
const read = () => page.evaluate(async () => {
const { api } = await import('/scripts/api.js');
return (await api.getHistory()).map(job => job.id);
});
assert.deepEqual(await read(), [], 'Missing traceback should reproduce the empty history page');
broken = false;
assert.deepEqual((await read()).sort(), ['previous', 'recovered']);
const detail = await page.evaluate(async () => {
const { api } = await import('/scripts/api.js');
return api.getJobDetail('recovered');
});
assert.equal(detail.id, 'recovered');
assert.deepEqual(detail.execution_error.traceback, []);
console.log('PASS: real frontend reproduces empty history and accepts repaired legacy recovery list/detail');
} finally {
await browser.close();
}
})().catch(error => { console.error(error); process.exitCode = 1; });
+2
View File
@@ -69,6 +69,8 @@ class ApiRecoveryTests(unittest.TestCase):
visible = self.history.query(owner_id="user-a")["job-1"]
self.assertEqual("error", visible["status"]["status_str"])
self.assertFalse(visible["status"]["completed"])
self.assertEqual([], visible["status"]["messages"][1][1]["traceback"])
self.assertIsInstance(visible["prompt"][3]["create_time"], int)
self.assertEqual({}, self.history.query(owner_id="user-b"))
videos = [ref for output in visible["outputs"].values() for ref in output.get("video", [])]
self.assertEqual(1, len(videos))
+20
View File
@@ -1,7 +1,9 @@
import importlib.util
import os
import sqlite3
import tempfile
import unittest
from contextlib import closing
from pathlib import Path
@@ -116,6 +118,24 @@ class HistoryStoreTests(unittest.TestCase):
list(self.store.query(owner_id="user-a", offset=1, max_items=2, for_jobs=True)),
)
def test_legacy_recovery_is_readable_without_rewriting_or_deleting_history(self):
item = self.item("user-a", 1)
item["api_recovery"] = {"task": {"state": "pending"}}
item["status"] = {"status_str": "error", "messages": [
["execution_start", {"timestamp": 123}],
["execution_error", {"node_type": "APIRecovery", "exception_message": "Interrupted"}],
]}
self.store.save("recovered", item, 10)
self.store.save("previous", self.item("user-a", 2), 10)
for rows in (self.store.load(10), self.store.query(), self.store.query(for_jobs=True)):
self.assertEqual({"recovered", "previous"}, set(rows))
recovered = rows["recovered"]
self.assertEqual([], recovered["status"]["messages"][1][1]["traceback"])
self.assertEqual(123, recovered["prompt"][3]["create_time"])
with closing(sqlite3.connect(self.database)) as db:
stored = db.execute("SELECT data FROM history WHERE prompt_id='recovered'").fetchone()[0]
self.assertNotIn('traceback', stored)
if __name__ == "__main__":
unittest.main()
+18
View File
@@ -5,6 +5,20 @@ import time
from contextlib import closing
def _repair_recovery_fields(item: dict) -> None:
# Earlier recovery records omitted fields required by the frontend Jobs schema.
if not item.get("api_recovery"):
return
for event, data in (item.get("status") or {}).get("messages", []):
if event == "execution_start" and data.get("timestamp") is not None:
if item["prompt"][3].get("create_time") is None:
item["prompt"][3]["create_time"] = data["timestamp"]
if event == "execution_error" and data.get("node_type") == "APIRecovery":
data.setdefault("traceback", [])
data.setdefault("current_inputs", {})
data.setdefault("current_outputs", [])
class HistoryStore:
"""Persist completed ComfyUI history entries in a small SQLite database."""
@@ -55,6 +69,7 @@ class HistoryStore:
item = json.loads(data)
if not isinstance(item, dict):
continue
_repair_recovery_fields(item)
item["user_id"] = owner_id
history[prompt_id] = item
return history
@@ -122,6 +137,7 @@ class HistoryStore:
for current_prompt_id, current_owner_id, data in rows:
item = json.loads(data)
if isinstance(item, dict):
_repair_recovery_fields(item)
item["user_id"] = current_owner_id
result[current_prompt_id] = item
return result
@@ -176,6 +192,7 @@ class HistoryStore:
["execution_start", {"timestamp": int(task["submitted_at"] * 1000)}],
["execution_error", {"node_id": task["node_id"], "node_type": "APIRecovery",
"exception_type": "WorkerInterrupted", "exception_message": "服务器中断,工作流未完整执行。",
"traceback": [], "current_inputs": {}, "current_outputs": [],
"timestamp": int(time.time() * 1000)}],
]},
}
@@ -189,6 +206,7 @@ class HistoryStore:
if videos:
output["video"] = videos
item.setdefault("outputs", {})["api_recovery_" + task["id"]] = output
_repair_recovery_fields(item)
connection.execute("""
INSERT INTO history(prompt_id,owner_id,data) VALUES(?,?,?)
ON CONFLICT(prompt_id) DO UPDATE SET data=excluded.data