diff --git a/src/jira_hub/queue.py b/src/jira_hub/queue.py index 9521e85..6544caa 100644 --- a/src/jira_hub/queue.py +++ b/src/jira_hub/queue.py @@ -108,10 +108,21 @@ def _definer_status(conn, batch_id, temp_key): return r["status"] if r else "failed" +def _definer_stale(conn, batch_id, temp_key) -> bool: + """Определитель взят агентом и давно не отвечает (старше STALE_MINUTES).""" + cutoff = (db.utcnow() - timedelta(minutes=STALE_MINUTES)).isoformat() + return conn.execute("SELECT 1 FROM command WHERE batch_id = ? AND type = 'ensure_issue' AND status = 'claimed' " + "AND claimed_at <= ? AND json_extract(params, '$.temp_key') = ?", + (batch_id, cutoff, temp_key)).fetchone() is not None + + def _next(conn) -> dict | None: rows = conn.execute("SELECT c.* FROM command c JOIN batch b ON b.id = c.batch_id " "WHERE c.status = 'queued' AND b.cancelled_at IS NULL ORDER BY b.created_at, b.id, c.seq").fetchall() + blocked_batches = set() # пакеты, ждущие зависшего определителя (R-13) for row in rows: + if row["batch_id"] in blocked_batches: + continue cmd = _cmd(row) needed = C.temp_keys_used(cmd) if needed: @@ -119,7 +130,10 @@ def _next(conn) -> dict | None: unresolved = [k for k in needed if k not in mapping] pending = [k for k in unresolved if _definer_status(conn, cmd["batch_id"], k) not in TERMINAL] if pending: - return None # создание ещё не завершено — ждём + if all(_definer_stale(conn, cmd["batch_id"], k) for k in pending): + blocked_batches.add(cmd["batch_id"]) # определитель завис: остаток пакета ждёт человека, + continue # очередь идёт к следующему пакету + return None # создание ещё не завершено — ждём (одна в полёте) cmd["params"] = C.substitute(cmd["type"], cmd["params"], mapping) if cmd["type"] == "refresh": cmd["params"]["keys"] = [k for k in cmd["params"].get("keys", []) if not C.is_temp(k)] diff --git a/tests/test_queue.py b/tests/test_queue.py index b092239..b3b6399 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -265,3 +265,33 @@ def test_reconcile_does_not_overwrite_finished_command(monkeypatch): assert queue.reconcile(conn) == [] assert queue.get_command(conn, c["id"])["status"] == "cancelled" assert mirror.get_note(conn, "new:1") is not None # rename_note не выполнялся + + +def _two_batches_with_dependent(conn): + a = queue.create_batch(conn, "A", "claude", [ + {"type": "ensure_issue", "params": {"temp_key": "new:1", "issue_type": "task", "summary": "Тема"}}, + {"type": "ensure_link", "params": {"source": "new:1", "target": "A-1", "link_type": "связанные"}}]) + first = queue.claim(conn) + assert first["type"] == "ensure_issue" + b = queue.create_batch(conn, "B", "claude", [{"type": "refresh", "params": {"jql": "project = X"}}]) + return a, b, first + + +def test_stale_definer_does_not_block_other_batches(): + """R-13: зависший ensure_issue с зависимыми не держит очередь целиком.""" + conn = fresh() + a, b, first = _two_batches_with_dependent(conn) + conn.execute("UPDATE command SET claimed_at = '2020-01-01T00:00:00+00:00' WHERE id = ?", (first["id"],)) + got = queue.claim(conn) + assert got is not None and got["batch_id"] == b["id"] and got["type"] == "refresh" + # остаток пакета A остаётся в очереди, не выдаётся и не пропускается + assert [c["status"] for c in queue.batch_view(conn, a["id"])["commands"]][1:] == ["queued", "queued"] + + +def test_fresh_definer_still_blocks_queue(): + conn = fresh() + a, b, first = _two_batches_with_dependent(conn) + assert queue.claim(conn) is None # одна в полёте + conn.execute("UPDATE command SET status = 'done', result = '{\"key\": \"A-9\"}' WHERE id = ?", (first["id"],)) + nxt = queue.claim(conn) + assert nxt["batch_id"] == a["id"] and nxt["params"]["source"] == "A-9"