diff --git a/backend/cortex_backend/execution/cleanup.py b/backend/cortex_backend/execution/cleanup.py index 3c0ee20..22f0f8b 100644 --- a/backend/cortex_backend/execution/cleanup.py +++ b/backend/cortex_backend/execution/cleanup.py @@ -178,7 +178,7 @@ def run_once(self) -> bool: raise RuntimeError("cleanup returned an invalid result") # After the rows, so a directory the pass just emptied is # already gone by the time the sweep looks at it. - swept = self.repository.sweep_artifact_root(limit=self.batch_size) + swept = self._sweep() if self._lease_lost.is_set(): raise RuntimeError("cleanup lease lost") self._increment( @@ -215,6 +215,24 @@ def run_once(self) -> bool: finally: self._run_lock.release() + def _sweep(self) -> int: + """Sweep the artifact root, keeping a failure from undoing the retention pass. + + The retention pass has already committed by now, so the sweep failing + (a locked store, an unreadable directory) must not record the whole + pass as failed and drop its counters. The failure is logged by type + only, kept in ``last_error`` and simply retried on the next pass. + """ + + try: + return self.repository.sweep_artifact_root(limit=self.batch_size) + except Exception as exc: + self._increment(last_error=type(exc).__name__) + _LOGGER.warning( + "Cortex artifact-root sweep failed (%s); it will be retried.", type(exc).__name__ + ) + return 0 + def _run(self) -> None: while not self._stop_event.is_set(): self.run_once() diff --git a/backend/cortex_backend/execution/finish.py b/backend/cortex_backend/execution/finish.py index 7014fda..e33931d 100644 --- a/backend/cortex_backend/execution/finish.py +++ b/backend/cortex_backend/execution/finish.py @@ -52,19 +52,20 @@ def finish_unsuccessful_job( against; if a concurrent actor moved the job first, the decision is made again from the new state. A job that is already terminal is left alone. Nothing here raises: this runs in a worker's error path, where the original - outcome is what matters. + outcome is what matters. That covers the read of the job and the + ``cancel_requested`` probe as well as the write. """ for _ in range(_ATTEMPTS): - current = repository.get_job(job_id) - if current is None or current.status in TerminalExecutionStatus: - return - cancelled = ( - failure_code == "cancelled" - or cancel_requested() - or current.status == "cancelling" - ) try: + current = repository.get_job(job_id) + if current is None or current.status in TerminalExecutionStatus: + return + cancelled = ( + failure_code == "cancelled" + or cancel_requested() + or current.status == "cancelling" + ) repository.transition( job_id, status="cancelled" if cancelled else "failed", diff --git a/backend/cortex_backend/execution/repository.py b/backend/cortex_backend/execution/repository.py index 4ec53ff..7aac3c7 100644 --- a/backend/cortex_backend/execution/repository.py +++ b/backend/cortex_backend/execution/repository.py @@ -34,6 +34,16 @@ ) +# The schema ladder. 1-2: earlier builds. 3: every job is owned by the +# installation principal. An artifact's path, and the paths of its cleanup +# tombstone, may be stored relative to the artifact root so a moved data +# directory keeps working. That is a change to what new rows hold, not to a +# table or a column, so it is not a version step: the version stays 3, rows +# written by an earlier build keep their absolute paths, and both forms are +# read (see ``ExecutionRepository._stored_path``). Leaving the version alone is +# what lets an earlier build open a store this one wrote: a build that only +# knows an older version sets a newer store aside, jobs and all, instead of +# reading it. SCHEMA_VERSION = 3 MAX_EVENT_BYTES = 64 * 1024 MAX_APPROVAL_TTL_SECONDS = 300.0 @@ -71,8 +81,13 @@ _SWEEP_ENTRY_BUDGET = 2_000 _SWEEP_CHILD_LIMIT = 64 # A path-component-safe job id, and a printable owner. The job id becomes a -# directory name under the artifact root, so it is held to the same shape as an -# artifact name; the owner never reaches the filesystem. +# directory name under the artifact root, so it is held to a stricter shape +# than an artifact name: it may not end in a dot (Windows drops trailing dots +# and spaces, so ``a.`` would share a directory with ``a``) and it may not be a +# Windows device name, with or without an extension (``nul`` and ``nul.txt`` name +# the NUL device, not a directory). The owner never reaches the filesystem. +_SAFE_JOB_ID = re.compile(r"[A-Za-z0-9](?:[A-Za-z0-9._-]{0,198}[A-Za-z0-9_-])?") +_WINDOWS_DEVICE_NAME = re.compile(r"(?:con|prn|aux|nul|com[1-9]|lpt[1-9])(?:\..*)?", re.IGNORECASE) _SAFE_OWNER = re.compile(r"[^\x00-\x1f\x7f]{1,200}") _SAFE_INSTALLATION_PRINCIPAL = re.compile(r"^[0-9a-f]{64}$") _SAFE_MIME = re.compile(r"^[a-z0-9][a-z0-9.+-]{0,31}/[a-z0-9][a-z0-9.+-]{0,63}$") @@ -95,6 +110,16 @@ def _is_reparse_point(path: Path) -> bool: return bool(is_junction is not None and is_junction()) +def _is_safe_job_id(value: object) -> bool: + """Whether ``value`` can name a job directory on every platform Cortex runs on.""" + + return ( + isinstance(value, str) + and _SAFE_JOB_ID.fullmatch(value) is not None + and _WINDOWS_DEVICE_NAME.fullmatch(value) is None + ) + + def _has_reparse_parent(path: Path) -> bool: """Reject a configured root that would traverse a link in an ancestor.""" @@ -674,7 +699,7 @@ def create_job( # The job id names the job's artifact directory, so an id that could # lead out of the artifact root is refused when the job is created, not # only when its first artifact is published. - if not isinstance(job_id, str) or _SAFE_NAME.fullmatch(job_id) is None: + if not _is_safe_job_id(job_id): raise ValueError("job_id must be a bounded path-safe identifier") if not isinstance(owner, str) or _SAFE_OWNER.fullmatch(owner) is None: raise ValueError("owner must be a bounded printable identifier") @@ -1757,7 +1782,7 @@ def publish_artifact( raise ArtifactLimitError("Artifact exceeds the configured size limit.") if retention_seconds <= 0: raise ValueError("retention_seconds must be positive") - if not isinstance(job_id, str) or _SAFE_NAME.fullmatch(job_id) is None: + if not _is_safe_job_id(job_id): raise ExecutionRepositoryError("Execution job does not exist.") if self.get_job(job_id) is None: raise ExecutionRepositoryError("Execution job does not exist.") @@ -1781,13 +1806,21 @@ def publish_artifact( stream: Any = None try: with self._directory_lock: - job_root.mkdir(parents=True, exist_ok=True) - resolved_job_root = job_root.resolve(strict=True) + # An OSError from either of these carries the absolute path, + # so it is reported as this repository's own error instead. + try: + job_root.mkdir(parents=True, exist_ok=True) + resolved_job_root = job_root.resolve(strict=True) + except OSError: + raise ExecutionRepositoryError("Artifact directory is unavailable.") from None if not resolved_job_root.is_relative_to(root) or _is_reparse_point(resolved_job_root): raise ExecutionRepositoryError("Artifact path escaped the artifact root.") # The temporary file exists before the lock is released, so the # directory is never empty for a sweep to remove. - stream = temporary.open("xb") + try: + stream = temporary.open("xb") + except OSError: + raise ExecutionRepositoryError("Artifact file could not be created.") from None with stream: stream.write(content) stream.flush() @@ -1818,7 +1851,7 @@ def publish_artifact( mime_type, len(content), digest, - str(target), + self._stored_text(target), now.isoformat(), expires.isoformat(), ), @@ -1826,8 +1859,13 @@ def publish_artifact( except Exception: if stream is not None: stream.close() - target.unlink(missing_ok=True) - temporary.unlink(missing_ok=True) + for leftover in (target, temporary): + try: + leftover.unlink(missing_ok=True) + except OSError: + # Never replaces the error being raised. What stays is a + # file no row names, which the artifact-root sweep reclaims. + pass # The directory this call created (or found empty) is not left behind. self._remove_empty_artifact_directory(job_root) raise @@ -1876,7 +1914,7 @@ def get_artifact( mime_type=row["mime_type"], size=int(row["size"]), sha256=row["sha256"], - path=row["path"], + path=str(self._stored_path(row["path"])), created_at=row["created_at"], expires_at=row["expires_at"], ) @@ -1906,7 +1944,7 @@ def delete_artifact(self, artifact_id: str, *, owner: str | None = None) -> None ).fetchone() if row is None: return - path = self._validated_cleanup_path(Path(row["path"])) + path = self._validated_cleanup_path(self._stored_path(row["path"])) connection.execute( "DELETE FROM execution_artifacts WHERE artifact_id = ?", (artifact_id,), @@ -1941,7 +1979,7 @@ def read_artifact(self, artifact_id: str, *, owner: str | None = None) -> bytes: raise ExecutionRepositoryError("Artifact integrity check failed.") from None if not 0 <= expected_size <= self.max_artifact_bytes: raise ExecutionRepositoryError("Artifact integrity check failed.") - original_path = Path(row["path"]) + original_path = self._stored_path(row["path"]) if _is_reparse_point(original_path): raise ExecutionRepositoryError("Artifact path is unavailable.") try: @@ -2021,7 +2059,7 @@ def cleanup_expired( artifact_id = str(artifact_row["artifact_id"]) quarantine = self.quarantine_root / f"{artifact_id}-{uuid4().hex}.artifact" try: - path = self._validated_cleanup_path(Path(artifact_row["path"])) + path = self._validated_cleanup_path(self._stored_path(artifact_row["path"])) self._validated_quarantine_path(quarantine) self._record_artifact_cleanup(artifact_id, path, quarantine) except ArtifactCleanupRejected as exc: @@ -2083,6 +2121,31 @@ def cleanup_expired( skipped=skipped, ) + def _stored_path(self, text: str) -> Path: + """The location a row's path column names, under this build's artifact root. + + New rows record a path relative to the artifact root, so moving the + data directory does not orphan them; rows written by an earlier build + hold an absolute path, and that is returned as written. Nothing here + decides whether the location is acceptable: an absolute path, a + relative one that climbs out with ``..`` and a drive-relative one all + come back as they resolve, and the validators that follow + (:meth:`_validated_cleanup_path`, :meth:`_validated_quarantine_path` + and the containment check in :meth:`read_artifact`) refuse any that + does not end up inside the root. + """ + + path = Path(text) + return path if path.is_absolute() else self.artifact_root / path + + def _stored_text(self, path: Path) -> str: + """How a row records ``path``: relative to the artifact root when it lies under it.""" + + try: + return path.relative_to(self.artifact_root).as_posix() + except ValueError: + return str(path) + def _validated_cleanup_path(self, path: Path) -> Path: """Validate a source path without following an untrusted reparse hop. @@ -2139,7 +2202,7 @@ def _record_artifact_cleanup( VALUES (?, ?, ?, 'pending', ?) ON CONFLICT(artifact_id) DO NOTHING """, - (artifact_id, str(path), str(quarantine), self._now()), + (artifact_id, self._stored_text(path), self._stored_text(quarantine), self._now()), ) def _resume_artifact_cleanup(self, *, limit: int) -> tuple[int, int]: @@ -2218,7 +2281,7 @@ def _reclaim_quarantined_file(self, artifact_id: str, quarantine_text: str) -> b """ try: - quarantine = self._validated_quarantine_path(Path(quarantine_text)) + quarantine = self._validated_quarantine_path(self._stored_path(quarantine_text)) except ArtifactCleanupBlocked: return False except ArtifactCleanupRejected: @@ -2283,8 +2346,8 @@ def _requeue_artifact_cleanup(self, artifact_id: str) -> None: def _advance_artifact_cleanup( self, artifact_id: str, path_text: str, quarantine_text: str, state: str ) -> int: - path = self._validated_cleanup_path(Path(path_text)) - quarantine = self._validated_quarantine_path(Path(quarantine_text)) + path = self._validated_cleanup_path(self._stored_path(path_text)) + quarantine = self._validated_quarantine_path(self._stored_path(quarantine_text)) if state not in {"pending", "quarantined", "finalized"}: raise ArtifactCleanupRejected("Artifact cleanup state is invalid.") removed = 0 @@ -2402,9 +2465,18 @@ def sweep_artifact_root(self, *, now: datetime | None = None, limit: int = 100) job; and * a link, a reparse point or anything of another name is never touched. - The pass is bounded: it removes at most ``limit`` things and looks at - at most ``_SWEEP_ENTRY_BUDGET`` entries, resuming where it stopped on - the next call. Best effort: a failure on one entry never stops the pass. + The pass is bounded, and the bounds are exact where they matter. It + removes at most ``limit`` things, counting every file, every job + directory and every staging directory as one; a job directory that + would need more than what is left is worked on until the limit is + reached and picked up again by the next call, at that same directory. + It stops taking on new top-level entries once it has looked at + ``_SWEEP_ENTRY_BUDGET`` entries (a directory counts as itself plus the + children it reads). That check runs between top-level entries, so a + pass can look at up to ``_SWEEP_CHILD_LIMIT`` + 1 entries more than the + budget; it never removes more than ``limit``. The next call resumes + where this one stopped. Best effort: a failure on one entry never + stops the pass. """ if isinstance(limit, bool) or not 1 <= limit <= 10_000: @@ -2427,55 +2499,76 @@ def sweep_artifact_root(self, *, now: datetime | None = None, limit: int = 100) if removed >= limit or examined >= _SWEEP_ENTRY_BUDGET: finished = False break - gone, looked_at = self._sweep_entry(connection, self.artifact_root / name, stale_before) + gone, looked_at, complete = self._sweep_entry( + connection, self.artifact_root / name, stale_before, limit - removed + ) removed += gone examined += looked_at + if not complete: + # The limit ran out inside this entry. What it removed is + # gone for good, so the next pass starts here and finishes it. + finished = False + break self._sweep_cursor = name if finished: self._sweep_cursor = "" return removed def _sweep_entry( - self, connection: sqlite3.Connection, entry: Path, stale_before: float - ) -> tuple[int, int]: - """Sweep one entry directly under the artifact root; return (removed, examined).""" + self, + connection: sqlite3.Connection, + entry: Path, + stale_before: float, + allowance: int, + ) -> tuple[int, int, bool]: + """Sweep one entry directly under the artifact root. + + Returns ``(removed, examined, complete)``. ``removed`` never exceeds + ``allowance``; ``complete`` is false only when the allowance ran out + with more of this entry left to remove, so the caller must come back + to it instead of moving on. + """ name = entry.name try: if _is_reparse_point(entry) or not stat.S_ISDIR(entry.lstat().st_mode): - return 0, 1 + return 0, 1, True except OSError: - return 0, 1 + return 0, 1, True if name == _LEGACY_QUARANTINE_NAME: - return int(self._remove_empty_directory(entry)), 1 + return int(self._remove_empty_directory(entry)), 1, True staging = _RECIPE_STAGING_NAME.fullmatch(name) if staging is not None: job_id = staging["job_id"] if _SAFE_NAME.fullmatch(job_id) is None: - return 0, 1 + return 0, 1, True job = connection.execute( "SELECT status FROM execution_jobs WHERE job_id = ?", (job_id,) ).fetchone() if job is not None and job["status"] not in TerminalExecutionStatus: - return 0, 1 # its worker may still be writing here + return 0, 1, True # its worker may still be writing here try: shutil.rmtree(entry) except OSError: - return 0, 1 - return 1, 1 + return 0, 1, True + return 1, 1, True if name.startswith(".") or _SAFE_NAME.fullmatch(name) is None: - return 0, 1 - return self._sweep_job_directory(connection, entry, stale_before) + return 0, 1, True + return self._sweep_job_directory(connection, entry, stale_before, allowance) def _sweep_job_directory( - self, connection: sqlite3.Connection, directory: Path, stale_before: float - ) -> tuple[int, int]: + self, + connection: sqlite3.Connection, + directory: Path, + stale_before: float, + allowance: int, + ) -> tuple[int, int, bool]: removed = 0 examined = 1 try: children = list(itertools.islice(directory.iterdir(), _SWEEP_CHILD_LIMIT)) except OSError: - return 0, examined + return 0, examined, True for child in children: examined += 1 try: @@ -2491,6 +2584,8 @@ def _sweep_job_directory( (stored["artifact_id"],), ).fetchone() is not None: continue + if removed >= allowance: + return removed, examined, False try: child.unlink() except OSError: @@ -2499,9 +2594,13 @@ def _sweep_job_directory( named = connection.execute( "SELECT 1 FROM execution_artifacts WHERE job_id = ? LIMIT 1", (directory.name,) ).fetchone() - if named is None and self._remove_empty_directory(directory): - removed += 1 - return removed, examined + if named is None: + if removed >= allowance: + # The directory may be empty now, and taking it is one more removal. + return removed, examined, False + if self._remove_empty_directory(directory): + removed += 1 + return removed, examined, True @staticmethod def _encode_event(data: Mapping[str, Any]) -> str: diff --git a/backend/cortex_backend/services/attachments.py b/backend/cortex_backend/services/attachments.py index 924acc3..90c2f42 100644 --- a/backend/cortex_backend/services/attachments.py +++ b/backend/cortex_backend/services/attachments.py @@ -26,6 +26,12 @@ from PIL import Image +from cortex_backend.execution.attachment_staging import ( + ABANDONED_STAGE_SECONDS, + ATTACHMENT_INTERRUPTED, + ATTACHMENT_INTERRUPTED_MESSAGE, + STAGING_LEASE_SECONDS, +) from cortex_backend.execution.repository import ExecutionRepository, ExecutionRepositoryError @@ -357,8 +363,50 @@ def stage( ) if not created: return self._existing(job, payload, content) + + # A stager that dies between creating the job and finishing it leaves a + # queued job that recovery cannot see unless it holds a lease, and that + # nothing else ever retires. Holding a short lease while the bytes are + # staged makes the job findable, and it is released on every exit that + # runs code. This is the same window the execution staging closes. + lease_owner = f"chat-attachment-stager-{uuid4().hex}" + try: + self.repository.claim_lease( + job.job_id, lease_owner=lease_owner, ttl_seconds=STAGING_LEASE_SECONDS + ) + except ExecutionRepositoryError as exc: + self._fail(job.job_id) + raise ChatAttachmentError("attachment_persist_failed") from exc + try: + return self._stage_created( + job, + safe_name=safe_name, + kind=kind, + mime_type=mime_type, + content=content, + digest=digest, + ) + finally: + try: + self.repository.release_lease(job.job_id, lease_owner=lease_owner) + except Exception: + pass # It expires by itself; nothing here may replace the real outcome. + + def _stage_created( + self, + job: Any, + *, + safe_name: str, + kind: str, + mime_type: str, + content: bytes, + digest: str, + ) -> ChatAttachment: + repository = self.repository + if repository is None: + raise ChatAttachmentError("attachment_unavailable") try: - artifact = self.repository.publish_artifact( + artifact = repository.publish_artifact( job.job_id, name=f"chat-{digest[:24]}.bin", content=content, @@ -374,7 +422,7 @@ def stage( kind=kind, expires_at=artifact.expires_at, ) - self.repository.transition( + completed = repository.transition( job.job_id, status="succeeded", event="completed", @@ -382,20 +430,47 @@ def stage( data={"message": "Chat attachment staged."}, result=descriptor.as_dict(), ) - return descriptor except (ExecutionRepositoryError, OSError) as exc: - try: - self.repository.transition( - job.job_id, - status="failed", - event="failed", - phase="failed", - data={"message": "Chat attachment staging failed."}, - error="attachment_persist_failed", - ) - except Exception: - pass + self._fail(job.job_id) raise ChatAttachmentError("attachment_persist_failed") from exc + if completed.status != "succeeded": + # A finished job is immutable: the write above changed nothing and + # only reported the job as it stands. That is what a retry that + # retired this job as abandoned leaves behind when the lease ran + # out while this call was still staging. Handing out a descriptor + # for a job the store says failed would leave the artifact behind + # with no result that names it, and resolve() would then refuse it + # at send time. + self._discard_artifact(artifact.artifact_id) + raise ChatAttachmentError("attachment_persist_failed") + return descriptor + + def _discard_artifact(self, artifact_id: str) -> None: + """Remove an artifact no job result names; best effort, because the caller is already raising.""" + + if self.repository is None: + return + try: + self.repository.delete_artifact(artifact_id) + except Exception: + pass # Retention reclaims the row and the file. + + def _fail(self, job_id: str) -> None: + """Record a failed staging job; best effort, because the caller is already raising.""" + + if self.repository is None: + return + try: + self.repository.transition( + job_id, + status="failed", + event="failed", + phase="failed", + data={"message": "Chat attachment staging failed."}, + error="attachment_persist_failed", + ) + except Exception: + pass def resolve(self, *, owner: str, descriptor: Mapping[str, Any] | ChatAttachment) -> ResolvedChatAttachment: _safe_owner(owner) @@ -478,8 +553,36 @@ def _record_expired(record: _MemoryRecord, now: datetime) -> bool: return True return expiry <= now + def _retire_if_abandoned(self, job: Any) -> Any: + """Fail a job whose stager died, and return the job as it now reads. + + The store decides, atomically, that nothing is working on it: no live + lease and no write for ABANDONED_STAGE_SECONDS. A job that is merely + slow, or that finished in the meantime, comes back as it is. + """ + + if self.repository is None: + return job + try: + current = self.repository.retire_abandoned_job( + job.job_id, + idle_seconds=ABANDONED_STAGE_SECONDS, + error=ATTACHMENT_INTERRUPTED, + message=ATTACHMENT_INTERRUPTED_MESSAGE, + ) + except ExecutionRepositoryError: + return job + return job if current is None else current + def _existing(self, job: Any, payload: Mapping[str, Any], content: bytes) -> ChatAttachment: - if job.profile != CHAT_ATTACHMENT_PROFILE or dict(job.payload) != dict(payload) or job.status != "succeeded": + if job.profile != CHAT_ATTACHMENT_PROFILE or dict(job.payload) != dict(payload): + raise ChatAttachmentError("attachment_request_conflict") + if job.status in {"queued", "running"}: + # Only the request that matches the job may retire it, and only a + # job nobody is working on: a stager killed part-way otherwise + # left it queued for good. + job = self._retire_if_abandoned(job) + if job.status != "succeeded": raise ChatAttachmentError("attachment_request_conflict") result = job.result descriptor = self._normalize_descriptor(result) diff --git a/tests/test_artifact_reclamation.py b/tests/test_artifact_reclamation.py index 39668ea..9eca170 100644 --- a/tests/test_artifact_reclamation.py +++ b/tests/test_artifact_reclamation.py @@ -315,6 +315,47 @@ def test_the_sweep_never_follows_or_removes_a_link(tmp_path): assert link.is_symlink() +def _make_junction(link: Path, target: Path) -> None: + """A real directory junction: a reparse point ``lstat`` reports as a plain directory.""" + + if os.name != "nt": + pytest.skip("directory junctions exist only on Windows") + if not hasattr(Path, "is_junction"): + pytest.skip("this interpreter cannot recognise a junction (Path.is_junction arrived in 3.12)") + import _winapi # type: ignore[import-not-found] + + try: + _winapi.CreateJunction(str(target), str(link)) + except OSError: + pytest.skip("directory junctions cannot be created here") + + +@pytest.mark.parametrize("name", ("job-junction", ".recipe-job-gone-abcd1234")) +def test_the_sweep_never_follows_or_removes_a_junction(tmp_path, name): + """A symbolic link reads as a link to ``lstat``; a junction reads as a directory. + + Only the reparse-point check stands between the sweep and the files the + junction leads to, so it needs a real junction to be tested at all. + """ + + repository = _repository(tmp_path) + outside = tmp_path / "outside" + outside.mkdir() + (outside / f".tmp-{'e' * 32}").write_bytes(b"not ours") + _age(outside / f".tmp-{'e' * 32}") + (outside / "output").write_bytes(b"not ours either") + junction = repository.artifact_root / name + _make_junction(junction, outside) + try: + assert repository.sweep_artifact_root() == 0 + + assert (outside / f".tmp-{'e' * 32}").read_bytes() == b"not ours" + assert (outside / "output").read_bytes() == b"not ours either" + assert junction.exists(), "the junction itself was removed" + finally: + os.rmdir(junction) # removes the junction itself and nothing behind it + + def test_the_sweep_is_bounded_and_resumes_where_it_stopped(tmp_path, monkeypatch): repository = _repository(tmp_path) directories = [] @@ -340,6 +381,103 @@ def test_the_sweep_is_bounded_and_resumes_where_it_stopped(tmp_path, monkeypatch repository.sweep_artifact_root(limit=0) +def _stale_temporary_files(directory: Path, count: int) -> list[Path]: + """``count`` stranded ``.tmp-*`` files, old enough for the sweep to remove.""" + + made = [] + for index in range(count): + stale = directory / f".tmp-{index:032x}" + stale.write_bytes(b"half a write") + _age(stale) + made.append(stale) + return made + + +@pytest.mark.parametrize("limit", (1, 2, 7, 64, 65, 100)) +def test_a_pass_never_removes_more_than_its_limit_even_inside_one_directory(tmp_path, limit): + """The limit was checked only between entries. + + With ``limit=1`` one directory of 64 stale files lost all 64 of them and + then itself -- 65 removals -- in a single pass. + """ + + repository = _repository(tmp_path) + directory = repository.artifact_root / "job-big" + directory.mkdir() + _stale_temporary_files(directory, repository_module._SWEEP_CHILD_LIMIT) + + per_pass = [] + for _ in range(80): # bounded: 65 things to remove + per_pass.append(repository.sweep_artifact_root(limit=limit)) + if not directory.exists(): + break + + assert max(per_pass) <= limit + assert sum(per_pass) == repository_module._SWEEP_CHILD_LIMIT + 1 # every file, then the directory + assert not directory.exists() + + +def test_an_entry_cut_short_by_the_limit_is_resumed_not_skipped(tmp_path): + repository = _repository(tmp_path) + first = repository.artifact_root / "job-a" + second = repository.artifact_root / "job-b" + for directory, count in ((first, 3), (second, 1)): + directory.mkdir() + _stale_temporary_files(directory, count) + + # Six things to remove, two per pass. A pass that ran out of allowance inside + # job-a must come back to job-a, not move on and leave the rest of it behind. + assert repository.sweep_artifact_root(limit=2) == 2 + assert len(list(first.iterdir())) == 1, "the first pass should have stopped inside job-a" + assert len(list(second.iterdir())) == 1 + + assert repository.sweep_artifact_root(limit=2) == 2 # job-a's last file, then job-a itself + assert not first.exists(), "job-a was not finished before the sweep went on to job-b" + assert len(list(second.iterdir())) == 1, "job-b was touched before job-a was finished" + + assert repository.sweep_artifact_root(limit=2) == 2 # job-b's file, then job-b itself + assert not second.exists() + assert repository.sweep_artifact_root(limit=2) == 0 + + +def test_the_limit_bounds_staging_and_legacy_directories_as_well_as_job_directories(tmp_path): + repository = _repository(tmp_path) + _job(repository, "job-done", finished=True) + for index in range(3): + staging = repository.artifact_root / f".recipe-job-done-abcd123{index}" + staging.mkdir() + (staging / "output").write_bytes(b"synthetic") + (repository.artifact_root / ".artifact_quarantine").mkdir() + + per_pass = [] + for _ in range(8): # bounded: four things to remove + per_pass.append(repository.sweep_artifact_root(limit=1)) + if per_pass[-1] == 0: + break + + assert per_pass == [1, 1, 1, 1, 0] + assert [entry.name for entry in repository.artifact_root.iterdir()] == [".quarantine"] + + +def test_the_entry_budget_stops_a_pass_early_and_the_next_one_carries_on(tmp_path, monkeypatch): + """The clause was never exercised: the earlier test would pass without it.""" + + repository = _repository(tmp_path) + monkeypatch.setattr(repository_module, "_SWEEP_ENTRY_BUDGET", 3) + for index in range(6): + (repository.artifact_root / f"job-{index}").mkdir() + + per_pass = [] + for _ in range(10): # bounded: at most one pass per entry + per_pass.append(repository.sweep_artifact_root(limit=100)) + if not any((repository.artifact_root / f"job-{index}").exists() for index in range(6)): + break + + assert max(per_pass) <= 3, "one pass looked at more entries than its budget allows" + assert len([removed for removed in per_pass if removed]) >= 2 + assert sum(per_pass) == 6 + + def test_the_sweep_never_removes_an_unexpired_artifact_or_a_directory_a_publish_just_made(tmp_path): repository = _repository(tmp_path) job_id = _job(repository, "job-1") diff --git a/tests/test_attachment_staging.py b/tests/test_attachment_staging.py index f1ba5db..05051f2 100644 --- a/tests/test_attachment_staging.py +++ b/tests/test_attachment_staging.py @@ -324,6 +324,40 @@ def test_a_retry_never_changes_the_payload_check_or_an_existing_success(executio assert conflict.value.code == "request_conflict" +def test_a_repeated_request_looks_up_and_reads_the_staged_artifact_as_its_owner_never_unscoped( + execution_repository, monkeypatch +): + """Both repository calls that follow a repeated request are owner-scoped. + + The lookup and the read are separate queries, so either one left unscoped + would trust whatever an artifact id happens to name. + """ + + repository, _boundary, service = _service_over(execution_repository) + content = _image_bytes() + service.stage(owner=OWNER, request_id="attach-owner", content=content) + real_get = repository.get_artifact + real_read = repository.read_artifact + looked_up_as: list[str | None] = [] + read_as: list[str | None] = [] + + def recording_get(artifact_id, *, owner=None): + looked_up_as.append(owner) + return real_get(artifact_id, owner=owner) + + def recording_read(artifact_id, *, owner=None): + read_as.append(owner) + return real_read(artifact_id, owner=owner) + + monkeypatch.setattr(repository, "get_artifact", recording_get) + monkeypatch.setattr(repository, "read_artifact", recording_read) + + service.stage(owner=OWNER, request_id="attach-owner", content=content) + + assert looked_up_as == [OWNER] + assert read_as == [OWNER] + + def test_a_stager_finishing_after_its_job_was_retired_reports_failure_and_drops_the_artifact( execution_repository, monkeypatch ): diff --git a/tests/test_chat_attachments.py b/tests/test_chat_attachments.py index 8b6f01e..eac4a07 100644 --- a/tests/test_chat_attachments.py +++ b/tests/test_chat_attachments.py @@ -6,6 +6,7 @@ import binascii from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta, timezone +import hashlib from io import BytesIO from pathlib import Path import struct @@ -20,6 +21,7 @@ from cortex_backend.core.generation import GenerationAttachment import cortex_backend.services.attachments as attachments_module from cortex_backend.services.attachments import ( + CHAT_ATTACHMENT_PROFILE, ChatAttachmentError, ChatAttachmentService, MAX_CHAT_IMAGE_DECODED_BYTES, @@ -28,7 +30,12 @@ ) from cortex_backend.services.llm import PromptTemplate from cortex_backend.testing.fake_ollama import FakeOllamaState -from cortex_backend.execution.repository import ExecutionRepository +from cortex_backend.execution.attachment_staging import ( + ABANDONED_STAGE_SECONDS, + ATTACHMENT_INTERRUPTED, + STAGING_LEASE_SECONDS, +) +from cortex_backend.execution.repository import ExecutionRepository, ExecutionRepositoryError from support import session_headers as _session @@ -343,6 +350,287 @@ def test_durable_staging_is_idempotent_and_integrity_checked(tmp_path: Path): assert foreign.value.code == "attachment_unavailable" +# -- A stager that dies part-way must not leave its job queued for good --------------- + + +class _HardKill(BaseException): + """Stands in for a killed process: nothing in a ``finally`` runs when one dies.""" + + +_OWNER = "owner-a" +_STAGED = b"durable reference" + + +def _durable(tmp_path: Path) -> tuple[ExecutionRepository, ChatAttachmentService]: + repository = ExecutionRepository(tmp_path / "execution.sqlite", tmp_path / "artifacts") + return repository, ChatAttachmentService(repository) + + +def _stage(service: ChatAttachmentService, request_id: str, content: bytes = _STAGED): + return service.stage( + owner=_OWNER, request_id=request_id, filename="reference.txt", content=content + ) + + +def _queued_job(repository: ExecutionRepository, request_id: str, job_id: str): + """A chat attachment job whose stager is gone: created, never finished.""" + + payload = { + "filename": "reference.txt", + "mime_type": "text/plain", + "kind": "document", + "size": len(_STAGED), + "sha256": hashlib.sha256(_STAGED).hexdigest(), + } + job, _ = repository.create_job( + job_id=job_id, + owner=_OWNER, + request_id=request_id, + profile=CHAT_ATTACHMENT_PROFILE, + payload=payload, + ) + return job + + +def _stage_and_die_holding_the_lease(service, repository, monkeypatch, request_id: str) -> str: + """Run a stage that is killed mid-way; return the job it left behind.""" + + def killed(*_args, **_kwargs): + raise _HardKill + + with monkeypatch.context() as crash: + crash.setattr(repository, "publish_artifact", killed) + # A killed process never reaches its ``finally`` blocks. + crash.setattr(repository, "release_lease", lambda *_a, **_k: None) + with pytest.raises(_HardKill): + _stage(service, request_id) + (job,) = repository.list_jobs(owner=_OWNER) + return job.job_id + + +def test_the_staging_lease_is_held_while_bytes_are_staged_and_released_afterwards( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + repository, service = _durable(tmp_path) + real_publish = repository.publish_artifact + seen_during: list[str | None] = [] + + def observing(job_id, **kwargs): + seen_during.append(repository.lease_holder(job_id)) + return real_publish(job_id, **kwargs) + + monkeypatch.setattr(repository, "publish_artifact", observing) + _stage(service, "lease-held") + + (job,) = repository.list_jobs(owner=_OWNER, include_terminal=True) + assert len(seen_during) == 1 and seen_during[0] is not None + assert job.status == "succeeded" + assert repository.lease_holder(job.job_id) is None + + +def test_the_staging_lease_is_released_and_the_job_failed_when_staging_fails( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + repository, service = _durable(tmp_path) + + def refuses(*_args, **_kwargs): + raise OSError(28, "No space left on device") + + monkeypatch.setattr(repository, "publish_artifact", refuses) + with pytest.raises(ChatAttachmentError) as failure: + _stage(service, "lease-failed") + + assert failure.value.code == "attachment_persist_failed" + (job,) = repository.list_jobs(owner=_OWNER, include_terminal=True) + assert (job.status, job.error) == ("failed", "attachment_persist_failed") + assert repository.lease_holder(job.job_id) is None + + +def test_a_lease_that_cannot_be_taken_fails_the_job_instead_of_leaving_it_queued( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + repository, service = _durable(tmp_path) + + def refuses(*_args, **_kwargs): + raise ExecutionRepositoryError("SQLite execution operation failed.") + + monkeypatch.setattr(repository, "claim_lease", refuses) + with pytest.raises(ChatAttachmentError) as failure: + _stage(service, "lease-refused") + + assert failure.value.code == "attachment_persist_failed" + (job,) = repository.list_jobs(owner=_OWNER, include_terminal=True) + assert job.status == "failed" + with repository.connect() as connection: + assert connection.execute("SELECT COUNT(*) FROM execution_artifacts").fetchone()[0] == 0 + + +def test_a_stager_that_died_holding_its_lease_is_found_by_recovery( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, frozen_clock +): + repository, service = _durable(tmp_path) + job_id = _stage_and_die_holding_the_lease(service, repository, monkeypatch, "crash-recovery") + + assert repository.get_job(job_id).status == "queued" + assert repository.lease_holder(job_id) is not None + frozen_clock.advance(STAGING_LEASE_SECONDS + 1) + assert repository.recover_expired_leases() == [job_id] + + +def test_a_retry_after_a_crashed_stage_retires_the_job_instead_of_leaving_it_queued_forever( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, frozen_clock +): + """A kill between creating the job and finishing it left it queued for good. + + Nothing ever finished it, so it stayed in the task tray, and the request id + answered the same conflict on every retry. The retry now retires a job + nobody is working on; the answer is unchanged and stable. + """ + + repository, service = _durable(tmp_path) + job_id = _stage_and_die_holding_the_lease(service, repository, monkeypatch, "crash-retry") + frozen_clock.advance(ABANDONED_STAGE_SECONDS + 1) + + for _ in range(2): # the answer is stable, not a one-off + with pytest.raises(ChatAttachmentError) as retry: + _stage(service, "crash-retry") + assert retry.value.code == "attachment_request_conflict" + + job = repository.get_job(job_id) + assert job is not None + assert (job.status, job.error) == ("failed", ATTACHMENT_INTERRUPTED) + assert [event.event for event in repository.events(job_id)].count("failed") == 1 + assert repository.lease_holder(job_id) is None + assert repository.list_jobs(owner=_OWNER) == [] # no longer a task in the tray + + +def test_a_recent_or_leased_stage_is_left_alone_by_a_retry(tmp_path: Path, frozen_clock): + repository, service = _durable(tmp_path) + job = _queued_job(repository, "quiet", "job-quiet") + frozen_clock.advance(ABANDONED_STAGE_SECONDS - 5) + + with pytest.raises(ChatAttachmentError) as recent: + _stage(service, "quiet") + assert recent.value.code == "attachment_request_conflict" + assert repository.get_job(job.job_id).status == "queued" + + # Old enough to be abandoned, but another stager holds a live lease on it. + frozen_clock.advance(ABANDONED_STAGE_SECONDS) + repository.claim_lease(job.job_id, lease_owner="another-stager", ttl_seconds=STAGING_LEASE_SECONDS) + with pytest.raises(ChatAttachmentError): + _stage(service, "quiet") + assert repository.get_job(job.job_id).status == "queued" + + +def test_a_conflicting_retry_does_not_retire_the_job_it_conflicts_with(tmp_path: Path, frozen_clock): + repository, service = _durable(tmp_path) + job = _queued_job(repository, "conflict", "job-abandoned") + frozen_clock.advance(ABANDONED_STAGE_SECONDS * 2) + + with pytest.raises(ChatAttachmentError) as conflict: + _stage(service, "conflict", content=b"different bytes") + + assert conflict.value.code == "attachment_request_conflict" + assert repository.get_job(job.job_id).status == "queued" # only the matching request may retire it + + +def test_a_retry_never_changes_an_existing_success(tmp_path: Path, frozen_clock): + repository, service = _durable(tmp_path) + first = _stage(service, "done") + frozen_clock.advance(ABANDONED_STAGE_SECONDS * 10) + + again = _stage(service, "done") + + assert again == first + (job,) = repository.list_jobs(owner=_OWNER, include_terminal=True) + assert job.status == "succeeded" + + +def _retire_between_publish_and_completion( + repository: ExecutionRepository, + service: ChatAttachmentService, + frozen_clock, + monkeypatch: pytest.MonkeyPatch, + request_id: str, +) -> None: + """After the bytes are published, let the lease lapse and a retry retire the job. + + This is the suspend/resume (or clock jump) window: the stager holds its + bytes, its 60 s lease is gone and nothing has written for longer than the + abandonment threshold, so the matching retry fails the job as interrupted. + """ + + real_publish = repository.publish_artifact + + def publish_then_get_retired(job_id, **kwargs): + artifact = real_publish(job_id, **kwargs) + frozen_clock.advance(ABANDONED_STAGE_SECONDS + STAGING_LEASE_SECONDS) + with pytest.raises(ChatAttachmentError) as retry: + _stage(service, request_id) + assert retry.value.code == "attachment_request_conflict" + return artifact + + monkeypatch.setattr(repository, "publish_artifact", publish_then_get_retired) + + +def _artifact_rows_and_files(repository: ExecutionRepository) -> tuple[int, list[Path]]: + with repository.connect() as connection: + rows = connection.execute("SELECT COUNT(*) FROM execution_artifacts").fetchone()[0] + files = [path for path in repository.artifact_root.rglob("*") if path.is_file()] + return rows, files + + +def test_a_stager_retired_between_publish_and_completion_fails_and_leaves_no_artifact( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, frozen_clock +): + """The call that lost its job must not report success for it. + + Finishing a job the store already failed writes nothing and returns the + failed job. Ignoring that returned a descriptor whose artifact no job result + named, and ``resolve`` then refused it at send time. + """ + + repository, service = _durable(tmp_path) + _retire_between_publish_and_completion( + repository, service, frozen_clock, monkeypatch, "retired-mid-stage" + ) + + with pytest.raises(ChatAttachmentError) as lost: + _stage(service, "retired-mid-stage") + + assert lost.value.code == "attachment_persist_failed" + (job,) = repository.list_jobs(owner=_OWNER, include_terminal=True) + assert (job.status, job.error, job.result) == ("failed", ATTACHMENT_INTERRUPTED, None) + assert [event.event for event in repository.events(job.job_id)].count("completed") == 0 + assert repository.lease_holder(job.job_id) is None + rows, files = _artifact_rows_and_files(repository) + assert rows == 0, "the published artifact row was left behind" + assert files == [], "the published artifact file was left behind" + + +def test_a_failed_removal_of_the_orphaned_artifact_never_replaces_the_staging_error( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, frozen_clock +): + repository, service = _durable(tmp_path) + _retire_between_publish_and_completion(repository, service, frozen_clock, monkeypatch, "retired-stuck") + secret_location = str(tmp_path / "artifacts") + + def refuses(*_args, **_kwargs): + raise ExecutionRepositoryError(f"Artifact cleanup failed at {secret_location}.") + + monkeypatch.setattr(repository, "delete_artifact", refuses) + with pytest.raises(ChatAttachmentError) as lost: + _stage(service, "retired-stuck") + + assert lost.value.code == "attachment_persist_failed" + assert secret_location not in str(lost.value) + assert lost.value.__cause__ is None, "the failure text would carry the store's location" + (job,) = repository.list_jobs(owner=_OWNER, include_terminal=True) + assert (job.status, job.error) == ("failed", ATTACHMENT_INTERRUPTED) + # Nothing removed it, so it waits for retention; no job result names it either. + assert _artifact_rows_and_files(repository)[0] == 1 + + def test_api_reports_non_vision_models_and_returns_only_attachment_metadata(): state = FakeOllamaState( installed_models={"local-chat:7b"}, diff --git a/tests/test_execution_cleanup.py b/tests/test_execution_cleanup.py index 71cd313..65c98af 100644 --- a/tests/test_execution_cleanup.py +++ b/tests/test_execution_cleanup.py @@ -429,6 +429,75 @@ def test_cleanup_supervisor_reports_rows_it_could_not_reclaim_and_still_succeeds assert str(tmp_path) not in caplog.text +def test_a_failing_sweep_does_not_discard_the_retention_pass_that_already_finished( + tmp_path, monkeypatch, caplog +): + """The sweep shared one ``try`` with the retention pass. + + ``connect()`` failing inside the sweep, after the rows were already + reclaimed, recorded the whole pass as failed and dropped its counters, so + the work looked like it had not happened. + """ + + repository = _repository(tmp_path) + expired = _terminal_job(repository, "expired") + artifact = repository.publish_artifact( + expired.job_id, name="expired.txt", content=b"remove", mime_type="text/plain" + ) + with repository.connect() as connection: + connection.execute( + "UPDATE execution_artifacts SET expires_at = ? WHERE artifact_id = ?", + ("2000-01-01T00:00:00+00:00", artifact.artifact_id), + ) + _terminal_job(repository, "idle") + + def sweep_cannot_open_the_store(**_kwargs): + raise ExecutionRepositoryError(f"SQLite execution operation failed at {tmp_path}") + + monkeypatch.setattr(repository, "sweep_artifact_root", sweep_cannot_open_the_store) + supervisor = ExecutionCleanupSupervisor(repository, terminal_job_retention_seconds=0) + + with caplog.at_level(logging.WARNING, logger="cortex.execution.cleanup"): + assert supervisor.run_once() is True + + metrics = supervisor.metrics + assert (metrics.successes, metrics.failures) == (1, 0) + assert (metrics.artifacts, metrics.jobs, metrics.events) == (1, 2, 4) + assert metrics.swept == 0 + assert metrics.last_error == "ExecutionRepositoryError" + assert "sweep failed (ExecutionRepositoryError)" in caplog.text + assert str(tmp_path) not in caplog.text + # The pass really did its work, and the lease is free for the next one. + assert repository.get_job("expired") is None + assert repository.get_artifact(artifact.artifact_id) is None + repository.claim_cleanup_lease(lease_owner="next", ttl_seconds=1) + + +def test_a_sweep_that_fails_is_retried_by_the_next_pass(tmp_path, monkeypatch): + repository = _repository(tmp_path) + (repository.artifact_root / "job-empty").mkdir() + real_sweep = repository.sweep_artifact_root + outcomes: list[str] = [] + + def flaky_sweep(**kwargs): + if not outcomes: + outcomes.append("failed") + raise ExecutionRepositoryError("SQLite execution operation failed.") + outcomes.append("ran") + return real_sweep(**kwargs) + + monkeypatch.setattr(repository, "sweep_artifact_root", flaky_sweep) + supervisor = ExecutionCleanupSupervisor(repository) + + assert supervisor.run_once() is True + assert (repository.artifact_root / "job-empty").is_dir() + assert supervisor.run_once() is True + + assert outcomes == ["failed", "ran"] + assert supervisor.metrics.swept == 1 + assert not (repository.artifact_root / "job-empty").exists() + + def test_cleanup_supervisor_skips_live_peer_and_local_overlap(tmp_path): repository = _repository(tmp_path) first = ExecutionCleanupSupervisor(repository) diff --git a/tests/test_execution_finish.py b/tests/test_execution_finish.py index 6fc65d0..a5f3470 100644 --- a/tests/test_execution_finish.py +++ b/tests/test_execution_finish.py @@ -5,7 +5,11 @@ from threading import Event from cortex_backend.execution.finish import UnsuccessfulJobWording, finish_unsuccessful_job -from cortex_backend.execution.repository import ExecutionRepository, ExecutionTransitionConflict +from cortex_backend.execution.repository import ( + ExecutionRepository, + ExecutionRepositoryError, + ExecutionTransitionConflict, +) WORDING = UnsuccessfulJobWording( cancelled_event="cancelled", @@ -178,6 +182,60 @@ def broken(job_id, **kwargs): assert "simulated persistence outage" not in caplog.text +def test_a_job_that_cannot_be_read_is_contained_like_a_job_that_cannot_be_written( + execution_repository, monkeypatch, caplog +): + """``get_job`` sat outside the ``try``, so a store failure escaped a function that says nothing raises. + + This runs in a worker's error path; an exception here replaces the outcome + the caller was about to report. + """ + + job = _running_job(execution_repository) + + def unreadable(job_id, **kwargs): + raise ExecutionRepositoryError("SQLite execution operation failed.") + + monkeypatch.setattr(execution_repository, "get_job", unreadable) + + with caplog.at_level("WARNING", logger="cortex.execution.finish"): + finish_unsuccessful_job( + execution_repository, + job.job_id, + failure_code="worker_failed", + cancel_requested=Event().is_set, + wording=WORDING, + ) + + monkeypatch.undo() + assert execution_repository.get_job(job.job_id).status == "running" # nothing was written + assert any("ExecutionRepositoryError" in record.getMessage() for record in caplog.records) + assert "SQLite execution operation failed" not in caplog.text + assert job.job_id not in caplog.text + + +def test_a_cancel_probe_that_raises_is_contained_and_logged_by_type_only( + execution_repository, caplog +): + job = _running_job(execution_repository) + + def probe() -> bool: + raise RuntimeError("simulated probe failure") + + with caplog.at_level("WARNING", logger="cortex.execution.finish"): + finish_unsuccessful_job( + execution_repository, + job.job_id, + failure_code="worker_failed", + cancel_requested=probe, + wording=WORDING, + ) + + assert execution_repository.get_job(job.job_id).status == "running" + assert any("RuntimeError" in record.getMessage() for record in caplog.records) + assert "simulated probe failure" not in caplog.text + + def test_the_scratch_profile_does_not_overwrite_a_stop_with_a_failure( coordinator, monkeypatch ): diff --git a/tests/test_execution_path_safety.py b/tests/test_execution_path_safety.py new file mode 100644 index 0000000..303bad2 --- /dev/null +++ b/tests/test_execution_path_safety.py @@ -0,0 +1,267 @@ +"""A job id names a directory, so it must be a name every platform can create. + +``create_job("nul")`` used to succeed and the first ``publish_artifact`` then +raised a raw ``OSError`` carrying the absolute path, because ``nul`` is the +Windows NUL device and not a directory name. ``a.`` shared a directory with +``a`` for the same reason: Windows drops trailing dots. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from cortex_backend.execution import repository as repository_module +from cortex_backend.execution.repository import ExecutionRepository, ExecutionRepositoryError + + +def _repository(tmp_path: Path) -> ExecutionRepository: + return ExecutionRepository(tmp_path / "execution.sqlite", tmp_path / "artifacts") + + +def _create(repository: ExecutionRepository, job_id: str): + return repository.create_job( + job_id=job_id, + owner=repository.installation_principal_id, + request_id=f"request-{job_id}", + profile="scratch.auto.v1", + payload={}, + ) + + +_REFUSED_JOB_IDS = ( + "nul", + "NUL", + "Nul", + "con", + "prn", + "aux", + "com1", + "COM9", + "lpt1", + "LPT9", + "nul.txt", + "con.tar.gz", + "aux.", # a device name and a trailing dot + "a.", + "a..", + "job-1.", + "x" * 199 + ".", + "x" * 201, + "", + ".hidden", + "-lead", + "a b", + "a/b", + "a\\b", + "..", + "a\n", +) + +_ACCEPTED_JOB_IDS = ( + "a", + "job-1", + "a.b", + "a..b", + "0123456789abcdef0123456789abcdef", + "null", + "console", + "com0", + "com10", + "lpt0", + "nul-1", + "nul_", + "auxiliary", + "x" * 200, +) + + +def _label(job_id: str) -> str: + return job_id if 0 < len(job_id) <= 24 else f"{len(job_id)}-chars-ending-{job_id[-1:]!r}" + + +@pytest.mark.parametrize("job_id", _REFUSED_JOB_IDS, ids=_label) +def test_a_job_id_that_is_not_a_portable_directory_name_is_refused_when_the_job_is_created( + tmp_path: Path, job_id: str +) -> None: + repository = _repository(tmp_path) + + with pytest.raises(ValueError, match="job_id"): + _create(repository, job_id) + + assert repository.list_jobs(owner=repository.installation_principal_id, include_terminal=True) == [] + + +@pytest.mark.parametrize("job_id", _ACCEPTED_JOB_IDS, ids=_label) +def test_ordinary_job_ids_are_still_accepted(tmp_path: Path, job_id: str) -> None: + repository = _repository(tmp_path) + + job, created = _create(repository, job_id) + artifact = repository.publish_artifact(job.job_id, name="out.txt", content=b"synthetic") + + assert created is True + assert repository.read_artifact(artifact.artifact_id) == b"synthetic" + + +@pytest.mark.parametrize("job_id", ("nul", "aux", "a.", "COM1")) +def test_publishing_for_a_job_that_was_stored_under_an_older_shape_touches_nothing( + tmp_path: Path, job_id: str +) -> None: + """A row an earlier build accepted is refused at publish, before any path is built.""" + + repository = _repository(tmp_path) + with repository.connect() as connection: + connection.execute( + """ + INSERT INTO execution_jobs + (job_id, owner, request_id, profile, status, sequence, payload_json, created_at, updated_at) + VALUES (?, ?, 'request-old', 'scratch.auto.v1', 'queued', 0, '{}', 'x', 'x') + """, + (job_id, repository.installation_principal_id), + ) + before = sorted(entry.name for entry in repository.artifact_root.iterdir()) + + with pytest.raises(ExecutionRepositoryError) as refused: + repository.publish_artifact(job_id, name="out.txt", content=b"synthetic") + + assert str(tmp_path) not in str(refused.value) + assert sorted(entry.name for entry in repository.artifact_root.iterdir()) == before + + +def test_a_directory_that_cannot_be_created_is_reported_without_its_path( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + repository = _repository(tmp_path) + job, _ = _create(repository, "job-1") + real_mkdir = Path.mkdir + + def refuses(self: Path, *args: object, **kwargs: object) -> None: + if self.name == "job-1": + raise PermissionError(13, "Access is denied", str(self)) + real_mkdir(self, *args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(Path, "mkdir", refuses) + + with pytest.raises(ExecutionRepositoryError) as failure: + repository.publish_artifact(job.job_id, name="out.txt", content=b"synthetic") + + assert str(tmp_path) not in str(failure.value) + assert failure.value.__cause__ is None + assert failure.value.__suppress_context__ is True + assert not (repository.artifact_root / "job-1").exists() + + +def test_a_file_that_cannot_be_opened_is_reported_without_its_path_and_leaves_nothing( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + repository = _repository(tmp_path) + job, _ = _create(repository, "job-1") + real_open = Path.open + + def refuses(self: Path, mode: str = "r", *args: object, **kwargs: object): + if self.name.startswith(".tmp-"): + raise PermissionError(13, "Access is denied", str(self)) + return real_open(self, mode, *args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(Path, "open", refuses) + + with pytest.raises(ExecutionRepositoryError) as failure: + repository.publish_artifact(job.job_id, name="out.txt", content=b"synthetic") + + assert str(tmp_path) not in str(failure.value) + assert failure.value.__cause__ is None + assert not (repository.artifact_root / "job-1").exists(), "the directory this call made was left behind" + + +def test_a_failing_cleanup_after_a_refused_write_does_not_replace_the_refusal( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The handler unlinks partial files; an OSError there must not become the error raised.""" + + repository = _repository(tmp_path) + job, _ = _create(repository, "job-1") + real_open = Path.open + real_unlink = Path.unlink + + def refuses(self: Path, mode: str = "r", *args: object, **kwargs: object): + if self.name.startswith(".tmp-"): + raise PermissionError(13, "Access is denied", str(self)) + return real_open(self, mode, *args, **kwargs) # type: ignore[arg-type] + + def cannot_unlink(self: Path, *args: object, **kwargs: object) -> None: + if self.parent.name == "job-1": + raise PermissionError(13, "Access is denied", str(self)) + real_unlink(self, *args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(Path, "open", refuses) + monkeypatch.setattr(Path, "unlink", cannot_unlink) + + with pytest.raises(ExecutionRepositoryError) as failure: + repository.publish_artifact(job.job_id, name="out.txt", content=b"synthetic") + + assert str(tmp_path) not in str(failure.value) + + +def _directory_link(link: Path, target: Path) -> None: + """Make ``link`` a symbolic link, or on Windows a junction, to ``target``; skip if neither can be made.""" + + try: + link.symlink_to(target, target_is_directory=True) + return + except (OSError, NotImplementedError): + pass + try: + import _winapi # type: ignore[import-not-found] + + _winapi.CreateJunction(str(target), str(link)) + except (ImportError, AttributeError, OSError): + pytest.skip("neither symbolic links nor junctions can be created here") + + +def test_a_job_directory_that_is_a_link_is_refused_before_anything_is_created( + tmp_path: Path, +) -> None: + repository = _repository(tmp_path) + job, _ = _create(repository, "job-1") + outside = tmp_path / "outside" + outside.mkdir() + _directory_link(repository.artifact_root / "job-1", outside) + + with pytest.raises(ExecutionRepositoryError, match="escaped"): + repository.publish_artifact(job.job_id, name="out.txt", content=b"synthetic") + + assert list(outside.iterdir()) == [] + + +def test_the_containment_check_alone_refuses_a_link_before_the_directory_is_touched( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """``planned.parent != root`` is the check that does not depend on link detection. + + Older interpreters cannot see a junction (``Path.is_junction`` arrived in + 3.12), so ``_is_reparse_point`` says no; the resolved location is then the + only thing that says the job directory is somewhere else. It must run + before ``mkdir``, which would otherwise be pointed at the link's target. + """ + + repository = _repository(tmp_path) + job, _ = _create(repository, "job-1") + outside = tmp_path / "outside" + outside.mkdir() + _directory_link(repository.artifact_root / "job-1", outside) + monkeypatch.setattr(repository_module, "_is_reparse_point", lambda _path: False) + made: list[Path] = [] + real_mkdir = Path.mkdir + + def recording_mkdir(self: Path, *args: object, **kwargs: object) -> None: + made.append(self) + real_mkdir(self, *args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(Path, "mkdir", recording_mkdir) + + with pytest.raises(ExecutionRepositoryError, match="escaped"): + repository.publish_artifact(job.job_id, name="out.txt", content=b"synthetic") + + assert made == [], "a refused job directory was still handed to mkdir" + assert list(outside.iterdir()) == [] diff --git a/tests/test_execution_relative_paths.py b/tests/test_execution_relative_paths.py new file mode 100644 index 0000000..c7622a6 --- /dev/null +++ b/tests/test_execution_relative_paths.py @@ -0,0 +1,465 @@ +"""Artifact paths are stored relative to the artifact root, additively. + +A row used to record the absolute path it was written with, so moving the data +directory made every existing row point at nothing: live attachments stopped +reading and expired ones were never reclaimed. New rows record their path +relative to the artifact root. This adds no table and no column, so the schema +version does not move: it stays 3, nothing is rewritten on upgrade -- rows an +earlier build wrote keep their absolute path -- and both forms are read. + +Leaving the version alone is what makes going back safe. A build that only +knows version 3 opens the same store, keeps every job, fails closed on reading +an artifact whose path it cannot resolve, and discards only the expired rows +it cannot reclaim; a version bump would have made it set the whole store aside. +""" + +from __future__ import annotations + +import os +from pathlib import Path, PurePosixPath +import shutil + +import pytest + +from cortex_backend.execution.repository import ( + SCHEMA_VERSION, + ExecutionRepository, + ExecutionRepositoryError, +) + +# The version the previous release build reads. It is spelled out, not taken from +# SCHEMA_VERSION, so that moving the constant fails these tests instead of following it. +_PREVIOUS_BUILDS_VERSION = 3 + +_LONG_AGO = "2000-01-01T00:00:00+00:00" + + +def _open(data: Path) -> ExecutionRepository: + return ExecutionRepository(data / "execution.sqlite", data / "artifacts") + + +def _job(repository: ExecutionRepository, job_id: str) -> str: + repository.create_job( + job_id=job_id, + owner=repository.installation_principal_id, + request_id=f"request-{job_id}", + profile="scratch.auto.v1", + payload={}, + ) + repository.transition(job_id, status="succeeded", event="completed") + return job_id + + +def _publish(repository: ExecutionRepository, job_id: str, name: str = "out.txt", *, expired: bool = False): + artifact = repository.publish_artifact( + _job(repository, job_id), name=name, content=b"synthetic", mime_type="text/plain" + ) + if expired: + with repository.connect() as connection: + connection.execute( + "UPDATE execution_artifacts SET expires_at = ? WHERE artifact_id = ?", + (_LONG_AGO, artifact.artifact_id), + ) + return artifact + + +def _cleanup(repository: ExecutionRepository): + """One retention pass at the real time: only rows made expired on purpose are due.""" + + return repository.cleanup_expired() + + +def _rows(repository: ExecutionRepository, sql: str, *params: object): + with repository.connect() as connection: + return connection.execute(sql, params).fetchall() + + +def _artifact_row(repository: ExecutionRepository, artifact_id: str): + with repository.connect() as connection: + return connection.execute( + "SELECT path FROM execution_artifacts WHERE artifact_id = ?", (artifact_id,) + ).fetchone() + + +def _tombstones(repository: ExecutionRepository) -> list[tuple[str, str, str]]: + with repository.connect() as connection: + return [ + (row["path"], row["quarantine_path"], row["state"]) + for row in connection.execute( + "SELECT path, quarantine_path, state FROM execution_artifact_cleanup ORDER BY artifact_id" + ).fetchall() + ] + + +def _leave_a_pending_tombstone( + repository: ExecutionRepository, monkeypatch: pytest.MonkeyPatch, *, stuck: int = 1 +) -> None: + """Run a cleanup whose move into quarantine is refused, so the tombstone stays pending. + + ``stuck`` is how many tombstones the pass is expected to be unable to move: the + new one and any that an earlier call left behind. + """ + + def refused(*_args: object, **_kwargs: object) -> Path: + raise PermissionError(13, "held open by a scanner") + + with monkeypatch.context() as blocked: + blocked.setattr(Path, "replace", refused) + result = _cleanup(repository) + assert (result.artifacts, result.skipped) == (0, stuck) + + +def _as_the_previous_build_wrote_it(repository: ExecutionRepository) -> None: + """Rewrite every path that is still relative the way the previous build recorded it: absolute.""" + + root = repository.artifact_root + with repository.connect() as connection: + for row in connection.execute("SELECT artifact_id, path FROM execution_artifacts").fetchall(): + connection.execute( + "UPDATE execution_artifacts SET path = ? WHERE artifact_id = ?", + (str(root / row["path"]), row["artifact_id"]), + ) + for row in connection.execute( + "SELECT artifact_id, path, quarantine_path FROM execution_artifact_cleanup" + ).fetchall(): + connection.execute( + "UPDATE execution_artifact_cleanup SET path = ?, quarantine_path = ? WHERE artifact_id = ?", + (str(root / row["path"]), str(root / row["quarantine_path"]), row["artifact_id"]), + ) + + +def _version(repository: ExecutionRepository) -> int: + with repository.connect() as connection: + return int(connection.execute("SELECT version FROM execution_schema WHERE id = 1").fetchone()[0]) + + +# -- A store written by this build ------------------------------------------------------------- + + +def test_new_rows_name_their_files_relative_to_the_artifact_root(tmp_path): + repository = _open(tmp_path / "data") + artifact = _publish(repository, "job-1", "out.txt") + + stored = _artifact_row(repository, artifact.artifact_id)["path"] + + assert stored == f"job-1/{Path(artifact.path).name}" + assert not os.path.isabs(stored) and ":" not in stored and "\\" not in stored + # What a caller is handed is still a real location, as it always was. + assert Path(artifact.path) == repository.artifact_root / "job-1" / Path(artifact.path).name + reread = repository.get_artifact(artifact.artifact_id) + assert reread is not None and reread.path == artifact.path + assert repository.read_artifact(artifact.artifact_id) == b"synthetic" + + +def test_a_cleanup_tombstone_records_relative_paths(tmp_path, monkeypatch): + repository = _open(tmp_path / "data") + artifact = _publish(repository, "job-1", expired=True) + + _leave_a_pending_tombstone(repository, monkeypatch) + + ((path, quarantine, state),) = _tombstones(repository) + assert state == "pending" + assert path == f"job-1/{Path(artifact.path).name}" + assert PurePosixPath(quarantine).parent.as_posix() == ".quarantine" + assert quarantine.endswith(".artifact") and not os.path.isabs(quarantine) + assert Path(artifact.path).is_file(), "the refused move left the file where it was" + + result = _cleanup(repository) # the move is allowed again + assert (result.artifacts, result.skipped) == (1, 0) + assert not Path(artifact.path).exists() + assert _tombstones(repository) == [] + + +def test_a_moved_data_directory_keeps_live_and_expiring_artifacts_working(tmp_path, monkeypatch): + first = tmp_path / "data" + repository = _open(first) + live = _publish(repository, "job-live", "live.txt") + stuck = _publish(repository, "job-stuck", "stuck.txt", expired=True) + _leave_a_pending_tombstone(repository, monkeypatch) + waiting = _publish(repository, "job-waiting", "waiting.txt", expired=True) + old_location = Path(stuck.path) + + moved = tmp_path / "moved" + shutil.move(str(first), str(moved)) + reopened = _open(moved) + + # The live artifact still reads, and it reports where it is now. + assert reopened.read_artifact(live.artifact_id) == b"synthetic" + found = reopened.get_artifact(live.artifact_id) + assert found is not None and Path(found.path) == moved / "artifacts" / "job-live" / Path(live.path).name + # The pending tombstone and the expired row are both reclaimed, with nothing skipped. + result = _cleanup(reopened) + assert (result.artifacts, result.skipped) == (2, 0) + assert reopened.get_artifact(stuck.artifact_id) is None + assert reopened.get_artifact(waiting.artifact_id) is None + assert not (moved / "artifacts" / "job-stuck").exists() + assert not (moved / "artifacts" / "job-waiting").exists() + assert not old_location.exists() + assert reopened.read_artifact(live.artifact_id) == b"synthetic" + assert _tombstones(reopened) == [] + + +# -- A store written by the previous build ------------------------------------------------------ + + +def test_a_store_the_previous_build_wrote_is_read_as_it_was_and_is_not_rewritten(tmp_path, monkeypatch): + data = tmp_path / "data" + repository = _open(data) + live = _publish(repository, "job-live", "live.txt") + stuck = _publish(repository, "job-stuck", "stuck.txt", expired=True) + _leave_a_pending_tombstone(repository, monkeypatch) + _as_the_previous_build_wrote_it(repository) + assert _version(repository) == _PREVIOUS_BUILDS_VERSION + before = { + "artifacts": sorted( + (row["artifact_id"], row["path"]) + for row in _rows(repository, "SELECT artifact_id, path FROM execution_artifacts") + ), + "tombstones": _tombstones(repository), + } + assert all(os.path.isabs(path) for _id, path in before["artifacts"]) + + upgraded = _open(data) + + assert _version(upgraded) == _PREVIOUS_BUILDS_VERSION + after = { + "artifacts": sorted( + (row["artifact_id"], row["path"]) + for row in _rows(upgraded, "SELECT artifact_id, path FROM execution_artifacts") + ), + "tombstones": _tombstones(upgraded), + } + assert after == before, "the upgrade rewrote rows it had no need to touch" + # Both forms read, and an absolute row reports the path it holds. + assert upgraded.read_artifact(live.artifact_id) == b"synthetic" + found = upgraded.get_artifact(live.artifact_id) + assert found is not None and found.path == dict(before["artifacts"])[live.artifact_id] + result = _cleanup(upgraded) + assert (result.artifacts, result.skipped) == (1, 0) + assert upgraded.get_artifact(stuck.artifact_id) is None + assert not Path(stuck.path).exists() + assert _tombstones(upgraded) == [] + # A row written by this build is relative, next to the absolute one. + fresh = _publish(upgraded, "job-fresh", "fresh.txt") + assert not os.path.isabs(_artifact_row(upgraded, fresh.artifact_id)["path"]) + assert os.path.isabs(_artifact_row(upgraded, live.artifact_id)["path"]) + + +def test_a_store_holding_absolute_and_relative_rows_opens_and_serves_and_reclaims_both(tmp_path, monkeypatch): + """The store a rollback and a second upgrade leave: rows of both forms, live and expired, + and tombstones of both forms waiting in the same queue.""" + + data = tmp_path / "data" + repository = _open(data) + old_live = _publish(repository, "job-old-live", "old-live.txt") + old_expired = _publish(repository, "job-old-expired", "old-expired.txt", expired=True) + _leave_a_pending_tombstone(repository, monkeypatch) + _as_the_previous_build_wrote_it(repository) # everything so far is absolute + new_live = _publish(repository, "job-new-live", "new-live.txt") + new_expired = _publish(repository, "job-new-expired", "new-expired.txt", expired=True) + _leave_a_pending_tombstone(repository, monkeypatch, stuck=2) # the old tombstone is retried too + old, new = (old_live, old_expired), (new_live, new_expired) + assert all(os.path.isabs(_artifact_row(repository, a.artifact_id)["path"]) for a in old) + assert not any(os.path.isabs(_artifact_row(repository, a.artifact_id)["path"]) for a in new) + forms = sorted(os.path.isabs(path) for path, _quarantine, _state in _tombstones(repository)) + assert forms == [False, True], "one tombstone of each form is waiting" + before = sorted( + (row["artifact_id"], row["path"]) + for row in _rows(repository, "SELECT artifact_id, path FROM execution_artifacts") + ) + + reopened = _open(data) + + assert _version(reopened) == _PREVIOUS_BUILDS_VERSION + assert ( + sorted( + (row["artifact_id"], row["path"]) + for row in _rows(reopened, "SELECT artifact_id, path FROM execution_artifacts") + ) + == before + ) + for live in (old_live, new_live): + assert reopened.read_artifact(live.artifact_id) == b"synthetic" + result = _cleanup(reopened) + assert (result.artifacts, result.skipped) == (2, 0) + for gone in (old_expired, new_expired): + assert reopened.get_artifact(gone.artifact_id) is None + assert not Path(gone.path).exists() + assert _tombstones(reopened) == [] + for live in (old_live, new_live): + assert reopened.read_artifact(live.artifact_id) == b"synthetic" + + +def test_new_rows_leave_the_schema_version_where_the_previous_build_reads_it(tmp_path): + """Going back must not cost the store: a build that knows only version 3 keeps a version-3 store. + + A store stamped with a version an older build does not know is set aside whole, + jobs and installation principal included, so writing relative paths may not + move the stamp. + """ + + assert SCHEMA_VERSION == _PREVIOUS_BUILDS_VERSION + data = tmp_path / "data" + repository = _open(data) + artifact = _publish(repository, "job-1") + before = _artifact_row(repository, artifact.artifact_id)["path"] + assert not os.path.isabs(before) + assert _version(repository) == _PREVIOUS_BUILDS_VERSION + + again = _open(data) + + assert _version(again) == _PREVIOUS_BUILDS_VERSION + assert _artifact_row(again, artifact.artifact_id)["path"] == before + assert again.read_artifact(artifact.artifact_id) == b"synthetic" + assert again.get_job("job-1") is not None + assert again.installation_principal_id == repository.installation_principal_id + + +def test_an_absolute_row_in_a_moved_data_directory_is_contained_as_it_always_was(tmp_path): + """The limitation is only for rows an earlier build wrote: they are refused, not followed.""" + + first = tmp_path / "data" + repository = _open(first) + artifact = _publish(repository, "job-1", expired=True) + _as_the_previous_build_wrote_it(repository) + moved = tmp_path / "moved" + shutil.move(str(first), str(moved)) + reopened = _open(moved) + survivor = moved / "artifacts" / "job-1" / Path(artifact.path).name + + with pytest.raises(ExecutionRepositoryError): + reopened.read_artifact(artifact.artifact_id) + result = _cleanup(reopened) + + assert (result.artifacts, result.skipped) == (0, 1) + assert reopened.get_artifact(artifact.artifact_id) is None # the unusable row is dropped + assert survivor.is_file() # and nothing outside a validated location was touched + + +# -- Failure paths: a relative row is confined exactly as an absolute one is ---------------------- + + +_ESCAPES = ( + "../outside.txt", + pytest.param( + "..\\outside.txt", + marks=pytest.mark.skipif(os.name != "nt", reason="a backslash is a separator only on Windows"), + ), + "job-1/../../outside.txt", + "./../outside.txt", +) + + +@pytest.mark.parametrize("stored", _ESCAPES) +def test_a_relative_artifact_row_that_climbs_out_of_the_root_is_refused_and_touches_nothing(tmp_path, stored): + repository = _open(tmp_path / "data") + outside = tmp_path / "data" / "outside.txt" + outside.write_bytes(b"not ours") + artifact = _publish(repository, "job-1", expired=True) + with repository.connect() as connection: + connection.execute( + "UPDATE execution_artifacts SET path = ? WHERE artifact_id = ?", (stored, artifact.artifact_id) + ) + + with pytest.raises(ExecutionRepositoryError): + repository.read_artifact(artifact.artifact_id) + with pytest.raises(ExecutionRepositoryError): + repository.delete_artifact(artifact.artifact_id) + assert outside.read_bytes() == b"not ours" + result = _cleanup(repository) + + assert (result.artifacts, result.skipped) == (0, 1) + assert outside.read_bytes() == b"not ours" + assert repository.get_artifact(artifact.artifact_id) is None + + +@pytest.mark.skipif(os.name != "nt", reason="drive-relative and rooted paths are a Windows form") +@pytest.mark.parametrize("form", ("rooted", "rooted-forward", "other-drive")) +def test_a_drive_relative_or_rooted_row_is_not_followed_out_of_the_root(tmp_path, form): + """Joining one of these to the root must not land anywhere but under it. + + A rooted path keeps only the root's drive, and a path on another drive + replaces the root altogether; either way it ends up outside and is refused. + """ + + other_drive = "Y:" if tmp_path.drive.upper() == "Z:" else "Z:" + stored = { + "rooted": "\\outside.txt", + "rooted-forward": "/outside.txt", + "other-drive": f"{other_drive}outside.txt", + }[form] + repository = _open(tmp_path / "data") + artifact = _publish(repository, "job-1", expired=True) + with repository.connect() as connection: + connection.execute( + "UPDATE execution_artifacts SET path = ? WHERE artifact_id = ?", (stored, artifact.artifact_id) + ) + + with pytest.raises(ExecutionRepositoryError): + repository.read_artifact(artifact.artifact_id) + result = _cleanup(repository) + + assert (result.artifacts, result.skipped) == (0, 1) + + +@pytest.mark.parametrize("stored", ("../outside.txt", ".quarantine/foreign.artifact", "job-1")) +def test_a_relative_tombstone_that_names_something_it_may_not_touch_is_discarded_untouched(tmp_path, stored): + """Outside the root, inside quarantine (which is not an artifact) or a directory.""" + + repository = _open(tmp_path / "data") + outside = tmp_path / "data" / "outside.txt" + outside.write_bytes(b"not ours") + foreign = repository.quarantine_root / "foreign.artifact" + foreign.write_bytes(b"belongs to another tombstone") + artifact = _publish(repository, "job-1", expired=True) + with repository.connect() as connection: + connection.execute( + """ + INSERT INTO execution_artifact_cleanup (artifact_id, path, quarantine_path, state, created_at) + VALUES (?, ?, '.quarantine/own.artifact', 'pending', ?) + """, + (artifact.artifact_id, stored, _LONG_AGO), + ) + + result = _cleanup(repository) + + assert result.skipped == 1 and result.artifacts == 0 + assert _tombstones(repository) == [] + assert outside.read_bytes() == b"not ours" + assert foreign.read_bytes() == b"belongs to another tombstone" + assert Path(artifact.path).parent.is_dir() + + +def test_a_relative_quarantine_path_outside_quarantine_is_never_unlinked(tmp_path): + repository = _open(tmp_path / "data") + victim = repository.artifact_root / "job-other" / "keep.txt" + victim.parent.mkdir() + victim.write_bytes(b"live artifact bytes") + artifact = _publish(repository, "job-1", expired=True) + with repository.connect() as connection: + connection.execute( + """ + INSERT INTO execution_artifact_cleanup (artifact_id, path, quarantine_path, state, created_at) + VALUES (?, ?, ?, 'finalized', ?) + """, + (artifact.artifact_id, f"job-1/{Path(artifact.path).name}", "job-other/keep.txt", _LONG_AGO), + ) + + _cleanup(repository) + + assert victim.read_bytes() == b"live artifact bytes" + + +# -- The two conversions ---------------------------------------------------------------------- + + +def test_the_path_conversions_round_trip_and_leave_foreign_absolute_paths_alone(tmp_path): + repository = _open(tmp_path / "data") + inside = repository.artifact_root / "job-1" / "file.bin" + foreign = tmp_path / "elsewhere" / "file.bin" + + assert repository._stored_text(inside) == "job-1/file.bin" + assert repository._stored_path("job-1/file.bin") == inside + assert repository._stored_text(foreign) == str(foreign) + assert repository._stored_path(str(foreign)) == foreign + assert repository._stored_path(repository._stored_text(inside)) == inside diff --git a/tests/test_installation_principal.py b/tests/test_installation_principal.py index e48722d..84fd25e 100644 --- a/tests/test_installation_principal.py +++ b/tests/test_installation_principal.py @@ -114,6 +114,45 @@ def test_ambiguous_legacy_owner_migration_fails_closed_without_rewriting_jobs(tm assert owners == {"session-a", "session-b"} +def test_a_current_store_is_not_put_through_the_owner_migration_again(tmp_path): + """Two owners sharing a request id is legal at version 3; only a version-2 store is ambiguous. + + The step that folds every owner into the installation principal runs for a + store older than the current version and for no other. If it ran on every + open, this store would be refused as ambiguous, or its owners rewritten. + """ + + database = tmp_path / "execution.sqlite" + artifacts = tmp_path / "artifacts" + repository = ExecutionRepository(database, artifacts) + principal = repository.installation_principal_id + for job_id, owner in (("current-a", "session-a"), ("current-b", "session-b")): + repository.create_job( + job_id=job_id, + owner=owner, + request_id="same-request", + profile="fake.v1", + payload={"provider": "fake-v1"}, + ) + with repository.connect() as connection: + assert connection.execute( + "SELECT version FROM execution_schema WHERE id = 1" + ).fetchone()[0] == 3 + + reopened = ExecutionRepository(database, artifacts) # must not raise "ambiguous" + + assert reopened.installation_principal_id == principal + with reopened.connect() as connection: + owners = { + row["job_id"]: row["owner"] + for row in connection.execute("SELECT job_id, owner FROM execution_jobs").fetchall() + } + assert connection.execute( + "SELECT version FROM execution_schema WHERE id = 1" + ).fetchone()[0] == 3 + assert owners == {"current-a": "session-a", "current-b": "session-b"} + + def test_restart_reattaches_execution_to_same_installation_principal(tmp_path): database = tmp_path / "execution.sqlite" artifacts = tmp_path / "artifacts" diff --git a/tests/test_recipe_coordinator.py b/tests/test_recipe_coordinator.py index 64c77df..4cb8870 100644 --- a/tests/test_recipe_coordinator.py +++ b/tests/test_recipe_coordinator.py @@ -200,6 +200,27 @@ def test_coordinator_rejects_request_conflict_and_wrong_owner_artifact(tmp_path: assert missing.value.code == "input_artifact_unavailable" +def test_the_source_artifact_is_read_as_its_owner_never_unscoped(tmp_path: Path, monkeypatch): + repository, _source_job_id, source_artifact_id = _repository(tmp_path) + coordinator = RecipeExecutionCoordinator(repository, lambda _job: _FakeAttempt(_output())) + real_read = repository.read_artifact + owners: list[str | None] = [] + + def recording(artifact_id, *, owner=None): + owners.append(owner) + return real_read(artifact_id, owner=owner) + + monkeypatch.setattr(repository, "read_artifact", recording) + + assert coordinator._load_input(OWNER, source_artifact_id) == _image_bytes() + assert owners == [OWNER] + + with pytest.raises(RecipeExecutionError) as foreign: + coordinator._load_input("b" * 64, source_artifact_id) + assert foreign.value.code == "input_artifact_unavailable" + assert owners == [OWNER], "another owner's artifact was read before it was refused" + + def test_coordinator_failure_is_redacted_and_does_not_publish(tmp_path: Path): repository, _source_job_id, source_artifact_id = _repository(tmp_path) attempt = _FakeAttempt(error="provider_failed")