diff --git a/changelog.d/2264.fixed.md b/changelog.d/2264.fixed.md index 0379545e9f..5c774cf912 100644 --- a/changelog.d/2264.fixed.md +++ b/changelog.d/2264.fixed.md @@ -1,6 +1,6 @@ - **The upload retention stop logs an audit gap only when one exists, and names the files.** At - shutdown the retention runner gives a sweep in flight 5 seconds, then cancels it. It used to log - an ERROR saying files may have lost their `upload.prune` row, even when the sweep had removed + shutdown the retention runner gives a sweep in flight a bounded wait, then cancels it. It used to + log an ERROR saying files may have lost their `upload.prune` row, even when the sweep had removed nothing. It now tracks each file the sweep removed and each row written. The ERROR names every removed file with no row yet, and a sweep that lost no row logs a WARNING instead. A runner with no audit callback says it owed no rows. (`BACKLOG #2264`) diff --git a/changelog.d/2322.fixed.md b/changelog.d/2322.fixed.md index 8deb2f45cf..fe9427ae92 100644 --- a/changelog.d/2322.fixed.md +++ b/changelog.d/2322.fixed.md @@ -2,8 +2,8 @@ Sealing it would list an upload with no usable body, and would seal a planted sidecar too. `rotate-key` now counts such an upload as unsealable rather than skipped. The upload stays inside retention: the prune removes it at expiry, timed from the sidecar's mtime, and the quota bills - it. A holder of `files:access_any` can delete it with `DELETE /uploads/{file_id}`, and still - gets 423 on browse and resend. Everyone else, its owner included, still gets 404. The startup + it. A holder of both `files:access_any` and `files:delete` can delete it with + `DELETE /uploads/{file_id}`, after step-up, and still gets 423 on browse and resend. Everyone else, its owner included, still gets 404. The startup warning no longer tells the operator to run `rotate-key` for it, and the 423 text no longer promises that `rotate-key` will seal it. It stays out of the listing. At least an oversized non-text body, and one that is not a regular file, are not covered; `docs/PHI.md` section 3 diff --git a/changelog.d/3184.changed.md b/changelog.d/3184.changed.md new file mode 100644 index 0000000000..83f4fe89f2 --- /dev/null +++ b/changelog.d/3184.changed.md @@ -0,0 +1,6 @@ +- **An upload no longer runs the uploaded-logs retention sweep on its own request.** Under `serve`, + the upload route wakes the retention runner, which sweeps in the background and audits each + pruned file as before. An app built without the runner still sweeps at save time. The sweep now + writes each `upload.prune` row as soon as it removes that file, so a shutdown that cuts a sweep + short loses at most the row of the file in progress. Each quota check now walks the uploads + directory once rather than twice. (`BACKLOG #3184`) diff --git a/changelog.d/3240.fixed.md b/changelog.d/3240.fixed.md new file mode 100644 index 0000000000..ca01cd9a7e --- /dev/null +++ b/changelog.d/3240.fixed.md @@ -0,0 +1,5 @@ +- **`rotate-key` counts a plaintext upload body it sealed even when that upload's sidecar could + not be read.** Such an upload is now in the sealed-plaintext count as well as in `skipped`, so a + planted body is not sealed uncounted. An upload whose two plaintext halves are sealed in two runs + counts in both. `DELETE /uploads/{file_id}` now answers 404, not 423, for an upload whose sidecar + is removed while the request runs. (`BACKLOG #3240`) diff --git a/changelog.d/3256.fixed.md b/changelog.d/3256.fixed.md new file mode 100644 index 0000000000..654e9ed9a9 --- /dev/null +++ b/changelog.d/3256.fixed.md @@ -0,0 +1,7 @@ +- **On Postgres and SQL Server, the engine-shard upload reservation now reads the database server's + clock.** The row's activity stamp and its staleness test used each engine host's own clock, so a + host whose clock ran ahead could hold a leaked slot for as long as the skew. Each heartbeat touch + of a live upload's reservation is now bounded, so a stalled store cannot hold the heartbeat + indefinitely. `serve` now bounds uvicorn's connection drain at a stop to 3 seconds, so an open + request can no longer use up the service's stop window before the engine shuts down. + (`BACKLOG #3256`) diff --git a/docs/PHI.md b/docs/PHI.md index 600fc3579f..f9cb3ba75b 100644 --- a/docs/PHI.md +++ b/docs/PHI.md @@ -171,7 +171,7 @@ destruction) are documented in [§3](#3-encryption-at-rest) under the matching h | `attachment_chunk.ciphertext` (ADR 0105 / #149) | all three | **Yes** — one slice of a detached very-large document (e.g. a base64 PDF from OBX-5.5) | **Yes, when a key is set** — store cipher, **sealed per chunk on write**; AAD `("attachment_chunk","ciphertext",attachment_id,seq)`; store DEK | **PL-1** | The document is detached at ingress, content-addressed (`sha256` of the verbatim plaintext) and chunked; the message keeps only an `mfdoc:v1:ref:` handle. Read back via `GET /messages/{id}/attachments/{id}` (§3) | ``rides `[security].delete_message_bodies_after_days` `` | | `attachment` header row (`content_type`, `total_bytes`, `refcount`, `created_at`) + `message_attachment` linkage | all three | **No** — size/type/linkage only | No (metadata, deliberately not ciphered) | **PL-4** | The linkage row is the security crux of the download route: it scopes a content address to a message the caller may already read | ``rides `[security].delete_message_bodies_after_days` `` | | `response.body` (ADR 0013 captured replies; ADR 0021 `kind='ack_sent'`) | all three | **Yes** — the partner's reply body, or the ACK/NAK the engine returned | **Yes, when a key is set** — store cipher; AAD `("response","body",message_id,destination_name,response_seq)`; store DEK | **PL-1** | Composite PK, so it rides its own migration/rotation pass. An **ACK body is stored only when the store cipher is active** — on a keyless store it is `NULL` rather than plaintext (fail-safe), and a NAK never stores a body at all | ``rides `[security].delete_message_bodies_after_days` `` | -| `[store].uploads_dir/*.blob` + `*.meta` — the cipher cells `uploaded_file.body` / `uploaded_file.meta` (offline uploaded logs, ADR 0134) | all (filesystem, not the DB) | **Yes** — an operator-uploaded diagnostic message file, held for offline browsing decoupled from any connection | **Yes, when a key is set** — the **same store cipher** (`build_store_cipher`); AAD `("uploaded_file","body")` / `("uploaded_file","meta")` + `file_id`; store DEK. **identity/plaintext-on-disk otherwise** (the File-connector-spill tier below) | **PL-1** | A PHI-at-rest location **outside** the message store, opt-in (unset ⇒ the subsystem is disabled — no surface). On-disk identity is a random 32-hex `file_id` (path-traversal guard); the operator filename is display-only. Every access is `files:*`-gated, browse is step-up + PHI-hop-guarded, all audited (metadata only). No upload audit row carries the file name, which an operator may have chosen to hold a patient identifier: a stored file is named by its `file_id`, and a refused one by its reason (vault BACKLOG #2619). `rotate-key` re-seals each upload it can read under the active key, and seals a plaintext one. **On a keyed store a plaintext upload is refused on read until it does** (BACKLOG #1169, owner ruling 2026-09-23; see [§3](#3-encryption-at-rest), which also names the plaintext upload it never seals and how retention still removes it, BACKLOG #2322). The dir is created `0o700` and the sidecar written `0o600` **best-effort, and both are no-ops on Windows** — the engine applies no ACL here (it does not call the `icacls` enforcer). **Retention + quotas (ASVS 5.2.4):** uploaded files auto-prune after `[store].uploads_retention_days` (default **30**) — swept opportunistically at save time and by a periodic task. A pruned pair is audited (`upload.prune`, file_id + uploader only, never content). Which pairs a pass counts as pruned is stated once, in `uploads.UploadStore.prune_expired`. **Not every removal gets a row.** At least these go unaudited: a write leftover the orphan sweep removes (no metadata exists, so it is a count and a WARNING); a pair a runner sweep removed and had not audited when its bounded shutdown wait (5 s) ran out. The runner then cancels the sweep and logs each such `file_id` at ERROR. An audit call that raises is logged per file. A save-time sweep that shutdown cancels, while it finishes for a cancelled request, is logged at ERROR (BACKLOG #2261). A save cancelled after its file landed removes that pair, since no `upload.create` row was written for it (BACKLOG #2262). A removal the filesystem refuses is logged at ERROR. A request cancelled later, in its own `upload.create` audit write, still keeps the file with no creation row. Per-uploader caps `[store].max_upload_files_per_user` (default **100**) / `[store].max_upload_total_bytes_per_user` (default **250 MiB**) bound the at-rest volume; a would-be over-quota upload is refused **HTTP 409** with an `upload.reject_quota` audit before anything is written (defaults-ON, `ge=1` floors). The quota is scoped to the **`uploads_dir`, not to the process** — the sidecar scan is uncached, so engine shards sharing one dir enforce **one** budget between them (measured 2026-08-10); shards given separate dirs get separate budgets by construction. The check and the write it authorizes are one critical section per process, and a shard mid-upload is held as an atomic reservation on the unified store every shard shares (`Store.reserve_upload_quota`). Each upload reserves, reads the in-flight total back, then scans the disk, and is refused unless the disk, the other shards' in-flight uploads and this file all fit (ASVS 2.3.4, BACKLOG #1941). The residuals that survive it, and their bounds, are stated once in `uploads.UploadQuotaError`. Harden the dir + volume encryption ([§10](#10-secure-deployment--operations-checklist)) | `` `[store].uploads_retention_days` `` | +| `[store].uploads_dir/*.blob` + `*.meta` — the cipher cells `uploaded_file.body` / `uploaded_file.meta` (offline uploaded logs, ADR 0134) | all (filesystem, not the DB) | **Yes** — an operator-uploaded diagnostic message file, held for offline browsing decoupled from any connection | **Yes, when a key is set** — the **same store cipher** (`build_store_cipher`); AAD `("uploaded_file","body")` / `("uploaded_file","meta")` + `file_id`; store DEK. **identity/plaintext-on-disk otherwise** (the File-connector-spill tier below) | **PL-1** | A PHI-at-rest location **outside** the message store, opt-in (unset ⇒ the subsystem is disabled — no surface). On-disk identity is a random 32-hex `file_id` (path-traversal guard); the operator filename is display-only. Every access is `files:*`-gated, browse is step-up + PHI-hop-guarded, all audited (metadata only). No upload audit row carries the file name, which an operator may have chosen to hold a patient identifier: a stored file is named by its `file_id`, and a refused one by its reason (vault BACKLOG #2619). `rotate-key` re-seals each upload it can read under the active key, and seals a plaintext one, except at least a plaintext sidecar over a body that is missing or is not text, or over a body it could not read. **On a keyed store a plaintext upload is refused on read until it does** (BACKLOG #1169, owner ruling 2026-09-23; see [§3](#3-encryption-at-rest), which also names the plaintext upload it never seals and how retention still removes it, BACKLOG #2322). The dir is created `0o700` and the sidecar written `0o600` **best-effort, and both are no-ops on Windows** — the engine applies no ACL here (it does not call the `icacls` enforcer). **Retention + quotas (ASVS 5.2.4):** uploaded files auto-prune after `[store].uploads_retention_days` (default **30**) — swept by a periodic task, which each save wakes under `serve`, and at save time in an app with no such task (BACKLOG #3184). A pruned pair is audited (`upload.prune`, file_id + uploader only, never content). Which pairs a pass counts as pruned is stated once, in `uploads.UploadStore.prune_expired`. **Not every removal gets a row.** At least these go unaudited: a write leftover the orphan sweep removes (no metadata exists, so it is a count and a WARNING); a pair a runner sweep removed and had not audited when its bounded shutdown wait ran out (5 s at most, `uploads._DEFAULT_STOP_TIMEOUT_SECONDS`; under `serve` it is a share of its teardown step's bound, which is shorter). The runner then cancels the sweep and logs each such `file_id` at ERROR. An audit call that raises is logged per file. A save-time sweep that shutdown cancels, while it finishes for a cancelled request, is logged at ERROR (BACKLOG #2261). A save cancelled after its file landed removes that pair, since no `upload.create` row was written for it (BACKLOG #2262). A removal the filesystem refuses is logged at ERROR. A request cancelled later, in its own `upload.create` audit write, still keeps the file with no creation row. Per-uploader caps `[store].max_upload_files_per_user` (default **100**) / `[store].max_upload_total_bytes_per_user` (default **250 MiB**) bound the at-rest volume; a would-be over-quota upload is refused **HTTP 409** with an `upload.reject_quota` audit before anything is written (defaults-ON, `ge=1` floors). The quota is scoped to the **`uploads_dir`, not to the process** — the sidecar scan is uncached, so engine shards sharing one dir enforce **one** budget between them (measured 2026-08-10); engine shards given separate dirs get separate budgets by construction. The check and the write it authorizes are one critical section per process, and an engine shard mid-upload is held as an atomic reservation on the unified store every engine shard shares (`Store.reserve_upload_quota`). Each upload reserves, reads the in-flight total back, then scans the disk, and is refused unless the disk, the other engine shards' in-flight uploads and this file all fit (ASVS 2.3.4, BACKLOG #1941). The residuals that survive it, and their bounds, are stated once in `uploads.UploadQuotaError`. Harden the dir + volume encryption ([§10](#10-secure-deployment--operations-checklist)) | `` `[store].uploads_retention_days` `` | | `[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` | @@ -593,11 +593,14 @@ for defense-in-depth without swapping the `aiosqlite` connector. in the future. The quota bills it to the uploader its sidecar names, at the body's size on disk. When the prune removes its body, the audit row, like the store's `delete()`, carries its path id with blank owner fields, since its owner cannot be verified. A missing body leaves no row, as for - any body-less pair. An operator holding `files:access_any` can also delete it through - `DELETE /uploads/{id}`, which writes the same blank-owner `upload.delete` row. Its owner cannot, + any body-less pair. An operator holding both `files:access_any` and `files:delete` can also + delete it through `DELETE /uploads/{id}`, after step-up, which writes the same blank-owner + `upload.delete` row. Its owner cannot, and gets 404. At least these are still outside retention, and the API cannot delete them: a - blank or non-UTF-8 sidecar, and a plaintext sidecar over a body that is oversized and not text, - or is not a regular file. + blank sidecar, or one that is not UTF-8 within the part the class check reads, and a plaintext + sidecar over a body that is oversized and not text, or is not a regular file. A plaintext + sidecar that is not UTF-8 only past that part is removed by the prune at expiry, but the API + cannot delete it either: its read fails before the class check. **Under `cipher_provider = "vault_transit"` a plaintext upload still reads back as plaintext.** `rotate-key` refuses to run in that mode: it needs a local active key, and the store's own rotation raises there (BACKLOG #1165). So no command could ever seal the upload, and refusing it diff --git a/messagefoundry/__main__.py b/messagefoundry/__main__.py index c2af797ee0..ada2e04bb5 100644 --- a/messagefoundry/__main__.py +++ b/messagefoundry/__main__.py @@ -2479,6 +2479,22 @@ def _missing_service_config_refused(service_config: str | None, *, fleet: str = return True +#: How long uvicorn waits, at a stop, for open connections and their requests to finish before it +#: cancels them and runs the lifespan teardown (BACKLOG #3256). The whole stop shares NSSM's +#: AppStopMethodConsole window (scripts/service/install-service.ps1) with that teardown and +#: engine.stop(). Unset, this wait had no bound, so one long request could spend the window and +#: the service would be killed mid-teardown. A request cut off here is cancelled like one past its +#: deadline, and uvicorn starts the teardown without waiting for it, so its cancel paths run +#: alongside the teardown. **Any route that acts in a worker thread and then writes its audit row +#: can lose that row here,** such as ``DELETE /uploads/{id}``, whose unlink finishes in its thread +#: while the request is cancelled before its ``upload.delete`` row. At least these upload cases +#: lose too: a save whose write is still running waits for it, with no bound, and may then pay its +#: quota slot back to a closed store, leaving the slot to go stale; and a save cancelled in its own +#: ``upload.create`` audit write keeps its file with no creation row. Before this bound, a request +#: still open at the window's end was killed with the process instead, at any point. +_UVICORN_DRAIN_SECONDS = 3.0 + + def _serve(args: argparse.Namespace) -> int: # Vault BACKLOG #2700: first in this function, ahead of the imports below and of config, # because a script injected before the hook is in place runs. It is NOT first in the process: @@ -4643,6 +4659,7 @@ def registry_filter(reg: Registry) -> Registry: # noqa: F811 (local shard-bound "server_header": False, "http": floored_http, "ws": floored_ws, + "timeout_graceful_shutdown": _UVICORN_DRAIN_SECONDS, # BACKLOG #3256; see the constant } from messagefoundry.api.tls import build_api_ssl_context diff --git a/messagefoundry/api/app.py b/messagefoundry/api/app.py index 052fb355f5..74217d3eb3 100644 --- a/messagefoundry/api/app.py +++ b/messagefoundry/api/app.py @@ -40,7 +40,7 @@ import re import shutil import time -from collections.abc import AsyncIterator, Callable, Coroutine, Mapping, Sequence +from collections.abc import AsyncIterator, Awaitable, Callable, Coroutine, Mapping, Sequence from contextlib import asynccontextmanager, suppress from dataclasses import asdict, dataclass, fields, replace from datetime import UTC, datetime @@ -216,7 +216,8 @@ MultipartTooLargeError, parse_single_file_upload, ) -from messagefoundry.api.outlive import OutlivingOperations +from messagefoundry.api.outlive import CANCEL_GRACE_SECONDS, OutlivingOperations +from messagefoundry.api.outlive import DRAIN_TIMEOUT_SECONDS as OUTLIVING_DRAIN_SECONDS from messagefoundry.api.paging import page_total from messagefoundry.api.request_timeout import RequestTimeoutMiddleware from messagefoundry.api.security import ( @@ -1045,37 +1046,255 @@ async def _reconcile_approvals_after_restart(gate: ApprovalGate) -> None: ) -async def _drain_approval_gate(app: FastAPI) -> None: - """Let approval outcome writes still running land before the store closes (BACKLOG #2087). - - A write whose caller was cancelled, such as by a request timeout, finishes on its own, and would - otherwise meet a closed store. ``drain()`` is bounded, and a failure is logged rather than - raised, so the teardown steps after it still run. The gate is read through getattr - because a startup that failed early may not have built one.""" +#: BACKLOG #3191: the wall-clock budget of everything the managed lifespan does after its yield, +#: engine.stop() included. NSSM sends Ctrl+C and escalates after AppStopMethodConsole, 15000 ms +#: (scripts/service/install-service.ps1). Before the lifespan sees the shutdown, uvicorn drains the +#: open connections, for at most ``__main__._UVICORN_DRAIN_SECONDS``, the bound serve passes +#: uvicorn.run (BACKLOG #3256). This budget plus that bound, plus a cut step's grace (below) for +#: each step cut, must stay inside the window with room left for the interpreter to exit. +_SHUTDOWN_BUDGET_SECONDS: Final = 10.0 +#: The share of that budget kept back for engine.stop(). Reserved, never a bound: the stop is not cut +#: short, because it ends in store.close(), and a store left open keeps a non-daemon worker thread +#: alive, which is the hang BACKLOG #2087 fixed. The stop's own waits are bounded inside the engine, +#: and nothing here holds them to this reserve. +_ENGINE_STOP_RESERVE_SECONDS: Final = 3.0 +#: The part of a step's bound handed to the step's own graceful wait, so that wait ends, and cancels +#: what it owns, before the teardown cuts the step. +_INNER_SHARE: Final = 0.8 +#: How long a step cut at its bound gets to unwind its cancellation before the teardown stops waiting +#: for it and moves on, leaving it to finish or fail on its own. +_CUT_GRACE_SECONDS: Final = 0.1 + +#: Steps the teardown stopped waiting for. The event loop holds only a weak reference to a task, so +#: this keeps one alive until it ends. +_left_running: set[asyncio.Future[object]] = set() + + +@dataclass(frozen=True) +class _TeardownStep: + """One shutdown step for :func:`_run_teardown`. + + ``floor`` is the time the step is guaranteed whatever the steps before it took. ``run`` is given + the step's bound and returns what to await. A step that does not apply is left out of the list, + which frees its floor. ``bounded=False`` runs the step to completion and is never cut. At least + two steps run that way: the reap of the managed lifespan's store-reading tasks, since a task left + running would meet the closed store, and engine.stop(), whose ``floor`` is the reserve the steps + before it leave it.""" + + label: str + floor: float + run: Callable[[float], Awaitable[object]] + bounded: bool = True + + +def _log_late_outcome(label: str) -> Callable[[asyncio.Future[object]], None]: + """Log how a step the teardown left running ended, since nobody else reads it.""" + + def _done(work: asyncio.Future[object]) -> None: + _left_running.discard(work) + if not work.cancelled() and (error := work.exception()) is not None: + _log.error("shutdown: %s failed after the teardown moved on", label, exc_info=error) + + return _done + + +async def _await_step(work: asyncio.Future[object], bound: float | None, label: str) -> bool: + """Wait for one step, through any cancel of the caller, for at most ``bound`` seconds. + + A cancel of the caller is noted once and does not reach the step: the step finishes, or is cut + at its bound, as it would have been with no cancel. At the bound the step is cancelled and gets + :data:`_CUT_GRACE_SECONDS` to unwind; one still running then is left to end on its own. Returns + whether the step ended on its own, so its outcome is the caller's to read. + + The wait is asyncio.wait, never asyncio.shield: once the caller is cancelled, shield reports a + later failure of the step to the loop's exception handler as well, so it would be logged twice + (BACKLOG #2087, #2263).""" + loop = asyncio.get_running_loop() + give_up = None if bound is None else loop.time() + bound + cut = noted = False + while not work.done(): + remaining = None if give_up is None else give_up - loop.time() + if remaining is not None and remaining <= 0: + if cut: + _log.error( + "shutdown: %s was still running %.2fs after it was cut; continuing the " + "teardown without it", + label, + _CUT_GRACE_SECONDS, + ) + _left_running.add(work) + work.add_done_callback(_log_late_outcome(label)) + return False + _log.error("shutdown: %s did not finish within %.2fs; cutting it", label, bound) + cut = True + work.cancel() + give_up = loop.time() + _CUT_GRACE_SECONDS + continue + try: + await asyncio.wait((work,), timeout=remaining) + except asyncio.CancelledError: + # Once per step: a cancel that is re-delivered on every pass of the loop, as an anyio + # cancel scope does, would otherwise log a line per pass. + if not noted: + noted = True + _log.warning( + "shutdown: cancelled during %s; it and the remaining steps run before the " + "cancel propagates", + label, + ) + return not (cut and work.cancelled()) + + +async def _run_teardown(steps: Sequence[_TeardownStep], budget: float) -> None: + """Run every step in order, whatever the steps before it did (BACKLOG #3191). + + Each step runs as its own task. A step that raises, or overruns its bound and is cut, is logged, + and the next step runs. Each bounded step gets what is left of ``budget`` after the floors of the + steps still to come, and never less than its own floor, so a slow step spends slack and not a + later step's share. The whole run takes at most ``budget``, plus :data:`_CUT_GRACE_SECONDS` for + each step cut, plus whatever an unbounded step overruns its reserve by. + + A cancel of the caller reaches no step (:func:`_await_step`), and is raised once every step has + run. So is the first other exception a step ends with that is not an :class:`Exception`: the + steps after it still run, so engine.stop() still closes the store. KeyboardInterrupt and + SystemExit are not among them, since asyncio raises those out of the event loop itself.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + budget + current = asyncio.current_task() + # Compared at the end, so a cancel that arrived while this ran is raised, and one already on its + # way out when the teardown began is left to the exception carrying it. + baseline = 0 if current is None else current.cancelling() + later = sum(step.floor for step in steps) + fatal: BaseException | None = None + for step in steps: + later -= step.floor + bound = max(step.floor, deadline - loop.time() - later) + # Called inside the task, so an error raised while building the awaitable is the step's + # failure like any other, and cannot skip the steps after it. + work = asyncio.ensure_future(_invoke(step.run, bound)) + if not await _await_step(work, bound if step.bounded else None, step.label): + continue + try: + work.result() + except asyncio.CancelledError: + _log.error("shutdown: %s ended cancelled; continuing the teardown", step.label) + except Exception: + _log.exception("shutdown: %s failed; continuing the teardown", step.label) + except BaseException as exc: # re-raised below, once the later steps have run + _log.exception("shutdown: %s raised; it is re-raised after the teardown", step.label) + fatal = fatal or exc + if fatal is not None: + raise fatal + if current is not None and current.cancelling() > baseline: + raise asyncio.CancelledError + + +async def _invoke(run: Callable[[float], Awaitable[object]], bound: float) -> object: + """A step's body, as the coroutine :func:`_run_teardown` runs in the step's own task.""" + return await run(bound) + + +def _sync(fn: Callable[[], object]) -> Callable[[float], Awaitable[None]]: + """A synchronous call as a teardown step, so :func:`_run_teardown` guards its failure too.""" + + async def _step(_bound: float) -> None: + fn() + + return _step + + +def _app_drain_steps( + app: FastAPI, store: Store | None, auth: AuthService | None, *, owns_auth: bool +) -> list[_TeardownStep]: + """The drains an app runs at shutdown while the store is still open, in order. + + Both lifespans run them: the managed one before engine.stop(), and the one + :func:`create_app` gives an app built around a caller's engine, before the caller stops it. + + 1. Long operations (vault BACKLOG #2751-#2753): a DR activate or release, or a config reload, + whose caller was cancelled is still running. It finishes, or is cancelled so its own + rollback arm restores state and records 'interrupted'. Before the approval drain, because a + released reload's outcome write follows the reload. + 2. Approval outcome writes (BACKLOG #2087), which would otherwise meet a closed store. + 3. Lock notices (BACKLOG #2216), which read and write the audit log and hand their mail to the + security notifier, so they drain before that notifier stops. + 4. The open summary-access windows (M-5, BACKLOG #1640). Every clean restart otherwise dropped + the open hour's PHI-summary access audit, the rows that control exists to produce. + + Each drain that has a graceful wait of its own is handed part of its bound for it, so it cancels + what it owns before the hard bound cuts it. The gate is read through getattr, because a startup + that failed early may not have built one. The summary auditor is read directly: create_app + always sets it, so a rename fails loudly inside the step's guard rather than silently skipping + the flush.""" + outliving: OutlivingOperations = app.state.outliving_operations + + async def _outliving(bound: float) -> None: + # At most a quarter of it waiting: the grace after is what lets a cancelled operation's + # rollback arm record 'interrupted' before the store closes. With the managed lifespan's + # full set of steps that grace is below the DR hook reap, and the NSSM window leaves the + # budget no room to hold it. CANCEL_GRACE_SECONDS (api/outlive.py) states what that costs. + wait = min(OUTLIVING_DRAIN_SECONDS, bound * _INNER_SHARE / 4) + await outliving.drain(wait, grace=min(CANCEL_GRACE_SECONDS, bound * _INNER_SHARE - wait)) + + steps = [_TeardownStep("the long-operations drain", 2.0, _outliving)] gate: ApprovalGate | None = getattr(app.state, "approval_gate", None) - if gate is None: - return - try: - await gate.drain() - except Exception: - _log.exception("approval gate: the shutdown drain failed; continuing the teardown") + if gate is not None: + steps.append( + _TeardownStep( + "the approval outcome-write drain", 0.5, lambda b: gate.drain(b * _INNER_SHARE) + ) + ) + if auth is not None and owns_auth: + # close_background is the shutdown call the service names, with its own bound (its drain + # and cancel grace, in auth/service.py). Only a bound below that cuts it before it cancels + # its notices itself; a notice cut that way meets the closed store and logs its own failure. + steps.append( + _TeardownStep("the lock-notice drain", 1.0, lambda _b: auth.close_background()) + ) + elif auth is not None: + # A service the caller owns is joined, never closed: closing it would make every later + # notice over it run inline on the sign-in path, which #2216 moved them off. The caller + # closes it before it stops its engine, as close_background asks. + steps.append( + _TeardownStep("the lock-notice drain", 1.0, lambda _b: auth.drain_background()) + ) + if store is not None: + steps.append( + _TeardownStep( + # Named for its loss: the flush clears its windows before it writes them, so a cut + # loses every row not yet written. + "the summary-access flush (a cut loses the unwritten windows' audit rows)", + 0.5, + lambda _b: app.state.summary_auditor.flush(store), + ) + ) + return steps @asynccontextmanager -async def _embedded_approvals_lifespan(app: FastAPI) -> AsyncIterator[None]: +async def _embedded_lifespan(app: FastAPI) -> AsyncIterator[None]: """The lifespan :func:`create_app` gives an app built around a caller's engine. - The caller starts and stops the engine; this runs only the approval gate's own steps, which the - managed lifespan runs too. At start, :meth:`ApprovalGate.reconcile_after_restart`, before the - app serves an approval. At shutdown, :meth:`ApprovalGate.drain` (BACKLOG #2087), so a caller - that stops its engine after the app has shut down loses no outcome write. A caller that passes - its own ``lifespan`` replaces this one and runs these steps itself, with the store open.""" + The caller starts and stops the engine. At start, this runs + :meth:`ApprovalGate.reconcile_after_restart`, before the app serves an approval. At shutdown it + runs the same drains the managed lifespan runs before engine.stop() (:func:`_app_drain_steps`, + BACKLOG #3191), so a caller that stops its engine after the app has shut down loses no outcome + write, notice or audit row. The lock notices of an auth service the caller passed in are waited + for, not closed: the service is the caller's, which closes it with + :meth:`AuthService.close_background` before it stops its engine. A caller that passes its own + ``lifespan`` replaces this one and runs these steps itself, with the store open.""" # create_app installs this only beside an engine, and always builds the gate then. await _reconcile_approvals_after_restart(app.state.approval_gate) try: yield finally: - await _drain_approval_gate(app) + # The caller still has its engine to stop, so its reserve is left out of this budget. + await _run_teardown( + _app_drain_steps( + app, app.state.engine.store, getattr(app.state, "auth", None), owns_auth=False + ), + _SHUTDOWN_BUDGET_SECONDS - _ENGINE_STOP_RESERVE_SECONDS, + ) def _build_approval_gate( @@ -2946,8 +3165,8 @@ def create_app( if engine is not None and lifespan is None: # The embedded path: the caller owns the engine, and this app still owns its approval # gate's start and stop (BACKLOG #1562). A caller passing its own lifespan runs those - # steps itself; _embedded_approvals_lifespan names them. - lifespan = _embedded_approvals_lifespan + # steps itself; _embedded_lifespan names them. + lifespan = _embedded_lifespan app = FastAPI( title="MessageFoundry", version=__version__, @@ -7153,7 +7372,7 @@ async def upload_file( client=client_ip(request), ) - # ASVS 5.2.4: opportunistic age-based retention sweep at save time (off-loop, best-effort). A prune + # ASVS 5.2.4: an age-based retention sweep triggered by the save (off-loop, best-effort). A prune # error must never fail the upload the operator just made — it is logged and retried by the # periodic task. Each pruned file is audited (file_id + uploader, never content). # @@ -7168,6 +7387,11 @@ async def upload_file( # rejects exactly this pairing for dual-control config reload). The owner survives as DATA in # `detail.uploader`, which is where it belongs. # + # BACKLOG #3184: where the lifespan's retention runner is running, the sweep is the runner's, + # woken here, so it is never on this request's path or under its deadline. Only an app with + # no runner (create_app alone) sweeps inline. + if us.wake_retention(): + return _upload_info(meta) # BACKLOG #2261: the request deadline can cancel this handler mid-sweep. prune_on_save then # stops the sweep at its next file and still writes a row for each file it removed. prune_store = engine.store @@ -10330,7 +10554,7 @@ async def _audit_convergence_reload(outcome: ReloadOutcome) -> None: # or opt out in writing via [alerts].security_notifications_required) is enforced fail-closed # at startup by the serve gate (messagefoundry/__main__.py), which checks these SAME two # conditions — not here. This task is owned by the lifespan (started here, drained + closed - # after the engine in the finally below). + # in the finally below, before engine.stop() closes the store it records in). if auth_settings.notify_security_events and alerts_settings is not None: security_notifier = security_notifier_from_settings( alerts_settings, @@ -10583,80 +10807,84 @@ async def _audit_upload_prune(meta: UploadedFileMeta) -> None: ) yield finally: - # BACKLOG #2087: every step before engine.stop() sits in this try, so a cancel delivered - # to the lifespan mid-teardown, or a step that raises, still stops the engine. Without - # it the store's non-daemon worker keeps the process alive. The three store-reading tasks - # are cancelled first, before any await, so a step that raises or a cancel cannot leave - # one running when engine.stop() closes the store. + # BACKLOG #2087, #3191: every step after the yield runs whatever the steps before it + # did. A step that raises, overruns its bound or is cut by a cancel is logged and the + # next one runs, so a cancel delivered mid-teardown still drains, stops the engine and + # closes both notifiers, and is re-raised once they have run. Without engine.stop() + # the store's non-daemon worker keeps the process alive. The three store-reading tasks + # are cancelled first, before any await, so no step can leave one running when + # engine.stop() closes the store. _SHUTDOWN_BUDGET_SECONDS bounds the steps that can be cut. background = [t for t in (reconciler, reaper, credential_reminder) if t is not None] for task in background: task.cancel() - try: - if upload_retention_runner is not None: - await upload_retention_runner.stop() - # gather(return_exceptions): absorbs both our cancellation AND any exception a - # previously-died task stored, so it can't skip the steps below (review M-33). - await asyncio.gather(*background, return_exceptions=True) - # vault BACKLOG #2751-#2753: a DR activate or release, or a config reload, whose - # caller was cancelled is still running on its own. Let it finish, or cancel it so - # its own rollback arm restores state and records 'interrupted', while the store is - # open. Before the approval drain: a released reload's outcome write follows it. - try: - await app.state.outliving_operations.drain() - except Exception: - _log.exception( - "long operations: the shutdown drain failed; continuing the teardown" + + steps: list[_TeardownStep] = [] + if upload_retention_runner is not None: + # Handed part of its bound, so the runner's own wait ends, names any unaudited + # file and cancels the sweep before the teardown cuts it. With the managed set of + # steps that bound is below the runner's own default wait, which then never runs + # in full here. + steps.append( + _TeardownStep( + "the upload retention runner's stop", + 0.5, + lambda b: upload_retention_runner.stop(timeout=b * _INNER_SHARE), ) - # BACKLOG #2087: let approval outcome writes still running land before the store closes. - await _drain_approval_gate(app) - # BACKLOG #2216: a lock notice still in flight reads and writes the audit log, then - # hands its mail to the security notifier, so it drains before engine.stop() closes - # the store and before that notifier stops in the finally below. Bounded, and it logs - # and cancels what it cannot finish; guarded like the drains above so a failure - # cannot skip a later step. - if auth is not None: - try: - await auth.close_background() - except Exception: - _log.exception( - "auth: the shutdown drain of background notices failed; continuing the " - "teardown" - ) - # M-5 (BACKLOG #1640): flush the open summary-access window before the store closes. - # `_SummaryAuditCoalescer.flush` documents itself as the engine-shutdown path and NOTHING - # called it, so every clean restart dropped the open hour's PHI-summary access audit -- - # the rows the control exists to produce, lost exactly when an operator restarts after a - # bulk census fetch. - # - # BEFORE engine.stop(), because that ends in store.close() and the emit needs the store. - # Guarded like the reaper above, so a store error here is logged rather than raised - # out of the teardown. Read directly, not through getattr: create_app always sets the - # auditor, so a rename should fail loudly inside this guard rather than silently skip - # the flush. - try: - await app.state.summary_auditor.flush(store) - except Exception: - _log.exception( - "summary-access coalescer: the shutdown flush failed, so at least one open " - "window's audit row is lost; continuing the teardown" + ) + # return_exceptions absorbs both our cancellation AND any exception a task that died + # earlier stored (review M-33). + steps.append( + _TeardownStep( + "the store-reading tasks", + 0.1, + lambda _b: asyncio.gather(*background, return_exceptions=True), + # Never cut, as before: a task left running here would meet the store that + # engine.stop() closes. + bounded=False, + ) + ) + steps += _app_drain_steps(app, store, auth, owns_auth=True) + if security_notifier is not None: + # It drains queued user emails, and records a notice it could not send in the audit + # log, so it closes while the store is open, after the lock notices that hand it + # mail (BACKLOG #3191). + steps.append( + _TeardownStep( + "the security notifier's close", 0.25, lambda _b: security_notifier.aclose() ) - finally: - if remove_security_signals is not None: - remove_security_signals() - await engine.stop() - # B11: shut down the harness-only instrumented executor (None in production / other tests). - # The engine is stopped (no more to_thread work), so a non-blocking shutdown is clean. - shim_executor = getattr(app.state, "connscale_executor", None) - if shim_executor is not None: - shim_executor.shutdown(wait=False) - if security_notifier is not None: - await ( - security_notifier.aclose() - ) # drain queued user emails, bounded by SMTP timeout - if notifier is not None: - # Stop accepting alerts last (after the engine quiesces) so any final - # connection_stopped/queue_buildup still drains; bounded by the transport timeouts. - await notifier.aclose() + ) + if remove_security_signals is not None: + steps.append( + _TeardownStep( + "the security-signal removal", 0.05, _sync(remove_security_signals) + ) + ) + steps.append( + _TeardownStep( + "engine.stop()", + _ENGINE_STOP_RESERVE_SECONDS, + lambda _b: engine.stop(), + bounded=False, + ) + ) + # B11: the harness-only instrumented executor (None in production and other tests). The + # engine is stopped, so no more to_thread work: a non-blocking shutdown is clean. + shim_executor = getattr(app.state, "connscale_executor", None) + if shim_executor is not None: + steps.append( + _TeardownStep( + "the instrumented executor's shutdown", + 0.05, + _sync(partial(shim_executor.shutdown, wait=False)), + ) + ) + if notifier is not None: + # Last, after the engine quiesces, so a final connection_stopped or queue_buildup + # alert still drains. + steps.append( + _TeardownStep("the alert notifier's close", 0.5, lambda _b: notifier.aclose()) + ) + await _run_teardown(steps, _SHUTDOWN_BUDGET_SECONDS) # Settings never select the open mode: only the caller's opt-in, checked at the top, does # (vault BACKLOG #2611, #2825). diff --git a/messagefoundry/api/outlive.py b/messagefoundry/api/outlive.py index b1b00a58d7..640574d8cb 100644 --- a/messagefoundry/api/outlive.py +++ b/messagefoundry/api/outlive.py @@ -38,17 +38,20 @@ log = logging.getLogger(__name__) #: How long :meth:`OutlivingOperations.drain` lets a running operation finish at shutdown before it -#: cancels it. It shares NSSM's 15 s graceful-stop window with the upload runner's stop and the -#: approval gate's drain, so it is short; an operation it cancels records ``interrupted``. +#: cancels it. It shares NSSM's 15 s graceful-stop window with the other shutdown steps, so it is +#: short; an operation it cancels records ``interrupted``. The API lifespans pass at most this, and +#: less when their shutdown budget is short (``_SHUTDOWN_BUDGET_SECONDS`` in ``api/app.py``). DRAIN_TIMEOUT_SECONDS = 0.5 #: How long the drain then waits for the cancelled operations' rollback arms to finish, so their #: interrupted rows reach the store before ``engine.stop()`` closes it. Longer than the drain itself #: because at least one arm waits on a process: a cancelled DR takeover hook is killed and reaped #: for up to ``_HOOK_REAP_SECONDS`` before the activation's arm writes its row, so this is that -#: bound plus a second for the write. Every shutdown bound together (the upload runner's stop, this -#: drain and grace, the approval gate's drain) comes close to NSSM's 15 s window only when all of -#: them are busy at once. +#: bound plus a second for the write. It is an upper limit: the API lifespans hand the drain a share +#: of one shutdown budget (``_SHUTDOWN_BUDGET_SECONDS`` in ``api/app.py``, BACKLOG #3191). With the +#: managed lifespan's full set of steps that share is below this, and below the reap alone, even +#: when they are idle, so at least a takeover hook that takes most of its reap can still be rolling +#: back when the store closes, and the drain logs it. CANCEL_GRACE_SECONDS = _HOOK_REAP_SECONDS + 1.0 @@ -131,7 +134,7 @@ async def drain( return [] cancelled = sorted(self._inflight.get(task, "?") for task in left) log.warning( - "shutdown: cancelling %d operation(s) still running after %.0fs: %s", + "shutdown: cancelling %d operation(s) still running after %.1fs: %s", len(left), timeout, ", ".join(cancelled), @@ -141,7 +144,7 @@ async def drain( _done, stuck = await asyncio.wait(left, timeout=grace) if stuck: log.error( - "shutdown: %d cancelled operation(s) had not finished rolling back after %.0fs", + "shutdown: %d cancelled operation(s) had not finished rolling back after %.1fs", len(stuck), grace, ) diff --git a/messagefoundry/store/base.py b/messagefoundry/store/base.py index a089041088..95ae096341 100644 --- a/messagefoundry/store/base.py +++ b/messagefoundry/store/base.py @@ -1513,7 +1513,8 @@ async def reserve_upload_quota( and its slot would otherwise consume the uploader's budget forever. A row with no activity for longer than ``stale_after`` seconds is reset to zero before the add. Activity is an applied reserve or any release, and both move the row's clock to now, never backwards, so a - host whose clock lags cannot age a row a sibling keeps fresh. A release of zero + host whose clock lags cannot age a row a sibling keeps fresh. Which clock "now" is, per + backend, is stated in ``messagefoundry.uploads.UploadQuotaError`` (BACKLOG #3256). A release of zero files and zero bytes is therefore a pure touch, which is how a live save keeps its slot past ``stale_after`` (BACKLOG #2648). So the reset normally reclaims only slots no live process holds. When it does reclaim a live one, it can only restore the pre-ledger diff --git a/messagefoundry/store/postgres.py b/messagefoundry/store/postgres.py index 3cc1d25830..58098fc177 100644 --- a/messagefoundry/store/postgres.py +++ b/messagefoundry/store/postgres.py @@ -332,6 +332,10 @@ _FIFO_HEADS_LANE_CHUNK = 500 # ADR 0066 §3.1: release_claimed id-chunk bound (ids per UPDATE statement). _RELEASE_CHUNK = 500 +# BACKLOG #3256: the database server's clock as epoch seconds, the unit `upload_quota.since` holds. +# statement_timestamp() is one value for the whole statement, so every reference agrees. The cluster +# lease reads clock_timestamp() inline (pipeline/cluster.py), which moves within a statement. +_PG_EPOCH_NOW = "EXTRACT(EPOCH FROM statement_timestamp())::double precision" # BACKLOG #1169: rows per batch for the `attachment_chunk` at-rest migration. Every OTHER cipher pass # batches 500 because its rows are kilobyte-shaped; one attachment_chunk row is a whole # DETACH_CHUNK_BYTES (1 MiB) slice, ~1.33 MiB once base64+GCM sealed. 16 keeps a batch near 21 MiB, @@ -8478,46 +8482,48 @@ async def reserve_upload_quota( One statement carries the budget predicate into the ``DO UPDATE``, so no other connection can land between the read and the write. ``RETURNING`` (absent on a refusal) discriminates applied - from refused.""" - now = time.time() + from refused. + + "Now" is the database server's clock, ``_PG_EPOCH_NOW`` (BACKLOG #3256; + ``uploads.UploadQuotaError`` says why).""" if files <= 0: # RELEASE — unconditional, clamped at zero (a double release cannot mint budget). await self._execute( "UPDATE upload_quota SET" " inflight_files = GREATEST(0, inflight_files + $2)," " inflight_bytes = GREATEST(0, inflight_bytes + $3)," - " since = GREATEST(since, $4)" # never backwards (BACKLOG #2648) + f" since = GREATEST(since, {_PG_EPOCH_NOW})" # never backwards (BACKLOG #2648) " WHERE uploader_id = $1", uploader_id, int(files), int(size_bytes), - now, ) return True if files > max_files or size_bytes > max_total_bytes: # Fail closed before touching the row: the insert branch applies unconditionally. return False - stale = now - max(0.0, stale_after) + stale = f"({_PG_EPOCH_NOW} - $4)" row = await self._fetchone( "INSERT INTO upload_quota (uploader_id, inflight_files, inflight_bytes, since)" - " VALUES ($1,$2,$3,$4)" + f" VALUES ($1,$2,$3,{_PG_EPOCH_NOW})" " ON CONFLICT (uploader_id) DO UPDATE SET" " inflight_files =" - " CASE WHEN upload_quota.since <= $5 THEN 0 ELSE upload_quota.inflight_files END + $2," + f" CASE WHEN upload_quota.since <= {stale} THEN 0" + " ELSE upload_quota.inflight_files END + $2," " inflight_bytes =" - " CASE WHEN upload_quota.since <= $5 THEN 0 ELSE upload_quota.inflight_bytes END + $3," + f" CASE WHEN upload_quota.since <= {stale} THEN 0" + " ELSE upload_quota.inflight_bytes END + $3," # Every applied reserve refreshes it, never backwards (BACKLOG #2648). - " since = GREATEST(upload_quota.since, $4)" - " WHERE (CASE WHEN upload_quota.since <= $5 THEN 0" - " ELSE upload_quota.inflight_files END) + $2 <= $6" - " AND (CASE WHEN upload_quota.since <= $5 THEN 0" - " ELSE upload_quota.inflight_bytes END) + $3 <= $7" + f" since = GREATEST(upload_quota.since, {_PG_EPOCH_NOW})" + f" WHERE (CASE WHEN upload_quota.since <= {stale} THEN 0" + " ELSE upload_quota.inflight_files END) + $2 <= $5" + f" AND (CASE WHEN upload_quota.since <= {stale} THEN 0" + " ELSE upload_quota.inflight_bytes END) + $3 <= $6" " RETURNING 1 AS applied", uploader_id, int(files), int(size_bytes), - now, - stale, + float(max(0.0, stale_after)), int(max_files), int(max_total_bytes), ) diff --git a/messagefoundry/store/sqlserver.py b/messagefoundry/store/sqlserver.py index de45f24c5b..c18d136065 100644 --- a/messagefoundry/store/sqlserver.py +++ b/messagefoundry/store/sqlserver.py @@ -326,6 +326,19 @@ _FIFO_HEADS_LANE_CHUNK = 500 # ADR 0066 §3.1: release_claimed id-chunk bound (ids per UPDATE statement). _RELEASE_CHUNK = 500 +# BACKLOG #3256: the database server's clock as epoch seconds, the unit `upload_quota.since` holds. +# Microseconds since the epoch fit a BIGINT for about 292,000 years. Each reference to +# SYSUTCDATETIME() is read on its own, so two references in one statement can disagree: two +# SYSDATETIME() calls in one INSERT ... VALUES row differed in 122 of 100,000 rows (Louis Davidson, +# "Row-wise system function calls, always the same?", Simple Talk, 2018). So each statement here +# references it once: the quota release reads it into a DECLARE @now, and the reserve MERGE names it +# once, in its one-row USING source. The cluster lease has its own copy at millisecond grain +# (pipeline/cluster_sqlserver.py `_DB_NOW`). Making this one the canonical copy for both is a +# pipeline change outside BACKLOG #3256. +_MSSQL_EPOCH_NOW = ( + "(CAST(DATEDIFF_BIG(MICROSECOND, CAST('1970-01-01' AS DATETIME2), SYSUTCDATETIME()) AS FLOAT)" + " / 1000000.0)" +) # BACKLOG #1169: rows per batch for the `attachment_chunk` at-rest migration. Every OTHER cipher pass # batches 500 because its rows are kilobyte-shaped; one attachment_chunk row is a whole # DETACH_CHUNK_BYTES (1 MiB) slice, ~1.33 MiB once base64+GCM sealed. 16 keeps a batch near 21 MiB, @@ -12114,23 +12127,27 @@ async def reserve_upload_quota( MERGE with HOLDLOCK is the SQL Server upsert that is safe under the concurrent opens of an engine-shard fleet (a bare IF EXISTS/INSERT races) — the same pattern :meth:`add_cipher_invocations` uses. The budget predicate rides ``WHEN MATCHED AND``, so a - refusal matches no rows and ``OUTPUT`` returns nothing.""" - now = time.time() + refusal matches no rows and ``OUTPUT`` returns nothing. + + "Now" is the database server's clock, ``_MSSQL_EPOCH_NOW``, read once: in the MERGE's + ``USING`` row, and into ``@now`` for the release (BACKLOG #3256; ``uploads.UploadQuotaError`` + says why).""" if files <= 0: # RELEASE — unconditional, clamped at zero (a double release cannot mint budget). await self._execute( - "UPDATE upload_quota SET" + # One read of the clock, so the CASE's test and its result are the same instant. + f"DECLARE @now FLOAT = {_MSSQL_EPOCH_NOW};" + " UPDATE upload_quota SET" " inflight_files = CASE WHEN inflight_files + ? < 0 THEN 0 ELSE inflight_files + ? END," " inflight_bytes = CASE WHEN inflight_bytes + ? < 0 THEN 0 ELSE inflight_bytes + ? END," - " since = CASE WHEN since > ? THEN since ELSE ? END" # never backwards (#2648) + # never backwards (#2648) + " since = CASE WHEN since > @now THEN since ELSE @now END" " WHERE uploader_id = ?", ( int(files), int(files), int(size_bytes), int(size_bytes), - now, - now, uploader_id, ), ) @@ -12138,13 +12155,14 @@ async def reserve_upload_quota( if files > max_files or size_bytes > max_total_bytes: # Fail closed before touching the row: WHEN NOT MATCHED inserts unconditionally. return False - stale = now - max(0.0, stale_after) async with self._acquire() as conn, self._cursor(conn) as cur: try: await cur.execute( "MERGE upload_quota WITH (HOLDLOCK) AS t" - " USING (SELECT ? AS uploader_id, ? AS files, ? AS size_bytes, ? AS now_ts," - " ? AS stale_ts, ? AS max_files, ? AS max_total_bytes) AS s" + " USING (SELECT ? AS uploader_id, ? AS files, ? AS size_bytes," + " n.now_ts AS now_ts, n.now_ts - ? AS stale_ts," + " ? AS max_files, ? AS max_total_bytes" + f" FROM (SELECT {_MSSQL_EPOCH_NOW} AS now_ts) AS n) AS s" " ON t.uploader_id = s.uploader_id" " WHEN MATCHED AND" " (CASE WHEN t.since <= s.stale_ts THEN 0 ELSE t.inflight_files END)" @@ -12167,8 +12185,7 @@ async def reserve_upload_quota( uploader_id, int(files), int(size_bytes), - now, - stale, + float(max(0.0, stale_after)), int(max_files), int(max_total_bytes), ), diff --git a/messagefoundry/uploads.py b/messagefoundry/uploads.py index 7dcc55dbda..0a52d175e9 100644 --- a/messagefoundry/uploads.py +++ b/messagefoundry/uploads.py @@ -91,7 +91,7 @@ # sits far outside any single write, which is bounded by ``max_bytes``. _ORPHAN_MIN_AGE_SECONDS = 3600.0 -# How long a cancelled save waits for each cross-shard ledger call, the reserve and the release, to +# How long a cancelled save waits for each engine-shard ledger call, the reserve and the release, to # finish (BACKLOG #2263). It is per call, so a save cancelled during a slow reserve can wait up to # two bounds. A healthy statement takes well under a second. The bound exists so a stuck store # cannot hold ``_quota_lock``, and the request's cancellation, indefinitely. Past it the call is @@ -99,17 +99,21 @@ # its audit rows after a cancelled upload (UploadStore._run_to_completion, BACKLOG #2261). _LEDGER_CANCEL_WAIT_SECONDS = 5.0 -# How often a save that holds a cross-shard reservation touches the ledger row, so a slow save keeps -# its slot past ``store.base.UPLOAD_RESERVATION_STALE_AFTER`` (300 s) instead of having it reclaimed -# by a sibling shard's reserve (BACKLOG #2648). A third of that window, which leaves room for touches -# that are slow as well as one that fails. Written out rather than imported to keep this module a -# leaf; a test pins the ratio. +# How often a save that holds an engine-shard reservation touches the ledger row, so a slow save +# keeps its slot past ``store.base.UPLOAD_RESERVATION_STALE_AFTER`` (300 s) instead of having it +# reclaimed by a sibling engine shard's reserve (BACKLOG #2648). A third of that window, which +# leaves room for touches that are slow as well as one that fails. Written out rather than imported +# to keep this module a leaf; a test pins the ratio. _RESERVATION_HEARTBEAT_SECONDS = 100.0 # How long a save keeps touching its row at most. A write that never finishes would otherwise pin # the row, and any leaked slot in it, for as long as the process lives. Past this the heartbeat # stops with a WARNING, and the row goes stale unless other activity keeps it fresh. A write is # bounded by ``max_upload_bytes`` (512 MiB at most), so an hour is far past any healthy save. _RESERVATION_HEARTBEAT_MAX_SECONDS = 3600.0 +# How long the heartbeat waits for one touch before it goes on without it (BACKLOG #3256). The +# touch is left running, never cancelled, so a slow touch that lands still refreshes the row. Well +# under the interval, so the wait never delays the next beat by much. +_RESERVATION_TOUCH_WAIT_SECONDS = 10.0 # How much of a refused plaintext sidecar the quota reads to find the uploader it bills (BACKLOG # #2322). A sidecar the engine wrote is a few hundred bytes. Past this the owner counts as unknown. @@ -243,11 +247,14 @@ class UploadQuotaError(UploadError): ``since`` to now, never backwards (BACKLOG #2648), and the row holds one clock for all of its slots. **So a leaked slot self-heals only once that uploader is otherwise IDLE** for the whole window, with no bound while activity continues. Any reserve, release or heartbeat for that - uploader, from any shard, keeps it alive. A host whose clock runs ahead holds it longer by the - skew, with no cap: the stamp never moves backwards, so one write from a host a year ahead - holds a slot leaked later for a year. A cap read from the writer's own clock cannot tell that - stamp from a fresh one seen by a host that lags, and the lagging host would then reset a - sibling's live slots, so the server-clock fix below is the one that closes it. + uploader, from any shard, keeps it alive. **On Postgres and SQL Server the stamp and the + staleness test both read the database server's clock (BACKLOG #3256),** so an engine host's + skew no longer moves either. A server clock stepped forward still holds a leaked slot longer by + the step, since the stamp never moves backwards, and so does a failover to a replica whose + clock lags the old primary's; a replica whose clock runs ahead can reclaim a live slot + early. So the database servers' own time sync now carries what each engine host's did. SQLite + reads the engine host's clock, and only one host ever writes a SQLite store, because + ``require_unified_store`` refuses SQLite for more than one engine shard. * **A reclaimed live reservation (BACKLOG #2648).** The store reclaims a row with no activity for ``UPLOAD_RESERVATION_STALE_AFTER``, and it cannot tell a dead holder from a slow one. Every applied reserve moves the row's clock, and a live save touches it every @@ -255,16 +262,14 @@ class UploadQuotaError(UploadError): keeps its slot. A live slot is still reclaimed in at least these cases: its holder lands no touch for the whole window while it is alive (a suspended process or VM, an event loop starved for minutes, or a store this shard cannot reach while a sibling can); its save - outlives ``_RESERVATION_HEARTBEAT_MAX_SECONDS``; or the holder's host clock lags a sibling's - by a large part of the window. ``since`` is stamped from the writer's ``time.time()`` and - compared against the reader's, and the stamp never moves backwards, so a lagging host can no - longer age a row a sibling keeps fresh, but a row only it touches still looks old to the - sibling. A sibling then counts neither that file nor its slot, and the holder's later release - erases a sibling's slot, so the undercount lasts until the ledger next drains to zero. That - restores the N-1 bound above while it lasts, never worse than the pre-#1112 behaviour. The - window is accepted. The skew case alone could be closed inside the existing statements, by - stamping and comparing with the database server's clock. The rest needs one ledger row per - reservation, a store-contract change across three backends. + outlives ``_RESERVATION_HEARTBEAT_MAX_SECONDS``; its touches stall on the store for the + whole window; or one touch never returns, since no second touch starts while one runs + (BACKLOG #3256). A sibling then counts neither that file nor its slot, and + the holder's later release erases a sibling's slot, so the undercount lasts until the ledger + next drains to zero. That restores the N-1 bound above while it lasts, never worse than the + pre-#1112 behaviour. The window is accepted. Host clock skew is not one of these cases on the + server backends (the bullet above says why). The rest needs one ledger row per reservation, + a store-contract change across three backends. The ledger is checked and paid back around the write, not in the same transaction as it, because the body lives on the filesystem rather than in the store. The ordering above is what stands in @@ -327,18 +332,26 @@ class UploadedFileMeta: @dataclass(frozen=True) class _UnsealableUpload: - """An upload ``rotate-key`` cannot seal, kept inside retention and the quota (BACKLOG #2322). + """One kind of upload ``rotate-key`` cannot seal, which the retention prune and the quota still + read (BACKLOG #2322). It is a plaintext sidecar that a keyed store refuses, over a body that is missing or is not - text. ``rotate-key`` leaves that sidecar unsealed, because sealing it would list an upload with - no usable body and would seal a planted lone sidecar too. So it is refused on every read, for - good, and nothing here ever serves or lists it. - - Nothing in the plaintext is trusted where trusting it could do harm. ``file_id`` comes from the - PATH, so a planted sidecar cannot name another upload. ``modified`` is the sidecar's mtime, which - is the expiry clock. ``size`` is the body's size on disk, never the sidecar's claim, so it cannot - be negative. ``uploader_id`` is read from the plaintext and used ONLY to bill the quota. A plant - can bill another uploader that way, which errs toward refusing an upload until the plant expires. + text (:meth:`UploadStore._unsealable_sync` is the test). ``rotate-key`` leaves that sidecar + unsealed, because sealing it would list an upload with no usable body and would seal a planted + lone sidecar too. So it is refused on every read, for good. The listing and the by-id reads never + serve it, and :meth:`UploadStore.delete` removes it. + + **Not every refused upload is one, so this class does not keep every refused upload inside + retention** (BACKLOG #3240). At least these are not: a blank sidecar, or one that is not UTF-8 + within the part :meth:`UploadStore._unsealable_sync` reads, and a plaintext sidecar over a body + that is oversized and not text, is not a regular file, or cannot be read for another cause. ``docs/PHI.md`` §3 names the ones outside retention. + + At least these fields are not taken from the plaintext. ``file_id`` comes from the PATH, so a + planted sidecar cannot name another upload. ``modified`` is the sidecar's mtime, which is the + expiry clock. ``size`` is the body's size on disk, never the sidecar's claim, so it cannot be + negative. ``uploader_id`` IS read from the plaintext, and is used here only to bill the quota. A + plant can bill another uploader that way, which errs toward refusing an upload until the plant + expires. """ file_id: str @@ -526,8 +539,11 @@ class ResealResult: unsealable: int = 0 #: How many UPLOADS had a plaintext half that this pass sealed (BACKLOG #1169). It counts files, #: not values, so it matches the count ``serve`` logs at startup (:meth:`UploadStore.warn_if_unsealed`). - #: A keyed store refuses those uploads until this pass seals them, so this is how many it turned - #: from refused into readable. + #: A keyed store refuses those uploads until this pass seals them. An upload also counted in + #: ``skipped`` counts here too when one half was sealed, since a planted body under a sealed + #: sidecar would otherwise be sealed without ever being counted (BACKLOG #3240). The cost is + #: that an upload whose two plaintext halves are sealed in two runs counts in both, so after a + #: skip the total across runs can exceed the startup count without a plant. sealed_plaintext: int = 0 @@ -628,6 +644,35 @@ def __init__( # Body name -> ((mtime_ns, size), decodes as text). Checking a body means reading it whole, # and every save's quota scan asks, so a body that has not changed is not read twice. self._body_checks: dict[str, tuple[tuple[int, int], bool]] = {} + # The running UploadRetentionRunner's wake, or None (BACKLOG #3184). See wake_retention. + self._retention_wake: Callable[[], bool] | None = None + + def wake_retention(self) -> bool: + """Ask the running :class:`UploadRetentionRunner` over this store to sweep now, and return + whether one is running (BACKLOG #3184). + + The upload route calls this after a save, in place of sweeping on the request path, so a + large prune never adds to an upload's latency or meets its deadline. The runner's sweep is + audited like any other. It returns at once: the sweep runs in the runner's own task, and a + wake during a sweep starts one more sweep after it, never two at once. ``False`` means no + runner is running, as in an app built with ``create_app`` alone, and the caller sweeps + itself (:meth:`prune_on_save`). The cost: the sweep now finishes after the response, so an + upload sent straight after it can still be billed for expired files not yet removed. That + errs toward refusing, and clears once the sweep lands.""" + wake = self._retention_wake + return wake is not None and wake() + + def bind_retention_wake(self, wake: Callable[[], bool]) -> None: + """Make ``wake`` what :meth:`wake_retention` calls. It returns whether its runner's loop is + still running, so a loop that ended without a stop sends the caller back to sweeping for + itself. :meth:`UploadRetentionRunner.start` binds it. One runner per store: a second + binding replaces the first.""" + self._retention_wake = wake + + def unbind_retention_wake(self, wake: Callable[[], bool]) -> None: + """Undo :meth:`bind_retention_wake` for ``wake``, and leave any other binding alone.""" + if self._retention_wake == wake: + self._retention_wake = None @property def max_bytes(self) -> int: @@ -687,6 +732,14 @@ def _passes_unmarked(self) -> bool: its reason are stated.""" return not isinstance(self._cipher, AesGcmCipher) + @property + def _refuses_plaintext(self) -> bool: + """True when this surface refuses every plaintext sidecar, which is the only place an upload + ``rotate-key`` cannot seal exists (:class:`_UnsealableUpload`). Under + ``[store].allow_unmarked_ciphertext`` the listing scan reads those sidecars itself.""" + cipher = self._cipher + return isinstance(cipher, AesGcmCipher) and not cipher.allow_unmarked + def _decrypt_blob(self, stored: str, file_id: str) -> bytes: b64 = self._cipher.decrypt( stored, @@ -740,7 +793,9 @@ def _iter_sidecars(self) -> Iterator[tuple[str, Path]]: if _FILE_ID_RE.match(fid): yield fid, entry - def _scan_metas_sync(self) -> list[UploadedFileMeta]: + def _scan_metas_sync( + self, held: list[_UnsealableUpload] | None = None + ) -> list[UploadedFileMeta]: """Walk the uploads root and decrypt every well-formed ``.meta`` sidecar (UNSORTED). A bad/foreign/undecryptable sidecar is skipped **with its cause named** (never a body in the log), so a rotated-away key can neither sink the listing nor silently drop a quota/retention @@ -767,15 +822,31 @@ def _scan_metas_sync(self) -> list[UploadedFileMeta]: residual for all of that. The residual is named in ``docs/PHI.md`` §3: a refused upload is outside retention and the quota until ``rotate-key`` seals it, and ``serve`` logs the count at startup so the operator knows to run it. One kind ``rotate-key`` can never seal is - brought back by :meth:`_scan_unsealable_sync` instead, without trusting its sidecar. At least - a blank or non-UTF-8 sidecar is still left out. + brought back through ``held`` instead, without trusting its sidecar. At least a blank + sidecar, or one that is not UTF-8 within the part :meth:`_unsealable_sync` reads, is still + left out. + + **``held`` is for the quota and the prune only (BACKLOG #2322, #3184).** When it is a list + and the cipher refuses plaintext, each refused sidecar that is an upload ``rotate-key`` + cannot seal is appended to it, from this same directory walk, so a quota check walks the + directory once rather than twice. Nothing that lists or serves an upload passes it, so the + strict read is not loosened. Only a sidecar the cipher refused, or one that is not UTF-8 + as a whole, is classed; the second covers a plaintext sidecar that is not UTF-8 only past + the part :meth:`_unsealable_sync` reads. A sidecar the cipher decrypted is sealed, and + :meth:`_unsealable_sync` would decline it anyway. The class is keyed by the PATH id, never + by the id in a sidecar's JSON. The cipher's own message is safe to log — every ``CipherError`` carries only key ids, marker versions and algorithm names, never a decrypted value. The malformed-shape branch logs only the exception TYPE, because a ``ValueError`` from coercing a metadata field can echo that field's value back into the message.""" out: list[UploadedFileMeta] = [] + sink = held if self._refuses_plaintext else None + bodies: set[str] = set() for fid, entry in self._iter_sidecars(): + if sink is not None: + bodies.add(f"{fid}{_BLOB_SUFFIX}") + classify = False try: out.append(self._decrypt_meta(entry.read_text(encoding="utf-8"), fid)) except CipherError as exc: @@ -783,45 +854,30 @@ def _scan_metas_sync(self) -> list[UploadedFileMeta]: # another cell, or the refusal of an unmarked (legacy, planted or downgraded) # sidecar. Named on its own so these can be told apart from a damaged file. _log.warning("uploaded-file sidecar %s: cipher declined it: %s", fid, exc) + classify = True except (OSError, UnicodeDecodeError) as exc: # The bytes never reached the cipher — unreadable file, or not valid UTF-8. _log.warning( "uploaded-file sidecar %s: unreadable on disk: %s", fid, codec_safe_str(exc) ) + # A decode error is classed too: a plaintext sidecar that is not UTF-8 only past + # the part the class check reads is still one rotate-key can never seal. An + # OSError is not, so a stalled read is not paid twice under _quota_lock. + classify = isinstance(exc, UnicodeDecodeError) except (ValueError, TypeError, KeyError) as exc: # Decrypted, but the JSON is malformed or a field will not coerce. Type only. _log.warning( "uploaded-file sidecar %s: malformed metadata (%s)", fid, type(exc).__name__ ) - return out - - def _scan_unsealable_sync(self) -> list[_UnsealableUpload]: - """Every upload ``rotate-key`` cannot seal that the listing scan refuses (BACKLOG #2322). - - :meth:`_scan_metas_sync` drops each refused sidecar. Most wait for ``rotate-key``, but this - kind never stops being refused, so without this pass it would sit outside retention and the - quota forever, with a body that may hold message content. Only the quota and the prune read - this list. Nothing that lists or serves an upload does, so the strict read is not loosened. - - Empty unless the cipher refuses plaintext. Under ``[store].allow_unmarked_ciphertext`` the - listing scan reads these sidecars itself, and counting them here too would bill them twice. - It opens every sidecar again rather than skip the ids that scan decrypted: a decrypted id is - the one in the sidecar's JSON, and a sealed plant can name another upload's id there. Sync: - the caller runs it off the event loop.""" - cipher = self._cipher - if not isinstance(cipher, AesGcmCipher) or cipher.allow_unmarked: - return [] - out: list[_UnsealableUpload] = [] - bodies: set[str] = set() - for fid, sidecar in self._iter_sidecars(): - bodies.add(f"{fid}{_BLOB_SUFFIX}") - held = self._unsealable_sync(fid, sidecar) - if held is not None: - out.append(held) - # Forget the checks of uploads that are gone, so the cache cannot outgrow the directory. A - # copy, because a scan on another worker thread may be writing to it. - for name in self._body_checks.copy().keys() - bodies: - self._body_checks.pop(name, None) + if classify and sink is not None: + lost = self._unsealable_sync(fid, entry) + if lost is not None: + sink.append(lost) + if sink is not None: + # Forget the checks of uploads that are gone, so the cache cannot outgrow the directory. + # A copy, because a scan on another worker thread may be writing to it. + for name in self._body_checks.copy().keys() - bodies: + self._body_checks.pop(name, None) return out def _unsealable_sync(self, fid: str, sidecar: Path) -> _UnsealableUpload | None: @@ -1079,7 +1135,7 @@ def _discard_cancelled_sync(self, file_id: str) -> None: ) async def _reserve_across_shards(self, *, uploader_id: str, uploader: str, size: int) -> bool: - """Take this uploader's cross-shard in-flight reservation; return whether one is held. + """Take this uploader's engine-shard in-flight reservation; return whether one is held. ``False`` means there is no ledger bound (the store-less construction path) — not that the reservation was refused. A refusal raises :class:`UploadQuotaError`, the same TYPE the @@ -1094,11 +1150,12 @@ async def _reserve_across_shards(self, *, uploader_id: str, uploader: str, size: two same-moment uploads be refused at the ledger, rather than both reserving and each then refusing the other. - That makes two sidecar scans per save, this one and the one in ``_build_and_write``. Each - reads and decrypts EVERY sidecar in the directory, for every uploader, then filters, and walks - it once more for an upload ``rotate-key`` cannot seal (BACKLOG #2322). Both - run under the per-process ``_quota_lock``. They run off the event loop, on the operator - diagnostic surface rather than the data plane. + That makes two sidecar scans per save, this one and the one in ``_build_and_write``, and the + ordering argument needs both: this one is before the reserve and that one after the ledger + read. Each is ONE walk of the directory (BACKLOG #3184) that reads and decrypts EVERY sidecar, + for every uploader, then filters, and classes each refused one for an upload ``rotate-key`` + cannot seal (BACKLOG #2322). Both run under the per-process ``_quota_lock``. They run off the + event loop, on the operator diagnostic surface rather than the data plane. An uploader already over budget on disk is refused here, with the on-disk wording, before the ledger is consulted. Otherwise a zero-headroom ledger would refuse first, and its message would @@ -1148,7 +1205,7 @@ def _reserve_took_a_slot(reserve: asyncio.Future[bool], uploader_id: str) -> boo committed would subtract a sibling shard's live slot and let an upload past the cap.""" if reserve.cancelled(): _log.warning( - "the cross-shard upload reserve for %s was cancelled before it finished; a slot " + "the engine-shard upload reserve for %s was cancelled before it finished; a slot " "it took is not paid back", uploader_id, ) @@ -1156,7 +1213,7 @@ def _reserve_took_a_slot(reserve: asyncio.Future[bool], uploader_id: str) -> boo exc = reserve.exception() if exc is not None: _log.warning( - "the cross-shard upload reserve for %s failed while the upload was being " + "the engine-shard upload reserve for %s failed while the upload was being " "cancelled; a slot it took is not paid back", uploader_id, exc_info=exc, @@ -1169,34 +1226,70 @@ def _start_heartbeat(self, uploader_id: str, stop: asyncio.Event) -> None: set, so a slow save keeps its slot (BACKLOG #2648; :class:`UploadQuotaError` states the residual). The touch is a release of zero files and zero bytes: it changes no count and only moves the row's clock. So a touch that lands after the save's release is harmless, and - ``stop`` is a signal, never a wait. A failed touch is logged and the next one tried.""" + ``stop`` is a signal, never a wait. A failed touch is logged and the next one tried. + + **The wait for a touch is bounded, and a touch is never cancelled (BACKLOG #3256).** The + heartbeat waits ``_RESERVATION_TOUCH_WAIT_SECONDS`` for it, then goes on, so a touch stuck + on the store no longer holds the heartbeat or its lifetime cap, and a ``stop`` during the + wait ends it at once. The stuck touch is left to land on its own, which still moves the + row's clock when it does, and no second touch starts while it runs, so a stalled store never + collects one abandoned statement per interval. The cost: a touch that never returns ends + the refreshing for that save, which :class:`UploadQuotaError` lists.""" ledger = self._ledger if ledger is None: return + def _missed(touch: asyncio.Future[Any]) -> None: + if not touch.cancelled() and touch.exception() is not None: + _log.warning( + "could not refresh the engine-shard upload reservation for %s; it may be " + "reclaimed if the upload outlasts the staleness window", + uploader_id, + exc_info=touch.exception(), + ) + async def _beat() -> None: deadline = time.monotonic() + _RESERVATION_HEARTBEAT_MAX_SECONDS - while not stop.is_set(): - try: - await asyncio.wait_for(stop.wait(), _RESERVATION_HEARTBEAT_SECONDS) - except TimeoutError: - if time.monotonic() >= deadline: - _log.warning( - "an upload by %s is still writing after %gs; its cross-shard " - "reservation is no longer refreshed and may be reclaimed", - uploader_id, - _RESERVATION_HEARTBEAT_MAX_SECONDS, - ) - return + touch: asyncio.Future[bool] | None = None + try: + while not stop.is_set(): try: - await ledger.reserve_upload_quota(uploader_id, files=0, size_bytes=0) - except Exception: # noqa: BLE001 — a missed touch must not fail the upload - _log.warning( - "could not refresh the cross-shard upload reservation for %s; it may " - "be reclaimed if the upload outlasts the staleness window", - uploader_id, - exc_info=True, + await asyncio.wait_for(stop.wait(), _RESERVATION_HEARTBEAT_SECONDS) + except TimeoutError: + if time.monotonic() >= deadline: + _log.warning( + "an upload by %s is still writing after %gs; its engine-shard " + "reservation is no longer refreshed and may be reclaimed", + uploader_id, + _RESERVATION_HEARTBEAT_MAX_SECONDS, + ) + return + if touch is not None and not touch.done(): + continue # the last touch is still running; it is not doubled + touch = asyncio.ensure_future( + ledger.reserve_upload_quota(uploader_id, files=0, size_bytes=0) ) + # On the touch itself, so a failure that lands after the wait is logged. + touch.add_done_callback(_missed) + stopped = asyncio.ensure_future(stop.wait()) + try: + await asyncio.wait( + {touch, stopped}, + timeout=_RESERVATION_TOUCH_WAIT_SECONDS, + return_when=asyncio.FIRST_COMPLETED, + ) + finally: + stopped.cancel() + if not touch.done() and not stop.is_set(): + _log.warning( + "an engine-shard upload reservation touch for %s did not finish " + "within %gs; it was left running, and the next waits for it", + uploader_id, + _RESERVATION_TOUCH_WAIT_SECONDS, + ) + finally: + if touch is not None and not touch.done(): + self._hold(touch) # still referenced; _missed still logs its outcome task = asyncio.ensure_future(_beat()) self._heartbeats.add(task) @@ -1209,7 +1302,7 @@ def _leave_running( A reserve that then turns out to hold a slot is paid back by a fresh release.""" _log.warning( - "the cross-shard upload %s for %s did not finish within %gs of a cancellation; it " + "the engine-shard upload %s for %s did not finish within %gs of a cancellation; it " "was left running", what, uploader_id, @@ -1226,7 +1319,7 @@ def _done(done: asyncio.Future[Any]) -> None: ) elif done.cancelled(): _log.warning( - "the cross-shard upload release for %s was cancelled before it finished; " + "the engine-shard upload release for %s was cancelled before it finished; " "its slot is not paid back", uploader_id, ) @@ -1268,7 +1361,7 @@ def _on_disk_refusal( """The per-uploader quota verdict against what is ALREADY on disk, or ``None`` if it fits. One string, two callers: the under-lock check inside ``save``'s build-and-write, and the - cross-shard reservation's pre-check. The bucket key is the IMMUTABLE ``uploader_id`` (the + engine-shard reservation's pre-check. The bucket key is the IMMUTABLE ``uploader_id`` (the caller filters on it, the same value the ownership check uses), so the budget and the ownership rule can never disagree about who a file belongs to and a recycled username is never billed for files it cannot read. The message names the human username, because an @@ -1315,7 +1408,7 @@ async def _release() -> None: ) except Exception: # noqa: BLE001 — a release failure must not fail an upload that landed _log.warning( - "could not release the cross-shard upload reservation for %s; it will be " + "could not release the engine-shard upload reservation for %s; it will be " "reclaimed when it goes stale", uploader_id, exc_info=True, @@ -1334,10 +1427,12 @@ def _observed_sync(self, uploader_id: str) -> tuple[int, int]: the budget. Sync: the caller runs it off the event loop. An upload ``rotate-key`` cannot seal counts too, at its body's size on disk (BACKLOG #2322). - Its owner is unverified, so this errs toward refusing; :class:`_UnsealableUpload` says why.""" - metas = self._scan_metas_sync() + Its owner is unverified, so this errs toward refusing; :class:`_UnsealableUpload` says why. + Both kinds come from one walk of the directory (BACKLOG #3184).""" + held: list[_UnsealableUpload] = [] + metas = self._scan_metas_sync(held) sizes = [m.size for m in metas if m.uploader_id == uploader_id] - sizes += [h.size for h in self._scan_unsealable_sync() if h.uploader_id == uploader_id] + sizes += [h.size for h in held if h.uploader_id == uploader_id] return len(sizes), sum(sizes) async def list_files(self) -> list[UploadedFileMeta]: @@ -1397,7 +1492,12 @@ async def delete(self, file_id: str, *, sidecar_refused: bool = False) -> Upload ``sidecar_refused`` says the caller's own :meth:`get_meta` was just refused. The sidecar is then not decrypted again, so one delete raises one cipher WARNING and one alert, not two. - Only the never-sealable kind is removed either way.""" + It does not widen what is removed: of the uploads whose sidecar is refused, only the + never-sealable kind is removed, whether the refusal was the caller's read or this call's own. + The flag does not authorize anything either. The route decides who may pass it. + + A sidecar that is gone by the time it is classed answers :class:`UploadNotFoundError`, as an + absent one does (BACKLOG #3240). It is checked again only after the class is refused.""" def _delete() -> UploadedFileMeta: blob_path, meta_path = self._paths(file_id) @@ -1405,14 +1505,9 @@ def _delete() -> UploadedFileMeta: refusal = _REFUSED_NOT_UNSEALABLE if sidecar_refused: # The store checks the caller's claim: only a cipher that refuses plaintext can - # hold the never-sealable kind, as in _scan_unsealable_sync. - cipher = self._cipher - if not isinstance(cipher, AesGcmCipher) or cipher.allow_unmarked: + # hold the never-sealable kind. + if not self._refuses_plaintext: raise UploadUnreadableError(refusal) - try: - meta_path.stat() - except FileNotFoundError as exc: - raise UploadNotFoundError(file_id) from exc else: try: meta = self._decrypt_meta(meta_path.read_text(encoding="utf-8"), file_id) @@ -1423,6 +1518,14 @@ def _delete() -> UploadedFileMeta: if meta is None: held = self._unsealable_sync(file_id, meta_path) if held is None: + # _unsealable_sync declines a sidecar it cannot open, so tell an absent one, + # whether it was never there or went since the read, from a refused one. + # Only FileNotFoundError means absent. Any other OSError propagates, as it does + # from the read above, and is never answered as a missing file. + try: + meta_path.lstat() + except FileNotFoundError as exc: + raise UploadNotFoundError(file_id) from exc raise UploadUnreadableError(refusal) meta = held.as_meta() # Remove the body first, then the sidecar (best-effort on the body — the sidecar is the @@ -1570,10 +1673,11 @@ def _reseal_to_active_sync(self) -> ResealResult: # one is read, roughly doubling peak memory for no reason. del stored resealed += 1 - # Once per pair, whichever half failed. A skipped pair is not readable yet, so its - # sealed plaintext half did not turn it "from refused into readable". + # Once per pair, whichever half failed. A skipped pair counts in sealed_plaintext too + # when it had a plaintext half sealed (BACKLOG #3240): its sidecar may be sealed + # already, and then no later run sees any plaintext in it to count. skipped += pair_skipped - sealed_plaintext += had_plaintext and not pair_skipped + sealed_plaintext += had_plaintext if resealed or skipped or unsealable: _log.info( "re-sealed %d uploaded-file value(s) under the active key, sealing %d plaintext " @@ -1674,6 +1778,7 @@ async def prune_expired( retention_days: int | None = None, abort: threading.Event | None = None, tally: _SweepTally | None = None, + audit: Callable[[UploadedFileMeta], Awaitable[None]] | None = None, ) -> PruneResult: """Age-based retention sweep (ASVS 5.2.4): delete every (blob, meta) pair whose ``uploaded_at`` is older than ``retention_days`` (default: the configured ``retention_days``), then sweep the write @@ -1685,7 +1790,8 @@ async def prune_expired( destroy data). One exception is an upload ``rotate-key`` can never seal (BACKLOG #2322): it is pruned once its sidecar's mtime passes the cutoff, and is reported with blank owner fields (:class:`_UnsealableUpload`). Runs off the event loop; the periodic runner + the - opportunistic save-time sweep both drive it. + opportunistic save-time sweep both drive it. A save wakes the runner, and sweeps itself only + where none is running (:meth:`wake_retention`). A pair is reported only when this pass removed its body, the PHI the audit row is about. A refused body unlink leaves the whole pair for the next pass; a refused sidecar unlink leaves @@ -1697,7 +1803,16 @@ async def prune_expired( A pair is never half-deleted by an abort. ``tally`` records, as the pass goes, each pair it removes and the pair it is removing. The - retention runner uses it to name the files a timed-out stop loses (:class:`_SweepTally`).""" + retention runner uses it to name the files a timed-out stop loses (:class:`_SweepTally`). + + **``audit`` is called for each pair right after its removal, before the next pair is + touched (BACKLOG #3184).** The pass used to delete every due pair, then sweep the orphans, + and only then write the rows, so a stop that timed out mid-pass lost the row of every pair + removed so far. Now a timed-out stop loses at most the pair in progress and its row. Each + pair is removed in its own worker-thread call, so the rows are written on the event loop + between them, and a cancellation stops the pass at the pair in progress. A failed ``audit`` + call is logged and the pass goes on, so one bad row never costs the others. ``tally`` + counts each call once it returns or raises.""" days = self._retention_days if retention_days is None else max(1, int(retention_days)) at = time.time() if now is None else now cutoff = at - days * _SECONDS_PER_DAY @@ -1705,39 +1820,57 @@ async def prune_expired( stop = abort if abort is not None else threading.Event() sweep = tally if tally is not None else _SweepTally() - def _prune() -> PruneResult: + def _due() -> list[UploadedFileMeta]: # Checked before the scan too: the scan decrypts every sidecar, which is wasted on a stop. - metas = [] if stop.is_set() else self._scan_metas_sync() - due = [meta for meta in metas if meta.uploaded_at < cutoff] - # An upload rotate-key cannot seal ages by its sidecar's mtime (BACKLOG #2322). - if not stop.is_set(): - due += [ - held.as_meta() - for held in self._scan_unsealable_sync() - if held.expired(cutoff, at) - ] - for meta in due: - if not sweep.begin(meta, stop): - break - removed = False - try: - removed = self._prune_pair_sync(meta) - finally: - sweep.end(removed) - pruned = list(sweep.removed) if stop.is_set(): - return PruneResult(pruned=pruned) - # The orphan sweep runs AFTER the deletions, so a failure in it must not raise out of - # here: that would drop `pruned`, and with it the audit rows (BACKLOG #2261). + return [] + held: list[_UnsealableUpload] = [] + metas = self._scan_metas_sync(held) + # An upload rotate-key cannot seal ages by its sidecar's mtime (BACKLOG #2322). + return [meta for meta in metas if meta.uploaded_at < cutoff] + [ + lost.as_meta() for lost in held if lost.expired(cutoff, at) + ] + + def _remove(meta: UploadedFileMeta) -> bool | None: + """Remove one due pair; ``None`` when the pass was stopped before it.""" + if not sweep.begin(meta, stop): + return None + removed = False try: - orphans = self._sweep_orphans_sync(now=at, abort=stop) - except Exception: # noqa: BLE001 — logged; a raise would drop the pass's audit rows - # ERROR, as the runner logged this before the catch moved here. + removed = self._prune_pair_sync(meta) + finally: + sweep.end(removed) + return removed + + def _orphans() -> int: + # A failure here must not raise out of the pass: that would drop `pruned` (BACKLOG + # #2261). ERROR, as the runner logged this before the catch moved here. + try: + return self._sweep_orphans_sync(now=at, abort=stop) + except Exception: # noqa: BLE001 — logged; a raise would drop the pass's result _log.exception("uploaded-logs orphan sweep failed; will retry next pass") - orphans = 0 - return PruneResult(pruned=pruned, orphans_removed=orphans) + return 0 - return await asyncio.to_thread(_prune) + for meta in await asyncio.to_thread(_due): + removed = await asyncio.to_thread(_remove, meta) + if removed is None: + break + if removed and audit is not None: + try: + await audit(meta) + except Exception: # noqa: BLE001 — logged; one bad row must not cost the rest + # ERROR: the file is gone and this was its only record, the level every other + # upload.prune audit gap logs at. + _log.exception( + "uploaded-logs prune audit failed for %s; it has no upload.prune audit row", + meta.file_id, + ) + sweep.audited += 1 + pruned = list(sweep.removed) + if stop.is_set(): + return PruneResult(pruned=pruned) + # The orphan sweep runs AFTER the deletions and their rows. + return PruneResult(pruned=pruned, orphans_removed=await asyncio.to_thread(_orphans)) def _prune_pair_sync(self, meta: UploadedFileMeta) -> bool: """Remove one aged pair for :meth:`prune_expired`; True when THIS call removed its body. @@ -1783,40 +1916,12 @@ def _prune_pair_sync(self, meta: UploadedFileMeta) -> bool: ) return True - async def prune_and_audit( - self, - audit: Callable[[UploadedFileMeta], Awaitable[None]] | None, - *, - now: float | None = None, - abort: threading.Event | None = None, - tally: _SweepTally | None = None, - ) -> PruneResult: - """One :meth:`prune_expired` pass, then one ``audit`` call per pruned file. - - The shared body of the retention runner and the save-time sweep. A failed audit call is - logged and the rest still run, so one bad row never costs the others. ``tally`` is - :meth:`prune_expired`'s, and also counts each audit call once it returns or raises.""" - result = await self.prune_expired(now=now, abort=abort, tally=tally) - if audit is not None: - for meta in result.pruned: - try: - await audit(meta) - except Exception: # noqa: BLE001 — logged; one bad row must not cost the rest - # ERROR: the file is gone and this was its only record, the level every other - # upload.prune audit gap logs at. - _log.exception( - "uploaded-logs prune audit failed for %s; it has no upload.prune audit row", - meta.file_id, - ) - if tally is not None: - tally.audited += 1 - return result - async def prune_on_save( self, audit: Callable[[UploadedFileMeta], Awaitable[None]] ) -> PruneResult: - """The save-time sweep: :meth:`prune_and_audit` that a cancelled caller cannot cut short - (BACKLOG #2261). + """The save-time sweep: an audited :meth:`prune_expired` that a cancelled caller cannot cut + short (BACKLOG #2261). The upload route runs it only where no retention runner is running + to wake instead (:meth:`wake_retention`, BACKLOG #3184). The request deadline cancels the upload route, and a cancel used to drop the result naming what the sweep's thread had deleted, so no ``upload.prune`` row was written for those files. @@ -1844,7 +1949,7 @@ def _late(task: asyncio.Future[Any]) -> None: ) return await self._run_to_completion( - self.prune_and_audit(audit, abort=abort, tally=tally), + self.prune_expired(audit=audit, abort=abort, tally=tally), what="the save-time upload prune", on_cancel=lambda: tally.abort(abort), on_late=_late, @@ -1918,9 +2023,10 @@ def _sweep_orphans_sync(self, *, now: float, abort: threading.Event | None = Non published half-written, nothing is left behind, and the operator retries. An hour against a 25 MiB default bound is far outside that window. - ``abort`` is :meth:`prune_expired`'s, checked before each entry. The pass's audit rows are - written only once this returns, so a slow sweep over a large dir must not hold them past a - stop (BACKLOG #2264). An aborted sweep returns what it removed so far; the next one resumes.""" + ``abort`` is :meth:`prune_expired`'s, checked before each entry, so a slow sweep over a large + dir does not hold a stop past its bound (BACKLOG #2264). The pass's audit rows are already + written when this runs (BACKLOG #3184). An aborted sweep returns what it removed so far; the + next one resumes.""" stop = abort if abort is not None else threading.Event() root = self._root if stop.is_set() or not root.is_dir(): @@ -1969,21 +2075,24 @@ def _sweep_orphans_sync(self, *, now: float, abort: threading.Event | None = Non return removed -# One prune sweep per hour is ample for a day-granularity retention window (the opportunistic save-time -# sweep covers the between-ticks case), and it keeps the background wakeups negligible. +# One prune sweep per hour is ample for a day-granularity retention window (a save wakes the runner +# for the between-ticks case, UploadStore.wake_retention), and it keeps the background wakeups +# negligible. _DEFAULT_PRUNE_INTERVAL_SECONDS = 3600.0 # How long ``UploadRetentionRunner.stop`` waits for a sweep already in flight to finish and write its # audit rows. The sweep stops at its next file once asked, so this bounds a hung filesystem call, a # sidecar scan already under way, and the audit writes for what the sweep had deleted. The runner -# stops FIRST in the API lifespan's teardown, and engine.stop() still has to run after it. +# stops FIRST in the API lifespan's teardown, and engine.stop() still has to run after it. It is an +# upper limit: that teardown passes ``stop(timeout=...)`` a share of the step's bound in one shutdown +# budget (``_SHUTDOWN_BUDGET_SECONDS`` in ``api/app.py``, BACKLOG #3191), which is shorter. # # It shares NSSM's 15 s graceful-stop window (AppStopMethodConsole in # scripts/service/install-service.ps1) with uvicorn's connection drain, which runs BEFORE the -# lifespan teardown. ``serve`` passes uvicorn no ``timeout_graceful_shutdown``, so that drain has no -# bound: a request still open at Ctrl+C can use up the window before this wait starts. A shorter -# bound here would not fix that, and would cost audit rows on a slow disk, so 5 s stands. The fix is -# to bound the drain where ``serve`` calls ``uvicorn.run``. +# lifespan teardown. ``serve`` bounds that drain with ``timeout_graceful_shutdown`` +# (``__main__._UVICORN_DRAIN_SECONDS``, BACKLOG #3256), so a request still open at Ctrl+C can no +# longer use up the window before this wait starts. Since the sweep writes each row as it goes +# (BACKLOG #3184), a stop that runs out loses at most the pair in progress. _DEFAULT_STOP_TIMEOUT_SECONDS = 5.0 @@ -1995,7 +2104,7 @@ class _SweepTally: ``removed`` lists each pair the sweep's thread removed, in the order it is audited, and ``removing`` is the pair it is removing now. ``audited`` counts audit calls that returned or - raised; a raised call is counted because :meth:`UploadStore.prune_and_audit` already logs it. + raised; a raised call is counted because :meth:`UploadStore.prune_expired` already logs it. So ``unaudited()`` names every removed file whose row has not been written. It also names the pair in progress, which may turn out not to be removed, and a row whose interrupted write may still land, so it can name more than are lost, never fewer. @@ -2062,48 +2171,58 @@ def __init__( self._clock = clock self._stop_timeout = float(stop_timeout_seconds) self._stop = asyncio.Event() + # Set by stop() and by wake(): either ends the wait between sweeps (BACKLOG #3184). + self._kick = asyncio.Event() # The sweep's worker thread reads this one; an asyncio.Event is not thread-safe. self._abort = threading.Event() self._task: asyncio.Task[None] | None = None self._tally = _SweepTally() # the sweep in flight, or the last one def start(self) -> None: - """Spawn the supervised prune loop (idempotent).""" + """Spawn the supervised prune loop (idempotent), and bind it as the store's + :meth:`UploadStore.wake_retention`.""" if self._task is not None: return # A fresh event, not a clear(), for the same reason as _abort below: a loop a stop() is # still waiting on holds the old one, and clearing it would keep that loop sweeping beside # the new one, each replacing the tally the other's stop() reads. self._stop = asyncio.Event() + self._kick = asyncio.Event() # A fresh event, not a clear(): a sweep thread a timed-out stop() left running still holds # the old one, and clearing it would let that thread resume deleting with nobody to audit. self._abort = threading.Event() - self._task = asyncio.create_task(self._run(self._stop)) + self._task = asyncio.create_task(self._run(self._stop, self._kick)) + self._store.bind_retention_wake(self.wake) _log.info( "uploaded-logs retention prune enabled: older than %d days, every %gs", self._store.retention_days, self._interval, ) - async def stop(self) -> None: + async def stop(self, timeout: float | None = None) -> None: """Signal the loop and await its exit, audit rows included (idempotent). **The loop is asked to stop, not cancelled (BACKLOG #2065).** A sweep deletes files in a worker thread, and cancelling the task that awaits it leaves the thread deleting while the result naming those files is dropped, so no ``upload.prune`` row is ever written for them. Instead the sweep is told to stop at its next file, and this waits for it to return and for - its audit rows to land, up to ``stop_timeout_seconds``. + its audit rows to land, up to ``stop_timeout_seconds``, or ``timeout`` when that is shorter. + A caller with a shutdown budget of its own passes ``timeout``, so this wait ends, and cancels + the sweep, before the caller cuts it. A sweep can outlast that bound: one stuck filesystem call, the sidecar scan or orphan sweep - already under way, or the audit writes for a long list of deletions. The task is then - cancelled. When the sweep had started deleting a file it had not yet audited, that loss is - logged as an ERROR. It covers every such file, not only one whose unlink is still running. - A sweep that had no such file loses no row, so it is logged as a WARNING instead. + already under way, or one stuck audit write. The task is then cancelled. The sweep writes + each pair's row before it touches the next pair (BACKLOG #3184), so at most the pair in + progress has no row. When the sweep had started deleting a file it had not yet audited, that + loss is logged as an ERROR, naming the file. A sweep that had no such file loses no row, so + it is logged as a WARNING instead. This never raises into the caller's teardown. If the caller is itself cancelled while waiting, the sweep is cancelled the same way and this returns, as it did before #2065, so the API lifespan's later steps, ``engine.stop()`` among them, still run.""" self._stop.set() + self._kick.set() + self._store.unbind_retention_wake(self.wake) # an upload sweeps for itself again task = self._task self._task = None if task is None: @@ -2114,12 +2233,13 @@ async def stop(self) -> None: tally = self._tally tally.abort(self._abort) self._abort = threading.Event() + wait = self._stop_timeout if timeout is None else min(self._stop_timeout, timeout) try: # shield: a timeout must not cancel the task mid-audit; the branches below decide that. - await asyncio.wait_for(asyncio.shield(task), self._stop_timeout) + await asyncio.wait_for(asyncio.shield(task), wait) return except TimeoutError: - why = f"did not finish within {self._stop_timeout:g}s of shutdown" + why = f"did not finish within {wait:g}s of shutdown" except asyncio.CancelledError: why = "was still running when shutdown itself was cancelled" # Read before the cancel. Every id is logged, since these lines are the only record of them. @@ -2135,30 +2255,44 @@ async def stop(self) -> None: with contextlib.suppress(asyncio.CancelledError): await task - async def _run(self, stop: asyncio.Event) -> None: - # One isolated sweep per interval; an error in a pass is logged and the loop continues (a prune - # must never take the engine down). Cooperatively cancellable via ``stop``, this loop's own. + def wake(self) -> bool: + """Start the next sweep now rather than at the end of the interval (BACKLOG #3184), and + return whether the loop is running. A wake during a sweep starts one more sweep after it. + ``False``, doing nothing, when the loop is not running, including one that ended without a + :meth:`stop`.""" + task = self._task + if task is None or task.done(): + return False + self._kick.set() + return True + + async def _run(self, stop: asyncio.Event, kick: asyncio.Event) -> None: + # One isolated sweep per interval or wake; an error in a pass is logged and the loop + # continues (a prune must never take the engine down). Cooperatively cancellable via + # ``stop``, this loop's own, which stop() pairs with a kick so the wait ends at once. while not stop.is_set(): + # Cleared before the sweep, so a wake that lands during it is kept for the next one. + kick.clear() try: # The loop's sweep alone owns self._tally, so a direct run_once() never hides it. self._tally = _SweepTally() await self._sweep(self._tally) except Exception: _log.exception("uploaded-logs retention prune failed; will retry next interval") - try: # noqa: SIM105 — wake immediately on stop so shutdown isn't held by the interval - await asyncio.wait_for(stop.wait(), self._interval) + try: # noqa: SIM105 — a stop or a wake ends the wait; only the interval times out + await asyncio.wait_for(kick.wait(), self._interval) except TimeoutError: pass async def run_once(self, now: float | None = None) -> PruneResult: - """One :meth:`UploadStore.prune_and_audit` pass for ``now`` (default: the injected clock). - The pass's orphan count rides back in the result; it is logged by the sweep and carries no - metadata to audit (see :class:`PruneResult`).""" + """One audited :meth:`UploadStore.prune_expired` pass for ``now`` (default: the injected + clock). The pass's orphan count rides back in the result; it is logged by the sweep and + carries no metadata to audit (see :class:`PruneResult`).""" return await self._sweep(None, now) async def _sweep(self, tally: _SweepTally | None, now: float | None = None) -> PruneResult: - return await self._store.prune_and_audit( - self._audit, + return await self._store.prune_expired( + audit=self._audit, now=self._clock() if now is None else now, abort=self._abort, tally=tally, @@ -2185,7 +2319,7 @@ async def _to_thread_to_completion[T, A](func: Callable[[A], T], arg: A) -> T: """``asyncio.to_thread``, except that a cancellation waits for the thread to finish first. A bare ``to_thread`` re-raises a cancellation at once while its thread keeps running. ``save`` - then pays its cross-shard reservation back and drops ``_quota_lock`` while the file is still + then pays its engine-shard reservation back and drops ``_quota_lock`` while the file is still being written, so a sibling can read the ledger and scan the disk and count neither. That breaks the premise :class:`UploadQuotaError`'s ordering argument rests on. The save's thread runs the quota's sidecar scan and then the write. The write is bounded by ``max_bytes``. The scan is @@ -2253,7 +2387,7 @@ async def _wait_to_completion[T]( ) -> T: """Await ``fut``; on a cancellation, wait for it to finish, then raise ``CancelledError``. - The shared body of :func:`_to_thread_to_completion` and the cross-shard ledger calls (BACKLOG + The shared body of :func:`_to_thread_to_completion` and the engine-shard ledger calls (BACKLOG #2263). Further cancellations while waiting are absorbed. A ledger call is a coroutine, so it runs as a Task, and ``asyncio.run``'s shutdown cancellation can still end it early. That leaves the reservation as a killed process would. diff --git a/tests/test_api_tls.py b/tests/test_api_tls.py index 4ba3fb9ade..e29e910f16 100644 --- a/tests/test_api_tls.py +++ b/tests/test_api_tls.py @@ -994,6 +994,10 @@ def test_serve_forwarded_allow_ips_empty_when_no_proxy( assert captured["forwarded_allow_ips"] == [] # WP-L3-07 (ASVS 13.4.6): the `Server: uvicorn` banner is suppressed. assert captured["server_header"] is False + # BACKLOG #3256: the connection drain before the lifespan teardown is bounded. + from messagefoundry.__main__ import _UVICORN_DRAIN_SECONDS + + assert captured["timeout_graceful_shutdown"] == _UVICORN_DRAIN_SECONDS # --- #200: Posture-B (upstream TLS termination) declared-floor validation -------------------- diff --git a/tests/test_lifespan_teardown.py b/tests/test_lifespan_teardown.py new file mode 100644 index 0000000000..0fdcc0a576 --- /dev/null +++ b/tests/test_lifespan_teardown.py @@ -0,0 +1,489 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +# Copyright (C) 2026 MessageFoundry Foundation, LLC and contributors +"""BACKLOG #3191: the API lifespans' shutdown runs every step whatever the steps before it did. + +Three findings from engine PR 2115. A cancel delivered mid-teardown skipped the drains after the one +it interrupted. engine.stop() and the security notifier's close sat unguarded, so a raise in one +skipped the rest. And the lifespan create_app(engine=...) installs drained only the approval gate. +The whole shutdown also has to fit NSSM's graceful-stop window. +""" + +from __future__ import annotations + +import asyncio +import logging +import re +import time +from collections.abc import AsyncIterator, Awaitable, Callable +from pathlib import Path +from typing import Any + +import pytest + +from messagefoundry.api import app as app_module +from messagefoundry.api import create_app, create_managed_app +from messagefoundry.api.app import _run_teardown, _TeardownStep +from messagefoundry.config.settings import ApprovalsSettings, EgressSettings +from messagefoundry.pipeline.engine import Engine + +_WAIT_S = 10.0 +_REPO = Path(__file__).resolve().parents[1] + + +async def _hang(_bound: float) -> None: + await asyncio.Event().wait() + + +def _record(calls: list[str], name: str) -> Callable[[float], Awaitable[None]]: + async def _step(_bound: float) -> None: + calls.append(name) + + return _step + + +class _Fatal(BaseException): + """A BaseException a task carries as its result, unlike KeyboardInterrupt, which asyncio + re-raises out of the event loop itself.""" + + +# --- _run_teardown, the step runner ------------------------------------------------------------- + + +async def test_a_step_that_raises_does_not_skip_the_next() -> None: + calls: list[str] = [] + + async def _boom(_bound: float) -> None: + calls.append("boom") + raise RuntimeError("PROBE: deliberate step failure") + + await _run_teardown( + [_TeardownStep("boom", 0.1, _boom), _TeardownStep("next", 0.1, _record(calls, "next"))], + 1.0, + ) + assert calls == ["boom", "next"] + + +async def test_a_step_that_raises_while_it_is_built_does_not_skip_the_next() -> None: + calls: list[str] = [] + + def _raises_at_once(_bound: float) -> Awaitable[None]: + raise AttributeError("PROBE: an attribute the step reads is gone") + + await _run_teardown( + [ + _TeardownStep("raises at once", 0.1, _raises_at_once), + _TeardownStep("stop", 0.1, _record(calls, "stop"), bounded=False), + ], + 1.0, + ) + assert calls == ["stop"] + + +async def test_a_base_exception_runs_the_later_steps_then_is_raised() -> None: + calls: list[str] = [] + + async def _fatal(_bound: float) -> None: + raise _Fatal + + with pytest.raises(_Fatal): + await _run_teardown( + [ + _TeardownStep("fatal", 0.1, _fatal), + _TeardownStep("stop", 0.1, _record(calls, "stop"), bounded=False), + ], + 1.0, + ) + assert calls == ["stop"] + + +async def test_a_cancel_reaches_no_step_and_propagates_after_the_last() -> None: + """The step running when the cancel lands finishes its own work, the later steps run, and only + then is the cancel raised.""" + calls: list[str] = [] + entered, release = asyncio.Event(), asyncio.Event() + + async def _first(_bound: float) -> None: + entered.set() + await release.wait() + calls.append("first finished") + + task = asyncio.create_task( + _run_teardown( + [ + _TeardownStep("first", 0.1, _first), + _TeardownStep("second", 0.1, _record(calls, "second")), + _TeardownStep("third", 0.1, _record(calls, "third")), + ], + # Long enough that the bound cannot end the first step during the test. + 60.0, + ) + ) + await asyncio.wait_for(entered.wait(), _WAIT_S) + task.cancel() + await asyncio.sleep(0.05) + assert not task.done(), "the cancel cut the running step short" + release.set() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, _WAIT_S) + assert calls == ["first finished", "second", "third"] + + +async def test_a_step_that_overruns_is_cut_and_every_later_step_keeps_its_floor() -> None: + bounds: dict[str, float] = {} + + async def _hangs(bound: float) -> None: + bounds["hangs"] = bound + await asyncio.Event().wait() + + async def _later(bound: float) -> None: + bounds["later"] = bound + + started = time.monotonic() + await _run_teardown( + [_TeardownStep("hangs", 0.02, _hangs), _TeardownStep("later", 0.1, _later)], 0.3 + ) + elapsed = time.monotonic() - started + # The hanging step got the budget less the later floor, not the whole budget. + assert bounds["hangs"] == pytest.approx(0.2, abs=0.05) + assert bounds["later"] >= 0.1 + assert elapsed < 0.3 + app_module._CUT_GRACE_SECONDS + 0.25 + + +async def test_a_step_that_ignores_its_cut_is_left_running( + caplog: pytest.LogCaptureFixture, +) -> None: + """The bound is hard: a step that swallows its cancel and goes on waiting is left behind after + the cut grace, and the teardown moves on. How it later ends is still logged.""" + calls: list[str] = [] + release = asyncio.Event() + + async def _stubborn(_bound: float) -> None: + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + await release.wait() + raise RuntimeError("PROBE: failed after the teardown moved on") from None + + started = time.monotonic() + with caplog.at_level(logging.ERROR, logger="messagefoundry.api.app"): + await _run_teardown( + [ + _TeardownStep("stubborn", 0.05, _stubborn), + _TeardownStep("after", 0.05, _record(calls, "after")), + ], + 0.2, + ) + elapsed = time.monotonic() - started + assert calls == ["after"] + assert elapsed < 0.2 + app_module._CUT_GRACE_SECONDS + 0.25 + release.set() + for _ in range(50): + if any("after the teardown moved on" in r.getMessage() for r in caplog.records): + break + await asyncio.sleep(0.01) + assert any("after the teardown moved on" in r.getMessage() for r in caplog.records) + + +async def test_an_unbounded_step_runs_to_completion_through_a_cancel() -> None: + calls: list[str] = [] + entered, release = asyncio.Event(), asyncio.Event() + + async def _stop(_bound: float) -> None: + entered.set() + await release.wait() + calls.append("stop finished") + + task = asyncio.create_task( + _run_teardown( + [ + _TeardownStep("stop", 0.1, _stop, bounded=False), + _TeardownStep("after", 0.1, _record(calls, "after")), + ], + # Shorter than the stop takes: an unbounded step is never cut by the budget either. + 0.01, + ) + ) + await asyncio.wait_for(entered.wait(), _WAIT_S) + task.cancel() + await asyncio.sleep(0.05) + assert not task.done(), "the unbounded step was cut short" + release.set() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, _WAIT_S) + assert calls == ["stop finished", "after"] + + +async def test_a_step_that_ends_cancelled_on_its_own_is_not_taken_for_a_cancel() -> None: + """A CancelledError nobody asked for is a step failure: raising it at the end would cancel the + caller's shutdown for no reason.""" + calls: list[str] = [] + + async def _cancels_itself(_bound: float) -> None: + inner = asyncio.ensure_future(asyncio.sleep(10)) + inner.cancel() + await inner + + # Returns rather than raising: nobody cancelled the caller. + await _run_teardown( + [ + _TeardownStep("cancels itself", 0.1, _cancels_itself), + _TeardownStep("after", 0.1, _record(calls, "after")), + ], + 1.0, + ) + assert calls == ["after"] + + +def test_the_shutdown_budget_fits_the_nssm_stop_window() -> None: + """The install script's AppStopMethodConsole is the window. The budget, plus uvicorn's own drain + bound where serve sets one (BACKLOG #3256), must leave room in it for the interpreter to exit.""" + script = (_REPO / "scripts" / "service" / "install-service.ps1").read_text(encoding="utf-8") + match = re.search(r"AppStopMethodConsole\s+(\d+)", script) + assert match is not None, "the install script no longer sets AppStopMethodConsole" + window = int(match.group(1)) / 1000.0 + import messagefoundry.__main__ as cli + + shutdown = app_module._SHUTDOWN_BUDGET_SECONDS + cli._UVICORN_DRAIN_SECONDS + assert shutdown + 1.0 <= window + assert app_module._ENGINE_STOP_RESERVE_SECONDS < app_module._SHUTDOWN_BUDGET_SECONDS + + +# --- the managed lifespan ----------------------------------------------------------------------- + + +def _managed(db_path: Path) -> Any: + return create_managed_app( + db_path=db_path, egress_settings=EgressSettings(deny_by_default=False) + ) + + +class _Spies: + """Swap each teardown step the managed app runs for a recorder, keeping the real engine.stop() + so the test can still stop the engine if the teardown never did.""" + + def __init__(self) -> None: + self.calls: list[str] = [] + self.real_stop: Callable[[], Awaitable[None]] | None = None + + def attach(self, app: Any) -> None: + calls = self.calls + engine = app.state.engine + real_stop = self.real_stop = engine.stop + + async def _outliving(timeout: float = 0.0, *, grace: float = 0.0) -> list[str]: + calls.append("outliving") + return [] + + async def _approvals(timeout: float = 0.0) -> list[str]: + calls.append("approvals") + return [] + + async def _flush(store: Any) -> None: + calls.append("flush") + + async def _stop() -> None: + calls.append("stop") + await real_stop() + + class _Executor: + def shutdown(self, wait: bool) -> None: + calls.append("executor") + + app.state.outliving_operations.drain = _outliving + app.state.approval_gate.drain = _approvals + app.state.summary_auditor.flush = _flush + app.state.connscale_executor = _Executor() + engine.stop = _stop + + async def stop_if_needed(self) -> None: + if self.real_stop is not None and "stop" not in self.calls: + await self.real_stop() + + +async def test_a_cancel_during_the_first_drain_still_runs_every_later_step(tmp_path: Path) -> None: + app = _managed(tmp_path / "cancel.db") + spies = _Spies() + entered, leave, draining, release = (asyncio.Event() for _ in range(4)) + + async def _held(timeout: float = 0.0, *, grace: float = 0.0) -> list[str]: + draining.set() + await release.wait() + spies.calls.append("outliving finished") + return [] + + async def _run() -> None: + async with app.router.lifespan_context(app): + spies.attach(app) + app.state.outliving_operations.drain = _held + entered.set() + await leave.wait() + + task = asyncio.create_task(_run()) + try: + await asyncio.wait_for(entered.wait(), _WAIT_S) + leave.set() + await asyncio.wait_for(draining.wait(), _WAIT_S) + task.cancel() + await asyncio.sleep(0.05) + release.set() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, _WAIT_S) + finally: + await spies.stop_if_needed() + assert spies.calls == ["outliving finished", "approvals", "flush", "stop", "executor"] + + +async def test_an_engine_stop_that_raises_does_not_skip_the_steps_after_it( + tmp_path: Path, +) -> None: + app = _managed(tmp_path / "stop-raises.db") + spies = _Spies() + try: + async with app.router.lifespan_context(app): + spies.attach(app) + real_stop = spies.real_stop + assert real_stop is not None + + async def _stop_raises() -> None: + spies.calls.append("stop") + await real_stop() + raise RuntimeError("PROBE: deliberate engine.stop() failure") + + app.state.engine.stop = _stop_raises + finally: + await spies.stop_if_needed() + assert spies.calls == ["outliving", "approvals", "flush", "stop", "executor"] + + +async def test_the_retention_runner_s_stop_waits_less_than_its_step_s_bound( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The runner's own stop wait is longer than its teardown step's bound. The lifespan hands it a + share of that bound, so its wait ends, and it cancels its own sweep, before the step is cut.""" + pytest.importorskip("psutil") + from messagefoundry import uploads as uploads_module + from messagefoundry.config.settings import StoreSettings + from messagefoundry.uploads import UploadRetentionRunner + + bounds: list[float] = [] + timeouts: list[float | None] = [] + real_invoke, real_stop = app_module._invoke, UploadRetentionRunner.stop + + async def _invoke(run: Callable[[float], Awaitable[object]], bound: float) -> object: + bounds.append(bound) + return await real_invoke(run, bound) + + async def _stop(self: UploadRetentionRunner, timeout: float | None = None) -> None: + timeouts.append(timeout) + await real_stop(self, timeout) + + monkeypatch.setattr(app_module, "_invoke", _invoke) + monkeypatch.setattr(UploadRetentionRunner, "stop", _stop) + settings = StoreSettings(path=str(tmp_path / "runner.db"), uploads_dir=str(tmp_path / "up")) + app = create_managed_app( + store_settings=settings, egress_settings=EgressSettings(deny_by_default=False) + ) + async with app.router.lifespan_context(app): + assert app.state.upload_store is not None + # The runner's stop is the first step. + [timeout] = timeouts + assert timeout is not None + assert timeout == pytest.approx(bounds[0] * app_module._INNER_SHARE) + assert timeout < bounds[0] + # The premise: even with fewer steps than serve runs, the share is the shorter of the two waits. + assert timeout < uploads_module._DEFAULT_STOP_TIMEOUT_SECONDS + + +async def test_the_teardown_is_bounded_when_every_drain_hangs( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Every drain before engine.stop() hangs, and the whole teardown still ends within the budget, + the cut graces and what the real engine.stop() takes. The budget just covers the floors of the + steps that apply here, as the shipped one covers the managed set's.""" + budget = 3.5 + monkeypatch.setattr(app_module, "_SHUTDOWN_BUDGET_SECONDS", budget) + monkeypatch.setattr(app_module, "_ENGINE_STOP_RESERVE_SECONDS", 0.2) + app = _managed(tmp_path / "bounded.db") + spies = _Spies() + stop_took: list[float] = [] + try: + async with app.router.lifespan_context(app): + spies.attach(app) + real_stop = spies.real_stop + assert real_stop is not None + + async def _hangs(*_a: Any, **_k: Any) -> list[str]: + await asyncio.Event().wait() + return [] + + async def _timed_stop() -> None: + spies.calls.append("stop") + began = time.monotonic() + await real_stop() + stop_took.append(time.monotonic() - began) + + app.state.outliving_operations.drain = _hangs + app.state.approval_gate.drain = _hangs + app.state.summary_auditor.flush = _hangs + app.state.engine.stop = _timed_stop + started = time.monotonic() + elapsed = time.monotonic() - started + finally: + await spies.stop_if_needed() + assert spies.calls == ["stop", "executor"] + graces = 3 * app_module._CUT_GRACE_SECONDS + assert elapsed < budget + graces + stop_took[0] + 0.5, f"teardown took {elapsed:.2f}s" + + +# --- the embedded lifespan (create_app(engine=...)) --------------------------------------------- + + +@pytest.fixture +async def engine(tmp_path: Path) -> AsyncIterator[Engine]: + eng = await Engine.create( + tmp_path / "embedded.db", + poll_interval=0.02, + egress_settings=EgressSettings(deny_by_default=False), + ) + yield eng + await eng.stop() + + +_ON = ApprovalsSettings(enabled=True, operations=["dead_letter_replay"], min_dwell_seconds=0.0) + + +async def test_the_embedded_app_runs_the_same_drains_at_shutdown(engine: Engine) -> None: + app = create_app(engine, approvals=_ON) + calls: list[str] = [] + + async def _outliving(timeout: float = 0.0, *, grace: float = 0.0) -> list[str]: + calls.append("outliving") + return [] + + async def _approvals(timeout: float = 0.0) -> list[str]: + calls.append("approvals") + return [] + + class _Auth: + """Joined, never closed: the caller owns the service.""" + + async def drain_background(self) -> None: + calls.append("lock notices") + + async with app.router.lifespan_context(app): + app.state.outliving_operations.drain = _outliving + app.state.approval_gate.drain = _approvals + app.state.auth = _Auth() + assert calls == ["outliving", "approvals", "lock notices"] + + +async def test_the_embedded_app_flushes_the_open_summary_window_before_its_engine_stops( + engine: Engine, +) -> None: + """The row lands before the app's shutdown returns, so before the caller stops its engine.""" + app = create_app(engine, approvals=_ON) + async with app.router.lifespan_context(app): + await app.state.summary_auditor.note(engine.store, "checker", None, 3, time.time()) + assert await engine.store.list_audit(action="summary_access") == [] + rows = await engine.store.list_audit(action="summary_access") + assert len(rows) == 1 diff --git a/tests/test_upload_api.py b/tests/test_upload_api.py index 4fbbd60974..8043f9cfeb 100644 --- a/tests/test_upload_api.py +++ b/tests/test_upload_api.py @@ -39,7 +39,7 @@ ) from messagefoundry.pipeline import Engine from messagefoundry.store import MessageStatus -from messagefoundry.store.crypto import generate_key, make_cipher +from messagefoundry.store.crypto import AesGcmCipher, generate_key, make_cipher from messagefoundry.store.store import AuditAppend from messagefoundry.uploads import UploadStore from tests._admin_account import create_local_user_chosen @@ -1057,6 +1057,65 @@ def _paths(file_id: str) -> tuple[Path, Path]: assert len(deleted) == 2, f"expected the sweep to stop after its second pair, got {deleted}" +async def test_an_upload_wakes_the_retention_runner_instead_of_sweeping_inline( + engine: Engine, tmp_path: Path +) -> None: + """BACKLOG #3184 item 4. Where a retention runner is running, the upload route wakes it and + returns, so the sweep is never on the request path or under its deadline. The runner's sweep + still prunes the aged file and writes its row. Red before the fix: the route swept inline.""" + import dataclasses + import time + + pytest.importorskip("psutil") + from messagefoundry.api import create_app + from messagefoundry.api.app import _record_upload_prune + from messagefoundry.uploads import UploadedFileMeta, UploadRetentionRunner + + service = await _make_user(engine, Role.OPERATOR, name="op") + app = create_app(engine, auth=service, store_settings=_uploads_settings(tmp_path)) + us = app.state.upload_store + inline: list[object] = [] + + async def _inline(audit: Any) -> Any: + inline.append(audit) + raise AssertionError("the route swept on the request path") + + us.prune_on_save = _inline + + async def _audit(meta: UploadedFileMeta) -> None: + await _record_upload_prune(engine.store, meta) + + runner = UploadRetentionRunner(us, audit=_audit, interval_seconds=3600) + transport = reproving_transport(PW, app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://t") as c: + h = await _login(c, "op") + r = await c.post("/uploads", files={"file": ("a.hl7", BATCH, "text/plain")}, headers=h) + assert r.status_code == 200, r.text + assert inline, "control: with no runner running, the route sweeps inline" + inline.clear() + aged = r.json()["file_id"] + meta = next(m for m in await us.list_files() if m.file_id == aged) + (tmp_path / "uploads" / f"{aged}.meta").write_text( + us._encrypt_meta(dataclasses.replace(meta, uploaded_at=time.time() - 31 * 86_400)), + encoding="utf-8", + ) + runner.start() + try: + # The runner's first sweep may already have taken the file; either way the upload must + # not sweep inline, and the row must land from the runner. + r = await c.post("/uploads", files={"file": ("b.hl7", BATCH, "text/plain")}, headers=h) + assert r.status_code == 200, r.text + rows: list[Any] = [] + deadline = time.monotonic() + 10.0 + while not rows and time.monotonic() < deadline: + await asyncio.sleep(0.02) + rows = list(await engine.store.list_audit(action="upload.prune")) + finally: + await runner.stop() + assert not inline, "the route swept inline although a runner was running" + assert [json.loads(str(a["detail"]))["file_id"] for a in rows] == [aged] + + async def test_a_prune_failure_of_any_kind_never_fails_the_upload( engine: Engine, tmp_path: Path ) -> None: @@ -1333,7 +1392,13 @@ async def test_an_override_holder_deletes_an_upload_rotate_key_cannot_seal( blob, sidecar = root / f"{fid}.blob", root / f"{fid}.meta" break_body(blob) had_body = blob.exists() - app.state.upload_store = _keyed_store(root) # the key is now enabled + # The key is now enabled. The refusal hook is what raises the upload-cipher alert in the + # engine, so counting its calls counts the alerts (BACKLOG #3240). + cipher = make_cipher(generate_key(), write_v2=True) + assert isinstance(cipher, AesGcmCipher) + alerts: list[tuple[str, str]] = [] + cipher.set_refusal_hook(lambda table, column: alerts.append((table, column))) + app.state.upload_store = UploadStore(root, cipher, max_bytes=10**6) r = await c.delete(f"/uploads/{fid}", headers=h) assert r.status_code == 404, r.text # the owner is unreadable: no existence oracle @@ -1342,12 +1407,14 @@ async def test_an_override_holder_deletes_an_upload_rotate_key_cannot_seal( ) caplog.clear() + alerts.clear() with caplog.at_level(logging.WARNING, logger="messagefoundry.store.crypto"): r = await c.delete(f"/uploads/{fid}", headers=ha) assert r.status_code == 200, r.text assert r.json() == {"file_id": fid, "filename": "", "deleted": True} refusals = [m for m in caplog.messages if m.startswith("refused an unmarked value")] assert len(refusals) == 1, refusals + assert alerts == [("uploaded_file", "meta")], alerts assert not sidecar.exists() and not blob.exists(), "the delete left a file behind" rows = [a for a in await engine.store.list_audit() if a["action"] == "upload.delete"] diff --git a/tests/test_uploads.py b/tests/test_uploads.py index 6b068ea5f8..607a801e51 100644 --- a/tests/test_uploads.py +++ b/tests/test_uploads.py @@ -30,7 +30,6 @@ UploadRetentionRunner, UploadStore, UploadTooLargeError, - _SweepTally, sanitize_filename, validate_upload_content, ) @@ -349,8 +348,8 @@ async def test_concurrent_uploads_cannot_double_book_the_quota( real_scan = store._scan_metas_sync - def _slow_scan() -> list[UploadedFileMeta]: - out = real_scan() + def _slow_scan(held: list[Any] | None = None) -> list[UploadedFileMeta]: + out = real_scan(held) time.sleep(0.05) # widen the window so an unlocked check-then-write WOULD lose the race return out @@ -522,20 +521,24 @@ async def _stop_stuck_runner( release: threading.Event, caplog: pytest.LogCaptureFixture, audit: Callable[[UploadedFileMeta], Awaitable[None]] | None = None, + *, + stop_timeout_seconds: float = 0.05, + timeout: float | None = None, ) -> logging.LogRecord: - """Start a runner over ``store``, wait for its sweep to park on ``paused``, stop it with a - 0.05 s bound, release the sweep, and return the one record ``stop()`` logged about it.""" + """Start a runner over ``store``, wait for its sweep to park on ``paused``, stop it, release the + sweep, and return the one record ``stop()`` logged about it. The runner's own bound is + ``stop_timeout_seconds``, and ``timeout`` is the caller's, passed to ``stop()``.""" runner = UploadRetentionRunner( store, audit=audit or _audited_to([]), clock=lambda: time.time() + 31 * 86_400, - stop_timeout_seconds=0.05, + stop_timeout_seconds=stop_timeout_seconds, ) runner.start() try: assert await asyncio.to_thread(paused.wait, 10), "the sweep never reached its park point" with caplog.at_level(logging.WARNING, logger="messagefoundry.uploads"): - await asyncio.wait_for(runner.stop(), 10) + await asyncio.wait_for(runner.stop(timeout=timeout), 10) finally: release.set() await runner.stop() # a no-op once stop() has run; stops the loop if an assert failed @@ -549,16 +552,130 @@ async def test_a_sweep_stuck_past_the_stop_bound_is_cancelled_and_logged( """BACKLOG #2065, the bound. A sweep stuck inside one filesystem call cannot be stopped, so ``stop()`` waits only ``stop_timeout_seconds`` and then names the audit gap at ERROR rather than holding shutdown open forever or returning in silence. The sweep is parked on its SECOND pair, - past that pair's abort check, so two pairs go with no audit row and the ERROR counts both.""" + past that pair's abort check. The first pair's row was written before the second pair was + touched (BACKLOG #3184), so only the pair in progress is named. Before #3184 the rows waited + for the whole pass, and the ERROR counted both.""" store = _quota_store(tmp_path, retention_days=30) for i in range(2): await store.save( data=f"aging {i}\n".encode(), filename=f"f{i}.txt", uploader="op", uploader_id="u-op" ) - record = await _stop_stuck_runner(store, *_pause_sweep_at_pair(store, 2), caplog) + audited: list[str] = [] + record = await _stop_stuck_runner( + store, *_pause_sweep_at_pair(store, 2), caplog, audit=_audited_to(audited) + ) assert record.levelno == logging.ERROR, record.getMessage() assert "did not finish within 0.05s of shutdown" in record.getMessage() - assert "2 file(s) it removed, or was removing, may have no" in record.getMessage() + assert "1 file(s) it removed, or was removing, may have no" in record.getMessage() + [named] = record.getMessage().rsplit(": ", 1)[1].split(", ") + assert len(audited) == 1 and audited[0] != named + + +async def test_a_stop_timeout_shorter_than_the_runner_s_own_bound_is_the_one_waited( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """The API lifespan passes ``stop`` a share of its teardown step's bound, which is shorter than + the runner's own. The runner's wait must end first, so it names the audit gap and cancels the + sweep itself, rather than being cut by the teardown and logging that shutdown was cancelled.""" + store = _quota_store(tmp_path, retention_days=30) + for i in range(2): + await store.save( + data=f"aging {i}\n".encode(), filename=f"f{i}.txt", uploader="op", uploader_id="u-op" + ) + # The helper's 10 s wait_for would time out on a stop that waited the runner's own 60 s. + record = await _stop_stuck_runner( + store, *_pause_sweep_at_pair(store, 2), caplog, stop_timeout_seconds=60.0, timeout=0.05 + ) + assert "did not finish within 0.05s of shutdown" in record.getMessage() + + +async def test_a_timed_out_stop_keeps_the_row_of_every_pair_already_removed( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """BACKLOG #3184 item 3. The sweep audited only after its whole pass, so a stop that timed out + mid-pass lost the row of every pair it had removed. Parked on its THIRD pair, the sweep has + removed two, and both rows are written: only the pair in progress is unaudited. Red before the + fix: no row at all, and the ERROR named all three.""" + store = _quota_store(tmp_path, retention_days=30) + for i in range(4): + await store.save( + data=f"aging {i}\n".encode(), filename=f"f{i}.txt", uploader="op", uploader_id="u-op" + ) + audited: list[str] = [] + record = await _stop_stuck_runner( + store, *_pause_sweep_at_pair(store, 3), caplog, audit=_audited_to(audited) + ) + assert record.levelno == logging.ERROR, record.getMessage() + assert "1 file(s) it removed, or was removing, may have no" in record.getMessage() + [named] = record.getMessage().rsplit(": ", 1)[1].split(", ") + assert len(audited) == 2 and named not in audited + # The parked thread finishes the pair it is on once released; the fourth pair is never touched. + for _ in range(200): + if len(await store.list_files()) == 1: + break + await asyncio.sleep(0.01) + assert len(await store.list_files()) == 1 + + +async def test_a_quota_check_walks_the_uploads_directory_once(tmp_path: Path) -> None: + """BACKLOG #3184 item 1. The quota scan walked the directory twice, once to decrypt the sidecars + and again to find an upload ``rotate-key`` cannot seal. It is one walk now, and it still finds + that kind: a plaintext sidecar over a missing body, on a store that refuses plaintext.""" + root = tmp_path / "uploads" + plain = await UploadStore(root, make_cipher(None), max_bytes=4096).save( + data=b"legacy\n", filename="a.txt", uploader="op", uploader_id="u-op" + ) + (root / f"{plain.file_id}.blob").unlink() + store = _quota_store(tmp_path) + await store.save(data=b"sealed\n", filename="b.txt", uploader="op", uploader_id="u-op") + real_iter = store._iter_sidecars + walks = 0 + + def _counting_iter() -> Any: + nonlocal walks + walks += 1 + return real_iter() + + store._iter_sidecars = _counting_iter # type: ignore[method-assign] + assert store._observed_sync("u-op") == (2, len(b"sealed\n")) + assert walks == 1 + + +async def test_a_save_wakes_the_running_retention_runner(tmp_path: Path) -> None: + """BACKLOG #3184 item 4. ``wake_retention`` reports whether a runner is running and starts its + next sweep at once, rather than at the end of the interval. A stopped runner unbinds, so a + save then sweeps for itself.""" + store = _quota_store(tmp_path, retention_days=30) + assert store.wake_retention() is False + audited: list[str] = [] + now = [time.time()] + runner = UploadRetentionRunner( + store, audit=_audited_to(audited), clock=lambda: now[0], interval_seconds=3600 + ) + runner.start() + try: + meta = await store.save( + data=b"aging\n", filename="a.txt", uploader="op", uploader_id="u-op" + ) + now[0] += 31 * 86_400 # the next sweep finds the file expired + assert store.wake_retention() is True + for _ in range(500): + if audited: + break + await asyncio.sleep(0.01) + assert audited == [meta.file_id], "the wake did not start a sweep inside the interval" + finally: + await runner.stop() + assert store.wake_retention() is False + + # A loop that ended without stop() is not reported as running, so a save sweeps for itself. + runner.start() + assert runner._task is not None + runner._task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await runner._task + assert store.wake_retention() is False + await runner.stop() async def test_a_stuck_sweep_that_deleted_nothing_logs_no_audit_gap( @@ -566,41 +683,39 @@ async def test_a_stuck_sweep_that_deleted_nothing_logs_no_audit_gap( ) -> None: """BACKLOG #2264 item 1. ``stop()`` logged its possible-audit-gap ERROR even when the sweep had deleted nothing. Parked in its sidecar scan, the sweep has deleted nothing, and once released - it sees the abort and deletes nothing more, so no row is lost: a WARNING, not an ERROR.""" + it deletes nothing more, so no row is lost: a WARNING, not an ERROR. Since BACKLOG #3184 each + pair is its own worker-thread call, so the cancelled sweep never reaches the pair at all.""" store = _quota_store(tmp_path, retention_days=30) meta = await store.save(data=b"aging\n", filename="a.txt", uploader="op", uploader_id="u-op") - paused, release, decided = threading.Event(), threading.Event(), threading.Event() + paused, release, scanned = threading.Event(), threading.Event(), threading.Event() real_scan = store._scan_metas_sync - real_begin = _SweepTally.begin - began: list[bool] = [] - def _stuck_scan() -> list[UploadedFileMeta]: + def _stuck_scan(held: list[Any] | None = None) -> list[UploadedFileMeta]: paused.set() release.wait(timeout=10) - return real_scan() - - def _begin(tally: _SweepTally, m: UploadedFileMeta, abort: threading.Event) -> bool: - began.append(real_begin(tally, m, abort)) - decided.set() - return began[-1] + try: + return real_scan(held) + finally: + scanned.set() monkeypatch.setattr(store, "_scan_metas_sync", _stuck_scan) - monkeypatch.setattr(_SweepTally, "begin", _begin) record = await _stop_stuck_runner(store, paused, release, caplog) assert record.levelno == logging.WARNING, record.getMessage() assert "no upload.prune row is lost" in record.getMessage() - # Nothing joins a cancelled to_thread job, so wait for the released thread's decision on the - # pair. The abort was set before the release, so it declines, and the pair stays whole. - assert await asyncio.to_thread(decided.wait, 10), "the released sweep never reached the pair" - assert began == [False] + # Nothing joins a cancelled to_thread job, so wait for the released scan to end. The sweep was + # cancelled during it, and the pair stays whole. That the abort check stops a pass before its + # next pair is pinned by test_stopping_the_runner_mid_sweep_audits_every_file_it_deleted. + assert await asyncio.to_thread(scanned.wait, 10), "the released scan never ended" assert [m.file_id for m in await store.list_files()] == [meta.file_id] async def test_a_sweep_stuck_in_its_audit_writes_names_each_unaudited_file( tmp_path: Path, caplog: pytest.LogCaptureFixture ) -> None: - """BACKLOG #2264, code review. The sweep removed three pairs and its audit write hangs on the - second, so two rows are missing. The ERROR names exactly those two files, by ``file_id``.""" + """BACKLOG #2264, code review. The audit write hangs on the second removed pair, so its row is + missing. The ERROR names exactly that file, by ``file_id``. Since BACKLOG #3184 each row is + written before the next pair is touched, so the third pair is never removed: before it, the + sweep removed all three first and the ERROR named two.""" store = _quota_store(tmp_path, retention_days=30) for i in range(3): await store.save( @@ -617,10 +732,11 @@ async def _audit(m: UploadedFileMeta) -> None: record = await _stop_stuck_runner(store, paused, threading.Event(), caplog, audit=_audit) assert record.levelno == logging.ERROR, record.getMessage() - assert "2 file(s) it removed, or was removing, may have no" in record.getMessage() - assert await store.list_files() == [] + assert "1 file(s) it removed, or was removing, may have no" in record.getMessage() named = set(record.getMessage().rsplit(": ", 1)[1].split(", ")) - assert len(written) == 1 and written[0] not in named and len(named) == 2 + remaining = [m.file_id for m in await store.list_files()] + assert len(written) == 1 and written[0] not in named and len(named) == 1 + assert len(remaining) == 1 and not named & set(remaining), "the stop did not end the pass" async def test_a_refused_unlink_is_not_counted_as_an_audit_gap( @@ -696,7 +812,7 @@ async def test_a_pair_another_pass_already_removed_is_not_reported_twice(tmp_pat await store.save(data=b"aging\n", filename="a.txt", uploader="op", uploader_id="u-op") seen = store._scan_metas_sync() first = await store.prune_expired(now=time.time() + 31 * 86_400) - store._scan_metas_sync = lambda: seen # type: ignore[method-assign] + store._scan_metas_sync = lambda held=None: seen # type: ignore[method-assign] second = await store.prune_expired(now=time.time() + 31 * 86_400) assert len(first.pruned) == 1 @@ -1399,10 +1515,11 @@ async def _write() -> None: async def test_a_save_time_sweep_shutdown_cancels_names_each_unaudited_file( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture ) -> None: - """BACKLOG #2264, code review. The save-time sweep removed three pairs and its audit write - hangs on the second. Its request is cancelled, the bound passes, and shutdown then cancels - the held sweep. The ERROR names exactly the two files with no row, as the runner's stop does, - rather than a bare "may have" that also fires for a sweep that removed nothing.""" + """BACKLOG #2264, code review. The save-time sweep's audit write hangs on the second pair it + removed. Its request is cancelled, the bound passes, and shutdown then cancels the held sweep. + The ERROR names exactly the file with no row, as the runner's stop does, rather than a bare + "may have" that also fires for a sweep that removed nothing. Since BACKLOG #3184 each row is + written before the next pair is touched, so the third pair is never removed.""" monkeypatch.setattr("messagefoundry.uploads._LEDGER_CANCEL_WAIT_SECONDS", 0.05) store = _quota_store(tmp_path, retention_days=30) ids = await _save_aged(store, tmp_path / "uploads", 3) @@ -1429,9 +1546,11 @@ async def _audit(m: UploadedFileMeta) -> None: [record] = [r for r in caplog.records if "upload.prune" in r.getMessage()] assert record.levelno == logging.ERROR, record.getMessage() assert record.getMessage().startswith("the save-time upload prune was cancelled") - assert "2 file(s) it removed, or was removing, may have no" in record.getMessage() + assert "1 file(s) it removed, or was removing, may have no" in record.getMessage() named = set(record.getMessage().rsplit(": ", 1)[1].split(", ")) - assert named == ids - set(written) and len(written) == 1 + remaining = {m.file_id for m in await store.list_files()} + assert len(written) == 1 and len(named) == 1 and len(remaining) == 1 + assert named | remaining == ids - set(written) async def test_an_orphan_sweep_failure_keeps_the_pruned_pairs_reported(tmp_path: Path) -> None: diff --git a/tests/test_uploads_cross_process_quota.py b/tests/test_uploads_cross_process_quota.py index c206874393..9036e75a4a 100644 --- a/tests/test_uploads_cross_process_quota.py +++ b/tests/test_uploads_cross_process_quota.py @@ -38,6 +38,7 @@ from collections.abc import AsyncIterator from contextlib import asynccontextmanager from pathlib import Path +from typing import Any import pytest @@ -78,8 +79,8 @@ async def _run() -> object: ) real_scan = uploads._scan_metas_sync - def _slow_scan() -> list[UploadedFileMeta]: - out = real_scan() + def _slow_scan(held: list[Any] | None = None) -> list[UploadedFileMeta]: + out = real_scan(held) time.sleep(_SCAN_OVERLAP_SECONDS) # widen the scan -> write window return out @@ -324,6 +325,87 @@ async def _run() -> None: asyncio.run(_run()) +class _StuckTouchLedger: + """A ledger whose reserve and release answer at once and whose touch never returns: a store + that stalls on exactly the statement the heartbeat sends.""" + + def __init__(self) -> None: + self.touches = 0 + self.cut_off = 0 + + async def reserve_upload_quota( + self, + uploader_id: str, + *, + files: int, + size_bytes: int, + max_files: int = 0, + max_total_bytes: int = 0, + ) -> bool: + if files == 0 and size_bytes == 0: + self.touches += 1 + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + self.cut_off += 1 + raise + return True + + async def upload_quota_in_flight(self, uploader_id: str) -> tuple[int, int]: + return (1, 5) + + +def test_a_stuck_heartbeat_touch_no_longer_holds_the_heartbeat( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """BACKLOG #3256 item 3. The heartbeat awaited its touch with no time bound, so one touch stuck + on the store held the heartbeat, and its lifetime cap, for as long as the store stalled. The + wait is bounded now and the touch is left running, never cancelled, and never doubled. Red + before the fix: the heartbeat never reached its cap.""" + monkeypatch.setattr(uploads_mod, "_RESERVATION_HEARTBEAT_SECONDS", 0.02) + monkeypatch.setattr(uploads_mod, "_RESERVATION_TOUCH_WAIT_SECONDS", 0.01) + monkeypatch.setattr(uploads_mod, "_RESERVATION_HEARTBEAT_MAX_SECONDS", 0.3) + ledger = _StuckTouchLedger() + + async def _run() -> None: + uploads = UploadStore( + tmp_path / "uploads", make_cipher(generate_key()), max_bytes=4096, store=ledger + ) + entered, gate = threading.Event(), threading.Event() + real_encrypt = uploads._encrypt_blob + + def _held(data: bytes, file_id: str) -> str: + entered.set() + gate.wait(timeout=30) + return real_encrypt(data, file_id) + + uploads._encrypt_blob = _held # type: ignore[method-assign] + save = asyncio.ensure_future( + uploads.save(data=b"slow\n", filename="a.txt", uploader="alice", uploader_id="u-alice") + ) + try: + assert await asyncio.to_thread(entered.wait, 30), "the save never reached its write" + for _ in range(250): + if "no longer refreshed" in caplog.text: + break + await asyncio.sleep(0.02) + assert "no longer refreshed" in caplog.text, "a stuck touch held the heartbeat" + await asyncio.sleep(0.05) # the task leaves the set on its done callback + assert not uploads._heartbeats + # The stuck touch was left running, held, and never doubled or cancelled. + assert (ledger.touches, ledger.cut_off) == (1, 0) + assert uploads._stragglers, "the stuck touch was dropped instead of held" + finally: + gate.set() + await asyncio.gather(save, return_exceptions=True) + for held in set(uploads._stragglers): + held.cancel() # stands in for asyncio.run's shutdown, which would cancel it + + with caplog.at_level("WARNING", logger=uploads_mod._log.name): + asyncio.run(_run()) + assert "did not finish within" in caplog.text + + def test_a_lagging_clock_never_moves_the_row_backwards(tmp_path: Path) -> None: """BACKLOG #2648. ``since`` is stamped from each writer's own clock. A shard whose clock lags must not age a row a sibling keeps fresh, or the sibling's next reserve reclaims live slots. diff --git a/tests/test_uploads_reseal.py b/tests/test_uploads_reseal.py index ed71c78246..73ee6f47f2 100644 --- a/tests/test_uploads_reseal.py +++ b/tests/test_uploads_reseal.py @@ -27,6 +27,7 @@ import logging from collections.abc import Callable from pathlib import Path +from typing import Any import pytest @@ -230,15 +231,47 @@ async def test_a_skipped_plaintext_body_leaves_its_sidecar_untouched_and_unliste async def test_an_unreadable_sidecar_still_counts_its_pair_once(tmp_path: Path) -> None: """The other half of the unit: a readable body is sealed, and the failed sidecar adds ONE skip. - The plaintext body it sealed does not count in ``sealed_plaintext``: that field counts uploads - turned from refused into readable, and this one is still unreadable. + The plaintext body it sealed counts in ``sealed_plaintext`` too (BACKLOG #3240). It used not + to, and a planted plaintext body under a sealed sidecar whose read failed once was then sealed + without ever being counted: the next run sees no plaintext in that upload at all. """ root = tmp_path / "uploads" fid = await _seed(_keyed_store(root, None)) _garble(root / f"{fid}.meta") result = await _keyed_store(root, generate_key()).reseal_to_active() - assert result == ResealResult(resealed=1, skipped=1, sealed_plaintext=0) + assert result == ResealResult(resealed=1, skipped=1, sealed_plaintext=1) + + +async def test_a_planted_body_under_a_sealed_sidecar_is_counted_when_its_read_fails_once( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """BACKLOG #3240, the case the count missed. A genuine sealed upload has its body replaced with + plaintext, and the first pass cannot read the sidecar. That pass seals the planted body, so the + second pass finds nothing plaintext. The seal must still be counted once, in the first pass.""" + root = tmp_path / "uploads" + key = generate_key() + keyed = _keyed_store(root, key) + fid = await _seed(keyed) + plain = await _seed(_keyed_store(root / "plain", None)) + (root / f"{fid}.blob").write_bytes((root / "plain" / f"{plain}.blob").read_bytes()) + sidecar = root / f"{fid}.meta" + real_open = Path.open + failed = False + + def _open_failing_once(self: Path, *args: Any, **kwargs: Any) -> Any: + nonlocal failed + if self == sidecar and not failed: + failed = True + raise PermissionError("held open by another process") + return real_open(self, *args, **kwargs) + + with monkeypatch.context() as m: + m.setattr(Path, "open", _open_failing_once) + first = await _keyed_store(root, key).reseal_to_active() + second = await _keyed_store(root, key).reseal_to_active() + assert first.sealed_plaintext + second.sealed_plaintext == 1, (first, second) + assert first == ResealResult(resealed=1, skipped=1, sealed_plaintext=1) async def test_a_pair_with_both_halves_unreadable_counts_once(tmp_path: Path) -> None: diff --git a/tests/test_uploads_shard_quota_interleave.py b/tests/test_uploads_shard_quota_interleave.py index 764cc6ae61..1ee0a3c608 100644 --- a/tests/test_uploads_shard_quota_interleave.py +++ b/tests/test_uploads_shard_quota_interleave.py @@ -31,6 +31,7 @@ import asyncio import os import threading +import time import uuid from collections.abc import AsyncIterator, Callable from contextlib import AbstractAsyncContextManager, asynccontextmanager @@ -93,9 +94,9 @@ def _b_observed(uploader_id: str) -> tuple[int, int]: armed = False scans = 0 - def _a_scan() -> list[UploadedFileMeta]: + def _a_scan(held: list[Any] | None = None) -> list[UploadedFileMeta]: nonlocal scans - out = real_a_scan() + out = real_a_scan(held) if armed: scans += 1 if scans == 2: # 1 = the pre-reserve scan, 2 = the scan inside _build_and_write @@ -269,3 +270,69 @@ async def _reserve(store: Any, stale_after: float = 300.0) -> bool: # Drain the row, so a server-backend run leaves nothing in flight behind it. await shard_c.reserve_upload_quota(uploader_id, files=-3, size_bytes=-30) assert await shard_c.upload_quota_in_flight(uploader_id) == (0, 0) + + +class _HostClockAhead: + """A store module's ``time``, with ``time()`` a decade ahead and everything else passed through: + an engine host whose clock is badly wrong.""" + + def __getattr__(self, name: str) -> Any: + return getattr(time, name) + + def time(self) -> float: + return time.time() + 10 * 365 * 86_400 + + +@pytest.mark.parametrize( + ("pair", "module"), + [ + pytest.param( + _BACKENDS[1].values[0], + "messagefoundry.store.postgres", + id="postgres", + marks=_BACKENDS[1].marks, + ), + pytest.param( + _BACKENDS[2].values[0], + "messagefoundry.store.sqlserver", + id="sqlserver", + marks=_BACKENDS[2].marks, + ), + ], +) +async def test_the_server_backends_stamp_since_with_the_database_clock( + tmp_path: Path, pair: _Pair, module: str, monkeypatch: pytest.MonkeyPatch +) -> None: + """BACKLOG #3256 item 1. ``since`` was stamped from the engine host's clock, so one host a + decade ahead stamped a row no sibling could age for a decade. The reserve, the release and the + touch now read the database server's clock. Shard A's host runs a decade ahead and its slot + leaks. Shard B, on a host with a true clock, then reserves with ``stale_after=0``, which + reclaims anything stamped at or before now: A's leaked slot goes. Red before the fix: A's + stamp is a decade ahead, nothing is reclaimed, and B counts two slots.""" + import importlib + + uploader_id = f"u-3256-{uuid.uuid4().hex}" + store_module = importlib.import_module(module) + async with pair(tmp_path) as (shard_a, shard_b): + + async def _reserve(store: Any, stale_after: float = 300.0) -> bool: + return bool( + await store.reserve_upload_quota( + uploader_id, + files=1, + size_bytes=10, + max_files=10, + max_total_bytes=1000, + stale_after=stale_after, + ) + ) + + with monkeypatch.context() as m: + m.setattr(store_module, "time", _HostClockAhead()) + assert await _reserve(shard_a) # shard A, never released + await shard_a.reserve_upload_quota(uploader_id, files=0, size_bytes=0) # a touch + await asyncio.sleep(0.05) # past A's stamp on the server's own clock + assert await _reserve(shard_b, stale_after=0.0) + assert await shard_b.upload_quota_in_flight(uploader_id) == (1, 10) + await shard_b.reserve_upload_quota(uploader_id, files=-1, size_bytes=-10) + assert await shard_b.upload_quota_in_flight(uploader_id) == (0, 0) diff --git a/tests/test_uploads_strict_ciphertext.py b/tests/test_uploads_strict_ciphertext.py index 9a4fb69093..1314155251 100644 --- a/tests/test_uploads_strict_ciphertext.py +++ b/tests/test_uploads_strict_ciphertext.py @@ -54,6 +54,7 @@ UploadQuotaError, UploadStore, UploadUnreadableError, + _UnsealableUpload, ) _ADT = "MSH|^~\\&|A|B|C|D|202601011200||ADT^A01|MSGID1|P|2.5\rPID|1||MRN123^^^HOSP||DOE^JOHN\r" @@ -482,6 +483,40 @@ async def test_a_delete_told_the_sidecar_was_refused_still_classifies_it(tmp_pat await store.delete("f" * 32, sidecar_refused=True) +@_BREAKS +async def test_a_delete_told_the_sidecar_was_refused_removes_an_unsealable_upload( + tmp_path: Path, break_body: Callable[[Path], None] +) -> None: + """BACKLOG #3240, the positive arm the store test lacked: with ``sidecar_refused`` the + never-sealable kind is removed, both files, and reported with blank owner fields.""" + root = tmp_path / "uploads" + fid = await _unsealable_upload(root, break_body) + store = UploadStore(root, _keyed(generate_key()), max_bytes=1 << 20) + meta = await store.delete(fid, sidecar_refused=True) + assert (meta.file_id, meta.uploader_id, meta.filename) == (fid, "", "") + assert _files(root) == [] + + +@pytest.mark.parametrize("sidecar_refused", [False, True], ids=["read-here", "read-by-caller"]) +async def test_a_sidecar_gone_before_it_is_classed_answers_not_found( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, sidecar_refused: bool +) -> None: + """BACKLOG #3240. A sidecar that vanished between the refused read and the class check, as when + a prune removes it, answered 423 (refused) where an absent sidecar answers 404.""" + root = tmp_path / "uploads" + fid = await _plaintext_upload(root) + store = UploadStore(root, _keyed(generate_key()), max_bytes=1 << 20) + real = store._unsealable_sync + + def _gone_first(file_id: str, sidecar: Path) -> object: + sidecar.unlink() + return real(file_id, sidecar) + + monkeypatch.setattr(store, "_unsealable_sync", _gone_first) + with pytest.raises(UploadNotFoundError): + await store.delete(fid, sidecar_refused=sidecar_refused) + + @_BREAKS async def test_an_unsealable_upload_counts_toward_the_quota_and_is_never_listed( tmp_path: Path, break_body: Callable[[Path], None], caplog: pytest.LogCaptureFixture @@ -574,11 +609,32 @@ async def test_a_body_the_engine_could_not_have_written_is_never_read(tmp_path: fid = await _plaintext_upload(root) (root / f"{fid}.blob").write_bytes(b"\xff" * (3 * 1024 + 4096)) store = UploadStore(root, _keyed(generate_key()), max_bytes=1024) - assert store._scan_unsealable_sync() == [] + held: list[_UnsealableUpload] = [] + assert store._scan_metas_sync(held) == [] and held == [] assert (await store.prune_expired(now=10**12)).pruned == [] assert (root / f"{fid}.blob").exists() +async def test_a_sidecar_undecodable_only_past_the_class_read_is_still_classed( + tmp_path: Path, +) -> None: + """The listing read of the whole sidecar fails, but the class check reads only its head, which + decodes. The quota and the prune still class it, as the separate unsealable scan did before the + two walks were folded into one (BACKLOG #3184), so it ages out rather than sit forever.""" + root = tmp_path / "uploads" + fid = await _unsealable_upload(root, _remove) + sidecar = root / f"{fid}.meta" + # Well past the owner read and the decoder's chunk after it, then one byte that is not UTF-8. + sidecar.write_bytes(sidecar.read_bytes() + b" " * (4 * 64 * 1024) + b"\xff") + store = UploadStore(root, _keyed(generate_key()), max_bytes=1 << 20, retention_days=30) + held: list[_UnsealableUpload] = [] + assert store._scan_metas_sync(held) == [] + # The owner is not read from a sidecar that was not read whole. + assert [(h.file_id, h.uploader_id, h.size) for h in held] == [(fid, "", 0)] + assert (await store.prune_expired(now=10**12)).pruned == [] # no body, so no row + assert _files(root) == [] + + def test_rotate_key_notes_an_unsealable_upload_and_does_not_warn( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] ) -> None: