clanker-saddle/test/bundles/ghostworking.py

"""The ghostworking bundle (goal 28): the single-saddle ghost 'working' row.

A saddle reboot can kill a round without any transition push; on a
single-saddle fleet the recovery passes that COULD clear the row never
fire — requeue_tasks_on_offline_saddles needs the saddle row offline
(the rebooted saddle reads online while running something else), and
requeue_interrupted_sessions rides an /next-task ask (the one saddle
is busy running another session, so no ask ever comes). The dispatcher
tick's ghost-working sweep requeues a stale open session on an
assigned/working task after STALE_WORKING_S without either crutch.
"""
from harness import *  # noqa: F401,F403 — the shared wire helpers

# TaskService.STALE_WORKING_S is 20 minutes; the sweep reads it from
# SADDLE_STALE_WORKING_S so the test can compress the window.
STALE_ENV = {"SADDLE_STALE_WORKING_S": "6"}


def run(ctx):
    auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token

    print("== goal 28: a ghost 'working' row requeues from the dispatcher tick ==")
    saddles_state = slot_public(auth, "/saddles", "saddles")
    act(auth, saddles_state["create_capability"], {"name": "ghostwake"})
    cap_id = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")["capabilities"]
                  if c["name"] == "ghostwake"][0])

    # The ghost: a daemon that claims a task, pushes working, then
    # DIES silently (no further pushes, no transition — a reboot).
    ghost = FakeSaddle()
    ghost.ask_on_register = False
    ghost.hold_state = True
    ghost.serve()
    act(auth, slot_public(auth, "/saddles", "saddles")["register"],
        {"name": "ghost-daemon", "host": "127.0.0.1:%d" % ghost.port})
    act(auth, saddle_card(auth, "ghost-daemon")["attach_capability"],
        {"capability_id": str(cap_id)})

    task_id = create_queued(auth, {
        "project_id": "1", "title": "Ghost: dies without a transition",
        "priority": "2", "capability_id": [str(cap_id)]})
    ask_status, assignment = ghost.ask_now()
    check(ask_status == 200 and int(assignment["task"]["id"]) == task_id,
          "the ask claims the queued task")
    ghost.state = "working"
    check(ghost.push_results("working") == 200, "the working push opens the round")

    conn = db()
    row = conn.execute(
        "select status from tasks where id = ?", (task_id,)).fetchone()
    check(row is not None and row[0] == "working", "the task row reads working")

    # The LIVE peer on another saddle: pushes keep coming, so the sweep
    # must never touch it (the seeded clock + fresh pushes protect it).
    live = FakeSaddle()
    live.ask_on_register = False
    live.hold_state = True
    live.serve()
    act(auth, slot_public(auth, "/saddles", "saddles")["register"],
        {"name": "live-daemon", "host": "127.0.0.1:%d" % live.port})
    act(auth, saddle_card(auth, "live-daemon")["attach_capability"],
        {"capability_id": str(cap_id)})
    live_task = create_queued(auth, {
        "project_id": "1", "title": "Live: keeps pushing", "priority": "1",
        "capability_id": [str(cap_id)]})
    ask_status, live_assignment = live.ask_now()
    check(ask_status == 200 and int(live_assignment["task"]["id"]) == live_task,
          "the live peer claims its task")
    live.state = "working"
    check(live.push_results("working") == 200, "the live round pushes working")

    # Silence the ghost: no pushes for the (compressed) stale window
    # while the live daemon keeps pushing. The sweep must requeue ONLY
    # the ghost's row, and the requeued task must be claimable again.
    deadline = time.time() + 40
    requeued = False
    while time.time() < deadline:
        live.state = "working"
        live.push_results("working")
        row = conn.execute(
            "select status from tasks where id = ?", (task_id,)).fetchone()
        if row is not None and row[0] == "queued":
            requeued = True
            break
        time.sleep(2.0)
    check(requeued,
          "the ghost 'working' row requeues once its silence passes the stale window")

    row = conn.execute(
        "select status from tasks where id = ?", (live_task,)).fetchone()
    check(row is not None and row[0] == "working",
          "the live, still-pushing round is NOT touched by the sweep")

    # The requeued task is claimable again — the fleet is unwedged.
    claim_status, reclaim = ghost.ask_now()
    check(claim_status == 200 and int(reclaim["task"]["id"]) == task_id,
          "the requeued ghost task is claimable again")

    ghost.state = "done"
    ghost.push_results("done", summary="# done")
    live.state = "done"
    live.push_results("done", summary="# done")
    wait_for(lambda: task_state(auth, task_id).get("status") == "done"
             or False, timeout=30, what="ghost task done")
    wait_for(lambda: task_state(auth, live_task).get("status") == "done"
             or False, timeout=30, what="live task done")