Add persistent lifecycle checkpoints and resumable inbox processing
This commit is contained in:
Nico committed
1 parent
f832b038a5
commit
e985e1d5f0
15 files changed
+1401
-1
No files matched your search
@@ -4,6 +4,8 @@ Project-independent foundations for reading, validating and extending ZIP case
|
||||
packages. The authoritative design is [docs/architecture.md](docs/architecture.md).
|
||||
The implemented v1 contract is documented in
|
||||
[docs/package-format-v1.md](docs/package-format-v1.md).
|
||||
Job state, checkpoints and inbox admission are documented in
|
||||
[docs/lifecycle-v1.md](docs/lifecycle-v1.md).
|
||||
|
||||
Requires Python 3.11 or newer.
|
||||
|
||||
@@ -13,5 +15,6 @@ python -m venv .venv
|
||||
.\.venv\Scripts\python.exe -m pytest
|
||||
```
|
||||
|
||||
This milestone includes no lifecycle worker, CaseBuilder, LLM,
|
||||
The lifecycle foundation supports one-pass inbox processing and resumable package
|
||||
checkpoints. It includes no analysis worker, CaseBuilder, LLM,
|
||||
case interpretation, hypothesis generation or investigation tools.
|
||||
@@ -0,0 +1,166 @@
|
||||
# Lifecycle foundation v1
|
||||
|
||||
`architecture.md` remains the authoritative design. This milestone provides only
|
||||
deterministic job state, checkpoint storage, locking and inbox admission. It does
|
||||
not execute CaseBuilder, interpretation, hypotheses, diagnosis, tools or LLMs.
|
||||
State APIs allow future phase implementations to confirm their work; recording
|
||||
a state is not evidence that this foundation executed that phase.
|
||||
|
||||
## Package state and service state
|
||||
|
||||
The package manifest is authoritative for identity, generation, revision, state
|
||||
and routing. Detailed progress is stored as immutable JSON artifacts at
|
||||
`history/checkpoints/<hash-of-step-id>.json`. The manifest's
|
||||
`artifacts.lifecycle_checkpoint` points to the latest artifact. Each checkpoint
|
||||
links to its predecessor, with format/version, identities, output generation,
|
||||
analysis revision, previous/current state, routing snapshot, optional resume state, step identity,
|
||||
request hash, source generation and content fingerprint, creation time and opaque
|
||||
structured `details`. `checkpoint-v1.schema.json` defines the JSON contract.
|
||||
Readers additionally check paths, cycles, state values, generation/revision order,
|
||||
identity and agreement with the manifest for a matching generation. Checkpoint
|
||||
JSON reads use the configurable manifest-byte limit as their security cap.
|
||||
|
||||
Existing `analysis/current.json` is not repurposed as worker state. The checkpoint
|
||||
namespace and manifest checkpoint pointer are reserved to the lifecycle service.
|
||||
Checkpoints are outside frozen analysis revision directories. Every successful
|
||||
checkpoint uses PackageWriter to create and validate one new package generation;
|
||||
all old artifacts remain. No checkpoint exists only in an external database.
|
||||
|
||||
Service layout:
|
||||
|
||||
```text
|
||||
<workspace>/
|
||||
inbox/ incoming files of arbitrary names/extensions
|
||||
quarantine/<unique-id>/ package.zip and validation_error.json
|
||||
cache/<raw-archive-sha256>.zip exact immutable input archives
|
||||
cases/<case-hash>/<package-hash>/
|
||||
g00000001.zip immutable confirmed generations
|
||||
current.json reconstructible local lookup index
|
||||
outbox/ reserved for later analyzed-package delivery
|
||||
review/ reserved for later human-review delivery
|
||||
runtime/
|
||||
current_job.json coarse worker summary, or null when idle
|
||||
locks/worker.lock persistent lock inode/file
|
||||
locks/case-<case-hash>.lock
|
||||
```
|
||||
|
||||
Opaque IDs are hashed before use in filesystem paths, including for locking.
|
||||
Stored-generation filenames are internal lookup conventions and are always
|
||||
checked against the manifest; incoming filenames never supply identity or state.
|
||||
Runtime includes identities, input generation, analysis revision, target stage,
|
||||
start time and last confirmed checkpoint time. It contains no detailed progress.
|
||||
Runtime and local index JSON are written via temporary file, flush/fsync and
|
||||
atomic replacement. They can be stale after a crash and are never authoritative.
|
||||
Current generation is recovered from fully validated managed packages. Corrupt
|
||||
managed storage is reported rather than silently rolling back to an older file.
|
||||
|
||||
## Transitions and retry
|
||||
|
||||
Normal transitions follow the architecture's linear sequence from NEW through
|
||||
HANDED_OFF. REOPENED proceeds to BUILDING_CASE. Active phases may checkpoint
|
||||
progress in the same state, become BLOCKED, or become FAILED_TO_ANALYZE.
|
||||
HANDED_OFF and FAILED_TO_ANALYZE can become REOPENED. Skips, regressions and
|
||||
unrelated BLOCKED resumes are rejected.
|
||||
|
||||
The initial VALIDATED to BUILDING_CASE transition starts revision 1 if it is
|
||||
still zero. REOPENED increments analysis revision exactly once; subsequently
|
||||
entering BUILDING_CASE preserves it. The revision summary counter is maintained
|
||||
without altering other history counters. Package generation changes independently.
|
||||
These are explicit controller operations, not inferred from filenames.
|
||||
|
||||
BLOCKED requires a nonblank reason and timezone-aware retry timestamp. The
|
||||
checkpoint records the interrupted active state; routing becomes inbox with an
|
||||
incremented retry count. Resume returns only to that state, only after the retry
|
||||
time, and clears the reason/timestamp while retaining retry count. The retry
|
||||
clock is UTC-aware. FAILED_TO_ANALYZE routes logically to review; OUTPUT_READY
|
||||
and HANDED_OFF route logically to outbox; active/reopened work routes to cases.
|
||||
The state and routing fields remain separate, and intake preserves a returning
|
||||
package's routing. Only fresh NEW packages are advanced to VALIDATED on admission.
|
||||
|
||||
This milestone does not create diagnoses, failure reports or Change Requests and
|
||||
does not deliver packages to coding agents. Outbox/review are initialized but
|
||||
physical output delivery belongs to the future phase controller. Detailed
|
||||
checkpoint payloads and artifact pointer updates are caller-supplied contracts;
|
||||
they receive structural validation, not diagnostic interpretation.
|
||||
|
||||
## Idempotence, locking and recovery
|
||||
|
||||
The content fingerprint hashes sorted internal paths and uncompressed bytes,
|
||||
with canonical JSON for the manifest. Archive compression, order and filename
|
||||
do not affect identity. Re-delivery of a known input generation and fingerprint
|
||||
does not create a new generation. Each checkpoint carries its source receipt,
|
||||
so this remains true when the package moves to a service without old local files.
|
||||
A seen generation with different content is quarantined as a conflict. Newer
|
||||
returning packages must preserve current protected files, checkpoint lineage and
|
||||
analysis revision; they cannot replace prior content or erase history. The input
|
||||
manifest remains authoritative for externally performed work on newer generations.
|
||||
|
||||
`advance` requires a stable nonblank `step_id`. Repeating it with the same target,
|
||||
details, artifact content and retry request returns the latest confirmed package
|
||||
without rerunning a write. Reusing the ID for a different request is an error.
|
||||
Artifact additions and the checkpoint are committed in the same package. External
|
||||
side effects are outside this transaction and need their own idempotence policy.
|
||||
|
||||
Nonblocking native OS locks use msvcrt on Windows and flock on POSIX. The coarse
|
||||
runtime file models one worker, guarded by a worker lock. Case locks additionally
|
||||
use case identity, independent of package identity. There is no timeout-based lock
|
||||
stealing and no waiting on blocked cases. The OS releases locks when a process
|
||||
exits, including a crash. Lock files remain in place to avoid inode replacement
|
||||
races; do not delete them while workers are active. Shared network filesystems
|
||||
require a separate distributed locking strategy and are not supported here.
|
||||
|
||||
Commit sequence:
|
||||
|
||||
1. Validate input, checkpoint lineage and requested transition under locks.
|
||||
2. Atomically write the coarse runtime summary.
|
||||
3. Publish the fully validated next generation via PackageWriter.
|
||||
4. Update the reconstructible local index, then clear runtime to null.
|
||||
5. For inbox admission, acknowledge the original only after publication, checking
|
||||
that its raw bytes still match the cached snapshot.
|
||||
|
||||
Before step 3, retry repeats only the unconfirmed operation. After step 3, the
|
||||
published package and embedded receipt prove completion even if steps 4 or 5
|
||||
never happened. A retry repairs local metadata and does not increment generation.
|
||||
File fsync and atomic visibility are provided, without claiming portable
|
||||
directory-metadata durability through power loss. Crashes may leave unpublished
|
||||
temporary files or incomplete quarantine directories; those are not treated as
|
||||
confirmed generations. Garbage collection is outside this milestone.
|
||||
|
||||
## Public API
|
||||
|
||||
```python
|
||||
from viano_bug_analysis.lifecycle import (
|
||||
Workspace, InboxProcessor, CheckpointStore, State, LifecycleError,
|
||||
)
|
||||
|
||||
workspace = Workspace("service-data")
|
||||
processor = InboxProcessor(workspace)
|
||||
results = processor.poll(max_jobs=100)
|
||||
|
||||
store = CheckpointStore(workspace)
|
||||
try:
|
||||
package = store.advance(
|
||||
"case-A", "stable-package", State.BUILDING_CASE,
|
||||
step_id="revision-1:start-case-building",
|
||||
details={"inputs_collected": []},
|
||||
)
|
||||
except LifecycleError as error:
|
||||
print(error.code, str(error))
|
||||
```
|
||||
|
||||
`poll` is a single pass, not a daemon. It validates bounded cached snapshots,
|
||||
admits valid cases, preserves invalid originals in quarantine with a structured
|
||||
error report, and continues to the next input after per-input errors. Oversized
|
||||
archives are moved directly to quarantine without creating an oversized cache
|
||||
copy. Due BLOCKED local cases are resumed once; future retries remain deferred.
|
||||
Deferred or busy inputs do not consume the `max_jobs` admission limit, so they
|
||||
cannot indefinitely crowd fresh work out of a poll. Such a pass may return more
|
||||
than `max_jobs` results and inspect all pending files. No diagnostic work runs.
|
||||
|
||||
Results are ACCEPTED, DUPLICATE, QUARANTINED, DEFERRED, BUSY, RESUMED or ERROR,
|
||||
with paths and optional error code/message. Conflicting/corrupt incoming records
|
||||
are quarantined; infrastructure failures leave inbox input pending. A changed
|
||||
producer file is not acknowledged. Workspace storage must be service-owned on a
|
||||
local filesystem supporting hard links, and producers must publish complete
|
||||
files atomically without modifying them during processing. Symlink inputs are
|
||||
rejected. This is not a hardened multi-tenant filesystem boundary.
|
||||
@@ -0,0 +1,10 @@
|
||||
from .checkpoints import CheckpointStore
|
||||
from .errors import LifecycleError, LockBusy
|
||||
from .job import Job
|
||||
from .locking import FileLock
|
||||
from .queue import InboxProcessor, IntakeResult
|
||||
from .state import State
|
||||
from .storage import Workspace
|
||||
|
||||
__all__ = ["CheckpointStore", "LifecycleError", "LockBusy", "Job", "FileLock",
|
||||
"InboxProcessor", "IntakeResult", "State", "Workspace"]
|
||||
@@ -0,0 +1,290 @@
|
||||
import hashlib
|
||||
import json
|
||||
from contextlib import contextmanager
|
||||
from datetime import datetime, timezone
|
||||
from importlib.resources import files
|
||||
from pathlib import Path
|
||||
|
||||
from jsonschema import Draft202012Validator
|
||||
from jsonschema.exceptions import ValidationError
|
||||
|
||||
from viano_bug_analysis.package import PackageReader, PackageReadError, PackageWriter, PackageWriteError
|
||||
from viano_bug_analysis.package.manifest import FORMAT_CHECKER, parse_manifest, safe_path
|
||||
from viano_bug_analysis.package.validator import CHUNK_SIZE
|
||||
|
||||
from .errors import LifecycleError
|
||||
from .job import Job, timestamp
|
||||
from .locking import FileLock
|
||||
from .state import ACTIVE, State, check_transition
|
||||
from .storage import Workspace, atomic_json, json_bytes, key
|
||||
|
||||
CHECKPOINT_POINTER = "lifecycle_checkpoint"
|
||||
CHECKPOINT_PREFIX = "history/checkpoints/"
|
||||
|
||||
|
||||
def _validator():
|
||||
schema = json.loads(files("viano_bug_analysis.schemas").joinpath("checkpoint-v1.schema.json").read_text())
|
||||
return Draft202012Validator(schema, format_checker=FORMAT_CHECKER)
|
||||
|
||||
|
||||
def fingerprint(reader: PackageReader) -> str:
|
||||
"""Content identity independent of archive filename, ordering and compression."""
|
||||
digest = hashlib.sha256()
|
||||
for info in sorted(reader._archive.infolist(), key=lambda item: item.filename):
|
||||
encoded = info.filename.encode("utf-8")
|
||||
digest.update(len(encoded).to_bytes(8, "big"))
|
||||
digest.update(encoded)
|
||||
if info.filename == "manifest.json":
|
||||
data = json_bytes(reader.manifest)
|
||||
digest.update(len(data).to_bytes(8, "big"))
|
||||
digest.update(data)
|
||||
else:
|
||||
digest.update(info.file_size.to_bytes(8, "big"))
|
||||
with reader._archive.open(info) as stream:
|
||||
while chunk := stream.read(CHUNK_SIZE):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def checkpoint_chain(reader: PackageReader) -> list[tuple[str, dict]]:
|
||||
validator = _validator()
|
||||
manifest = reader.manifest
|
||||
ref = manifest["artifacts"].get(CHECKPOINT_POINTER)
|
||||
result = []
|
||||
visited = set()
|
||||
upper_generation = int(manifest["package_generation"]) + 1
|
||||
upper_revision = int(manifest["analysis_revision"])
|
||||
steps = set()
|
||||
try:
|
||||
while ref is not None:
|
||||
if not safe_path(ref) or not ref.startswith(CHECKPOINT_PREFIX) or ref in visited:
|
||||
raise ValueError("Unsafe or cyclic checkpoint chain")
|
||||
visited.add(ref)
|
||||
if ref not in reader.files:
|
||||
raise ValueError("Missing checkpoint parent")
|
||||
with reader._archive.open(ref) as stream:
|
||||
raw = reader.validator._read_bounded(stream, min(reader.validator.limits.max_manifest_bytes, reader.validator.limits.max_file_bytes))
|
||||
data = parse_manifest(raw)
|
||||
validator.validate(data)
|
||||
State(data["state"])
|
||||
State(data["previous_state"])
|
||||
if data["resume_state"] is not None:
|
||||
if State(data["resume_state"]) not in ACTIVE or data["state"] != State.BLOCKED:
|
||||
raise ValueError("Invalid checkpoint resume state")
|
||||
elif data["state"] == State.BLOCKED:
|
||||
raise ValueError("Blocked checkpoint has no resume state")
|
||||
if data["case_id"] != manifest["case_id"] or data["package_id"] != manifest["package_id"]:
|
||||
raise ValueError("Checkpoint identity differs from manifest")
|
||||
generation = data["package_generation"]
|
||||
if generation >= upper_generation or data["source_generation"] != generation - 1:
|
||||
raise ValueError("Checkpoint generations are inconsistent")
|
||||
if data["analysis_revision"] > upper_revision or data["step_id"] in steps:
|
||||
raise ValueError("Checkpoint revision or step history is inconsistent")
|
||||
steps.add(data["step_id"])
|
||||
upper_generation = generation
|
||||
upper_revision = int(data["analysis_revision"])
|
||||
result.append((ref, data))
|
||||
ref = data["parent"]
|
||||
if result and result[0][1]["package_generation"] == manifest["package_generation"]:
|
||||
if (result[0][1]["state"] != manifest["state"] or result[0][1]["analysis_revision"] != manifest["analysis_revision"]
|
||||
or result[0][1]["routing"] != manifest["routing"]):
|
||||
raise ValueError("Latest checkpoint disagrees with manifest")
|
||||
return result
|
||||
except (ValueError, TypeError, KeyError, RecursionError, PackageReadError) as error:
|
||||
raise LifecycleError("CHECKPOINT_INVALID", str(error)) from error
|
||||
except ValidationError as error:
|
||||
raise LifecycleError("CHECKPOINT_INVALID", str(error)) from error
|
||||
|
||||
|
||||
class CheckpointStore:
|
||||
def __init__(self, workspace: Workspace, limits=None):
|
||||
self.workspace = workspace
|
||||
self.writer = PackageWriter(limits)
|
||||
self.limits = self.writer.validator.limits
|
||||
|
||||
@contextmanager
|
||||
def _locks(self, case_id: str):
|
||||
with FileLock(self.workspace.runtime / "locks" / "worker.lock"):
|
||||
with FileLock(self.workspace.runtime / "locks" / f"case-{key(case_id)}.lock"):
|
||||
yield
|
||||
|
||||
def latest(self, case_id: str, package_id: str) -> Path | None:
|
||||
directory = self.workspace.case_directory(case_id, package_id)
|
||||
candidates = {}
|
||||
for path in directory.glob("g*.zip"):
|
||||
if path.is_symlink():
|
||||
raise LifecycleError("UNSAFE_WORKSPACE", "Managed packages cannot be symlinks")
|
||||
try:
|
||||
with PackageReader(path, self.limits) as reader:
|
||||
manifest = reader.manifest
|
||||
generation = int(manifest["package_generation"])
|
||||
if manifest["case_id"] != case_id or manifest["package_id"] != package_id or path.name != f"g{generation:08d}.zip":
|
||||
raise LifecycleError("STORAGE_CONFLICT", "Managed package identity or generation is inconsistent")
|
||||
chain = checkpoint_chain(reader)
|
||||
if not chain or chain[0][1]["package_generation"] != generation:
|
||||
raise LifecycleError("CHECKPOINT_INVALID", "Managed generation has no matching checkpoint")
|
||||
candidates[generation] = path
|
||||
except PackageReadError as error:
|
||||
raise LifecycleError("STORAGE_PACKAGE_INVALID", str(error)) from error
|
||||
except LifecycleError as error:
|
||||
if error.code == "CHECKPOINT_INVALID":
|
||||
raise LifecycleError("STORAGE_CHECKPOINT_INVALID", str(error)) from error
|
||||
raise
|
||||
return candidates[max(candidates)] if candidates else None
|
||||
|
||||
def _index(self, path: Path, manifest: dict) -> None:
|
||||
atomic_json(path.parent / "current.json", {
|
||||
"package": path.name, "package_id": manifest["package_id"],
|
||||
"case_id": manifest["case_id"], "package_generation": manifest["package_generation"],
|
||||
})
|
||||
|
||||
def _commit(self, reader, step_id, request_hash, state, *, updates=None, details=None, additions=None, resume_state=None):
|
||||
source_manifest = reader.manifest
|
||||
generation = int(source_manifest["package_generation"]) + 1
|
||||
revision = (updates or {}).get("analysis_revision", source_manifest["analysis_revision"])
|
||||
record = {
|
||||
"format": "viano-lifecycle-checkpoint", "format_version": 1,
|
||||
"case_id": source_manifest["case_id"], "package_id": source_manifest["package_id"],
|
||||
"package_generation": generation, "analysis_revision": revision,
|
||||
"step_id": step_id, "request_hash": request_hash,
|
||||
"source_generation": source_manifest["package_generation"], "source_fingerprint": fingerprint(reader),
|
||||
"previous_state": source_manifest["state"], "state": state,
|
||||
"routing": source_manifest["routing"] | (updates or {}).get("routing", {}),
|
||||
"resume_state": resume_state,
|
||||
"parent": source_manifest["artifacts"].get(CHECKPOINT_POINTER),
|
||||
"created_at": timestamp(), "details": details or {},
|
||||
}
|
||||
ref = CHECKPOINT_PREFIX + key(step_id) + ".json"
|
||||
try:
|
||||
_validator().validate(record)
|
||||
except ValidationError as error:
|
||||
raise LifecycleError("CHECKPOINT_INVALID", str(error)) from error
|
||||
artifacts = dict((updates or {}).get("artifacts", {}))
|
||||
artifacts[CHECKPOINT_POINTER] = ref
|
||||
changes = dict(additions or {})
|
||||
if any(name.startswith(CHECKPOINT_PREFIX) for name in changes):
|
||||
raise LifecycleError("RESERVED_ARTIFACT", "Checkpoint namespace is service-managed")
|
||||
encoded = json_bytes(record)
|
||||
if len(encoded) > min(self.limits.max_manifest_bytes, self.limits.max_file_bytes):
|
||||
raise LifecycleError("CHECKPOINT_LIMIT", "Checkpoint exceeds configured JSON byte limit")
|
||||
changes[ref] = encoded
|
||||
destination = self.workspace.case_directory(record["case_id"], record["package_id"]) / f"g{generation:08d}.zip"
|
||||
chain = checkpoint_chain(reader)
|
||||
job = Job.from_manifest(source_manifest | {"analysis_revision": revision}, state,
|
||||
chain[0][1]["created_at"] if chain else None)
|
||||
atomic_json(self.workspace.runtime / "current_job.json", job.to_dict())
|
||||
result = self.writer.write(reader.path, destination, additions=changes,
|
||||
manifest_updates=dict(updates or {}) | {"state": state, "artifacts": artifacts})
|
||||
self._index(destination, result.manifest)
|
||||
atomic_json(self.workspace.runtime / "current_job.json", None)
|
||||
return destination
|
||||
|
||||
def intake(self, source: Path) -> tuple[str, Path]:
|
||||
try:
|
||||
with PackageReader(source, self.limits) as inbound:
|
||||
manifest = inbound.manifest
|
||||
with self._locks(manifest["case_id"]):
|
||||
latest = self.latest(manifest["case_id"], manifest["package_id"])
|
||||
digest = fingerprint(inbound)
|
||||
chain = checkpoint_chain(inbound)
|
||||
if latest is not None:
|
||||
with PackageReader(latest, self.limits) as current:
|
||||
old = current.manifest
|
||||
old_chain = checkpoint_chain(current)
|
||||
known = [(cp["source_generation"], cp["source_fingerprint"]) for _, cp in old_chain]
|
||||
known.append((old["package_generation"], fingerprint(current)))
|
||||
if (manifest["package_generation"], digest) in known:
|
||||
self._index(latest, old)
|
||||
atomic_json(self.workspace.runtime / "current_job.json", None)
|
||||
return "DUPLICATE", latest
|
||||
if manifest["package_generation"] <= old["package_generation"]:
|
||||
raise LifecycleError("GENERATION_CONFLICT", "Previously seen generation has different content")
|
||||
if manifest["analysis_revision"] < old["analysis_revision"]:
|
||||
raise LifecycleError("REVISION_REGRESSION", "Returning case lost its analysis revision")
|
||||
if not {ref for ref, _ in old_chain}.issubset({ref for ref, _ in chain}):
|
||||
raise LifecycleError("CHECKPOINT_HISTORY_LOST", "Returning case lost checkpoint lineage")
|
||||
self.writer._verify_preservation(current, source, {"analysis/current.json"})
|
||||
if manifest["state"] == State.BLOCKED:
|
||||
if not chain or chain[0][1]["state"] != State.BLOCKED:
|
||||
raise LifecycleError("CHECKPOINT_INVALID", "Blocked input requires its resume checkpoint")
|
||||
if not manifest["routing"]["reason"] or manifest["routing"]["retry_not_before"] is None:
|
||||
raise LifecycleError("INVALID_RETRY", "Blocked input requires a reason and retry timestamp")
|
||||
if not self.retry_due(manifest):
|
||||
return "DEFERRED", source
|
||||
state = State.VALIDATED if manifest["state"] == State.NEW else manifest["state"]
|
||||
step_id = f"intake:{int(manifest['package_generation'])}:{digest}"
|
||||
request_hash = key(step_id)
|
||||
path = self._commit(inbound, step_id, request_hash, state,
|
||||
resume_state=chain[0][1]["resume_state"] if state == State.BLOCKED else None,
|
||||
details={"operation": "intake"},
|
||||
updates={"routing": {"destination": "cases"}} if manifest["state"] == State.NEW else {})
|
||||
return "ACCEPTED", path
|
||||
except (PackageReadError, PackageWriteError) as error:
|
||||
if isinstance(error, PackageWriteError) and error.result.issues[0].code == "PRESERVATION_FAILED":
|
||||
raise LifecycleError("PRESERVATION_FAILED", str(error)) from error
|
||||
raise LifecycleError("PACKAGE_ERROR", str(error)) from error
|
||||
except (OSError, ValueError, TypeError, RecursionError) as error:
|
||||
raise LifecycleError("LIFECYCLE_ERROR", str(error)) from error
|
||||
|
||||
@staticmethod
|
||||
def retry_due(manifest: dict, now: datetime | None = None) -> bool:
|
||||
value = manifest["routing"]["retry_not_before"]
|
||||
return value is None or datetime.fromisoformat(value.upper().replace("Z", "+00:00")) <= (now or datetime.now(timezone.utc))
|
||||
|
||||
def advance(self, case_id: str, package_id: str, target: str, *, step_id: str,
|
||||
reason: str | None = None, retry_not_before: str | None = None,
|
||||
details: dict | None = None, additions: dict[str, bytes] | None = None,
|
||||
artifacts: dict[str, str | None] | None = None) -> Path:
|
||||
try:
|
||||
if not isinstance(step_id, str) or not step_id.strip():
|
||||
raise LifecycleError("INVALID_STEP", "A nonblank stable step_id is required")
|
||||
if details is not None and not isinstance(details, dict):
|
||||
raise LifecycleError("CHECKPOINT_INVALID", "Checkpoint details must be a JSON object")
|
||||
if CHECKPOINT_POINTER in (artifacts or {}):
|
||||
raise LifecycleError("RESERVED_ARTIFACT", "Lifecycle checkpoint pointer is service-managed")
|
||||
payload = {"target": target, "reason": reason, "retry_not_before": retry_not_before,
|
||||
"details": details or {}, "artifacts": artifacts or {},
|
||||
"additions": {name: hashlib.sha256(data).hexdigest() for name, data in (additions or {}).items()}}
|
||||
request_hash = hashlib.sha256(json_bytes(payload)).hexdigest()
|
||||
with self._locks(case_id):
|
||||
latest = self.latest(case_id, package_id)
|
||||
if latest is None:
|
||||
raise LifecycleError("CASE_NOT_FOUND", "Case has not been admitted")
|
||||
with PackageReader(latest, self.limits) as reader:
|
||||
manifest = reader.manifest
|
||||
chain = checkpoint_chain(reader)
|
||||
for _, checkpoint in chain:
|
||||
if checkpoint["step_id"] == step_id:
|
||||
if checkpoint["request_hash"] != request_hash:
|
||||
raise LifecycleError("STEP_CONFLICT", "step_id already used for different work")
|
||||
self._index(latest, manifest)
|
||||
atomic_json(self.workspace.runtime / "current_job.json", None)
|
||||
return latest
|
||||
resume = chain[0][1]["resume_state"] if manifest["state"] == State.BLOCKED else None
|
||||
check_transition(manifest["state"], target, resume)
|
||||
if manifest["state"] == State.BLOCKED and not self.retry_due(manifest):
|
||||
raise LifecycleError("RETRY_NOT_DUE", "Retry timestamp has not been reached")
|
||||
routing = {"destination": "cases", "reason": None, "retry_not_before": None}
|
||||
if target == State.BLOCKED:
|
||||
if not isinstance(reason, str) or not reason.strip() or not isinstance(retry_not_before, str) or not FORMAT_CHECKER.conforms(retry_not_before, "date-time"):
|
||||
raise LifecycleError("INVALID_RETRY", "BLOCKED needs a reason and timezone-aware retry timestamp")
|
||||
routing.update(destination="inbox", reason=reason, retry_not_before=retry_not_before,
|
||||
retry_count=int(manifest["routing"]["retry_count"]) + 1)
|
||||
resume = manifest["state"]
|
||||
else:
|
||||
resume = None
|
||||
if target == State.FAILED_TO_ANALYZE:
|
||||
routing.update(destination="review", reason=reason)
|
||||
elif target in {State.OUTPUT_READY, State.HANDED_OFF}:
|
||||
routing["destination"] = "outbox"
|
||||
revision = int(manifest["analysis_revision"])
|
||||
if target == State.REOPENED or (manifest["state"] == State.VALIDATED and target == State.BUILDING_CASE and revision == 0):
|
||||
revision += 1
|
||||
history = {"analysis_revisions": max(int(manifest["history"]["analysis_revisions"]), revision)}
|
||||
return self._commit(reader, step_id, request_hash, target, details=details, additions=additions,
|
||||
resume_state=resume, updates={"routing": routing, "analysis_revision": revision,
|
||||
"history": history, "artifacts": artifacts or {}})
|
||||
except (PackageReadError, PackageWriteError) as error:
|
||||
raise LifecycleError("PACKAGE_ERROR", str(error)) from error
|
||||
except (OSError, ValueError, TypeError, RecursionError) as error:
|
||||
raise LifecycleError("LIFECYCLE_ERROR", str(error)) from error
|
||||
@@ -0,0 +1,11 @@
|
||||
class LifecycleError(Exception):
|
||||
"""Controlled storage, checkpoint or lifecycle rejection."""
|
||||
|
||||
def __init__(self, code: str, message: str):
|
||||
self.code = code
|
||||
super().__init__(message)
|
||||
|
||||
|
||||
class LockBusy(LifecycleError):
|
||||
def __init__(self):
|
||||
super().__init__("LOCK_BUSY", "Another worker holds this lock")
|
||||
@@ -0,0 +1,25 @@
|
||||
from dataclasses import asdict, dataclass
|
||||
from datetime import datetime, timezone
|
||||
|
||||
|
||||
def timestamp() -> str:
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Job:
|
||||
case_id: str
|
||||
package_id: str
|
||||
package_generation: int
|
||||
analysis_revision: int
|
||||
stage: str
|
||||
started_at: str
|
||||
last_checkpoint_at: str | None = None
|
||||
|
||||
@classmethod
|
||||
def from_manifest(cls, manifest: dict, stage: str, last_checkpoint_at: str | None = None):
|
||||
return cls(manifest["case_id"], manifest["package_id"], int(manifest["package_generation"]),
|
||||
int(manifest["analysis_revision"]), stage, timestamp(), last_checkpoint_at)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return asdict(self)
|
||||
@@ -0,0 +1,49 @@
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from .errors import LifecycleError, LockBusy
|
||||
|
||||
|
||||
class FileLock:
|
||||
"""Nonblocking OS lock; automatically released when the process exits.
|
||||
|
||||
Lock files persist and must not be removed while workers are running.
|
||||
"""
|
||||
|
||||
def __init__(self, path: Path):
|
||||
self.path = path
|
||||
self._stream = None
|
||||
|
||||
def __enter__(self):
|
||||
if self._stream is not None:
|
||||
raise RuntimeError("Lock already held")
|
||||
try:
|
||||
if self.path.is_symlink():
|
||||
raise LifecycleError("UNSAFE_LOCK", "Lock path is a symlink")
|
||||
self._stream = open(self.path, "a+b")
|
||||
if os.fstat(self._stream.fileno()).st_size == 0:
|
||||
self._stream.write(b"\0")
|
||||
self._stream.flush()
|
||||
self._stream.seek(0)
|
||||
try:
|
||||
if os.name == "nt":
|
||||
import msvcrt
|
||||
msvcrt.locking(self._stream.fileno(), msvcrt.LK_NBLCK, 1)
|
||||
else:
|
||||
import fcntl
|
||||
fcntl.flock(self._stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
except OSError as error:
|
||||
raise LockBusy() from error
|
||||
return self
|
||||
except (OSError, LifecycleError) as error:
|
||||
if self._stream is not None:
|
||||
self._stream.close()
|
||||
self._stream = None
|
||||
if isinstance(error, LifecycleError):
|
||||
raise
|
||||
raise LifecycleError("LOCK_ERROR", str(error)) from error
|
||||
|
||||
def __exit__(self, *args):
|
||||
if self._stream is not None:
|
||||
self._stream.close()
|
||||
self._stream = None
|
||||
@@ -0,0 +1,161 @@
|
||||
import hashlib
|
||||
import os
|
||||
import tempfile
|
||||
import uuid
|
||||
from dataclasses import asdict, dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from viano_bug_analysis.package import PackageReader, PackageReadError, PackageValidator
|
||||
from viano_bug_analysis.package.validator import CHUNK_SIZE
|
||||
|
||||
from .checkpoints import CheckpointStore, checkpoint_chain
|
||||
from .errors import LifecycleError, LockBusy
|
||||
from .state import State
|
||||
from .storage import Workspace, atomic_json
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class IntakeResult:
|
||||
status: str
|
||||
source: Path
|
||||
package: Path | None = None
|
||||
code: str | None = None
|
||||
message: str | None = None
|
||||
|
||||
|
||||
class InboxProcessor:
|
||||
"""One inbox pass, no analysis execution or waits for external conditions."""
|
||||
|
||||
def __init__(self, workspace: Workspace, limits=None):
|
||||
self.workspace = workspace
|
||||
self.store = CheckpointStore(workspace, limits)
|
||||
self.validator = PackageValidator(self.store.limits)
|
||||
|
||||
def _snapshot(self, source: Path) -> tuple[Path, str]:
|
||||
temporary = None
|
||||
try:
|
||||
digest = hashlib.sha256()
|
||||
total = 0
|
||||
with tempfile.NamedTemporaryFile(dir=self.workspace.cache, prefix=".snapshot-", delete=False) as copied:
|
||||
temporary = Path(copied.name)
|
||||
with source.open("rb") as original:
|
||||
while chunk := original.read(CHUNK_SIZE):
|
||||
total += len(chunk)
|
||||
if total > self.store.limits.max_archive_bytes:
|
||||
raise LifecycleError("ARCHIVE_LIMIT", "Input exceeds configured archive byte limit")
|
||||
copied.write(chunk)
|
||||
digest.update(chunk)
|
||||
copied.flush()
|
||||
os.fsync(copied.fileno())
|
||||
raw_hash = digest.hexdigest()
|
||||
destination = self.workspace.cache / f"{raw_hash}.zip"
|
||||
if destination.is_symlink():
|
||||
raise LifecycleError("CACHE_CONFLICT", "Cached original is a symlink")
|
||||
try:
|
||||
os.link(temporary, destination)
|
||||
except FileExistsError:
|
||||
if self._raw_hash(destination) != raw_hash:
|
||||
raise LifecycleError("CACHE_CONFLICT", "Cached original no longer matches its hash")
|
||||
return destination, raw_hash
|
||||
except OSError as error:
|
||||
raise LifecycleError("STORAGE_ERROR", str(error)) from error
|
||||
finally:
|
||||
if temporary is not None:
|
||||
temporary.unlink(missing_ok=True)
|
||||
|
||||
@staticmethod
|
||||
def _raw_hash(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as stream:
|
||||
while chunk := stream.read(CHUNK_SIZE):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
def _acknowledge(self, source: Path, raw_hash: str) -> None:
|
||||
if source.is_symlink() or self._raw_hash(source) != raw_hash:
|
||||
raise LifecycleError("SOURCE_CHANGED", "Producer changed inbox input during processing")
|
||||
source.unlink()
|
||||
|
||||
def _quarantine(self, source: Path, issues: list[dict], raw_hash: str | None = None) -> IntakeResult:
|
||||
directory = self.workspace.quarantine / uuid.uuid4().hex
|
||||
directory.mkdir()
|
||||
atomic_json(directory / "validation_error.json", {
|
||||
"format": "viano-validation-error", "format_version": 1,
|
||||
"source_name": source.name, "issues": issues,
|
||||
})
|
||||
# Same workspace filesystem: preserve the exact archive without extraction.
|
||||
if source.is_symlink() or (raw_hash is not None and self._raw_hash(source) != raw_hash):
|
||||
raise LifecycleError("SOURCE_CHANGED", "Input changed before quarantine")
|
||||
os.replace(source, directory / "package.zip")
|
||||
return IntakeResult("QUARANTINED", source, directory / "package.zip", issues[0]["code"], issues[0]["message"])
|
||||
|
||||
def process(self, source: Path) -> IntakeResult:
|
||||
source = Path(source)
|
||||
try:
|
||||
if source.parent.resolve() != self.workspace.inbox or source.is_symlink() or not source.is_file():
|
||||
raise LifecycleError("UNSAFE_INPUT", "Expected a regular file directly in the inbox")
|
||||
try:
|
||||
snapshot, raw_hash = self._snapshot(source)
|
||||
except LifecycleError as error:
|
||||
if error.code == "ARCHIVE_LIMIT":
|
||||
return self._quarantine(source, [{"code": error.code, "message": str(error), "path": None}])
|
||||
raise
|
||||
result = self.validator.validate(snapshot)
|
||||
if not result.valid:
|
||||
return self._quarantine(source, [asdict(issue) for issue in result.issues], raw_hash)
|
||||
try:
|
||||
status, managed = self.store.intake(snapshot)
|
||||
except LockBusy:
|
||||
return IntakeResult("BUSY", source, code="LOCK_BUSY")
|
||||
except LifecycleError as error:
|
||||
if error.code in {"GENERATION_CONFLICT", "REVISION_REGRESSION", "CHECKPOINT_HISTORY_LOST", "CHECKPOINT_INVALID", "PRESERVATION_FAILED", "INVALID_RETRY"}:
|
||||
return self._quarantine(source, [{"code": error.code, "message": str(error), "path": None}], raw_hash)
|
||||
raise
|
||||
if status != "DEFERRED":
|
||||
self._acknowledge(source, raw_hash)
|
||||
return IntakeResult(status, source, managed)
|
||||
except (LifecycleError, OSError, PackageReadError) as error:
|
||||
return IntakeResult("ERROR", source, code=getattr(error, "code", "STORAGE_ERROR"), message=str(error))
|
||||
|
||||
def poll(self, *, max_jobs: int = 100) -> list[IntakeResult]:
|
||||
if isinstance(max_jobs, bool) or not isinstance(max_jobs, int) or max_jobs <= 0:
|
||||
raise ValueError("max_jobs must be a positive integer")
|
||||
outcomes = []
|
||||
completed = 0
|
||||
for source in sorted(self.workspace.inbox.iterdir()):
|
||||
if completed >= max_jobs:
|
||||
break
|
||||
if source.is_file() or source.is_symlink():
|
||||
result = self.process(source)
|
||||
outcomes.append(result)
|
||||
completed += result.status not in {"DEFERRED", "BUSY"}
|
||||
# Waiting cases do not occupy a worker. Resume only due local checkpoints.
|
||||
identities = set()
|
||||
for path in self.workspace.cases.glob("*/*/g*.zip"):
|
||||
try:
|
||||
with PackageReader(path, self.store.limits) as reader:
|
||||
manifest = reader.manifest
|
||||
identities.add((manifest["case_id"], manifest["package_id"]))
|
||||
except (PackageReadError, OSError) as error:
|
||||
outcomes.append(IntakeResult("ERROR", path, code="PACKAGE_ERROR", message=str(error)))
|
||||
for case_id, package_id in sorted(identities):
|
||||
if completed >= max_jobs:
|
||||
break
|
||||
path = self.workspace.cases
|
||||
try:
|
||||
path = self.store.latest(case_id, package_id)
|
||||
with PackageReader(path, self.store.limits) as reader:
|
||||
manifest = reader.manifest
|
||||
if manifest["state"] != State.BLOCKED or not self.store.retry_due(manifest):
|
||||
continue
|
||||
resume = checkpoint_chain(reader)[0][1]["resume_state"]
|
||||
resumed = self.store.advance(case_id, package_id, resume,
|
||||
step_id=f"retry:{int(manifest['package_generation'])}")
|
||||
outcomes.append(IntakeResult("RESUMED", path, resumed))
|
||||
completed += 1
|
||||
except LockBusy:
|
||||
outcomes.append(IntakeResult("BUSY", path, code="LOCK_BUSY"))
|
||||
except (LifecycleError, PackageReadError, OSError) as error:
|
||||
outcomes.append(IntakeResult("ERROR", path, code=getattr(error, "code", "STORAGE_ERROR"), message=str(error)))
|
||||
completed += 1
|
||||
return outcomes
|
||||
@@ -0,0 +1,40 @@
|
||||
from enum import StrEnum
|
||||
|
||||
from .errors import LifecycleError
|
||||
|
||||
|
||||
class State(StrEnum):
|
||||
NEW = "NEW"
|
||||
VALIDATED = "VALIDATED"
|
||||
BUILDING_CASE = "BUILDING_CASE"
|
||||
INTERPRETING_CASE = "INTERPRETING_CASE"
|
||||
GENERATING_HYPOTHESES = "GENERATING_HYPOTHESES"
|
||||
INVESTIGATING_HYPOTHESES = "INVESTIGATING_HYPOTHESES"
|
||||
BUILDING_DIAGNOSIS = "BUILDING_DIAGNOSIS"
|
||||
OUTPUT_READY = "OUTPUT_READY"
|
||||
HANDED_OFF = "HANDED_OFF"
|
||||
BLOCKED = "BLOCKED"
|
||||
FAILED_TO_ANALYZE = "FAILED_TO_ANALYZE"
|
||||
REOPENED = "REOPENED"
|
||||
|
||||
|
||||
ORDER = [State.NEW, State.VALIDATED, State.BUILDING_CASE, State.INTERPRETING_CASE,
|
||||
State.GENERATING_HYPOTHESES, State.INVESTIGATING_HYPOTHESES,
|
||||
State.BUILDING_DIAGNOSIS, State.OUTPUT_READY, State.HANDED_OFF]
|
||||
ACTIVE = frozenset(ORDER[1:-2]) | {State.REOPENED}
|
||||
|
||||
|
||||
def check_transition(previous: str, target: str, resume_state: str | None = None) -> None:
|
||||
try:
|
||||
old, new = State(previous), State(target)
|
||||
except ValueError as error:
|
||||
raise LifecycleError("INVALID_STATE", str(error)) from error
|
||||
permitted = old == new and old in ACTIVE
|
||||
if old in ORDER[:-1]:
|
||||
permitted |= new == ORDER[ORDER.index(old) + 1]
|
||||
permitted |= old in ACTIVE and new in {State.BLOCKED, State.FAILED_TO_ANALYZE}
|
||||
permitted |= old == State.REOPENED and new == State.BUILDING_CASE
|
||||
permitted |= old in {State.HANDED_OFF, State.FAILED_TO_ANALYZE} and new == State.REOPENED
|
||||
permitted |= old == State.BLOCKED and new == resume_state and new in ACTIVE
|
||||
if not permitted:
|
||||
raise LifecycleError("INVALID_TRANSITION", f"Cannot transition {old} to {new}")
|
||||
@@ -0,0 +1,58 @@
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
from .errors import LifecycleError
|
||||
|
||||
|
||||
def key(value: str) -> str:
|
||||
return hashlib.sha256(value.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def json_bytes(value: object) -> bytes:
|
||||
return json.dumps(value, sort_keys=True, ensure_ascii=False, allow_nan=False).encode("utf-8")
|
||||
|
||||
|
||||
def atomic_json(path: Path, value: object) -> None:
|
||||
temporary = None
|
||||
try:
|
||||
with tempfile.NamedTemporaryFile(dir=path.parent, prefix=".json-", delete=False) as stream:
|
||||
temporary = Path(stream.name)
|
||||
stream.write(json_bytes(value))
|
||||
stream.flush()
|
||||
os.fsync(stream.fileno())
|
||||
os.replace(temporary, path)
|
||||
except (OSError, ValueError, TypeError, RecursionError) as error:
|
||||
raise LifecycleError("STORAGE_ERROR", str(error)) from error
|
||||
finally:
|
||||
if temporary is not None:
|
||||
temporary.unlink(missing_ok=True)
|
||||
|
||||
|
||||
class Workspace:
|
||||
"""Service-owned directories. Case identifiers never become path segments."""
|
||||
|
||||
def __init__(self, root: str | os.PathLike[str]):
|
||||
self.root = Path(root).resolve()
|
||||
try:
|
||||
for name in ("inbox", "quarantine", "cache", "cases", "outbox", "review", "runtime"):
|
||||
path = self.root / name
|
||||
if path.is_symlink():
|
||||
raise LifecycleError("UNSAFE_WORKSPACE", "Workspace directories cannot be symlinks")
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
setattr(self, name, path)
|
||||
lock_directory = self.runtime / "locks"
|
||||
if lock_directory.is_symlink():
|
||||
raise LifecycleError("UNSAFE_WORKSPACE", "Lock directory cannot be a symlink")
|
||||
lock_directory.mkdir(exist_ok=True)
|
||||
except OSError as error:
|
||||
raise LifecycleError("STORAGE_ERROR", str(error)) from error
|
||||
|
||||
def case_directory(self, case_id: str, package_id: str) -> Path:
|
||||
path = self.cases / key(case_id) / key(package_id)
|
||||
if path.parent.is_symlink() or path.is_symlink():
|
||||
raise LifecycleError("UNSAFE_WORKSPACE", "Case storage cannot be symlinked")
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
return path
|
||||
@@ -0,0 +1,24 @@
|
||||
{
|
||||
"$schema": "https://json-schema.org/draft/2020-12/schema",
|
||||
"type": "object",
|
||||
"required": ["format", "format_version", "case_id", "package_id", "package_generation", "analysis_revision", "step_id", "request_hash", "source_generation", "source_fingerprint", "previous_state", "state", "routing", "resume_state", "parent", "created_at", "details"],
|
||||
"properties": {
|
||||
"format": {"const": "viano-lifecycle-checkpoint"},
|
||||
"format_version": {"type": "integer", "const": 1},
|
||||
"case_id": {"type": "string", "minLength": 1},
|
||||
"package_id": {"type": "string", "minLength": 1},
|
||||
"package_generation": {"type": "integer", "minimum": 1},
|
||||
"analysis_revision": {"type": "integer", "minimum": 0},
|
||||
"step_id": {"type": "string", "minLength": 1, "pattern": "\\S"},
|
||||
"request_hash": {"type": "string", "pattern": "^[0-9a-f]{64}$"},
|
||||
"source_generation": {"type": "integer", "minimum": 0},
|
||||
"source_fingerprint": {"type": "string", "pattern": "^[0-9a-f]{64}$"},
|
||||
"previous_state": {"type": "string"},
|
||||
"state": {"type": "string"},
|
||||
"routing": {"type": "object", "required": ["destination", "reason", "retry_count", "retry_not_before"]},
|
||||
"resume_state": {"type": ["string", "null"]},
|
||||
"parent": {"type": ["string", "null"]},
|
||||
"created_at": {"type": "string", "format": "date-time"},
|
||||
"details": {"type": "object"}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,227 @@
|
||||
import json
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
from viano_bug_analysis.lifecycle import CheckpointStore, LifecycleError, State, Workspace
|
||||
from viano_bug_analysis.lifecycle.checkpoints import CHECKPOINT_POINTER, checkpoint_chain
|
||||
from viano_bug_analysis.package import PackageReader, PackageValidator, PackageWriter
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admitted(make_package, tmp_path):
|
||||
workspace = Workspace(tmp_path / "service")
|
||||
store = CheckpointStore(workspace)
|
||||
status, path = store.intake(make_package(entries={"unknown": b"keep"}))
|
||||
assert status == "ACCEPTED"
|
||||
return workspace, store, path
|
||||
|
||||
|
||||
def test_checkpoint_inside_package_and_runtime_is_coarse(admitted):
|
||||
workspace, store, path = admitted
|
||||
with PackageReader(path) as reader:
|
||||
manifest = reader.manifest
|
||||
assert manifest["state"] == State.VALIDATED
|
||||
assert manifest["package_generation"] == 1
|
||||
assert manifest["analysis_revision"] == 0
|
||||
chain = checkpoint_chain(reader)
|
||||
assert chain[0][1]["previous_state"] == State.NEW
|
||||
assert chain[0][1]["source_generation"] == 0
|
||||
assert chain[0][1]["details"] == {"operation": "intake"}
|
||||
assert reader.read_bytes("unknown") == b"keep"
|
||||
assert json.loads((workspace.runtime / "current_job.json").read_text()) is None
|
||||
|
||||
|
||||
def test_progress_idempotence_and_conflicting_step(admitted):
|
||||
_, store, _ = admitted
|
||||
path = store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start")
|
||||
with PackageReader(path) as reader:
|
||||
assert reader.manifest["package_generation"] == 2
|
||||
assert reader.manifest["analysis_revision"] == 1
|
||||
assert store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start") == path
|
||||
with pytest.raises(LifecycleError) as error:
|
||||
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start", details={"different": True})
|
||||
assert error.value.code == "STEP_CONFLICT"
|
||||
next_path = store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-progress", details={"collected": ["input-A"]}, additions={"case/inputs/a": b"generic"})
|
||||
assert store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start") == next_path
|
||||
with PackageReader(next_path) as reader:
|
||||
assert len(checkpoint_chain(reader)) == 3
|
||||
assert reader.manifest["analysis_revision"] == 1
|
||||
assert reader.read_bytes("case/inputs/a") == b"generic"
|
||||
|
||||
|
||||
def test_full_lifecycle_and_reopen(admitted):
|
||||
_, store, _ = admitted
|
||||
states = [State.BUILDING_CASE, State.INTERPRETING_CASE, State.GENERATING_HYPOTHESES, State.INVESTIGATING_HYPOTHESES, State.BUILDING_DIAGNOSIS, State.OUTPUT_READY, State.HANDED_OFF, State.REOPENED, State.BUILDING_CASE]
|
||||
for index, state in enumerate(states):
|
||||
path = store.advance("case-A", "stable-package", state, step_id=f"step-{index}")
|
||||
with PackageReader(path) as reader:
|
||||
assert reader.manifest["analysis_revision"] == 2
|
||||
assert reader.manifest["history"]["analysis_revisions"] == 2
|
||||
assert reader.manifest["package_generation"] == 10
|
||||
assert len(checkpoint_chain(reader)) == 10
|
||||
assert PackageValidator().validate(path).valid
|
||||
|
||||
|
||||
def test_failure_routes_review_then_reopen(admitted):
|
||||
_, store, _ = admitted
|
||||
path = store.advance("case-A", "stable-package", State.FAILED_TO_ANALYZE, step_id="failure", reason="No available analysis")
|
||||
with PackageReader(path) as reader:
|
||||
assert reader.manifest["routing"]["destination"] == "review"
|
||||
path = store.advance("case-A", "stable-package", State.REOPENED, step_id="review-return")
|
||||
with PackageReader(path) as reader:
|
||||
assert reader.manifest["analysis_revision"] == 1
|
||||
|
||||
|
||||
def test_block_retry_and_resume(admitted):
|
||||
_, store, _ = admitted
|
||||
future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()
|
||||
path = store.advance("case-A", "stable-package", State.BLOCKED, step_id="blocked", reason="External dependency", retry_not_before=future)
|
||||
with PackageReader(path) as reader:
|
||||
assert reader.manifest["routing"]["retry_count"] == 1
|
||||
assert checkpoint_chain(reader)[0][1]["resume_state"] == State.VALIDATED
|
||||
with pytest.raises(LifecycleError) as error:
|
||||
store.advance("case-A", "stable-package", State.VALIDATED, step_id="resume")
|
||||
assert error.value.code == "RETRY_NOT_DUE"
|
||||
with pytest.raises(LifecycleError):
|
||||
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="wrong-resume")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("retry", [None, "not-a-date", "2026-01-01T00:00:00", "2026-02-30T00:00:00Z"])
|
||||
def test_invalid_retry_rejected_before_publication(retry, admitted):
|
||||
_, store, path = admitted
|
||||
with pytest.raises(LifecycleError):
|
||||
store.advance("case-A", "stable-package", State.BLOCKED, step_id="bad", reason="External dependency", retry_not_before=retry)
|
||||
assert store.latest("case-A", "stable-package") == path
|
||||
|
||||
|
||||
def test_restart_uses_package_not_current_index(admitted):
|
||||
workspace, _, path = admitted
|
||||
(path.parent / "current.json").write_text("broken index")
|
||||
(workspace.runtime / "current_job.json").write_text("broken runtime summary")
|
||||
restarted = CheckpointStore(workspace)
|
||||
assert restarted.latest("case-A", "stable-package") == path
|
||||
assert restarted.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start").exists()
|
||||
|
||||
|
||||
def test_crash_after_package_publication_before_index(admitted, monkeypatch):
|
||||
workspace, store, original = admitted
|
||||
def crash(*args):
|
||||
raise LifecycleError("SIMULATED_CRASH", "after publication")
|
||||
monkeypatch.setattr(store, "_index", crash)
|
||||
with pytest.raises(LifecycleError):
|
||||
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start")
|
||||
summary = json.loads((workspace.runtime / "current_job.json").read_text())
|
||||
assert set(summary) == {"case_id", "package_id", "package_generation", "analysis_revision", "stage", "started_at", "last_checkpoint_at"}
|
||||
restarted = CheckpointStore(workspace)
|
||||
path = restarted.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start")
|
||||
assert path != original
|
||||
assert len(list(path.parent.glob("g*.zip"))) == 2
|
||||
assert json.loads((workspace.runtime / "current_job.json").read_text()) is None
|
||||
|
||||
|
||||
def test_crash_before_publication_repeats_only_pending_step(admitted, monkeypatch):
|
||||
workspace, store, original = admitted
|
||||
def crash(*args, **kwargs):
|
||||
raise LifecycleError("SIMULATED_CRASH", "before publication")
|
||||
monkeypatch.setattr(store.writer, "write", crash)
|
||||
with pytest.raises(LifecycleError):
|
||||
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start")
|
||||
restarted = CheckpointStore(workspace)
|
||||
assert restarted.latest("case-A", "stable-package") == original
|
||||
path = restarted.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start")
|
||||
with PackageReader(path) as reader:
|
||||
assert reader.manifest["package_generation"] == 2
|
||||
|
||||
|
||||
def test_import_portable_package_without_service_memory(admitted, tmp_path):
|
||||
_, store, _ = admitted
|
||||
original = store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start", details={"progress": "retained"})
|
||||
other = CheckpointStore(Workspace(tmp_path / "other-service"))
|
||||
status, imported = other.intake(original)
|
||||
assert status == "ACCEPTED"
|
||||
with PackageReader(imported) as reader:
|
||||
assert checkpoint_chain(reader)[1][1]["details"] == {"progress": "retained"}
|
||||
# Previous work remains idempotent even without the previous service's files.
|
||||
assert other.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start", details={"progress": "retained"}) == imported
|
||||
|
||||
|
||||
def test_newer_return_preserves_local_history(admitted, tmp_path):
|
||||
_, store, original = admitted
|
||||
returned = tmp_path / "returned"
|
||||
PackageWriter().write(original, returned, additions={"feedback/new": b"new feedback"})
|
||||
status, path = store.intake(returned)
|
||||
assert status == "ACCEPTED"
|
||||
with PackageReader(path) as reader:
|
||||
assert reader.manifest["package_generation"] == 3
|
||||
assert reader.read_bytes("feedback/new") == b"new feedback"
|
||||
|
||||
|
||||
def test_invalid_details_not_published(admitted):
|
||||
_, store, original = admitted
|
||||
with pytest.raises(LifecycleError):
|
||||
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="invalid", details=["wrong"])
|
||||
assert store.latest("case-A", "stable-package") == original
|
||||
|
||||
|
||||
def test_corrupt_managed_package_fails_without_rollback(admitted):
|
||||
_, store, path = admitted
|
||||
newer = store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start")
|
||||
newer.write_bytes(b"broken")
|
||||
with pytest.raises(LifecycleError) as error:
|
||||
store.latest("case-A", "stable-package")
|
||||
assert error.value.code == "STORAGE_PACKAGE_INVALID"
|
||||
assert path.exists()
|
||||
|
||||
|
||||
def test_real_process_crash_after_publication(admitted):
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
workspace, _, _ = admitted
|
||||
script = "from viano_bug_analysis.lifecycle import Workspace,CheckpointStore,State; import os,sys; s=CheckpointStore(Workspace(sys.argv[1])); s._index=lambda *args: os._exit(23); s.advance('case-A','stable-package',State.BUILDING_CASE,step_id='process-build')"
|
||||
env = dict(os.environ, PYTHONPATH=str(Path(__file__).resolve().parents[1] / "src"))
|
||||
process = subprocess.run([sys.executable, "-c", script, str(workspace.root)], env=env, capture_output=True, timeout=15)
|
||||
assert process.returncode == 23, process.stderr.decode()
|
||||
restarted = CheckpointStore(workspace)
|
||||
path = restarted.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="process-build")
|
||||
with PackageReader(path) as reader:
|
||||
assert reader.manifest["package_generation"] == 2
|
||||
assert len(checkpoint_chain(reader)) == 2
|
||||
assert len(list(path.parent.glob("g*.zip"))) == 2
|
||||
|
||||
|
||||
def test_reserved_checkpoint_artifacts_rejected(admitted):
|
||||
_, store, original = admitted
|
||||
with pytest.raises(LifecycleError) as error:
|
||||
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="forbidden", artifacts={CHECKPOINT_POINTER: None})
|
||||
assert error.value.code == "RESERVED_ARTIFACT"
|
||||
with pytest.raises(LifecycleError):
|
||||
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="forbidden", additions={"history/checkpoints/user.json": b"{}"})
|
||||
assert store.latest("case-A", "stable-package") == original
|
||||
|
||||
|
||||
def test_invalid_block_request_types(admitted):
|
||||
_, store, _ = admitted
|
||||
for reason, retry in [(3, "2026-01-01T00:00:00Z"), ("reason", 4)]:
|
||||
with pytest.raises(LifecycleError) as error:
|
||||
store.advance("case-A", "stable-package", State.BLOCKED, step_id="bad", reason=reason, retry_not_before=retry)
|
||||
assert error.value.code == "INVALID_RETRY"
|
||||
|
||||
|
||||
def test_oversized_checkpoint_not_published(make_package, tmp_path):
|
||||
from viano_bug_analysis.package import PackageLimits
|
||||
store = CheckpointStore(Workspace(tmp_path / "service"), PackageLimits(max_manifest_bytes=4000))
|
||||
_, original = store.intake(make_package())
|
||||
with pytest.raises(LifecycleError) as error:
|
||||
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="large", details={"data": "x" * 5000})
|
||||
assert error.value.code == "CHECKPOINT_LIMIT"
|
||||
assert store.latest("case-A", "stable-package") == original
|
||||
|
||||
|
||||
def test_empty_array_details_rejected(admitted):
|
||||
_, store, original = admitted
|
||||
with pytest.raises(LifecycleError):
|
||||
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="bad", details=[])
|
||||
assert store.latest("case-A", "stable-package") == original
|
||||
@@ -0,0 +1,248 @@
|
||||
import json
|
||||
import shutil
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
from viano_bug_analysis.lifecycle import FileLock, InboxProcessor, LifecycleError, State, Workspace
|
||||
from viano_bug_analysis.lifecycle.storage import key
|
||||
from viano_bug_analysis.package import PackageLimits, PackageReader
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def queue(tmp_path):
|
||||
return InboxProcessor(Workspace(tmp_path / "service"))
|
||||
|
||||
|
||||
def deliver(queue, source, name="anything.bin"):
|
||||
path = queue.workspace.inbox / name
|
||||
shutil.copyfile(source, path)
|
||||
return path
|
||||
|
||||
|
||||
def test_intake_and_renamed_duplicate(queue, make_package):
|
||||
source = make_package(entries={"unknown": b"retained"})
|
||||
before = source.read_bytes()
|
||||
inbound = deliver(queue, source)
|
||||
result = queue.process(inbound)
|
||||
assert result.status == "ACCEPTED"
|
||||
assert not inbound.exists()
|
||||
assert next(queue.workspace.cache.glob("*.zip")).read_bytes() == before
|
||||
duplicate = deliver(queue, source, "renamed.zip")
|
||||
again = queue.process(duplicate)
|
||||
assert again.status == "DUPLICATE"
|
||||
assert again.package == result.package
|
||||
assert not duplicate.exists()
|
||||
assert len(list(result.package.parent.glob("g*.zip"))) == 1
|
||||
|
||||
|
||||
def test_repacked_duplicate(queue, make_package):
|
||||
import zipfile
|
||||
first = make_package(entries={"unknown": b"retained"}, name="first")
|
||||
original = queue.process(deliver(queue, first)).package
|
||||
repacked = make_package(entries={"unknown": b"retained"}, name="second", compression=zipfile.ZIP_DEFLATED)
|
||||
result = queue.process(deliver(queue, repacked))
|
||||
assert result.status == "DUPLICATE"
|
||||
assert result.package == original
|
||||
|
||||
|
||||
def test_generation_conflict_quarantined(queue, make_package):
|
||||
queue.process(deliver(queue, make_package(entries={"unknown": b"old"})))
|
||||
conflict = deliver(queue, make_package(entries={"unknown": b"different"}))
|
||||
result = queue.process(conflict)
|
||||
assert result.status == "QUARANTINED"
|
||||
assert result.code == "GENERATION_CONFLICT"
|
||||
assert json.loads((result.package.parent / "validation_error.json").read_text())["issues"][0]["code"] == "GENERATION_CONFLICT"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("raw", [b"broken", b"{}", b"[]"])
|
||||
def test_invalid_packages_preserved_in_quarantine(raw, queue, make_package):
|
||||
source = make_package(raw=raw)
|
||||
before = source.read_bytes()
|
||||
result = queue.process(deliver(queue, source))
|
||||
assert result.status == "QUARANTINED"
|
||||
assert result.package.read_bytes() == before
|
||||
assert (result.package.parent / "validation_error.json").exists()
|
||||
|
||||
|
||||
def test_oversized_archive_quarantined_without_caching(tmp_path):
|
||||
queue = InboxProcessor(Workspace(tmp_path / "service"), PackageLimits(max_archive_bytes=20))
|
||||
source = queue.workspace.inbox / "huge"
|
||||
source.write_bytes(b"x" * 100)
|
||||
result = queue.process(source)
|
||||
assert result.status == "QUARANTINED"
|
||||
assert result.package.read_bytes() == b"x" * 100
|
||||
assert not list(queue.workspace.cache.glob("*.zip"))
|
||||
|
||||
|
||||
def test_case_lock_contention_leaves_input_pending(queue, make_package):
|
||||
source = deliver(queue, make_package())
|
||||
with FileLock(queue.workspace.runtime / "locks" / f"case-{key('case-A')}.lock"):
|
||||
result = queue.process(source)
|
||||
assert result.status == "BUSY"
|
||||
assert source.exists()
|
||||
assert queue.process(source).status == "ACCEPTED"
|
||||
|
||||
|
||||
def test_worker_lock_contention(queue, make_package):
|
||||
source = deliver(queue, make_package())
|
||||
with FileLock(queue.workspace.runtime / "locks" / "worker.lock"):
|
||||
assert queue.process(source).status == "BUSY"
|
||||
assert queue.process(source).status == "ACCEPTED"
|
||||
|
||||
|
||||
def test_future_blocked_case_does_not_hold_worker(queue, make_package, manifest):
|
||||
first = queue.process(deliver(queue, make_package()))
|
||||
future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()
|
||||
blocked = queue.store.advance("case-A", "stable-package", State.BLOCKED, step_id="external-unavailable", reason="External dependency", retry_not_before=future)
|
||||
# A returned blocked case keeps its retry time and does not occupy the worker.
|
||||
other_service = InboxProcessor(Workspace(queue.workspace.root.parent / "other"))
|
||||
deferred = other_service.process(deliver(other_service, blocked))
|
||||
assert deferred.status == "DEFERRED"
|
||||
manifest.update(case_id="case-B", package_id="other-package")
|
||||
fresh = deliver(other_service, make_package(name="new-case"), "second.bin")
|
||||
outcomes = other_service.poll()
|
||||
assert any(outcome.status == "ACCEPTED" and outcome.source == fresh for outcome in outcomes)
|
||||
|
||||
|
||||
def test_due_blocked_case_resumes_once(queue, make_package):
|
||||
queue.process(deliver(queue, make_package()))
|
||||
past = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat()
|
||||
queue.store.advance("case-A", "stable-package", State.BLOCKED, step_id="external-unavailable", reason="External dependency", retry_not_before=past)
|
||||
outcomes = queue.poll()
|
||||
assert [item.status for item in outcomes] == ["RESUMED"]
|
||||
with PackageReader(outcomes[0].package) as reader:
|
||||
assert reader.manifest["state"] == State.VALIDATED
|
||||
assert reader.manifest["routing"]["retry_not_before"] is None
|
||||
assert reader.manifest["routing"]["retry_count"] == 1
|
||||
from viano_bug_analysis.lifecycle.checkpoints import checkpoint_chain
|
||||
blocked_checkpoint = checkpoint_chain(reader)[1][1]
|
||||
assert blocked_checkpoint["routing"]["reason"] == "External dependency"
|
||||
assert blocked_checkpoint["routing"]["retry_not_before"] == past
|
||||
assert queue.poll() == []
|
||||
|
||||
|
||||
def test_crash_before_acknowledgement_does_not_generate_again(queue, make_package, monkeypatch):
|
||||
source = deliver(queue, make_package())
|
||||
def crash(*args):
|
||||
raise LifecycleError("SIMULATED_CRASH", "before acknowledgement")
|
||||
monkeypatch.setattr(queue, "_acknowledge", crash)
|
||||
assert queue.process(source).status == "ERROR"
|
||||
assert source.exists()
|
||||
restarted = InboxProcessor(queue.workspace)
|
||||
outcome = restarted.process(source)
|
||||
assert outcome.status == "DUPLICATE"
|
||||
assert len(list(outcome.package.parent.glob("g*.zip"))) == 1
|
||||
|
||||
|
||||
def test_changed_inbox_file_is_not_deleted(queue, make_package, monkeypatch):
|
||||
source = deliver(queue, make_package())
|
||||
original = queue.store.intake
|
||||
def change(snapshot):
|
||||
result = original(snapshot)
|
||||
source.write_bytes(b"new producer content")
|
||||
return result
|
||||
monkeypatch.setattr(queue.store, "intake", change)
|
||||
outcome = queue.process(source)
|
||||
assert outcome.status == "ERROR"
|
||||
assert outcome.code == "SOURCE_CHANGED"
|
||||
assert source.read_bytes() == b"new producer content"
|
||||
|
||||
|
||||
def test_poll_limit_and_continues_after_bad_archive(queue, make_package):
|
||||
(queue.workspace.inbox / "a-invalid").write_bytes(b"not zip")
|
||||
deliver(queue, make_package(), "b-valid")
|
||||
assert len(queue.poll(max_jobs=1)) == 1
|
||||
assert queue.poll()[0].status == "ACCEPTED"
|
||||
with pytest.raises(ValueError):
|
||||
queue.poll(max_jobs=0)
|
||||
|
||||
|
||||
def test_unsafe_source_rejected(queue, make_package):
|
||||
source = make_package()
|
||||
assert queue.process(source).code == "UNSAFE_INPUT"
|
||||
assert source.exists()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("change", ["cycle", "identity", "version", "generation", "state", "parent_path"])
|
||||
def test_invalid_checkpoint_chain_quarantined(queue, make_package, change, tmp_path):
|
||||
import hashlib
|
||||
import zipfile
|
||||
from viano_bug_analysis.lifecycle.checkpoints import CHECKPOINT_POINTER
|
||||
admitted = queue.process(deliver(queue, make_package())).package
|
||||
with zipfile.ZipFile(admitted) as archive:
|
||||
entries = {name: archive.read(name) for name in archive.namelist()}
|
||||
manifest = json.loads(entries["manifest.json"])
|
||||
ref = manifest["artifacts"][CHECKPOINT_POINTER]
|
||||
checkpoint = json.loads(entries[ref])
|
||||
if change == "cycle":
|
||||
checkpoint["parent"] = ref
|
||||
elif change == "identity":
|
||||
checkpoint["case_id"] = "other"
|
||||
elif change == "version":
|
||||
checkpoint["format_version"] = 2
|
||||
elif change == "generation":
|
||||
checkpoint["source_generation"] = 42
|
||||
elif change == "state":
|
||||
checkpoint["state"] = "BUILDING_CASE"
|
||||
else:
|
||||
checkpoint["parent"] = "../escape"
|
||||
entries[ref] = json.dumps(checkpoint).encode()
|
||||
manifest["integrity"][ref].update(size=len(entries[ref]), sha256=hashlib.sha256(entries[ref]).hexdigest())
|
||||
entries["manifest.json"] = json.dumps(manifest).encode()
|
||||
bad = tmp_path / "bad-checkpoint"
|
||||
with zipfile.ZipFile(bad, "w") as archive:
|
||||
for name, data in entries.items():
|
||||
archive.writestr(name, data)
|
||||
other = InboxProcessor(Workspace(tmp_path / "other-service"))
|
||||
result = other.process(deliver(other, bad))
|
||||
assert result.status == "QUARANTINED"
|
||||
assert result.code == "CHECKPOINT_INVALID"
|
||||
|
||||
|
||||
def test_return_without_old_files_is_quarantined(queue, make_package, manifest):
|
||||
queue.process(deliver(queue, make_package(entries={"unknown": b"required"})))
|
||||
manifest["package_generation"] = 2
|
||||
manifest["state"] = "VALIDATED"
|
||||
result = queue.process(deliver(queue, make_package()))
|
||||
assert result.status == "QUARANTINED"
|
||||
assert result.code == "CHECKPOINT_HISTORY_LOST"
|
||||
|
||||
|
||||
def test_managed_corruption_does_not_quarantine_valid_inbound(queue, make_package):
|
||||
source = make_package()
|
||||
admitted = queue.process(deliver(queue, source)).package
|
||||
admitted.write_bytes(b"corrupt managed package")
|
||||
inbound = deliver(queue, source)
|
||||
result = queue.process(inbound)
|
||||
assert result.status == "ERROR"
|
||||
assert result.code == "STORAGE_PACKAGE_INVALID"
|
||||
assert inbound.exists()
|
||||
|
||||
|
||||
def test_runtime_stale_on_crash_does_not_block_other_cases(queue, make_package, manifest, monkeypatch):
|
||||
original_index = queue.store._index
|
||||
def crash(*args):
|
||||
raise LifecycleError("SIMULATED_CRASH", "after publication")
|
||||
monkeypatch.setattr(queue.store, "_index", crash)
|
||||
source = deliver(queue, make_package())
|
||||
assert queue.process(source).status == "ERROR"
|
||||
monkeypatch.setattr(queue.store, "_index", original_index)
|
||||
manifest.update(case_id="second", package_id="second-package")
|
||||
second = deliver(queue, make_package(name="second"), "second-case")
|
||||
assert queue.process(second).status == "ACCEPTED"
|
||||
assert queue.process(source).status == "DUPLICATE"
|
||||
|
||||
|
||||
def test_deferred_jobs_do_not_consume_admission_limit(queue, make_package, manifest, tmp_path):
|
||||
queue.process(deliver(queue, make_package()))
|
||||
future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()
|
||||
blocked = queue.store.advance("case-A", "stable-package", State.BLOCKED, step_id="block", reason="Dependency", retry_not_before=future)
|
||||
other = InboxProcessor(Workspace(tmp_path / "another-service"))
|
||||
for index in range(3):
|
||||
deliver(other, blocked, f"a-blocked-{index}")
|
||||
manifest.update(case_id="new-case", package_id="new-package")
|
||||
fresh = deliver(other, make_package(name="new"), "z-fresh")
|
||||
outcomes = other.poll(max_jobs=1)
|
||||
assert [item.status for item in outcomes] == ["DEFERRED", "DEFERRED", "DEFERRED", "ACCEPTED"]
|
||||
assert not fresh.exists()
|
||||
@@ -0,0 +1,29 @@
|
||||
import pytest
|
||||
|
||||
from viano_bug_analysis.lifecycle import LifecycleError, State
|
||||
from viano_bug_analysis.lifecycle.state import ACTIVE, ORDER, check_transition
|
||||
|
||||
|
||||
@pytest.mark.parametrize("before,after", list(zip(ORDER, ORDER[1:])))
|
||||
def test_forward_transitions(before, after):
|
||||
check_transition(before, after)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("state", list(ACTIVE))
|
||||
def test_block_failure_and_progress_checkpoint(state):
|
||||
check_transition(state, State.BLOCKED)
|
||||
check_transition(state, State.FAILED_TO_ANALYZE)
|
||||
check_transition(state, state)
|
||||
check_transition(State.BLOCKED, state, state)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("before,after", [(State.NEW, State.BUILDING_CASE), (State.HANDED_OFF, State.VALIDATED), (State.BLOCKED, State.INTERPRETING_CASE), (State.FAILED_TO_ANALYZE, State.BUILDING_CASE), (State.OUTPUT_READY, State.NEW)])
|
||||
def test_forbidden_transitions(before, after):
|
||||
with pytest.raises(LifecycleError, match="Cannot transition"):
|
||||
check_transition(before, after)
|
||||
|
||||
|
||||
def test_reopening():
|
||||
check_transition(State.HANDED_OFF, State.REOPENED)
|
||||
check_transition(State.FAILED_TO_ANALYZE, State.REOPENED)
|
||||
check_transition(State.REOPENED, State.BUILDING_CASE)
|
||||
@@ -0,0 +1,59 @@
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
from viano_bug_analysis.lifecycle import FileLock, LifecycleError, LockBusy, Workspace
|
||||
from viano_bug_analysis.lifecycle.storage import atomic_json
|
||||
|
||||
|
||||
def test_atomic_json_failure_keeps_old_file(tmp_path, monkeypatch):
|
||||
path = tmp_path / "index.json"
|
||||
atomic_json(path, {"old": True})
|
||||
def fail(*args):
|
||||
raise OSError("replace failure")
|
||||
monkeypatch.setattr(os, "replace", fail)
|
||||
with pytest.raises(LifecycleError):
|
||||
atomic_json(path, {"new": True})
|
||||
assert json.loads(path.read_text()) == {"old": True}
|
||||
assert not list(tmp_path.glob(".json-*"))
|
||||
|
||||
|
||||
def test_case_ids_do_not_become_paths(tmp_path):
|
||||
workspace = Workspace(tmp_path / "service")
|
||||
path = workspace.case_directory("../../case", "C:/package")
|
||||
assert path.is_relative_to(workspace.cases)
|
||||
assert len(path.name) == 64
|
||||
assert len(path.parent.name) == 64
|
||||
|
||||
|
||||
def test_competing_lock_and_release(tmp_path):
|
||||
path = tmp_path / "lock"
|
||||
with FileLock(path):
|
||||
with pytest.raises(LockBusy):
|
||||
with FileLock(path):
|
||||
pass
|
||||
with FileLock(path):
|
||||
pass
|
||||
|
||||
|
||||
def test_lock_released_on_process_exit(tmp_path):
|
||||
path = tmp_path / "process.lock"
|
||||
script = "from pathlib import Path; from viano_bug_analysis.lifecycle import FileLock; import os,sys; lock=FileLock(Path(sys.argv[1])); lock.__enter__(); print('locked',flush=True); sys.stdin.readline(); os._exit(17)"
|
||||
env = dict(os.environ, PYTHONPATH=str(__import__('pathlib').Path(__file__).resolve().parents[1] / "src"))
|
||||
process = subprocess.Popen([sys.executable, "-c", script, str(path)], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env)
|
||||
try:
|
||||
assert process.stdout.readline().strip() == "locked"
|
||||
with pytest.raises(LockBusy):
|
||||
with FileLock(path):
|
||||
pass
|
||||
process.communicate("exit\n", timeout=10)
|
||||
assert process.returncode == 17
|
||||
with FileLock(path):
|
||||
pass
|
||||
finally:
|
||||
if process.poll() is None:
|
||||
process.kill()
|
||||
process.wait()
|
||||
Reference in new issue
Block a user