"""Append-only local queue and reload state for generic authenticated jobs."""
|
|
from __future__ import annotations
|
|
import hashlib
|
import json
|
import os
|
import secrets
|
from contextlib import contextmanager
|
from dataclasses import dataclass
|
from pathlib import Path
|
from typing import Any, Callable, Iterator
|
|
from .constants import (
|
COMPLETION_CLOSURE_REQUIRED,
|
EXTENSION_BUILD,
|
ERROR_CODE_RE,
|
HANDOFF_ID_RE,
|
JOB_ID_RE,
|
MESSAGE_ID_RE,
|
MAX_CLAIM_ATTEMPTS,
|
INTERNAL_ONLY_PAGE_PENDING_CODES,
|
LEGACY_MISSING_DIAGNOSTIC_REPLAY_LINES,
|
LEGACY_PAGE_METADATA_REPLAY_LINES,
|
LEGACY_PAGE_METADATA_REPLAY_PREFIX_BYTES,
|
LEGACY_PAGE_METADATA_REPLAY_PREFIX_SHA256,
|
LEGACY_PAGE_METADATA_REPLAY_STATE_SUFFIX,
|
PREPARELESS_REJECT_CODES,
|
QUEUE_LEASE_SECONDS,
|
QUEUE_SCHEMA_VERSION,
|
SUCCESSOR_QUEUE_SCHEMA_VERSION,
|
UPPER_SHA256_RE,
|
RELOAD_BACKOFF_SECONDS,
|
canonical_url,
|
stable_job_id,
|
stable_successor_job_id,
|
validate_bvid,
|
validate_creator_uid,
|
validate_prepareless_terminal,
|
validate_postprocess_terminal,
|
validate_runtime_diagnostic,
|
)
|
from .protocol import (
|
ProtocolError, encode_json, strict_json_loads, validate_job,
|
validate_media_complete_identity,
|
)
|
|
|
@dataclass(frozen=True)
|
class AuthorizedSuccessor:
|
creator_uid: str
|
bvid: str
|
predecessor_job_id: str
|
retry_generation: int
|
terminal_error_code: str
|
|
|
@dataclass(frozen=True)
|
class ReleaseApproval:
|
"""Immutable value emitted only by the producer's deployment trust gate."""
|
|
authorization_message_id: str
|
authorization_handoff_id: str
|
authorization_sha256: str
|
repair_review_result_message_id: str
|
repair_audit_id: str
|
repair_audit_bytes: int
|
repair_audit_sha256: str
|
successors: tuple[AuthorizedSuccessor, ...]
|
|
|
def _read_jsonl(path: Path, *, missing_ok: bool) -> list[dict[str, Any]]:
|
if not path.exists():
|
if missing_ok:
|
return []
|
raise ProtocolError("E_QUEUE")
|
if not path.is_file() or path.is_symlink():
|
raise ProtocolError("E_QUEUE")
|
payload = path.read_bytes()
|
if len(payload) > 16 * 1024 * 1024:
|
raise ProtocolError("E_QUEUE")
|
if payload and not payload.endswith(b"\n"):
|
raise ProtocolError("E_QUEUE_PARTIAL")
|
result: list[dict[str, Any]] = []
|
for line in payload.splitlines():
|
if not line or len(line) > 4096:
|
raise ProtocolError("E_QUEUE")
|
result.append(strict_json_loads(line))
|
return result
|
|
|
def _append_jsonl(path: Path, value: dict[str, Any]) -> None:
|
payload = encode_json(value, 4096) + b"\n"
|
with path.open("ab", buffering=0) as stream:
|
if stream.write(payload) != len(payload):
|
raise ProtocolError("E_QUEUE_WRITE")
|
stream.flush()
|
os.fsync(stream.fileno())
|
|
|
def _media_complete_durability_test_seam(_stage: str) -> None:
|
"""No-op production seam for crash-window durability counterexamples."""
|
|
return None
|
|
|
def _postprocess_recovery_claim_test_seam(_stage: str) -> None:
|
"""No-op production seam for recovery-claim crash counterexamples."""
|
|
return None
|
|
|
def _append_jsonl_batch(path: Path, values: list[dict[str, Any]]) -> None:
|
if not values:
|
return
|
payload = b"".join(encode_json(value, 4096) + b"\n" for value in values)
|
with path.open("ab", buffering=0) as stream:
|
if stream.write(payload) != len(payload):
|
raise ProtocolError("E_QUEUE_WRITE")
|
stream.flush()
|
os.fsync(stream.fileno())
|
|
|
def _append_bytes(path: Path, payload: bytes) -> None:
|
if not payload:
|
return
|
with path.open("ab", buffering=0) as stream:
|
if stream.write(payload) != len(payload):
|
raise ProtocolError("E_QUEUE_WRITE")
|
stream.flush()
|
os.fsync(stream.fileno())
|
|
|
def _canonical_line(value: dict[str, Any]) -> bytes:
|
return encode_json(value, 4096) + b"\n"
|
|
|
@dataclass(frozen=True)
|
class _QueueSnapshot:
|
jobs: dict[str, dict[str, Any]]
|
records: dict[str, dict[str, Any]]
|
children: dict[str, str]
|
blocks: tuple[tuple[str, str, str, tuple[str, ...], bytes], ...]
|
committed_bytes: bytes
|
pending_suffix: bytes
|
|
|
_LINEAGE_KEYS = {
|
"predecessor_job_id", "retry_generation", "predecessor_terminal_error_code",
|
"authorization_message_id", "authorization_handoff_id", "authorization_sha256",
|
"repair_review_result_message_id", "repair_audit_id", "repair_audit_bytes",
|
"repair_audit_sha256",
|
}
|
_SUCCESSOR_RECORD_KEYS = {
|
"schema", "record_type", "job_id", "bvid", "creator_uid",
|
"expected_duration_ms", "discovered_at_unix_ms", "published_at", "title", "lineage",
|
}
|
_BEGIN_KEYS = {
|
"schema", "record_type", "authorization_message_id", "authorization_handoff_id",
|
"authorization_sha256", "successor_count",
|
}
|
_COMMIT_KEYS = _BEGIN_KEYS | {"block_sha256"}
|
|
|
def _validate_authorization_identity(message_id: Any, handoff_id: Any, sha256: Any) -> None:
|
if (
|
not isinstance(message_id, str) or not MESSAGE_ID_RE.fullmatch(message_id)
|
or not isinstance(handoff_id, str) or not HANDOFF_ID_RE.fullmatch(handoff_id)
|
or not isinstance(sha256, str) or not UPPER_SHA256_RE.fullmatch(sha256)
|
):
|
raise ProtocolError("E_LINEAGE")
|
|
|
def _successor_record(value: dict[str, Any], allowed_creators: frozenset[str]) -> dict[str, Any]:
|
if set(value) != _SUCCESSOR_RECORD_KEYS or value.get("schema") != SUCCESSOR_QUEUE_SCHEMA_VERSION or value.get("record_type") != "SUCCESSOR_JOB":
|
raise ProtocolError("E_LINEAGE")
|
creator = value.get("creator_uid")
|
if creator not in allowed_creators:
|
raise ProtocolError("E_ALLOWLIST")
|
discovered = value.get("discovered_at_unix_ms")
|
if isinstance(discovered, bool) or not isinstance(discovered, int) or discovered <= 0:
|
raise ProtocolError("E_LINEAGE")
|
lineage = value.get("lineage")
|
if not isinstance(lineage, dict) or set(lineage) != _LINEAGE_KEYS:
|
raise ProtocolError("E_LINEAGE")
|
runtime = {
|
"job_id": value["job_id"],
|
"bvid": value["bvid"],
|
"creator_uid": creator,
|
"canonical_url": canonical_url(value["bvid"]),
|
"expected_duration_ms": value["expected_duration_ms"],
|
"published_at": value["published_at"],
|
"title": value["title"],
|
"lineage": lineage,
|
}
|
try:
|
validate_job(runtime)
|
except (ProtocolError, ValueError) as exc:
|
raise ProtocolError("E_LINEAGE") from exc
|
return runtime
|
|
|
def _queue_snapshot(payload: bytes, allowed_creators: frozenset[str]) -> _QueueSnapshot:
|
"""Parse committed schema-1 records and schema-2 blocks, retaining one tail suffix."""
|
if len(payload) > 16 * 1024 * 1024:
|
raise ProtocolError("E_QUEUE")
|
jobs: dict[str, dict[str, Any]] = {}
|
records: dict[str, dict[str, Any]] = {}
|
ingress: dict[str, dict[str, Any]] = {}
|
children: dict[str, str] = {}
|
blocks: list[tuple[str, str, str, tuple[str, ...], bytes]] = []
|
lines: list[tuple[int, int, bytes, dict[str, Any]]] = []
|
offset = 0
|
for raw_line in payload.splitlines(keepends=True):
|
start = offset
|
offset += len(raw_line)
|
if not raw_line.endswith(b"\n"):
|
break
|
content = raw_line[:-1]
|
if content.endswith(b"\r"):
|
content = content[:-1]
|
if not content or len(content) > 4096:
|
raise ProtocolError("E_QUEUE")
|
value = strict_json_loads(content)
|
if not isinstance(value, dict):
|
raise ProtocolError("E_QUEUE")
|
lines.append((start, offset, raw_line, value))
|
|
index = 0
|
committed_end = 0
|
while index < len(lines):
|
start, end, raw_line, value = lines[index]
|
if value.get("schema") == QUEUE_SCHEMA_VERSION:
|
if value.get("record_type") is not None:
|
raise ProtocolError("E_QUEUE")
|
job = _ingress_job(value, allowed_creators)
|
job_id = job["job_id"]
|
previous = ingress.get(job_id)
|
if previous is not None and previous != value:
|
raise ProtocolError("E_QUEUE_CONFLICT")
|
ingress.setdefault(job_id, value)
|
prior_job = jobs.get(job_id)
|
if prior_job is not None and prior_job != job:
|
raise ProtocolError("E_QUEUE_CONFLICT")
|
jobs.setdefault(job_id, job)
|
records.setdefault(job_id, value)
|
committed_end = end
|
index += 1
|
continue
|
|
if value.get("schema") != SUCCESSOR_QUEUE_SCHEMA_VERSION or value.get("record_type") != "SUCCESSOR_BEGIN":
|
raise ProtocolError("E_LINEAGE_RECOVERY")
|
if set(value) != _BEGIN_KEYS or raw_line != _canonical_line(value):
|
raise ProtocolError("E_LINEAGE")
|
_validate_authorization_identity(
|
value.get("authorization_message_id"), value.get("authorization_handoff_id"),
|
value.get("authorization_sha256"),
|
)
|
count = value.get("successor_count")
|
if isinstance(count, bool) or not isinstance(count, int) or not 1 <= count <= 100:
|
raise ProtocolError("E_LINEAGE")
|
needed = count + 2
|
if index + needed > len(lines):
|
break
|
block_lines = lines[index:index + needed]
|
job_ids: list[str] = []
|
block_records: list[tuple[dict[str, Any], dict[str, Any]]] = []
|
for _, _, job_line, raw_job in block_lines[1:-1]:
|
if job_line != _canonical_line(raw_job):
|
raise ProtocolError("E_LINEAGE")
|
runtime = _successor_record(raw_job, allowed_creators)
|
lineage = runtime["lineage"]
|
if (
|
lineage["authorization_message_id"] != value["authorization_message_id"]
|
or lineage["authorization_handoff_id"] != value["authorization_handoff_id"]
|
or lineage["authorization_sha256"] != value["authorization_sha256"]
|
):
|
raise ProtocolError("E_LINEAGE")
|
job_ids.append(runtime["job_id"])
|
block_records.append((raw_job, runtime))
|
if job_ids != sorted(job_ids) or len(set(job_ids)) != count:
|
raise ProtocolError("E_LINEAGE")
|
_, block_end, commit_line, commit = block_lines[-1]
|
if set(commit) != _COMMIT_KEYS or commit.get("schema") != SUCCESSOR_QUEUE_SCHEMA_VERSION or commit.get("record_type") != "SUCCESSOR_COMMIT" or commit_line != _canonical_line(commit):
|
raise ProtocolError("E_LINEAGE")
|
for key in _BEGIN_KEYS - {"record_type"}:
|
if commit.get(key) != value.get(key):
|
raise ProtocolError("E_LINEAGE")
|
block_prefix = b"".join(item[2] for item in block_lines[:-1])
|
block_hash = hashlib.sha256(block_prefix).hexdigest().upper()
|
if commit.get("block_sha256") != block_hash:
|
raise ProtocolError("E_LINEAGE")
|
full_block = block_prefix + commit_line
|
for raw_job, runtime in block_records:
|
job_id = runtime["job_id"]
|
parent = runtime["lineage"]["predecessor_job_id"]
|
if job_id in jobs or parent in children:
|
raise ProtocolError("E_LINEAGE_CONFLICT")
|
jobs[job_id] = runtime
|
records[job_id] = raw_job
|
children[parent] = job_id
|
blocks.append((
|
value["authorization_message_id"], value["authorization_handoff_id"],
|
value["authorization_sha256"], tuple(job_ids), full_block,
|
))
|
committed_end = block_end
|
index += needed
|
|
return _QueueSnapshot(
|
jobs=jobs,
|
records=records,
|
children=children,
|
blocks=tuple(blocks),
|
committed_bytes=payload[:committed_end],
|
pending_suffix=payload[committed_end:],
|
)
|
|
|
def _ingress_job(value: dict[str, Any], allowed_creators: frozenset[str]) -> dict[str, Any]:
|
expected = {"schema", "bvid", "creator_uid", "expected_duration_ms", "discovered_at_unix_ms", "published_at", "title"}
|
if set(value) != expected or value.get("schema") != QUEUE_SCHEMA_VERSION:
|
raise ProtocolError("E_QUEUE")
|
try:
|
bvid = validate_bvid(value["bvid"])
|
creator = validate_creator_uid(value["creator_uid"])
|
except ValueError as exc:
|
raise ProtocolError("E_JOB") from exc
|
duration = value["expected_duration_ms"]
|
discovered = value["discovered_at_unix_ms"]
|
if (
|
creator not in allowed_creators
|
or isinstance(duration, bool) or not isinstance(duration, int) or not 1_000 <= duration <= 86_400_000
|
or isinstance(discovered, bool) or not isinstance(discovered, int) or discovered <= 0
|
):
|
raise ProtocolError("E_ALLOWLIST" if creator not in allowed_creators else "E_JOB")
|
job = {
|
"job_id": stable_job_id(creator, bvid),
|
"bvid": bvid,
|
"creator_uid": creator,
|
"canonical_url": canonical_url(bvid),
|
"expected_duration_ms": duration,
|
"published_at": value["published_at"],
|
"title": value["title"],
|
}
|
try:
|
return validate_job(job)
|
except ProtocolError as exc:
|
raise ProtocolError("E_JOB") from exc
|
|
|
def validate_ingress_record(
|
value: dict[str, Any], allowed_creators: frozenset[str]
|
) -> dict[str, Any]:
|
"""Public producer/consumer boundary for one exact schema-1 record."""
|
return _ingress_job(value, allowed_creators)
|
|
|
_EVENTS = {
|
"CLAIMED", "STARTED", "MEDIA_COMPLETE", "POSTPROCESS_CLAIMED",
|
"COMPLETE", "FAILED", "POSTPROCESS_FAILED",
|
}
|
_TERMINAL_EVENTS = {"COMPLETE", "FAILED", "POSTPROCESS_FAILED"}
|
_POSTPROCESS_RECOVERY_KEYS = {
|
"media_complete_event_sha256", "media_complete_lease_id", "media",
|
}
|
|
|
def _media_complete_event_sha256(value: dict[str, Any]) -> str:
|
if value.get("event") != "MEDIA_COMPLETE":
|
raise ProtocolError("E_QUEUE_STATE")
|
return hashlib.sha256(_canonical_line(value)).hexdigest().upper()
|
|
|
def _postprocess_recovery_binding(value: dict[str, Any]) -> dict[str, Any]:
|
if value.get("event") != "MEDIA_COMPLETE" or "media" not in value:
|
raise ProtocolError("E_QUEUE_STATE")
|
return {
|
"media_complete_event_sha256": _media_complete_event_sha256(value),
|
"media_complete_lease_id": value["lease_id"],
|
"media": value["media"],
|
}
|
|
|
def _validate_postprocess_recovery_binding(
|
value: object, job: dict[str, Any],
|
) -> dict[str, Any]:
|
if not isinstance(value, dict) or set(value) != _POSTPROCESS_RECOVERY_KEYS:
|
raise ProtocolError("E_QUEUE_STATE")
|
digest = value["media_complete_event_sha256"]
|
prior_lease = value["media_complete_lease_id"]
|
if not isinstance(digest, str) or UPPER_SHA256_RE.fullmatch(digest) is None:
|
raise ProtocolError("E_QUEUE_STATE")
|
if (
|
not isinstance(prior_lease, str) or len(prior_lease) != 32
|
or any(ch not in "0123456789abcdef" for ch in prior_lease)
|
):
|
raise ProtocolError("E_QUEUE_STATE")
|
try:
|
media = validate_media_complete_identity(value["media"], job)
|
except ProtocolError as exc:
|
raise ProtocolError("E_QUEUE_STATE") from exc
|
return {
|
"media_complete_event_sha256": digest,
|
"media_complete_lease_id": prior_lease,
|
"media": media,
|
}
|
|
|
def _event(
|
value: dict[str, Any], jobs: dict[str, dict[str, Any]], *,
|
allow_exact_legacy_replay: bool = False,
|
allow_exact_missing_diagnostic_replay: bool = False,
|
) -> dict[str, Any]:
|
expected = {
|
"schema", "event", "job_id", "bvid", "creator_uid", "lease_id",
|
"at_unix_ms", "lease_expires_unix_ms", "error_code",
|
}
|
allowed_shapes = (
|
expected, expected | {"diagnostic"}, expected | {"media"},
|
expected | {"recovery"},
|
)
|
if set(value) not in allowed_shapes or value.get("schema") != QUEUE_SCHEMA_VERSION or value.get("event") not in _EVENTS:
|
raise ProtocolError("E_QUEUE_STATE")
|
if not isinstance(value["job_id"], str) or not JOB_ID_RE.fullmatch(value["job_id"]):
|
raise ProtocolError("E_QUEUE_STATE")
|
job = jobs.get(value["job_id"])
|
if job is None or value["creator_uid"] != job["creator_uid"] or value["bvid"] != job["bvid"]:
|
raise ProtocolError("E_QUEUE_STATE")
|
lease = value["lease_id"]
|
if not isinstance(lease, str) or len(lease) != 32 or any(ch not in "0123456789abcdef" for ch in lease):
|
raise ProtocolError("E_QUEUE_STATE")
|
for name in ("at_unix_ms", "lease_expires_unix_ms"):
|
if isinstance(value[name], bool) or not isinstance(value[name], int) or value[name] <= 0:
|
raise ProtocolError("E_QUEUE_STATE")
|
error = value["error_code"]
|
if error is not None and (not isinstance(error, str) or not error.startswith("E_")):
|
raise ProtocolError("E_QUEUE_STATE")
|
if value["event"] in {"FAILED", "POSTPROCESS_FAILED"} and error is None:
|
raise ProtocolError("E_QUEUE_STATE")
|
if value["event"] not in {"FAILED", "POSTPROCESS_FAILED"} and error is not None:
|
raise ProtocolError("E_QUEUE_STATE")
|
if value["event"] == "MEDIA_COMPLETE":
|
if "media" not in value:
|
raise ProtocolError("E_QUEUE_STATE")
|
try:
|
validate_media_complete_identity(value["media"], job)
|
except ProtocolError as exc:
|
raise ProtocolError("E_QUEUE_STATE") from exc
|
elif "media" in value:
|
raise ProtocolError("E_QUEUE_STATE")
|
if value["event"] == "POSTPROCESS_CLAIMED":
|
if "recovery" not in value:
|
raise ProtocolError("E_QUEUE_STATE")
|
_validate_postprocess_recovery_binding(value["recovery"], job)
|
elif "recovery" in value:
|
raise ProtocolError("E_QUEUE_STATE")
|
if error in INTERNAL_ONLY_PAGE_PENDING_CODES and not allow_exact_legacy_replay:
|
raise ProtocolError("E_QUEUE_STATE")
|
if "diagnostic" in value:
|
if value["event"] not in {"FAILED", "POSTPROCESS_FAILED"}:
|
raise ProtocolError("E_QUEUE_STATE")
|
try:
|
if value["event"] == "POSTPROCESS_FAILED":
|
validate_postprocess_terminal(error, value["diagnostic"])
|
elif error in PREPARELESS_REJECT_CODES:
|
validate_prepareless_terminal(error, value["diagnostic"])
|
else:
|
validate_runtime_diagnostic(value["diagnostic"])
|
except ValueError as exc:
|
raise ProtocolError("E_QUEUE_STATE") from exc
|
elif value["event"] == "POSTPROCESS_FAILED":
|
try:
|
validate_postprocess_terminal(error, None)
|
except ValueError as exc:
|
raise ProtocolError("E_QUEUE_STATE") from exc
|
elif error in PREPARELESS_REJECT_CODES and not allow_exact_missing_diagnostic_replay:
|
raise ProtocolError("E_QUEUE_STATE")
|
return value
|
|
|
def _terminal_request(
|
job: dict[str, Any], lease_id: str, now_ms: int, *, complete: bool,
|
error_code: str | None, diagnostic: dict[str, object] | None,
|
) -> dict[str, Any]:
|
"""Validate one live terminal request before any idempotent decision."""
|
|
validated_job = validate_job(job)
|
if not isinstance(complete, bool):
|
raise ProtocolError("E_QUEUE_STATE")
|
if error_code == COMPLETION_CLOSURE_REQUIRED:
|
raise ProtocolError("E_QUEUE_STATE")
|
payload: dict[str, Any] = {
|
"schema": QUEUE_SCHEMA_VERSION,
|
"event": "COMPLETE" if complete else "FAILED",
|
"job_id": validated_job["job_id"],
|
"bvid": validated_job["bvid"],
|
"creator_uid": validated_job["creator_uid"],
|
"lease_id": lease_id,
|
"at_unix_ms": now_ms,
|
"lease_expires_unix_ms": now_ms + QUEUE_LEASE_SECONDS * 1_000,
|
"error_code": error_code,
|
}
|
if diagnostic is not None:
|
payload["diagnostic"] = diagnostic
|
return _event(payload, {validated_job["job_id"]: validated_job})
|
|
|
def _is_governed_legacy_state_path(path: Path) -> bool:
|
if not path.is_absolute():
|
return False
|
suffix = tuple(part.casefold() for part in LEGACY_PAGE_METADATA_REPLAY_STATE_SUFFIX)
|
parts = tuple(part.casefold() for part in path.parts)
|
return len(parts) >= len(suffix) and parts[-len(suffix):] == suffix
|
|
|
def _replay_state_events(path: Path, jobs: dict[str, dict[str, Any]]) -> list[dict[str, Any]]:
|
if not path.exists():
|
return []
|
if not path.is_file() or path.is_symlink():
|
raise ProtocolError("E_QUEUE")
|
payload = path.read_bytes()
|
if len(payload) > 16 * 1024 * 1024:
|
raise ProtocolError("E_QUEUE")
|
if payload and not payload.endswith(b"\n"):
|
raise ProtocolError("E_QUEUE_PARTIAL")
|
|
governed = (
|
_is_governed_legacy_state_path(path)
|
and len(payload) >= LEGACY_PAGE_METADATA_REPLAY_PREFIX_BYTES
|
)
|
if governed and hashlib.sha256(
|
payload[:LEGACY_PAGE_METADATA_REPLAY_PREFIX_BYTES]
|
).hexdigest().upper() != LEGACY_PAGE_METADATA_REPLAY_PREFIX_SHA256:
|
raise ProtocolError("E_QUEUE_STATE")
|
|
expected = {
|
line_number: (line_bytes, line_sha256)
|
for line_number, line_bytes, line_sha256 in LEGACY_PAGE_METADATA_REPLAY_LINES
|
}
|
expected_missing = {
|
line_number: (line_bytes, line_sha256, error_code)
|
for line_number, line_bytes, line_sha256, error_code
|
in LEGACY_MISSING_DIAGNOSTIC_REPLAY_LINES
|
}
|
observed: list[tuple[int, int, str]] = []
|
observed_missing: list[tuple[int, int, str, str]] = []
|
events: list[dict[str, Any]] = []
|
latest_by_job: dict[str, dict[str, Any]] = {}
|
media_complete_by_job: dict[str, dict[str, Any]] = {}
|
for line_number, line in enumerate(payload.splitlines(keepends=True), 1):
|
if not line.endswith(b"\n") or len(line) <= 1 or len(line) - 1 > 4096:
|
raise ProtocolError("E_QUEUE")
|
value = strict_json_loads(line[:-1])
|
legacy = value.get("error_code") in INTERNAL_ONLY_PAGE_PENDING_CODES
|
allow = False
|
allow_missing = False
|
if legacy:
|
identity = (len(line), hashlib.sha256(line).hexdigest().upper())
|
allow = governed and expected.get(line_number) == identity
|
if not allow:
|
raise ProtocolError("E_QUEUE_STATE")
|
observed.append((line_number, *identity))
|
elif value.get("error_code") in PREPARELESS_REJECT_CODES and "diagnostic" not in value:
|
identity_with_error = (
|
len(line), hashlib.sha256(line).hexdigest().upper(), value["error_code"],
|
)
|
allow_missing = governed and expected_missing.get(line_number) == identity_with_error
|
if not allow_missing:
|
raise ProtocolError("E_QUEUE_STATE")
|
observed_missing.append((line_number, *identity_with_error))
|
parsed = _event(
|
value, jobs, allow_exact_legacy_replay=allow,
|
allow_exact_missing_diagnostic_replay=allow_missing,
|
)
|
job_id = parsed["job_id"]
|
latest = latest_by_job.get(job_id)
|
if parsed["event"] == "MEDIA_COMPLETE":
|
if job_id in media_complete_by_job:
|
raise ProtocolError("E_QUEUE_STATE")
|
if (
|
latest is None or latest["event"] != "STARTED"
|
or latest["lease_id"] != parsed["lease_id"]
|
):
|
raise ProtocolError("E_QUEUE_STATE")
|
media_complete_by_job[job_id] = parsed
|
elif parsed["event"] == "POSTPROCESS_CLAIMED":
|
prior_media = media_complete_by_job.get(job_id)
|
if (
|
prior_media is None
|
or latest is None
|
or latest["event"] not in {"MEDIA_COMPLETE", "POSTPROCESS_CLAIMED"}
|
or parsed["recovery"] != _postprocess_recovery_binding(prior_media)
|
):
|
raise ProtocolError("E_QUEUE_STATE")
|
elif latest is not None and latest["event"] == "MEDIA_COMPLETE":
|
if (
|
parsed["event"] not in {"COMPLETE", "POSTPROCESS_FAILED"}
|
or parsed["lease_id"] != latest["lease_id"]
|
):
|
raise ProtocolError("E_QUEUE_STATE")
|
elif latest is not None and latest["event"] == "POSTPROCESS_CLAIMED":
|
if (
|
parsed["event"] not in {"COMPLETE", "POSTPROCESS_FAILED"}
|
or parsed["lease_id"] != latest["lease_id"]
|
):
|
raise ProtocolError("E_QUEUE_STATE")
|
events.append(parsed)
|
latest_by_job[job_id] = parsed
|
if governed and tuple(observed) != LEGACY_PAGE_METADATA_REPLAY_LINES:
|
raise ProtocolError("E_QUEUE_STATE")
|
if governed and tuple(observed_missing) != LEGACY_MISSING_DIAGNOSTIC_REPLAY_LINES:
|
raise ProtocolError("E_QUEUE_STATE")
|
return events
|
|
|
class QueueStore:
|
def __init__(
|
self,
|
queue_path: Path,
|
state_path: Path,
|
lock_path: Path,
|
allowed_creators: frozenset[str],
|
) -> None:
|
self.queue_path = queue_path
|
self.state_path = state_path
|
self.lock_path = lock_path
|
self.allowed_creators = allowed_creators
|
|
@contextmanager
|
def _locked(self) -> Iterator[None]:
|
import msvcrt
|
|
self.lock_path.parent.mkdir(parents=True, exist_ok=True)
|
with self.lock_path.open("a+b") as stream:
|
if stream.seek(0, os.SEEK_END) == 0:
|
stream.write(b"\0")
|
stream.flush()
|
os.fsync(stream.fileno())
|
stream.seek(0)
|
try:
|
msvcrt.locking(stream.fileno(), msvcrt.LK_NBLCK, 1)
|
except OSError as exc:
|
raise ProtocolError("E_QUEUE_BUSY") from exc
|
try:
|
yield
|
finally:
|
stream.seek(0)
|
msvcrt.locking(stream.fileno(), msvcrt.LK_UNLCK, 1)
|
|
def _jobs(self) -> list[dict[str, Any]]:
|
payload = self.queue_path.read_bytes() if self.queue_path.exists() else b""
|
snapshot = _queue_snapshot(payload, self.allowed_creators)
|
if snapshot.pending_suffix:
|
raise ProtocolError("E_QUEUE_PARTIAL")
|
return list(snapshot.jobs.values())
|
|
def append_ingress_jobs(self, records: list[dict[str, Any]]) -> dict[str, int]:
|
"""Append a prevalidated producer batch under the consumer's exact lock.
|
|
The full incoming batch and the existing queue are checked before the
|
first append. Replaying byte-equivalent schema-1 records is idempotent;
|
any field drift for a stable job id fails closed.
|
"""
|
if not isinstance(records, list) or not 1 <= len(records) <= 10_000:
|
raise ProtocolError("E_QUEUE")
|
incoming: dict[str, tuple[dict[str, Any], dict[str, Any]]] = {}
|
order: list[str] = []
|
for raw in records:
|
if not isinstance(raw, dict):
|
raise ProtocolError("E_QUEUE")
|
job = _ingress_job(raw, self.allowed_creators)
|
job_id = job["job_id"]
|
previous = incoming.get(job_id)
|
if previous is not None and previous[0] != raw:
|
raise ProtocolError("E_QUEUE_CONFLICT")
|
if previous is None:
|
incoming[job_id] = (raw, job)
|
order.append(job_id)
|
|
with self._locked():
|
payload = self.queue_path.read_bytes() if self.queue_path.exists() else b""
|
snapshot = _queue_snapshot(payload, self.allowed_creators)
|
if snapshot.pending_suffix:
|
raise ProtocolError("E_QUEUE_PARTIAL")
|
existing = snapshot.records
|
|
to_append: list[dict[str, Any]] = []
|
unchanged = 0
|
for job_id in order:
|
raw = incoming[job_id][0]
|
previous = existing.get(job_id)
|
if previous is None:
|
to_append.append(raw)
|
elif previous == raw:
|
unchanged += 1
|
else:
|
raise ProtocolError("E_QUEUE_CONFLICT")
|
_append_jsonl_batch(self.queue_path, to_append)
|
return {"appended": len(to_append), "unchanged": unchanged}
|
|
@staticmethod
|
def _validate_release(approval: ReleaseApproval) -> None:
|
if not isinstance(approval, ReleaseApproval):
|
raise ProtocolError("E_AUTH_TRUST")
|
_validate_authorization_identity(
|
approval.authorization_message_id, approval.authorization_handoff_id,
|
approval.authorization_sha256,
|
)
|
if (
|
not MESSAGE_ID_RE.fullmatch(approval.repair_review_result_message_id)
|
or not isinstance(approval.repair_audit_id, str)
|
or not approval.repair_audit_id.startswith("DEV-AUDIT-")
|
or isinstance(approval.repair_audit_bytes, bool)
|
or not isinstance(approval.repair_audit_bytes, int)
|
or approval.repair_audit_bytes <= 0
|
or not UPPER_SHA256_RE.fullmatch(approval.repair_audit_sha256)
|
or not isinstance(approval.successors, tuple)
|
or not 1 <= len(approval.successors) <= 100
|
):
|
raise ProtocolError("E_AUTH_TRUST")
|
identities: list[tuple[str, str, str]] = []
|
for item in approval.successors:
|
if not isinstance(item, AuthorizedSuccessor):
|
raise ProtocolError("E_AUTH_TRUST")
|
try:
|
validate_creator_uid(item.creator_uid)
|
validate_bvid(item.bvid)
|
except ValueError as exc:
|
raise ProtocolError("E_AUTH_TRUST") from exc
|
if (
|
not JOB_ID_RE.fullmatch(item.predecessor_job_id)
|
or isinstance(item.retry_generation, bool)
|
or not isinstance(item.retry_generation, int)
|
or not 1 <= item.retry_generation <= 1_000_000
|
or not ERROR_CODE_RE.fullmatch(item.terminal_error_code)
|
):
|
raise ProtocolError("E_AUTH_TRUST")
|
identities.append((item.creator_uid, item.bvid, item.predecessor_job_id))
|
if identities != sorted(identities) or len(set(identities)) != len(identities):
|
raise ProtocolError("E_AUTH_TRUST")
|
|
@staticmethod
|
def _build_successor_block(
|
approval: ReleaseApproval, records: list[dict[str, Any]]
|
) -> tuple[bytes, tuple[str, ...]]:
|
ordered = sorted(records, key=lambda value: value["job_id"])
|
begin = {
|
"schema": SUCCESSOR_QUEUE_SCHEMA_VERSION,
|
"record_type": "SUCCESSOR_BEGIN",
|
"authorization_message_id": approval.authorization_message_id,
|
"authorization_handoff_id": approval.authorization_handoff_id,
|
"authorization_sha256": approval.authorization_sha256,
|
"successor_count": len(ordered),
|
}
|
prefix = _canonical_line(begin) + b"".join(_canonical_line(value) for value in ordered)
|
commit = {
|
**begin,
|
"record_type": "SUCCESSOR_COMMIT",
|
"block_sha256": hashlib.sha256(prefix).hexdigest().upper(),
|
}
|
block = prefix + _canonical_line(commit)
|
if len(block) > 1024 * 1024:
|
raise ProtocolError("E_LINEAGE")
|
return block, tuple(value["job_id"] for value in ordered)
|
|
def append_authorized_successors(
|
self,
|
approval_loader: Callable[[], ReleaseApproval],
|
catalog_records: list[dict[str, Any]],
|
) -> dict[str, int]:
|
"""Append or recover one deterministic authorized schema-2 block."""
|
catalog: dict[tuple[str, str], dict[str, Any]] = {}
|
if not isinstance(catalog_records, list) or not catalog_records:
|
raise ProtocolError("E_CATALOG")
|
for raw in catalog_records:
|
if not isinstance(raw, dict):
|
raise ProtocolError("E_CATALOG")
|
job = _ingress_job(raw, self.allowed_creators)
|
key = (job["creator_uid"], job["bvid"])
|
previous = catalog.get(key)
|
if previous is not None and previous != raw:
|
raise ProtocolError("E_CATALOG_CONFLICT")
|
catalog.setdefault(key, raw)
|
|
with self._locked():
|
approval = approval_loader()
|
self._validate_release(approval)
|
payload = self.queue_path.read_bytes() if self.queue_path.exists() else b""
|
snapshot = _queue_snapshot(payload, self.allowed_creators)
|
events = self._events(snapshot.jobs)
|
latest: dict[str, dict[str, Any]] = {}
|
for event in events:
|
latest[event["job_id"]] = event
|
successor_records: list[dict[str, Any]] = []
|
for item in approval.successors:
|
predecessor = snapshot.jobs.get(item.predecessor_job_id)
|
source = snapshot.records.get(item.predecessor_job_id)
|
current_catalog = catalog.get((item.creator_uid, item.bvid))
|
if predecessor is None or source is None or current_catalog is None:
|
raise ProtocolError("E_LINEAGE_TERMINAL")
|
if predecessor["creator_uid"] != item.creator_uid or predecessor["bvid"] != item.bvid:
|
raise ProtocolError("E_LINEAGE_CONFLICT")
|
state = latest.get(item.predecessor_job_id)
|
failed_terminal = (
|
state is not None and state["event"] in {"FAILED", "POSTPROCESS_FAILED"}
|
and item.terminal_error_code != COMPLETION_CLOSURE_REQUIRED
|
and state["error_code"] == item.terminal_error_code
|
)
|
completion_closure = (
|
state is not None and state["event"] == "COMPLETE"
|
and state["error_code"] is None
|
and item.terminal_error_code == COMPLETION_CLOSURE_REQUIRED
|
)
|
if not (failed_terminal or completion_closure):
|
raise ProtocolError("E_LINEAGE_TERMINAL")
|
predecessor_lineage = predecessor.get("lineage")
|
expected_generation = 1 if predecessor_lineage is None else predecessor_lineage["retry_generation"] + 1
|
if item.retry_generation != expected_generation:
|
raise ProtocolError("E_LINEAGE_GENERATION")
|
for key in (
|
"bvid", "creator_uid", "expected_duration_ms", "discovered_at_unix_ms",
|
"published_at", "title",
|
):
|
if source.get(key) != current_catalog.get(key):
|
raise ProtocolError("E_CATALOG_CONFLICT")
|
lineage = {
|
"predecessor_job_id": item.predecessor_job_id,
|
"retry_generation": item.retry_generation,
|
"predecessor_terminal_error_code": item.terminal_error_code,
|
"authorization_message_id": approval.authorization_message_id,
|
"authorization_handoff_id": approval.authorization_handoff_id,
|
"authorization_sha256": approval.authorization_sha256,
|
"repair_review_result_message_id": approval.repair_review_result_message_id,
|
"repair_audit_id": approval.repair_audit_id,
|
"repair_audit_bytes": approval.repair_audit_bytes,
|
"repair_audit_sha256": approval.repair_audit_sha256,
|
}
|
try:
|
job_id = stable_successor_job_id(
|
item.creator_uid, item.bvid, item.predecessor_job_id,
|
item.retry_generation, item.terminal_error_code,
|
approval.authorization_message_id, approval.authorization_handoff_id,
|
approval.authorization_sha256, approval.repair_review_result_message_id,
|
approval.repair_audit_id, approval.repair_audit_bytes,
|
approval.repair_audit_sha256,
|
)
|
except ValueError as exc:
|
raise ProtocolError("E_LINEAGE") from exc
|
successor_records.append({
|
"schema": SUCCESSOR_QUEUE_SCHEMA_VERSION,
|
"record_type": "SUCCESSOR_JOB",
|
"job_id": job_id,
|
"bvid": item.bvid,
|
"creator_uid": item.creator_uid,
|
"expected_duration_ms": source["expected_duration_ms"],
|
"discovered_at_unix_ms": source["discovered_at_unix_ms"],
|
"published_at": source["published_at"],
|
"title": source["title"],
|
"lineage": lineage,
|
})
|
|
block, target_ids = self._build_successor_block(approval, successor_records)
|
matching_blocks = [
|
existing for existing in snapshot.blocks
|
if existing[:3] == (
|
approval.authorization_message_id, approval.authorization_handoff_id,
|
approval.authorization_sha256,
|
)
|
]
|
if matching_blocks:
|
if len(matching_blocks) != 1 or matching_blocks[0][3] != target_ids or matching_blocks[0][4] != block or snapshot.pending_suffix:
|
raise ProtocolError("E_LINEAGE_CONFLICT")
|
return {"appended": 0, "unchanged": len(target_ids), "recovered": 0}
|
for item, target_id in zip(approval.successors, (
|
value["job_id"] for value in successor_records
|
)):
|
existing_child = snapshot.children.get(item.predecessor_job_id)
|
if existing_child is not None and existing_child != target_id:
|
raise ProtocolError("E_LINEAGE_CONFLICT")
|
suffix = snapshot.pending_suffix
|
if suffix and not block.startswith(suffix):
|
raise ProtocolError("E_LINEAGE_RECOVERY")
|
remaining = block[len(suffix):]
|
_append_bytes(self.queue_path, remaining)
|
final_payload = self.queue_path.read_bytes()
|
if not final_payload.startswith(payload) or final_payload != snapshot.committed_bytes + block:
|
raise ProtocolError("E_LINEAGE_RECOVERY")
|
final_snapshot = _queue_snapshot(final_payload, self.allowed_creators)
|
if final_snapshot.pending_suffix or not any(existing[3] == target_ids and existing[4] == block for existing in final_snapshot.blocks):
|
raise ProtocolError("E_LINEAGE_RECOVERY")
|
return {
|
"appended": len(target_ids),
|
"unchanged": 0,
|
"recovered": len(target_ids) if suffix else 0,
|
}
|
|
def _events(self, jobs: dict[str, dict[str, Any]] | None = None) -> list[dict[str, Any]]:
|
if jobs is None:
|
payload = self.queue_path.read_bytes() if self.queue_path.exists() else b""
|
snapshot = _queue_snapshot(payload, self.allowed_creators)
|
if snapshot.pending_suffix:
|
raise ProtocolError("E_QUEUE_PARTIAL")
|
jobs = snapshot.jobs
|
return _replay_state_events(self.state_path, jobs)
|
|
def claim_next(self, now_ms: int) -> tuple[dict[str, Any], str] | None:
|
with self._locked():
|
payload = self.queue_path.read_bytes() if self.queue_path.exists() else b""
|
snapshot = _queue_snapshot(payload, self.allowed_creators)
|
if snapshot.pending_suffix:
|
raise ProtocolError("E_QUEUE_PARTIAL")
|
jobs = list(snapshot.jobs.values())
|
events = self._events(snapshot.jobs)
|
latest: dict[str, dict[str, Any]] = {}
|
for item in events:
|
latest[item["job_id"]] = item
|
for job in jobs:
|
current = latest.get(job["job_id"])
|
if current and current["event"] in _TERMINAL_EVENTS:
|
continue
|
if current and current["event"] in {"MEDIA_COMPLETE", "POSTPROCESS_CLAIMED"}:
|
if (
|
current["event"] == "POSTPROCESS_CLAIMED"
|
and current["lease_expires_unix_ms"] >= now_ms
|
):
|
continue
|
prior = [
|
item for item in events
|
if item["job_id"] == job["job_id"]
|
and item["event"] == "MEDIA_COMPLETE"
|
]
|
if len(prior) != 1:
|
raise ProtocolError("E_QUEUE_STATE")
|
lease = secrets.token_hex(16)
|
recovery = _postprocess_recovery_binding(prior[0])
|
_postprocess_recovery_claim_test_seam("BEFORE_APPEND")
|
self._append_event(
|
job, lease, "POSTPROCESS_CLAIMED", now_ms, None,
|
recovery=recovery,
|
)
|
_postprocess_recovery_claim_test_seam("AFTER_APPEND_BEFORE_READBACK")
|
reread = [
|
item for item in self._events(snapshot.jobs)
|
if item["job_id"] == job["job_id"]
|
]
|
if (
|
not reread or reread[-1]["event"] != "POSTPROCESS_CLAIMED"
|
or reread[-1]["lease_id"] != lease
|
or reread[-1].get("recovery") != recovery
|
):
|
raise ProtocolError("E_QUEUE_WRITE")
|
_postprocess_recovery_claim_test_seam("AFTER_READBACK")
|
return job, lease
|
if current and current["event"] == "STARTED":
|
if current["lease_expires_unix_ms"] < now_ms:
|
self._append_event(job, current["lease_id"], "FAILED", now_ms, "E_ORPHANED")
|
continue
|
if current and current["event"] == "CLAIMED":
|
if current["lease_expires_unix_ms"] >= now_ms:
|
continue
|
attempts = sum(
|
item["event"] == "CLAIMED" and item["job_id"] == job["job_id"]
|
for item in events
|
)
|
if attempts >= MAX_CLAIM_ATTEMPTS:
|
self._append_event(job, current["lease_id"], "FAILED", now_ms, "E_CLAIM_EXPIRED")
|
continue
|
lease = secrets.token_hex(16)
|
self._append_event(job, lease, "CLAIMED", now_ms, None)
|
return job, lease
|
return None
|
|
def _append_event(
|
self, job: dict[str, Any], lease_id: str, event: str, now_ms: int, error_code: str | None,
|
diagnostic: dict[str, object] | None = None,
|
media: dict[str, Any] | None = None,
|
recovery: dict[str, Any] | None = None,
|
) -> None:
|
validate_job(job)
|
if error_code == COMPLETION_CLOSURE_REQUIRED:
|
raise ProtocolError("E_QUEUE_STATE")
|
if error_code in INTERNAL_ONLY_PAGE_PENDING_CODES:
|
raise ProtocolError("E_QUEUE_STATE")
|
if error_code in PREPARELESS_REJECT_CODES:
|
try:
|
diagnostic = validate_prepareless_terminal(error_code, diagnostic)
|
except ValueError as exc:
|
raise ProtocolError("E_QUEUE_STATE") from exc
|
if event != "FAILED":
|
raise ProtocolError("E_QUEUE_STATE")
|
elif event == "POSTPROCESS_FAILED":
|
try:
|
diagnostic = validate_postprocess_terminal(error_code, diagnostic)
|
except ValueError as exc:
|
raise ProtocolError("E_QUEUE_STATE") from exc
|
elif diagnostic is not None:
|
try:
|
validate_runtime_diagnostic(diagnostic)
|
except ValueError as exc:
|
raise ProtocolError("E_QUEUE_STATE") from exc
|
if event not in {"FAILED", "POSTPROCESS_FAILED"}:
|
raise ProtocolError("E_QUEUE_STATE")
|
payload = {
|
"schema": QUEUE_SCHEMA_VERSION,
|
"event": event,
|
"job_id": job["job_id"],
|
"bvid": job["bvid"],
|
"creator_uid": job["creator_uid"],
|
"lease_id": lease_id,
|
"at_unix_ms": now_ms,
|
"lease_expires_unix_ms": now_ms + QUEUE_LEASE_SECONDS * 1_000,
|
"error_code": error_code,
|
}
|
if diagnostic is not None:
|
payload["diagnostic"] = diagnostic
|
if event == "MEDIA_COMPLETE":
|
try:
|
payload["media"] = validate_media_complete_identity(media, job)
|
except ProtocolError as exc:
|
raise ProtocolError("E_QUEUE_STATE") from exc
|
elif media is not None:
|
raise ProtocolError("E_QUEUE_STATE")
|
if event == "POSTPROCESS_CLAIMED":
|
payload["recovery"] = _validate_postprocess_recovery_binding(recovery, job)
|
elif recovery is not None:
|
raise ProtocolError("E_QUEUE_STATE")
|
_append_jsonl(
|
self.state_path,
|
payload,
|
)
|
|
def assert_claim(self, job: dict[str, Any], lease_id: str, now_ms: int) -> None:
|
validate_job(job)
|
with self._locked():
|
latest = None
|
for item in self._events():
|
if item["job_id"] == job["job_id"]:
|
latest = item
|
if (
|
latest is None or latest["event"] != "CLAIMED" or latest["lease_id"] != lease_id
|
or latest["lease_expires_unix_ms"] < now_ms
|
):
|
raise ProtocolError("E_LEASE")
|
|
def mark_started(self, job: dict[str, Any], lease_id: str, now_ms: int) -> None:
|
with self._locked():
|
latest = None
|
for item in self._events():
|
if item["job_id"] == job["job_id"]:
|
latest = item
|
if latest is None or latest["event"] != "CLAIMED" or latest["lease_id"] != lease_id:
|
raise ProtocolError("E_LEASE")
|
self._append_event(job, lease_id, "STARTED", now_ms, None)
|
|
def postprocess_recovery_claim(
|
self, job: dict[str, Any], lease_id: str,
|
) -> dict[str, Any] | None:
|
"""Return the exact durable recovery binding, or None for an ordinary claim."""
|
|
validated_job = validate_job(job)
|
with self._locked():
|
matching = [
|
item for item in self._events()
|
if item["job_id"] == validated_job["job_id"]
|
]
|
latest = matching[-1] if matching else None
|
if latest is None or latest["lease_id"] != lease_id:
|
raise ProtocolError("E_LEASE")
|
if latest["event"] == "CLAIMED":
|
return None
|
if latest["event"] != "POSTPROCESS_CLAIMED":
|
raise ProtocolError("E_LEASE")
|
prior = [item for item in matching if item["event"] == "MEDIA_COMPLETE"]
|
if (
|
len(prior) != 1
|
or latest.get("recovery") != _postprocess_recovery_binding(prior[0])
|
):
|
raise ProtocolError("E_QUEUE_STATE")
|
return dict(latest["recovery"])
|
|
def mark_media_complete(
|
self, job: dict[str, Any], lease_id: str, now_ms: int,
|
media: dict[str, Any],
|
) -> dict[str, Any]:
|
"""Persist verified user media before any reentrant postprocess outcome."""
|
validated_job = validate_job(job)
|
validated_media = validate_media_complete_identity(media, validated_job)
|
with self._locked():
|
matching = [
|
item for item in self._events()
|
if item["job_id"] == job["job_id"]
|
]
|
latest = matching[-1] if matching else None
|
prior = [item for item in matching if item["event"] == "MEDIA_COMPLETE"]
|
if prior:
|
if (
|
len(prior) != 1
|
or prior[0].get("media") != validated_media
|
):
|
raise ProtocolError("E_LEASE")
|
if latest is None or latest["lease_id"] != lease_id:
|
raise ProtocolError("E_LEASE")
|
if prior[0]["lease_id"] == lease_id:
|
if latest["event"] not in {
|
"MEDIA_COMPLETE", "COMPLETE", "POSTPROCESS_FAILED",
|
}:
|
raise ProtocolError("E_LEASE")
|
else:
|
recovery = [
|
item for item in matching
|
if item["event"] == "POSTPROCESS_CLAIMED"
|
and item["lease_id"] == lease_id
|
]
|
if (
|
len(recovery) != 1
|
or recovery[0].get("recovery") != _postprocess_recovery_binding(prior[0])
|
or latest["event"] not in {
|
"POSTPROCESS_CLAIMED", "COMPLETE", "POSTPROCESS_FAILED",
|
}
|
):
|
raise ProtocolError("E_LEASE")
|
return prior[0]
|
if latest is None or latest["event"] != "STARTED" or latest["lease_id"] != lease_id:
|
raise ProtocolError("E_LEASE")
|
_media_complete_durability_test_seam("BEFORE_APPEND")
|
self._append_event(
|
validated_job, lease_id, "MEDIA_COMPLETE", now_ms, None,
|
media=validated_media,
|
)
|
_media_complete_durability_test_seam("AFTER_APPEND_BEFORE_READBACK")
|
matching = [
|
item for item in self._events()
|
if item["job_id"] == validated_job["job_id"]
|
]
|
if not matching or matching[-1].get("media") != validated_media:
|
raise ProtocolError("E_QUEUE_WRITE")
|
_media_complete_durability_test_seam("AFTER_READBACK")
|
return matching[-1]
|
|
def mark_postprocess_failed(
|
self, job: dict[str, Any], lease_id: str, now_ms: int, *,
|
error_code: str, diagnostic: dict[str, object] | None = None,
|
) -> None:
|
"""Terminalize postprocess without erasing the prior MEDIA_COMPLETE fact."""
|
validated_job = validate_job(job)
|
try:
|
diagnostic = validate_postprocess_terminal(error_code, diagnostic)
|
except ValueError as exc:
|
raise ProtocolError("E_QUEUE_STATE") from exc
|
if (
|
not isinstance(error_code, str)
|
or not error_code.startswith("E_")
|
):
|
raise ProtocolError("E_QUEUE_STATE")
|
requested = {
|
"schema": QUEUE_SCHEMA_VERSION,
|
"event": "POSTPROCESS_FAILED",
|
"job_id": validated_job["job_id"],
|
"bvid": validated_job["bvid"],
|
"creator_uid": validated_job["creator_uid"],
|
"lease_id": lease_id,
|
"at_unix_ms": now_ms,
|
"lease_expires_unix_ms": now_ms + QUEUE_LEASE_SECONDS * 1_000,
|
"error_code": error_code,
|
}
|
if diagnostic is not None:
|
requested["diagnostic"] = diagnostic
|
requested = _event(requested, {validated_job["job_id"]: validated_job})
|
with self._locked():
|
matching = [
|
item for item in self._events()
|
if item["job_id"] == job["job_id"]
|
]
|
latest = matching[-1] if matching else None
|
if latest and latest["event"] in _TERMINAL_EVENTS:
|
identity_fields = (
|
"event", "job_id", "bvid", "creator_uid", "lease_id", "error_code",
|
)
|
if (
|
any(latest[name] != requested[name] for name in identity_fields)
|
or ("diagnostic" in latest) != ("diagnostic" in requested)
|
or latest.get("diagnostic") != requested.get("diagnostic")
|
):
|
raise ProtocolError("E_LEASE")
|
return
|
if (
|
latest is None
|
or latest["event"] not in {"MEDIA_COMPLETE", "POSTPROCESS_CLAIMED"}
|
or latest["lease_id"] != lease_id
|
):
|
raise ProtocolError("E_LEASE")
|
self._append_event(
|
job, lease_id, "POSTPROCESS_FAILED", now_ms,
|
error_code, requested.get("diagnostic"),
|
)
|
|
def reject_claim(
|
self, job: dict[str, Any], lease_id: str, now_ms: int, error_code: str,
|
diagnostic: dict[str, object] | None = None,
|
) -> None:
|
validate_job(job)
|
try:
|
validated_diagnostic = validate_prepareless_terminal(error_code, diagnostic)
|
except ValueError as exc:
|
raise ProtocolError("E_REJECT")
|
with self._locked():
|
latest = None
|
for item in self._events():
|
if item["job_id"] == job["job_id"]:
|
latest = item
|
if latest and latest["event"] == "FAILED" and latest["lease_id"] == lease_id:
|
if latest["error_code"] != error_code or latest.get("diagnostic") != validated_diagnostic:
|
raise ProtocolError("E_LEASE")
|
return
|
if latest is None or latest["event"] != "CLAIMED" or latest["lease_id"] != lease_id:
|
raise ProtocolError("E_LEASE")
|
self._append_event(job, lease_id, "FAILED", now_ms, error_code, validated_diagnostic)
|
|
def mark_terminal(
|
self, job: dict[str, Any], lease_id: str, now_ms: int, *, complete: bool,
|
error_code: str | None, diagnostic: dict[str, object] | None = None,
|
) -> None:
|
requested = _terminal_request(
|
job, lease_id, now_ms, complete=complete,
|
error_code=error_code, diagnostic=diagnostic,
|
)
|
with self._locked():
|
latest = None
|
for item in self._events():
|
if item["job_id"] == job["job_id"]:
|
latest = item
|
if latest and latest["event"] in _TERMINAL_EVENTS:
|
identity_fields = (
|
"event", "job_id", "bvid", "creator_uid", "lease_id", "error_code",
|
)
|
if (
|
any(latest[name] != requested[name] for name in identity_fields)
|
or ("diagnostic" in latest) != ("diagnostic" in requested)
|
or latest.get("diagnostic") != requested.get("diagnostic")
|
):
|
raise ProtocolError("E_LEASE")
|
return
|
if latest is None or latest["lease_id"] != lease_id:
|
raise ProtocolError("E_LEASE")
|
if latest["event"] == "MEDIA_COMPLETE" and requested["event"] != "COMPLETE":
|
raise ProtocolError("E_QUEUE_STATE")
|
if latest["event"] == "POSTPROCESS_CLAIMED" and requested["event"] != "COMPLETE":
|
raise ProtocolError("E_QUEUE_STATE")
|
self._append_event(
|
job, lease_id, requested["event"], now_ms,
|
requested["error_code"], requested.get("diagnostic"),
|
)
|
|
|
class ReloadStore:
|
def __init__(self, path: Path, generation: str) -> None:
|
self.path = path
|
self.generation = generation
|
|
def _records(self) -> list[dict[str, Any]]:
|
records = _read_jsonl(self.path, missing_ok=True)
|
for value in records:
|
if set(value) != {"schema", "generation", "event", "token", "from_build", "to_build", "at_unix_ms"}:
|
raise ProtocolError("E_RELOAD_STATE")
|
if value["schema"] != 1 or value["event"] not in {"OFFERED", "BEGIN", "APPLIED"}:
|
raise ProtocolError("E_RELOAD_STATE")
|
if not all(isinstance(value[name], str) for name in ("generation", "token", "from_build", "to_build")):
|
raise ProtocolError("E_RELOAD_STATE")
|
if len(value["token"]) != 32 or any(ch not in "0123456789abcdef" for ch in value["token"]):
|
raise ProtocolError("E_RELOAD_STATE")
|
if isinstance(value["at_unix_ms"], bool) or not isinstance(value["at_unix_ms"], int):
|
raise ProtocolError("E_RELOAD_STATE")
|
return records
|
|
def status(self, current_build: str, now_ms: int) -> dict[str, Any]:
|
records = [item for item in self._records() if item["generation"] == self.generation]
|
if current_build == EXTENSION_BUILD:
|
if not records or records[-1]["event"] != "APPLIED":
|
token = records[-1]["token"] if records else secrets.token_hex(16)
|
_append_jsonl(self.path, {"schema": 1, "generation": self.generation, "event": "APPLIED", "token": token, "from_build": current_build, "to_build": EXTENSION_BUILD, "at_unix_ms": now_ms})
|
return {"required_extension_build": EXTENSION_BUILD, "reload_required": False, "reload_token": None, "retry_after_unix_ms": 0}
|
begun = next((item for item in reversed(records) if item["event"] == "BEGIN"), None)
|
if begun is not None:
|
return {"required_extension_build": EXTENSION_BUILD, "reload_required": False, "reload_token": None, "retry_after_unix_ms": begun["at_unix_ms"] + RELOAD_BACKOFF_SECONDS * 1_000}
|
offered = records[-1] if records and records[-1]["event"] == "OFFERED" else None
|
if offered is None:
|
offered = {"schema": 1, "generation": self.generation, "event": "OFFERED", "token": secrets.token_hex(16), "from_build": current_build, "to_build": EXTENSION_BUILD, "at_unix_ms": now_ms}
|
_append_jsonl(self.path, offered)
|
return {"required_extension_build": EXTENSION_BUILD, "reload_required": True, "reload_token": offered["token"], "retry_after_unix_ms": 0}
|
|
def begin(self, current_build: str, token: str, now_ms: int) -> None:
|
records = [item for item in self._records() if item["generation"] == self.generation]
|
if current_build == EXTENSION_BUILD or not records:
|
raise ProtocolError("E_RELOAD")
|
latest = records[-1]
|
if latest["event"] != "OFFERED" or latest["token"] != token or latest["from_build"] != current_build:
|
raise ProtocolError("E_RELOAD")
|
_append_jsonl(self.path, {**latest, "event": "BEGIN", "at_unix_ms": now_ms})
|
