|
|
@@ -7,20 +7,34 @@ LLM), this suite spawns everything it needs:
|
|
|
- the clanker-saddle app binary (path from argv[1], or builddir default)
|
|
|
in a scratch directory with its own web-config.json (statum keys copied
|
|
|
from the repo, plus `saddle_callback_base_url`) and SQLite database
|
|
|
-- an in-process fake saddle daemon speaking the wire contract: POST
|
|
|
- /start (captures the payload), GET /status (scripted state machine),
|
|
|
- GET /output (empty), GET /log?since= (raw PTY chunks, one new chunk
|
|
|
- per request so since-paging is exercised)
|
|
|
+- in-process fake saddle daemons speaking the wire contract: POST
|
|
|
+ /register (plan 10: receives {saddle_token, web_base_url}, answers
|
|
|
+ {ok, ssh_public_key, version}), POST /wake (records the wake; the
|
|
|
+ daemon then asks POST /api/saddle/next-task like a phase-1 daemon),
|
|
|
+ POST /start (captures the payload — the rollback path), GET /status
|
|
|
+ (scripted state machine), GET /output (empty), GET /log?since= (raw
|
|
|
+ PTY chunks, one new chunk per request so since-paging is exercised)
|
|
|
|
|
|
It then drives the app the way statum.js does (entrypoint loads + action
|
|
|
POSTs) and asserts the Workstream B features against the wire contract:
|
|
|
|
|
|
3. plan-review blocks hide their duplicate body preview
|
|
|
-5. rules CRUD/attach + the effective `rules` array in /start
|
|
|
+5. rules CRUD/attach + the effective `rules` array in assignments
|
|
|
6. project notes, POST /api/saddle/note, `project_notes` + `callback`
|
|
|
- in /start
|
|
|
+ in assignments
|
|
|
plus the raw-log side: session_raw_log polling + terminal raw view
|
|
|
|
|
|
+Plan 10 phase 1 (work acquisition flip):
|
|
|
+
|
|
|
+- per-saddle tokens: registration mints + /register-delivers them,
|
|
|
+ legacy tokens migrate onto the row at startup, removal revokes
|
|
|
+- token auth on every /api/saddle/* endpoint
|
|
|
+- next-task allocation: concurrent asks race exactly one winner,
|
|
|
+ affinity pinning for parked tasks, resume_notice on suspended resumes
|
|
|
+- dispatcher: wake -> ask -> allocate roundtrip, the lost-wake guard
|
|
|
+ (no ask after a wake -> offline, re-wake the next capable), and the
|
|
|
+ pre-flip /answer + /start resume paths kept as the rollback route
|
|
|
+
|
|
|
Prints PASS/FAIL lines Statum-style; exits non-zero on any failure.
|
|
|
"""
|
|
|
import json
|
|
|
@@ -194,44 +208,65 @@ def db():
|
|
|
|
|
|
|
|
|
class FakeSaddle:
|
|
|
- """In-process saddle daemon implementing the wire contract."""
|
|
|
+ """In-process saddle daemon implementing the wire contract.
|
|
|
+
|
|
|
+ Phase-1 ask behaviour: the daemon asks POST /api/saddle/next-task
|
|
|
+ whenever idle-awake — at registration (its token arrives via
|
|
|
+ /register), on POST /wake, after its task reaches a terminal state,
|
|
|
+ and whenever a /status poll observes it parked idle (throttled).
|
|
|
+ Every trigger can be disabled per test for determinism; ask_now()
|
|
|
+ performs a synchronous manual ask.
|
|
|
+ """
|
|
|
|
|
|
PROMPT = "bbarrow@fedora:/workspace/task-1$ "
|
|
|
|
|
|
def __init__(self):
|
|
|
self.lock = threading.Lock()
|
|
|
- self.start_payloads = [] # captured POST /start bodies
|
|
|
+ self.start_payloads = [] # captured assignment payloads (/start + next-task 200s)
|
|
|
self.chunks = [] # [{seq, text}] — the raw log ring
|
|
|
self.next_seq = 1
|
|
|
self.state = "idle"
|
|
|
self.summary = ""
|
|
|
self.state_entered = time.monotonic()
|
|
|
- self.fail_next = False # next /start drives the task to failed
|
|
|
- self.start_counts = {} # task id -> /start count
|
|
|
+ self.fail_next = False # next task drives the task to failed
|
|
|
+ self.start_counts = {} # task id -> assignment count
|
|
|
self.resumed_pending = False # crash-resume flag on the next /status
|
|
|
- self.suspend_after_start = False # next task parks as suspended
|
|
|
+ self.suspend_on_title = None # this task (by title) parks as suspended
|
|
|
self.pending_output = [] # entries drained by the next /output poll
|
|
|
self.output_seq = 100
|
|
|
self.last_activity_epoch = int(time.time())
|
|
|
self.subtask_status_events = [] # recorded POST /subtask-status payloads
|
|
|
self.fail_subtask_status = False # exercise the web's retry path
|
|
|
- self.hang_next = {} # path suffix -> hang count (never responds)
|
|
|
+ self.hang_next = {} # path suffix -> hang count (never responds)
|
|
|
self.fail_log_count = 0 # /log requests failing (500) while > 0
|
|
|
self.scripted_log = None # fixed chunk set served by /log
|
|
|
self.scripted_next = 0 # the daemon-style next for it
|
|
|
self.scripted_corpus = None # chunks "produced" when a task starts
|
|
|
self.race_texts = None # chunks appended DURING /log response
|
|
|
- # construction (the daemon-side bug)
|
|
|
+ # construction (the daemon-side bug)
|
|
|
self.answer_bodies = [] # recorded POST /answer payloads
|
|
|
self.abort_failures = 0 # /abort requests failing (500) while > 0
|
|
|
self.abort_log = [] # task ids aborted via POST /abort
|
|
|
self.running_task = 0 # task id the daemon believes it runs
|
|
|
+ self.running_title = "" # its title (suspend targeting)
|
|
|
self.zombie_409 = 0 # /start 409s once per unit (then decrements)
|
|
|
self.conflict_running = 0 # while > 0, /start persistently 409s with this id
|
|
|
self.hold_state = False # freeze /status lifecycle transitions
|
|
|
- self.ssh_public_key = "" # reported on /status (plan-9 contract)
|
|
|
+ self.ssh_public_key = "" # reported on /status + /register
|
|
|
self.started_mono = time.monotonic()
|
|
|
self.freeze_activity = False # hold last_activity_epoch still
|
|
|
+ # --- plan 10 phase 1: ask/wake/register ---
|
|
|
+ self.token = None # per-saddle token from POST /register
|
|
|
+ self.register_log = [] # recorded /register bodies
|
|
|
+ self.wake_log = [] # recorded POST /wake arrivals
|
|
|
+ self.asks = [] # [{"status": int, "payload": dict|None}]
|
|
|
+ self.ask_on_register = True
|
|
|
+ self.ask_on_wake = True
|
|
|
+ self.ask_on_terminal = True
|
|
|
+ self.ask_on_idle_status = True
|
|
|
+ self.serve_register = True # False: 404 /register (pre-flip daemon)
|
|
|
+ self.last_ask_trigger = 0.0
|
|
|
+ self.port = None
|
|
|
|
|
|
def set_state(self, state, summary=None):
|
|
|
"""Advances the scripted lifecycle.
|
|
|
@@ -279,6 +314,63 @@ class FakeSaddle:
|
|
|
chunks = [c for c in self.chunks if c["seq"] >= since]
|
|
|
return chunks, self.next_seq
|
|
|
|
|
|
+ # --- plan 10 phase 1: the ask path ---
|
|
|
+
|
|
|
+ def ask_now(self):
|
|
|
+ """Synchronously asks the web for work (a phase-1 daemon's idle
|
|
|
+ ask). A 200 assignment is accepted exactly like POST /start;
|
|
|
+ returns (status, payload|None)."""
|
|
|
+ token = self.token
|
|
|
+ if token is None:
|
|
|
+ raise SystemExit("fake daemon asked before receiving its token")
|
|
|
+ status, payload = saddle_api("POST", "/api/saddle/next-task", token)
|
|
|
+ with self.lock:
|
|
|
+ self.asks.append({"status": status, "payload": payload})
|
|
|
+ if status == 200:
|
|
|
+ self.accept_assignment(payload)
|
|
|
+ return status, payload
|
|
|
+
|
|
|
+ def _trigger_ask(self, reason):
|
|
|
+ """Schedules a delayed ask (the daemon woke / went idle-awake)."""
|
|
|
+ def run():
|
|
|
+ try:
|
|
|
+ self.ask_now()
|
|
|
+ except OSError:
|
|
|
+ pass # app restarting / shutting down mid-test
|
|
|
+ threading.Thread(target=lambda: (time.sleep(0.2), run()), daemon=True).start()
|
|
|
+
|
|
|
+ def maybe_trigger_ask(self, reason):
|
|
|
+ now = time.monotonic()
|
|
|
+ if now - self.last_ask_trigger < 1.0:
|
|
|
+ return
|
|
|
+ self.last_ask_trigger = now
|
|
|
+ self._trigger_ask(reason)
|
|
|
+
|
|
|
+ def accept_assignment(self, payload):
|
|
|
+ """A 200 next-task assignment: adopted exactly like POST /start."""
|
|
|
+ self.start_payloads.append(payload)
|
|
|
+ task = payload.get("task") or {}
|
|
|
+ task_id = task.get("id")
|
|
|
+ self.running_task = int(task_id or 0)
|
|
|
+ self.running_title = task.get("title") or ""
|
|
|
+ count = self.start_counts.get(task_id, 0) + 1
|
|
|
+ self.start_counts[task_id] = count
|
|
|
+ if count > 1:
|
|
|
+ # Crash-resume contract: re-pushing a task whose
|
|
|
+ # workspace still exists restores it; the first
|
|
|
+ # /status after the restore reports "resumed": true.
|
|
|
+ self.resumed_pending = True
|
|
|
+ self.append_chunk(self.PROMPT)
|
|
|
+ self.append_chunk("echo hello\r\n")
|
|
|
+ self.append_chunk("hello\r\n")
|
|
|
+ self.set_state("failing" if self.fail_next else "working")
|
|
|
+ self.fail_next = False
|
|
|
+ if self.scripted_corpus is not None:
|
|
|
+ with self.lock:
|
|
|
+ for chunk in self.scripted_corpus:
|
|
|
+ self.scripted_log.append(dict(chunk))
|
|
|
+ self.scripted_next = self.scripted_corpus[-1]["seq"] + 1
|
|
|
+
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
|
daemon = None # set on the class before serving
|
|
|
protocol_version = "HTTP/1.1"
|
|
|
@@ -307,6 +399,34 @@ class FakeSaddle:
|
|
|
self.daemon.touch_activity()
|
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
|
raw = self.rfile.read(length).decode() if length else "{}"
|
|
|
+ if self.path == "/register":
|
|
|
+ # Plan 10: the web delivers the per-saddle token + its
|
|
|
+ # own base URL; the daemon persists the token for its
|
|
|
+ # calls home and reports its ssh key + version.
|
|
|
+ if not self.daemon.serve_register:
|
|
|
+ self._send(404, {"error": "no such route"})
|
|
|
+ return
|
|
|
+ try:
|
|
|
+ payload = json.loads(raw)
|
|
|
+ except ValueError:
|
|
|
+ payload = {"raw": raw}
|
|
|
+ self.daemon.register_log.append(payload)
|
|
|
+ if payload.get("saddle_token"):
|
|
|
+ self.daemon.token = payload["saddle_token"]
|
|
|
+ self._send(200, {"ok": True,
|
|
|
+ "ssh_public_key": self.daemon.ssh_public_key,
|
|
|
+ "version": "0.1-fake"})
|
|
|
+ if self.daemon.ask_on_register:
|
|
|
+ self.daemon.maybe_trigger_ask("register")
|
|
|
+ return
|
|
|
+ if self.path == "/wake":
|
|
|
+ # Plan 10: a capable task is ready; a sleeping daemon
|
|
|
+ # asks for work.
|
|
|
+ self.daemon.wake_log.append(time.monotonic())
|
|
|
+ self._send(200, {"ok": True})
|
|
|
+ if self.daemon.ask_on_wake:
|
|
|
+ self.daemon.maybe_trigger_ask("wake")
|
|
|
+ return
|
|
|
if self.path == "/start":
|
|
|
if self.daemon.conflict_running > 0 or self.daemon.zombie_409 > 0:
|
|
|
# Wire contract: a machine-readable conflict.
|
|
|
@@ -327,26 +447,7 @@ class FakeSaddle:
|
|
|
payload = json.loads(raw)
|
|
|
except ValueError:
|
|
|
payload = {"raw": raw}
|
|
|
- self.daemon.start_payloads.append(payload)
|
|
|
- task_id = (payload.get("task") or {}).get("id")
|
|
|
- self.daemon.running_task = int(task_id or 0)
|
|
|
- count = self.daemon.start_counts.get(task_id, 0) + 1
|
|
|
- self.daemon.start_counts[task_id] = count
|
|
|
- if count > 1:
|
|
|
- # Crash-resume contract: re-pushing a task whose
|
|
|
- # workspace still exists restores it; the first
|
|
|
- # /status after the restore reports "resumed": true.
|
|
|
- self.daemon.resumed_pending = True
|
|
|
- self.daemon.append_chunk(self.daemon.PROMPT)
|
|
|
- self.daemon.append_chunk("echo hello\r\n")
|
|
|
- self.daemon.append_chunk("hello\r\n")
|
|
|
- self.daemon.set_state("failing" if self.daemon.fail_next else "working")
|
|
|
- self.daemon.fail_next = False
|
|
|
- if self.daemon.scripted_corpus is not None:
|
|
|
- with self.daemon.lock:
|
|
|
- for chunk in self.daemon.scripted_corpus:
|
|
|
- self.daemon.scripted_log.append(dict(chunk))
|
|
|
- self.daemon.scripted_next = self.daemon.scripted_corpus[-1]["seq"] + 1
|
|
|
+ self.daemon.accept_assignment(payload)
|
|
|
self._send(200, {})
|
|
|
elif self.path == "/answer":
|
|
|
try:
|
|
|
@@ -365,6 +466,8 @@ class FakeSaddle:
|
|
|
self.daemon.running_task = 0
|
|
|
self.daemon.abort_log.append(aborted)
|
|
|
self._send(200, {"ok": True, "aborted_task_id": aborted})
|
|
|
+ # An aborted daemon returns to its idle loop: it asks.
|
|
|
+ self.daemon.maybe_trigger_ask("aborted")
|
|
|
elif self.path == "/subtask-status":
|
|
|
try:
|
|
|
payload = json.loads(raw)
|
|
|
@@ -399,7 +502,7 @@ class FakeSaddle:
|
|
|
self._send(200, response)
|
|
|
return
|
|
|
if self.daemon.state == "failing":
|
|
|
- # Fails fast (a rejected /start script).
|
|
|
+ # Fails fast (a rejected assignment script).
|
|
|
if time.monotonic() - self.daemon.state_entered >= 1:
|
|
|
self.daemon.set_state("failed",
|
|
|
"bash -n rejected the script (simulated)")
|
|
|
@@ -408,13 +511,21 @@ class FakeSaddle:
|
|
|
# (dispatcher polls, stale-busy healing, refreshes)
|
|
|
# must not advance the lifecycle.
|
|
|
if time.monotonic() - self.daemon.state_entered >= 8:
|
|
|
- if self.daemon.suspend_after_start:
|
|
|
- # `saddle suspend`: park until a fresh /start
|
|
|
- # resumes the workspace (one-shot flag).
|
|
|
+ if self.daemon.suspend_on_title \
|
|
|
+ and self.daemon.suspend_on_title == self.daemon.running_title:
|
|
|
+ # `saddle suspend`: park until a fresh assignment
|
|
|
+ # resumes the workspace (title-targeted one-shot,
|
|
|
+ # so an interleaved dispatch cannot eat the flag).
|
|
|
self.daemon.set_state("suspended")
|
|
|
- self.daemon.suspend_after_start = False
|
|
|
+ self.daemon.suspend_on_title = None
|
|
|
else:
|
|
|
self.daemon.set_state("done", "fake daemon finished")
|
|
|
+ # A parked/terminal daemon is idle-awake: a phase-1
|
|
|
+ # daemon asks (and sleeps on 204). Throttled so status
|
|
|
+ # poll bursts do not hammer the ask endpoint.
|
|
|
+ if self.daemon.ask_on_idle_status and self.daemon.state in (
|
|
|
+ "idle", "done", "failed", "suspended"):
|
|
|
+ self.daemon.maybe_trigger_ask("idle-status")
|
|
|
self._send(200, response)
|
|
|
elif self.path.startswith("/output"):
|
|
|
with self.daemon.lock:
|
|
|
@@ -454,10 +565,51 @@ class FakeSaddle:
|
|
|
self._send(404, {"error": "no such route"})
|
|
|
self.close_connection = True
|
|
|
|
|
|
- def serve(self):
|
|
|
+ def run_lifecycle(self):
|
|
|
+ """Applies the scripted lifecycle on a timer (like a real daemon
|
|
|
+ whose task ends on its own, not when the web happens to poll):
|
|
|
+ working ages into done/suspended, failing into failed — and a
|
|
|
+ task END makes the idle-awake daemon ask for work again."""
|
|
|
+ while True:
|
|
|
+ time.sleep(0.5)
|
|
|
+ if self.hold_state:
|
|
|
+ continue
|
|
|
+ if self.state == "failing":
|
|
|
+ if time.monotonic() - self.state_entered >= 1:
|
|
|
+ self.set_state("failed", "bash -n rejected the script (simulated)")
|
|
|
+ if self.ask_on_terminal:
|
|
|
+ self.maybe_trigger_ask("terminal")
|
|
|
+ elif self.state == "working":
|
|
|
+ if time.monotonic() - self.state_entered >= 8:
|
|
|
+ if self.suspend_on_title and self.suspend_on_title == self.running_title:
|
|
|
+ self.set_state("suspended")
|
|
|
+ self.suspend_on_title = None
|
|
|
+ # A parked daemon is idle-awake: it asks (and
|
|
|
+ # sleeps on 204) so a later wake can reach it.
|
|
|
+ if self.ask_on_idle_status:
|
|
|
+ self.maybe_trigger_ask("parked")
|
|
|
+ else:
|
|
|
+ self.set_state("done", "fake daemon finished")
|
|
|
+ if self.ask_on_terminal:
|
|
|
+ self.maybe_trigger_ask("terminal")
|
|
|
+
|
|
|
+ def serve(self, port=None):
|
|
|
+ """Binds the daemon's HTTP server. A requested port that is
|
|
|
+ already taken (free_port() hands out recyclable ports — two
|
|
|
+ calls can collide) falls back to an OS-assigned one instead of
|
|
|
+ silently dying and impersonating whatever owns that port."""
|
|
|
+ if port is None:
|
|
|
+ port = DAEMON_PORT
|
|
|
handler = type("BoundHandler", (FakeSaddle.Handler,), {"daemon": self})
|
|
|
- self.server = ThreadingHTTPServer(("127.0.0.1", DAEMON_PORT), handler)
|
|
|
+ self.server = None
|
|
|
+ try:
|
|
|
+ self.server = ThreadingHTTPServer(("127.0.0.1", port), handler)
|
|
|
+ except OSError:
|
|
|
+ print("(fake daemon: port %d taken, falling back)" % port)
|
|
|
+ self.server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
|
|
|
+ self.port = self.server.server_address[1]
|
|
|
threading.Thread(target=self.server.serve_forever, daemon=True).start()
|
|
|
+ threading.Thread(target=self.run_lifecycle, daemon=True).start()
|
|
|
return self.server
|
|
|
|
|
|
def kill(self):
|
|
|
@@ -486,12 +638,16 @@ def start_app():
|
|
|
"SADDLE_RAW_PROBE_INTERVAL_S": "5",
|
|
|
"SADDLE_STRAND_GRACE_S": "10",
|
|
|
"SADDLE_WEDGE_THRESHOLD_S": "8",
|
|
|
+ "SADDLE_LOST_WAKE_S": "2",
|
|
|
# Never inherit a caller-poisoned LD_LIBRARY_PATH (e.g. one that
|
|
|
# resolves the ancient /lib64 copies): empty makes the loader use
|
|
|
# the binary's RUNPATH, which is the dev stack it was built
|
|
|
# against.
|
|
|
"LD_LIBRARY_PATH": "",
|
|
|
})
|
|
|
+ log_previous = os.path.join(WORK, "app.log.1")
|
|
|
+ if os.path.exists(os.path.join(WORK, "app.log")):
|
|
|
+ shutil.move(os.path.join(WORK, "app.log"), log_previous)
|
|
|
log = open(os.path.join(WORK, "app.log"), "w")
|
|
|
command = [APP, str(APP_PORT)]
|
|
|
if os.environ.get("WEBSTACK_GDB"):
|
|
|
@@ -612,6 +768,30 @@ def main():
|
|
|
fake_id = int(registered[0]["id"])
|
|
|
fake_token = registered[0]["token"]
|
|
|
|
|
|
+ # Plan 10 phase 1: registration delivers the per-saddle token to
|
|
|
+ # the daemon via POST /register {saddle_token, web_base_url}.
|
|
|
+ check(wait_for(lambda: len(fake.register_log) > 0 or False,
|
|
|
+ timeout=15, what="/register delivery") is not None,
|
|
|
+ "registration POSTs /register to the daemon")
|
|
|
+ register_body = fake.register_log[0]
|
|
|
+ check(register_body.get("saddle_token") == fake_token,
|
|
|
+ "/register carries the minted per-saddle token")
|
|
|
+ check(register_body.get("web_base_url") == WEB,
|
|
|
+ "/register carries the web base URL")
|
|
|
+ check(fake.token == fake_token,
|
|
|
+ "daemon persisted the token (its asks use it)")
|
|
|
+ token_conn = db()
|
|
|
+ row_token = token_conn.execute("select token from saddles where id = ?",
|
|
|
+ (fake_id,)).fetchone()[0]
|
|
|
+ token_conn.close()
|
|
|
+ check(row_token == fake_token,
|
|
|
+ "per-saddle token lives on the saddle row")
|
|
|
+ # The register-time ask (idle-awake daemon): 204, nothing
|
|
|
+ # offerable yet — the saddle is asleep and wakeable.
|
|
|
+ check(wait_for(lambda: any(a["status"] == 204 for a in fake.asks) or False,
|
|
|
+ timeout=15, what="register-time ask") is not None,
|
|
|
+ "the daemon's idle ask returns 204 (nothing offerable)")
|
|
|
+
|
|
|
conn.execute("insert into sessions (task_id, saddle_id, started_at, ended_at, outcome)"
|
|
|
" values (?, ?, strftime('%s','now'), strftime('%s','now'), 'done')",
|
|
|
(plan_task, fake_id))
|
|
|
@@ -816,20 +996,18 @@ def main():
|
|
|
"dispatched task completes against the fake daemon")
|
|
|
|
|
|
print("== feature 6: project notes + callback ==")
|
|
|
- conn = db()
|
|
|
- callback_rows = conn.execute(
|
|
|
- "select token from api_tokens where saddle_id = 0").fetchall()
|
|
|
- conn.close()
|
|
|
- check(len(callback_rows) == 1 and callback_rows[0][0].startswith("saddle-callback-"),
|
|
|
- "dedicated saddle-callback token minted at startup")
|
|
|
- callback_token = callback_rows[0][0]
|
|
|
+ # Plan 10 phase 1: the shared saddle-callback token is gone —
|
|
|
+ # the per-saddle token authorises the callback endpoints.
|
|
|
+ callback_token = fake_token
|
|
|
+ check(callback_token.startswith("saddle-") and "saddle-callback-" not in callback_token,
|
|
|
+ "note callback authorised by the per-saddle token")
|
|
|
|
|
|
status, _ = saddle_api("POST", "/api/saddle/note", None,
|
|
|
{"project": "demo", "body": "x"})
|
|
|
check(status == 401, "note endpoint rejects a missing token")
|
|
|
status, _ = saddle_api("POST", "/api/saddle/note", "saddle-callback-not-real",
|
|
|
{"project": "demo", "body": "x"})
|
|
|
- check(status == 401, "note endpoint rejects a wrong token")
|
|
|
+ check(status == 401, "note endpoint rejects a wrong (legacy-style) token")
|
|
|
|
|
|
status, body = saddle_api("POST", "/api/saddle/note", callback_token,
|
|
|
{"task_id": rule_task, "project": "demo",
|
|
|
@@ -1341,7 +1519,7 @@ def main():
|
|
|
and "fake daemon finished" in (task_state(auth, rejected_task).get("summary") or ""),
|
|
|
"rejected entries never fail the task")
|
|
|
print("== B7: suspended state + per-subtask resume ==")
|
|
|
- fake.suspend_after_start = True
|
|
|
+ fake.suspend_on_title = "Suspend parent"
|
|
|
pushes_before = len(fake.start_payloads)
|
|
|
act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
|
|
|
"project_id": "1",
|
|
|
@@ -1412,7 +1590,7 @@ def main():
|
|
|
|
|
|
print("== R3 B1: subtask status push ==")
|
|
|
# A parked (suspended) parent gives delivery a stable open session.
|
|
|
- fake.suspend_after_start = True
|
|
|
+ fake.suspend_on_title = "Push parent"
|
|
|
act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
|
|
|
"project_id": "1",
|
|
|
"title": "Push parent",
|
|
|
@@ -1456,8 +1634,11 @@ def main():
|
|
|
"dispatch pushes a working status event")
|
|
|
|
|
|
# Best-effort retry: while the daemon 500s, events stay queued and
|
|
|
- # land once it recovers.
|
|
|
+ # land once it recovers. The parent's resumed run re-parks (the
|
|
|
+ # ask-flow resume races this section otherwise: its completion
|
|
|
+ # would close the parked session the events are delivered to).
|
|
|
fake.fail_subtask_status = True
|
|
|
+ fake.suspend_on_title = "Push parent"
|
|
|
act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
|
|
|
"project_id": "1",
|
|
|
"title": "Push child two",
|
|
|
@@ -1497,7 +1678,33 @@ def main():
|
|
|
"the busy@0 heal is logged")
|
|
|
|
|
|
# Contradiction: the daemon reports idle while the web believes the
|
|
|
- # saddle is running the orphaned task.
|
|
|
+ # saddle is running the orphaned task. Drain everything in flight
|
|
|
+ # first — forcing the daemon's state mid-run requeues whatever it
|
|
|
+ # is really executing.
|
|
|
+ def in_flight_count():
|
|
|
+ conn = db()
|
|
|
+ try:
|
|
|
+ # The plan-review fixture (task 1) is inserted 'working'
|
|
|
+ # on purpose and never dispatches; it is not in flight.
|
|
|
+ return conn.execute(
|
|
|
+ "select count(*) from tasks where status in"
|
|
|
+ " ('queued', 'assigned', 'working')"
|
|
|
+ " and title != 'Plan review task'").fetchone()[0]
|
|
|
+ finally:
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ def in_flight_titles():
|
|
|
+ conn = db()
|
|
|
+ try:
|
|
|
+ return [r[0] for r in conn.execute(
|
|
|
+ "select title from tasks where status in"
|
|
|
+ " ('queued', 'assigned', 'working')"
|
|
|
+ " and title != 'Plan review task'")]
|
|
|
+ finally:
|
|
|
+ conn.close()
|
|
|
+ check(wait_for(lambda: in_flight_count() == 0 or False,
|
|
|
+ timeout=90, what="in-flight work settles (%s)" % in_flight_titles()) is not None,
|
|
|
+ "in-flight work settles before the contradiction run")
|
|
|
conn = db()
|
|
|
conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
|
|
|
" values (1, 'Orphan task', 'working', 2, strftime('%s','now'))")
|
|
|
@@ -1517,15 +1724,11 @@ def main():
|
|
|
|
|
|
|
|
|
print("== R4 B1: synchronous subtask callback ==")
|
|
|
- conn = db()
|
|
|
- callback_rows = conn.execute(
|
|
|
- "select token from api_tokens where saddle_id = 0"
|
|
|
- " and token like 'saddle-callback-%'").fetchall()
|
|
|
- conn.close()
|
|
|
- r4_token = callback_rows[0][0] if callback_rows else callback_token
|
|
|
+ # Plan 10: the subtask callback is per-saddle-token authenticated.
|
|
|
+ r4_token = fake_token
|
|
|
|
|
|
# Parent with a live parked session (suspended) hosts the callback.
|
|
|
- fake.suspend_after_start = True
|
|
|
+ fake.suspend_on_title = "Callback parent"
|
|
|
act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
|
|
|
"project_id": "1",
|
|
|
"title": "Callback parent",
|
|
|
@@ -1623,7 +1826,7 @@ def main():
|
|
|
# Parent parks; its toggle then covers all descendants. Children are
|
|
|
# created with an unmet 'gpu' requirement so they stay queued (and
|
|
|
# untouched) until the requirement is lifted for the final-resume leg.
|
|
|
- fake.suspend_after_start = True
|
|
|
+ fake.suspend_on_title = "Auto parent"
|
|
|
act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
|
|
|
"project_id": "1",
|
|
|
"title": "Auto parent",
|
|
|
@@ -1731,44 +1934,288 @@ def main():
|
|
|
"resume notice ends with the final-review instruction")
|
|
|
|
|
|
|
|
|
- print("== R4 B3: dispatch-failure observability ==")
|
|
|
- # A dead saddle that alone carries the required tag attracts the
|
|
|
- # dispatch; the push fails and the task learns why via a bot comment.
|
|
|
+ print("== phase 1: ask/allocation — token auth + concurrent race ==")
|
|
|
+ # Plan 10 phase 1. Registration mints per-saddle tokens; the
|
|
|
+ # ask endpoint authenticates by them; concurrent asks race to
|
|
|
+ # exactly one winner through the allocation transaction.
|
|
|
+ status, _ = saddle_api("POST", "/api/saddle/next-task", None)
|
|
|
+ check(status == 401, "next-task rejects a missing token")
|
|
|
+ status, _ = saddle_api("POST", "/api/saddle/next-task", "saddle-bogus-token")
|
|
|
+ check(status == 401, "next-task rejects a wrong token")
|
|
|
+ status, body = saddle_api("POST", "/api/saddle/next-task", fake_token)
|
|
|
+ check(status == 204, "next-task answers 204 (nothing offerable — daemon sleeps)")
|
|
|
+
|
|
|
+ # Two more racers: real daemons on their own ports; fake2 reports
|
|
|
+ # its ssh key at /register so the key lands on the row directly.
|
|
|
+ fake2 = FakeSaddle()
|
|
|
+ fake2.ssh_public_key = "ssh-ed25519 AAAAFake2RacerKey saddle-2@fedora"
|
|
|
+ fake2.serve(port=free_port())
|
|
|
+ fake3 = FakeSaddle()
|
|
|
+ fake3.serve(port=free_port())
|
|
|
+ saddles_state = slot_public(auth, "/saddles", "saddles")
|
|
|
+ act(auth, saddles_state["register"], {"name": "racer-two",
|
|
|
+ "host": "127.0.0.1:%d" % fake2.port})
|
|
|
+ act(auth, saddles_state["register"], {"name": "racer-three",
|
|
|
+ "host": "127.0.0.1:%d" % fake3.port})
|
|
|
+ check(wait_for(lambda: len(fake2.register_log) > 0 and len(fake3.register_log) > 0
|
|
|
+ or False, timeout=15, what="racers registered") is not None,
|
|
|
+ "both racers received /register")
|
|
|
+ check(wait_for(lambda: "AAAAFake2RacerKey" in
|
|
|
+ [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
|
|
|
+ if s["name"] == "racer-two"][0].get("ssh_public_key", "") or False,
|
|
|
+ timeout=15, what="register-reported key") is not None,
|
|
|
+ "the ssh key reported at /register lands on the saddle row")
|
|
|
+
|
|
|
+ racer_cards = {}
|
|
|
+ for name, daemon in (("fake-saddle", fake), ("racer-two", fake2), ("racer-three", fake3)):
|
|
|
+ card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
|
|
|
+ if s["name"] == name][0]
|
|
|
+ racer_cards[daemon] = {"id": int(card["id"]), "token": card["token"]}
|
|
|
+ # Revocation: a registered saddle's token works until the saddle
|
|
|
+ # is removed, then nothing authorises its calls.
|
|
|
+ fake4 = FakeSaddle()
|
|
|
+ fake4.ask_on_register = False # a pre-flip daemon: never asks
|
|
|
+ fake4.serve(port=free_port())
|
|
|
+ saddles_state = slot_public(auth, "/saddles", "saddles")
|
|
|
+ act(auth, saddles_state["register"], {"name": "old-daemon",
|
|
|
+ "host": "127.0.0.1:%d" % fake4.port})
|
|
|
+ old_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
|
|
|
+ if s["name"] == "old-daemon"][0]
|
|
|
+ status, _ = saddle_api("POST", "/api/saddle/next-task", old_card["token"])
|
|
|
+ check(status in (200, 204),
|
|
|
+ "the old daemon's fresh token authenticates next-task")
|
|
|
+ status, _ = saddle_api("POST", "/api/saddle/note", old_card["token"],
|
|
|
+ {"project": "demo", "body": "from the old daemon"})
|
|
|
+ check(status == 200, "note accepts any registered saddle's token")
|
|
|
+ act(auth, old_card["unregister"], {})
|
|
|
+ status, _ = saddle_api("POST", "/api/saddle/next-task", old_card["token"])
|
|
|
+ check(status == 401, "removing the saddle revokes its token")
|
|
|
+ status, _ = saddle_api("POST", "/api/saddle/note", old_card["token"],
|
|
|
+ {"project": "demo", "body": "should fail"})
|
|
|
+ check(status == 401, "revocation covers the note endpoint too")
|
|
|
+
|
|
|
+ # Deterministic race: settle the queue first (the auto-approve
|
|
|
+ # stragglers must finish — their parked resumes stay offerable
|
|
|
+ # otherwise and would steal the race's asks), freeze every
|
|
|
+ # daemon's self-ask triggers, put all three to sleep (204),
|
|
|
+ # then fire one task worth of concurrent asks.
|
|
|
+ check(wait_task_done(auth, auto_child, timeout=90) is not None
|
|
|
+ and wait_task_done(auth, auto_grand, timeout=90) is not None
|
|
|
+ and wait_task_done(auth, auto_parent, timeout=90) is not None,
|
|
|
+ "the auto-approve stragglers settle before the race")
|
|
|
+ for daemon in (fake, fake2, fake3):
|
|
|
+ daemon.ask_on_register = False
|
|
|
+ daemon.ask_on_wake = False
|
|
|
+ daemon.ask_on_terminal = False
|
|
|
+ daemon.ask_on_idle_status = False
|
|
|
+ time.sleep(1.5) # let any in-flight triggered asks land
|
|
|
+ def queued_count():
|
|
|
+ conn = db()
|
|
|
+ try:
|
|
|
+ return conn.execute(
|
|
|
+ "select count(*) from tasks where status = 'queued'").fetchone()[0]
|
|
|
+ finally:
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ check(wait_for(lambda: queued_count() == 0 or False,
|
|
|
+ timeout=60, what="queue settles before the race") is not None,
|
|
|
+ "the queue settles before the race")
|
|
|
+
|
|
|
+ sleep_statuses = [daemon.ask_now()[0] for daemon in (fake, fake2, fake3)]
|
|
|
+ check(sleep_statuses == [204, 204, 204],
|
|
|
+ "all three racers sleep (204) before the task exists")
|
|
|
+
|
|
|
+ act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
|
|
|
+ "project_id": "1",
|
|
|
+ "title": "Race task",
|
|
|
+ "priority": "2",
|
|
|
+ })
|
|
|
+ race_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
|
|
|
+ if t["title"] == "Race task"][0])
|
|
|
+
|
|
|
+ race_results = []
|
|
|
+ def race_ask(daemon):
|
|
|
+ # A real racing daemon accepts the assignment it wins — the
|
|
|
+ # manual ask path does exactly that on a 200.
|
|
|
+ race_results.append(daemon.ask_now())
|
|
|
+ threads = [threading.Thread(target=race_ask, args=(daemon,))
|
|
|
+ for daemon in (fake, fake2, fake3)]
|
|
|
+ for thread in threads:
|
|
|
+ thread.start()
|
|
|
+ for thread in threads:
|
|
|
+ thread.join()
|
|
|
+ race_statuses = sorted(r[0] for r in race_results)
|
|
|
+ check(race_statuses == [200, 204, 204],
|
|
|
+ "concurrent asks race to exactly one winner (%s)" % race_statuses)
|
|
|
+ winner = [r for r in race_results if r[0] == 200][0][1]
|
|
|
+ check(int(winner["task"]["id"]) == race_task, "the winner's payload carries the task")
|
|
|
+ check(int(winner["session_id"]) > 0, "the assignment carries the session id")
|
|
|
+ conn = db()
|
|
|
+ race_sessions = conn.execute(
|
|
|
+ "select count(*) from sessions where task_id = ?", (race_task,)).fetchone()[0]
|
|
|
+ conn.close()
|
|
|
+ check(race_sessions == 1,
|
|
|
+ "exactly one session row — the transaction serialised the claims (%d)"
|
|
|
+ % race_sessions)
|
|
|
+ check(wait_task_done(auth, race_task, timeout=60) is not None,
|
|
|
+ "the raced task completes on the winning daemon")
|
|
|
+ for daemon in (fake, fake2, fake3):
|
|
|
+ daemon.ask_on_register = True
|
|
|
+ daemon.ask_on_wake = True
|
|
|
+ daemon.ask_on_terminal = True
|
|
|
+ daemon.ask_on_idle_status = True
|
|
|
+ # Re-establish sleeping explicitly: the winner's post-race
|
|
|
+ # re-ask timing is not deterministic, and the affinity section
|
|
|
+ # below needs every racer provably asleep. Any stray offerable
|
|
|
+ # drains into these asks first.
|
|
|
+ def all_asleep():
|
|
|
+ return all(daemon.ask_now()[0] == 204 for daemon in (fake, fake2, fake3))
|
|
|
+ check(wait_for(lambda: all_asleep() or False, timeout=30,
|
|
|
+ what="racers re-sleep after the race") is not None,
|
|
|
+ "all racers re-sleep after the race")
|
|
|
+
|
|
|
+ print("== phase 1: workspace affinity pins parked tasks ==")
|
|
|
+ # A suspended parent parked on racer-two may be resumed ONLY by
|
|
|
+ # racer-two: a tag-capable non-affine saddle asking gets 204, and
|
|
|
+ # the wake targets the affine saddle. Only racer-two carries the
|
|
|
+ # tag at dispatch time (the race's winner is nondeterministic, so
|
|
|
+ # the park must not depend on which racers are free); racer-three
|
|
|
+ # gains it afterwards to prove the pinning is by workspace, not
|
|
|
+ # tags.
|
|
|
saddles_state = slot_public(auth, "/saddles", "saddles")
|
|
|
- act(auth, saddles_state["register"], {"name": "dead-saddle",
|
|
|
- "host": "127.0.0.1:1"})
|
|
|
- dead_row = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
|
|
|
- if s["name"] == "dead-saddle"][0]
|
|
|
- act(auth, dead_row["attach_capability"], {"capability_id": str(gpu_option)})
|
|
|
+ act(auth, saddles_state["create_capability"], {"name": "affinity-tag"})
|
|
|
+ affinity_cap = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")["capabilities"]
|
|
|
+ if c["name"] == "affinity-tag"][0])
|
|
|
+ racer2_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
|
|
|
+ if s["name"] == "racer-two"][0]
|
|
|
+ racer3_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
|
|
|
+ if s["name"] == "racer-three"][0]
|
|
|
+ act(auth, racer2_card["attach_capability"], {"capability_id": str(affinity_cap)})
|
|
|
+
|
|
|
+ fake2.suspend_on_title = "Affinity parent"
|
|
|
act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
|
|
|
"project_id": "1",
|
|
|
- "title": "Dead saddle task",
|
|
|
- "capability_id": [str(gpu_option)],
|
|
|
+ "title": "Affinity parent",
|
|
|
+ "priority": "2",
|
|
|
+ "capability_id": [str(affinity_cap)],
|
|
|
})
|
|
|
- dead_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
|
|
|
- if t["title"] == "Dead saddle task"][0])
|
|
|
- check(wait_for(lambda: any("assignment push failed" in c["body"]
|
|
|
- for c in task_state(auth, dead_task).get("comments", []))
|
|
|
- or False, timeout=60, what="push-failure comment") is not None,
|
|
|
- "failed push writes the reason on the task")
|
|
|
- check(task_state(auth, dead_task)["status"] == "queued",
|
|
|
- "push-failed task returns to the queue")
|
|
|
- dead_rows = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
|
|
|
- if s["name"] == "dead-saddle"]
|
|
|
- check(dead_rows[0]["status"] == "offline",
|
|
|
- "the failing saddle goes offline")
|
|
|
- act(auth, dead_rows[0]["unregister"], {})
|
|
|
+ affinity_parent = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
|
|
|
+ if t["title"] == "Affinity parent"][0])
|
|
|
+ check(wait_for(lambda: task_state(auth, affinity_parent).get("status") == "suspended"
|
|
|
+ or False, timeout=60, what="affinity parent parks") is not None,
|
|
|
+ "affinity parent parks (suspended) on the tag-carrying racer")
|
|
|
conn = db()
|
|
|
- conn.execute("delete from task_capabilities where capability_id = ?"
|
|
|
- " and task_id = ?", (gpu_option, dead_task))
|
|
|
+ affinity_session = conn.execute(
|
|
|
+ "select saddle_id from sessions where task_id = ? and ended_at is null",
|
|
|
+ (affinity_parent,)).fetchone()[0]
|
|
|
+ conn.close()
|
|
|
+ check(affinity_session == int(racer2_card["id"]),
|
|
|
+ "the parked workspace sits on racer-two (the only tag-carrier)")
|
|
|
+
|
|
|
+ # Now racer-three carries the tag too — the pinning is by
|
|
|
+ # workspace, not tags.
|
|
|
+
|
|
|
+ # A terminal subtask makes the parked task resume-offerable —
|
|
|
+ # affine to racer-two only. The child's completion is stamped
|
|
|
+ # slightly in the future: the poller may not have observed (and
|
|
|
+ # stamped) the parent's suspension yet, and the resume trigger
|
|
|
+ # needs completed_at > suspended_at. Freeze racer-two's ask
|
|
|
+ # triggers so the manual ask below is the only contender (its
|
|
|
+ # park-time idle-status ask would otherwise race this claim).
|
|
|
+ conn = db()
|
|
|
+ conn.execute("insert into tasks (project_id, title, status, priority, parent_task_id,"
|
|
|
+ " summary, completed_at, created_at) values (1, 'Affinity child', 'done',"
|
|
|
+ " 5, ?, 'affinity child finished', strftime('%s','now','+15 seconds'),"
|
|
|
+ " strftime('%s','now'))", (affinity_parent,))
|
|
|
+ conn.commit()
|
|
|
+ conn.close()
|
|
|
+ fake2.ask_on_wake = False
|
|
|
+ fake2.ask_on_idle_status = False
|
|
|
+ time.sleep(1.2) # any in-flight park-time trigger ask lands (204)
|
|
|
+ act(auth, racer3_card["attach_capability"], {"capability_id": str(affinity_cap)})
|
|
|
+ status, _ = fake3.ask_now()
|
|
|
+ check(status == 204,
|
|
|
+ "a tag-capable non-affine saddle is refused the parked task (204)")
|
|
|
+
|
|
|
+ wakes_before_2 = len(fake2.wake_log)
|
|
|
+ check(wait_for(lambda: len(fake2.wake_log) > wakes_before_2 or False,
|
|
|
+ timeout=30, what="affine wake") is not None,
|
|
|
+ "the dispatcher's wake targets the affine saddle")
|
|
|
+ status, resume_payload = fake2.ask_now()
|
|
|
+ fake2.ask_on_wake = True # restored whatever the claim returned
|
|
|
+ fake2.ask_on_idle_status = True
|
|
|
+ check(status == 200 and int(resume_payload["task"]["id"]) == affinity_parent,
|
|
|
+ "the affine saddle's ask claims the parked task")
|
|
|
+ check("Affinity child" in (resume_payload.get("resume_notice") or "")
|
|
|
+ and "affinity child finished" in (resume_payload.get("resume_notice") or ""),
|
|
|
+ "the resume assignment carries the subtask report notice")
|
|
|
+ check(wait_task_done(auth, affinity_parent, timeout=60) is not None,
|
|
|
+ "the resumed parent completes")
|
|
|
+
|
|
|
+ print("== phase 1: lost wake -> offline + re-wake of the next capable ==")
|
|
|
+ # racer-three sleeps through wakes (a daemon that died after its
|
|
|
+ # last 204): the guard window (SADDLE_LOST_WAKE_S=2) passes with
|
|
|
+ # no ask, the saddle goes offline, and the work moves to the
|
|
|
+ # next capable sleeping saddle.
|
|
|
+ fake3.ask_on_wake = False
|
|
|
+ saddles_state = slot_public(auth, "/saddles", "saddles")
|
|
|
+ act(auth, saddles_state["create_capability"], {"name": "lostwake"})
|
|
|
+ lostwake_cap = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")["capabilities"]
|
|
|
+ if c["name"] == "lostwake"][0])
|
|
|
+ racer3_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
|
|
|
+ if s["name"] == "racer-three"][0]
|
|
|
+ act(auth, racer3_card["attach_capability"], {"capability_id": str(lostwake_cap)})
|
|
|
+ # Racer-three must be asleep to be woken: park it with a manual
|
|
|
+ # 204 ask (nothing offerable for it — the task below is not
|
|
|
+ # created yet).
|
|
|
+ status, _ = fake3.ask_now()
|
|
|
+ check(status == 204, "racer-three sleeps before the lost-wake run")
|
|
|
+
|
|
|
+ act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
|
|
|
+ "project_id": "1",
|
|
|
+ "title": "Lost wake task",
|
|
|
+ "priority": "2",
|
|
|
+ "capability_id": [str(lostwake_cap)],
|
|
|
+ })
|
|
|
+ lost_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
|
|
|
+ if t["title"] == "Lost wake task"][0])
|
|
|
+ check(wait_for(lambda: len(fake3.wake_log) > 0 or False,
|
|
|
+ timeout=30, what="wake sent to racer-three") is not None,
|
|
|
+ "the only capable sleeping saddle is woken")
|
|
|
+ check(wait_for(lambda: [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
|
|
|
+ if s["name"] == "racer-three"][0]["status"] == "offline"
|
|
|
+ or False, timeout=30, what="lost-wake -> offline") is not None,
|
|
|
+ "no ask after the wake: the saddle goes offline (lost-wake guard)")
|
|
|
+ check(task_state(auth, lost_task)["status"] == "queued",
|
|
|
+ "the task is untouched while its wake was pending")
|
|
|
+
|
|
|
+ # Re-wake the next capable: the healthy saddle gains the tag and
|
|
|
+ # picks the work up via its own wake -> ask -> allocate roundtrip.
|
|
|
+ fake_card = saddle_card("fake-saddle")
|
|
|
+ act(auth, fake_card["attach_capability"], {"capability_id": str(lostwake_cap)})
|
|
|
+ wakes_before = len(fake.wake_log)
|
|
|
+ check(wait_for(lambda: len(fake.wake_log) > wakes_before or False,
|
|
|
+ timeout=30, what="re-wake of the next capable") is not None,
|
|
|
+ "the next capable sleeping saddle is woken for the task")
|
|
|
+ check(wait_task_done(auth, lost_task, timeout=60) is not None,
|
|
|
+ "wake -> ask -> allocate: the task completes on the healthy saddle")
|
|
|
+ check(wait_for(lambda: any(int(p.get("task", {}).get("id") or 0) == lost_task
|
|
|
+ for p in fake.start_payloads) or False,
|
|
|
+ timeout=10, what="lost task assignment captured") is not None,
|
|
|
+ "the re-woken saddle's ask produced the assignment")
|
|
|
+ fake3.ask_on_wake = True
|
|
|
+
|
|
|
+ # The racers leave the dispatch pool: the remaining sections
|
|
|
+ # (rollback resume, R5-R11) predate multi-daemon routing and
|
|
|
+ # assert against the primary fake's payloads. racer-three is
|
|
|
+ # already offline (the lost-wake guard); racer-two joins it.
|
|
|
+ conn = db()
|
|
|
+ conn.execute("update saddles set status = 'offline' where name = 'racer-two'")
|
|
|
conn.commit()
|
|
|
conn.close()
|
|
|
- check(wait_task_done(auth, dead_task, timeout=60) is not None,
|
|
|
- "the task recovers on the healthy saddle")
|
|
|
|
|
|
# Tasks caught working on a saddle that went offline requeue with a
|
|
|
# bot comment each tick.
|
|
|
- fake.suspend_after_start = False
|
|
|
act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
|
|
|
"project_id": "1",
|
|
|
"title": "Offline saddle task",
|
|
|
@@ -1916,7 +2363,7 @@ def main():
|
|
|
|
|
|
|
|
|
print("== R6 F2: refresh must not corrupt parked bookkeeping ==")
|
|
|
- fake.suspend_after_start = True
|
|
|
+ fake.suspend_on_title = "Parked refresh parent"
|
|
|
act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
|
|
|
"project_id": "1",
|
|
|
"title": "Parked refresh parent",
|
|
|
@@ -1945,7 +2392,6 @@ def main():
|
|
|
|
|
|
|
|
|
print("== R6 F3: raw-log probe retries instead of deciding once ==")
|
|
|
- fake.suspend_after_start = False
|
|
|
fake.fail_log_count = 1 # the seed-time probe fails exactly once
|
|
|
act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
|
|
|
"project_id": "1",
|
|
|
@@ -2055,95 +2501,54 @@ def main():
|
|
|
"queued work dispatches once the fence clears")
|
|
|
|
|
|
|
|
|
- print("== R7 B2: 409-aware dispatch ==")
|
|
|
- # Zombie: the daemon (per the fake) still runs an old completed task;
|
|
|
- # one 409, then abort + retry recovers without churning sessions.
|
|
|
- fake.state = "idle"
|
|
|
- fake.zombie_409 = 1
|
|
|
- act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
|
|
|
- "project_id": "1",
|
|
|
- "title": "Zombie recovery",
|
|
|
- "priority": "2",
|
|
|
- })
|
|
|
- zombie_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
|
|
|
- if t["title"] == "Zombie recovery"][0])
|
|
|
- check(wait_task_done(auth, zombie_task, timeout=90) is not None,
|
|
|
- "zombie conflict recovers and the task completes")
|
|
|
- detail = task_state(auth, zombie_task)
|
|
|
- check(any("recovered zombie session" in c["body"] for c in detail.get("comments", [])),
|
|
|
- "the zombie recovery is narrated")
|
|
|
- conn = db()
|
|
|
- zombie_sessions = conn.execute(
|
|
|
- "select count(*) from sessions where task_id = ?", (zombie_task,)).fetchone()[0]
|
|
|
- conn.close()
|
|
|
- check(zombie_sessions == 1,
|
|
|
- "no session churn: one row across the failed and retried push (%d)"
|
|
|
- % zombie_sessions)
|
|
|
+ print("== phase 1: rollback resume path (pre-flip daemon) ==")
|
|
|
+ # Phase 1 keeps the /answer + /start resume machinery for daemons
|
|
|
+ # that never ask (the rollback route; the 409-aware /start push
|
|
|
+ # machinery itself is rollback-only now that dispatch flows
|
|
|
+ # through asks). A pre-flip daemon's parked suspended parent is
|
|
|
+ # resumed by a /start push carrying the subtask report.
|
|
|
+ fake4.ask_on_register = False
|
|
|
+ fake4.ask_on_wake = False
|
|
|
+ fake4.ask_on_terminal = False
|
|
|
+ fake4.ask_on_idle_status = False
|
|
|
+ fake4.serve_register = False # a pre-flip daemon has no /register
|
|
|
+ registers_before = len(fake4.register_log)
|
|
|
+ saddles_state = slot_public(auth, "/saddles", "saddles")
|
|
|
+ act(auth, saddles_state["register"], {"name": "preflip-daemon",
|
|
|
+ "host": "127.0.0.1:%d" % fake4.port})
|
|
|
+ preflip_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
|
|
|
+ if s["name"] == "preflip-daemon"][0]
|
|
|
+ preflip_id = int(preflip_card["id"])
|
|
|
+ check(len(fake4.register_log) == registers_before,
|
|
|
+ "the pre-flip daemon 404s /register (registration tolerated)")
|
|
|
|
|
|
- # Live conflict: the daemon claims to run a task the web still sees
|
|
|
- # live — fence + alert, never abort live work.
|
|
|
- act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
|
|
|
- "project_id": "1",
|
|
|
- "title": "Conflict live task",
|
|
|
- "priority": "2",
|
|
|
- })
|
|
|
- conflict_live = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
|
|
|
- if t["title"] == "Conflict live task"][0])
|
|
|
- check(wait_for(lambda: task_state(auth, conflict_live).get("status") == "working"
|
|
|
- or False, timeout=60, what="conflict task working") is not None,
|
|
|
- "live task reaches working")
|
|
|
- # Hold the daemon mid-working and desynchronise the web's view:
|
|
|
- # double-booking means the web believes the saddle free while the
|
|
|
- # daemon still claims the live task.
|
|
|
- fake.hold_state = True
|
|
|
- fake.conflict_running = conflict_live
|
|
|
conn = db()
|
|
|
- conn.execute("update saddles set status = 'idle', current_task_id = 0,"
|
|
|
- " allocated_at = null where name = 'fake-saddle'")
|
|
|
+ conn.execute("insert into tasks (project_id, title, status, priority, suspended_at,"
|
|
|
+ " created_at) values (1, 'Preflip parked parent', 'suspended', 2,"
|
|
|
+ " strftime('%s','now','-60 seconds'), strftime('%s','now','-120 seconds'))")
|
|
|
+ preflip_parent = conn.execute("select max(id) from tasks").fetchone()[0]
|
|
|
+ conn.execute("insert into sessions (task_id, saddle_id, started_at)"
|
|
|
+ " values (?, ?, strftime('%s','now','-90 seconds'))",
|
|
|
+ (preflip_parent, preflip_id))
|
|
|
+ conn.execute("insert into tasks (project_id, title, status, priority, parent_task_id,"
|
|
|
+ " summary, completed_at, created_at) values (1, 'Preflip child', 'done',"
|
|
|
+ " 5, ?, 'child of the pre-flip era',"
|
|
|
+ " strftime('%s','now','-30 seconds'),"
|
|
|
+ " strftime('%s','now','-100 seconds'))", (preflip_parent,))
|
|
|
conn.commit()
|
|
|
conn.close()
|
|
|
- act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
|
|
|
- "project_id": "1",
|
|
|
- "title": "Conflict newcomer",
|
|
|
- "priority": "3",
|
|
|
- })
|
|
|
- conflict_new = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
|
|
|
- if t["title"] == "Conflict newcomer"][0])
|
|
|
- check(wait_for(lambda: saddle_card("fake-saddle")["status"] == "recovering" or False,
|
|
|
- timeout=30, what="double-booked saddle fenced") is not None,
|
|
|
- "double-booked saddle is fenced")
|
|
|
- check(any("double-booked" in c["body"]
|
|
|
- for c in task_state(auth, conflict_live).get("comments", [])),
|
|
|
- "the live task gets the alert comment")
|
|
|
- check(any("dispatch refused" in c["body"]
|
|
|
- for c in task_state(auth, conflict_new).get("comments", [])),
|
|
|
- "the refused task gets its comment")
|
|
|
- conn = db()
|
|
|
- new_sessions = conn.execute(
|
|
|
- "select outcome from sessions where task_id = ?", (conflict_new,)).fetchall()
|
|
|
- conn.close()
|
|
|
- check(len(new_sessions) == 1 and new_sessions[0][0] == "push_failed",
|
|
|
- "failed push marks the session push_failed (no churn)")
|
|
|
- check(conflict_live not in fake.abort_log,
|
|
|
- "the live task is never aborted")
|
|
|
- fake.conflict_running = 0
|
|
|
- fake.hold_state = False
|
|
|
- fake.state = "idle"
|
|
|
- act(auth, task_state(auth, conflict_live)["complete"],
|
|
|
- {"summary": "finished before the conflict"})
|
|
|
- check(wait_for(lambda: task_state(auth, conflict_live).get("status") == "done"
|
|
|
- or False, timeout=30, what="live task completed") is not None,
|
|
|
- "the live task completes without an abort")
|
|
|
- check(wait_for(lambda: saddle_card("fake-saddle")["status"] == "idle" or False,
|
|
|
- timeout=40, what="fence clears after conflict") is not None,
|
|
|
- "fence clears once the daemon goes idle")
|
|
|
- conn = db()
|
|
|
- after_clear = conn.execute(
|
|
|
- "select count(*) from sessions where task_id = ?", (conflict_new,)).fetchone()[0]
|
|
|
- conn.close()
|
|
|
- check(after_clear == 1, "still one session row for the refused task")
|
|
|
- check(wait_task_done(auth, conflict_new, timeout=90) is not None,
|
|
|
- "the refused task dispatches after the fence clears")
|
|
|
+
|
|
|
+ pushes_before = len(fake4.start_payloads)
|
|
|
+ check(wait_for(lambda: len(fake4.start_payloads) > pushes_before or False,
|
|
|
+ timeout=45, what="pre-flip /start resume push") is not None,
|
|
|
+ "the pre-flip daemon's parked parent resumes via POST /start")
|
|
|
+ preflip_push = [p for p in fake4.start_payloads
|
|
|
+ if int(p.get("task", {}).get("id") or 0) == preflip_parent][-1]
|
|
|
+ check("Preflip child" in (preflip_push.get("resume_notice") or "")
|
|
|
+ and "child of the pre-flip era" in (preflip_push.get("resume_notice") or ""),
|
|
|
+ "the rollback /start resume carries the subtask report notice")
|
|
|
+ check(wait_task_done(auth, preflip_parent, timeout=60) is not None,
|
|
|
+ "the pre-flip parent completes through the rollback path")
|
|
|
|
|
|
|
|
|
print("== R7 B3: wedged daemon, health panel, raw log, vanishing daemon ==")
|
|
|
@@ -2534,6 +2939,58 @@ def main():
|
|
|
check(row.get("unreachable") is True,
|
|
|
"queue row shows the unreachable state in the window")
|
|
|
|
|
|
+ print("== phase 1: startup migrates the per-saddle token ==")
|
|
|
+ # Saddles registered before the flip hold legacy api_tokens rows.
|
|
|
+ # On startup every saddle row gets its canonical token: the
|
|
|
+ # legacy value it already holds (daemons keep their credential)
|
|
|
+ # or a freshly minted one.
|
|
|
+ conn = db()
|
|
|
+ conn.execute("insert into api_tokens (saddle_id, token, created_at)"
|
|
|
+ " values (?, 'saddle-legacy-copied-token', strftime('%s','now'))",
|
|
|
+ (int(racer_cards[fake2]["id"]),))
|
|
|
+ conn.execute("update saddles set token = null where name = 'racer-two'")
|
|
|
+ conn.execute("insert into saddles (name, host, status, current_task_id)"
|
|
|
+ " values ('tokenless-orphan', '127.0.0.1:1', 'offline', 0)")
|
|
|
+ conn.commit()
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ app.terminate()
|
|
|
+ app.wait(timeout=10)
|
|
|
+ app = start_app()
|
|
|
+ auth = login("admin", "adminpass123")
|
|
|
+
|
|
|
+ def saddle_row(name):
|
|
|
+ conn = db()
|
|
|
+ try:
|
|
|
+ return conn.execute("select id, token from saddles where name = ?",
|
|
|
+ (name,)).fetchone()
|
|
|
+ finally:
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ check(wait_for(lambda: saddle_row("tokenless-orphan") is not None
|
|
|
+ and (saddle_row("tokenless-orphan")[1] or "").startswith("saddle-")
|
|
|
+ or False, timeout=20, what="tokenless row minted") is not None,
|
|
|
+ "startup mints a token for a saddle row without one")
|
|
|
+ check(wait_for(lambda: saddle_row("racer-two") is not None
|
|
|
+ and saddle_row("racer-two")[1] == "saddle-legacy-copied-token"
|
|
|
+ or False, timeout=20, what="legacy token copied") is not None,
|
|
|
+ "startup copies the legacy api_tokens value onto the row")
|
|
|
+ fake2.token = "saddle-legacy-copied-token" # the daemon asks with it
|
|
|
+ status, _ = saddle_api("POST", "/api/saddle/next-task", "saddle-legacy-copied-token")
|
|
|
+ check(status in (200, 204),
|
|
|
+ "the legacy token still authenticates (copied, not replaced)")
|
|
|
+ conn = db()
|
|
|
+ conn.execute("delete from saddles where name = 'tokenless-orphan'")
|
|
|
+ conn.commit()
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ # The restart wiped the in-memory ask bookkeeping; the fleet's
|
|
|
+ # daemons re-establish sleeping with a fresh ask each (the real
|
|
|
+ # daemons' idle-awake ask / 15-minute safety self-ask).
|
|
|
+ for daemon in (fake2, fake3):
|
|
|
+ daemon.ask_now()
|
|
|
+ check(True, "fleet re-asks after the restart (sleeping re-established)")
|
|
|
+
|
|
|
|
|
|
finally:
|
|
|
# Exit-code diagnostic: distinguishes a crashed app (negative
|
