diff --git a/.env.example b/.env.example index a773847..1d8093d 100644 --- a/.env.example +++ b/.env.example @@ -20,3 +20,8 @@ # CLAMAV_HOST= empty: no virus scan. Set it and start the stack with --profile clamav # CLAMAV_PORT=3310 # CLAMAV_TIMEOUT_SECONDS=30 +# +# Archive location (phase 10, Gitea issue 7). See docs/DEPLOY.md for the share: +# ARCHIVE_ROOT= empty: the archive is off. Set it to the mounted archive share, e.g. /archive. +# The Admin picks a folder under it (Admin, File locations); filed packets are +# copied there by the archive-copy job. diff --git a/deploy/ci/ci.env b/deploy/ci/ci.env index 51cee16..69b8f23 100644 --- a/deploy/ci/ci.env +++ b/deploy/ci/ci.env @@ -59,6 +59,9 @@ SEAL_KEY_FILE=/run/secrets/seal_key ASSETS_DIR=/data/assets # Built frontend. Compose sets it too. STATIC_DIR=/app/static +# The archive share (Gitea issue 7). compose.ci.yaml mounts the ci-archive volume here, with /archive/packets +# writable by uid 10001. Empty turns the archive off. +ARCHIVE_ROOT=/archive # ---- LDAP (the throwaway Samba DC from deploy/samba, domain PSTEST.LOCAL; never PESCO's AD) ---- # The DC's alias on the backend network. LDAPS with the CA it generates on first start. diff --git a/deploy/compose.ci.yaml b/deploy/compose.ci.yaml index 9996580..e6d638a 100644 --- a/deploy/compose.ci.yaml +++ b/deploy/compose.ci.yaml @@ -36,12 +36,36 @@ services: volumes: - ci-dc-export:/etc/ssl/pstest:ro # the DC's CA, LDAP_CA_BUNDLE=/etc/ssl/pstest/ca.pem + # The archive share stand-in (Gitea issue 7): a volume at /archive (ARCHIVE_ROOT in ci.env) with the folder + # /archive/packets owned by uid 10001, so the e2e scenario can save "packets" under File locations. + archive-init: + build: *build + image: ${IMAGE:-packetstation:ci} + pull_policy: build + user: "0:0" + # Once per volume: root may not chmod the folder after it belongs to 10001 (no CAP_FOWNER). + entrypoint: + - sh + - -c + - test -d /archive/packets || { mkdir /archive/packets && chmod 0750 /archive/packets && chown 10001:10001 /archive/packets; } + volumes: + - ci-archive:/archive + network_mode: none + security_opt: + - no-new-privileges:true + cap_drop: + - ALL + cap_add: + - CHOWN + restart: "no" + app: build: *build image: ${IMAGE:-packetstation:ci} pull_policy: build volumes: - ci-dc-export:/etc/ssl/pstest:ro + - ci-archive:/archive # Playwright reaches the app as ps-app, never as "app": "app" is a top-level domain on # Chromium's HSTS preload list, so http://app:8000 is forced to TLS and fails. networks: @@ -52,6 +76,8 @@ services: depends_on: samba-dc: condition: service_healthy + archive-init: + condition: service_completed_successfully worker: build: *build @@ -60,6 +86,10 @@ services: volumes: - ci-dc-export:/etc/ssl/pstest:ro - ../tests/fixtures/generated:/fixtures:ro # synthetic feed and drawings; the scenario switches state + - ci-archive:/archive + depends_on: + archive-init: + condition: service_completed_successfully # Throwaway Samba AD DC (PSTEST.LOCAL) so LDAP sign-in runs over real TLS in CI. # Built from deploy/samba/. Runs as root, as Samba needs. No published port. @@ -127,6 +157,7 @@ services: volumes: ci-dc-data: {} ci-dc-export: {} + ci-archive: {} secrets: samba_admin_password: diff --git a/docs/FEED.md b/docs/FEED.md index 5e9b39a..8e70156 100644 --- a/docs/FEED.md +++ b/docs/FEED.md @@ -88,8 +88,9 @@ At start the worker runs the same stack guard as the app: the test stack refuses | `backup-check` | Daily at 07:15 site time | no | The newest dump is there, not empty, under 26 hours old, and not followed by a failure (phase 9). | | `notify-send` | Every 60 seconds | no | Rule seed, the version check, the watch checks, immediate delivery and the digest queue (phase 9). | | `notify-digest` | Every 5 minutes | no | Sends each digest that is due, one message per address (phase 9). | +| `archive-copy` | Every 15 minutes, plus a request from every filing, every folder save and "Retry now" | no | Copies each filed packet's data book, files and a manifest to the archive folder (phase 10). | -`job_schedule` holds each job's next run, interval and pause state for the Admin tab. Migration 0004 seeds rows for `visual-feed`, `drawing-lookup` and `housekeeping`; 0010 adds the six phase 9 jobs. +`job_schedule` holds each job's next run, interval and pause state for the Admin tab. Migration 0004 seeds rows for `visual-feed`, `drawing-lookup` and `housekeeping`; 0010 adds the six phase 9 jobs; 0012 adds `archive-copy`. ## Phase 9 jobs @@ -141,6 +142,34 @@ A failed or retried send makes the run degraded. For each rule whose digest time (site time) has passed today and that has outbox rows queued before it: recipients resolved per event, then grouped per address across every rule. One message per address lists every due event (`Daily digest: N notifications`), with one `notifications_sent` row per message (event type `digest`). An outbox row is marked sent only when every address got it. A failed address leaves the rows for the next run, which skips addresses that already got an event, and the job backs off 30 minutes (open question 43). Nothing queued: the run ends without touching AD. +## archive-copy (every 15 minutes, phase 10) + +Gitea issue 7. Not pausable. Off when `ARCHIVE_ROOT` is unset or blank: the run does nothing (counts `enabled: false`) and FILE asks for nothing. With the root set but no folder saved on Admin, File locations, the run idles (`skipped: no archive folder saved`). + +FILE inserts a `job_requests` row for `archive-copy` and NOTIFYs `ps_job_request` inside its own transaction when the archive is on and a folder is saved, so the copy starts right after the filing commits. The request is made in a savepoint: if it fails, the filing still commits and the next scheduled run copies the packet. A failed copy never blocks filing and never touches the packet. Saving a folder and "Retry now" also request a run. + +Each run: + +1. Checks the saved folder with the same rules and probe as the Admin's Check (below). A failed check raises `archive_failed`, makes the run degraded and copies nothing. The job audits its own check (`archive.checked`, source `job`) only when the result differs from the last check on record. +2. Queues every unbroken seal with a data book that has no `archive_copies` row, so turning the archive on later copies earlier filings too. +3. Copies the pending rows, then the failed rows whose `next_try_at` has passed, up to 200 per run. Each copy is audited (`archive.copied` or `archive.copy_failed`). A failed copy waits 5 minutes, then 15, then 60, then an hour between tries. "Retry now" makes every failed row due at once. +4. Raises `archive_failed` (condition, critical, subject `archive`) when any copy has failed 3 times or more, and resolves it when no copy is failed. + +Counts: `enabled`, `archive_dir`, `queued`, `copied`, `failed` (this run), `pending`, `failed_total`, `done_total`, and `check` (the message) when the folder check failed. A failed copy makes the run degraded. + +**The folder rules.** `ARCHIVE_ROOT` is a fixed path on the server (the NFS mount). The Admin enters the folder relative to it; an absolute path is accepted only under the root and is stored relative. Refused: empty, a `..` component, an absolute path elsewhere, the root itself, a NUL character, more than 400 characters, a folder that does not exist (nothing creates it), something that is not a folder, and a path whose real path (links followed) leaves the real root. **The check** writes `.packetstation-probe-` in the folder, fsyncs it, reads it back, compares and deletes it, and reports the free space. `packetstation doctor` runs the same check on the saved folder (SKIP when the archive is off or no folder is saved). + +**What is copied.** One folder per seal: `///-seal/`, for example `packets/L51234/L51234001/L51234-Lot001_D-5001_Manufacturing_Data_Book-seal7/`. In it: + +- the sealed data book PDF (the asset on the seal); +- the files of the packet's drawing pages (live and void, each file once); +- the packet documents that are not removed (NCRs and test reports included); +- the current Q.C. folder documents of the packet's unit and its project (what the data book's section 5 reads); +- the MTRs on the BOM lines; +- `manifest.json`: `packet_id`, `packet_label`, `project`, `unit`, `drawing`, `rev`, `seal_id`, `seal_hash`, `key_id`, `sealed_at`, `sealed_by`, `filed_at`, `filed_by`, `build`, and per file `name`, `kind` (databook, page, doc, qc_doc, mtr), `sha256`, `size`, `asset_id`. + +File names are the stored original names made safe (no separators or reserved characters) and unique in the folder (`name (2).pdf`). Each file is written to a temp name in the folder, fsynced, checked against the asset's sha256 and renamed; the manifest goes last. An existing identical file counts as done; a different file is never overwritten (the copy fails and names the file). A manifest already there for the same seal and files means the copy is done. Nothing in the archive is ever deleted. A refile after a reopen is a new seal and a new folder. + ### Events from the feed and the drawing job After each `visual-feed` and `drawing-lookup` run, `worker/runs.py` raises or resolves the feed conditions: @@ -171,6 +200,7 @@ After each `visual-feed` and `drawing-lookup` run, `worker/runs.py` raises or re | `qc_reject` | A Q.C. reject on a packet, an ITP line reject | one-shot, department | none | | `version_deployed` | notify-send, first run on a new build | one-shot | none | | `assets_quota` | The storage check after every upload and in housekeeping | condition | usage back under the threshold | +| `archive_failed` | archive-copy: a copy failed 3 times, or the folder check failed | condition, subject `archive` | no copy failed | A condition sends once when it opens and a "Cleared: " message to the same rule when it closes. Raised again while open, it only counts (`raised_count`). The department of an ITP reject is the one department among the unit's live packets that works to the line's page; none or several gives none. diff --git a/docs/SCHEMA.md b/docs/SCHEMA.md index 12052ac..3caf1be 100644 --- a/docs/SCHEMA.md +++ b/docs/SCHEMA.md @@ -353,7 +353,7 @@ Phase 3, migration 0004: - `job_requests`: one row per "Run now". `job`, `requested_at`, `requested_by`, `note`, `claimed_run_id` and `claimed_at` (set by the run that took it). Index on `job` where `claimed_at` is null. The app inserts the row and NOTIFYs `ps_job_request` in the same transaction. - `job_runs`: one row per run. `job`, `trigger` (schedule, request, chain, cli), `requested_by` uuid[] (the requests claimed), `started_at`, `finished_at`, `status` (running, ok, degraded, failed, skipped, abandoned), `counts` jsonb, `errors` jsonb (a list of redacted strings), `build_version`, `visual_target` (`"<server> / <database>"`), `host`. Index on (`job`, `id DESC`). -- `job_schedule`: one row per job (`job` PK). `next_run_at`, `interval_seconds`, `paused`, `updated_at`. Written by the worker after each run for the Admin tab. Seeded for `visual-feed`, `drawing-lookup` and `housekeeping`; 0010 adds `ad-reconcile`, `session-refresh`, `integrity`, `backup-check`, `notify-send` and `notify-digest`. +- `job_schedule`: one row per job (`job` PK). `next_run_at`, `interval_seconds`, `paused`, `updated_at`. Written by the worker after each run for the Admin tab. Seeded for `visual-feed`, `drawing-lookup` and `housekeeping`; 0010 adds `ad-reconcile`, `session-refresh`, `integrity`, `backup-check`, `notify-send` and `notify-digest`; 0012 adds `archive-copy`. - `worker_heartbeat`: a single row (`id` = 1 by CHECK). `at`, `build`, `host`, `pid`, `running` text[] (the jobs running now). Upserted every 30 seconds. ## Notifications (phases 4 and 9) @@ -389,6 +389,16 @@ Migration 0011 serves changes C7 and C8 in [PLAN.md](PLAN.md). The others need n - `packets.vclosed_at` and `vclosed_by` (no migration; Gitea issue 31, change C5): on a VISUAL packet the feed is now the one writer. It sets `vclosed_at` (with `vclosed_by` null) when every leg is closed or cancelled and clears both when a leg shows open again. On a hand packet Clerical still sets them. See [FEED.md](FEED.md), "The apply". - `qc_inspections.ncr` (no migration; Gitea issue 24, change C2) is now also written by the accept that carries the NCR number of an earlier reject. +## Archive location (phase 10, migration 0012) + +Gitea issue 7. Filed packets are copied to a folder on the archive share (see [FEED.md](FEED.md), "archive-copy"). + +- `archive_copies` (new): one row per filed seal to copy. `seal_id` (unique, references `packet_seals`), `packet_id`, `status` (pending, done or failed by CHECK), `attempts` (tries so far, the successful one included), `last_error` (the last failure, for the Admin), `next_try_at` (when a failed row is due again), `folder` (the copy's folder relative to `ARCHIVE_ROOT`, for example `packets/L51234/L51234001/L51234-Lot001_D-5001_Manufacturing_Data_Book-seal7`), `manifest_sha256` (sha256 of the `manifest.json` written there), `created_at`, `done_at`. A done row must have `folder` and `done_at` (CHECK). Indexed on (`next_try_at`, `id`) where not done, and on `packet_id`. `ps_app` has SELECT, INSERT and UPDATE; DELETE and TRUNCATE are revoked, so a copy record is never removed. The worker queues a row for every unbroken seal that has a data book and no row yet, so turning the archive on later also copies earlier filings. A refile after a reopen is a new seal and a new row. +- `job_schedule` gains the `archive-copy` row. +- Settings row `files` (no migration): `{"archive_dir": "<folder relative to ARCHIVE_ROOT>"}`, saved by the Admin on File locations with compare and set on `version`. No row, or no `archive_dir`, means no folder is saved and nothing is copied. +- Audit actions: `admin.file_locations_updated` (detail `archive_dir: {from, to}`), `archive.checked` (every check: `source` admin, save, doctor or job, with `archive_dir`, `ok`, `resolved`, `free_bytes`, `message`; the job records its own check only when the result changes), `archive.copied` (`seal_id`, `folder`, `files`, `manifest_sha256`), `archive.copy_failed` (`seal_id`, `attempts`, `error`), `admin.archive_retry` (the "Retry now" job request). +- Notification event type `archive_failed` (seeded by the code like the others; condition, subject `archive`). + ## VISUAL and drawing caches (phase 3) Migration 0004. Every ID part is text (`wo_base`, `lot`, `split`, `sub`, part IDs, order IDs), with `''` in place of null in a key. `seq` and `piece` are int. Closed sets are text with a CHECK. The worker fills these tables. Screens read them, never VISUAL. [FEED.md](FEED.md) says how each is written. diff --git a/server/packetstation/api/admin_files.py b/server/packetstation/api/admin_files.py new file mode 100644 index 0000000..eb2f96d --- /dev/null +++ b/server/packetstation/api/admin_files.py @@ -0,0 +1,141 @@ +"""Admin: File locations (Gitea issue 7). The archive folder under ARCHIVE_ROOT, its check, and the archive copies. + +Every route needs the admin role and a password session. Writes answer 409 `archive_off` while ARCHIVE_ROOT is unset. +The check and the save are audited (`archive.checked`, `admin.file_locations_updated`); "Retry now" is audited as +`admin.archive_retry` through the job request. +""" + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, Depends, Request +from fastapi.encoders import jsonable_encoder +from pydantic import BaseModel, ConfigDict, Field +from starlette.concurrency import run_in_threadpool +from starlette.responses import JSONResponse + +from ..domain import archive +from .deps import Db, Principal, admin_principal, get_db, settings_of +from .errors import ApiError +from .jobs import request_run +from .packets import WRITE + +router = APIRouter(prefix="/api/admin/file-locations", tags=["admin"]) + + +class CheckBody(BaseModel): + model_config = ConfigDict(extra="forbid") + + archive_dir: str = Field(max_length=archive.DIR_MAX + 100) + + +class PutBody(BaseModel): + model_config = ConfigDict(extra="forbid") + + archive_dir: str = Field(max_length=archive.DIR_MAX + 100) + version: int = Field(ge=0) + + +def _out(conn: Any, request: Request) -> dict[str, Any]: + root = archive.root_of(settings_of(request)) + archive_dir, version, _, _ = archive.load(conn) + resolved = None + if root is not None and archive_dir is not None: + try: + resolved = str(archive.resolve_dir(root, archive_dir)[1]) + except archive.PathRefused: + resolved = None + last = archive.last_check(conn) + return { + "enabled": root is not None, + "root": str(root) if root is not None else None, + "archive_dir": archive_dir, + "version": version, + "resolved": resolved, + "last_check": {k: last[k] for k in ("ok", "at", "by", "message")} if last else None, + "copies": archive.copies_status(conn), + } + + +def _root_or_409(request: Request) -> Any: + root = archive.root_of(settings_of(request)) + if root is None: + raise ApiError(409, "archive_off", archive.OFF_MESSAGE) + return root + + +@router.get("") +def file_locations( + request: Request, + p: Principal = Depends(admin_principal), # noqa: B008 + d: Db = Depends(get_db), # noqa: B008 +) -> dict[str, Any]: + return _out(d.conn, request) + + +@router.post("/check", dependencies=WRITE) +async def file_locations_check( + body: CheckBody, + request: Request, + p: Principal = Depends(admin_principal), # noqa: B008 + d: Db = Depends(get_db), # noqa: B008 +) -> dict[str, Any]: + root = _root_or_409(request) + rel, result = await run_in_threadpool(archive.check, root, body.archive_dir) + await run_in_threadpool( + archive.audit_check, d.conn, result, archive_dir=rel or body.archive_dir, source="admin", who=p + ) + if rel is None: + # The path rules refused it (not a failed probe): 422 with the reason, still audited above. + raise ApiError(422, "bad_path", result.message) + return result.as_dict() + + +@router.put("", dependencies=WRITE) +async def file_locations_put( + body: PutBody, + request: Request, + p: Principal = Depends(admin_principal), # noqa: B008 + d: Db = Depends(get_db), # noqa: B008 +) -> Any: + root = _root_or_409(request) + conn = d.conn + rel, result = await run_in_threadpool(archive.check, root, body.archive_dir) + await run_in_threadpool( + archive.audit_check, conn, result, archive_dir=rel or body.archive_dir, source="save", who=p + ) + if rel is None: + raise ApiError(422, "bad_path", result.message) + if not result.ok: + raise ApiError(422, "check_failed", result.message, extra={"check": result.as_dict()}) + + def save() -> dict[str, Any]: + try: + archive.save(conn, who=p, archive_dir=rel, version=body.version) + except archive.VersionConflict: + raise ApiError( + 409, + "version_conflict", + "Someone else changed the file locations. Reload and try again.", + extra={"current": jsonable_encoder(_out(conn, request))}, + ) from None + with conn.transaction(): + archive.request_copy(conn, p.user_id, "archive folder saved") + return _out(conn, request) + + return await run_in_threadpool(save) + + +@router.post("/retry", dependencies=WRITE) +def file_locations_retry( + request: Request, + p: Principal = Depends(admin_principal), # noqa: B008 + d: Db = Depends(get_db), # noqa: B008 +) -> JSONResponse: + _root_or_409(request) + conn = d.conn + with conn.transaction(): + reset = archive.reset_backoff(conn) + out = request_run(conn, p, archive.JOB, note="Retry now (File locations)", action="admin.archive_retry") + return JSONResponse(status_code=202, content={**out, "reset": reset}) diff --git a/server/packetstation/cli.py b/server/packetstation/cli.py index ec251f1..37b7bfc 100644 --- a/server/packetstation/cli.py +++ b/server/packetstation/cli.py @@ -213,7 +213,7 @@ def build_parser() -> argparse.ArgumentParser: m.set_defaults(func=cmd_migrate) d = sub.add_parser("doctor", help="Check configuration and every dependency") d.add_argument("--json", action="store_true", help="Print the checks as a JSON list") - d.add_argument("--skip", metavar="LIST", help="Comma list of: visual,mfiles,ldap,smtp,db,volume") + d.add_argument("--skip", metavar="LIST", help="Comma list of: visual,mfiles,ldap,smtp,db,volume,clamav,archive") d.add_argument("--send-test", metavar="ADDR", help="Also send one test message to ADDR") d.set_defaults(func=cmd_doctor) g = sub.add_parser("ad-groups", help="The AD groups Packet Station maps to roles") diff --git a/server/packetstation/doctor.py b/server/packetstation/doctor.py index ecb7dd3..414a98d 100644 --- a/server/packetstation/doctor.py +++ b/server/packetstation/doctor.py @@ -23,7 +23,7 @@ from . import __version__ from .settings import SecretFileError, Settings PASS, WARN, FAIL, SKIP = "PASS", "WARN", "FAIL", "SKIP" -SKIPPABLE = ("visual", "mfiles", "ldap", "smtp", "db", "volume", "clamav") +SKIPPABLE = ("visual", "mfiles", "ldap", "smtp", "db", "volume", "clamav", "archive") FREE_WARN_BYTES = 2 * 1024**3 EXPECTED_PS_GROUPS = 50 IN_CHAIN = "1.2.840.113556.1.4.1941" @@ -261,6 +261,46 @@ def check_volume(s: Settings) -> list[Check]: return out +# ---------------------------------------------------------------- archive + + +def check_archive(s: Settings) -> list[Check]: + """The archive folder check (probe write, read back, delete) against the saved folder, audited like the Admin's.""" + import psycopg + + from . import db + from .domain import archive + + root = archive.root_of(s) + if root is None: + return [Check("archive", SKIP, "ARCHIVE_ROOT not set: filed packets are not copied to an archive")] + try: + with db.connect_as("app", s, autocommit=True) as conn: + archive_dir = archive.load(conn)[0] + except (SecretFileError, psycopg.Error) as e: + return [Check("archive", FAIL, f"can't read the saved archive folder: {_first_line(e)}")] + if archive_dir is None: + return [Check("archive", SKIP, f"ARCHIVE_ROOT={root}, no archive folder saved (Admin, File locations)")] + _, result = archive.check(root, archive_dir) + note = "" + try: + with db.connect_as("app", s, autocommit=True) as conn: + archive.audit_check(conn, result, archive_dir=archive_dir, source="doctor") + except (SecretFileError, psycopg.Error) as e: + note = f" (not audited: {_first_line(e)})" + if not result.ok: + return [ + Check( + "archive", + FAIL, + f"{archive_dir} under {root}: {result.message}{note}", + "is the share mounted and writable by uid 10001?", + ) + ] + free = f", {result.free_bytes / 1024**3:.1f} GB free" if result.free_bytes is not None else "" + return [Check("archive", WARN if note else PASS, f"{result.resolved} writable{free}{note}")] + + # ---------------------------------------------------------------- clamav @@ -439,6 +479,7 @@ def run( ("db", check_db), ("volume", check_volume), ("clamav", check_clamav), + ("archive", check_archive), ("ldap", check_ldap), ("smtp", lambda st: check_smtp(st, send_test)), ] diff --git a/server/packetstation/domain/archive.py b/server/packetstation/domain/archive.py new file mode 100644 index 0000000..61571e0 --- /dev/null +++ b/server/packetstation/domain/archive.py @@ -0,0 +1,646 @@ +"""The archive location (Gitea issue 7): filed packets are copied to a folder under ARCHIVE_ROOT. + +ARCHIVE_ROOT is a fixed path on the server (an NFS mount). Unset or blank turns the feature off. The Admin picks the +archive folder as a path under the root; the `files` settings row stores it relative (`{"archive_dir": "packets"}`). +A path that leaves the root is refused: empty, `..`, an absolute path elsewhere, NUL, a link that resolves outside +the root, or a folder that does not exist. The check writes a probe file, reads it back and removes it. Nothing here +creates folders or deletes anything in the archive. + +FILE inserts a job request for `archive-copy` in its own transaction. The worker job (worker/jobs/archive_copy.py) +queues every unbroken seal with a data book that has no `archive_copies` row, then copies the pending and due failed +rows. A failed copy never blocks filing: it is retried with a backoff and raised to Admin as `archive_failed`. + +Folder layout under the archive folder: `<project>/<unit>/<data book name>-seal<seal id>/` with the data book PDF, +the packet's files (drawing page files, packet documents, the Q.C. folder documents of its unit and project, MTRs) +and `manifest.json`. Each file goes to a temp name, is fsynced and renamed. An existing identical file counts as +done; a different one is never overwritten. +""" + +from __future__ import annotations + +import contextlib +import hashlib +import json +import logging +import os +import re +import secrets +import shutil +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from datetime import datetime, timedelta +from pathlib import Path, PurePath +from typing import Any + +from psycopg.rows import dict_row +from psycopg.types.json import Jsonb + +from .. import audit +from ..settings import Settings + +log = logging.getLogger(__name__) + +KEY = "files" +JOB = "archive-copy" +JOB_CHANNEL = "ps_job_request" +EVENT = "archive_failed" +EVENT_SUBJECT = "archive" +PROBE_PREFIX = ".packetstation-probe-" +MANIFEST = "manifest.json" +DIR_MAX = 400 +FAILS_TO_RAISE = 3 +# Minutes before the next try after the 1st, 2nd, 3rd failure; hourly after that. +BACKOFF_MINUTES = (5, 15, 60) +RECENT_FAILURES = 20 +CHUNK = 1024 * 1024 +OFF_MESSAGE = "The archive is off. Set ARCHIVE_ROOT on the server first." + +_UNSAFE = re.compile(r'[\x00-\x1f\x7f<>:"/\\|?*]+') + + +class PathRefused(ValueError): + """An archive folder the rules refuse. The message is for the Admin.""" + + +class VersionConflict(Exception): + def __init__(self, version: int): + super().__init__(f"version is now {version}") + self.version = version + + +class CopyError(Exception): + """One copy failed. The message goes into archive_copies.last_error and the audit row.""" + + +@dataclass(frozen=True) +class CheckResult: + ok: bool + resolved: str | None + free_bytes: int | None + message: str + + def as_dict(self) -> dict[str, Any]: + return {"ok": self.ok, "resolved": self.resolved, "free_bytes": self.free_bytes, "message": self.message} + + +def root_of(settings: Settings) -> Path | None: + """ARCHIVE_ROOT, or None when the feature is off.""" + root = settings.archive_root + if root is None or not str(root).strip(): + return None + return Path(root) + + +# ---------------------------------------------------------------- path rules + + +def _parts(text: str) -> list[str]: + return [p for p in re.split(r"[\\/]+", text) if p not in ("", ".")] + + +def _is_absolute(text: str) -> bool: + return text.startswith(("/", "\\")) or PurePath(text).is_absolute() + + +def relative_dir(root: Path, archive_dir: str) -> str: + """The folder as a clean relative path under the root ("a/b"). Raises PathRefused. Lexical only.""" + text = (archive_dir or "").strip() + if not text: + raise PathRefused("Enter a folder under the archive root.") + if "\x00" in text: + raise PathRefused("The folder name holds a NUL character.") + if len(text) > DIR_MAX: + raise PathRefused(f"The folder path is longer than {DIR_MAX} characters.") + if ".." in re.split(r"[\\/]+", text): + raise PathRefused("The folder may not contain '..'.") + if _is_absolute(text): + root_parts = _parts(os.path.normpath(str(root))) + given = _parts(os.path.normpath(text)) + norm = (lambda xs: [x.lower() for x in xs]) if os.name == "nt" else (lambda xs: xs) + if norm(given[: len(root_parts)]) != norm(root_parts): + raise PathRefused(f"An absolute path must lie under the archive root {root}.") + rel = given[len(root_parts) :] + if not rel: + raise PathRefused("Pick a folder under the archive root, not the root itself.") + return "/".join(rel) + return "/".join(_parts(text)) + + +def resolve_dir(root: Path, archive_dir: str) -> tuple[str, Path]: + """(relative folder, real path) for an archive folder. Raises PathRefused when the rules refuse it.""" + rel = relative_dir(root, archive_dir) + try: + real_root = root.resolve(strict=True) + except OSError: + raise PathRefused(f"The archive root {root} is not there. Is the share mounted?") from None + if not real_root.is_dir(): + raise PathRefused(f"The archive root {root} is not a folder.") + target = root.joinpath(*rel.split("/")) + try: + real = target.resolve(strict=True) + except OSError: + raise PathRefused(f"The folder {rel} does not exist under {root}. Create it on the share first.") from None + if real != real_root and not real.is_relative_to(real_root): + raise PathRefused(f"The folder {rel} points outside the archive root.") + if real == real_root: + raise PathRefused("Pick a folder under the archive root, not the root itself.") + if not real.is_dir(): + raise PathRefused(f"{rel} is not a folder.") + return rel, real + + +def probe(folder: Path) -> CheckResult: + """Write a probe file in the folder, fsync, read it back, compare and remove it.""" + name = folder / f"{PROBE_PREFIX}{secrets.token_hex(8)}" + data = secrets.token_bytes(64) + try: + with open(name, "xb") as f: + f.write(data) + f.flush() + os.fsync(f.fileno()) + try: + back = name.read_bytes() + finally: + name.unlink() + except OSError as e: + with contextlib.suppress(OSError): + name.unlink() + return CheckResult(False, str(folder), None, f"Cannot write in {folder}: {e.strerror or type(e).__name__}.") + if back != data: + return CheckResult(False, str(folder), None, f"The probe file read back from {folder} differs.") + try: + free = int(shutil.disk_usage(folder).free) + except OSError: + free = None + return CheckResult(True, str(folder), free, "Write, read back and delete all worked.") + + +def check(root: Path | None, archive_dir: str) -> tuple[str | None, CheckResult]: + """(relative folder or None, result) for a candidate folder. Never raises for a refused path.""" + if root is None: + return None, CheckResult(False, None, None, OFF_MESSAGE) + try: + rel, real = resolve_dir(root, archive_dir) + except PathRefused as e: + return None, CheckResult(False, None, None, str(e)) + return rel, probe(real) + + +# ---------------------------------------------------------------- the settings row + + +def load(conn: Any) -> tuple[str | None, int, Any, Any]: + """(archive_dir or None, version, updated_at, updated_by). Version 0 when the row is missing.""" + row = conn.execute("SELECT value, version, updated_at, updated_by FROM settings WHERE key = %s", (KEY,)).fetchone() + if row is None: + return None, 0, None, None + value = row[0] if isinstance(row[0], dict) else {} + d = value.get("archive_dir") + return (str(d) if isinstance(d, str) and d.strip() else None), int(row[1]), row[2], row[3] + + +def save(conn: Any, *, who: Any, archive_dir: str, version: int) -> int: + """Compare-and-set the folder (already checked). Returns the new version. Audited with the old and new value.""" + with conn.transaction(): + row = conn.execute("SELECT value, version FROM settings WHERE key = %s FOR UPDATE", (KEY,)).fetchone() + cur_version = int(row[1]) if row else 0 + if int(version) != cur_version: + raise VersionConflict(cur_version) + value = dict(row[0]) if row and isinstance(row[0], dict) else {} + old = value.get("archive_dir") + value["archive_dir"] = archive_dir + new_version = conn.execute( + "INSERT INTO settings (key, value, version, updated_at, updated_by) VALUES (%s, %s, 1, now(), %s)" + " ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, version = settings.version + 1," + " updated_at = now(), updated_by = EXCLUDED.updated_by RETURNING version", + (KEY, Jsonb(value), who.user_id), + ).fetchone()[0] + audit.append( + conn, + action="admin.file_locations_updated", + user_id=who.user_id, + session_id=getattr(who, "session_id", None), + role_snapshot=getattr(who, "role_snapshot", None), + subject_type="settings", + subject_id=KEY, + detail={"archive_dir": {"from": old, "to": archive_dir}}, + ip=getattr(who, "ip", None), + ) + return int(new_version) + + +def audit_check(conn: Any, result: CheckResult, *, archive_dir: str | None, source: str, who: Any = None) -> None: + """One `archive.checked` audit row: source admin (the Check button), save, or job (the worker's own check).""" + audit.append( + conn, + action="archive.checked", + user_id=getattr(who, "user_id", None), + session_id=getattr(who, "session_id", None), + role_snapshot=getattr(who, "role_snapshot", None), + subject_type="archive", + subject_id=(archive_dir or "")[:200] or None, + detail={"source": source, "archive_dir": archive_dir, **result.as_dict()}, + ip=getattr(who, "ip", None), + ) + + +def last_check(conn: Any) -> dict[str, Any] | None: + """The newest check from the audit log: {ok, at, by, message, archive_dir, source}.""" + with conn.cursor(row_factory=dict_row) as cur: + r = cur.execute( + "SELECT e.at, e.detail, u.display_name FROM audit_events e LEFT JOIN users u ON u.id = e.user_id" + " WHERE e.action = 'archive.checked' ORDER BY e.id DESC LIMIT 1" + ).fetchone() + if r is None: + return None + d = r["detail"] or {} + return { + "ok": bool(d.get("ok")), + "at": r["at"], + "by": r["display_name"] or ("Worker" if d.get("source") == "job" else None), + "message": d.get("message") or "", + "archive_dir": d.get("archive_dir"), + "source": d.get("source"), + } + + +# ---------------------------------------------------------------- copies: status and requests + + +def copies_status(conn: Any) -> dict[str, Any]: + """{pending, failed, done, recent_failures: [...]}.""" + counts = {"pending": 0, "failed": 0, "done": 0} + for status, n in conn.execute("SELECT status, count(*) FROM archive_copies GROUP BY status").fetchall(): + counts[str(status)] = int(n) + with conn.cursor(row_factory=dict_row) as cur: + rows = cur.execute( + "SELECT c.seal_id, c.packet_id, c.attempts, c.last_error, c.next_try_at," + " pr.number AS project, u.lot, p.drawing" + " FROM archive_copies c JOIN packets p ON p.id = c.packet_id JOIN units u ON u.id = p.unit_id" + " JOIN projects pr ON pr.id = u.project_id" + " WHERE c.status = 'failed' ORDER BY c.next_try_at NULLS FIRST, c.id LIMIT %s", + (RECENT_FAILURES,), + ).fetchall() + return { + **counts, + "recent_failures": [ + { + "seal_id": r["seal_id"], + "packet_id": r["packet_id"], + "packet_label": packet_label(r["project"], r["lot"], r["drawing"]), + "attempts": r["attempts"], + "last_error": r["last_error"], + "next_try_at": r["next_try_at"], + } + for r in rows + ], + } + + +def stuck_failures(conn: Any) -> list[dict[str, Any]]: + """The failed copies that have failed FAILS_TO_RAISE times or more, oldest first.""" + with conn.cursor(row_factory=dict_row) as cur: + rows = cur.execute( + "SELECT c.seal_id, c.packet_id, c.attempts, c.last_error, pr.number AS project, u.lot, p.drawing" + " FROM archive_copies c JOIN packets p ON p.id = c.packet_id JOIN units u ON u.id = p.unit_id" + " JOIN projects pr ON pr.id = u.project_id" + " WHERE c.status = 'failed' AND c.attempts >= %s ORDER BY c.id", + (FAILS_TO_RAISE,), + ).fetchall() + return [{**r, "packet_label": packet_label(r["project"], r["lot"], r["drawing"])} for r in rows] + + +def packet_label(project: Any, lot: Any, drawing: Any) -> str: + return " ".join(x for x in (str(project or ""), f"Lot {lot}" if lot else "", str(drawing or "")) if x) + + +def request_copy(conn: Any, user_id: Any = None, note: str | None = None) -> None: + """Ask the worker for an archive-copy run: a job request and a NOTIFY, in the caller's transaction.""" + conn.execute("INSERT INTO job_requests (job, requested_by, note) VALUES (%s, %s, %s)", (JOB, user_id, note)) + conn.execute("SELECT ps_notify(%s, %s)", (JOB_CHANNEL, json.dumps({"job": JOB}, separators=(",", ":")))) + + +def on_filed(conn: Any, settings: Settings, user_id: Any, packet_id: int) -> bool: + """FILE's hook: request a prompt copy when the archive is on and a folder is saved. Never raises; a failure + here only means the copy waits for the next scheduled run. Returns whether a request went in.""" + if root_of(settings) is None: + return False + try: + with conn.transaction(): + if load(conn)[0] is None: + return False + request_copy(conn, user_id, f"filed packet {packet_id}") + return True + except Exception: + log.exception("archive copy request for packet %s not made; the scheduled run picks it up", packet_id) + return False + + +def reset_backoff(conn: Any) -> int: + """Make every failed copy due now. Returns how many.""" + return int(conn.execute("UPDATE archive_copies SET next_try_at = now() WHERE status = 'failed'").rowcount or 0) + + +def queue_missing(conn: Any) -> int: + """Queue every unbroken seal with a data book and no archive_copies row. Returns how many.""" + cur = conn.execute( + "INSERT INTO archive_copies (seal_id, packet_id)" + " SELECT s.id, s.packet_id FROM packet_seals s" + " WHERE s.broken_at IS NULL AND s.databook_asset_id IS NOT NULL" + " AND NOT EXISTS (SELECT 1 FROM archive_copies c WHERE c.seal_id = s.id)" + " ORDER BY s.id ON CONFLICT (seal_id) DO NOTHING" + ) + return int(cur.rowcount or 0) + + +def due_rows(conn: Any, limit: int) -> list[dict[str, Any]]: + with conn.cursor(row_factory=dict_row) as cur: + rows: list[dict[str, Any]] = cur.execute( + "SELECT * FROM archive_copies WHERE status = 'pending'" + " OR (status = 'failed' AND (next_try_at IS NULL OR next_try_at <= now()))" + " ORDER BY (status = 'failed'), id LIMIT %s", + (limit,), + ).fetchall() + return rows + + +def backoff(attempts: int) -> timedelta: + """The wait after the given number of failed attempts: 5, 15, 60 minutes, then hourly.""" + i = max(1, attempts) - 1 + return timedelta(minutes=BACKOFF_MINUTES[i] if i < len(BACKOFF_MINUTES) else 60) + + +def mark_done(conn: Any, row: Mapping[str, Any], folder: str, manifest_sha: str, files: int) -> None: + with conn.transaction(): + conn.execute( + "UPDATE archive_copies SET status = 'done', attempts = attempts + 1, last_error = NULL, next_try_at = NULL," + " folder = %s, manifest_sha256 = %s, done_at = now() WHERE id = %s", + (folder, manifest_sha, row["id"]), + ) + audit.append( + conn, + action="archive.copied", + subject_type="seal", + subject_id=str(row["seal_id"]), + packet_id=int(row["packet_id"]), + detail={"seal_id": row["seal_id"], "folder": folder, "files": files, "manifest_sha256": manifest_sha}, + ) + + +def mark_failed(conn: Any, row: Mapping[str, Any], error: str) -> int: + """Record one failed attempt. Returns the attempts so far.""" + attempts = int(row["attempts"] or 0) + 1 + error = " ".join(error.split())[:500] + with conn.transaction(): + conn.execute( + "UPDATE archive_copies SET status = 'failed', attempts = %s, last_error = %s," + " next_try_at = now() + %s WHERE id = %s", + (attempts, error, backoff(attempts), row["id"]), + ) + audit.append( + conn, + action="archive.copy_failed", + subject_type="seal", + subject_id=str(row["seal_id"]), + packet_id=int(row["packet_id"]), + detail={"seal_id": row["seal_id"], "attempts": attempts, "error": error}, + ) + return attempts + + +# ---------------------------------------------------------------- one copy + + +def safe_part(text: Any, fallback: str) -> str: + """One path component: no separators or reserved characters, no leading dots, at most 120 characters.""" + s = _UNSAFE.sub("_", str(text or "")).strip(" .") + return s[:120].strip(" .") or fallback + + +@dataclass +class Item: + asset_id: str + kind: str # databook, page, doc, qc_doc, mtr + name: str + sha256: str + size: int + path: Path + + +def _unique(name: str, taken: set[str]) -> str: + stem, dot, ext = name.rpartition(".") + if not dot: + stem, ext = name, "" + out, n = name, 2 + while out.lower() in taken or out.lower() == MANIFEST: + out = f"{stem} ({n}).{ext}" if ext else f"{stem} ({n})" + n += 1 + taken.add(out.lower()) + return out + + +def _asset_rows(conn: Any, ids: Iterable[Any]) -> dict[str, dict[str, Any]]: + keys = list(dict.fromkeys(str(i) for i in ids if i)) + if not keys: + return {} + with conn.cursor(row_factory=dict_row) as cur: + rows = cur.execute("SELECT * FROM assets WHERE id = ANY(%s::uuid[])", (keys,)).fetchall() + return {str(r["id"]): dict(r) for r in rows} + + +def packet_files(conn: Any, packet_id: int, unit_id: int, project_id: int) -> list[tuple[str, str]]: + """(asset id, kind) of the packet's files in a fixed order, each asset once: drawing page files (live and void), + packet documents (not removed), the current Q.C. folder documents of its unit and project, MTRs.""" + out: list[tuple[str, str]] = [] + q: list[tuple[str, str, tuple[Any, ...]]] = [ + ( + "page", + "SELECT asset_id FROM packet_pages WHERE packet_id = %s AND asset_id IS NOT NULL" + " GROUP BY asset_id ORDER BY min(n)", + (packet_id,), + ), + ( + "doc", + "SELECT asset_id FROM packet_docs WHERE packet_id = %s AND removed_at IS NULL ORDER BY added_at, id", + (packet_id,), + ), + ( + "qc_doc", + "SELECT asset_id FROM qc_docs WHERE superseded_by IS NULL AND (unit_id = %s OR project_id = %s)" + " ORDER BY (unit_id IS NULL), added_at, id", + (unit_id, project_id), + ), + ( + "mtr", + "SELECT mtr_asset_id FROM bom_lines WHERE packet_id = %s AND mtr_asset_id IS NOT NULL ORDER BY item", + (packet_id,), + ), + ] + seen: set[str] = set() + for kind, sql, params in q: + for (aid,) in conn.execute(sql, params).fetchall(): + key = str(aid) + if key not in seen: + seen.add(key) + out.append((key, kind)) + return out + + +def _sha_file(path: Path) -> str: + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(CHUNK), b""): + h.update(chunk) + return h.hexdigest() + + +def _fsync_dir(path: Path) -> None: + if os.name != "posix": + return + fd = os.open(str(path), os.O_RDONLY) + try: + os.fsync(fd) + finally: + os.close(fd) + + +def _write_once(folder: Path, name: str, source: Path | None, data: bytes | None, sha: str) -> None: + """Put one file in the folder: temp name, fsync, rename. Identical existing file: done. Different: refused.""" + target = folder / name + if target.exists(): + if target.is_file() and _sha_file(target) == sha: + return + raise CopyError(f"A different file is already at {name} in the archive; it was not overwritten.") + tmp = folder / f".{name[:80]}.tmp-{secrets.token_hex(6)}" + h = hashlib.sha256() + try: + with open(tmp, "xb") as out: + if data is not None: + out.write(data) + h.update(data) + else: + assert source is not None + with open(source, "rb") as src: + for chunk in iter(lambda: src.read(CHUNK), b""): + out.write(chunk) + h.update(chunk) + out.flush() + os.fsync(out.fileno()) + if h.hexdigest() != sha: + raise CopyError(f"{name} changed while it was copied (sha256 differs from the asset record).") + if target.exists(): + raise CopyError(f"A file appeared at {name} during the copy; it was not overwritten.") + os.rename(tmp, target) + finally: + with contextlib.suppress(OSError): + tmp.unlink() + + +def copy_seal(conn: Any, settings: Settings, base: Path, rel_base: str, seal_id: int) -> tuple[str, str, int]: + """Copy one filed seal into the archive folder `base` (its relative path `rel_base`). + Returns (folder relative to ARCHIVE_ROOT, manifest sha256, files). Raises CopyError.""" + from . import assets + + with conn.cursor(row_factory=dict_row) as cur: + s = cur.execute( + "SELECT s.*, p.drawing, p.rev, p.unit_id, u.lot, u.project_id, pr.number AS project," + " pr.number || u.lot AS unit_number, sb.display_name AS sealed_by_name," + " t.by AS filed_by, fb.display_name AS filed_by_name, t.at AS filed_at" + " FROM packet_seals s JOIN packets p ON p.id = s.packet_id JOIN units u ON u.id = p.unit_id" + " JOIN projects pr ON pr.id = u.project_id LEFT JOIN users sb ON sb.id = s.sealed_by" + " LEFT JOIN LATERAL (SELECT x.by, x.at FROM packet_transitions x WHERE x.packet_id = s.packet_id" + " AND x.action = 'file' AND x.at <= s.sealed_at + interval '1 minute' ORDER BY x.id DESC LIMIT 1) t" + " ON true" + " LEFT JOIN users fb ON fb.id = t.by WHERE s.id = %s", + (seal_id,), + ).fetchone() + if s is None: + raise CopyError(f"Seal {seal_id} is not there.") + if s["databook_asset_id"] is None: + raise CopyError(f"Seal {seal_id} has no data book.") + rows = _asset_rows(conn, [s["databook_asset_id"]]) + book = rows.get(str(s["databook_asset_id"])) + if book is None: + raise CopyError("The data book is not in the asset store.") + book_name = assets.safe_name(book.get("original_name")) or "Data_Book.pdf" + stem = book_name[:-4] if book_name.lower().endswith(".pdf") else book_name + rel_parts = [ + safe_part(s["project"], "project"), + safe_part(s["unit_number"], "unit"), + safe_part(f"{stem}-seal{seal_id}", f"seal{seal_id}"), + ] + folder = base.joinpath(*rel_parts) + + ids = [(str(s["databook_asset_id"]), "databook")] + ids += [ + x for x in packet_files(conn, int(s["packet_id"]), int(s["unit_id"]), int(s["project_id"])) if x[0] != ids[0][0] + ] + rows.update(_asset_rows(conn, [a for a, _ in ids[1:]])) + taken: set[str] = set() + items: list[Item] = [] + for aid, kind in ids: + r = rows.get(aid) + if r is None: + raise CopyError(f"Asset {aid} ({kind}) is not in the asset store.") + try: + path = assets.open_blob(settings, r) + except Exception: + raise CopyError(f"The stored file of asset {aid} ({kind}) is missing.") from None + name = assets.safe_name(r.get("original_name")) or f"{aid}{assets.EXT.get(str(r['mime']), '')}" + if kind == "databook": + name = book_name + items.append(Item(aid, kind, _unique(name, taken), bytes(r["sha256"]).hex(), int(r["size"]), path)) + + manifest = { + "packet_id": int(s["packet_id"]), + "packet_label": packet_label(s["project"], s["lot"], s["drawing"]), + "project": s["project"], + "unit": s["unit_number"], + "drawing": s["drawing"], + "rev": s["rev"], + "seal_id": int(seal_id), + "seal_hash": bytes(s["hash"]).hex(), + "key_id": s["key_id"], + "sealed_at": _iso(s["sealed_at"]), + "sealed_by": s["sealed_by_name"], + "filed_at": _iso(s["filed_at"] or s["sealed_at"]), + "filed_by": s["filed_by_name"] or s["sealed_by_name"], + "build": settings.ps_build, + "files": [ + {"name": i.name, "kind": i.kind, "sha256": i.sha256, "size": i.size, "asset_id": i.asset_id} for i in items + ], + } + data = (json.dumps(manifest, indent=2, ensure_ascii=False) + "\n").encode("utf-8") + try: + folder.mkdir(parents=True, exist_ok=True) + existing = folder / MANIFEST + if existing.exists(): + same = _same_manifest(existing, manifest) + if same is not None: + return "/".join([rel_base, *rel_parts]), same, len(items) + raise CopyError(f"A different {MANIFEST} is already in {'/'.join(rel_parts)}; it was not overwritten.") + for i in items: + _write_once(folder, i.name, i.path, None, i.sha256) + sha = hashlib.sha256(data).hexdigest() + _write_once(folder, MANIFEST, None, data, sha) + _fsync_dir(folder) + except OSError as e: + raise CopyError(f"Cannot write in the archive: {e.strerror or type(e).__name__}.") from None + return "/".join([rel_base, *rel_parts]), sha, len(items) + + +def _same_manifest(path: Path, manifest: Mapping[str, Any]) -> str | None: + """The sha256 of the manifest on disk when it describes the same copy (the build may differ), else None.""" + try: + raw = path.read_bytes() + old = json.loads(raw) + except (OSError, ValueError): + return None + keys = ("packet_id", "seal_id", "seal_hash", "files") + if isinstance(old, dict) and all(old.get(k) == manifest.get(k) for k in keys): + return hashlib.sha256(raw).hexdigest() + return None + + +def _iso(v: Any) -> str | None: + return v.isoformat() if isinstance(v, datetime) else (str(v) if v is not None else None) diff --git a/server/packetstation/domain/pipeline.py b/server/packetstation/domain/pipeline.py index 0c46b38..3e32f35 100644 --- a/server/packetstation/domain/pipeline.py +++ b/server/packetstation/domain/pipeline.py @@ -850,6 +850,11 @@ def _file(w: Write) -> None: w.log_also("Data book filed", f"{pages} pages") w.detail["seal_id"] = int(seal_id) w.detail["databook_pages"] = pages + # The archive copy runs in the worker after this commits. A failed copy never blocks filing. + from . import archive + + if archive.on_filed(w.conn, _settings(w), w.user_id, w.id): + w.detail["archive_requested"] = True def _file_databook(w: Write, row: Mapping[str, Any], seal: dict[str, Any], seal_id: int) -> int: diff --git a/server/packetstation/main.py b/server/packetstation/main.py index 574c2b4..98f777f 100644 --- a/server/packetstation/main.py +++ b/server/packetstation/main.py @@ -97,6 +97,7 @@ def create_app( from .api import ( account, admin, + admin_files, admin_org, admin_security, assets, @@ -125,6 +126,7 @@ def create_app( admin, admin_org, admin_security, + admin_files, departments, events_api, projects, diff --git a/server/packetstation/migrations/0012_archive.sql b/server/packetstation/migrations/0012_archive.sql new file mode 100644 index 0000000..f1db745 --- /dev/null +++ b/server/packetstation/migrations/0012_archive.sql @@ -0,0 +1,32 @@ +-- 0012_archive: phase 10. The archive location (Gitea issue 7). Runs as ps_migrate; 0001's default privileges give +-- ps_app SELECT, INSERT, UPDATE, DELETE on the new table. + +-- One row per filed seal to copy to the archive folder. The archive-copy job queues every unbroken seal with a data +-- book that has no row yet, then copies the pending rows and the failed rows that are due (next_try_at). +-- folder is the copy's folder relative to ARCHIVE_ROOT, set when the copy is done. manifest_sha256 is the sha256 of +-- the manifest.json written there. A copy is never removed; a refile after a reopen has its own seal and row. +CREATE TABLE archive_copies ( + id bigserial PRIMARY KEY, + seal_id bigint NOT NULL UNIQUE REFERENCES packet_seals (id), + packet_id bigint NOT NULL REFERENCES packets (id), + status text NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'done', 'failed')), + attempts int NOT NULL DEFAULT 0, + last_error text, + next_try_at timestamptz, + folder text, + manifest_sha256 text, + created_at timestamptz NOT NULL DEFAULT now(), + done_at timestamptz, + CHECK (status <> 'done' OR (folder IS NOT NULL AND done_at IS NOT NULL)) +); +CREATE INDEX archive_copies_due ON archive_copies (next_try_at, id) WHERE status <> 'done'; +CREATE INDEX archive_copies_packet ON archive_copies (packet_id); + +-- The job, so "Retry now" and FILE's request are accepted before the worker has started once. The notification +-- rule for archive_failed is seeded by the code (notify/seeding.py) like every other event type. +INSERT INTO job_schedule (job) VALUES ('archive-copy') ON CONFLICT (job) DO NOTHING; + +-- Explicit, after the default privileges: the app reads, queues and updates copies; it never deletes one. +REVOKE DELETE, TRUNCATE ON archive_copies FROM ps_app; +GRANT SELECT, INSERT, UPDATE ON archive_copies TO ps_app; +GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO ps_app; diff --git a/server/packetstation/notify/catalog.py b/server/packetstation/notify/catalog.py index 2ebae9f..764d467 100644 --- a/server/packetstation/notify/catalog.py +++ b/server/packetstation/notify/catalog.py @@ -72,6 +72,12 @@ _TYPES = ( ), EventType("version_deployed", "Version deployed", "The worker started on a new build."), EventType("assets_quota", "Assets storage warning", "The assets storage passed its warning threshold."), + EventType( + "archive_failed", + "Archive copy failing", + "A filed packet could not be copied to the archive folder after three tries, or the archive folder check" + " failed.", + ), ) EVENT_TYPES: dict[str, EventType] = {t.key: t for t in _TYPES} diff --git a/server/packetstation/settings.py b/server/packetstation/settings.py index b6d1d3e..e50f382 100644 --- a/server/packetstation/settings.py +++ b/server/packetstation/settings.py @@ -33,7 +33,7 @@ def read_secret(path: Path | str, what: str = "secret") -> str: return value -_OPTIONAL_PATHS = ("ldap_ca_bundle", "smtp_ca_bundle") +_OPTIONAL_PATHS = ("ldap_ca_bundle", "smtp_ca_bundle", "archive_root") class Settings(BaseSettings): @@ -61,6 +61,8 @@ class Settings(BaseSettings): assets_dir: Path = Path("/data/assets") backup_dir: Path = Path("/backups") + # The archive share (Gitea issue 7). Unset or blank turns the archive off. The Admin picks a folder under it. + archive_root: Path | None = None session_key_file: Path = Path("/run/secrets/session_key") seal_key_file: Path = Path("/run/secrets/seal_key") static_dir: Path = Path("/app/static") @@ -125,7 +127,7 @@ class Settings(BaseSettings): if isinstance(data, dict): data = dict(data) for key in list(data): - if key.lower() in _OPTIONAL_PATHS and data[key] in ("", None): + if key.lower() in _OPTIONAL_PATHS and (data[key] is None or not str(data[key]).strip()): data[key] = None for key in list(data): if key.lower() in ("stack", "feed_source", "drawing_source") and isinstance(data[key], str): diff --git a/server/packetstation/worker/jobs/__init__.py b/server/packetstation/worker/jobs/__init__.py index 2260c49..068cb07 100644 --- a/server/packetstation/worker/jobs/__init__.py +++ b/server/packetstation/worker/jobs/__init__.py @@ -106,6 +106,7 @@ LAZY = ( "backup_check", "notify_send", "notify_digest", + "archive_copy", ) diff --git a/server/packetstation/worker/jobs/archive_copy.py b/server/packetstation/worker/jobs/archive_copy.py new file mode 100644 index 0000000..c3d13a2 --- /dev/null +++ b/server/packetstation/worker/jobs/archive_copy.py @@ -0,0 +1,101 @@ +"""The archive copy (Gitea issue 7), every 15 minutes and on request (FILE asks for a run when it commits). + +Off when ARCHIVE_ROOT is unset, and idle until the Admin saves a folder. Each run checks the folder (the probe), queues +every unbroken seal with a data book that has no `archive_copies` row (so turning the archive on later copies earlier +filings too), then copies the pending rows and the failed rows that are due. A failed copy waits 5, 15, 60 minutes, +then an hour between tries. The `archive_failed` condition (subject `archive`) is raised when a copy has failed three +times or the folder check fails, and cleared when no copy is failed. A failed copy never touches the packet. +""" + +from __future__ import annotations + +from datetime import datetime + +from ...domain import archive +from ...feed.config import FeedSettings +from .. import jobs +from . import Counts, JobContext, JobSpec + +NAME = archive.JOB +INTERVAL_SECONDS = 15 * 60 +BATCH = 200 +TITLE = "Archive copy failing" + + +def _check_changed(ctx: JobContext, result: archive.CheckResult, archive_dir: str) -> None: + """Audit the job's own folder check only when its result differs from the last check on record.""" + last = archive.last_check(ctx.conn) + if last is None or bool(last["ok"]) != result.ok or last.get("archive_dir") != archive_dir: + archive.audit_check(ctx.conn, result, archive_dir=archive_dir, source="job") + + +def run(ctx: JobContext) -> Counts: + root = archive.root_of(ctx.settings) + counts: Counts = {"enabled": root is not None, "queued": 0, "copied": 0, "failed": 0} + if root is None: + return counts + archive_dir, _, _, _ = archive.load(ctx.conn) + counts["archive_dir"] = archive_dir + if archive_dir is None: + counts["skipped"] = "no archive folder saved" + return counts + rel, result = archive.check(root, archive_dir) + _check_changed(ctx, result, archive_dir) + if not result.ok or rel is None: + counts["check"] = result.message + ctx.error(f"archive folder check failed: {result.message}") + jobs.raise_event( + ctx.conn, + archive.EVENT, + subject=archive.EVENT_SUBJECT, + title=TITLE, + body=f"The archive folder check failed: {result.message}", + detail={"archive_dir": archive_dir, "message": result.message}, + severity="critical", + condition=True, + ) + return counts + base = root.joinpath(*rel.split("/")) + with ctx.conn.transaction(): + counts["queued"] = archive.queue_missing(ctx.conn) + for row in archive.due_rows(ctx.conn, BATCH): + try: + folder, sha, files = archive.copy_seal(ctx.conn, ctx.settings, base, rel, int(row["seal_id"])) + except archive.CopyError as e: + attempts = archive.mark_failed(ctx.conn, row, str(e)) + counts["failed"] += 1 + ctx.error(f"seal {row['seal_id']}: {e} (try {attempts})", degrade=True) + continue + except Exception as e: # anything unexpected is one failed copy, never a stopped run + attempts = archive.mark_failed(ctx.conn, row, f"{type(e).__name__}: {e}") + counts["failed"] += 1 + ctx.log.exception("archive copy of seal %s failed", row["seal_id"]) + ctx.error(f"seal {row['seal_id']}: {type(e).__name__} (try {attempts})", degrade=True) + continue + archive.mark_done(ctx.conn, row, folder, sha, files) + counts["copied"] += 1 + st = archive.copies_status(ctx.conn) + counts.update(pending=st["pending"], failed_total=st["failed"], done_total=st["done"]) + stuck = archive.stuck_failures(ctx.conn) + if stuck: + lines = [f"{f['packet_label']} (seal {f['seal_id']}): {f['last_error']}" for f in stuck[:10]] + jobs.raise_event( + ctx.conn, + archive.EVENT, + subject=archive.EVENT_SUBJECT, + title=TITLE, + body=f"{len(stuck)} filed packet(s) could not be copied to the archive:\n" + "\n".join(lines), + detail={"failed": st["failed"], "seal_ids": [f["seal_id"] for f in stuck]}, + severity="critical", + condition=True, + ) + elif st["failed"] == 0: + jobs.resolve_event(ctx.conn, archive.EVENT, subject=archive.EVENT_SUBJECT, title=TITLE) + return counts + + +def cadence(_: FeedSettings, now_site: datetime) -> int: + return INTERVAL_SECONDS + + +SPEC = JobSpec(name=NAME, run=run, cadence=cadence, pausable=False) diff --git a/tests/server/test_archive.py b/tests/server/test_archive.py new file mode 100644 index 0000000..a6e765c --- /dev/null +++ b/tests/server/test_archive.py @@ -0,0 +1,453 @@ +"""The archive location (Gitea issue 7): the path rules, the check, the File locations API, the archive-copy job +(copy, idempotence, failure and backoff, the Admin condition, old seals queued) and filing with the archive off or +broken. Synthetic data only; the archive root is a temp folder.""" + +from __future__ import annotations + +import hashlib +import json +import os +from collections.abc import Iterator +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any + +import pytest +from closehelpers import file_packet, filed, q1, qall, rich_complete, vclose +from fakedir import FakeDirectory +from pshelpers import make_settings, needs_pg +from shophelpers import Shop, ok, open_shop, refused, run + +from packetstation.domain import archive +from packetstation.settings import Settings +from packetstation.worker import jobs, runs +from packetstation.worker.jobs import archive_copy +from packetstation.worker.scheduler import Scheduler + +URL = "/api/admin/file-locations" + + +def share(tmp_path: Path) -> Path: + root = tmp_path / "share" + (root / "packets").mkdir(parents=True, exist_ok=True) + return root + + +# ---------------------------------------------------------------- settings and path rules + + +def test_archive_root_blank_is_off(tmp_path: Path) -> None: + assert make_settings(tmp_path).archive_root is None + assert make_settings(tmp_path, archive_root="").archive_root is None + assert make_settings(tmp_path, archive_root=" ").archive_root is None + assert archive.root_of(make_settings(tmp_path, archive_root=" ")) is None + assert make_settings(tmp_path, archive_root="/archive").archive_root == Path("/archive") + + +@pytest.mark.parametrize( + ("given", "why"), + [ + ("", "Enter a folder"), + (" ", "Enter a folder"), + ("..", "'..'"), + ("packets/../..", "'..'"), + ("a\\..\\b", "'..'"), + ("pack\x00ets", "NUL"), + ("x" * 401, "longer than"), + ], +) +def test_relative_dir_refusals(tmp_path: Path, given: str, why: str) -> None: + with pytest.raises(archive.PathRefused, match=why): + archive.relative_dir(share(tmp_path), given) + + +def test_relative_dir_clean_and_absolute(tmp_path: Path) -> None: + root = share(tmp_path) + assert archive.relative_dir(root, "packets") == "packets" + assert archive.relative_dir(root, "./packets//2026/") == "packets/2026" + assert archive.relative_dir(root, "packets\\2026") == "packets/2026" + assert archive.relative_dir(root, str(root / "packets" / "2026")) == "packets/2026" + with pytest.raises(archive.PathRefused, match="must lie under"): + archive.relative_dir(root, str(tmp_path / "elsewhere")) + with pytest.raises(archive.PathRefused, match="must lie under"): + archive.relative_dir(root, "/etc") + with pytest.raises(archive.PathRefused, match="not the root"): + archive.relative_dir(root, str(root)) + + +def test_resolve_dir(tmp_path: Path) -> None: + root = share(tmp_path) + rel, real = archive.resolve_dir(root, "packets") + assert rel == "packets" and real == (root / "packets").resolve() + with pytest.raises(archive.PathRefused, match="does not exist"): + archive.resolve_dir(root, "nope") + (root / "afile").write_bytes(b"x") + with pytest.raises(archive.PathRefused, match="not a folder"): + archive.resolve_dir(root, "afile") + with pytest.raises(archive.PathRefused, match="is not there"): + archive.resolve_dir(tmp_path / "unmounted", "packets") + + +def test_resolve_dir_refuses_a_link_out(tmp_path: Path) -> None: + root = share(tmp_path) + outside = tmp_path / "outside" + outside.mkdir() + try: + os.symlink(outside, root / "escape", target_is_directory=True) + except (OSError, NotImplementedError): + pytest.skip("this OS or user cannot make symlinks") + with pytest.raises(archive.PathRefused, match="outside the archive root"): + archive.resolve_dir(root, "escape") + (outside / "deep").mkdir() + with pytest.raises(archive.PathRefused, match="outside the archive root"): + archive.resolve_dir(root, "escape/deep") + # a link that stays inside the root is fine + os.symlink(root / "packets", root / "inside", target_is_directory=True) + assert archive.resolve_dir(root, "inside")[1] == (root / "packets").resolve() + + +def test_check_and_probe(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + root = share(tmp_path) + rel, res = archive.check(root, "packets") + assert rel == "packets" and res.ok and res.free_bytes and res.resolved == str((root / "packets").resolve()) + assert list((root / "packets").iterdir()) == [] # the probe is gone + assert archive.check(None, "packets") == (None, archive.CheckResult(False, None, None, archive.OFF_MESSAGE)) + rel, res = archive.check(root, "../x") + assert rel is None and not res.ok and "'..'" in res.message + + def broken_fsync(fd: int) -> None: + raise OSError(5, "Input/output error") + + monkeypatch.setattr(archive.os, "fsync", broken_fsync) + rel, res = archive.check(root, "packets") + assert rel == "packets" and not res.ok and "Cannot write" in res.message and "Input/output error" in res.message + assert list((root / "packets").iterdir()) == [] + + +def test_backoff() -> None: + assert [archive.backoff(n) for n in (1, 2, 3, 4, 9)] == [ + timedelta(minutes=5), + timedelta(minutes=15), + timedelta(minutes=60), + timedelta(minutes=60), + timedelta(minutes=60), + ] + + +def test_safe_names() -> None: + assert archive.safe_part("L5/12:3", "x") == "L5_12_3" + assert archive.safe_part("..", "x") == "x" + taken: set[str] = set() + assert [archive._unique(n, taken) for n in ("a.pdf", "A.pdf", "a.pdf", "manifest.json", "b")] == [ + "a.pdf", + "A (2).pdf", + "a (3).pdf", + "manifest (2).json", + "b", + ] + + +# ---------------------------------------------------------------- with PostgreSQL + + +@pytest.fixture +def on_settings(pg_settings: Settings, tmp_path: Path) -> Settings: + return pg_settings.model_copy(update={"archive_root": share(tmp_path)}) + + +@pytest.fixture +def shop(on_settings: Settings, fake_dir: FakeDirectory, dev_visual: None) -> Iterator[Shop]: + yield from open_shop(on_settings, fake_dir) + + +@pytest.fixture +def off_shop(pg_settings: Settings, fake_dir: FakeDirectory, dev_visual: None) -> Iterator[Shop]: + yield from open_shop(pg_settings, fake_dir) + + +def run_copy(settings: Settings) -> dict[str, Any]: + sch = Scheduler(settings, build="pytest", jobs={archive.JOB: archive_copy.SPEC}) + return runs.run_job(sch, archive.JOB, "schedule") + + +def save_dir(settings: Settings, archive_dir: str = "packets") -> None: + run( + settings, + "INSERT INTO settings (key, value) VALUES ('files', jsonb_build_object('archive_dir', %s::text))" + " ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, version = settings.version + 1", + (archive_dir,), + ) + + +def copies(settings: Settings) -> list[Any]: + return qall(settings, "SELECT seal_id, status, attempts, last_error, next_try_at, folder FROM archive_copies") + + +def audits(settings: Settings, action: str) -> list[dict[str, Any]]: + return [r[0] for r in qall(settings, "SELECT detail FROM audit_events WHERE action = %s ORDER BY id", (action,))] + + +def sha(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +@needs_pg +def test_api_off(off_shop: Shop) -> None: + ann = off_shop("admin.ann") + body = ok(ann.get(URL)) + assert body == { + "enabled": False, + "root": None, + "archive_dir": None, + "version": 0, + "resolved": None, + "last_check": None, + "copies": {"pending": 0, "failed": 0, "done": 0, "recent_failures": []}, + } + for r in ( + ann.post(f"{URL}/check", json={"archive_dir": "packets"}), + ann.put(URL, json={"archive_dir": "packets", "version": 0}), + ann.post(f"{URL}/retry"), + ): + refused(r, 409, "archive_off", archive.OFF_MESSAGE) + # Admin only, password session + assert off_shop("clerk.clara").get(URL).status_code == 403 + + +@needs_pg +def test_api_check_save_and_retry(shop: Shop) -> None: + s = shop.settings + root = Path(str(s.archive_root)) + ann = shop("admin.ann") + body = ok(ann.get(URL)) + assert body["enabled"] is True and body["root"] == str(root) and body["archive_dir"] is None + + refused(ann.post(f"{URL}/check", json={"archive_dir": "../etc"}), 422, "bad_path") + refused(ann.post(f"{URL}/check", json={"archive_dir": "missing"}), 422, "bad_path") + good = ok(ann.post(f"{URL}/check", json={"archive_dir": str(root / "packets")})) + assert good["ok"] is True and good["resolved"] == str((root / "packets").resolve()) and good["free_bytes"] > 0 + assert ok(ann.get(URL))["archive_dir"] is None # the check saves nothing + checks = audits(s, "archive.checked") + assert [(c["source"], c["ok"]) for c in checks] == [("admin", False), ("admin", False), ("admin", True)] + assert checks[2]["archive_dir"] == "packets" + + refused(ann.put(URL, json={"archive_dir": "missing", "version": 0}), 422, "bad_path") + refused(ann.put(URL, json={"archive_dir": "/elsewhere", "version": 0}), 422, "bad_path") + saved = ok(ann.put(URL, json={"archive_dir": str(root / "packets"), "version": 0})) + assert saved["archive_dir"] == "packets" and saved["version"] == 1 + assert saved["resolved"] == str((root / "packets").resolve()) + assert saved["last_check"]["ok"] is True and saved["last_check"]["by"] == "Ann Admin" + assert set(saved["last_check"]) == {"ok", "at", "by", "message"} + conflict = refused(ann.put(URL, json={"archive_dir": "packets", "version": 0}), 409, "version_conflict") + assert conflict["current"]["version"] == 1 + (root / "other").mkdir() + assert ok(ann.put(URL, json={"archive_dir": "other", "version": 1}))["archive_dir"] == "other" + assert [a["archive_dir"] for a in audits(s, "admin.file_locations_updated")] == [ + {"from": None, "to": "packets"}, + {"from": "packets", "to": "other"}, + ] + # a check that fails after the path rules (the probe) answers check_failed on save and keeps the old folder + (root / "ro").mkdir() + orig = archive.probe + try: + archive.probe = lambda folder: archive.CheckResult(False, str(folder), None, "Cannot write in it.") + r = refused(ann.put(URL, json={"archive_dir": "ro", "version": 2}), 422, "check_failed", "Cannot write in it.") + assert r["check"]["ok"] is False + finally: + archive.probe = orig + assert ok(ann.get(URL))["archive_dir"] == "other" + # saving asks for a copy run; Retry now resets the backoff and asks again (202) + r = ann.post(f"{URL}/retry") + assert r.status_code == 202 and r.json()["requested"] is True + assert q1(s, "SELECT count(*) FROM job_requests WHERE job = 'archive-copy'")[0] == 3 + assert audits(s, "admin.archive_retry") + assert shop("clerk.clara").post(f"{URL}/retry").status_code == 403 + + +@needs_pg +def test_copy_job(shop: Shop, monkeypatch: pytest.MonkeyPatch) -> None: + s = shop.settings + root = Path(str(s.archive_root)) + events: list[tuple[str, str, dict[str, Any]]] = [] + monkeypatch.setattr(jobs, "raise_event", lambda conn, t, **kw: events.append(("raise", t, kw))) + monkeypatch.setattr(jobs, "resolve_event", lambda conn, t, **kw: events.append(("resolve", t, kw))) + + # no folder saved: the job idles, nothing queued + row = run_copy(s) + assert row["status"] == "ok" and row["counts"]["skipped"] == "no archive folder saved" + save_dir(s) + + pkt = rich_complete(shop) + pid = pkt["id"] + # rich_complete keeps one short open (a packet from before change C7); close it so FILE accepts + run(s, "UPDATE bom_shorts SET closed_at = now() WHERE packet_id = %s", (pid,)) + out = ok(file_packet(shop, vclose(shop, pkt))) + assert out["status"] == "filed" + assert q1(s, "SELECT count(*) FROM job_requests WHERE job = 'archive-copy' AND claimed_at IS NULL")[0] == 1 + seal_id, book_id = q1(s, "SELECT id, databook_asset_id FROM packet_seals WHERE packet_id = %s", (pid,)) + + row = run_copy(s) + assert row["status"] == "ok", row + assert row["trigger"] == "request" and row["counts"]["queued"] == 1 and row["counts"]["copied"] == 1 + ((sid, status, attempts, err, nxt, folder),) = copies(s) + assert (sid, status, attempts, err, nxt) == (seal_id, "done", 1, None, None) + project, unit = q1( + s, + "SELECT pr.number, pr.number || u.lot FROM packets p JOIN units u ON u.id = p.unit_id" + " JOIN projects pr ON pr.id = u.project_id WHERE p.id = %s", + (pid,), + ) + assert folder.startswith(f"packets/{project}/{unit}/") and folder.endswith(f"-seal{seal_id}") + where = root.joinpath(*folder.split("/")) + manifest = json.loads((where / "manifest.json").read_text(encoding="utf-8")) + assert manifest["packet_id"] == pid and manifest["seal_id"] == seal_id and manifest["project"] == project + assert manifest["unit"] == unit and manifest["build"] == s.ps_build and manifest["filed_by"] == "Clara Clerk" + assert manifest["sealed_by"] == "Clara Clerk" and len(manifest["seal_hash"]) == 64 and manifest["key_id"] + kinds = [f["kind"] for f in manifest["files"]] + assert kinds[0] == "databook" and manifest["files"][0]["asset_id"] == str(book_id) + assert {"databook", "page", "doc", "qc_doc", "mtr"} <= set(kinds), kinds + assert kinds.count("mtr") == 2 + names = [f["name"] for f in manifest["files"]] + assert len({n.lower() for n in names}) == len(names) + assert sorted(p.name for p in where.iterdir()) == sorted([*names, "manifest.json"]) + for f in manifest["files"]: + assert sha(where / f["name"]) == f["sha256"] and (where / f["name"]).stat().st_size == f["size"] + assert q1(s, "SELECT manifest_sha256 FROM archive_copies")[0] == sha(where / "manifest.json") + (done,) = audits(s, "archive.copied") + assert done["seal_id"] == seal_id and done["folder"] == folder and done["files"] == len(names) + assert events == [("resolve", "archive_failed", {"subject": "archive", "title": "Archive copy failing"})] + + # idempotent: a second run copies nothing; a row set back to pending finds the identical files and is done + events.clear() + assert run_copy(s)["counts"]["copied"] == 0 + run(s, "UPDATE archive_copies SET status = 'pending'") + before = {p.name: p.stat().st_mtime_ns for p in where.iterdir()} + assert run_copy(s)["counts"]["copied"] == 1 + assert {p.name: p.stat().st_mtime_ns for p in where.iterdir()} == before + assert copies(s)[0][1] == "done" + + # a different file in the way: never overwritten; failed with a 5-minute backoff, then 15, then 60 and raised + run(s, "UPDATE archive_copies SET status = 'pending', folder = NULL, done_at = NULL, attempts = 0") + (where / "manifest.json").unlink() + victim = where / names[1] + victim.write_bytes(b"something else") + events.clear() + t0 = datetime.now(UTC) + row = run_copy(s) + assert row["status"] == "degraded" and row["counts"]["failed"] == 1 + _, status, attempts, err, nxt, _ = copies(s)[0] + assert (status, attempts) == ("failed", 1) and "not overwritten" in err and names[1] in err + assert timedelta(minutes=4) < nxt - t0 < timedelta(minutes=6) + assert victim.read_bytes() == b"something else" and not (where / "manifest.json").exists() + assert events == [] # one failure: not raised yet, not cleared either + assert run_copy(s)["counts"]["failed"] == 0 # not due + for n, wait in ((2, 15), (3, 60)): + run(s, "UPDATE archive_copies SET next_try_at = now() - interval '1 second'") + t0 = datetime.now(UTC) + run_copy(s) + _, status, attempts, _, nxt, _ = copies(s)[0] + assert (status, attempts) == ("failed", n) + assert timedelta(minutes=wait - 1) < nxt - t0 < timedelta(minutes=wait + 1) + raised = [e for e in events if e[0] == "raise"] + assert len(raised) == 1 and raised[0][1] == "archive_failed" and raised[0][2]["subject"] == "archive" + assert raised[0][2]["condition"] is True and f"seal {seal_id}" in raised[0][2]["body"] + assert len(audits(s, "archive.copy_failed")) == 3 + st = ok(shop("admin.ann").get(URL))["copies"] + assert (st["failed"], st["done"], st["pending"]) == (1, 0, 0) + (f,) = st["recent_failures"] + assert ( + f["seal_id"] == seal_id + and f["packet_id"] == pid + and f["attempts"] == 3 + and "not overwritten" in f["last_error"] + ) + assert f["packet_label"].startswith(project) and f["next_try_at"] + + # the Admin removes the stray file and presses Retry now: done, condition cleared + victim.unlink() + events.clear() + assert shop("admin.ann").post(f"{URL}/retry").status_code == 202 + row = run_copy(s) + assert row["counts"]["copied"] == 1 and copies(s)[0][1] == "done" + assert sha(victim) == manifest["files"][1]["sha256"] + assert events == [("resolve", "archive_failed", {"subject": "archive", "title": "Archive copy failing"})] + + +@needs_pg +def test_old_seals_are_queued_when_the_archive_is_turned_on(off_shop: Shop, tmp_path: Path) -> None: + s = off_shop.settings + pkt = filed(off_shop) + # archive off: FILE asks for nothing, and the job does nothing + assert q1(s, "SELECT count(*) FROM job_requests WHERE job = 'archive-copy'")[0] == 0 + assert run_copy(s)["counts"] == {"enabled": False, "queued": 0, "copied": 0, "failed": 0} + assert copies(s) == [] + + on = s.model_copy(update={"archive_root": share(tmp_path)}) + save_dir(on) + row = run_copy(on) + assert row["counts"]["queued"] == 1 and row["counts"]["copied"] == 1 + seal = q1(s, "SELECT id FROM packet_seals WHERE packet_id = %s", (pkt["id"],))[0] + assert copies(s)[0][:2] == (seal, "done") + + +@needs_pg +def test_filing_succeeds_with_the_archive_broken(shop: Shop, monkeypatch: pytest.MonkeyPatch) -> None: + s = shop.settings + root = Path(str(s.archive_root)) + events: list[tuple[str, str, dict[str, Any]]] = [] + monkeypatch.setattr(jobs, "raise_event", lambda conn, t, **kw: events.append(("raise", t, kw))) + monkeypatch.setattr(jobs, "resolve_event", lambda conn, t, **kw: events.append(("resolve", t, kw))) + save_dir(s, "gone") # saved, then the folder went away + pkt = filed(shop) + assert pkt["status"] == "filed" + assert q1(s, "SELECT count(*) FROM job_requests WHERE job = 'archive-copy'")[0] == 1 + + row = run_copy(s) + assert row["status"] == "degraded" and "does not exist" in row["counts"]["check"] + assert copies(s) == [] + ((kind, etype, kw),) = events + assert (kind, etype, kw["subject"], kw["condition"]) == ("raise", "archive_failed", "archive", True) + checks = audits(s, "archive.checked") + assert [(c["source"], c["ok"]) for c in checks] == [("job", False)] + run_copy(s) + assert len(audits(s, "archive.checked")) == 1 # the same result again is not audited again + + # the folder comes back: the copy is made and the condition cleared + (root / "gone").mkdir() + events.clear() + row = run_copy(s) + assert row["status"] == "ok" and row["counts"]["copied"] == 1 + assert events == [("resolve", "archive_failed", {"subject": "archive", "title": "Archive copy failing"})] + assert [(c["source"], c["ok"]) for c in audits(s, "archive.checked")] == [("job", False), ("job", True)] + + +@needs_pg +def test_filing_succeeds_when_the_request_fails(shop: Shop, monkeypatch: pytest.MonkeyPatch) -> None: + s = shop.settings + save_dir(s) + + def boom(*a: Any, **k: Any) -> None: + raise RuntimeError("no requests today") + + monkeypatch.setattr(archive, "request_copy", boom) + pkt = filed(shop) + assert pkt["status"] == "filed" + assert q1(s, "SELECT count(*) FROM job_requests WHERE job = 'archive-copy'")[0] == 0 + assert run_copy(s)["counts"]["copied"] == 1 # the scheduled run picks it up + + +@needs_pg +def test_doctor_archive_check(pg_settings: Settings, tmp_path: Path) -> None: + from packetstation import doctor + + assert doctor.check_archive(pg_settings)[0].status == "SKIP" + on = pg_settings.model_copy(update={"archive_root": share(tmp_path)}) + (c,) = doctor.check_archive(on) + assert c.status == "SKIP" and "no archive folder saved" in c.detail + save_dir(on) + (c,) = doctor.check_archive(on) + assert c.status == "PASS" and "writable" in c.detail + save_dir(on, "gone") + (c,) = doctor.check_archive(on) + assert c.status == "FAIL" and "does not exist" in c.detail + assert [(a["source"], a["ok"]) for a in audits(on, "archive.checked")] == [("doctor", True), ("doctor", False)] diff --git a/tests/server/test_notify_api.py b/tests/server/test_notify_api.py index 9daf1c2..80ae0cb 100644 --- a/tests/server/test_notify_api.py +++ b/tests/server/test_notify_api.py @@ -143,7 +143,7 @@ def test_get_seeds_every_rule(c: TestClient, base: Settings) -> None: body = c.get(ADMIN).json() assert body["dry_run"] is True and body["subject_prefix"] == "[Packet Station TEST]" assert body["smtp"] == {"host": base.smtp_host, "port": base.smtp_port, "from": base.smtp_from, "starttls": True} - assert len(body["rules"]) == 13 + assert len(body["rules"]) == 14 r = next(x for x in body["rules"] if x["event_type"] == "drawing_failing") assert r["label"] == "Drawing lookup failing" and r["param"] == {"name": "hours", "value": 24, "default": 24} assert r["enabled"] is True and r["delivery"] == "immediate" and r["digest_time"] == "07:00" diff --git a/tests/server/test_notify_db.py b/tests/server/test_notify_db.py index 5c4488b..490658d 100644 --- a/tests/server/test_notify_db.py +++ b/tests/server/test_notify_db.py @@ -136,12 +136,12 @@ def test_one_shot_dedupe_and_unknown_type(s: Settings, marker: int) -> None: def test_seed_to_admins_is_idempotent(s: Settings, marker: int) -> None: with app_conn(s) as conn: added = seed(conn, s) - assert len(added) == 13 + assert len(added) == 14 rows = conn.execute( "SELECT r.event_type, r.enabled, r.delivery::text, r.digest_time::text, r.param, c.kind::text," " c.group_dn, c.group_name FROM notification_rules r JOIN notification_recipients c USING (event_type)" ).fetchall() - assert len(rows) == 13 + assert len(rows) == 14 for r in rows: assert r[1] is True and r[2] == "immediate" and r[3] == "07:00:00" and r[5] == "group" assert r[6] == f"CN=PS-Admins,{s.ldap_group_base_dn}" and r[7] == "PS-Admins" @@ -522,7 +522,7 @@ def test_notify_send_job(s: Settings, marker: int, relay: FakeSmtp, ndir: Notify with app_conn(s) as conn: ctx = ctx_for(conn, build, ndir) counts = notify_send.run(ctx) - assert counts["seeded"] == 13 and counts["version_deployed"] == "sha-job" and counts["sent"] == 1 + assert counts["seeded"] == 14 and counts["version_deployed"] == "sha-job" and counts["sent"] == 1 assert counts["drawings_failing"] == 0 and counts["co_waiting"] == 0 again = notify_send.run(ctx) assert "version_deployed" not in again and "drawings_failing" not in again # watch waits 15 minutes diff --git a/tests/server/test_notify_smtp.py b/tests/server/test_notify_smtp.py index f558886..e9d5b06 100644 --- a/tests/server/test_notify_smtp.py +++ b/tests/server/test_notify_smtp.py @@ -44,7 +44,7 @@ def mail_settings(tmp: Path, server: FakeSmtp, ca: Path | None, **over: object) def test_catalog() -> None: - assert len(EVENT_TYPES) == 13 + assert len(EVENT_TYPES) == 14 assert SECURITY_TYPES == {"security_alert", "signin_burst", "integrity_failure"} assert EVENT_TYPES["drawing_failing"].param == "hours" and EVENT_TYPES["drawing_failing"].param_default == 24 assert EVENT_TYPES["co_waiting"].param == "days" and EVENT_TYPES["co_waiting"].param_default == 3 diff --git a/tests/server/test_schema.py b/tests/server/test_schema.py index c3c14d4..a4bfd7d 100644 --- a/tests/server/test_schema.py +++ b/tests/server/test_schema.py @@ -414,6 +414,7 @@ def test_feed_tables_and_app_grants(pg_settings: Settings) -> None: jobs = [r[0] for r in conn.execute("SELECT job FROM job_schedule ORDER BY job").fetchall()] assert jobs == [ "ad-reconcile", + "archive-copy", "backup-check", "drawing-lookup", "housekeeping", @@ -632,3 +633,19 @@ def test_qc_columns_checks_and_items(pg_settings: Settings) -> None: "ph_clean", "ph_qty", ] + + +def test_archive_copies_grants(pg_settings: Settings) -> None: + """0012: the app reads, queues and updates archive copies and never deletes one.""" + with app_conn(pg_settings) as conn: + for priv, want in ( + ("SELECT", True), + ("INSERT", True), + ("UPDATE", True), + ("DELETE", False), + ("TRUNCATE", False), + ): + got = one(conn, "SELECT has_table_privilege('ps_app', 'archive_copies', %s)", (priv,)) + assert got is want, priv + with pytest.raises(psycopg.errors.CheckViolation): + conn.execute("INSERT INTO archive_copies (seal_id, packet_id, status) VALUES (1, 1, 'done')")