fix(render): keep a seal's place until its preparation ends

run_seal held the seal's place in flight in the handler future, while
Seal::seal prepares the field on a blocking task. When a client
disconnected, the future and its place dropped, but the blocking work
ran on, so it went uncounted by the seal limit and the shutdown drain.
Seal::seal now takes the place and moves it into the blocking task,
which hands it back when it ends: a dropped seal keeps its place until
the preparation finishes, and the drain waits for it.

Signed-off-by: Jeremi Joslin <jeremi@joslin.fr>

Author: jeremi

crates/registry-render/src/seal.rs

@@ -878,27 +878,34 @@ impl Seal {
     /// must be a single-revision PDF; one that carries an update or bytes
     /// after its end marker is refused before anything is signed. Reading
     /// the PDF is blocking work, so it runs off the async workers; the
-    /// caller bounds `first_revision` by the output limit.
+    /// caller bounds `first_revision` by the output limit. `in_flight` is
+    /// the caller's place for this seal, held until the seal ends or, when
+    /// this future is dropped first, until the blocking work does.
     pub async fn seal(
         &self,
         first_revision: Vec<u8>,
         signing_time: OffsetDateTime,
+        in_flight: tokio::sync::OwnedSemaphorePermit,
     ) -> Result<SealedDocument, RenderProblem> {
         let internal = |detail: String| RenderProblem::new(ProblemKind::Internal, detail);
         let field = self.signature_field(signing_time)?;
-        let (document_digest, prepared) = tokio::task::spawn_blocking(move || {
+        // The place travels with the blocking work and back: if this future
+        // is dropped, the work runs on and the place is released only when
+        // it ends.
+        let (_in_flight, prepared) = tokio::task::spawn_blocking(move || {
             let document_digest = sha256_digest(&first_revision);
-            crate::pdf_signature::prepare(&first_revision, &field)
-                .map(|prepared| (document_digest, prepared))
+            let prepared = crate::pdf_signature::prepare(&first_revision, &field)
+                .map(|prepared| (document_digest, prepared));
+            (in_flight, prepared)
         })
         .await
         .map_err(|error| {
             internal(format!(
                 "seal {}: preparing the signature field stopped: {error}",
                 self.id
             ))
-        })?
-        .map_err(|error| {
+        })?;
+        let (document_digest, prepared) = prepared.map_err(|error| {
             internal(format!(
                 "seal {}: the rendered document cannot be sealed: {error}",
                 self.id
@@ -1296,6 +1303,58 @@ mod tests {
 
     use crate::worker::seal_tests::sealed_receipt;
 
+    /// A place for one seal in flight, from a service of one.
+    fn place() -> tokio::sync::OwnedSemaphorePermit {
+        Arc::new(tokio::sync::Semaphore::new(1))
+            .try_acquire_owned()
+            .expect("a free place")
+    }
+
+    /// A client that disconnects drops its seal's future, but preparing the
+    /// field is blocking work that runs on. The seal's place goes with that
+    /// work and comes back only when it ends, so the drain, which waits for
+    /// every place, waits for it too. With the one blocking thread busy, the
+    /// preparation waits in the queue, holding the place, until it is free.
+    #[test]
+    fn a_dropped_seal_keeps_its_place_until_its_blocking_work_ends() {
+        use std::future::Future as _;
+        use std::task::{Context, Waker};
+
+        let pki = TestPki::generate().expect("test PKI");
+        let test_seal = pki.issue_seal(SigningAlgorithm::Es256).expect("test seal");
+        let seal = Seal::for_test(
+            "ministry",
+            &test_seal,
+            Arc::new(test_seal.signer().expect("test signer")),
+        );
+        let first_revision = sealed_receipt(seal.worker_seal());
+        let places = Arc::new(tokio::sync::Semaphore::new(1));
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .max_blocking_threads(1)
+            .enable_all()
+            .build()
+            .expect("runtime");
+        runtime.block_on(async {
+            let (release, wait) = std::sync::mpsc::channel::<()>();
+            let busy = tokio::task::spawn_blocking(move || wait.recv());
+            let held = Arc::clone(&places).try_acquire_owned().expect("a place");
+            let mut sealing = Box::pin(seal.seal(first_revision, OffsetDateTime::now_utc(), held));
+            assert!(sealing
+                .as_mut()
+                .poll(&mut Context::from_waker(Waker::noop()))
+                .is_pending());
+            drop(sealing);
+            assert_eq!(
+                places.available_permits(),
+                0,
+                "the queued preparation holds the place"
+            );
+            release.send(()).expect("the busy task waits");
+            busy.await.expect("busy task").expect("released");
+            let _place = places.acquire().await.expect("the place comes back");
+        });
+    }
+
     /// A signer that counts its signatures, so a test can show that a
     /// refused answer was never signed.
     struct CountingSigner {
@@ -1358,7 +1417,7 @@ mod tests {
             ("bytes after the end marker", smuggled),
             ("an update of the worker's own", own_update),
         ] {
-            let Err(problem) = seal.seal(answer, now).await else {
+            let Err(problem) = seal.seal(answer, now, place()).await else {
                 panic!("{label}: sealed");
             };
             assert_eq!(problem.kind, ProblemKind::Internal, "{label}");
@@ -1369,7 +1428,10 @@ mod tests {
             "nothing signed"
         );
 
-        let sealed = seal.seal(honest.clone(), now).await.expect("seals");
+        let sealed = seal
+            .seal(honest.clone(), now, place())
+            .await
+            .expect("seals");
         assert_eq!(signer.signatures.load(Ordering::SeqCst), 1);
         assert_eq!(sealed.document_digest, sha256_digest(&honest));
         let expected = crate::pdf_signature::prepare(&honest, &field).expect("the parent's update");
@@ -1475,7 +1537,7 @@ mod tests {
 
         let first_revision = sealed_receipt(large_seal.worker_seal());
         large_seal
-            .seal(first_revision, OffsetDateTime::now_utc())
+            .seal(first_revision, OffsetDateTime::now_utc(), place())
             .await
             .expect("the signature fits the reserved space");
     }

crates/registry-render/src/server.rs

@@ -1219,7 +1219,8 @@ struct SealedRender {
 /// free, release the slot, then prepare the field, sign, timestamp and embed
 /// in this process, signed at `signing_time`. Preparing, signing and the
 /// timestamp exchange never hold a render slot; the whole call holds one of
-/// the service's places for seals in flight.
+/// the service's places for seals in flight, and the preparation keeps it
+/// until it ends even when the client has gone.
 async fn run_seal(
     service: &Service,
     seal: &Seal,
@@ -1228,7 +1229,7 @@ async fn run_seal(
 ) -> Result<(SealedRender, SealedDocument), RenderProblem> {
     // A seal beyond the places is refused, not queued: the queue behind a
     // slow authority would otherwise grow without a bound.
-    let _in_flight = service
+    let in_flight = service
         .seals_in_flight
         .clone()
         .try_acquire_owned()
@@ -1260,7 +1261,7 @@ async fn run_seal(
         data_sha256: answer.data_sha256,
         warnings: answer.warnings,
     };
-    let sealed = seal.seal(first_revision, signing_time).await?;
+    let sealed = seal.seal(first_revision, signing_time, in_flight).await?;
     Ok((rendered, sealed))
 }
 

products/render/SECURITY-MATRIX.md

@@ -16,7 +16,7 @@ conditions, follow the table.
 | 7 | Request data values or the API key leak into the audit log or logs | Audit events carry a closed, value-free field set; lifecycle logs use fixed dimensions (method class, route template, status, latency, trace id); the seal route's `seal refused` warning carries only the seal id and the problem and `reason` names | `serve::audit_events_are_value_free` (canary + key scans over the audit file) |
 | 8 | A render starts, or a document leaves, with no accepted audit record of it | The shared audit writer accepts a `file` entry only after `fsync` (owner-only file, single-writer lock) and a `stdout` entry only after it is written and flushed, then stops accepting after any failure. The `request` entry is accepted before the worker starts, else `503 audit-failed` and no render; the `response` entry is accepted before the document leaves, else `503 audit-failed` and no PDF or hash headers; `/ready` reports a stopped writer. Each call pairs its entries under a correlation the server draws, never the caller's `Idempotency-Key` (echoed only as `correlationId`), so calls sharing a key cannot be confused, and a call that ends before its outcome is written writes an `unfinished` response. Host-level tampering with the log is not defended here: the log carries no hash chain or signature (residual below) | `server::tests::the_request_entry_is_accepted_before_the_render_starts`; `server::tests::a_refused_request_entry_prevents_the_render`; `serve::a_refused_request_entry_prevents_the_render`; `serve::a_refused_response_entry_withholds_the_pdf`; `serve::a_render_writes_a_request_entry_then_a_response_entry_sharing_correlation` (file mode 0600); `server::tests::concurrent_calls_sharing_an_idempotency_key_pair_their_own_entries` |
 | 9 | A render succeeds but is wrong on paper (missing glyphs) | Check-time label-script coverage over the actual label characters; package validates before it writes any output; render-time warnings; `--strict` fails on warnings | `scaffold::package_refuses_a_broken_bundle_before_writing_output` (CJK label, no font, so the package directory is never created) |
-| 10 | Pathological template data exhausts CPU or memory | Serves render in a supervised worker process: killed at the configured timeout, which also bounds its exit after answering, `RLIMIT_AS` on Linux, recycled on panic; bounded concurrency, and as many seals in flight as render slots once a seal releases its slot; request body and output caps with hard ceilings | `serve::pathological_renders_are_bounded_and_the_service_recovers` (timeout or memory wall is audited; a fresh worker immediately serves a healthy render); `worker::tests::a_worker_that_lingers_after_answering_is_cut_off_at_the_deadline`; `serve_seal::seals_in_flight_beyond_the_render_concurrency_are_refused_as_rate_limited` |
+| 10 | Pathological template data exhausts CPU or memory | Serves render in a supervised worker process: killed at the configured timeout, which also bounds its exit after answering, `RLIMIT_AS` on Linux, recycled on panic; bounded concurrency, and as many seals in flight as render slots once a seal releases its slot, a seal counted until its preparation ends even after its client has gone; request body and output caps with hard ceilings | `serve::pathological_renders_are_bounded_and_the_service_recovers` (timeout or memory wall is audited; a fresh worker immediately serves a healthy render); `worker::tests::a_worker_that_lingers_after_answering_is_cut_off_at_the_deadline`; `serve_seal::seals_in_flight_beyond_the_render_concurrency_are_refused_as_rate_limited`; `seal::tests::a_dropped_seal_keeps_its_place_until_its_blocking_work_ends` |
 | 11 | Assets smuggle executables or balloons | base64-decoded, JPEG/PNG magic sniffed, per-asset 2 MiB / per-request 8 MiB caps, exact bytes covered by `dataDigest` | `golden::wrong_media_type_and_oversize_assets_are_refused` |
 | 12 | Deeply nested JSON parse DoS | `serde_json` default 128-depth recursion limit on top of the tower body-limit layer | platform-owned; body limit exercised by config ceiling (`runtime.rs` 64 MiB hard cap) |
 | 13 | Response header injection via echoed `Idempotency-Key` | `HeaderValue::from_str` rejects CR/LF/NUL; 128-char bound at read | code-reviewed; echo asserted in `serve::render_returns_pdf_with_hash_headers_matching_golden` |