scheduler: батч-фаза закрывает ТОЛЬКО ещё-running (пустой _log-стаб убран)
Воркер мог сам решить судьбу задачи: complete (done), requeue (ready), close в архив (closed). Дайджест тика больше не перезаписывает его решения TaskCompleted'ом — иначе requeue'd задачи нечестно закрывались бы дайджестом (поверх проверенного результата «последний-победил»). found при e-261003094432
This commit is contained in:
parent
d8a7c89ef9
commit
3b3e3d17e9
1 changed files with 17 additions and 13 deletions
|
|
@ -539,19 +539,23 @@ def run_level_tick(level_cfg, seq: int, model: dict):
|
||||||
# ЗАКРЫТЬ задачу сам (cli complete) ещё до батч-фазы — его TaskCompleted
|
# ЗАКРЫТЬ задачу сам (cli complete) ещё до батч-фазы — его TaskCompleted
|
||||||
# содержит целевой ответ; повторный append поверх перезаписал бы answer
|
# содержит целевой ответ; повторный append поверх перезаписал бы answer
|
||||||
# дайджестом тика и испортил бы проверенный результат (перезапись
|
# дайджестом тика и испортил бы проверенный результат (перезапись
|
||||||
# «последний-победил» в проекции). Закрываем только ещё-running.
|
# «последний-победил» в проекции). ЗАКРЫВАЕМ ТОЛЬКО ЕЩЁ-RUNNING:
|
||||||
_model = es.run_projections() or {} # стаб/None в тестах — не валим
|
# воркер мог решить сам — complete (done), requeue (готово) или
|
||||||
|
# закрыть в архив (closed); дайджест тика не перезаписывает его
|
||||||
|
# решения (авто-схлопывание фейлов, e-261003094432: без этого фикса
|
||||||
|
# requeue'd задачи нечестно закрывались бы дайджестом).
|
||||||
|
_model = es.run_projections()
|
||||||
|
if _model is None:
|
||||||
|
# стаб/None в тестах (mocked es) — прежний путь: закрываем все claimed
|
||||||
|
_resolve = list(claimed or [])
|
||||||
|
else:
|
||||||
_tasks = _model.get("tasks") or {}
|
_tasks = _model.get("tasks") or {}
|
||||||
_done = {tid for tid in (claimed or [])
|
_resolve = [tid for tid in (claimed or [])
|
||||||
if (_tasks.get(tid) or {}).get("status") == "done"}
|
if (_tasks.get(tid) or {}).get("status") == "running"]
|
||||||
_first = True
|
_skipped = sorted(set(claimed or []) - set(_resolve))
|
||||||
for tid in (claimed or []):
|
if _skipped:
|
||||||
if tid in _done:
|
_log("skip батч-догзакрытие задач, решённых воркером: " + ", ".join(_skipped))
|
||||||
if _first:
|
for tid in _resolve:
|
||||||
_log("skip батч-догзакрытие готовых задач: "
|
|
||||||
+ ", ".join(sorted(_done)))
|
|
||||||
_first = False
|
|
||||||
continue
|
|
||||||
es.append_event("task", tid, "TaskCompleted", {
|
es.append_event("task", tid, "TaskCompleted", {
|
||||||
"result_ref": trace_id, "answer": (text or "")[:2000],
|
"result_ref": trace_id, "answer": (text or "")[:2000],
|
||||||
})
|
})
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue