feat(observatory): expose attempt-bound recorded progress
This commit is contained in:
DCCONSTRUCTIONS
@@ -0,0 +1,339 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sqlite3
|
||||
import sys
|
||||
import threading
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
from test_observatory_recorded_jobs import _queue, _running_job
|
||||
from test_observatory_worker_api import WORKER_HEADERS, _claim_request, _services
|
||||
|
||||
from k1link.observatory.m49_timing_progress import M49TimingProgress
|
||||
from k1link.observatory.recorded_jobs import (
|
||||
ObservatoryRecordedQueueConflictError,
|
||||
ObservatoryRecordedQueueStaleClaimError,
|
||||
)
|
||||
from k1link.observatory.recorded_progress import (
|
||||
RecordedProgress,
|
||||
RecordedProgressTracker,
|
||||
observe_recorded_execution,
|
||||
report_recorded_progress,
|
||||
)
|
||||
|
||||
|
||||
def _progress(**changes: object) -> RecordedProgress:
|
||||
return RecordedProgress.model_validate(
|
||||
{
|
||||
"claim_generation": 1,
|
||||
"sequence": 1,
|
||||
"phase_index": 1,
|
||||
"phase": "computing",
|
||||
"unit": "frames",
|
||||
"completed": 3,
|
||||
"total": 10,
|
||||
"elapsed_seconds": 3.0,
|
||||
"phase_elapsed_seconds": 2.0,
|
||||
**changes,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"change",
|
||||
[
|
||||
{"completed": True},
|
||||
{"completed": 11},
|
||||
{"total": 0},
|
||||
{"phase": "done"},
|
||||
{"elapsed_seconds": float("nan")},
|
||||
{"phase_elapsed_seconds": 4.0},
|
||||
{"sequence": 2**53},
|
||||
{"path": "/secret"},
|
||||
],
|
||||
)
|
||||
def test_progress_rejects_unbounded_or_invented_values(change: dict[str, object]) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
_progress(**change)
|
||||
|
||||
|
||||
def test_progress_is_one_durable_snapshot_not_job_identity_or_lease(tmp_path: Path) -> None:
|
||||
queue = _queue(tmp_path)
|
||||
job, claim = _running_job(queue)
|
||||
original = queue.get(job.job_id).as_dict()
|
||||
assert queue.progress(job.job_id)["progress"] is None
|
||||
for sequence in range(1, 101):
|
||||
queue.report_progress(
|
||||
job.job_id,
|
||||
claim_token=claim.claim_token,
|
||||
claimant_id="recorded-worker",
|
||||
progress=_progress(sequence=sequence),
|
||||
)
|
||||
assert queue.get(job.job_id).as_dict() == original
|
||||
reopened = _queue(tmp_path)
|
||||
view = reopened.progress(job.job_id)
|
||||
assert view["definition_sha256"] == job.definition_sha256
|
||||
assert view["claim_generation"] == 1
|
||||
assert view["age_seconds"] == 0.0
|
||||
assert view["progress"]["sequence"] == 100
|
||||
with sqlite3.connect(queue.database_path) as connection:
|
||||
assert (
|
||||
connection.execute("SELECT COUNT(*) FROM observatory_recorded_progress").fetchone()[0]
|
||||
== 1
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"change",
|
||||
[
|
||||
{"sequence": 1, "completed": 4},
|
||||
{"sequence": 2, "completed": 2},
|
||||
{"sequence": 2, "total": 12},
|
||||
{"sequence": 2, "phase": "result-transfer"},
|
||||
{"sequence": 2, "elapsed_seconds": 2.0},
|
||||
],
|
||||
)
|
||||
def test_progress_retry_is_idempotent_and_regressions_rejected(tmp_path: Path, change) -> None:
|
||||
queue = _queue(tmp_path)
|
||||
job, claim = _running_job(queue)
|
||||
|
||||
def send(value):
|
||||
queue.report_progress(
|
||||
job.job_id, claim_token=claim.claim_token, claimant_id="recorded-worker", progress=value
|
||||
)
|
||||
|
||||
send(_progress())
|
||||
send(_progress())
|
||||
with pytest.raises(ObservatoryRecordedQueueConflictError):
|
||||
send(_progress(**change))
|
||||
send(_progress(sequence=3, phase_index=2, phase="result-assembly", completed=0, total=None))
|
||||
|
||||
|
||||
def test_progress_old_generation_cannot_write_or_project_as_current(tmp_path: Path) -> None:
|
||||
queue = _queue(tmp_path)
|
||||
job, claim = _running_job(queue)
|
||||
with pytest.raises(ObservatoryRecordedQueueStaleClaimError):
|
||||
queue.report_progress(
|
||||
job.job_id,
|
||||
claim_token=claim.claim_token,
|
||||
claimant_id="recorded-worker",
|
||||
progress=_progress(claim_generation=2),
|
||||
)
|
||||
with pytest.raises(ObservatoryRecordedQueueStaleClaimError):
|
||||
queue.report_progress(
|
||||
job.job_id,
|
||||
claim_token=claim.claim_token,
|
||||
claimant_id="another-worker",
|
||||
progress=_progress(),
|
||||
)
|
||||
# Synthetic old attempt retained on disk is not the new attempt's progress.
|
||||
with sqlite3.connect(queue.database_path) as connection:
|
||||
connection.execute(
|
||||
"INSERT INTO observatory_recorded_progress VALUES (?, ?, ?)",
|
||||
(job.job_id, _progress(claim_generation=2).model_dump_json(), job.created_at_utc),
|
||||
)
|
||||
assert queue.progress(job.job_id)["progress"] is None
|
||||
|
||||
|
||||
def test_progress_tracker_preserves_total_and_separates_phases() -> None:
|
||||
tracker = RecordedProgressTracker(2, clock=lambda: 10.0)
|
||||
tracker.update("computing", 0, 10)
|
||||
first = tracker.sample()
|
||||
tracker.update("computing", 2)
|
||||
second = tracker.sample()
|
||||
assert second.total == 10 and second.follows(first)
|
||||
tracker.update("result-assembly", unit="steps")
|
||||
third = tracker.sample()
|
||||
assert third.completed == 0 and third.total is None and third.follows(second)
|
||||
|
||||
|
||||
def test_m49_invoker_preserves_arguments_and_observes_real_output(tmp_path, monkeypatch):
|
||||
from k1link.observatory import m49_portable_executor as executor
|
||||
|
||||
script = tmp_path / "synthetic.py"
|
||||
script.write_text(
|
||||
"import os, sys\n"
|
||||
"from pathlib import Path\n"
|
||||
"assert sys.argv[1] == 'schedule.tsv'\n"
|
||||
"assert sys.argv[2] == 'outputs'\n"
|
||||
"assert os.environ['LANG'] == 'C' and os.environ['TZ'] == 'UTC'\n"
|
||||
"Path(sys.argv[3]).write_text('timeline_frame_index\\tsource_frame_index\\tother\\n'"
|
||||
" + '0\\t0\\t0\\t1\\t0\\t12\\t10\\t2\\t1.2\\t1.3\\n')\n"
|
||||
"print('synthetic runner completed')\n"
|
||||
)
|
||||
counters = []
|
||||
monkeypatch.setattr(
|
||||
executor, "report_recorded_progress", lambda phase, count: counters.append(count)
|
||||
)
|
||||
executor._invoke_exact_runner(
|
||||
binary=Path(sys.executable),
|
||||
sequence=script,
|
||||
schedule=Path("schedule.tsv"),
|
||||
output=Path("outputs"),
|
||||
timing=tmp_path / "timing.tsv",
|
||||
workspace=tmp_path,
|
||||
timeout_seconds=5,
|
||||
)
|
||||
assert counters[-1] == 1
|
||||
assert (tmp_path / "runner.stdout.log").read_text().strip() == "synthetic runner completed"
|
||||
assert (tmp_path / "runner.stderr.log").read_bytes() == b""
|
||||
|
||||
|
||||
def test_m49_invoker_timeout_reaps_only_its_child(tmp_path):
|
||||
import os
|
||||
|
||||
from k1link.observatory.m49_portable_executor import (
|
||||
M49PortableExecutorError,
|
||||
_invoke_exact_runner,
|
||||
)
|
||||
|
||||
script = tmp_path / "synthetic.py"
|
||||
script.write_text(
|
||||
"import os, time\nfrom pathlib import Path\n"
|
||||
"Path('synthetic.pid').write_text(str(os.getpid()))\ntime.sleep(30)\n"
|
||||
)
|
||||
with pytest.raises(M49PortableExecutorError, match="invocation failed"):
|
||||
_invoke_exact_runner(
|
||||
binary=Path(sys.executable),
|
||||
sequence=script,
|
||||
schedule=tmp_path / "schedule.tsv",
|
||||
output=tmp_path / "outputs",
|
||||
timing=tmp_path / "timing.tsv",
|
||||
workspace=tmp_path,
|
||||
timeout_seconds=1,
|
||||
)
|
||||
pid = int((tmp_path / "synthetic.pid").read_text())
|
||||
with pytest.raises(ProcessLookupError):
|
||||
os.kill(pid, 0)
|
||||
|
||||
|
||||
def test_progress_sender_failure_does_not_escape_or_leave_a_thread() -> None:
|
||||
sampled = threading.Event()
|
||||
|
||||
def unavailable(snapshot):
|
||||
assert isinstance(snapshot, RecordedProgress)
|
||||
sampled.set()
|
||||
raise RuntimeError("synthetic observer connection loss")
|
||||
|
||||
with observe_recorded_execution(1, unavailable, interval_seconds=0.01) as tracker:
|
||||
report_recorded_progress("computing", 5, 8)
|
||||
assert sampled.wait(1)
|
||||
assert tracker.sample().completed == 5
|
||||
assert not any(thread.name == "observatory-progress" for thread in threading.enumerate())
|
||||
report_recorded_progress("computing", 1, 2) # no leaked execution context
|
||||
assert tracker.sample().completed == 5
|
||||
|
||||
|
||||
def test_worker_keeps_compute_and_lease_when_progress_transport_fails(tmp_path: Path) -> None:
|
||||
from test_observatory_worker_agent import (
|
||||
BlockingExecutor,
|
||||
FakeTransport,
|
||||
_enqueue,
|
||||
_heartbeat_agent,
|
||||
)
|
||||
from test_observatory_worker_agent import (
|
||||
_queue as worker_queue,
|
||||
)
|
||||
|
||||
queue = worker_queue(tmp_path)
|
||||
job_id = _enqueue(queue)
|
||||
attempted = threading.Event()
|
||||
entered, release = threading.Event(), threading.Event()
|
||||
|
||||
class ObserverFailureTransport(FakeTransport):
|
||||
def report_progress(self, **_kwargs):
|
||||
attempted.set()
|
||||
raise RuntimeError("observer unavailable")
|
||||
|
||||
agent = _heartbeat_agent(
|
||||
ObserverFailureTransport(queue),
|
||||
BlockingExecutor(entered=entered, release=release),
|
||||
)
|
||||
reports = []
|
||||
thread = threading.Thread(target=lambda: reports.append(agent.run_once()))
|
||||
thread.start()
|
||||
assert entered.wait(2) and attempted.wait(2)
|
||||
release.set()
|
||||
thread.join(2)
|
||||
assert not thread.is_alive()
|
||||
assert reports[0].state == "succeeded"
|
||||
assert queue.get(job_id).state == "succeeded"
|
||||
|
||||
|
||||
def test_http_progress_uses_active_claim_and_short_bodyless_request(tmp_path: Path) -> None:
|
||||
import httpx
|
||||
from test_observatory_worker_http_transport import (
|
||||
BEARER_TOKEN,
|
||||
CLAIM_TOKEN,
|
||||
JOB_ID,
|
||||
_cache_claim,
|
||||
_claim_response,
|
||||
)
|
||||
|
||||
from k1link.observatory.worker_http_transport import ObservatoryWorkerHttpGateway
|
||||
|
||||
requests = []
|
||||
|
||||
def handle(request):
|
||||
if request.url.path.endswith("/claims"):
|
||||
return _claim_response()
|
||||
requests.append(request)
|
||||
assert request.url.path.endswith("/progress")
|
||||
assert request.extensions["timeout"]["read"] == 2.0
|
||||
assert json.loads(request.content)["claim_token"] == CLAIM_TOKEN
|
||||
return httpx.Response(204)
|
||||
|
||||
with ObservatoryWorkerHttpGateway(
|
||||
base_url="http://127.0.0.1:8000",
|
||||
bearer_token=BEARER_TOKEN,
|
||||
work_root=tmp_path,
|
||||
transport=httpx.MockTransport(handle),
|
||||
) as gateway:
|
||||
_cache_claim(gateway)
|
||||
gateway.report_progress(job_id=JOB_ID, claim_token=CLAIM_TOKEN, progress=_progress())
|
||||
assert len(requests) == 1
|
||||
|
||||
|
||||
def test_tgs_progress_counts_only_complete_ordered_flushed_rows(tmp_path: Path) -> None:
|
||||
path = tmp_path / "timing.tsv"
|
||||
observer = M49TimingProgress(path)
|
||||
assert observer.poll() == 0
|
||||
path.write_bytes(
|
||||
b"timeline_frame_index\tsource_frame_index\trest\n0\t0\t0\t1\t0\t1\t1\t0\t1\t1\n1\t1"
|
||||
)
|
||||
assert observer.poll() == 1
|
||||
with path.open("ab") as stream:
|
||||
stream.write(b"\t1\t1\t1\t1\t1\t0\t1\t1\n")
|
||||
assert observer.poll() == 2
|
||||
with path.open("ab") as stream:
|
||||
stream.write(b"4\t4\t4\t1\t4\t1\t1\t0\t1\t1\n")
|
||||
assert observer.poll() == 2 and observer.invalid
|
||||
|
||||
|
||||
def test_progress_api_requires_auth_exact_claim_and_strict_payload(tmp_path: Path) -> None:
|
||||
from test_observatory_worker_api import _enqueue
|
||||
|
||||
client, queue = _services(tmp_path)
|
||||
job_id = _enqueue(queue)
|
||||
response = client.post(
|
||||
"/api/v1/worker/observatory/recorded-jobs/claims",
|
||||
headers=WORKER_HEADERS,
|
||||
json=_claim_request("progress-claim"),
|
||||
)
|
||||
claim = response.json()
|
||||
token = claim["claim_token"]
|
||||
queue.start(job_id, claim_token=token)
|
||||
path = f"/api/v1/worker/observatory/recorded-jobs/{job_id}/progress"
|
||||
body = {
|
||||
"schema_version": "missioncore.observatory-worker-progress-request/v1",
|
||||
"claim_token": token,
|
||||
"progress": _progress().model_dump(mode="json"),
|
||||
}
|
||||
assert client.post(path, json=body).status_code == 401
|
||||
assert (
|
||||
client.post(path, headers=WORKER_HEADERS, json={**body, "command": "no"}).status_code == 422
|
||||
)
|
||||
assert client.post(path, headers=WORKER_HEADERS, json=body).status_code == 204
|
||||
assert queue.progress(job_id)["progress"] == json.loads(_progress().model_dump_json())
|
||||
Reference in New Issue
Block a user