diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index ab3673b98..fc901906f 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -1102,8 +1102,9 @@ disposition, and the audit trail stay intact — the Mirth Data-Pruner pattern); — `messages.metadata` is nulled in the same statement as the body (ASVS 14.2.7). The raw `[retention]` windows still default to `0`/`""` = keep/off (at least three keys ship on, and none of them purges anything: the `min_free_disk_mb` free-space floor, the `audit_log_warn_rows` growth warning and the hourly `wal_checkpoint_seconds` WAL truncate), **but `serve` applies a posture gate on top of them, so retention is *not* opt-in on a PHI instance**: each *unset* window that carries an auto-bound — -`[security].delete_message_bodies_after_days`, `[retention].dead_letter_days` and -`[retention].reference_snapshot_days` — is **defaulted to 30 days** at startup, under **both** +`[security].delete_message_bodies_after_days`, `[retention].dead_letter_days`, +`[retention].reference_snapshot_days` and `[retention].processed_archive_days` — +is **defaulted to 30 days** at startup, under **both** `[security].enforcement` dials, and the defaulted settings are named on stderr. A window set **explicitly to `0`** is not defaulted: that **refuses to start (exit 2)** under `enforce`, and warns under `warn`. This paragraph used to state the opposite split — refusal under `enforce`, auto-bound @@ -1147,6 +1148,7 @@ entry a Handler still reads. That stays true until state has an eviction key tha |---|---|---|---| | `messages_days` | | | **→ moved to `[security].delete_message_bodies_after_days`** (ADR 0118) — set it there; no longer accepted in `[retention]`. | | `dead_letter_days` | int | `0` | past N days, null the bodies of **dead-lettered** rows at **every stage**. A dead `ingress` or `routed` row carries the whole raw body, so the purge reaches it as well as a dead outbound row (BACKLOG #1188). This is their own window, because a dead row stays replayable until purged. Unset or `0`, this global window meets the startup posture gate described above this table; `0` = keep only where that gate allows it. An outbound's own `dead_letter_days` overrides it for that outbound's dead rows only (see *Per-connection overrides* below). An override of `0` meets its own gate, described there. A dead row at any other stage always takes this global window. *Corrected 2026-09-28 (BACKLOG #1186):* this row used to scope the purge to outbound rows, which BACKLOG #1188 made false. | +| `processed_archive_days` | int | `0` | past N days, **delete** the archived copies a File inbound left in its `.processed` directory (`processed_subdir`, written when `after_read = "move"`, the default). Each is a clear-text copy of an inbound payload, so this is the fourth window the startup posture gate described above this table auto-bounds, at the body window (owner ruling A2 of 2026-10-08, BACKLOG #1188). Keyed on **archive** time: the File source sets each copy's mtime as it archives it, so a partner's preserved old mtime cannot age a fresh copy out early; on POSIX the sweep keys on the later of mtime and ctime. The archive move sets ctime, but so does any later change to the copy's metadata (a `chmod`, `chown` or ACL change), and that restarts the copy's window. On Windows a copy whose stamp failed keeps the drop's own mtime and can age out early; the failure is logged as a warning. Regular files one level deep only, never through a link or junction; file content is never read and no file name is logged. Only a File inbound whose listener is bound, and whose `after_read` is `"move"`, is swept, so a stopped or undeployed inbound's archive waits until it runs again. An archive directory that is, or contains, the watch or `.error` directory of any inbound in the configuration (on every engine shard under `serve --shard`) or the target directory of any File outbound, bound or not, or one of the engine's own directories (at least the directories holding the store, key and TLS files the settings name, the config directory, and any `*_dir` setting or the backup `destination`, an archive inside one of those last two counting too), is skipped with a warning on every pass. The sweep deletes nothing in at least these further cases: for every connection, when one connection's directory settings cannot be resolved (a missing `env()` value); and for one connection, when a directory it must not reach exists, or may, but its identity cannot be read under that connection's identity (an unreachable share, a refused `stat`). Each is logged as a warning on every pass. Every regular file in the directory counts as an archive, so give `processed_subdir` a directory of its own. `.error` is never swept. `0` = keep only where the gate allows it | | `allow_unbounded_phi` | | | **→ moved to `[security].allow_keeping_phi_indefinitely`** (ADR 0118) — set it there; no longer accepted in `[retention]`. | | `state_max_age_days` | int | `0` | past N days, **delete** transform-state entries (ADR 0005) last written before the cutoff — keeps the in-memory state cache + table bounded. A simple global age purge (by `set_at`); per-namespace policy is a follow-up. `0` = keep | | `connection_event_retention_hours` | int | `0` | past N **hours**, **delete** `connection_event` rows (the `[diagnostics]` #46 transport/lifecycle log — high-volume under a connect-per-message sender or a probe storm, so its own short window in **hours**, not days). `0` = inherit the `messages_days` body window (the ADR 0021 §7.5 default). | diff --git a/docs/PHI.md b/docs/PHI.md index 75d2e80f0..eebd5dbe7 100644 --- a/docs/PHI.md +++ b/docs/PHI.md @@ -175,7 +175,8 @@ destruction) are documented in [§3](#3-encryption-at-rest) under the matching h | `[backup].destination/mefor-backup-*.mfbak` (ADR 0049 DR backup) | **SQLite only** carries bodies | **SQLite: Yes** — a consistent store snapshot (full inbound + outbound bodies) + the config bundle. **SQL Server / Postgres: No** — config bundle only | **Yes** — `.mfbak` chunked-AEAD codec under the **store DEK** (`resolve_active_key`); an identity-cipher (no-key) box is **refused** unless `[backup].allow_unencrypted` writes a `.mfbak.plain` | **PL-1** (SQLite) / **PL-4** (server backends) | On a **server-DB store `snapshot_to` raises `DbaDelegatedError`**, so the BackupRunner writes a **config-only** archive — or skips entirely when `[backup].config_only_on_server_db = false`. There is therefore **no `.mfbak` containing message bodies on SQL Server or Postgres**; the DB-tier backup there is `BACKUP DATABASE` / Always On / `pg_dump` / PITR, infra-owned. Where bodies *are* present it is a second at-rest PHI copy, bounded by keep-N retention; unlike `uploads_dir`, it is **not** re-encrypted by `rotate-key`. The share's own ACLs are infra-owned. **The keep-N bound covers the CANONICAL name only** — a `.failed` or `.part` archive at the same destination is deliberately outside the candidate set (ADR 0049, BACKLOG #1587) and nothing expires it; see [§8](#8-retention--purge)'s honest-gaps table | ``keep-N `[backup].retention_keep` `` | | `mefor-backup-*` / `mefor-verify-*` staging dirs (ADR 0049, BACKLOG #1174): **SQLite** — the store's own data dir; **SQL Server / Postgres** — `.mefor-staging` under `[backup].destination`; a **standalone verify** (`restore-verify`, the DR cold-seed activation) — a private `mefor-verify-*` dir under the OS temp dir | SQLite carries bodies; server backends config-only | **Yes** — a full store snapshot and its tar, and on verify a **decrypted** archive | **No** — the snapshot keeps the store's own column cipher, but the staging tar and the verify extraction are **plaintext on disk**. **SQLite store:** the engine applies the store's own best-effort `_secure_file` to each staged tar and extracted store as soon as it exists and before its first byte, as `mefor-restore-*` does, and `snapshot_to` applies it to the snapshot once its copy completes. What it leaves inside a temp directory's own DACL is not always owner-only (ADR 0163). **SQL Server / Postgres:** each run's dir under `.mefor-staging` is created by `mkdtemp` and must come out owner-only (mode `0700` on POSIX; on Windows a protected DACL naming only SYSTEM, Administrators, OWNER RIGHTS and its owner), and each staged file gets `_secure_file` before its first byte. The `.mefor-staging` dir itself still inherits what the destination grants, and holds no plaintext of its own. **Standalone verify, any backend:** the dir is created by `mkdtemp`, so it is mode `0700` on POSIX and on Windows carries the protected DACL Python writes for that mode (SYSTEM, Administrators, OWNER RIGHTS, measured on 3.14.6), and each staged file gets `_secure_file` before its first byte. **All of these, since BACKLOG #1174:** each run's dir, in the data dir, in `.mefor-staging` and under the temp dir alike, must come out owner-only, or nothing is staged. On a volume that reports modes or ACLs the engine did not ask for, a backup fails with a reason naming the dir, and a standalone verify returns `FAIL` saying the volume, not the archive, is at fault. Volumes that can do this include, for example, a CIFS or Samba share without POSIX extensions, WSL `/mnt/c` without `metadata`, a Docker Desktop bind mount of a Windows path, and FAT or exFAT. A backup's own verify refused this way leaves the written archive unpublished at its staging name, not quarantined. The check trusts the owner the file system reports, and on POSIX reads mode bits only, so a server-set owner or an inherited macOS or NFSv4 extended ACL can pass it | **PL-1** | A backup snapshots the store into a `mefor-backup-*` dir and tars it **plaintext** there before sealing it into the `.mfbak` (`pipeline/dr_backup.py`), and `[backup].verify_after_backup` (**default `true`**) decrypts the archive straight back out to a `mefor-verify-*` dir on **every** run — independent of `full_restore_verify`. `restore-verify` and the DR cold-seed activation do not stage there. They stage in a private dir under the OS temp dir (`%TEMP%` / `TMPDIR`) of the account that runs them. So they do not write to the archive's own directory, which on a DR box may be a read-only share. Nor do they write beside `[store].path`, the current directory under the default relative path. The exceptions are an archive or store that sits in the temp dir itself, and a box with no writable temp dir, where Python falls back to the current directory. That dir needs about twice the archive's size free on the temp volume while it runs; the verify checks that first and returns `FAIL` naming the volume, not the archive, when it is short (a backup checks its own staging and destination volumes the same way, and fails with kind `space`); on a RAM-backed `/tmp` it is held in memory and can reach swap. Each dir holds a lock file for its whole run and is removed on success, on an exception and on cancellation; a build cancelled on the event loop keeps running on its worker thread, which removes the dir once it lets go of the files. **A crash or `SIGKILL` still leaves the dir behind**, and the OS releases its lock with the process. The **next backup** removes it. A standalone verify's dir is removed by the next standalone verify, or the next backup, that runs under the same account and temp dir. Those two sweep the temp dir for dirs with the same owner as their own new dir, because the temp dir can be shared. A dir with any other owner is never touched. On Windows the owner is the process's default owner: for an elevated process that is Administrators, so elevated runs of any admin account, and SYSTEM, can share one owner, while a non-elevated run of the same user does not. The sweep takes only a dir whose lock it can acquire and that holds the marker a run creates once it holds its lock, and it never goes by age, so a sibling engine shard's live run survives it. The sweep never runs at `serve` start. A dir whose removal is refused — on Windows, a scanner or indexer still holding a file — is retried for about two seconds, then every file in it is truncated to zero bytes in place (BACKLOG #1721). Truncation empties the FILE; the disk blocks it freed still hold the bytes until reused, and only FDE on that volume covers them ([§10](#10-secure-deployment--operations-checklist)). A holder that denies write sharing or has the file mapped refuses the truncation too. If a file can be neither removed nor emptied, the verify says so and names the dir (a `PASS` becomes `FAIL`, another verdict keeps its status, and an exception carries the dir as a note), and a backup run still publishes its good archive but raises a `backup_failed` alert of kind `cleanup`, under its own subject `dr_backup:staging`, and names the dir in its audit row; either way the next sweep of that dir's root retries it. A hard-linked file is reported, never truncated. **Still unbounded, at least:** a dir left behind when no further backup runs (an on-demand-only box, or `[backup].enabled` turned off), a standalone verify's dir when no later standalone verify or backup runs under that account and temp dir (the engine's backups never sweep an operator's profile temp), a dir something still holds open each time the sweep runs, a dir the engine's account cannot open (such as one an administrator's interactive run left; the sweep logs it), a dir on a destination that cannot lock at all (it is staged unlocked and never swept), a dir left in the instant between its creation and its lock (it holds no plaintext yet, and the sweep cannot prove it abandoned), and the freed blocks above | `UNBOUNDED — honest gap` | | `mefor-restore-*` staging dirs (the **destination** volume, ADR 0049 / BACKLOG #1717) | SQLite only — `restore` refuses a config-only archive outright | **Yes** — the whole decrypted archive, then the whole extracted `store.db` | **No** — both staged files are **plaintext on disk**; the engine locks them down, calling the store's own `_secure_file` on each as soon as it exists, as it now does for the SQLite backup and verify staging in the row above. That ACL is owner-only for **the account that ran the restore**: `_secure_file` grants `_current_user()`, which is the account the process runs as. On this operator-driven path that is the operator's own account, not the service account | **PL-1** | `messagefoundry restore` stages on `--to`'s own volume rather than `%TEMP%`, so the restored store is published by a hard link instead of a second multi-GB copy. That choice moves the exposure off the shared temp volume and onto the data volume, where the engine's own file ACL reaches it — but the **directory** still inherits whatever the destination's parent grants on Windows, so the operator's directory ACL remains the backstop ([§10](#10-secure-deployment--operations-checklist)) — and it is the only control on a box where `_current_user()` resolves no name, since `_secure_file` is best-effort and logs rather than failing, leaving the file whatever the parent directory granted. Transient (the `TemporaryDirectory` unlinks on exit) but **not** on a crash or `SIGKILL`. A restore is an operator-driven one-shot rather than a scheduled repeat, so nothing accumulates across runs — but nothing expires what a crash leaves behind: unlike the backup and verify staging in the row above, no sweep removes a restore's directory | `UNBOUNDED — honest gap` | -| File-connector output / spill dirs (`.hl7`, `.processed`, `.error`) | all | **Yes** — plaintext on disk | **No** — no cipher at all on this path | **PL-1** | Written by the File transport; treat the directory as PHI and cover it with volume/share encryption + an ACL | `UNBOUNDED — honest gap` | +| File-connector output / spill dirs (`.hl7` destination output, `.error`) | all | **Yes** — plaintext on disk | **No** — no cipher at all on this path | **PL-1** | Written by the File transport; treat the directory as PHI and cover it with volume/share encryption + an ACL | `UNBOUNDED — honest gap` | +| File-source archive (`.processed`, the inbound's `processed_subdir`) | all (filesystem, not the DB) | **Yes** — a clear-text copy of every inbound payload | **No** — no cipher at all on this path | **PL-1** | Written by the File source when `after_read = "move"` (the default). The fourth auto-bound window (owner ruling A2 of 2026-10-08, BACKLOG #1188): an unset window takes the body window's 30 days on a PHI instance, and an explicit `0` is refused under `enforce` unless `[security].allow_keeping_phi_indefinitely` is set. Keyed on **archive** time: the File source sets each copy's mtime as it archives it, so a partner's preserved old mtime cannot age a fresh copy out early. On POSIX the sweep keys on the later of mtime and ctime, so a stamp the engine may not make (a partner-owned drop) does not age the copy early, because the archive move sets ctime. Any later change to the copy's metadata (a `chmod`, `chown` or ACL change) also sets ctime and restarts its window; on Windows a failed stamp logs a warning and the copy ages from its own earlier mtime. The retention pass deletes regular files one level deep, never through a link, and only for a File inbound whose listener is bound and whose `after_read` is `"move"`, so a stopped or undeployed inbound's archive waits until it runs again. It skips an archive directory that is, or holds, the watch or `.error` directory of any inbound in the configuration or the target directory of any File outbound, or one of the engine's own directories, and deletes nothing in the further cases listed under `processed_archive_days` in [CONFIGURATION.md](CONFIGURATION.md), and treats every regular file in the directory as an archive, so give `processed_subdir` a directory of its own. Cover the volume with encryption + an ACL meanwhile | `` `[retention].processed_archive_days` `` | | Application log files (`[logging].log_dir`; under NSSM, `\logs\service.out.log` and `service.err.log`) | all (filesystem, not the DB) | **Possibly** — redaction is best-effort; a single-token identifier can survive it | **No** — plaintext on disk, no app-level cipher | **PL-1** | NSSM captures stdout/stderr; the engine writes a log file of its own only when the opt-in `[logging].file` is set (#122, ADR 0162 — same three handler filters, engine-owned rotation, refused inside `log_dir`), together with the `*.broken-*` files a write failure rolls aside, which sit outside `log_dir` and are therefore NOT swept by `[retention].app_log_days`. The defence is the three handler filters + `safe_exc()`/`safe_text()` + the never-log-bodies rule ([§7](#7-logging--phi-redaction) row 1), and the residual is stated there. The directory ACL is the NSSM installer's **best-effort** `icacls /inheritance:r`; age deletion is `[retention].app_log_days` (files by **mtime** — content is never read, so nothing selective happens here) and optional in-place gzip is `[retention].app_log_compress_days` (the compressor **does** read a file's bytes to archive + integrity-verify them, but only in-process — nothing is logged, and the archive stays inside the same ACL'd directory at the source's mtime). A support bundle copies a 500-line tail of this file out of the ACL'd directory entirely ([§7](#7-logging--phi-redaction)). Cover the volume with FDE ([§10](#10-secure-deployment--operations-checklist)) | `` `[retention].app_log_days` `` | | Off-box forwarder spool (`[logging].forward_spool_dir`; default `log-spool/` beside `[store].path`, and `log-spool/supervisor` for the `supervise` process) | all (filesystem, not the DB) | **Possibly** — it holds the same redacted text the forwarder sends, and redaction is best-effort | **No** — plaintext JSONL on disk, no app-level cipher | **PL-1** | BACKLOG #1966, ADR 0200. Records the off-box collector has not yet taken (down, backing off, or still queued at shutdown), kept in order and sent when it answers again, best effort (ADR 0200 states the limits). The spool is fed from the far side of the forwarder hand-off queue, so it only ever holds text the three handler filters already processed; a test plants a PHI-shaped value and proves only its redacted form lands. The directory is created `0o700` and segments `0o600` best-effort (no-ops on Windows, where it inherits the parent ACL). A segment is deleted once every entry in it is sent. Bounded by **size**, not age: when full the newest record is dropped and reported. `0` turns the spool off. Cover the volume with FDE ([§10](#10-secure-deployment--operations-checklist)) | `` `[logging].forward_spool_max_bytes` `` | | `messages.summary` | all three | **Yes** — MRN / patient name / order | **Yes, when a key is set** — store cipher; AAD binds the row id and its `channel_id` (`message_cell_aad`, vault BACKLOG #2593); store DEK (EF-3) | **PL-2** | Ingest-derived; no SQL search or index exists on it, so encrypting it costs nothing. NULL/blank stay as-is | ``rides `[security].delete_message_bodies_after_days` `` | @@ -884,7 +885,10 @@ per tier which of those its code provides, and an operator carries the rest BACKLOG #1587). It is sealed under the store DEK exactly like a good archive, so the at-rest protection is identical and only the retention bound differs. **File-connector output / spill dirs have no engine-managed retention or destruction at all** — the File transport writes - `.hl7`/`.processed`/`.error` files and an operator or infra job is the only removal path. Application + `.hl7`/`.error` files and an operator or infra job is the only removal path. The File source's + `.processed` archive is the exception: `[retention].processed_archive_days` ages it out on archive time, + auto-bounded at the body window (owner ruling A2 of 2026-10-08, BACKLOG #1188; §2's row states its + limits). Application log files are age-deleted by `[retention].app_log_days` (by **mtime**; content is never inspected) and, optionally, gzipped in place first by `[retention].app_log_compress_days` — the compressor reads a file's bytes to archive and verify them **in-process, never logged or exported**, leaves the archive on @@ -2025,6 +2029,7 @@ method exists for `Store`-protocol completeness and deliberately does nothing), | `strip_embedded_documents` | per-inbound `prune_documents_after` + `prune_documents_min_bytes` (**no global default** — nothing is stripped without an override) | in-place strip of bulky base64 documents; sets `messages.documents_pruned` | enforced | **enforced** | **enforced** | | Streaming-attachment release (`release_message_attachments`) | rides the two body windows | refcount decref + GC at 0, plus a startup `sweep_orphan_attachments` | enforced | **enforced** | **enforced** | | Application **log-file** sweep + compression (`app_log_days`, `app_log_compress_days`) | `[retention].app_log_days` / `[retention].app_log_compress_days` over `[logging].log_dir` | `DELETE` of `.log`/`.txt` files by **mtime** (content never read), plus optional in-place **gzip** of aged files — free-space prechecked, and the archive is decompressed off disk and compared byte-for-byte **before** the original is removed (a failure keeps the original). Bytes are read to compress/verify but never logged or exported; the archive inherits the source mtime, so the delete window still ages it out | enforced | enforced | enforced | +| File-source **.processed** sweep (`processed_archive_days`) | `[retention].processed_archive_days` (auto-bounded at 30 on a PHI instance) over each bound File inbound's `processed_subdir` | `DELETE` of regular files one level deep by archive time: **mtime**, which the File source sets as it archives, and on POSIX the later of that and ctime, which a later change to the copy's metadata also moves; content never read, no file name logged, never through a link; refused for an archive directory that is or holds any configured inbound's watch or `.error` directory or any File outbound's target directory, or one of the engine's own directories; the refusals are listed under `processed_archive_days` in [CONFIGURATION.md](CONFIGURATION.md). Filesystem, so the store backend does not matter | enforced | enforced | enforced | | `wal_checkpoint` | `[retention].wal_checkpoint_seconds` (default `3600`, on) | `PRAGMA wal_checkpoint(TRUNCATE)`, after the pass's purges; PASSIVE, which does not truncate, while a DR snapshot copy runs (BACKLOG #1937) | enforced | **no-op (DBA-owned)** — log management is `.ldf` backup / recovery model | **no-op (DBA-owned)** — checkpointer/autovacuum | | `vacuum` | `[retention].vacuum_at` (a daily clock time, **not** a cron) | `VACUUM` — locks the whole DB, so off-peak; off by default | enforced | **no-op (DBA-owned)** — space reclamation is a DBA operation | **no-op (DBA-owned)** — autovacuum | | Size threshold (advisory) (`db_status`) | `[retention].max_db_mb` | `storage_threshold` alert + `WARNING`; **never** auto-deletes | enforced | **enforced** — `db_status().size_bytes` is implemented (`SUM(size)` over `sys.database_files`) | **enforced** — `pg_database_size()` | @@ -2102,7 +2107,7 @@ window purely so a replay can be richer — is precisely the defect ASVS 14.2.7 | `[backup].destination/*.failed` and `*.part` non-canonical archives | no window, and that is the price of the keep-N fix rather than an oversight. A backup is written to `.part` and only renamed onto the canonical name once it has passed every configured check (ADR 0049, BACKLOG #1587), because the canonical name IS the keep-N candidate set — an archive that failed its restore-verify holding that name spent a retention slot and evicted an older **good** copy. So a `.failed` archive (kept deliberately, for diagnosis) and a `.part` (left by any abort between the write and the rename — a dropped UNC share, a verify cancelled when the engine stops mid-backup, a failed rename) are both invisible to the prune, which means **no engine path expires either**. On a box that restarts during its backup window the `.part` files accumulate at one full-size archive per incident. Each is sealed under the store DEK exactly like a good archive, so the at-rest protection is identical and only the retention bound differs; clearing them is the operator's. The predecessor was worse, not better: those same aborts used to leave a **truncated** file wearing the canonical name, which keep-N then counted as a good backup | | `mefor-backup-*` / `mefor-verify-*` staging dirs | no window. Where these dirs live, what the engine applies to them, and how the next backup's lock-proven sweep removes what a crash leaves are stated once, in §2's row (BACKLOG #1174). That sweep runs only when a backup runs, so it is not a window, and §2's row lists what it leaves unbounded, at least. `verify_after_backup` (default `true`) still decrypts a full archive back out on **every** run. On a server-DB store each run's dir in `.mefor-staging` must be owner-only or the run refuses, but the blocks a removed file freed are still covered only by the operator's volume encryption: restrict the destination to the service account and encrypt its volume ([§10](#10-secure-deployment--operations-checklist)). A standalone verify stages in the OS temp dir instead, so there the cover is FDE on the temp volume ([§10](#10-secure-deployment--operations-checklist)) | | `mefor-restore-*` staging dirs | no window on a clean exit — the `TemporaryDirectory` unlinks, and a restore is an operator-driven one-shot rather than a scheduled repeat, so nothing accumulates. It survives a crash or `SIGKILL`, and unlike the backup and verify staging above no sweep removes it. The engine applies its file ACL here (`_secure_file` on the staged tar and on the extracted store), so the operator's directory ACL is the backstop rather than the only control. If the teardown itself fails -- on Windows, a scanner or the indexer still holding the extracted store open -- `restore` exits non-zero as `restore failed (cleanup)`, names the directory, and leaves the restored store and bundle in place rather than rolling them back; the directory still holds the decrypted archive and store, and is the operator's to delete once nothing holds it open (BACKLOG #1717) | -| File-connector output / spill dirs (`.hl7`, `.processed`, `.error`) | no window, and no engine sweep of any kind — `pipeline/retention.py`'s only filesystem pass is `_sweep_app_logs`, which scans the log directory alone. These are operator-configured output paths, so the engine deleting from them would be deleting data an operator or a downstream system owns. On a first deployment they would accumulate plaintext **PL-1** bodies indefinitely; the cover is the operator's: volume or share encryption plus an ACL ([§10](#10-secure-deployment--operations-checklist)) | +| File-connector output / spill dirs (`.hl7` destination output, `.error`) | no window, and no engine sweep of either. The File source's `.processed` archive is NOT in this row any more: since owner ruling A2 of 2026-10-08 (BACKLOG #1188) `[retention].processed_archive_days` bounds it, and §2's row for it states the limits. `.error` deletion was not ruled on (#3335). These are operator-configured output paths, so the engine deleting from them would be deleting data an operator or a downstream system owns. On a first deployment they would accumulate plaintext **PL-1** bodies indefinitely; the cover is the operator's: volume or share encryption plus an ACL ([§10](#10-secure-deployment--operations-checklist)) | | Test-harness stores (the [harness table](#the-test-harnesss-own-at-rest-data-harness) in §2) | no engine sweep removes them after a run, and that table states each one's retention. The stores an operator can feed real messages, and which would then accumulate them, are named there: at least the reconcile capture, its compare report and the Compose-fed file drop | | `users.totp_secret` | no window by design — it lives and dies with the user row | | `secret_rotation_meta.fingerprint` | no window by design — one row per tracked secret, replaced in place by the rotation watcher; not PHI (BACKLOG #3472) | diff --git a/messagefoundry/api/app.py b/messagefoundry/api/app.py index 1919aabf6..82e3586e8 100644 --- a/messagefoundry/api/app.py +++ b/messagefoundry/api/app.py @@ -11529,6 +11529,23 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: cluster_settings=cluster_settings, registry_filter=registry_filter, registry_guard=registry_guard, + # BACKLOG #1188: the settings the engine is not otherwise handed, read only for the + # directories they name (key files, CA files, spool dirs), which the `.processed` + # archive sweep must never reach. + owned_settings=( + auth_settings, + ai_settings, + security_settings, + alerts_settings, + secrets_settings, + tls_settings, + cluster_settings, + approvals_settings, + integrity_settings, + service_settings, + static_credential_settings, + egress_settings, + ), # BACKLOG #1142, slice 3: the audited preflight for every inbound CA that requires a # peer certificate (MLLP, the HTTP listener, the DICOM SCP), at the first load and at # every real reload. Each listener's own build then enforces again and loads the bytes diff --git a/messagefoundry/config/retention_classification.py b/messagefoundry/config/retention_classification.py index 6744d77e5..6242c3cd0 100644 --- a/messagefoundry/config/retention_classification.py +++ b/messagefoundry/config/retention_classification.py @@ -140,6 +140,18 @@ def is_acknowledged(self, security: object) -> bool: level="PL-2", auto_bound_days=30, ), + # PL-1 the File source's `.processed` archive: a clear-text copy of every inbound payload. The + # fourth auto-bound window, at the body window and keyed on ARCHIVE time (the File source sets the + # copy's mtime when it archives it), by owner ruling A2 of 2026-10-08 (BACKLOG #1188). It is safe + # to bound where transform state is not: nothing reads an archived copy back, so age since archive + # cannot delete data the engine still uses. `.error` is not covered (#3335). + RetentionWindow( + setting="[retention].processed_archive_days", + field="processed_archive_days", + reads_from="[retention]", + level="PL-1", + auto_bound_days=30, + ), # --- classified and WARNED, never auto-bounded (see auto_bound_days above) --------------------- RetentionWindow( setting="[retention].state_max_age_days", @@ -231,12 +243,13 @@ def is_acknowledged(self, security: object) -> bool: #: most of the entries leaves a non-empty tuple, and a startup gate would then check two windows while #: reporting success. #: -#: TEN since BACKLOG #1966 added the forwarder spool. It was NINE, and that number was corrected by its own drift test rather than by counting. It was first +#: ELEVEN since BACKLOG #1188 added the `.processed` archive window (owner ruling A2 of 2026-10-08). +#: TEN after BACKLOG #1966 added the forwarder spool. It was NINE before that, and that number was corrected by its own drift test rather than by counting. It was first #: written as 7 — the count I derived by hand from the classification. The two-way equality against #: docs/PHI.md §2 immediately reported `[backup].retention_keep` and #: `[retention].connection_event_retention_hours` as documented-but-absent. That is precisely the #: failure this floor exists to catch, caught in the constant it protects, before either had ever run. -MIN_PHI_RETENTION_WINDOWS: Final[int] = 10 +MIN_PHI_RETENTION_WINDOWS: Final[int] = 11 def auto_bounded_windows() -> tuple[RetentionWindow, ...]: @@ -375,7 +388,7 @@ def evaluate_retention_gate( prod = "production " if production else "" # AUTO-BOUND. Owner ruling 2026-07-30: the three PHI-BODY windows default to 30 days when - # UNSET, on BOTH dials — previously this ran only when `not enforcing`, so on the shipped + # UNSET (a fourth, the `.processed` archive, by owner ruling A2 of 2026-10-08), on BOTH dials — previously this ran only when `not enforcing`, so on the shipped # `enforce` posture an unset window took the refusal below instead of a default. # # THE SAFETY TRADE IS DELIBERATE AND WORTH STATING: a production PHI instance with an unset diff --git a/messagefoundry/config/settings.py b/messagefoundry/config/settings.py index 22271c0a2..9f01b5ee5 100644 --- a/messagefoundry/config/settings.py +++ b/messagefoundry/config/settings.py @@ -44,7 +44,7 @@ import re import string import tomllib -from collections.abc import Iterator, Mapping, Sequence +from collections.abc import Iterable, Iterator, Mapping, Sequence from contextvars import ContextVar from dataclasses import dataclass from datetime import date @@ -2912,6 +2912,16 @@ class RetentionSettings(_Section): # closed cell; do not let a classification table describe this window as covering `reference.value` # generally, which would machine-bless a false claim. reference_snapshot_days: int = 0 + # Past N days, DELETE the archived copies a File inbound left in its `.processed` directory + # (`processed_subdir`), keyed on ARCHIVE time: the File source sets each file's mtime to the moment + # it archives it, so a partner's preserved old mtime cannot age a fresh copy out early. Each copy + # is the full clear-text inbound payload, a second PL-1 copy beside the stored body, so it is + # auto-bounded at the body window (owner ruling A2 of 2026-10-08, BACKLOG #1188). One level, regular + # files only, never through a link; `.error` is NOT swept. Only a running File inbound's directory is + # swept, and one that overlaps a directory it must not reach is refused (the list is + # docs/CONFIGURATION.md, `processed_archive_days`). 0 = keep forever, which an enforcing PHI + # instance refuses unless `allow_unbounded_phi` is set. + processed_archive_days: int = 0 # Past N HOURS, DELETE connection_event rows (#46) — the Corepoint-style transport/lifecycle log can # be high-volume (a connect-per-message sender, a probe storm), so it has its own short window in # HOURS (not days). 0 = inherit the message-body window (messages_days), the ADR 0021 §7.5 default. @@ -3008,9 +3018,10 @@ class RetentionSettings(_Section): vacuum_at: str = "" # Secure-by-default opt-out (#186a, ASVS 14.2.4): `serve` refuses to start under # [security].enforcement=enforce (warns under enforcement = warn) when an auto-bounded PHI window - # is unbounded — `messages_days`, `dead_letter_days` and `reference_snapshot_days` - # (`auto_bounded_windows()` is the list; the first two keep FULL raw message bodies until - # purged, the third covers orphaned reference snapshots only) — so PHI does not + # is unbounded — `messages_days`, `dead_letter_days`, `reference_snapshot_days` and + # `processed_archive_days` (`auto_bounded_windows()` is the list; the first two keep FULL raw + # message bodies until purged, the third covers orphaned reference snapshots only, the fourth the + # File source's `.processed` copies) — so PHI does not # accumulate without bound. Unless this override is set, an UNSET window is auto-bounded to 30 # days, so only an explicit 0 trips the gate. Setting this true is the explicit, audited override # that lets an instance run with unbounded (keep-forever) retention. Operators write it as @@ -3032,6 +3043,7 @@ class RetentionSettings(_Section): "app_log_compress_days", "search_preset_days", "reference_snapshot_days", + "processed_archive_days", ) @classmethod def _non_negative_days(cls, value: int) -> int: @@ -10180,3 +10192,77 @@ def keyless_opt_out_refusal(store: StoreSettings, security: SecuritySettings) -> ): return KEYLESS_REFUSED_BY_NO_STRICT_ACK return None + + +#: Directory settings whose WHOLE tree the engine owns (BACKLOG #1188): an archive inside one is +#: refused as well as one that is or holds it. ``base_dir`` is the config-repo root, which commonly +#: holds watch directories, so it is guarded as a directory only. +_OWNED_TREE_NAMES: frozenset[str] = frozenset({"destination"}) +_NOT_A_TREE: frozenset[str] = frozenset({"base_dir"}) + + +def engine_owned_dirs( + models: Iterable[object], + *, + dirs: Iterable[str | Path | None] = (), + trees: Iterable[str | Path | None] = (), + files: Iterable[str | Path | None] = (), +) -> tuple[tuple[Path, ...], tuple[Path, ...]]: + """The engine's own directories, which no ``.processed`` archive sweep may reach (BACKLOG #1188), + as ``(dirs, trees)``: an archive directory that is, or holds, any of either is refused, and one + inside a tree is refused too. + + Read from every settings model given, recursively, by field name, so a setting added later is + covered without a list to keep: at least a ``*_file``, ``*_files`` or ``*_path`` setting (and the + store's ``path`` and ``[logging].file``) contributes its parent directory, and a ``*_dir`` setting or the backup + ``destination`` contributes a tree (``base_dir``, the config-repo root, only a directory). A value + that names a URL is skipped. ``dirs``, ``trees`` and ``files`` add paths the settings models do + not hold, the config directory or the served TLS certificate say. Relative paths are taken + against the working directory, as the engine opens them.""" + out_dirs: list[Path] = [] + out_trees: list[Path] = [] + + def usable(value: object) -> str | None: + if isinstance(value, Path): + value = str(value) + if not isinstance(value, str) or not value.strip() or "://" in value: + return None + return value + + def add_file(value: object) -> None: + if (v := usable(value)) is not None: + out_dirs.append(Path(os.path.abspath(v)).parent) + + def walk(model: object, seen: set[int]) -> None: + if not isinstance(model, BaseModel) or id(model) in seen: + return + seen.add(id(model)) + for name in type(model).model_fields: + value = getattr(model, name, None) + if isinstance(value, BaseModel): + walk(value, seen) + elif name.endswith("_url") or "redirect" in name: + continue + elif name.endswith("_files") and isinstance(value, list | tuple): + for item in value: + add_file(item) + elif name.endswith(("_file", "_path")) or name in ("path", "file"): + add_file(value) + elif (name.endswith("_dir") or name in _OWNED_TREE_NAMES) and ( + v := usable(value) + ) is not None: + target = out_dirs if name in _NOT_A_TREE else out_trees + target.append(Path(os.path.abspath(v))) + + seen: set[int] = set() + for model in models: + walk(model, seen) + for d in dirs: + if (v := usable(d)) is not None: + out_dirs.append(Path(os.path.abspath(v))) + for d in trees: + if (v := usable(d)) is not None: + out_trees.append(Path(os.path.abspath(v))) + for f in files: + add_file(f) + return tuple(dict.fromkeys(out_dirs)), tuple(dict.fromkeys(out_trees)) diff --git a/messagefoundry/config/wiring.py b/messagefoundry/config/wiring.py index 436c96d34..ba9f4d5be 100644 --- a/messagefoundry/config/wiring.py +++ b/messagefoundry/config/wiring.py @@ -5504,6 +5504,17 @@ class Registry: # not-deployed target (ADR 0111) and must decide that identically on every shard. Names and a # flag only, not the connections, for the same reason as `all_loopback_inbound`. all_pt_inbound: Mapping[str, bool] | None = None + # EVERY inbound's DIRECTORY settings in the deployment (NAME -> connector type + only the settings + # it reads its watch and `.error` directories from, before env() resolution; None for a connector + # that declares directories but not their settings), pinned beside the shard identity by + # pipeline.sharding (None on an unfiltered graph / single-shard config, where `inbound` IS the + # whole config). Read only by the `.processed` archive sweep's guard (BACKLOG #1188): a sibling + # shard's watch directory is a fact about the CONFIG, and an archive directory that overlaps it + # would otherwise age out that shard's undelivered drops. Never a whole spec: a credential setting + # does not travel with it, and every other reader of `inbound` keeps seeing only this shard's. + all_inbound_guard_settings: ( + Mapping[str, tuple[ConnectorType, Mapping[str, Any] | None]] | None + ) = None # The directory :func:`load_config` read this graph from, or None for a graph built in code. A # `[sandbox].mode=subprocess` worker re-loads its Router/Handler functions from here, so after # an operator reload from another root the workers load the root the running graph came from, diff --git a/messagefoundry/pipeline/engine.py b/messagefoundry/pipeline/engine.py index c4551b7eb..6e9916fc0 100644 --- a/messagefoundry/pipeline/engine.py +++ b/messagefoundry/pipeline/engine.py @@ -46,6 +46,7 @@ ShadowSettings, StoreSettings, UpdateCheckSettings, + engine_owned_dirs, ) from messagefoundry.config.snapshot import ConfigSnapshot from messagefoundry.config.tls_policy import HopPosture, TrustAnchorPolicy @@ -582,8 +583,13 @@ def __init__( registry_preflight: Callable[[Registry, Mapping[str, Any]], Awaitable[None]] | None = None, settings_preflight: Callable[[], Awaitable[tuple[str, ...]]] | None = None, lane_anchor_check: LaneAnchorCheck | None = None, + owned_settings: Sequence[object] = (), ) -> None: self.store = store + # Settings models this engine is not otherwise handed ([auth], [security], [tls], ...), read + # only for the directories they name, which the `.processed` archive sweep must never reach + # (BACKLOG #1188, engine_owned_dirs). + self._owned_settings = tuple(owned_settings) # [sandbox] opt-in Router/Handler subprocess isolation (ADR 0087, #197). None → the # SandboxSettings defaults (mode=off, byte-identical). Rendered into a SandboxPolicy passed to # every runner this engine builds; read ONCE here (a /config/reload does NOT re-read it — @@ -936,6 +942,32 @@ def __init__( ) = None self._convergence_failures: tuple[int, set[str]] | None = None + def _engine_owned_dirs(self) -> tuple[tuple[Path, ...], tuple[Path, ...]]: + """The engine's own directories and trees for the ``.processed`` sweep's guard (BACKLOG + #1188), from every settings model this engine holds plus the paths it was handed directly.""" + return engine_owned_dirs( + ( + *self._owned_settings, + self._retention_settings, + self._cert_monitor_settings, + self._secret_rotation_settings, + self._update_check_settings, + self._backup_settings, + self._store_settings, + self._dr_settings, + self._sandbox_settings, + self._reference_settings, + self._shadow_settings, + ), + dirs=(self.config_dir,), + trees=(self._log_dir,), + files=( + self._api_tls_cert_file, + self._audit_anchor_file, + *self._api_tls_client_cert_files, + ), + ) + @classmethod async def create( cls, @@ -2029,6 +2061,7 @@ async def start(self) -> None: # leader-only WRITE singleton (purges bodies + writes audit rows), so it is gated on the # coordinator: in a cluster a follower's runner ticks but no-ops; single-node always leads. if self._retention_settings is not None: + engine_dirs, engine_trees = self._engine_owned_dirs() self._retention_runner = RetentionRunner( self.store, self._retention_settings, @@ -2042,6 +2075,25 @@ async def start(self) -> None: ), # [logging].log_dir for application-log-file retention (#120); None → sweep no-ops. log_dir=self._log_dir, + # The bound File inbounds whose `.processed` archive processed_archive_days ages out + # (BACKLOG #1188), read live each pass for the same reason as registry_source. + archive_source=lambda: ( + self._registry_runner.processed_archives() + if self._registry_runner is not None + else () + ), + # Every inbound's watch and `.error` directories on every engine shard, and every + # File outbound's target directory, bound or not, which that sweep must never reach + # (BACKLOG #1188). + guard_source=lambda: ( + self._registry_runner.archive_guard_dirs() + if self._registry_runner is not None + else () + ), + # And the engine's own directories and trees (store, keys, TLS pair, config, backup + # destination, log directory), read once from the settings (BACKLOG #1188). + engine_dirs=engine_dirs, + engine_trees=engine_trees, ) self._retention_runner.start() # [backup] turnkey DR backup (ADR 0049, #60) — a maintenance task like retention: independent of diff --git a/messagefoundry/pipeline/retention.py b/messagefoundry/pipeline/retention.py index c90afeb78..6364afab3 100644 --- a/messagefoundry/pipeline/retention.py +++ b/messagefoundry/pipeline/retention.py @@ -34,9 +34,10 @@ import os import shutil import tempfile +import threading import time -from collections.abc import Callable, Mapping -from dataclasses import dataclass, field +from collections.abc import Callable, Iterable, Mapping, Sequence +from dataclasses import dataclass, field, replace from pathlib import Path from messagefoundry.config.settings import RetentionSettings, StoreBackend @@ -44,7 +45,14 @@ from messagefoundry.parsing.compression import CompressionError, gzip_compress, gzip_decompress from messagefoundry.pipeline.alerts import AlertSink, LoggingAlertSink from messagefoundry.pipeline.cluster import ClusterCoordinator, NullCoordinator +from messagefoundry.redaction import safe_exc from messagefoundry.store import Store, StripResult +from messagefoundry.transports.base import ( + ArchiveGuard, + ProcessedArchive, + archive_dir_key, + build_archive_guard, +) __all__ = [ "DiskFloorReading", @@ -269,6 +277,10 @@ class RetentionPass: # declares. Metadata only: a count, never a key or a value. 0 when the window is unset, when no # registry is wired, or when the registry declares no reference sets (the positive-signal skip). reference_snapshots_purged: int = 0 + # Archived copies DELETEd from running File inbounds' `.processed` directories this pass, past + # `processed_archive_days` (owner ruling A2 of 2026-10-08, BACKLOG #1188). A count only, never a + # file or connection name. 0 when the window is unset or no File inbound is bound. + processed_archives_deleted: int = 0 # This pass hit the `max_pass_seconds` between-phase duration cap (#121, ADR 0137) and SKIPPED its # remaining phases — the skipped tail (incl. any due WAL-checkpoint/VACUUM) re-runs next interval with # its last-run marker unadvanced. Always False when the cap is off (`max_pass_seconds<=0`, the @@ -311,6 +323,7 @@ def did_work(self) -> bool: or self.app_logs_compressed > 0 or self.search_presets_purged > 0 or self.reference_snapshots_purged > 0 + or self.processed_archives_deleted > 0 or self.vacuumed or self.over_limit or self.capped @@ -333,8 +346,25 @@ def __init__( coordinator: ClusterCoordinator | None = None, registry_source: Callable[[], Registry | None] | None = None, log_dir: str | None = None, + archive_source: Callable[[], Sequence[ProcessedArchive]] | None = None, + guard_source: Callable[[], Iterable[Path] | None] | None = None, + engine_dirs: Iterable[Path] = (), + engine_trees: Iterable[Path] = (), ) -> None: self._store = store + # The engine's own directories (store, keys, TLS pair, config) and trees (backup destination, + # log directory), which no archive sweep may reach either (BACKLOG #1188): an archive that is + # or holds one is refused, and so is one inside a tree. + self._engine_dirs = tuple(engine_dirs) + self._engine_trees = tuple(engine_trees) + # The bound inbounds whose `.processed` archive the `processed_archive_days` sweep ages out + # (BACKLOG #1188), read from the LIVE graph each pass like the registry. None (embedding, + # tests) = nothing to sweep. + self._archive_source = archive_source + # Every inbound's watch and `.error` directories across all engine shards, and every File + # outbound's target directory, bound or not, which no archive sweep may reach. Returning None + # means one could not be resolved, so the sweep deletes nothing. + self._guard_source = guard_source self._settings = settings # The SQLite store file the low-disk floor measures (BACKLOG #290); None when the floor # cannot apply to this store. See sqlite_store_file for the rule. @@ -385,6 +415,7 @@ def enabled(self) -> bool: or s.connection_event_retention_hours or s.search_preset_days or s.reference_snapshot_days + or s.processed_archive_days or s.max_db_mb or s.audit_log_warn_rows or self._wal_truncate_applies() @@ -418,7 +449,8 @@ def start(self) -> None: log.info( "retention enabled: messages_days=%d dead_letter_days=%d max_db_mb=%d " "min_free_disk_mb=%d audit_log_warn_rows=%d wal_checkpoint_seconds=%g vacuum_at=%r " - "app_log_days=%d app_log_compress_days=%d max_pass_seconds=%g (every %gs)", + "app_log_days=%d app_log_compress_days=%d processed_archive_days=%d " + "max_pass_seconds=%g (every %gs)", self._settings.messages_days, self._settings.dead_letter_days, self._settings.max_db_mb, @@ -428,6 +460,7 @@ def start(self) -> None: self._settings.vacuum_at, self._settings.app_log_days, self._settings.app_log_compress_days, + self._settings.processed_archive_days, self._settings.max_pass_seconds, self._settings.purge_interval_seconds, ) @@ -703,6 +736,16 @@ def _deadline_hit() -> bool: self._compress_app_logs, now, _deadline_hit ) + # The File source's `.processed` archive (owner ruling A2 of 2026-10-08, BACKLOG #1188): a + # clear-text copy of every inbound payload, aged out on ARCHIVE time at the body window. Each + # source does its own blocking I/O off the loop and under its own credential, re-reading the + # deadline per file, so a large directory releases the pass like the compressor does. + processed_archives_deleted = 0 + if not _deadline_hit() and s.processed_archive_days > 0: + processed_archives_deleted = await self._prune_processed_archives( + now - s.processed_archive_days * _SECONDS_PER_DAY, _deadline_hit + ) + # Maintenance phases (WAL-checkpoint, then daily VACUUM). Their cadence markers advance ONLY when # the phase actually runs, so a phase SKIPPED by the cap stays due for the next pass (#121). A # running VACUUM is non-interruptible: the deadline is checked only BEFORE it is dispatched. @@ -761,6 +804,7 @@ def _deadline_hit() -> bool: app_log_bytes_reclaimed=app_log_bytes_reclaimed, search_presets_purged=search_presets_purged, reference_snapshots_purged=reference_snapshots_purged, + processed_archives_deleted=processed_archives_deleted, capped=capped, below_disk_floor=below_disk_floor, wal_left_untruncated=wal_left_untruncated, @@ -770,6 +814,92 @@ def _deadline_hit() -> bool: await self._audit(result) return result + async def _prune_processed_archives( + self, older_than: float, deadline_hit: Callable[[], bool] + ) -> int: + """Run each bound archive's sweep, and return how many files went in all (BACKLOG #1188). + + Every archive is handed the watch and ``.error`` directories of every inbound in the + deployment, on every engine shard, and every File outbound's target directory, bound or not, + so one whose archive directory overlaps any connection's drops, quarantine or deliveries + deletes nothing. When a connection's directories cannot be read from its configuration the + whole sweep is skipped: the guard is only as good as its list. A guarded directory whose + identity cannot be read is read again by each archive under its own identity, and an archive + that still cannot read it is not swept (:class:`ArchiveGuard`), nor is one whose opened + directory has no readable identity. A directory two connections share is swept once. A + fault in one connection is logged and the others still run, because a sweep must never stop the pass: a pass that + raised here would lose the audit row for what it had already purged.""" + # Everything up to the sweep itself is contained: a fault here, a connector's declared + # directories raising say, must not escape the pass and lose its audit row. + try: + archives = list(self._archive_source()) if self._archive_source is not None else [] + if not any(a.archiving for a in archives): + return 0 + guarded: list[Path] = [p for a in archives for p in (a.directory, a.error_dir)] + if self._guard_source is not None: + extra = self._guard_source() + if extra is None: + log.warning( + "the [retention].processed_archive_days sweep deleted nothing this pass: a " + "connection's directories could not be resolved from its configuration, so " + "the sweep cannot tell what it must not reach" + ) + return 0 + guarded.extend(extra) + guarded.extend(self._engine_dirs) + guarded.extend(self._engine_trees) + protected, keys = await asyncio.to_thread( + _archive_keys, guarded, [a.processed_dir for a in archives], self._engine_trees + ) + except (OSError, ValueError, TypeError, KeyError) as exc: + log.warning( + "the [retention].processed_archive_days sweep deleted nothing this pass: the " + "directories it must not reach could not be listed (%s)", + safe_exc(exc), + ) + return 0 + # Set when this task is cancelled, so a sweep already running on a worker thread stops at its + # next file instead of deleting on after the engine has stopped. + cancelled = threading.Event() + + def stop_hit() -> bool: + return cancelled.is_set() or deadline_hit() + + swept: set[str] = set() + deleted = 0 + for archive, key in zip(archives, keys, strict=True): + if stop_hit(): + break + # Only a source that archives is swept, and only it claims the directory, so a leave + # source sharing a directory with a move source does not shadow it. + if not archive.archiving or key in swept: + continue + try: + outcome = await archive.prune_processed_archive( + older_than, protected=protected, deadline_hit=stop_hit + ) + except asyncio.CancelledError: + cancelled.set() + raise + except OSError as exc: + # A file fault is the sweep's own to absorb; this catches what escapes it, the + # alternate credential failing say, which the protocol raises as an OSError. + # safe_exc, because an OSError renders its path and below the archive directory a + # path is a partner's file name. + log.warning( + "the [retention].processed_archive_days sweep failed for the archive directory " + "%s (%s); nothing more was deleted there this pass", + archive.processed_dir.name, + safe_exc(exc), + ) + continue + # Only a sweep that ran claims the directory: one refused (an unreadable guarded + # directory under its identity, say) or failed leaves it to a sibling that shares it. + if outcome is not None: + swept.add(key) + deleted += outcome + return deleted + # --- per-connection retention overrides (#34, ADR 0027) ------------------ def _resolve_overrides(self) -> tuple[dict[str, int], dict[str, int]]: @@ -1268,6 +1398,10 @@ async def _audit(self, result: RetentionPass) -> None: # narrow which patient cohort was dropped; the count is what an assessor needs. "reference_snapshot_days": self._settings.reference_snapshot_days, "reference_snapshots_purged": result.reference_snapshots_purged, + # `.processed` archive copies deleted (BACKLOG #1188) + the window. The COUNT only: + # never a file name, which can carry an identifier, nor a connection name. + "processed_archive_days": self._settings.processed_archive_days, + "processed_archives_deleted": result.processed_archives_deleted, # Between-phase duration cap (#121, ADR 0137): the configured ceiling + whether THIS pass # hit it and skipped its remaining phases. Timing metadata only — no PHI. "max_pass_seconds": self._settings.max_pass_seconds, @@ -1302,3 +1436,27 @@ def _vacuum_due(self, now: float) -> bool: def _day_key(now: float) -> str: lt = time.localtime(now) return f"{lt.tm_year:04d}-{lt.tm_mon:02d}-{lt.tm_mday:02d}" + + +def _archive_keys( + guarded: Iterable[Path], archives: Sequence[Path], trees: Iterable[Path] = () +) -> tuple[ArchiveGuard, list[str]]: + """Build the guard over the protected directories once per pass, and each archive directory's + key for sweeping a shared directory once (BACKLOG #1188). Blocking: run off the loop. + + The once-only key is resolved but NOT case-folded: two directories that differ only in case on a + case-sensitive volume are two archives, and folding would leave the second unswept.""" + keys = [] + for path in archives: + try: + keys.append(os.path.normcase(os.path.realpath(path))) + except (OSError, ValueError): + keys.append(os.path.normcase(os.path.abspath(path))) + tree_keys = set() + for tree in trees: + try: + tree_keys.add(archive_dir_key(tree)) + except (OSError, ValueError): + tree_keys.add(os.path.normcase(os.path.abspath(tree)).casefold()) + guard = build_archive_guard(guarded) + return replace(guard, trees=frozenset(tree_keys)), keys diff --git a/messagefoundry/pipeline/sharding.py b/messagefoundry/pipeline/sharding.py index 848507ac5..1484db922 100644 --- a/messagefoundry/pipeline/sharding.py +++ b/messagefoundry/pipeline/sharding.py @@ -54,11 +54,14 @@ from __future__ import annotations import hashlib -from collections.abc import Iterable, Sequence +from collections.abc import Iterable, Mapping, Sequence from types import MappingProxyType +from typing import Any +from messagefoundry.config.models import ConnectorType from messagefoundry.config.settings import StoreBackend from messagefoundry.config.wiring import Registry +from messagefoundry.transports.base import guard_dir_settings #: The implicit shard every ``shard=None`` (untagged) inbound connection belongs to. A config with no #: shard tags at all is therefore a single-shard deployment named ``"default"`` — ``supervise`` spawns @@ -169,6 +172,21 @@ def owned_destination_set(registry: Registry, shard: str, ids: Sequence[str]) -> ) +def inbound_guard_settings( + registry: Registry, +) -> dict[str, tuple[ConnectorType, Mapping[str, Any] | None]]: + """Every deployment inbound's directory settings for the ``.processed`` archive sweep's guard + (BACKLOG #1188), NAME -> (type, :func:`guard_dir_settings`): the pinned unfiltered map when + ``registry`` is one engine shard's filtered view, else derived from its own inbounds. Only the + directory settings, never a whole spec, so no credential setting travels with it.""" + if registry.all_inbound_guard_settings is not None: + return dict(registry.all_inbound_guard_settings) + return { + name: (ic.spec.type, guard_dir_settings(ic.spec.type, ic.spec.settings)) + for name, ic in registry.inbound.items() + } + + def filter_registry_for_shard(registry: Registry, shard: str) -> Registry: """A :class:`Registry` exposing ONLY ``shard``'s inbound connections, sharing everything else. @@ -239,4 +257,11 @@ def filter_registry_for_shard(registry: Registry, shard: str) -> Registry: all_pt_inbound=( MappingProxyType(dict(registry.passthrough_inbounds())) if sharded else None ), + # Pin the UNFILTERED inbounds' directory settings for the `.processed` archive sweep's guard + # (BACKLOG #1188): an archive directory equal to, or holding, a sibling shard's watch or + # `.error` directory would otherwise age out that shard's drops. Derived from the SOURCE + # registry, as above. + all_inbound_guard_settings=( + MappingProxyType(inbound_guard_settings(registry)) if sharded else None + ), ) diff --git a/messagefoundry/pipeline/wiring_runner.py b/messagefoundry/pipeline/wiring_runner.py index 581c1c0a0..83a51db3d 100644 --- a/messagefoundry/pipeline/wiring_runner.py +++ b/messagefoundry/pipeline/wiring_runner.py @@ -37,6 +37,7 @@ from dataclasses import dataclass from datetime import UTC, datetime from enum import Enum, StrEnum, auto +from pathlib import Path from typing import Any, Literal, NamedTuple, Protocol, cast from messagefoundry.auth.ratelimit import SlidingWindowRateLimiter @@ -189,7 +190,7 @@ graph_shape, ) from messagefoundry.pipeline.saturation import SaturationDetector -from messagefoundry.pipeline.sharding import owner_shard_of_destination +from messagefoundry.pipeline.sharding import inbound_guard_settings, owner_shard_of_destination from messagefoundry.pipeline.stage_dispatcher import ( _REPEND_RETRY_CAP_SECONDS, HoldResult, @@ -229,9 +230,13 @@ IntakeAuditSink, IntakeGate, IntakeRateLimiter, + ProcessedArchive, ReplyBodyDropped, SyncReplyResolver, check_file_key, + guard_dir_settings, + guarded_destination_dirs, + guarded_source_dirs, ) from messagefoundry.transports.database import DatabaseLookupExecutor from messagefoundry.transports.fhir import FhirLookupExecutor @@ -3048,6 +3053,60 @@ def listening_inbounds(self) -> frozenset[str]: """The inbounds whose listener is bound now.""" return frozenset(self._sources) + def processed_archives(self) -> tuple[ProcessedArchive, ...]: + """The bound inbounds whose source archives what it read, such as the File source's + ``.processed`` directory, for the retention pass's age sweep (BACKLOG #1188). Only bound + ones: a stopped or undeployed inbound has no source object to resolve its directories.""" + return tuple(s for s in self._sources.values() if isinstance(s, ProcessedArchive)) + + def archive_guard_dirs(self) -> tuple[Path, ...] | None: + """The directories the ``.processed`` age sweep must never reach (BACKLOG #1188): the watch + and ``.error`` directories of EVERY inbound in the deployment, and the target directory of + every outbound, bound or not, deployed or not. Read from the configuration rather than the + running connections, so a stopped, failed or DR-parked one stays protected. Under + ``serve --shard`` this shard's ``inbound`` map holds only its own inbounds, so the inbounds + come from :func:`inbound_guard_settings`, the whole config's; outbounds are never + filtered. Only the settings a connector declares its directories from are resolved + (:func:`guard_dir_settings`). None when one of those cannot be resolved (a missing env() + value, or a connector that declares directories but not their settings): the caller then + deletes nothing, because a guard that silently drops a directory fails open.""" + connections = [ + (False, name, kind, declared, inbound_record_name(name)) + for name, (kind, declared) in inbound_guard_settings(self.registry).items() + ] + [ + ( + True, + name, + oc.spec.type, + guard_dir_settings(oc.spec.type, oc.spec.settings, outbound=True), + name, + ) + for name, oc in self.registry.outbound.items() + ] + dirs: list[Path] = [] + for outbound, name, kind, declared, record in connections: + try: + if declared is None: + raise TypeError("directories declared without the settings they come from") + if not declared: + continue # a fault in a connection with no directories cannot hide one + settings = resolve_env_settings(declared, self._env_values, connection=record) + dirs.extend( + guarded_destination_dirs(Destination(type=kind, name=name, settings=settings)) + if outbound + else guarded_source_dirs(Source(type=kind, name=name, settings=settings)) + ) + except (WiringError, ValueError, TypeError, KeyError) as exc: + log.warning( + "%s connection %r: its directories could not be resolved for the " + "[retention].processed_archive_days guard (%s)", + "outbound" if outbound else "inbound", + name, + type(exc).__name__, + ) + return None + return tuple(dirs) + async def park_intake(self, standby: Priority, *, keep: frozenset[str] = frozenset()) -> None: """Make this box a passive standby for intake and unbind every inbound, in one span of the reload lock, for ``POST /dr/release`` (vault BACKLOG #3140). The DR threshold is kept, so diff --git a/messagefoundry/transports/base.py b/messagefoundry/transports/base.py index 6bc58665a..b73223892 100644 --- a/messagefoundry/transports/base.py +++ b/messagefoundry/transports/base.py @@ -24,15 +24,18 @@ import abc import asyncio import contextlib +import errno import math import numbers import operator +import os import re import ssl -from collections.abc import Awaitable, Callable, Mapping +from collections.abc import Awaitable, Callable, Iterable, Mapping from dataclasses import dataclass, field from enum import Enum -from typing import Any, ClassVar, Protocol +from pathlib import Path +from typing import Any, ClassVar, Protocol, runtime_checkable from messagefoundry.config.models import ConnectorType, ContentType, Destination, Source from messagefoundry.config.settings import EgressSettings @@ -57,6 +60,14 @@ "ReplyOutcome", "SyncReplyResolver", "ProcessedFileLedger", + "ProcessedArchive", + "ArchiveGuard", + "archive_dir_key", + "build_archive_guard", + "guarded_destination_dirs", + "guard_dir_settings", + "guarded_setting_keys", + "guarded_source_dirs", "FILE_KEY_SCHEME", "FILE_KEY_SCHEMES", "FileKeyError", @@ -626,6 +637,187 @@ def __repr__(self) -> str: # pragma: no cover - trivial, but load-bearing for P SyncReplyResolver = Callable[[str], Awaitable[InboundReply]] +@runtime_checkable +class ProcessedArchive(Protocol): + """A running inbound that archives what it read into a directory the retention pass may age out: + the local File source's ``.processed`` (owner ruling A2 of 2026-10-08, BACKLOG #1188). + + The retention runner reads these from the live graph each pass and never imports a transport. It + hands every archive, as ``protected``, an :class:`ArchiveGuard` over each watch and ``.error`` + directory any inbound in the deployment names (:func:`guarded_source_dirs`) and each outbound + target directory (:func:`guarded_destination_dirs`), bound or not, so one whose archive directory + is, or holds, any of those deletes nothing: a misconfigured ``processed_subdir`` must never let an + age sweep reach undelivered drops, files awaiting a partner's pickup, or the ``.error`` copies, + whose deletion nobody has ruled on (#3335). ``archiving`` says whether this source writes the + archive at all; one that does not is protected but never swept.""" + + directory: Path + processed_dir: Path + error_dir: Path + + @property + def archiving(self) -> bool: ... + + async def prune_processed_archive( + self, + older_than: float, + *, + protected: ArchiveGuard, + deadline_hit: Callable[[], bool], + ) -> int | None: + """Delete archived files whose archive time is before ``older_than`` (epoch seconds), and + return how many went, or None when this archive refused or could not open its directory, + which leaves a directory it shares to a sibling. A file it cannot remove is skipped and + tried again next pass. Only an ``OSError`` may escape, and the caller contains it.""" + ... + + +@dataclass(frozen=True) +class ArchiveGuard: + """What an archive sweep must not reach (BACKLOG #1188). + + ``keys`` holds each guarded directory's :func:`archive_dir_key`, for the check before the archive + directory is opened. ``ids`` holds the ``(st_dev, st_ino)`` of each guarded directory and of every + readable directory above one, for the check once it is open: an archive directory that is, or + holds, a guarded one under another name (a UNC path and the local path of one share, a bind + mount, a DFS link, a lexical ``x/..``) is recognised by what was actually opened. + + ``unresolved`` holds each guarded directory that exists, or may, but whose identity could not be + read, an unreachable share or a refused ``stat`` say. Without that identity an alias of it would + pass the check once the archive is open, so each archive reads them again under its own + identity (an alternate credential may reach a share the engine's own account cannot), and an + archive that still cannot read one deletes nothing: the guard fails closed, as the sweep does + for an opened archive directory whose identity it cannot read.""" + + keys: frozenset[str] = frozenset() + ids: frozenset[tuple[int, int]] = frozenset() + unresolved: frozenset[Path] = frozenset() + #: The :func:`archive_dir_key` of each engine-owned tree (the backup destination, the log + #: directory, say): an archive directory INSIDE one is refused too, not only one that is or holds + #: it. Compared by resolved path only. + trees: frozenset[str] = frozenset() + + +def archive_dir_key(path: Path) -> str: + """The form in which an archive sweep compares directories (BACKLOG #1188): links resolved, and + case folded on EVERY platform. A case-sensitive volume holding two directories that differ only in + case then reads them as one, which can only make the sweep refuse more, never reach further. + Blocking (``realpath``), and it can raise ``OSError`` on an unreachable share.""" + return os.path.normcase(os.path.realpath(path)).casefold() + + +#: What an absent directory raises: one not created yet (a ``.error`` before the first quarantine, an +#: outbound directory before the first write) has nothing an archive could alias. Windows is read by +#: its own codes, an allowlist, because it maps an unreachable network path (``ERROR_BAD_NETPATH``, +#: ``ERROR_BAD_NET_NAME``) to ``ENOENT`` too, and those must fail closed. +_ABSENT_ERRNOS = frozenset({errno.ENOENT, errno.ENOTDIR}) +_ABSENT_WINERRORS = frozenset({2, 3}) # ERROR_FILE_NOT_FOUND, ERROR_PATH_NOT_FOUND + + +def _is_absent(exc: OSError) -> bool: + winerror = getattr(exc, "winerror", None) + if winerror is not None: + return winerror in _ABSENT_WINERRORS + return exc.errno in _ABSENT_ERRNOS + + +def build_archive_guard(dirs: Iterable[Path]) -> ArchiveGuard: + """The :class:`ArchiveGuard` over ``dirs``. Blocking (``realpath``, ``stat``). + + A directory that cannot be resolved is compared by its absolute path. Its identity, and that of + each directory above it, is recorded for the check once the archive is open. A directory that + does not exist yet stands in by its nearest existing ancestor. A directory whose identity cannot + be read for any other reason goes into :attr:`ArchiveGuard.unresolved`, and so does one none of + whose path can be read, and so does one whose stat reports inode 0: Windows answers a refused + directory from its parent's listing with inode 0 rather than raising, and some file systems + report 0 for every directory. An ancestor above a readable directory that cannot itself be read + (a traverse-only parent) is passed over, and an ancestor's inode 0 is not recorded, since it + would match every directory so reported. Each directory is stat'ed once per call.""" + keys: set[str] = set() + ids: set[tuple[int, int]] = set() + unresolved: set[Path] = set() + seen: set[Path] = set() + for path in dict.fromkeys(dirs): + try: + here = Path(os.path.realpath(path)) + except (OSError, ValueError): + here = Path(os.path.abspath(path)) + keys.add(os.path.normcase(str(here)).casefold()) + found = False + for candidate in (here, *here.parents): + if candidate in seen: + found = True + break # it, with its identity, and everything above it were recorded already + try: + st = os.stat(candidate) + except (OSError, ValueError) as exc: + if found: + continue # a traverse-only ancestor; the guarded directory itself was read + if isinstance(exc, OSError) and _is_absent(exc): + continue # not created yet; the nearest existing ancestor stands in for it + break # the guarded directory exists, or may, and cannot be read + if not st.st_ino and not found: + break # no identity for the guarded directory (or its stand-in): unresolved + found = True + if st.st_ino: + # Only a directory with an identity is memoised: one of inode 0 reached first as an + # ancestor must still be unresolved when it is itself a guarded directory. + seen.add(candidate) + ids.add((st.st_dev, st.st_ino)) + if not found: + unresolved.add(Path(path)) + return ArchiveGuard(frozenset(keys), frozenset(ids), frozenset(unresolved)) + + +def guarded_setting_keys(kind: ConnectorType, *, outbound: bool = False) -> tuple[str, ...] | None: + """The settings the connector registered for ``kind`` reads its guarded directories from + (:func:`guarded_source_dirs`, :func:`guarded_destination_dirs`): ``()`` when it declares no + directories, and None when it declares directories but not the settings they come from.""" + builder = (_DESTINATIONS if outbound else _SOURCES).get(kind) + if not callable(getattr(builder, "guarded_dirs", None)): + return () + keys = getattr(builder, "guarded_setting_keys", None) + return None if keys is None else tuple(keys) + + +def guard_dir_settings( + kind: ConnectorType, settings: Mapping[str, Any], *, outbound: bool = False +) -> dict[str, Any] | None: + """Only the settings a connection of ``kind`` reads its guarded directories from, copied out of + ``settings`` before ``env()`` resolution (:func:`guarded_setting_keys`): empty when it declares no + directories, and None when it declares directories but not their settings, which the guard + treats as unresolvable (fail closed) rather than handing ``guarded_dirs`` every setting. Nothing + else is carried, so a credential setting never travels with the guard, and another setting's + ``env()`` value, a not-deployed connection's secret say, is neither read nor able to switch the + sweep off (BACKLOG #1188).""" + keys = guarded_setting_keys(kind, outbound=outbound) + if keys is None: + return None + return {k: settings[k] for k in keys if k in settings} + + +def guarded_source_dirs(config: Source) -> tuple[Path, ...]: + """The directories an inbound built from ``config`` reads drops from or quarantines into, which no + archive sweep may reach (BACKLOG #1188). Empty for a connector that declares none. A connector + declares them with a ``guarded_dirs(config)`` callable on what it registered, and the settings + that callable reads in ``guarded_setting_keys``, so the pipeline asks without naming a connector + type.""" + declare = getattr(_SOURCES.get(config.type), "guarded_dirs", None) + if not callable(declare): + return () + return tuple(declare(config)) + + +def guarded_destination_dirs(config: Destination) -> tuple[Path, ...]: + """The directories an outbound built from ``config`` writes into, which no archive sweep may reach + (BACKLOG #1188): a written file can be awaiting a partner's pickup. Empty for a connector that + declares none. Declared like :func:`guarded_source_dirs`.""" + declare = getattr(_DESTINATIONS.get(config.type), "guarded_dirs", None) + if not callable(declare): + return () + return tuple(declare(config)) + + class ProcessedFileLedger(Protocol): """The store-backed **process-in-place dedup ledger** a leave-in-place (``after_read='leave'``) poll source uses (ADR 0129, BACKLOG #142). The runner injects an adapter closing over the store + the diff --git a/messagefoundry/transports/file.py b/messagefoundry/transports/file.py index a55309aaa..d3149fe11 100644 --- a/messagefoundry/transports/file.py +++ b/messagefoundry/transports/file.py @@ -37,7 +37,7 @@ from collections.abc import Callable, Iterator, Mapping from contextlib import contextmanager, suppress from pathlib import Path -from typing import Any, BinaryIO, Self, TypeVar +from typing import Any, BinaryIO, ClassVar, Self, TypeVar from messagefoundry.config.models import ( ConnectorType, @@ -54,6 +54,7 @@ from messagefoundry.transports import wincred from messagefoundry.transports.base import ( DEFAULT_MAX_ITEMS_PER_POLL, + ArchiveGuard, DeliveryError, DestinationConnector, DestinationStartupError, @@ -61,6 +62,8 @@ NegativeAckError, SourceConnector, SourceStartupError, + archive_dir_key, + build_archive_guard, encode_wire_body, file_key_forms, float_setting, @@ -462,6 +465,19 @@ def __init__(self, config: Destination) -> None: # Set once a directory fsync has failed here, so it is logged once and not retried (#1618). self._dir_fsync_unsupported = False + #: The settings :meth:`guarded_dirs` reads, the only ones the guard resolves. + guarded_setting_keys: ClassVar[tuple[str, ...]] = ("directory",) + + @staticmethod + def guarded_dirs(config: Destination) -> tuple[Path, ...]: + """The target directory ``config`` names, which no ``.processed`` archive sweep may reach + (BACKLOG #1188): a delivered file there may still be awaiting the partner's pickup. Computed + from the configuration alone, so a stopped or undeployed outbound is protected too.""" + s = config.settings + if "directory" not in s: + return () + return (Path(s["directory"]),) + async def _run_fs(self, fn: Callable[..., _T], *args: Any, **kwargs: Any) -> _T: """Run a blocking filesystem callable off the event loop — under the alternate credential (on a dedicated impersonated thread) when one is configured, else via the shared @@ -704,6 +720,8 @@ def __init__(self, config: Source) -> None: # BACKLOG #2507: paths already refused as links, so each is warned about once. See # _log_unconfined. In memory only and never logged. self._refused: set[str] = set() + # The connection, as the `.processed` age sweep names it in a log line (BACKLOG #1188). + self._label = config.name or str(self.directory) # Opt-in at-start directory validation (#114, ADR 0031 amendment). Default off = the historical # run-time deferral (a missing dir is logged-and-retried each poll, never fails start). self.validate_directory: bool = bool(s.get("validate_directory", False)) @@ -1662,9 +1680,11 @@ def _after_processing( safe_exc(exc, file_name=path.name), ) else: - self._move(path, self.processed_dir, read_id) + self._move(path, self.processed_dir, read_id, stamp=True) - def _move(self, path: Path, dest_dir: Path, read_id: _FileId | None = None) -> bool: + def _move( + self, path: Path, dest_dir: Path, read_id: _FileId | None = None, *, stamp: bool = False + ) -> bool: """Archive ``path`` into ``dest_dir`` under a name claimed ATOMICALLY (BACKLOG #1046). Returns whether a copy now sits in ``dest_dir``. A quarantine is recorded only then (BACKLOG #1621): a file that could not be archived at all is logged here and examined again next scan. @@ -1692,11 +1712,19 @@ def _move(self, path: Path, dest_dir: Path, read_id: _FileId | None = None) -> b the original would survive). If the unlink fails after the claim the file is archived AND left in place to be re-read — the same duplicate-read outcome the pre-existing failure arm already had, and logged the same way. A Windows rename by handle moves the file in one step, - so there the original cannot stay behind.""" + so there the original cannot stay behind. + + ``stamp`` sets the file's modification time to now just before it is archived, which is what + the ``.processed`` age sweep keys on (:meth:`prune_processed_archive`, BACKLOG #1188). Before + the claim, not after: a link or a rename keeps the drop's own mtime, which a partner's copy + tool may have preserved from long ago, and a sweep running between the claim and a later stamp + would delete a copy archived a moment earlier.""" pin = self._pinned(path, read_id) if pin is None: return False with pin: + if stamp: + self._stamp_archive_time(pin, path) try: with _open_dest(dest_dir, self.directory, self._root_real) as dest: gone = _archive(pin, dest, path.name) @@ -1724,6 +1752,327 @@ def _move(self, path: Path, dest_dir: Path, read_id: _FileId | None = None) -> b ) return True # archived, even if the original stayed and will be re-read + def _stamp_archive_time(self, pin: _Pin, path: Path) -> None: + """Set the pinned file's mtime to now, so the ``.processed`` sweep ages it from its archive. + + Through the pin, never by a fresh name lookup: POSIX names it relative to the checked + directory without following a link, and Windows reopens the checked handle itself. POSIX asks + for the current time (``UTIME_NOW``) rather than passing one: that needs only write access, + where an explicit time needs the file's owner, and a partner's own account often owns a drop. + A failure does not stop the archive, because refusing to archive would re-read the file on + every poll. On POSIX it is only a debug line: the sweep keys on the later of mtime and ctime, + and the archive move itself sets ctime, which nobody can set back. On Windows it is a warning, + because that copy then keeps its drop-time mtime and may age out sooner than its window.""" + try: + if pin.fd is None: + if not _UTIME_AT: + raise OSError(errno.ENOTSUP, "cannot set a time relative to a directory here") + os.utime(pin.name, dir_fd=pin.dir_fd, follow_symlinks=False) + else: + _win_stamp_write_time(pin.fd, time.time_ns()) + except OSError as exc: + logger.log( + logging.DEBUG if pin.fd is None else logging.WARNING, + "could not set the archive time on %s before moving it to %s (%s); the " + "[retention].processed_archive_days sweep will age that copy from its own earlier " + "modification time", + safe_name(path.name), + self.processed_dir.name, + _describe_os_error(exc, file_name=path.name), + ) + + #: The settings :meth:`guarded_dirs` reads, the only ones the guard resolves. + guarded_setting_keys: ClassVar[tuple[str, ...]] = ("directory", "error_subdir") + + @staticmethod + def guarded_dirs(config: Source) -> tuple[Path, ...]: + """The watch and ``.error`` directories ``config`` names, which no archive sweep may reach + (BACKLOG #1188). Computed from the configuration alone, as ``__init__`` computes them, so the + retention pass protects an inbound that is stopped, failed, parked or bound on another engine + shard as well as a running one.""" + s = config.settings + if "directory" not in s: + return () + directory = Path(s["directory"]) + return directory, directory / s.get("error_subdir", ".error") + + @property + def archiving(self) -> bool: + """Whether this source writes a ``.processed`` archive at all: only a move does.""" + return self.after_read == "move" + + async def prune_processed_archive( + self, + older_than: float, + *, + protected: ArchiveGuard, + deadline_hit: Callable[[], bool], + ) -> int | None: + """Delete files in ``processed_dir`` archived before ``older_than``; return how many went. + + The ``[retention].processed_archive_days`` sweep (owner ruling A2 of 2026-10-08, BACKLOG + #1188), run by the retention pass. Off the event loop, and under this connection's alternate + credential when one is set, through :meth:`_run_fs`. It also stops between files once this + source is stopping. See :meth:`_prune_processed`. A credential fault, or a credential worker + already shut down by a reload, is raised as an ``OSError``, the one error the protocol lets + escape, so the retention pass needs no transport's exception types.""" + + def stop_hit() -> bool: + return self._stop.is_set() or deadline_hit() + + try: + return await self._run_fs(self._prune_processed, older_than, protected, stop_hit) + except (wincred.CredentialError, RuntimeError) as exc: + raise OSError( + "the sweep could not run, under the alternate credential if one is set: " + f"{type(exc).__name__}" + ) from exc + + def _prune_processed( + self, older_than: float, protected: ArchiveGuard, deadline_hit: Callable[[], bool] + ) -> int | None: + """Blocking half of :meth:`prune_processed_archive`. + + What it deletes, and what it never does: + + * Nothing unless ``after_read`` is ``"move"``: only a move archives, so a leave or delete + source's ``.processed``, which may hold another system's files, is not this sweep's. + * Only REGULAR files directly in ``processed_dir``: one level, since every archive move names + its copy by the bare file name. A subdirectory, a link and a junction are left alone, and + the removal itself never follows a link (POSIX unlinks the name relative to the opened + directory; Windows deletes through a handle opened on the file as itself). + * Keyed on archive time: mtime, which :meth:`_move` sets as it archives, and on POSIX the + later of that and ctime, which the archive move sets and nobody can set back. Read again on + the file itself just before it is removed. File content is never read, and no file name is + logged: a drop's name can carry an identifier. + * Nothing at all when ``processed_dir`` is, or holds, this connection's watch directory or + ``.error``, or any directory ``protected`` guards, or when one of those exists but its + identity cannot be read under this connection's identity. A ``processed_subdir`` of + ``"."`` or ``".."`` would otherwise age out undelivered drops, and ``.error`` deletion is + unruled (#3335). The directory is compared by resolved, case-folded path before it is opened, and + by identity once it is (:class:`ArchiveGuard`). + * The directory is opened as an archive move opens it (:func:`_open_dest`), so below the + watch root a link or junction put where ``.processed`` was is refused, not swept through. + + Re-reads ``deadline_hit`` before each file, so a large directory releases the pass between + files. Never raises for a filesystem fault: what it cannot read or remove stays for the next + pass, and how many it could not remove is logged. Returns how many went, or None when the + sweep was refused or could not open the directory, so a connection sharing the directory + may still sweep it.""" + if not self.archiving: + return 0 + # This connection's own directories, and every guarded directory the pass could not read, + # read again here, under this connection's identity: an alternate credential may reach a + # share the engine's own account cannot. One it still cannot read refuses the sweep. + # Without an alternate credential this runs as the pass did, so a re-read cannot succeed: + # refuse without stat'ing a dead share once per archive. + own = build_archive_guard( + (self.directory, self.error_dir) + if self._cred_ctx is None + else (self.directory, self.error_dir, *protected.unresolved) + ) + unreadable = own.unresolved | (protected.unresolved if self._cred_ctx is None else set()) + if unreadable: + logger.warning( + "file source %s: the [retention].processed_archive_days sweep skips this connection " + "this pass, because %d directory(ies) it must not reach exist, or may, but their " + "identity could not be read under this connection's identity, so an alias of one " + "would not be recognised", + self._label, + len(unreadable), + ) + return None + guard = ArchiveGuard( + protected.keys | own.keys, protected.ids | own.ids, trees=protected.trees + ) + try: + mine = archive_dir_key(self.processed_dir) + except (OSError, ValueError): + mine = os.path.normcase(os.path.abspath(self.processed_dir)).casefold() + if _archive_overlaps(mine, guard.keys) or _archive_inside(mine, guard.trees): + self._log_prune_refusal() + return None + try: + dest = _open_dest(self.processed_dir, self.directory, self._root_real) + except FileNotFoundError: + return 0 # nothing archived yet + except OSError as exc: + logger.warning( + "file source %s: the [retention].processed_archive_days sweep could not open the " + "archive directory %s (%s); nothing was deleted this pass", + self._label, + self.processed_dir.name, + _describe_os_error(exc), + ) + return None + with dest: + opened = _opened_id(dest) + if opened is None or not opened[1]: + # Cannot tell what was opened, so sweep nothing: an alias of a guarded directory + # would otherwise pass, since a guarded identity of inode 0 is never recorded. + logger.warning( + "file source %s: the [retention].processed_archive_days sweep skips this " + "connection this pass, because the identity of its archive directory %s could " + "not be read, so it cannot be checked against the directories the sweep " + "must not reach", + self._label, + self.processed_dir.name, + ) + return None + if opened in guard.ids: + self._log_prune_refusal() + return None + deleted, failed = _prune_entries(dest, older_than, deadline_hit) + if failed: + logger.warning( + "file source %s: the [retention].processed_archive_days sweep could not remove %d " + "aged file(s) in %s this pass; they stay until a later pass removes them", + self._label, + failed, + self.processed_dir.name, + ) + return deleted + + def _log_prune_refusal(self) -> None: + # Every pass, not once: while it holds, this connection's archive has no age bound at all. + logger.warning( + "file source %s: the [retention].processed_archive_days sweep skips this connection, " + "because its archive directory is, or contains, a File inbound's watch or error " + "directory, a File outbound's target directory or one of the engine's own " + "directories; set processed_subdir to a directory of its own", + self._label, + ) + + +def _archive_inside(archive: str, trees: frozenset[str]) -> bool: + """Whether the archive directory key ``archive`` is inside any engine-owned tree key in + ``trees`` (both from :func:`archive_dir_key`).""" + for tree in trees: + try: + if os.path.commonpath([archive, tree]) == tree: + return True + except ValueError: + continue # another Windows drive + return False + + +def _archive_overlaps(archive: str, protected: frozenset[str]) -> bool: + """Whether the archive directory key ``archive`` is, or holds, any key in ``protected`` (both from + :func:`archive_dir_key`).""" + for other in protected: + try: + if os.path.commonpath([archive, other]) == archive: + return True + except ValueError: + continue # another Windows drive, so neither holds the other + return False + + +def _opened_id(dest: _Dest) -> _FileId | None: + """The identity of the archive directory actually opened, to check against + :attr:`ArchiveGuard.ids`. The path compare in :func:`_archive_overlaps` resolves links, while + :func:`_open_dest` walks a normalised path below the watch root, so the two can reach different + directories, and two paths can name one directory. None when the open directory cannot be + stat'ed; the caller then sweeps nothing, as it does for an inode of 0.""" + try: + return _file_id(os.fstat(dest.fd) if dest.fd is not None else os.stat(dest.path)) + except OSError: + return None + + +def _prune_entries( + dest: _Dest, older_than: float, deadline_hit: Callable[[], bool] +) -> tuple[int, int]: + """Remove each regular file directly in ``dest`` archived before ``older_than`` + (:func:`_archived_at`). Returns how many went and how many could not be removed.""" + deleted = failed = 0 + try: + listing = os.scandir(dest.fd if dest.fd is not None else dest.path) + except OSError as exc: + logger.warning( + "the [retention].processed_archive_days sweep could not list %s (%s)", + dest.path.name, + _describe_os_error(exc), + ) + return 0, 0 + with listing: + entries = iter(listing) + while not deadline_hit(): + try: + entry = next(entries) + except StopIteration: + break + except OSError as exc: + # The listing itself failed part way, a share dropping say: keep the count so far. + logger.warning( + "the [retention].processed_archive_days sweep stopped listing %s part way (%s); " + "the rest waits for the next pass", + dest.path.name, + _describe_os_error(exc), + ) + break + try: + if not entry.is_file(follow_symlinks=False): + continue + if dest.fd is not None: + listed = entry.stat(follow_symlinks=False) + if _archived_at(listed) >= older_than: + continue + gone = _unlink_if_unchanged(entry.name, dest.fd, listed) + else: + # The listing's own mtime first, so a young copy is never opened at all. + if entry.stat(follow_symlinks=False).st_mtime >= older_than: + continue + gone = _win_delete_aged(dest.path / entry.name, older_than) + except OSError: + failed += 1 # locked, refused or vanished: it stays, and the next pass tries again + continue + if gone: + deleted += 1 + return deleted, failed + + +def _archived_at(st: os.stat_result) -> float: + """POSIX: when a copy was archived, as the later of its mtime and ctime. The archive move links or + writes the copy, which sets ctime, and nothing can set ctime back, so a drop whose old mtime the + stamp could not replace (the partner owns it and the engine may not write it) still ages from its + archive.""" + return max(st.st_mtime, st.st_ctime) + + +def _unlink_if_unchanged(name: str, dir_fd: int, listed: os.stat_result) -> bool: + """POSIX: unlink ``name`` relative to ``dir_fd`` only if it is still the regular file listed, with + the same mtime and ctime. False when it changed, so it is kept. As :func:`_pin_confined` says, POSIX has no + call that unlinks a name only if it is a given file, so a file renamed over the name between this + check and the unlink, two system calls wide, is the one removed; renaming there needs the same + permission on the directory as removing it.""" + now = os.stat(name, dir_fd=dir_fd, follow_symlinks=False) + if ( + not stat.S_ISREG(now.st_mode) + or _file_id(now) != _file_id(listed) + or now.st_mtime_ns != listed.st_mtime_ns + or now.st_ctime_ns != listed.st_ctime_ns + ): + return False + os.unlink(name, dir_fd=dir_fd) + return True + + +def _win_delete_aged(path: Path, older_than: float) -> bool: + """Windows: open ``path`` as itself (:func:`_win_open_file`, refusing a link or a junction), check + on that handle that it is a regular file archived before ``older_than``, and delete that handle, + read-only attribute or not. Deciding and deleting on one handle means a file swapped in under the + name, or a name Windows normalises to another file, is judged as itself. False when it is kept.""" + fd = _win_open_file(path, _FILE_READ_ATTRIBUTES | _DELETE) + try: + st = os.fstat(fd) + if not stat.S_ISREG(st.st_mode) or st.st_mtime >= older_than: + return False + _delete_by_handle(fd, ignore_readonly=True) + finally: + os.close(fd) + return True + # --- helpers ----------------------------------------------------------------- # The pure magic-byte sniffers (_looks_like_hl7 / _lstrip_bom_ws / _content_matches_declared) were hoisted @@ -2180,10 +2529,14 @@ def _names_another_path(st: os.stat_result) -> bool: #: POSIX ``os.link`` follows a link at its source unless told not to. Windows ``CreateHardLinkW`` links #: the link itself already, and Windows ``os.link`` does not list the flag. _LINK_TAKES_NOFOLLOW = os.link in os.supports_follow_symlinks +#: The archive-time stamp (BACKLOG #1188) sets a time relative to the checked directory, never +#: following a link at the last name. +_UTIME_AT = os.utime in os.supports_dir_fd and os.utime in os.supports_follow_symlinks # Win32 values for the Windows walk and the handle-based move (BACKLOG #2535), from the Windows SDK. _GENERIC_READ = 0x80000000 _DELETE = 0x00010000 +_FILE_READ_ATTRIBUTES = 0x0080 _FILE_LIST_DIRECTORY = 0x0001 _SHARE_ALL = 0x7 # FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE #: Held on each directory the walk passes, so nobody can rename that directory until the file is open. @@ -2199,6 +2552,7 @@ def _names_another_path(st: os.stat_result) -> bool: _FILE_ATTRIBUTE_TAG_INFO = 9 _FILE_DISPOSITION_INFO_EX = 21 _DISPOSITION_DELETE_POSIX = 0x1 | 0x2 # FILE_DISPOSITION_FLAG_DELETE | ..._POSIX_SEMANTICS +_DISPOSITION_IGNORE_READONLY = 0x10 # FILE_DISPOSITION_FLAG_IGNORE_READONLY_ATTRIBUTE #: ERROR_INVALID_FUNCTION, ERROR_NOT_SUPPORTED, ERROR_INVALID_PARAMETER: "this volume does not do #: that", not "that failed". They send a POSIX-semantics delete to the classic one. _WIN_UNSUPPORTED = frozenset({1, 50, 87}) @@ -2636,20 +2990,65 @@ class _RenameInfo(ctypes.Structure): raise _win_error() -def _delete_by_handle(fd: int) -> None: +#: ``FILE_BASIC_INFO``'s class id, and the access ``ReOpenFile`` asks for to write it. +_FILE_BASIC_INFO = 0 +_FILE_WRITE_ATTRIBUTES = 0x0100 +#: 100-nanosecond intervals between 1601-01-01 (the FILETIME epoch) and 1970-01-01. +_FILETIME_EPOCH_OFFSET = 116_444_736_000_000_000 + + +class _BasicInfo(ctypes.Structure): + """``FILE_BASIC_INFO``. A zero time leaves that time as it is.""" + + _fields_ = ( + ("creation_time", ctypes.c_int64), + ("last_access_time", ctypes.c_int64), + ("last_write_time", ctypes.c_int64), + ("change_time", ctypes.c_int64), + ("file_attributes", ctypes.c_uint32), + ) + + +def _win_stamp_write_time(fd: int, when_ns: int) -> None: + """Windows: set the last-write time of the file open on ``fd`` to ``when_ns`` (epoch ns). + + The checked handle was opened for read and delete only, so it is reopened with + ``FILE_WRITE_ATTRIBUTES`` by ``ReOpenFile``, which names no path: the time lands on exactly the + file checked (BACKLOG #1188). Only the last-write time moves; zeros keep the others.""" + k32 = _kernel32() + handle = k32.ReOpenFile(_os_handle(fd), _FILE_WRITE_ATTRIBUTES, _SHARE_ALL, 0) + if handle is None or handle == _INVALID_HANDLE: + raise _win_error() + try: + info = _BasicInfo(0, 0, when_ns // 100 + _FILETIME_EPOCH_OFFSET, 0, 0) + if not k32.SetFileInformationByHandle( + handle, _FILE_BASIC_INFO, ctypes.byref(info), ctypes.sizeof(info) + ): + raise _win_error() + finally: + k32.CloseHandle(handle) + + +def _delete_by_handle(fd: int, *, ignore_readonly: bool = False) -> None: """Windows: delete the file open on ``fd``. POSIX semantics first, so the name goes at once even while another process holds the file open; the classic disposition where the volume has no such - thing, which removes the name when the last handle closes.""" + thing, which removes the name when the last handle closes. ``ignore_readonly`` deletes a file + with the read-only attribute too, as a POSIX unlink would (the ``.processed`` sweep, BACKLOG + #1188); the classic disposition cannot, and refuses such a file.""" k32 = _kernel32() handle = _os_handle(fd) - flags = ctypes.c_uint32(_DISPOSITION_DELETE_POSIX) - if k32.SetFileInformationByHandle( - handle, _FILE_DISPOSITION_INFO_EX, ctypes.byref(flags), ctypes.sizeof(flags) - ): - return - error = _win_error() - if getattr(error, "winerror", None) not in _WIN_UNSUPPORTED: - raise error + # A Windows build before the read-only flag existed refuses the whole request as unsupported, so + # retry without it before giving up the POSIX-semantics delete. A read-only file is then refused. + tries = [_DISPOSITION_DELETE_POSIX | _DISPOSITION_IGNORE_READONLY] if ignore_readonly else [] + for value in (*tries, _DISPOSITION_DELETE_POSIX): + flags = ctypes.c_uint32(value) + if k32.SetFileInformationByHandle( + handle, _FILE_DISPOSITION_INFO_EX, ctypes.byref(flags), ctypes.sizeof(flags) + ): + return + error = _win_error() + if getattr(error, "winerror", None) not in _WIN_UNSUPPORTED: + raise error delete = ctypes.c_ubyte(1) # FILE_DISPOSITION_INFO.DeleteFile if not k32.SetFileInformationByHandle( handle, _FILE_DISPOSITION_INFO, ctypes.byref(delete), ctypes.sizeof(delete) @@ -2706,6 +3105,8 @@ def _kernel32() -> Any: k32.CreateFileW.restype = wintypes.HANDLE k32.CloseHandle.argtypes = (wintypes.HANDLE,) k32.CloseHandle.restype = wintypes.BOOL + k32.ReOpenFile.argtypes = (wintypes.HANDLE, wintypes.DWORD, wintypes.DWORD, wintypes.DWORD) + k32.ReOpenFile.restype = wintypes.HANDLE for info_call in (k32.GetFileInformationByHandleEx, k32.SetFileInformationByHandle): info_call.argtypes = (wintypes.HANDLE, ctypes.c_int, wintypes.LPVOID, wintypes.DWORD) info_call.restype = wintypes.BOOL diff --git a/tests/test_api_tls.py b/tests/test_api_tls.py index eb5b7d8ba..94607996b 100644 --- a/tests/test_api_tls.py +++ b/tests/test_api_tls.py @@ -1784,6 +1784,7 @@ def test_serve_loopback_emits_no_new_stderr( "security.delete_message_bodies_after_days = 30\n" "retention.dead_letter_days = 30\n" "retention.reference_snapshot_days = 30\n" + "retention.processed_archive_days = 30\n" # The two PL-2 tiers the engine deliberately does NOT default, because each keys on a # timestamp that only moves on a write, so a silent default would delete data still in use. # They warn when unset, which on this test's own terms is a line on a stream it asserts is diff --git a/tests/test_file_path_join_census.py b/tests/test_file_path_join_census.py index 3f96cca6f..48591b735 100644 --- a/tests/test_file_path_join_census.py +++ b/tests/test_file_path_join_census.py @@ -770,6 +770,15 @@ def derived_script_sites() -> dict[str, int]: "messagefoundry/transports/file.py::FileSource.__init__": Site( 2, OPERATOR, "The operator's processed and error subdirectory settings." ), + "messagefoundry/transports/file.py::FileSource.guarded_dirs": Site( + 1, OPERATOR, "The operator's error subdirectory setting, as `__init__` joins it." + ), + "messagefoundry/transports/file.py::_prune_entries": Site( + 1, + REVIEWED, + "A bare name from listing the archive directory the sweep opened through no link, joined " + "into that held directory; the Windows delete then opens it as itself, refusing a link.", + ), "messagefoundry/transports/file.py::_archive": Site( 1, GUARDED, diff --git a/tests/test_file_source_bounded_stop.py b/tests/test_file_source_bounded_stop.py index 84db85516..46b643c4b 100644 --- a/tests/test_file_source_bounded_stop.py +++ b/tests/test_file_source_bounded_stop.py @@ -96,11 +96,11 @@ async def test_stop_never_cancels_a_hand_off_in_progress( source = _source(inbox) real_move = source._move - def slow_move(path: Path, dest_dir: Path, *rest: Any) -> bool: + def slow_move(path: Path, dest_dir: Path, *rest: Any, **kw: Any) -> bool: # Longer than the 0.25 s window before stop() next looks, so a stop() that skips the grace # restart cancels the poll task mid-move. time.sleep(0.5) - return real_move(path, dest_dir, *rest) + return real_move(path, dest_dir, *rest, **kw) monkeypatch.setattr(source, "_move", slow_move) in_handler = asyncio.Event() diff --git a/tests/test_phi_at_rest_inventory.py b/tests/test_phi_at_rest_inventory.py index 1d0f3bf66..bec5c23d6 100644 --- a/tests/test_phi_at_rest_inventory.py +++ b/tests/test_phi_at_rest_inventory.py @@ -53,10 +53,12 @@ import messagefoundry.store.sqlserver as store_sqlserver import messagefoundry.store.store as store_store from messagefoundry.config.settings import ServiceSettings +from messagefoundry.pipeline.retention import RetentionRunner from messagefoundry.store.base import Store from messagefoundry.store.postgres import PostgresStore from messagefoundry.store.sqlserver import SqlServerStore from messagefoundry.store.store import MessageStore +from messagefoundry.transports.file import FileSource from tests._ast_sites import ( call_sites, code_strings, @@ -1288,10 +1290,21 @@ def _purge_table_cells(section: str) -> dict[str, dict[str, str]]: return out -#: §8 rows whose operation is NOT a Store method: the app-log sweep is filesystem work owned by the -#: RetentionRunner itself, identical on every backend. Reviewed and pinned below, so it cannot be -#: used as a hiding place for a store operation that quietly stopped being implemented. -_NON_STORE_OPERATIONS = frozenset({"app_log_days"}) +#: §8 rows whose operation is NOT a Store method: filesystem work the RetentionRunner drives, +#: identical on every backend. The app-log sweep is the runner's own; the `.processed` sweep is run +#: by the runner and done by each File source. Reviewed and pinned below, each to the code that does +#: the work, so it cannot be used as a hiding place for an operation that quietly stopped existing. +_NON_STORE_OPERATIONS = frozenset({"app_log_days", "processed_archive_days"}) + +#: For each exempt operation, (class, method) pairs that must still exist, read off the imported +#: classes: a setting name alone survives in a log string after the code that honours it is gone. +_NON_STORE_IMPLEMENTATIONS: dict[str, tuple[tuple[type, str], ...]] = { + "app_log_days": (), + "processed_archive_days": ( + (RetentionRunner, "_prune_processed_archives"), + (FileSource, "prune_processed_archive"), + ), +} def _method_shape(cls: type, name: str) -> str: @@ -1344,6 +1357,10 @@ def test_per_backend_purge_verdicts_are_bound_to_the_method_bodies() -> None: f"{operation} is exempted as RetentionRunner-owned but no longer appears in " "pipeline/retention.py — the §8 row describes nothing." ) + for owner, method in _NON_STORE_IMPLEMENTATIONS[operation]: + assert callable(owner.__dict__.get(method)), ( + f"{operation}: {owner.__name__}.{method} is gone, so the §8 row describes nothing." + ) def _verdict_mismatches(table: dict[str, dict[str, str]], backends: dict[str, type]) -> list[str]: diff --git a/tests/test_processed_archive_retention.py b/tests/test_processed_archive_retention.py new file mode 100644 index 000000000..6dc7cddf6 --- /dev/null +++ b/tests/test_processed_archive_retention.py @@ -0,0 +1,971 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +# Copyright (C) 2026 MessageFoundry Foundation, LLC and contributors +"""The File source's `.processed` archive has a default-on age bound (BACKLOG #1188). + +Owner ruling A2 of 2026-10-08: bound `.processed` by default, as a fourth auto-bound window at the +body window, keyed on archive time. `.error` deletion was not ruled on and is not swept. + +What these tests hold the sweep to: it deletes only aged regular files directly in the archive +directory, keeps young ones, never touches a file outside it or reaches through a link, refuses an +archive directory that overlaps a watch or `.error` directory, and keys on the time the File source +archived the copy rather than on the drop's own mtime. +""" + +from __future__ import annotations + +import errno +import json +import os +import stat +import time +from collections.abc import AsyncIterator, Callable, Iterable +from pathlib import Path + +import pytest + +from messagefoundry.config.models import ConnectorType, Source +from messagefoundry.config.retention_classification import PHI_RETENTION_WINDOWS +from messagefoundry.config.settings import ( + ApiSettings, + BackupSettings, + EgressSettings, + LoggingSettings, + RetentionSettings, + StoreSettings, + engine_owned_dirs, +) +from messagefoundry.config.wiring import ( + ConnectionSpec, + InboundConnection, + OutboundConnection, + Registry, + env, +) +from messagefoundry.pipeline import wiring_runner +from messagefoundry.pipeline.retention import RetentionRunner +from messagefoundry.pipeline.sharding import filter_registry_for_shard +from messagefoundry.pipeline.wiring_runner import RegistryRunner +from messagefoundry.store import MessageStore +from messagefoundry.transports import base as transport_base +from messagefoundry.transports import file as file_transport +from messagefoundry.transports.base import ( + ArchiveGuard, + ProcessedArchive, + build_archive_guard, + guarded_source_dirs, +) +from messagefoundry.transports.file import FileSource + +DAY = 86_400.0 +#: The sweep's cutoff in these tests: anything archived more than 30 days before "now" goes. +_NOW = time.time() +_CUTOFF = _NOW - 30 * DAY + + +def _source(inbox: Path, **settings: object) -> FileSource: + return FileSource( + Source(type=ConnectorType.FILE, settings={"directory": str(inbox), **settings}) + ) + + +def _aged(path: Path, days: float, body: str = "MSH|^~\\&|SYNTHETIC") -> Path: + """Write ``path`` with an mtime ``days`` before now.""" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(body, encoding="ascii") + when = _NOW - days * DAY + os.utime(path, (when, when)) + return path + + +def _never() -> bool: + return False + + +def _prune(src: FileSource, protected: frozenset[Path] = frozenset()) -> int | None: + return src._prune_processed(_CUTOFF, build_archive_guard(protected), _never) + + +@pytest.fixture(autouse=True) +def _mtime_keyed(monkeypatch: pytest.MonkeyPatch) -> None: + """POSIX keys the sweep on the later of mtime and ctime, and a test cannot make an old ctime, so + these tests age a file by its mtime alone. The ctime test undoes this.""" + monkeypatch.setattr(file_transport, "_archived_at", lambda st: st.st_mtime) + + +def test_the_window_is_auto_bounded_at_the_body_window() -> None: + """The ruling's 'at the body window': the default bound equals the body window's. + + Mutation: give the new window `auto_bound_days=90`. Red.""" + by_setting = {w.setting: w for w in PHI_RETENTION_WINDOWS} + body = by_setting["[security].delete_message_bodies_after_days"] + processed = by_setting["[retention].processed_archive_days"] + assert processed.auto_bound_days is not None + assert processed.auto_bound_days == body.auto_bound_days + assert processed.level == "PL-1" + assert RetentionSettings().processed_archive_days == 0 # unset; the gate supplies the bound + + +def test_a_negative_window_is_refused() -> None: + with pytest.raises(ValueError, match=">= 0"): + RetentionSettings(processed_archive_days=-1) + + +def test_the_file_source_is_a_processed_archive(tmp_path: Path) -> None: + assert isinstance(_source(tmp_path / "in"), ProcessedArchive) + + +def test_aged_copies_go_and_young_ones_stay(tmp_path: Path) -> None: + inbox = tmp_path / "in" + src = _source(inbox) + old = _aged(src.processed_dir / "old.hl7", 31) + young = _aged(src.processed_dir / "young.hl7", 29) + + assert _prune(src) == 1 + assert not old.exists() + assert young.exists() + + +def test_nothing_outside_the_archive_directory_is_touched(tmp_path: Path) -> None: + """An undelivered drop, a `.error` copy and a file in a subdirectory are all old, and all stay. + + Mutation: make the sweep recursive, or point it at `src.directory`. Red.""" + inbox = tmp_path / "in" + src = _source(inbox) + drop = _aged(inbox / "waiting.hl7", 90) + quarantined = _aged(src.error_dir / "bad.hl7", 90) + nested = _aged(src.processed_dir / "sub" / "nested.hl7", 90) + archived = _aged(src.processed_dir / "done.hl7", 90) + + assert _prune(src) == 1 + assert not archived.exists() + assert drop.exists() and quarantined.exists() and nested.exists() + + +def test_a_link_in_the_archive_is_never_followed(tmp_path: Path) -> None: + """A link put in `.processed` that names an old file outside it: the outside file stays. + + Mutation: test `entry.is_file()` with links followed and unlink the resolved target. Red.""" + inbox = tmp_path / "in" + src = _source(inbox) + outside = _aged(tmp_path / "elsewhere" / "keep.hl7", 90) + src.processed_dir.mkdir(parents=True) + link = src.processed_dir / "link.hl7" + try: + link.symlink_to(outside) + except OSError as exc: + pytest.skip(f"cannot create a symbolic link here: {exc}") + + assert _prune(src) == 0 + assert outside.exists() + + +def test_a_linked_archive_directory_below_the_watch_root_is_refused(tmp_path: Path) -> None: + """`.processed` replaced by a link to another directory: nothing there is deleted.""" + inbox = tmp_path / "in" + inbox.mkdir() + elsewhere = tmp_path / "elsewhere" + outside = _aged(elsewhere / "keep.hl7", 90) + try: + (inbox / ".processed").symlink_to(elsewhere, target_is_directory=True) + except OSError as exc: + pytest.skip(f"cannot create a symbolic link here: {exc}") + src = _source(inbox) + + assert _prune(src) is None + assert outside.exists() + + +@pytest.mark.parametrize("subdir", [".", ".."]) +def test_an_archive_directory_holding_the_watch_directory_is_refused( + tmp_path: Path, subdir: str, caplog: pytest.LogCaptureFixture +) -> None: + """`processed_subdir = "."` would make every undelivered drop an 'archive'. Refused, and + warned about on every pass, since while it holds the archive has no age bound at all. + + Mutation: drop the overlap guard. Red: the waiting drop is deleted.""" + inbox = tmp_path / "in" + src = _source(inbox, processed_subdir=subdir) + drop = _aged(inbox / "waiting.hl7", 90) + + assert _prune(src) is None + assert _prune(src) is None + assert drop.exists() + refusals = [r for r in caplog.records if "skips this connection" in r.getMessage()] + assert len(refusals) == 2 + assert "waiting.hl7" not in caplog.text + + +def test_an_archive_directory_shared_with_error_is_refused(tmp_path: Path) -> None: + """`.error` deletion is unruled (#3335), so a `.processed` that IS `.error` deletes nothing.""" + inbox = tmp_path / "in" + src = _source(inbox, processed_subdir="quarantine", error_subdir="quarantine") + quarantined = _aged(src.error_dir / "bad.hl7", 90) + + assert _prune(src) is None + assert quarantined.exists() + + +@pytest.mark.parametrize("nested", [False, True]) +def test_another_connections_watch_directory_is_protected(tmp_path: Path, nested: bool) -> None: + """A's archive directory is, or holds, B's watch directory: A's sweep deletes nothing, not + even its own old copy.""" + shared = tmp_path / "shared" + b_inbox = shared / "b-inbox" if nested else shared + a = _source(tmp_path / "a", processed_subdir=str(shared)) + b_drop = _aged(b_inbox / "waiting.hl7", 90) + a_copy = _aged(a.processed_dir / "done.hl7", 90) + + assert _prune(a, frozenset({b_inbox})) is None + assert b_drop.exists() and a_copy.exists() + + +def test_the_overlap_guard_folds_case(tmp_path: Path) -> None: + """B's watch directory named in another case still protects it: a case-insensitive volume names + one directory both ways, and folding on a case-sensitive one only refuses more. + + Mutation: drop `.casefold()` from `archive_dir_key`. Red on a case-sensitive volume.""" + shared = tmp_path / "Shared" + a = _source(tmp_path / "a", processed_subdir=str(shared)) + a_copy = _aged(a.processed_dir / "done.hl7", 90) + + assert _prune(a, frozenset({tmp_path / "shared"})) is None + assert a_copy.exists() + + +@pytest.mark.parametrize("after_read", ["leave", "delete"]) +def test_a_source_that_does_not_move_sweeps_nothing(tmp_path: Path, after_read: str) -> None: + """Only a move archives. A leave source's `.processed` may hold another system's files. + + Mutation: drop the `after_read != "move"` return. Red.""" + src = _source(tmp_path / "in", after_read=after_read) + theirs = _aged(src.processed_dir / "theirs.hl7", 90) + + assert _prune(src) == 0 + assert theirs.exists() + + +def test_a_read_only_aged_copy_is_removed(tmp_path: Path) -> None: + """Windows refuses a plain delete of a file with the read-only attribute, where a POSIX unlink + ignores the mode. The sweep must bound both alike. + + Mutation: drop `ignore_readonly=True` from `_win_delete_aged`. Red on Windows.""" + src = _source(tmp_path / "in") + old = _aged(src.processed_dir / "old.hl7", 90) + os.chmod(old, stat.S_IREAD) + try: + assert _prune(src) == 1 + assert not old.exists() + finally: + if old.exists(): + os.chmod(old, stat.S_IREAD | stat.S_IWRITE) + + +async def test_a_stopping_source_stops_its_sweep(tmp_path: Path) -> None: + """A source that is stopping stops its sweep between files, deadline or not.""" + src = _source(tmp_path / "in") + old = _aged(src.processed_dir / "old.hl7", 90) + src._stop.set() + + assert ( + await src.prune_processed_archive(_CUTOFF, protected=ArchiveGuard(), deadline_hit=_never) + == 0 + ) + assert old.exists() + + +def test_the_file_source_declares_its_guarded_directories(tmp_path: Path) -> None: + """From configuration alone, so a stopped or other-shard inbound is protected too.""" + inbox = tmp_path / "in" + config = Source( + type=ConnectorType.FILE, settings={"directory": str(inbox), "error_subdir": "bad"} + ) + + assert guarded_source_dirs(config) == (inbox, inbox / "bad") + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX keys on ctime; Windows stamps by handle") +def test_posix_ages_a_copy_from_its_ctime(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A copy whose old mtime the stamp could not replace still ages from its archive: the move set + its ctime, which nobody can set back. + + Mutation: make `_archived_at` return `st.st_mtime`. Red.""" + monkeypatch.undo() + src = _source(tmp_path / "in") + copy = _aged(src.processed_dir / "kept-mtime.hl7", 90) # ctime is now + + assert _prune(src) == 0 + assert copy.exists() + + +def test_an_aliased_protected_directory_is_refused_once_opened(tmp_path: Path) -> None: + """The guard names B's watch directory by another path that the path compare cannot join (here a + link). The opened archive is recognised by identity. + + Mutation: drop the `_opened_id(dest) in guard.ids` check. Red.""" + real = tmp_path / "shared" + real.mkdir() + alias = tmp_path / "alias" + try: + alias.symlink_to(real, target_is_directory=True) + except OSError as exc: + pytest.skip(f"cannot create a symbolic link here: {exc}") + a = _source(tmp_path / "a", processed_subdir=str(real)) + a_copy = _aged(real / "done.hl7", 90) + guard = ArchiveGuard(keys=frozenset({"unrelated"}), ids=build_archive_guard({alias}).ids) + + assert a._prune_processed(_CUTOFF, guard, _never) is None + assert a_copy.exists() + + +@pytest.mark.parametrize("opened", [None, (7, 0)]) +def test_an_archive_directory_with_no_readable_identity_is_not_swept( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, opened: tuple[int, int] | None +) -> None: + """The opened directory's identity cannot be read, or is inode 0, which the guard never records. + Either way an alias of a guarded directory would pass the identity check, so nothing goes. + + Mutation: test only `_opened_id(dest) in guard.ids`. Red.""" + src = _source(tmp_path / "in") + copy = _aged(src.processed_dir / "done.hl7", 90) + monkeypatch.setattr(file_transport, "_opened_id", lambda dest: opened) + + assert _prune(src) is None + assert copy.exists() + + +def test_archived_at_is_the_later_of_mtime_and_ctime(monkeypatch: pytest.MonkeyPatch) -> None: + """The real key, which the autouse fixture replaces everywhere else. + + Mutation: return `st.st_mtime`, or `min(...)`. Red.""" + monkeypatch.undo() + st = os.stat_result((0o100644, 1, 1, 1, 0, 0, 0, 0, 100, 200)) # mtime 100, ctime 200 + + assert file_transport._archived_at(st) == 200 + assert ( + file_transport._archived_at(os.stat_result((0o100644, 1, 1, 1, 0, 0, 0, 0, 300, 200))) + == 300 + ) + + +def test_the_registry_runner_guards_an_unbound_file_inbound(tmp_path: Path) -> None: + """`archive_guard_dirs` reads the configuration, so an inbound that never started is named.""" + inbox = tmp_path / "stopped" + reg = Registry() + reg.add_inbound( + InboundConnection( + "IB_STOPPED", ConnectionSpec(ConnectorType.FILE, {"directory": str(inbox)}), router="r" + ) + ) + runner = RegistryRunner(reg, None, egress=EgressSettings(deny_by_default=False)) # type: ignore[arg-type] + + assert runner.processed_archives() == () + assert runner.archive_guard_dirs() == (inbox, inbox / ".error") + + +def _runner(reg: Registry) -> RegistryRunner: + return RegistryRunner(reg, None, egress=EgressSettings(deny_by_default=False)) # type: ignore[arg-type] + + +def _file_inbound(name: str, inbox: Path, shard: str | None = None) -> InboundConnection: + return InboundConnection( + name, + ConnectionSpec(ConnectorType.FILE, {"directory": str(inbox)}), + router="r", + shard=shard, + ) + + +def _file_outbound(name: str, out: Path) -> OutboundConnection: + return OutboundConnection(name, ConnectionSpec(ConnectorType.FILE, {"directory": str(out)})) + + +def _refuse_stat_of(monkeypatch: pytest.MonkeyPatch, refused: Path) -> None: + """Make `stat` of ``refused`` fail as an unreadable directory would, and nothing else.""" + real_stat = os.stat + target = Path(os.path.realpath(refused)) + + def refusing(path: object, *args: object, **kwargs: object) -> os.stat_result: + if isinstance(path, (str, os.PathLike)) and Path(path) == target: + raise PermissionError(errno.EACCES, "refused") + return real_stat(path, *args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(os, "stat", refusing) + + +def test_the_guard_names_a_sibling_engine_shards_inbound(tmp_path: Path) -> None: + """Under `serve --shard` the runner's registry holds only this shard's inbounds; the guard still + names the sibling shard's watch and `.error` directories, from the whole config's specs. + + Mutation: read `self.registry.inbound` in `archive_guard_dirs`, or drop + `all_inbound_guard_settings` + from the shard filter. Red.""" + reg = Registry() + reg.add_inbound(_file_inbound("IB_MINE", tmp_path / "mine", shard="s1")) + reg.add_inbound(_file_inbound("IB_SIBLING", tmp_path / "sibling", shard="s2")) + s1 = filter_registry_for_shard(reg, "s1") + assert set(s1.inbound) == {"IB_MINE"} # the precondition the guard must not rely on + + dirs = _runner(s1).archive_guard_dirs() + + assert dirs is not None + assert tmp_path / "sibling" in dirs and tmp_path / "sibling" / ".error" in dirs + assert tmp_path / "mine" in dirs + + +def test_the_shard_pin_carries_only_directory_settings(tmp_path: Path) -> None: + """The pinned sibling-shard view holds the directory settings, never a credential setting. + + Mutation: pin each inbound's whole settings in `inbound_guard_settings`. Red.""" + reg = Registry() + reg.add_inbound(_file_inbound("IB_MINE", tmp_path / "mine", shard="s1")) + reg.add_inbound( + InboundConnection( + "IB_SIBLING", + ConnectionSpec( + ConnectorType.FILE, + {"directory": str(tmp_path / "sib"), "credential_password": env("NOT_SET_HERE")}, + ), + router="r", + shard="s2", + ) + ) + + pinned = filter_registry_for_shard(reg, "s1").all_inbound_guard_settings + + assert pinned is not None + assert pinned["IB_SIBLING"] == (ConnectorType.FILE, {"directory": str(tmp_path / "sib")}) + + +def test_the_guard_names_every_file_outbound_target_directory(tmp_path: Path) -> None: + """A File outbound's target directory holds deliveries awaiting a partner's pickup. + + Mutation: drop the outbound loop in `archive_guard_dirs`. Red.""" + reg = Registry() + reg.add_outbound(_file_outbound("OB_PARTNER", tmp_path / "out")) + + assert _runner(reg).archive_guard_dirs() == (tmp_path / "out",) + + +def test_an_unresolvable_outbound_directory_fails_closed( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """An outbound whose settings cannot be resolved makes the guard None, as an inbound's does.""" + reg = Registry() + reg.add_outbound(_file_outbound("OB_PARTNER", tmp_path / "out")) + runner = _runner(reg) + + def unresolved(*args: object, **kwargs: object) -> object: + raise ValueError("unresolved") + + monkeypatch.setattr(wiring_runner, "resolve_env_settings", unresolved) + + assert runner.archive_guard_dirs() is None + + +def test_a_guarded_directory_whose_identity_cannot_be_read_is_recorded( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A guarded directory that exists but cannot be stat'ed adds no identity, so an alias of it + would pass the check once the archive is open. The guard records it as unresolved. + + Mutation: `continue` on every stat failure in `build_archive_guard`. Red.""" + guarded = tmp_path / "share" + guarded.mkdir() + _refuse_stat_of(monkeypatch, guarded) + + assert build_archive_guard({guarded}).unresolved == {guarded} + + +def _as_credentialed(src: FileSource, monkeypatch: pytest.MonkeyPatch) -> None: + """Mark ``src`` as running under an alternate credential, so it re-reads unresolved directories. + The blocking half under test runs on the calling thread either way.""" + monkeypatch.setattr(src, "_cred_ctx", object()) + + +@pytest.mark.parametrize("credentialed", [False, True]) +def test_an_archive_that_cannot_read_an_unresolved_directory_sweeps_nothing( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, credentialed: bool +) -> None: + """Without a credential the archive refuses outright; with one it reads the unresolved directory + again under its own identity and still cannot. + + Mutation: drop the `unreadable` refusal in `_prune_processed`. Red.""" + src = _source(tmp_path / "in") + if credentialed: + _as_credentialed(src, monkeypatch) + copy = _aged(src.processed_dir / "done.hl7", 90) + other = tmp_path / "share" + other.mkdir() + _refuse_stat_of(monkeypatch, other) + + assert ( + src._prune_processed(_CUTOFF, ArchiveGuard(unresolved=frozenset({other})), _never) is None + ) + assert copy.exists() + + +def test_an_archive_that_can_read_an_unresolved_directory_sweeps( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A directory the pass could not read, which this archive's alternate credential can reach, + does not stop this archive's sweep: its identity is used.""" + src = _source(tmp_path / "in") + _as_credentialed(src, monkeypatch) + copy = _aged(src.processed_dir / "done.hl7", 90) + other = tmp_path / "share" + other.mkdir() + + assert src._prune_processed(_CUTOFF, ArchiveGuard(unresolved=frozenset({other})), _never) == 1 + assert not copy.exists() + + +def test_an_unreadable_directory_that_is_the_archive_is_refused_once_reread( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The pass could not read B's watch directory; A's archive re-reads it, finds it is its own + archive directory, and refuses (here by the resolved path; identity covers an alias).""" + shared = tmp_path / "shared" + a = _source(tmp_path / "a", processed_subdir=str(shared)) + _as_credentialed(a, monkeypatch) + drop = _aged(shared / "waiting.hl7", 90) + guard = ArchiveGuard(keys=frozenset({"unrelated"}), unresolved=frozenset({shared})) + + assert a._prune_processed(_CUTOFF, guard, _never) is None + assert drop.exists() + + +def test_only_the_directory_settings_are_resolved_for_the_guard(tmp_path: Path) -> None: + """A not-deployed outbound's unrelated env() secret, missing in this environment, is not read and + does not switch the sweep off. + + Mutation: resolve `spec.settings` whole in `archive_guard_dirs`. Red.""" + reg = Registry() + reg.add_outbound( + OutboundConnection( + "OB_PARKED", + ConnectionSpec( + ConnectorType.FILE, + {"directory": str(tmp_path / "out"), "credential_password": env("NOT_SET_HERE")}, + ), + deployed=False, + ) + ) + + assert _runner(reg).archive_guard_dirs() == (tmp_path / "out",) + + +def test_an_unreachable_network_path_is_not_read_as_absent() -> None: + """Windows maps ERROR_BAD_NETPATH (53) to ENOENT; only its own not-found codes count as absent. + + Mutation: read `exc.errno` alone in `_is_absent`. Red.""" + + class _NetPath(OSError): + winerror = 53 + + class _NotFound(OSError): + winerror = 3 + + assert not transport_base._is_absent(_NetPath(errno.ENOENT, "unreachable")) + assert transport_base._is_absent(_NotFound(errno.ENOENT, "absent")) + + +def test_a_guarded_directory_not_created_yet_stands_in_by_its_parent(tmp_path: Path) -> None: + """A `.error` before the first quarantine does not switch the sweep off: its nearest existing + ancestor's identity is recorded instead.""" + guard = build_archive_guard({tmp_path / "in" / ".error"}) + + assert not guard.unresolved + st = os.stat(tmp_path) + assert not st.st_ino or (st.st_dev, st.st_ino) in guard.ids + + +def test_a_guarded_directory_of_inode_zero_is_unresolved( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Windows answers a refused directory's stat from its parent's listing, with inode 0, rather + than raising. That is no identity, so the directory is unresolved, not silently passed. + + Mutation: drop the `not st.st_ino and not found` check in `build_archive_guard`. Red.""" + guarded = tmp_path / "share" + guarded.mkdir() + real_stat = os.stat + target = Path(os.path.realpath(guarded)) + + def listing_answer(path: object, *args: object, **kwargs: object) -> os.stat_result: + st = real_stat(path, *args, **kwargs) # type: ignore[arg-type] + if isinstance(path, (str, os.PathLike)) and Path(path) == target: + return os.stat_result((st.st_mode, 0, 0, 1, 0, 0, 0, 0, 0, 0)) + return st + + monkeypatch.setattr(os, "stat", listing_answer) + + assert build_archive_guard({guarded}).unresolved == {guarded} + + +def test_an_inode_zero_ancestor_seen_first_is_still_unresolved_when_guarded( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A directory first reached as an ancestor, whose stat reports inode 0, is later itself a + guarded directory. It must be unresolved, not passed because it was walked already. + + Mutation: memoise every stat'ed directory in `build_archive_guard`. Red.""" + parent = tmp_path / "shares" + child = parent / "in" + child.mkdir(parents=True) + real_stat = os.stat + target = Path(os.path.realpath(parent)) + + def listing_answer(path: object, *args: object, **kwargs: object) -> os.stat_result: + st = real_stat(path, *args, **kwargs) # type: ignore[arg-type] + if isinstance(path, (str, os.PathLike)) and Path(path) == target: + return os.stat_result((st.st_mode, 0, 0, 1, 0, 0, 0, 0, 0, 0)) + return st + + monkeypatch.setattr(os, "stat", listing_answer) + + assert build_archive_guard([child, parent]).unresolved == {parent} + + +def test_a_missing_archive_directory_is_not_an_error(tmp_path: Path) -> None: + assert _prune(_source(tmp_path / "in")) == 0 + + +def test_the_deadline_stops_the_sweep_between_files(tmp_path: Path) -> None: + src = _source(tmp_path / "in") + for i in range(3): + _aged(src.processed_dir / f"old-{i}.hl7", 90) + + assert src._prune_processed(_CUTOFF, ArchiveGuard(), lambda: True) == 0 + assert len(list(src.processed_dir.iterdir())) == 3 + + +def test_archiving_stamps_the_archive_time(tmp_path: Path) -> None: + """A drop whose own mtime is 90 days old is archived now, so it must not age out now. + + Mutation: drop `stamp=True` from `_after_processing`. Red: the copy keeps its 90-day mtime and + the sweep deletes it the moment it lands.""" + inbox = tmp_path / "in" + src = _source(inbox) + src._prepare_subdirs() + drop = _aged(inbox / "old-drop.hl7", 90) + before = time.time() + + src._after_processing(drop) + + [copy] = list(src.processed_dir.iterdir()) + assert copy.stat().st_mtime >= before - 2 + assert copy.read_text(encoding="ascii") == "MSH|^~\\&|SYNTHETIC" # bytes untouched + assert _prune(src) == 0 + assert copy.exists() + + +def test_quarantine_is_not_stamped(tmp_path: Path) -> None: + """Only the `.processed` move stamps; a `.error` move keeps the drop's own mtime.""" + inbox = tmp_path / "in" + src = _source(inbox) + src._prepare_subdirs() + drop = _aged(inbox / "bad.hl7", 90) + + assert src._move(drop, src.error_dir) + + [copy] = list(src.error_dir.iterdir()) + assert copy.stat().st_mtime < _CUTOFF + + +# --- the retention pass --------------------------------------------------------------------------- + + +@pytest.fixture +async def store(tmp_path: Path) -> AsyncIterator[MessageStore]: + s = await MessageStore.open(tmp_path / "retention.db") + yield s + await s.close() + + +def _settings(**knobs: object) -> RetentionSettings: + off = {"min_free_disk_mb": 0, "wal_checkpoint_seconds": 0, "audit_log_warn_rows": 0} + return RetentionSettings.model_validate({**off, **knobs}) + + +async def test_the_pass_sweeps_each_bound_archive_and_audits_the_count( + store: MessageStore, tmp_path: Path +) -> None: + a = _source(tmp_path / "a") + b = _source(tmp_path / "b") + gone = [_aged(a.processed_dir / "x.hl7", 31), _aged(b.processed_dir / "y.hl7", 31)] + kept = _aged(b.processed_dir / "z.hl7", 1) + runner = RetentionRunner( + store, + _settings(processed_archive_days=30), + clock=lambda: _NOW, + archive_source=lambda: (a, b), + ) + assert runner.enabled + + result = await runner.run_once() + + assert result.processed_archives_deleted == 2 + assert not any(p.exists() for p in gone) and kept.exists() + [row] = [r for r in await store.list_audit(limit=10) if r["action"] == "retention_purge"] + detail = json.loads(row["detail"]) + assert detail["processed_archive_days"] == 30 + assert detail["processed_archives_deleted"] == 2 + assert "x.hl7" not in row["detail"] and "y.hl7" not in row["detail"] + + +async def test_the_pass_protects_every_connections_directories( + store: MessageStore, tmp_path: Path +) -> None: + """The runner hands each archive the watch and `.error` directories of all of them. + + Mutation: pass `protected=frozenset()` in `_prune_processed_archives`. Red.""" + b = _source(tmp_path / "shared") + a = _source(tmp_path / "a", processed_subdir=str(tmp_path / "shared")) + b_drop = _aged(b.directory / "waiting.hl7", 90) + runner = RetentionRunner( + store, + _settings(processed_archive_days=30), + clock=lambda: _NOW, + archive_source=lambda: (a, b), + ) + + result = await runner.run_once() + + assert result.processed_archives_deleted == 0 + assert b_drop.exists() + + +async def test_an_unbound_inbounds_directories_are_protected( + store: MessageStore, tmp_path: Path +) -> None: + """B is stopped, so it is not in the archive list; the guard source still names its watch + directory, which A's archive directory holds. + + Mutation: ignore `guard_source` in `_prune_processed_archives`. Red.""" + stopped_inbox = tmp_path / "shared" / "b-inbox" + a = _source(tmp_path / "a", processed_subdir=str(tmp_path / "shared")) + b_drop = _aged(stopped_inbox / "waiting.hl7", 90) + runner = RetentionRunner( + store, + _settings(processed_archive_days=30), + clock=lambda: _NOW, + archive_source=lambda: (a,), + guard_source=lambda: (stopped_inbox, stopped_inbox / ".error"), + ) + + result = await runner.run_once() + + assert result.processed_archives_deleted == 0 + assert b_drop.exists() + + +async def test_an_unresolvable_guard_sweeps_nothing(store: MessageStore, tmp_path: Path) -> None: + """A guard list that could not be built fails closed.""" + src = _source(tmp_path / "in") + old = _aged(src.processed_dir / "old.hl7", 90) + runner = RetentionRunner( + store, + _settings(processed_archive_days=30), + clock=lambda: _NOW, + archive_source=lambda: (src,), + guard_source=lambda: None, + ) + + result = await runner.run_once() + + assert result.processed_archives_deleted == 0 + assert old.exists() + + +async def test_the_pass_spares_a_sibling_shards_drops(store: MessageStore, tmp_path: Path) -> None: + """End to end: shard s1's archive directory is shard s2's watch directory. s2's drop survives. + + Mutation: as `test_the_guard_names_a_sibling_engine_shards_inbound`. Red.""" + sibling = tmp_path / "sibling" + reg = Registry() + reg.add_inbound(_file_inbound("IB_MINE", tmp_path / "mine", shard="s1")) + reg.add_inbound(_file_inbound("IB_SIBLING", sibling, shard="s2")) + runner = _runner(filter_registry_for_shard(reg, "s1")) + a = _source(tmp_path / "mine", processed_subdir=str(sibling)) + drop = _aged(sibling / "waiting.hl7", 90) + retention = RetentionRunner( + store, + _settings(processed_archive_days=30), + clock=lambda: _NOW, + archive_source=lambda: (a,), + guard_source=runner.archive_guard_dirs, + ) + + result = await retention.run_once() + + assert result.processed_archives_deleted == 0 + assert drop.exists() + + +async def test_the_pass_spares_a_file_outbounds_deliveries( + store: MessageStore, tmp_path: Path +) -> None: + """End to end: the archive directory is a File outbound's target. The delivery survives. + + Mutation: as `test_the_guard_names_every_file_outbound_target_directory`. Red.""" + out = tmp_path / "out" + reg = Registry() + reg.add_inbound(_file_inbound("IB_A", tmp_path / "a")) + reg.add_outbound(_file_outbound("OB_PARTNER", out)) + runner = _runner(reg) + a = _source(tmp_path / "a", processed_subdir=str(out)) + delivered = _aged(out / "for-pickup.hl7", 90) + retention = RetentionRunner( + store, + _settings(processed_archive_days=30), + clock=lambda: _NOW, + archive_source=lambda: (a,), + guard_source=runner.archive_guard_dirs, + ) + + result = await retention.run_once() + + assert result.processed_archives_deleted == 0 + assert delivered.exists() + + +async def test_the_pass_sweeps_nothing_when_a_guarded_identity_is_unreadable( + store: MessageStore, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """End to end: one guarded directory cannot be stat'ed, so no archive is swept. + + Mutation: as `test_an_archive_that_cannot_read_an_unresolved_directory_sweeps_nothing`. + Red.""" + src = _source(tmp_path / "in") + old = _aged(src.processed_dir / "old.hl7", 90) + other = tmp_path / "other-share" + other.mkdir() + _refuse_stat_of(monkeypatch, other) + retention = RetentionRunner( + store, + _settings(processed_archive_days=30), + clock=lambda: _NOW, + archive_source=lambda: (src,), + guard_source=lambda: (other,), + ) + + result = await retention.run_once() + + assert result.processed_archives_deleted == 0 + assert old.exists() + + +def test_the_engines_own_directories_are_read_from_its_settings(tmp_path: Path) -> None: + """A file setting gives its parent directory, a directory setting a tree. + + Mutation: skip `_file` settings, or `destination`, in `engine_owned_dirs`. Red.""" + dirs, trees = engine_owned_dirs( + ( + StoreSettings(path=str(tmp_path / "data" / "mf.db")), + ApiSettings(tls_cert_file=str(tmp_path / "tls" / "c.pem")), + BackupSettings(destination=str(tmp_path / "bak")), + LoggingSettings( + log_dir=str(tmp_path / "logs"), file=str(tmp_path / "applog" / "e.log") + ), + ), + dirs=(tmp_path / "config",), + ) + + assert {tmp_path / "data", tmp_path / "tls", tmp_path / "config"} <= set(dirs) + assert tmp_path / "applog" in dirs # [logging].file, a field named plain `file` + assert {tmp_path / "bak", tmp_path / "logs"} <= set(trees) + + +@pytest.mark.parametrize("where", ["equal", "holds", "inside-tree"]) +async def test_the_pass_spares_the_engines_own_directories( + store: MessageStore, tmp_path: Path, where: str +) -> None: + """An archive directory that is, or holds, an engine directory, or sits inside an engine tree, + deletes nothing there. + + Mutation: drop `engine_dirs`/`engine_trees` from the guard, or the `_archive_inside` check. Red.""" + data = tmp_path / "eng" / "data" + backups = tmp_path / "eng" / "backups" + archive = {"equal": data, "holds": tmp_path / "eng", "inside-tree": backups / "x"}[where] + a = _source(tmp_path / "a", processed_subdir=str(archive)) + victim = _aged(archive / "engine-owned.bin", 90) + retention = RetentionRunner( + store, + _settings(processed_archive_days=30), + clock=lambda: _NOW, + archive_source=lambda: (a,), + engine_dirs=(data,), + engine_trees=(backups,), + ) + + result = await retention.run_once() + + assert result.processed_archives_deleted == 0 + assert victim.exists() + + +class _InlineCredential: + """Stands in for an alternate-credential context: runs the call on the calling thread.""" + + async def run(self, fn: Callable[..., object], *args: object, **kwargs: object) -> object: + return fn(*args, **kwargs) + + +async def test_a_refused_sharer_leaves_the_directory_to_a_sibling( + store: MessageStore, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A and B share one archive directory. A, with no credential, cannot read a guarded directory + and refuses; B's credential can read it, so B still sweeps the shared directory. + + Mutation: mark the directory swept before the sweep runs. Red.""" + shared = tmp_path / "shared-archive" + a = _source(tmp_path / "a", processed_subdir=str(shared)) + b = _source(tmp_path / "b", processed_subdir=str(shared)) + monkeypatch.setattr(b, "_cred_ctx", _InlineCredential()) + old = _aged(shared / "old.hl7", 90) + other = tmp_path / "partner-share" + other.mkdir() + readable = {"now": False} + real_stat = os.stat + target = Path(os.path.realpath(other)) + + def refusing(path: object, *args: object, **kwargs: object) -> os.stat_result: + if not readable["now"] and isinstance(path, str | os.PathLike) and Path(path) == target: + raise PermissionError(errno.EACCES, "refused") + return real_stat(path, *args, **kwargs) # type: ignore[arg-type] + + real_build = build_archive_guard + + def build(dirs: Iterable[Path]) -> ArchiveGuard: + dirs = tuple(dirs) + readable["now"] = other in dirs # only B's credentialed re-read can reach the share + try: + return real_build(dirs) + finally: + readable["now"] = False + + monkeypatch.setattr(os, "stat", refusing) + monkeypatch.setattr(file_transport, "build_archive_guard", build) + retention = RetentionRunner( + store, + _settings(processed_archive_days=30), + clock=lambda: _NOW, + archive_source=lambda: (a, b), + guard_source=lambda: (other,), + ) + + result = await retention.run_once() + + assert result.processed_archives_deleted == 1 + assert not old.exists() + + +async def test_a_zero_window_sweeps_nothing(store: MessageStore, tmp_path: Path) -> None: + src = _source(tmp_path / "in") + old = _aged(src.processed_dir / "old.hl7", 365) + runner = RetentionRunner(store, _settings(), clock=lambda: _NOW, archive_source=lambda: (src,)) + + result = await runner.run_once() + + assert result.processed_archives_deleted == 0 + assert old.exists() diff --git a/tests/test_retention_classification_drift.py b/tests/test_retention_classification_drift.py index 45293467e..5548ceded 100644 --- a/tests/test_retention_classification_drift.py +++ b/tests/test_retention_classification_drift.py @@ -191,8 +191,9 @@ def test_the_floor_is_seven_not_six_and_is_enforced() -> None: Mutation: set MIN to 6 — reds, because the real length is checked against it AND against the documented reason for the value. """ - assert MIN_PHI_RETENTION_WINDOWS == 10, ( - "the floor changed. It is 10 (9 until BACKLOG #1966 added [logging].forward_spool_max_bytes) — and 9 was CORRECTED BY THIS TEST rather than counted: " + assert MIN_PHI_RETENTION_WINDOWS == 11, ( + "the floor changed. It is 11 (10 until BACKLOG #1188 added [retention].processed_archive_days, " + "9 until BACKLOG #1966 added [logging].forward_spool_max_bytes) — and 9 was CORRECTED BY THIS TEST rather than counted: " "it was first written as 7, and the two-way equality above immediately reported " "[backup].retention_keep and [retention].connection_event_retention_hours as " "documented-but-absent. If the classification genuinely grew or shrank, update this assertion " @@ -220,6 +221,8 @@ def test_the_auto_bound_and_warn_only_split_matches_the_owner_ruling() -> None: "[security].delete_message_bodies_after_days", "[retention].dead_letter_days", "[retention].reference_snapshot_days", + # Owner ruling A2 of 2026-10-08 (BACKLOG #1188): the fourth window. + "[retention].processed_archive_days", }, f"auto-bound set drifted from the owner ruling: {sorted(auto)}" assert "[retention].state_max_age_days" in warn, ( "state_max_age_days must NOT auto-bound — purge_state deletes by `set_at`, which a read never " diff --git a/tests/test_retention_gate_parity.py b/tests/test_retention_gate_parity.py index 75750903d..ff61cc520 100644 --- a/tests/test_retention_gate_parity.py +++ b/tests/test_retention_gate_parity.py @@ -34,6 +34,7 @@ _BODY = "[security].delete_message_bodies_after_days" _DEAD = "[retention].dead_letter_days" _REFERENCE = "[retention].reference_snapshot_days" +_PROCESSED = "[retention].processed_archive_days" _STATE_CAVEAT = ( "a window on it deletes transform state by write time, so a Handler's correlation " "entry could vanish while still in use; until state has a non-write-time eviction key, " @@ -95,12 +96,13 @@ def test_serve_auto_bound_notice_is_unchanged( rc, captured = _run_secure_serve(tmp_path, monkeypatch, _EGRESS + _SECURE_ALERTS) assert rc == 0 assert _gate_lines(capsys.readouterr().err) == [ - _auto_bound_notice(f"{_BODY}, {_DEAD}, {_REFERENCE}", "prod") + _auto_bound_notice(f"{_BODY}, {_DEAD}, {_REFERENCE}, {_PROCESSED}", "prod") ] retention = captured["retention_settings"] assert retention.messages_days == 30 # type: ignore[attr-defined] assert retention.dead_letter_days == 30 # type: ignore[attr-defined] assert retention.reference_snapshot_days == 30 # type: ignore[attr-defined] + assert retention.processed_archive_days == 30 # type: ignore[attr-defined] def test_serve_body_window_refusal_is_unchanged( @@ -116,7 +118,7 @@ def test_serve_body_window_refusal_is_unchanged( assert rc == 2 # The notice for the one unset window comes first, then the refusal for the explicit 0. assert _gate_lines(capsys.readouterr().err) == [ - _auto_bound_notice(_REFERENCE, "prod"), + _auto_bound_notice(f"{_REFERENCE}, {_PROCESSED}", "prod"), f"error: a data-retention window is explicitly disabled for {_BODY} on a production PHI " "instance ('prod'); refusing to start — PHI message bodies would be retained indefinitely " "(unbounded PHI at rest, ASVS 14.2.4/14.2.7). Set the window(s) to a positive number of " @@ -133,7 +135,7 @@ def test_serve_body_window_warning_is_unchanged( 'security.enforcement = "warn"\n' + _EGRESS + "security.delete_message_bodies_after_days = 0\n" - + "[retention]\ndead_letter_days = 30\nreference_snapshot_days = 30\n" + + "[retention]\ndead_letter_days = 30\nreference_snapshot_days = 30\nprocessed_archive_days = 30\n" + _SECURE_ALERTS, env="staging", ) @@ -159,7 +161,7 @@ def test_serve_acknowledged_body_windows_are_unchanged( assert rc == 0 # The acknowledgement turns the auto-bound off, so every unset window stays 0. assert captured["retention_settings"].messages_days == 0 # type: ignore[attr-defined] - windows = f"{_BODY}, {_DEAD}, {_REFERENCE}" + windows = f"{_BODY}, {_DEAD}, {_REFERENCE}, {_PROCESSED}" streams = capsys.readouterr() assert _gate_lines(streams.err) == [ f"warning: {_ACK}=true — a production PHI instance ('prod') retains PHI message bodies " @@ -179,7 +181,7 @@ def test_serve_warn_only_tier_refusal_is_unchanged( rc, _ = _run_secure_serve(tmp_path, monkeypatch, _EGRESS + _SECURE_ALERTS) assert rc == 2 assert _gate_lines(capsys.readouterr().err) == [ - _auto_bound_notice(f"{_BODY}, {_DEAD}, {_REFERENCE}", "prod"), + _auto_bound_notice(f"{_BODY}, {_DEAD}, {_REFERENCE}, {_PROCESSED}", "prod"), "error: these classified PHI tiers have no retention window on a PHI instance ('prod') " "and would accumulate without bound; refusing to start, because each needs a window or " f"its own audited acknowledgement (ASVS 14.2.7): {_WARN_ONLY_TIERS}.", @@ -198,7 +200,7 @@ def test_serve_warn_only_tier_warning_is_unchanged( ) assert rc == 0 assert _gate_lines(capsys.readouterr().err) == [ - _auto_bound_notice(f"{_BODY}, {_DEAD}, {_REFERENCE}", "staging"), + _auto_bound_notice(f"{_BODY}, {_DEAD}, {_REFERENCE}, {_PROCESSED}", "staging"), "warning: these classified PHI tiers have no retention window on a PHI instance " "('staging') and will accumulate without bound. They are deliberately NOT defaulted " "(owner ruling 2026-07-30); under enforcement=enforce this refuses to start: " @@ -223,7 +225,7 @@ def test_serve_acknowledged_warn_only_tiers_are_unchanged( assert rc == 0 streams = capsys.readouterr() assert _gate_lines(streams.err) == [ - _auto_bound_notice(f"{_BODY}, {_DEAD}, {_REFERENCE}", "prod") + _auto_bound_notice(f"{_BODY}, {_DEAD}, {_REFERENCE}, {_PROCESSED}", "prod") ] assert _audit_lines(streams.out) == [ "AUDIT: starting a production PHI instance (environment 'prod') with " @@ -312,7 +314,7 @@ def test_check_says_so_when_nothing_reads_as_unbounded(tmp_path: Path) -> None: toml = tmp_path / "messagefoundry.toml" toml.write_text( 'ai.environment = "prod"\nsecurity.delete_message_bodies_after_days = 30\n' - "[retention]\ndead_letter_days = 30\nreference_snapshot_days = 30\n", + "[retention]\ndead_letter_days = 30\nreference_snapshot_days = 30\nprocessed_archive_days = 30\n", encoding="utf-8", ) leg = _retention_leg(toml) @@ -366,7 +368,7 @@ def test_a_line_that_prints_no_environment_carries_no_placeholder_note(tmp_path: toml = tmp_path / "messagefoundry.toml" toml.write_text( "security.delete_message_bodies_after_days = 30\n" - "[retention]\ndead_letter_days = 30\nreference_snapshot_days = 30\n", + "[retention]\ndead_letter_days = 30\nreference_snapshot_days = 30\nprocessed_archive_days = 30\n", encoding="utf-8", ) assert _retention_leg(toml).detail == _SILENT @@ -453,7 +455,8 @@ def test_the_notice_names_the_bound_the_window_took(monkeypatch: pytest.MonkeyPa assert settings.retention.dead_letter_days == 7 and settings.retention.messages_days == 14 notice = outcome.lines[0].text assert ( - f"{_BODY} (14 days), {_DEAD} (7 days), {_REFERENCE} (14 days) defaulted ON for a PHI " + f"{_BODY} (14 days), {_DEAD} (7 days), {_REFERENCE} (14 days), {_PROCESSED} (14 days) " + "defaulted ON for a PHI " "instance ('dev')" in notice ) assert "30 days" not in notice @@ -467,4 +470,7 @@ def test_the_notice_names_the_bound_the_window_took(monkeypatch: pytest.MonkeyPa again = retention_classification.evaluate_retention_gate( _dev_settings(), enforcing=False, production=False, env_name="dev" ) - assert f"{_BODY}, {_DEAD}, {_REFERENCE} defaulted ON (14 days) for" in again.lines[0].text + assert ( + f"{_BODY}, {_DEAD}, {_REFERENCE}, {_PROCESSED} defaulted ON (14 days) for" + in again.lines[0].text + ) diff --git a/tests/test_security_doc_drift.py b/tests/test_security_doc_drift.py index 128a9ce94..2e7438364 100644 --- a/tests/test_security_doc_drift.py +++ b/tests/test_security_doc_drift.py @@ -3867,8 +3867,8 @@ def test_retention_gate_reader_detects_a_planted_violation() -> None: ), # A window dropped from the posture paragraph while a table row still names it. "window-unnamed": ( - "`[security].delete_message_bodies_after_days`, `[retention].dead_letter_days` and", - "`[retention].dead_letter_days` and", + "`[security].delete_message_bodies_after_days`, `[retention].dead_letter_days`,", + "`[retention].dead_letter_days`,", "delete_message_bodies_after_days", ), } diff --git a/tests/test_storage_floor.py b/tests/test_storage_floor.py index 6b983447a..c13a0c9e6 100644 --- a/tests/test_storage_floor.py +++ b/tests/test_storage_floor.py @@ -119,6 +119,7 @@ def _write_quiet_config(tmp_path: Path, extra: str = "") -> None: "security.delete_message_bodies_after_days = 30\n" "retention.dead_letter_days = 30\n" "retention.reference_snapshot_days = 30\n" + "retention.processed_archive_days = 30\n" "retention.state_max_age_days = 30\n" "retention.search_preset_days = 30\n" "security.local_access_only = true\n" + extra,