From 868b1345ebe5df4d2c64bb35d2fc110c7b99b840 Mon Sep 17 00:00:00 2001 From: Luke Valenta Date: Thu, 23 Jul 2026 12:15:09 -0400 Subject: [PATCH 01/11] mirror_worker: implement the add-entries submission API Read and verify entry packages against the target pending checkpoint, persist them as tlog-tiles entry bundles and hash tiles (see commit.rs), and advance the persisted-entry frontier. A complete upload cosigns the mirror checkpoint and returns 200; a client-truncated upload persists the verified prefix and returns 202 with the advanced next entry so the client can resume (C2SP/C2SP#253). The whole request body is buffered and all its packages are committed in a single pass. commit.rs persists resumably from the current frontier, so a client that truncated at a package boundary resumes cleanly on its next request. A follow-up commit adds incremental streaming so a large upload does not have to buffer the entire body. Includes body-decoding (transparent gzip), per-origin prefixed R2 storage, and the ticket-sealer for recovering a past pending checkpoint on resume. --- Cargo.lock | 48 ++ Cargo.toml | 2 + crates/mirror_worker/.dev.vars | 1 + crates/mirror_worker/Cargo.toml | 14 +- crates/mirror_worker/config/src/lib.rs | 4 + crates/mirror_worker/src/add_entries.rs | 898 ++++++++++++++++++++ crates/mirror_worker/src/body.rs | 127 +++ crates/mirror_worker/src/commit.rs | 466 ++++++++++ crates/mirror_worker/src/frontend_worker.rs | 23 +- crates/mirror_worker/src/lib.rs | 102 ++- crates/mirror_worker/src/storage.rs | 108 +++ crates/mirror_worker/wrangler.jsonc | 11 + 12 files changed, 1789 insertions(+), 15 deletions(-) create mode 100644 crates/mirror_worker/src/add_entries.rs create mode 100644 crates/mirror_worker/src/body.rs create mode 100644 crates/mirror_worker/src/commit.rs create mode 100644 crates/mirror_worker/src/storage.rs diff --git a/Cargo.lock b/Cargo.lock index e35b4f4a..8b28f4d9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,12 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "adler2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" + [[package]] name = "aead" version = "0.6.1" @@ -528,6 +534,15 @@ dependencies = [ "libc", ] +[[package]] +name = "crc32fast" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9481c1c90cbf2ac953f07c8d4a58aa3945c425b7185c9154d67a65e4230da511" +dependencies = [ + "cfg-if", +] + [[package]] name = "criterion" version = "0.5.1" @@ -996,6 +1011,16 @@ version = "0.4.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b7ac824320a75a52197e8f2d787f6a38b6718bb6897a35142d749af3c0e8f4fe" +[[package]] +name = "flate2" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c" +dependencies = [ + "crc32fast", + "miniz_oxide", +] + [[package]] name = "fluent-uri" version = "0.3.2" @@ -1840,6 +1865,16 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" +[[package]] +name = "miniz_oxide" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316" +dependencies = [ + "adler2", + "simd-adler32", +] + [[package]] name = "mio" version = "1.1.0" @@ -1860,11 +1895,14 @@ dependencies = [ "console_error_panic_hook", "console_log", "ed25519-dalek", + "flate2", + "futures-util", "generic_log_worker", "getrandom 0.3.4", "getrandom 0.4.2", "hex", "jsonschema", + "length_prefixed", "log", "mirror_worker_config", "ml-dsa", @@ -1872,11 +1910,15 @@ dependencies = [ "serde", "serde_json", "serde_with", + "sha2", "signed_note", "tlog_checkpoint", "tlog_core", "tlog_cosignature", + "tlog_mirror", + "tlog_tiles", "tlog_witness", + "tokio", "tower-service", "worker", ] @@ -3029,6 +3071,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "simd-adler32" +version = "0.3.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a219298ac11a56ea9a6d2120044824d6f01aeb034955e7af7bc16858527deea" + [[package]] name = "slab" version = "0.4.11" diff --git a/Cargo.toml b/Cargo.toml index 79594e3d..8633d705 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -121,6 +121,7 @@ der = { version = "0.8", features = ["oid"] } # (crypto-common 0.2 / signature 3.0), so it remains pinned to a pre-release. # The rest of the batch is now stable. ed25519-dalek = { version = "3.0.0", features = ["pem"] } +flate2 = { version = "1", default-features = false, features = ["rust_backend"] } futures-executor = "0.3.31" futures-util = "0.3.31" getrandom = { version = "0.4", features = ["wasm_js"] } @@ -157,6 +158,7 @@ tlog_checkpoint = { path = "crates/tlog_checkpoint", version = "0.2.0" } tlog_core = { path = "crates/tlog_core", version = "0.2.0" } tlog_cosignature = { path = "crates/tlog_cosignature", version = "0.2.0" } tlog_entry = { path = "crates/tlog_entry", version = "0.2.0" } +tlog_mirror = { path = "crates/tlog_mirror", version = "0.2.0" } tlog_tiles = { path = "crates/tlog_tiles", version = "0.2.0" } tlog_witness = { path = "crates/tlog_witness", version = "0.2.0" } tokio = { version = "1", features = ["sync"] } diff --git a/crates/mirror_worker/.dev.vars b/crates/mirror_worker/.dev.vars index c0d8d885..2150b9d6 100644 --- a/crates/mirror_worker/.dev.vars +++ b/crates/mirror_worker/.dev.vars @@ -1 +1,2 @@ MIRROR_SIGNING_KEY="-----BEGIN PRIVATE KEY-----\nMDQCAQAwCwYJYIZIAWUDBAMRBCKAIEJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJC\nQkJCQkJC\n-----END PRIVATE KEY-----\n" +MIRROR_TICKET_KEY="Nzc3Nzc3Nzc3Nzc3Nzc3Nzc3Nzc3Nzc3Nzc3Nzc3Nzc=" diff --git a/crates/mirror_worker/Cargo.toml b/crates/mirror_worker/Cargo.toml index 23f53a85..6443cc3d 100644 --- a/crates/mirror_worker/Cargo.toml +++ b/crates/mirror_worker/Cargo.toml @@ -32,33 +32,39 @@ serde_json.workspace = true [dependencies] axum.workspace = true +base64.workspace = true config = { path = "./config", package = "mirror_worker_config" } console_error_panic_hook.workspace = true console_log.workspace = true +flate2.workspace = true +futures-util.workspace = true generic_log_worker.workspace = true getrandom.workspace = true getrandom_03.workspace = true hex.workspace = true +length_prefixed.workspace = true log.workspace = true ml-dsa.workspace = true pkcs8.workspace = true serde.workspace = true serde_json.workspace = true serde_with.workspace = true +sha2.workspace = true signed_note.workspace = true tlog_checkpoint.workspace = true tlog_core.workspace = true tlog_cosignature.workspace = true +tlog_mirror.workspace = true +tlog_tiles.workspace = true tlog_witness.workspace = true tower-service.workspace = true worker = { workspace = true, features = ["http", "axum"] } [dev-dependencies] -# base64 is used only by the dev-config pin tests; ed25519-dalek only to -# check that a non-ML-DSA-44 MIRROR_SIGNING_KEY is rejected. The shipped -# worker is ML-DSA-44 only. -base64.workspace = true +# Ed25519 is used only by unit tests, to check that a non-ML-DSA-44 +# MIRROR_SIGNING_KEY is rejected. The shipped worker is ML-DSA-44 only. ed25519-dalek.workspace = true +tokio = { workspace = true, features = ["macros", "rt"] } [lints.rust] unexpected_cfgs = { level = "warn", check-cfg = [ diff --git a/crates/mirror_worker/config/src/lib.rs b/crates/mirror_worker/config/src/lib.rs index 1998dcf4..5197b506 100644 --- a/crates/mirror_worker/config/src/lib.rs +++ b/crates/mirror_worker/config/src/lib.rs @@ -141,6 +141,10 @@ impl AppConfig { /// signed-note key name length cap, since each origin is itself /// used as a checkpoint origin. /// + /// Simple single-field bounds (e.g. the log-number ranges) are + /// expressed in `config.schema.json` and enforced by the build + /// script, so they are not re-checked here. + /// /// `log_key_name` uniqueness across log entries is not checked here; /// it is enforced earlier, during deserialization (see /// [`deserialize_logs`]). A plain `serde_json` object silently keeps diff --git a/crates/mirror_worker/src/add_entries.rs b/crates/mirror_worker/src/add_entries.rs new file mode 100644 index 00000000..62e3909b --- /dev/null +++ b/crates/mirror_worker/src/add_entries.rs @@ -0,0 +1,898 @@ +// Copyright (c) 2025-2026 Cloudflare, Inc. All rights reserved. +// SPDX-License-Identifier: BSD-3-Clause + +//! `POST /add-entries` handler. +//! +//! Implements the [c2sp.org/tlog-mirror `add-entries`][add-e] endpoint: +//! read the (optionally gzip) request body, verify each [`EntryPackage`] +//! against the target pending checkpoint with a subtree consistency proof, +//! persist the verified entries as bundles and hash tiles (see +//! [`crate::commit`]), and advance the persisted-entry frontier. +//! +//! The whole body is buffered, then all its packages are committed in a +//! single pass; the mirror checkpoint is cosigned only once the upload is +//! durably committed. A follow-up commit adds incremental streaming so a +//! large upload does not have to buffer the entire body in memory. +//! +//! A complete upload writes the cosigned checkpoint and returns 200 with +//! the mirror's [cosignature][cosig] line(s); a client-truncated upload +//! keeps the persisted prefix and returns 202 with the advanced next entry +//! so the client can resume (see [Processing][proc]). +//! +//! [add-e]: https://c2sp.org/tlog-mirror#add-entries +//! [proc]: https://c2sp.org/tlog-mirror#processing +//! [cosig]: https://c2sp.org/tlog-cosignature +//! [`EntryPackage`]: tlog_mirror::EntryPackage + +use std::collections::HashMap; +use std::io::{Cursor, ErrorKind}; + +use axum::extract::State; +use axum::http::{StatusCode, header::CONTENT_TYPE}; +use axum::response::IntoResponse as _; +use signed_note::{Note, NoteError}; +use tlog_checkpoint::CheckpointText; +use tlog_core::{ + Hash, HashReader, Subtree, TlogError, stored_hash_index, stored_hashes_for_record_hash, + tree_hash, verify_subtree_consistency_proof, +}; +use tlog_mirror::{ + AddEntriesRequestHeader, EntryPackage, MIRROR_INFO_CONTENT_TYPE, MirrorInfo, PACKAGE_ALIGNMENT, + ParseError, package_ranges, +}; +#[allow(clippy::wildcard_imports)] +use worker::*; + +use generic_log_worker::{ObjectBackend, util::now_millis}; + +use crate::{ + body, commit, + frontend_worker::{ApiResult, AppError}, + load_mirror_signer, load_ticket_sealer, log_verifiers, + mirror_state_do::{ + AdvanceNextEntryRequest, CommitRequest, MirrorStateSnapshot, NextEntry, PendingCheckpoint, + state_stub, + }, + storage::load_origin_bucket, +}; + +/// Handle `POST /add-entries`. +/// +/// See the module-level comment for the full flow: read and verify entry +/// packages from the request body, persist the verified entries and +/// advance the persisted-entry frontier, and either cosign the mirror +/// checkpoint (200) or, for a truncated upload, persist the verified +/// prefix and return 202. +#[worker::send] +pub(crate) async fn add_entries( + State(env): State, + req: axum::extract::Request, +) -> ApiResult { + // No DefaultBodyLimit: Cloudflare enforces a request-body cap at the + // edge (100 MB, higher on paid plans) and 413s oversized bodies there. + // Clients on body-limited platforms truncate at a package boundary and + // resume via the 202 + advanced next_entry (tlog-mirror "Implementation + // Considerations"). The Workers runtime does not decompress request + // bodies, so gzip-encoded bodies are gunzipped here; unknown encodings + // are 415'd (see `crate::body`). + // + // The whole (decoded) body is buffered before processing; a follow-up + // commit streams it instead of buffering it all. + let (parts, body) = req.into_parts(); + let raw = body::read_decoded_body(&parts.headers, body).await?; + let mut cursor = Cursor::new(raw.as_slice()); + + let header = match AddEntriesRequestHeader::read_from(&mut cursor) { + Ok(header) => header, + Err(e) => { + log::warn!("add-entries: malformed header: {e:?}"); + return Err(AppError::BadRequest(e.to_string())); + } + }; + + let Some(verifiers) = log_verifiers(&header.log_origin) else { + return Err(AppError::UnknownLogOrigin); + }; + + let snapshot = fetch_snapshot(&env, &header.log_origin).await?; + + // No pending checkpoint accepted yet means there is nothing to + // authenticate the entries against, so this MUST be 422, not 409: the + // client can't make progress by retrying and must first drive an + // add-checkpoint (tlog-mirror "Processing"). Empty pending signed-note bytes + // reliably mean pristine state; once accepted, the DO retains the + // latest pending forever. + if snapshot.pending.signed_note_bytes.is_empty() { + log::info!( + "add-entries: no pending checkpoint for origin {:?}; returning 422", + header.log_origin, + ); + return Err(AppError::UnprocessableEntity( + "mirror has no pending checkpoint for this log".to_owned(), + )); + } + + let target = match resolve_target_pending(&env, &header, &snapshot, &verifiers) { + Ok(t) => t, + Err(reason) => { + log::info!( + "add-entries: rejecting target pending: {reason} \ + (origin={origin:?}, upload_end={ue}, pending_size={ps}, committed_size={cs})", + origin = header.log_origin, + ue = header.upload_end, + ps = snapshot.pending.size, + cs = snapshot.committed.size, + ); + return Ok(mirror_info_409(&env, &snapshot, &header.log_origin)); + } + }; + + // upload_start must be <= next_entry (the first index not yet + // persisted); a client resuming after a 202 sets it to the advertised + // next entry. A non-256-aligned value is accepted; see + // `first_package_prefix`. + if header.upload_start > snapshot.next_entry.size { + log::info!( + "add-entries: rejecting upload_start={us} > next_entry={ne}", + us = header.upload_start, + ne = snapshot.next_entry.size, + ); + return Ok(mirror_info_409(&env, &snapshot, &header.log_origin)); + } + + let first_prefix = first_package_prefix(&env, &header, snapshot.next_entry.size).await?; + + verify_and_persist( + &env, + &header, + &snapshot, + &target, + &mut cursor, + &first_prefix, + ) + .await +} + +/// Read, verify, and persist the entry packages for `[upload_start, +/// upload_end)`, then produce the HTTP response. +/// +/// Packages are read and verified in canonical order from the buffered +/// body. Entries below the frontier at request start are already +/// persisted, so they are verified but not re-saved (spec: "skip saving +/// already-written entries"); the remaining entries are committed in a +/// single [`commit::persist_entries`] call and the persisted-entry +/// frontier is advanced once in the DO. A follow-up commit flushes +/// incrementally instead of once at the end. +/// +/// Response cases (see [Processing][proc]): +/// +/// * every package received and the recomputed tree matches the target: +/// cosign the mirror checkpoint, advance it, and return 200 with the +/// cosignature line(s). +/// * client truncation after at least one complete package: persist the +/// verified prefix and return 202 with the advanced next entry. +/// * truncation before the first complete package, or a malformed body: +/// 400. A package that fails subtree-consistency verification: 422. +/// +/// # Errors +/// +/// Returns an error on a storage failure while persisting, or (500) if the +/// recomputed root of a complete upload disagrees with the target +/// checkpoint. +/// +/// [proc]: https://c2sp.org/tlog-mirror#processing +async fn verify_and_persist( + env: &Env, + header: &AddEntriesRequestHeader, + snapshot: &MirrorStateSnapshot, + target: &PendingCheckpoint, + cursor: &mut Cursor<&[u8]>, + first_prefix: &[Vec], +) -> ApiResult { + let bucket = load_origin_bucket(env, &header.log_origin)?; + + // Entries below the request-start frontier are already persisted; new + // persistence begins at this fixed boundary. + let initial_next = snapshot.next_entry.size; + + // Newly-received entries collected across all packages for a single + // commit, and the leaf index one past the last collected entry. + let mut entries: Vec> = Vec::new(); + let mut collected_end = initial_next; + let mut packages_received: u64 = 0; + let mut truncated = false; + + for (pkg_start, pkg_end) in package_ranges(header.upload_start, header.upload_end) { + let num_entries = pkg_end - pkg_start; + let pkg = match read_package(cursor, num_entries) { + PackageOutcome::Ok(pkg) => pkg, + // The body ended between or partway through a package: client + // truncation. Keep whatever complete packages were verified. + PackageOutcome::Eof => { + truncated = true; + break; + } + PackageOutcome::Err(e) => { + log::warn!("add-entries: malformed package [{pkg_start}, {pkg_end}): {e:?}"); + return Err(AppError::BadRequest(e.to_string())); + } + }; + + // Only the first package can be non-256-aligned, carrying the + // persisted-leaf prefix; later packages start on a 256 boundary + // (subtree start == pkg_start) and need no prefix. + let subtree_start = (pkg_start / PACKAGE_ALIGNMENT) * PACKAGE_ALIGNMENT; + let prefix: &[Vec] = if packages_received == 0 { + first_prefix + } else { + &[] + }; + if let Err(reason) = verify_package(prefix, &pkg, subtree_start, pkg_end, target) { + log::info!( + "add-entries: package [{pkg_start}, {pkg_end}) failed verification: {reason}" + ); + return Err(AppError::UnprocessableEntity(reason.to_owned())); + } + packages_received += 1; + + // Collect only the not-yet-persisted tail of this package. + if pkg_end > initial_next { + let skip = usize::try_from(initial_next.saturating_sub(pkg_start)) + .map_err(|_| Error::from("skip count overflows usize"))?; + entries.extend(pkg.entries.into_iter().skip(skip)); + collected_end = pkg_end; + } + } + + // Truncation before the first complete package is a hard 400. + if truncated && packages_received == 0 { + log::warn!("add-entries: body truncated before the first complete package"); + return Err(AppError::BadRequest( + "no complete entry package received".to_owned(), + )); + } + + // Persist all newly-received entries in a single commit, advancing the + // persisted-entry frontier once. + let mut frontier_size = initial_next; + let mut frontier_hash = snapshot.next_entry.hash; + if !entries.is_empty() { + let root = commit::persist_entries( + &bucket, + frontier_size, + frontier_hash, + collected_end, + &entries, + ) + .await?; + advance_next_entry(env, &header.log_origin, collected_end, root).await?; + frontier_size = collected_end; + frontier_hash = root; + } + let persisted_new = frontier_size > initial_next; + + if truncated { + log::info!( + "add-entries: client-truncated after {packages_received} complete packages; \ + persisted through {frontier_size}", + ); + return Ok(mirror_info_202( + env, + snapshot, + &header.log_origin, + frontier_size, + )); + } + + // Complete upload: no trailing bytes may remain after the last package. + let pos = usize::try_from(cursor.position()).unwrap_or(usize::MAX); + if pos < cursor.get_ref().len() { + log::warn!("add-entries: trailing data after the last entry package"); + return Err(AppError::BadRequest( + "trailing data after the last entry package".to_owned(), + )); + } + + // When we persisted new entries, the recomputed tree MUST match the + // pending checkpoint the packages were proven against. A mismatch means + // proof verification and tile computation disagree: an internal error, + // never a client fault. When nothing new was persisted (a re-upload of + // an already-persisted range), the log-signed target is trusted + // directly. + if persisted_new && (frontier_size != header.upload_end || frontier_hash != target.hash) { + log::error!( + "add-entries: recomputed frontier ({frontier_size}, {frontier_hash}) != target ({}, {})", + target.size, + target.hash, + ); + return Err(AppError::InternalServerError( + "recomputed root mismatch".to_owned(), + )); + } + + cosign_and_serve(env, header, target, &bucket).await +} + +/// Outcome of reading the next entry package from the buffered body. +enum PackageOutcome { + Ok(EntryPackage), + /// The body ended between packages or partway through one: a client + /// truncation. Complete packages already read are persisted (partial + /// progress); truncation before the first complete package is a 400. + Eof, + Err(ParseError), +} + +/// Read the next entry package from `cursor`, returning +/// [`PackageOutcome::Eof`] when the buffered body is exhausted (cleanly +/// between packages or mid-package) so the caller treats it as a client +/// truncation. +fn read_package(cursor: &mut Cursor<&[u8]>, num_entries: u64) -> PackageOutcome { + if usize::try_from(cursor.position()).unwrap_or(usize::MAX) >= cursor.get_ref().len() { + return PackageOutcome::Eof; + } + match EntryPackage::read_from(cursor, num_entries) { + Ok(pkg) => PackageOutcome::Ok(pkg), + Err(ParseError::Io(ref e)) if e.kind() == ErrorKind::UnexpectedEof => PackageOutcome::Eof, + Err(e) => PackageOutcome::Err(e), + } +} + +/// Cosign the target checkpoint with the mirror key, persist the served +/// checkpoint object, advance the durable mirror checkpoint via the DO, +/// and return the 200 response carrying the mirror cosignature line(s). +/// +/// Called only once the whole upload is durably committed (spec: the +/// mirror MUST NOT sign until all entries are committed). +/// +/// # Errors +/// +/// Returns an error if the target note/checkpoint fails to parse, signing +/// fails, or a storage/DO write fails. +async fn cosign_and_serve( + env: &Env, + header: &AddEntriesRequestHeader, + target: &PendingCheckpoint, + bucket: &O, +) -> ApiResult { + // The checkpoint text comes from the log-signed pending note; the + // response is the bare cosignature line(s), identical to a witness's + // add-checkpoint response. + let note = Note::from_bytes(&target.signed_note_bytes) + .map_err(|e| Error::from(format!("target note parse: {e:?}")))?; + let cp_text = CheckpointText::from_bytes(note.text()) + .map_err(|e| Error::from(format!("target checkpoint parse: {e:?}")))?; + let note_sig = load_mirror_signer(env)? + .as_checkpoint_signer() + .sign(now_millis(), &cp_text) + .map_err(|e| Error::from(format!("mirror cosign: {e:?}")))?; + let cosig_body = + tlog_witness::serialize_add_checkpoint_response(std::slice::from_ref(¬e_sig)); + + // Persist the served checkpoint (log note + mirror cosignature) before + // advancing the durable mirror checkpoint, so a monitor can never + // observe an advanced size without the matching checkpoint object. + let mut checkpoint_obj = target.signed_note_bytes.clone(); + checkpoint_obj.extend_from_slice(&cosig_body); + commit::write_checkpoint(bucket, checkpoint_obj.clone()).await?; + + dispatch_commit( + env, + &header.log_origin, + &CommitRequest { + size: header.upload_end, + hash: target.hash, + signed_note_bytes: checkpoint_obj, + }, + ) + .await?; + + Ok(( + StatusCode::OK, + [(CONTENT_TYPE, "text/plain; charset=utf-8")], + cosig_body, + ) + .into_response()) +} + +/// Read the persisted-leaf prefix required to verify a non-256-aligned +/// first package: the leaves `[subtree_start, upload_start)` where +/// `subtree_start` is `upload_start` rounded down to a 256 boundary. +/// +/// Returns an empty vec when `upload_start` is already 256-aligned (the +/// common case and every non-first package). Because `upload_start <= +/// next_entry.size` is enforced upstream, the requested leaves are always +/// present in storage. +/// +/// # Errors +/// +/// Returns an error if opening the origin bucket or reading the persisted +/// entry bundle fails. +async fn first_package_prefix( + env: &Env, + header: &AddEntriesRequestHeader, + persisted_size: u64, +) -> Result>> { + let subtree_start = (header.upload_start / PACKAGE_ALIGNMENT) * PACKAGE_ALIGNMENT; + if header.upload_start == subtree_start { + return Ok(Vec::new()); + } + let bucket = load_origin_bucket(env, &header.log_origin)?; + commit::read_persisted_leaves( + &bucket, + subtree_start, + header.upload_start - subtree_start, + persisted_size, + ) + .await +} + +/// POST the [`CommitRequest`] to the per-origin DO, advancing the mirror +/// checkpoint. A non-200 status (or a `/commit` beyond pending) is a +/// frontend/mirror bug, so it is surfaced as a transport error that the +/// handler maps to 500. +async fn dispatch_commit(env: &Env, origin: &str, commit_req: &CommitRequest) -> Result<()> { + let stub = state_stub(env, origin)?; + let mut resp = stub + .fetch_with_request(Request::new_with_init( + "http://do/commit", + &RequestInit { + method: Method::Post, + body: Some(serde_json::to_string(commit_req)?.into()), + headers: { + let h = Headers::new(); + h.set("content-type", "application/json")?; + h + }, + ..Default::default() + }, + )?) + .await?; + match resp.status_code() { + 200 => Ok(()), + status => { + let msg = resp.text().await.unwrap_or_default(); + log::error!("add-entries: DO /commit returned {status}: {msg}"); + Err(Error::from(format!("commit failed ({status})"))) + } + } +} + +/// POST an [`AdvanceNextEntryRequest`] to the per-origin DO, advancing +/// the persisted-entry frontier to `(size, hash)`. Returns the effective +/// frontier (which reflects any concurrent advance). A DO rejection +/// (`size > pending`, a mirror bug) or RPC failure is surfaced as a +/// transport error that the handler maps to 500. +async fn advance_next_entry(env: &Env, origin: &str, size: u64, hash: Hash) -> Result { + let stub = state_stub(env, origin)?; + let req = AdvanceNextEntryRequest { size, hash }; + let mut resp = stub + .fetch_with_request(Request::new_with_init( + "http://do/advance-next-entry", + &RequestInit { + method: Method::Post, + body: Some(serde_json::to_string(&req)?.into()), + headers: { + let h = Headers::new(); + h.set("content-type", "application/json")?; + h + }, + ..Default::default() + }, + )?) + .await?; + match resp.status_code() { + 200 => Ok(resp.json().await?), + status => { + let msg = resp.text().await.unwrap_or_default(); + log::error!("add-entries: DO /advance-next-entry returned {status}: {msg}"); + Err(Error::from(format!("advance-next-entry failed ({status})"))) + } + } +} + +/// Read the per-origin DO state snapshot. A non-200 status or RPC failure +/// is a transport-level error the handler maps to 500. +async fn fetch_snapshot(env: &Env, origin: &str) -> Result { + let stub = state_stub(env, origin)?; + let mut resp = stub + .fetch_with_request(Request::new_with_init( + "http://do/get-state", + &RequestInit { + method: Method::Post, + body: None, + headers: Headers::new(), + ..Default::default() + }, + )?) + .await?; + if resp.status_code() != 200 { + return Err(Error::from(format!( + "DO /get-state returned {}", + resp.status_code() + ))); + } + let snapshot: MirrorStateSnapshot = resp.json().await?; + Ok(snapshot) +} + +/// Resolve the target pending checkpoint that this `add-entries` +/// request is uploading toward. Either: +/// +/// * `upload_end == snapshot.pending.size`: use the current pending. +/// * The ticket round-trips and yields a past pending whose +/// embedded checkpoint has size `upload_end` and verifies against +/// the trusted log keys: use that. +/// +/// Returns `Err(reason)` (a static `&str` describing why) when +/// neither path produces a target. The frontend turns the error into +/// a 409 with `text/x.tlog.mirror-info`. +fn resolve_target_pending( + env: &Env, + header: &AddEntriesRequestHeader, + snapshot: &MirrorStateSnapshot, + verifiers: &signed_note::VerifierList, +) -> std::result::Result { + if header.upload_end == snapshot.pending.size { + // Per spec: `upload_end` must be at-or-above the mirror's + // committed checkpoint. The DO state guarantees pending >= + // committed (the `/commit` RPC enforces it on the write + // side), so checking against pending is sufficient. + if header.upload_end < snapshot.committed.size { + return Err("upload_end below committed checkpoint"); + } + return Ok(snapshot.pending.clone()); + } + + // Try the ticket. An empty ticket can't carry a past pending, so + // there's nothing to fall back to. + if header.ticket.is_empty() { + return Err("upload_end does not match current pending and no ticket provided"); + } + let sealer = match load_ticket_sealer(env) { + Ok(m) => m, + Err(e) => { + // A missing/malformed MIRROR_TICKET_KEY is an operator + // misconfiguration, not a client error. Surface as 409 + // (which we'd return anyway) and log so an operator + // notices. + log::error!("add-entries: ticket sealer unavailable: {e:?}"); + return Err("ticket key unavailable"); + } + }; + // The log origin is bound as associated data, so a ticket minted + // for one log cannot be opened against another. + let Ok(plaintext) = sealer.open(&header.ticket, header.log_origin.as_bytes()) else { + return Err("ticket authentication failed"); + }; + // The ticket plaintext is the full signed-note bytes of a + // previously accepted pending checkpoint. Re-parse and re-verify + // against the trusted log keys; tickets are mirror-keyed so we + // know they came from us, but the *embedded* signature is the + // log's, and we've already established the log key isn't + // self-signed by the ticket key. + let Ok(note) = Note::from_bytes(&plaintext) else { + return Err("ticket plaintext is not a valid signed note"); + }; + if let Err(e) = note.verify(verifiers) { + match e { + NoteError::UnverifiedNote | NoteError::InvalidSignature { .. } => { + return Err("ticket-bound note has no valid signatures from trusted log keys"); + } + _ => return Err("ticket-bound note failed structural verification"), + } + } + let Ok(cp_text) = CheckpointText::from_bytes(note.text()) else { + return Err("ticket-bound note text is not a valid checkpoint"); + }; + if cp_text.origin() != header.log_origin { + return Err("ticket-bound checkpoint has a different origin"); + } + if cp_text.size() != header.upload_end { + return Err("ticket-bound checkpoint size != upload_end"); + } + if cp_text.size() < snapshot.committed.size { + return Err("ticket-bound checkpoint size < committed checkpoint size"); + } + Ok(PendingCheckpoint { + size: cp_text.size(), + hash: *cp_text.hash(), + signed_note_bytes: plaintext, + }) +} + +/// Verify a single [`EntryPackage`] against the target pending +/// checkpoint at `upload_end`. +/// +/// A package's subtree is `[subtree_start, pkg_end)` where `subtree_start` +/// is `pkg_start` rounded down to a [`PACKAGE_ALIGNMENT`] (256) boundary. +/// Only the *first* package of a request can have `pkg_start > +/// subtree_start`, when the client's `upload_start` is not +/// 256-aligned; the leading leaves `[subtree_start, pkg_start)` are then +/// already in the log and supplied here as `prefix_leaves` (read from +/// storage by the caller). For every other package `prefix_leaves` is +/// empty and `subtree_start == pkg_start`. +/// +/// `prefix_leaves` and `pkg.entries` are the raw entry bytes for the +/// contiguous leaves `[subtree_start, pkg_end)`. +/// +/// Returns `Err(reason)` if proof verification fails. +fn verify_package( + prefix_leaves: &[Vec], + pkg: &EntryPackage, + subtree_start: u64, + pkg_end: u64, + target: &PendingCheckpoint, +) -> std::result::Result<(), &'static str> { + debug_assert!( + subtree_start.is_multiple_of(PACKAGE_ALIGNMENT), + "subtree_start must be 256-aligned" + ); + let prefix_len = u64::try_from(prefix_leaves.len()).map_err(|_| "prefix too large")?; + let pkg_start = subtree_start + prefix_len; + let received = u64::try_from(pkg.entries.len()).map_err(|_| "package has too many entries")?; + if received != pkg_end - pkg_start { + return Err("package entry count != range size"); + } + + // Reconstruct the package's subtree hash as a standalone Merkle tree + // over its `count = prefix_len + received` leaves (persisted + // `prefix_leaves` first, then uploaded `pkg.entries`). Leaves are + // replayed with *local* 0-based indices so a subtree-completing leaf + // merges only within the subtree; replaying with absolute indices + // would, at a subtree boundary, reach for a left sibling outside it + // (e.g. leaf 511 of `[256,512)` would reach for `[0,256)`). + let count = prefix_len + received; + let mut store: HashMap = + HashMap::with_capacity(usize::try_from(2 * count).unwrap_or(usize::MAX)); + let mut next_idx = stored_hash_index(0, 0); + for (local_index, entry) in prefix_leaves.iter().chain(pkg.entries.iter()).enumerate() { + let hashes = { + let reader = MapReader { store: &store }; + stored_hashes_for_record_hash( + local_index as u64, + tlog_core::record_hash(entry), + &reader, + ) + .map_err(|_| "failed to compute stored hashes for leaf")? + }; + for h in hashes { + store.insert(next_idx, h); + next_idx += 1; + } + } + let reader = MapReader { store: &store }; + let Ok(pkg_hash) = tree_hash(count, &reader) else { + return Err("failed to compute package subtree hash"); + }; + + // The package's subtree is `[subtree_start, pkg_end)`. `Subtree::new` + // requires `subtree_start` to be aligned to the next-power-of-2 >= + // `pkg_end - subtree_start`; the 256-aligned `subtree_start` and a + // span of at most 256 leaves guarantee that. + let subtree = + Subtree::new(subtree_start, pkg_end).map_err(|_| "package range is not a valid subtree")?; + + // Verify the consistency proof against the target tree size. + if verify_subtree_consistency_proof(&pkg.proof, target.size, target.hash, &subtree, pkg_hash) + .is_err() + { + return Err("subtree consistency proof failed"); + } + + Ok(()) +} + +/// Build a `text/x.tlog.mirror-info` response carrying the mirror's +/// current pending tree size, the advertised `next_entry`, and a sealed +/// ticket, at the given HTTP `status`. +/// +/// The ticket is sealed via [`tlog_mirror::TicketSealer`] (AES-256-GCM-SIV, log +/// origin bound as associated data) so the client can present it on +/// retry to recover the pending state without keeping it in DO storage. +/// If sealing fails (operator misconfigured `MIRROR_TICKET_KEY`), the +/// response still carries an empty ticket; the client falls back to a +/// `(0, 0)` initial query. +/// +/// Two status codes use this shape (see [`mirror_info_409`] / +/// [`mirror_info_202`]): +/// +/// * `409 Conflict`: the request could not be applied (stale +/// `upload_start`/`upload_end`, no matching pending). `next_entry` +/// reports the persisted frontier so the client can resume. +/// * `202 Accepted`: a partial run of packages was persisted; +/// `next_entry` reports the *advanced* frontier so the client +/// continues from there. +fn mirror_info_response( + env: &Env, + snapshot: &MirrorStateSnapshot, + origin: &str, + status: StatusCode, + next_entry: u64, +) -> axum::response::Response { + let ticket = if snapshot.pending.signed_note_bytes.is_empty() { + Vec::new() + } else { + match load_ticket_sealer(env) { + // Bind the log origin as associated data so the ticket can + // only be reopened for the same log. + Ok(m) => m.seal(&snapshot.pending.signed_note_bytes, origin.as_bytes()), + Err(e) => { + log::error!("add-entries: cannot seal ticket: {e:?}"); + Vec::new() + } + } + }; + let info = MirrorInfo { + tree_size: snapshot.pending.size, + next_entry, + ticket, + }; + ( + status, + [(CONTENT_TYPE, MIRROR_INFO_CONTENT_TYPE)], + info.to_body(), + ) + .into_response() +} + +/// 409 Conflict carrying the mirror's current state. Advertises the +/// persisted frontier (`next_entry.size`) as the resume point. +fn mirror_info_409( + env: &Env, + snapshot: &MirrorStateSnapshot, + origin: &str, +) -> axum::response::Response { + mirror_info_response( + env, + snapshot, + origin, + StatusCode::CONFLICT, + snapshot.next_entry.size, + ) +} + +/// 202 Accepted after a partial persist. Advertises the freshly-advanced +/// persisted frontier so the client resumes appending from there. +fn mirror_info_202( + env: &Env, + snapshot: &MirrorStateSnapshot, + origin: &str, + next_entry: u64, +) -> axum::response::Response { + mirror_info_response(env, snapshot, origin, StatusCode::ACCEPTED, next_entry) +} + +/// A [`HashReader`] backed by a sparse `HashMap` indexed +/// by absolute stored-hash index. Used during package verification +/// to reconstruct the subtree hash from leaves we just received, +/// without needing access to the mirror's full storage backend. +struct MapReader<'a> { + store: &'a HashMap, +} + +impl HashReader for MapReader<'_> { + fn read_hashes(&self, indexes: &[u64]) -> std::result::Result, TlogError> { + indexes + .iter() + .map(|i| { + self.store + .get(i) + .copied() + .ok_or(TlogError::IndexesNotInTree) + }) + .collect() + } +} + +#[cfg(test)] +mod tests { + use super::{MapReader, verify_package}; + use crate::mirror_state_do::PendingCheckpoint; + use std::collections::HashMap; + use tlog_core::{Hash, Subtree, stored_hash_index, stored_hashes, tree_hash}; + use tlog_mirror::EntryPackage; + + /// Deterministic distinct entry bytes for leaf `i`. + fn entry(i: u64) -> Vec { + format!("entry-{i}").into_bytes() + } + + /// Build a full in-memory stored-hash store for the first `n` leaves. + fn build_store(n: u64) -> HashMap { + let mut store: HashMap = HashMap::new(); + for i in 0..n { + let hashes = stored_hashes(i, &entry(i), &MapReader { store: &store }).unwrap(); + for (j, h) in hashes.iter().enumerate() { + store.insert(stored_hash_index(0, i) + j as u64, *h); + } + } + store + } + + /// Construct the target pending checkpoint and a verified package for + /// the subtree `[subtree_start, pkg_end)` of a tree of size `target`. + /// `pkg_start` is where the *uploaded* entries begin, so leaves + /// `[subtree_start, pkg_start)` become the persisted prefix. + fn fixture( + target: u64, + subtree_start: u64, + pkg_start: u64, + pkg_end: u64, + ) -> (Vec>, EntryPackage, PendingCheckpoint) { + let store = build_store(target); + let reader = MapReader { store: &store }; + let target_hash = tree_hash(target, &reader).unwrap(); + let subtree = Subtree::new(subtree_start, pkg_end).unwrap(); + let proof = tlog_core::subtree_consistency_proof(target, &subtree, &reader).unwrap(); + + let prefix: Vec> = (subtree_start..pkg_start).map(entry).collect(); + let pkg = EntryPackage { + entries: (pkg_start..pkg_end).map(entry).collect(), + proof, + }; + let cp = PendingCheckpoint { + size: target, + hash: target_hash, + signed_note_bytes: Vec::new(), + }; + (prefix, pkg, cp) + } + + #[test] + fn verify_aligned_package_ok() { + // Aligned first package: no persisted prefix. + let (prefix, pkg, cp) = fixture(1000, 256, 256, 512); + assert!(prefix.is_empty()); + verify_package(&prefix, &pkg, 256, 512, &cp).expect("aligned package verifies"); + } + + #[test] + fn verify_nonaligned_first_package_with_prefix_ok() { + // upload_start = 300 (non-aligned): subtree starts at 256, and the + // persisted leaves [256, 300) are supplied as the prefix. + let (prefix, pkg, cp) = fixture(1000, 256, 300, 512); + assert_eq!(prefix.len(), 44); + verify_package(&prefix, &pkg, 256, 512, &cp).expect("non-aligned package verifies"); + } + + #[test] + fn verify_first_ever_package_from_zero_ok() { + // Subtree rooted at 0 (first bundle), partial last package. + let (prefix, pkg, cp) = fixture(300, 0, 0, 256); + verify_package(&prefix, &pkg, 0, 256, &cp).expect("first bundle verifies"); + } + + #[test] + fn verify_rejects_wrong_prefix() { + // A prefix leaf that doesn't match the persisted entry yields the + // wrong subtree hash, so proof verification must fail. + let (mut prefix, pkg, cp) = fixture(1000, 256, 300, 512); + prefix[0] = b"tampered".to_vec(); + assert!(verify_package(&prefix, &pkg, 256, 512, &cp).is_err()); + } + + #[test] + fn verify_rejects_wrong_target_hash() { + let (prefix, pkg, mut cp) = fixture(1000, 256, 300, 512); + cp.hash = Hash([0xab; tlog_core::HASH_SIZE]); + assert!(verify_package(&prefix, &pkg, 256, 512, &cp).is_err()); + } + + #[test] + fn verify_rejects_entry_count_mismatch() { + let (prefix, mut pkg, cp) = fixture(1000, 256, 300, 512); + pkg.entries.push(entry(999)); // one too many + assert_eq!( + verify_package(&prefix, &pkg, 256, 512, &cp), + Err("package entry count != range size") + ); + } + + #[test] + fn verify_rejects_tampered_entry() { + let (prefix, mut pkg, cp) = fixture(1000, 256, 300, 512); + pkg.entries[0] = b"not the real entry".to_vec(); + assert!(verify_package(&prefix, &pkg, 256, 512, &cp).is_err()); + } +} diff --git a/crates/mirror_worker/src/body.rs b/crates/mirror_worker/src/body.rs new file mode 100644 index 00000000..298e50e5 --- /dev/null +++ b/crates/mirror_worker/src/body.rs @@ -0,0 +1,127 @@ +// Copyright (c) 2025-2026 Cloudflare, Inc. All rights reserved. +// SPDX-License-Identifier: BSD-3-Clause + +//! Request-body decoding for `add-entries`. +//! +//! [c2sp.org/tlog-mirror][spec] requires mirrors to accept +//! `Content-Encoding: gzip` request bodies (see [Request Body][reqbody]); +//! clients MAY send gzip without negotiating first. The Cloudflare Workers +//! runtime does not transparently decompress request bodies (unlike +//! responses, which it (de)compresses based on `Accept-Encoding`), so the +//! mirror must gunzip the body itself. +//! +//! This module reads the whole body into memory and inflates it in one +//! pass. A follow-up commit replaces this with incremental streaming so a +//! large upload does not have to buffer the entire body (Workers isolates +//! have a ~128 MB memory ceiling). +//! +//! [spec]: https://c2sp.org/tlog-mirror#add-entries +//! [reqbody]: https://c2sp.org/tlog-mirror#request-body + +use std::io::Read as _; + +use flate2::read::GzDecoder; +use futures_util::StreamExt as _; +#[allow(clippy::wildcard_imports)] +use worker::*; + +use crate::frontend_worker::{ApiResult, AppError}; + +/// Read the whole request body and decode it per `Content-Encoding`. +/// +/// `identity` (or an absent header) returns the body unchanged; +/// `gzip`/`x-gzip` is inflated. Any other encoding is unsupported: the +/// mirror can't authenticate a body it can't read, so this returns 415. +/// +/// # Errors +/// +/// Returns [`AppError::UnsupportedMediaType`] for an unrecognized +/// `Content-Encoding`, [`AppError::BadRequest`] for a malformed/truncated +/// gzip body, or a transport error while reading the body stream. +pub(crate) async fn read_decoded_body( + headers: &axum::http::HeaderMap, + body: axum::body::Body, +) -> ApiResult> { + let encoding = headers + .get(axum::http::header::CONTENT_ENCODING) + .and_then(|v| v.to_str().ok()) + .unwrap_or_default() + .trim() + .to_ascii_lowercase(); + + let mut raw = Vec::new(); + let mut stream = body.into_data_stream(); + while let Some(chunk) = stream.next().await { + let chunk = chunk.map_err(|e| Error::from(e.to_string()))?; + raw.extend_from_slice(&chunk); + } + + match encoding.as_str() { + "" | "identity" => Ok(raw), + "gzip" | "x-gzip" => { + let mut out = Vec::new(); + GzDecoder::new(raw.as_slice()) + .read_to_end(&mut out) + .map_err(|e| AppError::BadRequest(format!("gzip decode failed: {e}")))?; + Ok(out) + } + other => Err(AppError::UnsupportedMediaType(format!( + "Unsupported Content-Encoding: {other}" + ))), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use flate2::Compression; + use flate2::write::GzEncoder; + use std::io::Write as _; + + /// gzip-compress `data` into a single buffer for test input. + fn gzip(data: &[u8]) -> Vec { + let mut enc = GzEncoder::new(Vec::new(), Compression::default()); + enc.write_all(data).unwrap(); + enc.finish().unwrap() + } + + /// Inflate `compressed` the same way [`read_decoded_body`] does. + fn gunzip(compressed: &[u8]) -> Result> { + let mut out = Vec::new(); + GzDecoder::new(compressed) + .read_to_end(&mut out) + .map_err(|e| Error::from(format!("gzip decode failed: {e}")))?; + Ok(out) + } + + #[test] + fn gzip_roundtrips() { + let mut plain: Vec = Vec::new(); + for i in 0..20_000u32 { + plain.extend_from_slice(format!("entry-{i};").as_bytes()); + } + let decoded = gunzip(&gzip(&plain)).expect("roundtrip"); + assert_eq!(decoded, plain); + } + + #[test] + fn empty_payload_roundtrips() { + let decoded = gunzip(&gzip(b"")).expect("roundtrip"); + assert!(decoded.is_empty()); + } + + #[test] + fn truncated_gzip_errors() { + let mut compressed = gzip(b"the quick brown fox jumps over the lazy dog"); + compressed.truncate(compressed.len() - 6); + assert!(gunzip(&compressed).is_err(), "truncated gzip must error"); + } + + #[test] + fn corrupt_gzip_errors() { + let mut compressed = gzip(b"hello world, this is a test payload for corruption"); + let mid = compressed.len() / 2; + compressed[mid] ^= 0xff; + assert!(gunzip(&compressed).is_err(), "corrupt gzip must error"); + } +} diff --git a/crates/mirror_worker/src/commit.rs b/crates/mirror_worker/src/commit.rs new file mode 100644 index 00000000..08ed72bc --- /dev/null +++ b/crates/mirror_worker/src/commit.rs @@ -0,0 +1,466 @@ +// Copyright (c) 2025-2026 Cloudflare, Inc. All rights reserved. +// SPDX-License-Identifier: BSD-3-Clause + +//! Commit path for `add-entries`: persist verified entries as +//! [tlog-tiles][tiles] entry bundles and (re)compute the Merkle hash +//! tiles, growing the mirrored copy of a log from its persisted frontier +//! up to a pending checkpoint. +//! +//! One commit maps to one `add-entries` request: [`persist_entries`] runs +//! once per request over the full batch of entries received across that +//! request's packages, so the persisted frontier is read from R2 once per +//! request (see [`read_edge_tiles`]), not once per package or entry. +//! +//! This mirrors the sequencer's append path +//! ([`generic_log_worker::log_ops`]'s `sequence_entries`), with two +//! differences: the mirror is stateless between requests, so +//! [`persist_entries`] re-reads and authenticates the persisted frontier +//! from R2 on every commit rather than caching it; and it stores the +//! tlog-tiles entry-bundle framing directly (each leaf `uint16_be(len) || +//! entry`, leaf hash `record_hash(entry)`) instead of owning an +//! application entry type. +//! +//! Commits are safe to interrupt and repeat: entry bundles and hash tiles +//! are immutable and content-addressed, so re-uploading one is a harmless +//! identical overwrite. The caller advances the durable mirror checkpoint +//! only after [`persist_entries`] returns and the recomputed root matches. +//! +//! [tiles]: https://c2sp.org/tlog-tiles + +use std::collections::HashMap; + +use generic_log_worker::{ + ObjectBackend, + log_ops::{HashReaderWithOverlay, TileWithBytes, UploadOptions, read_edge_tiles}, +}; +use length_prefixed::{ReadLengthPrefixedBytesExt as _, WriteLengthPrefixedBytesExt as _}; +use tlog_core::{Hash, record_hash, stored_hash_index, stored_hashes_for_record_hash}; +use tlog_tiles::{PathElem, TlogTile}; +#[allow(clippy::wildcard_imports)] +use worker::*; + +/// tlog-tiles fixes a tile height of 8, i.e. 256 entries per full tile. +const TILE_WIDTH: u64 = TlogTile::FULL_WIDTH as u64; + +/// Object key for the (cosigned) checkpoint the mirror serves at +/// `//checkpoint`. Matches +/// [`generic_log_worker::log_ops::CHECKPOINT_KEY`]. +pub(crate) const CHECKPOINT_KEY: &str = "checkpoint"; + +/// Read `count` already-persisted log entries starting at leaf index +/// `start` back out of their entry bundle in object storage. +/// +/// `start` MUST be aligned to a 256-entry bundle boundary and `start + +/// count` MUST NOT exceed `persisted_size`, so the requested leaves all +/// live in the single entry bundle beginning at `start` (a package's +/// subtree spans at most 256 leaves). Returns the entries in order, +/// stripped of their uint16 length prefixes. +/// +/// Used by [`crate::add_entries`] to reconstruct the subtree hash of a +/// non-256-aligned first package, whose leading leaves `[subtree_start, +/// upload_start)` are already in the log and therefore absent from the +/// uploaded package. +/// +/// # Errors +/// +/// Returns an error if the bundle is missing from storage or is shorter +/// than `count` entries (i.e. the requested leaves were not actually +/// persisted). +pub(crate) async fn read_persisted_leaves( + object: &impl ObjectBackend, + start: u64, + count: u64, + persisted_size: u64, +) -> Result>> { + debug_assert!( + start.is_multiple_of(TILE_WIDTH), + "start must be bundle-aligned" + ); + debug_assert!(start + count <= persisted_size, "leaves must be persisted"); + + // The bundle's stored width is 256 if the whole tile is persisted, + // otherwise the trailing partial width `persisted_size - start`. The + // data-tile path encodes that width (`from_index` derives it from the + // last stored leaf), so pick the last persisted leaf in this tile. + let stored_width = TILE_WIDTH.min(persisted_size - start); + let last_leaf = start + stored_width - 1; + let tile = + TlogTile::from_index(stored_hash_index(0, last_leaf)).with_data_path(PathElem::Entries); + let bytes = object + .fetch(tile.path()) + .await? + .ok_or_else(|| Error::from(format!("persisted entry bundle missing: {}", tile.path())))?; + + let mut cur: &[u8] = &bytes; + let mut out = Vec::with_capacity(usize::try_from(count).unwrap_or(0)); + for i in 0..count { + let entry = cur.read_length_prefixed(2).map_err(|e| { + Error::from(format!( + "persisted bundle leaf {} truncated: {e}", + start + i + )) + })?; + out.push(entry); + } + Ok(out) +} + +/// Serialize one entry into its tlog-tiles entry-bundle framing (a +/// big-endian uint16 length prefix followed by the entry bytes) and +/// append it to `buf`. Matches `tlog_entry`'s `to_data_tile_entry` so +/// the mirror's bundles are byte-identical to a native tlog-tiles log's. +/// +/// # Errors +/// +/// Returns an error if `entry` exceeds 65535 bytes (the u16 length +/// prefix limit). Entries were parsed off the wire with the same uint16 +/// framing, so in practice this never fires. +fn push_tile_leaf(buf: &mut Vec, entry: &[u8]) -> Result<()> { + buf.write_length_prefixed(entry, 2) + .map_err(|e| Error::from(format!("entry too large for tile bundle: {e}"))) +} + +/// The `UploadOptions` for an immutable, content-addressed tile (entry +/// bundle or hash tile). +fn immutable_tile_opts() -> UploadOptions { + UploadOptions { + content_type: Some("application/octet-stream".to_owned()), + immutable: true, + } +} + +/// Persist entries `[persisted_size, target_size)` into object storage, +/// growing the mirrored tree and returning the recomputed root hash. +/// +/// Called once per `add-entries` request; `entries` spans all packages +/// received in that request. `entries[i]` MUST be the raw log-entry bytes +/// (no length prefix) for leaf `persisted_size + i`, covering exactly +/// `[persisted_size, target_size)`. +/// +/// Reads the persisted frontier from R2 (skipped at size 0), replays each +/// leaf through [`stored_hashes_for_record_hash`] while flushing full and +/// trailing-partial entry bundles, then (re)computes and uploads every +/// hash tile in [`TlogTile::new_tiles`]`(persisted_size, target_size)`. +/// +/// It does not write the checkpoint object or advance the mirror +/// checkpoint; the caller does that after verifying the returned root. +/// +/// # Errors +/// +/// Returns an error on any storage failure, if a persisted tile is +/// missing or fails authentication, or if `entries.len()` does not equal +/// `target_size - persisted_size`. +pub(crate) async fn persist_entries( + object: &impl ObjectBackend, + persisted_size: u64, + persisted_hash: Hash, + target_size: u64, + entries: &[Vec], +) -> Result { + let expected = target_size + .checked_sub(persisted_size) + .ok_or_else(|| Error::from("target_size < persisted_size"))?; + if entries.len() as u64 != expected { + return Err(Error::from(format!( + "commit entry count {} != range {persisted_size}..{target_size}", + entries.len() + ))); + } + if expected == 0 { + return Ok(persisted_hash); + } + + // Genesis (persisted_size 0): no persisted frontier exists yet, so there + // are no edge tiles to read; start from an empty overlay. + let mut edge_tiles = if persisted_size == 0 { + HashMap::new() + } else { + read_edge_tiles(object, persisted_size, &persisted_hash).await? + }; + + // Load the current partial entry bundle so we extend rather than + // overwrite it. Only exists when the frontier is mid-tile. + let mut data_tile = Vec::new(); + if persisted_size > 0 && !persisted_size.is_multiple_of(TILE_WIDTH) { + let partial = TlogTile::from_index(stored_hash_index(0, persisted_size - 1)) + .with_data_path(PathElem::Entries); + data_tile = object.fetch(partial.path()).await?.ok_or_else(|| { + Error::from(format!("partial entry bundle missing: {}", partial.path())) + })?; + } + + // Replay leaves, flushing entry bundles at 256-entry boundaries. + let mut overlay: HashMap = HashMap::new(); + let mut n = persisted_size; + for entry in entries { + push_tile_leaf(&mut data_tile, entry)?; + let hashes = stored_hashes_for_record_hash( + n, + record_hash(entry), + &HashReaderWithOverlay { + edge_tiles: &edge_tiles, + overlay: &overlay, + }, + ) + .map_err(|e| Error::from(format!("couldn't compute hashes for leaf {n}: {e}")))?; + for (i, h) in hashes.iter().enumerate() { + overlay.insert(stored_hash_index(0, n) + i as u64, *h); + } + n += 1; + if n.is_multiple_of(TILE_WIDTH) { + upload_entry_bundle(object, n, std::mem::take(&mut data_tile)).await?; + } + } + debug_assert_eq!(n, target_size); + // Trailing partial entry bundle. + if !target_size.is_multiple_of(TILE_WIDTH) { + upload_entry_bundle(object, target_size, std::mem::take(&mut data_tile)).await?; + } + + // (Re)compute and upload hash tiles. + for tile in TlogTile::new_tiles(persisted_size, target_size) { + let bytes = tile + .read_data(&HashReaderWithOverlay { + edge_tiles: &edge_tiles, + overlay: &overlay, + }) + .map_err(|e| Error::from(format!("couldn't build hash tile {tile:?}: {e}")))?; + object + .upload(tile.path(), bytes.clone(), &immutable_tile_opts()) + .await?; + // Keep edge_tiles current so read_data of a higher/next tile can + // still resolve persisted hashes it depends on. + edge_tiles.insert(tile.level(), TileWithBytes { tile, b: bytes }); + } + + // Recompute the root hash from the frontier we just built. + tlog_core::tree_hash( + target_size, + &HashReaderWithOverlay { + edge_tiles: &edge_tiles, + overlay: &overlay, + }, + ) + .map_err(|e| Error::from(format!("couldn't compute root hash: {e}"))) +} + +/// Write the mirror's served checkpoint object. Unlike tiles the +/// checkpoint is mutable (it advances as the mirror commits), so it is +/// stored with `no-store` caching (`immutable: false`). +/// +/// `bytes` MUST be the checkpoint note the mirror serves at +/// `//checkpoint`: the origin log's signed +/// checkpoint with the mirror's own cosignature line(s) appended. +/// +/// # Errors +/// +/// Returns an error if the storage write fails. +pub(crate) async fn write_checkpoint(object: &impl ObjectBackend, bytes: Vec) -> Result<()> { + object + .upload( + CHECKPOINT_KEY, + bytes, + &UploadOptions { + content_type: Some("text/plain; charset=utf-8".to_owned()), + immutable: false, + }, + ) + .await +} + +/// Upload one entry bundle (data tile). `n` is the tree size after the +/// bundle's last entry, so the bundle covers leaves ending at `n - 1`. +async fn upload_entry_bundle(object: &impl ObjectBackend, n: u64, bytes: Vec) -> Result<()> { + let tile = TlogTile::from_index(stored_hash_index(0, n - 1)).with_data_path(PathElem::Entries); + object + .upload(tile.path(), bytes, &immutable_tile_opts()) + .await +} + +#[cfg(test)] +mod tests { + use super::*; + use std::cell::RefCell; + use std::collections::HashMap; + use tlog_core::{EMPTY_HASH, HashReader, TlogError, stored_hashes}; + use tlog_tiles::{PreloadedTlogTileReader, TileHashReader, TlogTileRecorder}; + + /// In-memory [`ObjectBackend`] for exercising the commit path without + /// a live R2 bucket. + #[derive(Default)] + struct MemBackend { + store: RefCell>>, + } + + impl ObjectBackend for MemBackend { + async fn upload, D: Into>>( + &self, + key: S, + data: D, + _opts: &UploadOptions, + ) -> Result<()> { + self.store + .borrow_mut() + .insert(key.as_ref().to_owned(), data.into()); + Ok(()) + } + + async fn fetch>(&self, key: S) -> Result>> { + Ok(self.store.borrow().get(key.as_ref()).cloned()) + } + } + + /// Deterministic distinct entry bytes for leaf `i`. + fn entry(i: u64) -> Vec { + format!("entry-{i}").into_bytes() + } + + /// A [`HashReader`] over an in-memory stored-hash map, used to build + /// the reference tree hash independently of the commit path. + struct MapReader<'a>(&'a HashMap); + impl HashReader for MapReader<'_> { + fn read_hashes(&self, idx: &[u64]) -> std::result::Result, TlogError> { + idx.iter() + .map(|i| self.0.get(i).copied().ok_or(TlogError::IndexesNotInTree)) + .collect() + } + } + + /// Compute the reference tree hash for the first `n` leaves by + /// building a full in-memory stored-hash map with [`tlog_core`], so + /// we can check that what the commit path stored is correct. + fn reference_root(n: u64) -> Hash { + let mut store: HashMap = HashMap::new(); + for i in 0..n { + let hashes = stored_hashes(i, &entry(i), &MapReader(&store)).unwrap(); + for (j, h) in hashes.iter().enumerate() { + store.insert(stored_hash_index(0, i) + j as u64, *h); + } + } + tlog_core::tree_hash(n, &MapReader(&store)).unwrap() + } + + fn leaves(range: std::ops::Range) -> Vec> { + range.map(entry).collect() + } + + #[tokio::test] + async fn commit_from_empty_matches_reference() { + let obj = MemBackend::default(); + // 300 leaves crosses one full tile (256) + a 44-wide partial. + let root = persist_entries(&obj, 0, EMPTY_HASH, 300, &leaves(0..300)) + .await + .unwrap(); + assert_eq!(root, reference_root(300)); + } + + #[tokio::test] + async fn incremental_commit_reads_persisted_frontier() { + let obj = MemBackend::default(); + // First commit to a mid-tile size (300). + let root0 = persist_entries(&obj, 0, EMPTY_HASH, 300, &leaves(0..300)) + .await + .unwrap(); + assert_eq!(root0, reference_root(300)); + + // Second commit continues from the mid-tile frontier, exercising + // read_edge_tiles + partial-entry-bundle reload from storage. + let root1 = persist_entries(&obj, 300, root0, 800, &leaves(300..800)) + .await + .unwrap(); + assert_eq!(root1, reference_root(800)); + } + + #[tokio::test] + async fn entry_bundles_roundtrip() { + use length_prefixed::ReadLengthPrefixedBytesExt as _; + let obj = MemBackend::default(); + persist_entries(&obj, 0, EMPTY_HASH, 260, &leaves(0..260)) + .await + .unwrap(); + + // First full bundle: leaves [0, 256). + let full = TlogTile::from_index(stored_hash_index(0, 255)) + .with_data_path(PathElem::Entries) + .path(); + let bytes = obj.fetch(&full).await.unwrap().expect("full bundle stored"); + let mut cur: &[u8] = &bytes; + for i in 0..256u64 { + let got = cur.read_length_prefixed(2).unwrap(); + assert_eq!(got, entry(i), "leaf {i} mismatch in full bundle"); + } + assert!(cur.is_empty(), "full bundle has trailing bytes"); + + // Trailing partial bundle: leaves [256, 260). + let partial = TlogTile::from_index(stored_hash_index(0, 259)) + .with_data_path(PathElem::Entries) + .path(); + let bytes = obj + .fetch(&partial) + .await + .unwrap() + .expect("partial bundle stored"); + let mut cur: &[u8] = &bytes; + for i in 256..260u64 { + let got = cur.read_length_prefixed(2).unwrap(); + assert_eq!(got, entry(i), "leaf {i} mismatch in partial bundle"); + } + assert!(cur.is_empty(), "partial bundle has trailing bytes"); + } + + #[tokio::test] + async fn read_persisted_leaves_from_full_and_partial_bundles() { + let obj = MemBackend::default(); + // 300 leaves: one full bundle [0,256) + a partial bundle [256,300). + persist_entries(&obj, 0, EMPTY_HASH, 300, &leaves(0..300)) + .await + .unwrap(); + + // Prefix within the full bundle (persisted_size 300 -> stored + // width 256 for tile 0). + let got = read_persisted_leaves(&obj, 0, 44, 300).await.unwrap(); + assert_eq!(got, leaves(0..44)); + + // Whole full bundle. + let got = read_persisted_leaves(&obj, 0, 256, 300).await.unwrap(); + assert_eq!(got, leaves(0..256)); + + // Prefix within the trailing partial bundle (tile 1, stored width + // 300 - 256 = 44). + let got = read_persisted_leaves(&obj, 256, 20, 300).await.unwrap(); + assert_eq!(got, leaves(256..276)); + } + + #[tokio::test] + async fn read_persisted_leaves_missing_bundle_errors() { + let obj = MemBackend::default(); + assert!(read_persisted_leaves(&obj, 0, 10, 300).await.is_err()); + } + + #[tokio::test] + async fn hash_tiles_authenticate_against_root() { + let obj = MemBackend::default(); + let root = persist_entries(&obj, 0, EMPTY_HASH, 500, &leaves(0..500)) + .await + .unwrap(); + + // A TileHashReader over the stored tiles must authenticate an + // arbitrary leaf hash against the recomputed root. + let idx = [stored_hash_index(0, 499)]; + let recorder = TlogTileRecorder::default(); + let probe = TileHashReader::new(500, root, &recorder); + assert!(matches!( + probe.read_hashes(&idx), + Err(TlogError::RecordedTilesOnly) + )); + let mut fetched: HashMap> = HashMap::new(); + for tile in recorder.0.into_inner() { + let bytes = obj.fetch(tile.path()).await.unwrap().expect("tile stored"); + fetched.insert(tile, bytes); + } + let reader = PreloadedTlogTileReader(fetched); + let hash_reader = TileHashReader::new(500, root, &reader); + let got = hash_reader.read_hashes(&idx).expect("authenticates"); + assert_eq!(got[0], record_hash(&entry(499))); + } +} diff --git a/crates/mirror_worker/src/frontend_worker.rs b/crates/mirror_worker/src/frontend_worker.rs index a6fb6f77..b7ea7c51 100644 --- a/crates/mirror_worker/src/frontend_worker.rs +++ b/crates/mirror_worker/src/frontend_worker.rs @@ -69,12 +69,17 @@ async fn fetch( // handler is captured and shipped before the WASM isolate is torn // down. `Router`'s `Service::Error` is `Infallible`; the `?` below // performs the trivial conversion into `worker::Error`. + // + // `/add-entries` streams a potentially large (optionally gzip) body, + // so it uses the raw `Request` extractor with no `DefaultBodyLimit`; + // the buffered endpoints cap their bodies via the layer. let response = generic_log_worker::obs::sentry::catch_unwind_and_flush(async { Router::new() .route( "/add-checkpoint", post(add_checkpoint).layer(DefaultBodyLimit::max(MAX_ADD_CHECKPOINT_BODY_SIZE)), ) + .route("/add-entries", post(crate::add_entries::add_entries)) .route("/metadata", get(metadata)) .route("/", get(root)) .with_state(env) @@ -97,16 +102,20 @@ async fn root() -> impl IntoResponse { } /// Error type for the mirror's axum handlers, mapped to an HTTP status by -/// [`IntoResponse`]. -enum AppError { +/// [`IntoResponse`]. Success and special-body responses (the 200 +/// cosignature, the 409/202 `text/x.tlog.mirror-info` and `text/x.tlog.size` +/// bodies) are built as axum responses directly, not via this enum. +pub(crate) enum AppError { InternalServerError(String), BadRequest(String), + UnsupportedMediaType(String), + UnprocessableEntity(String), UnknownLogOrigin, NoValidSignatures, } /// Result type for the mirror's axum handlers. -type ApiResult = std::result::Result; +pub(crate) type ApiResult = std::result::Result; impl From for AppError { fn from(err: worker::Error) -> Self { @@ -124,6 +133,14 @@ impl IntoResponse for AppError { AppError::BadRequest(e) => { (StatusCode::BAD_REQUEST, format!("Bad request: {e}")).into_response() } + AppError::UnsupportedMediaType(e) => { + (StatusCode::UNSUPPORTED_MEDIA_TYPE, e).into_response() + } + AppError::UnprocessableEntity(e) => ( + StatusCode::UNPROCESSABLE_ENTITY, + format!("Unprocessable Entity: {e}"), + ) + .into_response(), AppError::UnknownLogOrigin => { (StatusCode::NOT_FOUND, "Unknown log origin").into_response() } diff --git a/crates/mirror_worker/src/lib.rs b/crates/mirror_worker/src/lib.rs index a839e009..54af74a4 100644 --- a/crates/mirror_worker/src/lib.rs +++ b/crates/mirror_worker/src/lib.rs @@ -4,9 +4,11 @@ //! A transparency-log mirror implementing [c2sp.org/tlog-mirror][mirror] on //! Cloudflare Workers, specialized for MTC issuance logs. //! -//! This worker handles the [`add-checkpoint`][add-cp] submission endpoint, -//! which updates the pending checkpoint for a log origin, and publishes -//! the mirror's identity and per-log configuration at `/metadata`. +//! This worker handles the [`add-checkpoint`][add-cp] and +//! [`add-entries`][add-e] submission endpoints, and publishes the mirror's +//! identity and per-log configuration at `/metadata`. The +//! [tlog-tiles][tiles] read interface is served directly from object +//! storage (see the `storage` module). //! //! Per-origin persistent state lives in a `MirrorState` Durable Object, //! one per log origin. Its single-threaded execution model provides the @@ -15,7 +17,10 @@ //! //! [mirror]: https://c2sp.org/tlog-mirror //! [add-cp]: https://c2sp.org/tlog-mirror#add-checkpoint +//! [add-e]: https://c2sp.org/tlog-mirror#add-entries +//! [tiles]: https://c2sp.org/tlog-tiles +use base64::Engine as _; use config::AppConfig; use ml_dsa::pkcs8::{DecodePrivateKey as _, EncodePublicKey as _}; use ml_dsa::{MlDsa44, VerifyingKey as MlDsaVerifyingKey}; @@ -23,12 +28,17 @@ use pkcs8::{PrivateKeyInfoRef, SecretDocument, der::oid::db::fips204::ID_ML_DSA_ use signed_note::{KeyName, NoteVerifier, VerifierList}; use std::collections::HashMap; use std::sync::{Arc, LazyLock, OnceLock}; -use tlog_cosignature::SubtreeV1NoteVerifier; +use tlog_cosignature::{SubtreeV1CheckpointSigner, SubtreeV1NoteVerifier}; +use tlog_mirror::TicketSealer; #[allow(clippy::wildcard_imports)] use worker::*; +mod add_entries; +mod body; +mod commit; mod frontend_worker; mod mirror_state_do; +mod storage; /// The binding name used in `wrangler.jsonc` for the `MirrorState` DO. pub(crate) const MIRROR_STATE_BINDING: &str = "MIRROR_STATE"; @@ -124,12 +134,15 @@ pub(crate) fn log_verifiers(origin: &str) -> Option { /// /// The mirror is an MTC cosigner, which per [c2sp.org/mtc-tlog][mtc] MUST /// use an ML-DSA-44 key and produce [`subtree/v1`][cosig] messages, so -/// this worker supports only that algorithm. Holds the DER-encoded -/// `SubjectPublicKeyInfo` computed once at load and served by `/metadata`. +/// this worker supports only that algorithm. Holds the signer plus the +/// DER-encoded `SubjectPublicKeyInfo` computed once at load and served by +/// `/metadata`. The signer is boxed because the expanded ML-DSA-44 key is +/// large (~64 KiB). /// /// [mtc]: https://c2sp.org/mtc-tlog /// [cosig]: https://c2sp.org/tlog-cosignature pub(crate) struct MirrorSigner { + signer: Box, public_key_der: Vec, } @@ -147,6 +160,15 @@ impl MirrorSigner { pub(crate) fn algorithm(&self) -> &'static str { "subtree/v1" } + + /// The inner [`CheckpointSigner`] trait object, used by the + /// `add-entries` handler to emit the mirror's checkpoint cosignature + /// on a successful upload. + /// + /// [`CheckpointSigner`]: tlog_checkpoint::CheckpointSigner + pub(crate) fn as_checkpoint_signer(&self) -> &dyn tlog_checkpoint::CheckpointSigner { + &*self.signer + } } /// Cached mirror signer, so the PKCS#8 parse happens at most once per @@ -178,6 +200,8 @@ pub(crate) fn load_mirror_signer(env: &Env) -> Result<&'static MirrorSigner> { /// any other algorithm is rejected (the mirror's cosigner must be an MTC /// cosigner, see [`MirrorSigner`]). fn build_mirror_signer(pem: &str) -> Result { + let name = KeyName::new(CONFIG.mirror_name.clone()) + .map_err(|e| Error::from(format!("invalid mirror_name: {e:?}")))?; let (_label, doc) = SecretDocument::from_pem(pem).map_err(|e| Error::from(format!("PEM parse: {e}")))?; let pk_info = PrivateKeyInfoRef::try_from(doc.as_bytes()) @@ -185,7 +209,8 @@ fn build_mirror_signer(pem: &str) -> Result { match pk_info.algorithm.oid { ID_ML_DSA_44 => { // ml-dsa's PKCS#8 stores only the 32-byte seed; `from_pkcs8_der` - // expands it on the way in. + // expands it on the way in. The expanded key never leaves this + // worker. let expanded = ml_dsa::ExpandedSigningKey::::from_pkcs8_der(doc.as_bytes()) .map_err(|e| Error::from(format!("ML-DSA-44 PKCS#8 parse: {e}")))?; let public_key_der = expanded @@ -193,7 +218,10 @@ fn build_mirror_signer(pem: &str) -> Result { .to_public_key_der() .map_err(|e| Error::from(format!("ML-DSA-44 SPKI encode: {e}")))? .to_vec(); - Ok(MirrorSigner { public_key_der }) + Ok(MirrorSigner { + signer: Box::new(SubtreeV1CheckpointSigner::new(name, expanded)), + public_key_der, + }) } oid => Err(Error::from(format!( "unsupported MIRROR_SIGNING_KEY algorithm OID {oid}: expected id-ml-dsa-44 \ @@ -213,6 +241,48 @@ pub(crate) fn load_mirror_public_key_der(env: &Env) -> Result<&'static [u8]> { Ok(load_mirror_signer(env)?.public_key_der()) } +// --------------------------------------------------------------------------- +// Ticket key +// --------------------------------------------------------------------------- + +/// Cached ticket authenticator, built lazily on first request. +/// +/// The mirror's ticket scheme (base64 blobs returned in the +/// `text/x.tlog.mirror-info` 409 response body and round-tripped via the +/// `add-entries` request) is sealed with AES-256-GCM-SIV, a fresh random +/// nonce per ticket, with the log origin bound as associated data. See +/// [`tlog_mirror::TicketSealer`]; this static holds a single instance +/// keyed off the `MIRROR_TICKET_KEY` secret. +static TICKET_SEALER: OnceLock = OnceLock::new(); + +/// Load (or return the already-cached) ticket authenticator. +/// +/// The `MIRROR_TICKET_KEY` secret is 32 raw bytes encoded as standard +/// base64 (RFC 4648 section 4, no URL-safe variant). Operators can +/// generate one with `head -c 32 /dev/urandom | base64` and load it via +/// `wrangler secret put MIRROR_TICKET_KEY`. +/// +/// # Errors +/// +/// Returns an error if the `MIRROR_TICKET_KEY` secret is missing, is not +/// valid base64, or does not decode to exactly 32 bytes. +pub(crate) fn load_ticket_sealer(env: &Env) -> Result<&'static TicketSealer> { + if let Some(t) = TICKET_SEALER.get() { + return Ok(t); + } + let b64 = env.secret("MIRROR_TICKET_KEY")?.to_string(); + let raw = base64::engine::general_purpose::STANDARD + .decode(b64.trim()) + .map_err(|e| Error::from(format!("MIRROR_TICKET_KEY base64 decode: {e}")))?; + let key: [u8; 32] = raw.try_into().map_err(|v: Vec| { + Error::from(format!( + "MIRROR_TICKET_KEY must decode to exactly 32 bytes; got {}", + v.len() + )) + })?; + Ok(TICKET_SEALER.get_or_init(|| TicketSealer::new(&key))) +} + /// Initialize sentry from the `SENTRY_DSN` environment variable. /// /// Does nothing when the variable is absent or empty, allowing @@ -393,4 +463,20 @@ mod dev_config_tests { "dev MIRROR_SIGNING_KEY must load as a subtree/v1 signer", ); } + + /// `MIRROR_TICKET_KEY` in `.dev.vars` is base64 of exactly 32 bytes, + /// the precondition for [`crate::load_ticket_sealer`]. + #[test] + fn dev_vars_ticket_key_is_32_bytes_base64() { + let b64 = dev_var("MIRROR_TICKET_KEY"); + let raw = BASE64_STANDARD + .decode(b64.trim()) + .expect("MIRROR_TICKET_KEY must be valid base64"); + assert_eq!( + raw.len(), + 32, + "MIRROR_TICKET_KEY must decode to exactly 32 bytes; got {}", + raw.len() + ); + } } diff --git a/crates/mirror_worker/src/storage.rs b/crates/mirror_worker/src/storage.rs new file mode 100644 index 00000000..4105eb80 --- /dev/null +++ b/crates/mirror_worker/src/storage.rs @@ -0,0 +1,108 @@ +// Copyright (c) 2025-2026 Cloudflare, Inc. All rights reserved. +// SPDX-License-Identifier: BSD-3-Clause + +//! Object storage for the mirrored copy of each origin log. +//! +//! A single R2 bucket (bound as [`PUBLIC_BUCKET_BINDING`]) backs every +//! origin the mirror serves. Objects for a given origin are stored under +//! a distinct `/` key prefix, where `origin hash` is the +//! lowercase hex SHA-256 of the log's origin, the same identifier the +//! [c2sp.org/tlog-mirror][spec] monitoring interface uses in its URL +//! layout (`//...`; see the spec +//! [Introduction][intro]). +//! Keeping the storage prefix identical to the served path means the read +//! interface can proxy R2 keys to URLs without +//! translation. +//! +//! [`OriginBucket`] is a thin [`ObjectBackend`] adapter that prepends the +//! prefix to every key, so the commit path can use bare +//! [tlog-tiles][tiles] paths (`checkpoint`, `tile/...`, `tile/entries/...`) +//! exactly as [`generic_log_worker`] and [`tlog_tiles`] produce them. +//! +//! [spec]: https://c2sp.org/tlog-mirror +//! [intro]: https://c2sp.org/tlog-mirror#introduction +//! [tiles]: https://c2sp.org/tlog-tiles + +use generic_log_worker::{ObjectBackend, ObjectBucket, log_ops::UploadOptions}; +use sha2::{Digest as _, Sha256}; +#[allow(clippy::wildcard_imports)] +use worker::*; + +/// `wrangler.jsonc` binding name for the mirror's public R2 bucket. +pub(crate) const PUBLIC_BUCKET_BINDING: &str = "PUBLIC_BUCKET"; + +/// Compute a log's *origin hash*: the lowercase hex-encoded SHA-256 of +/// the origin string, per [c2sp.org/tlog-mirror][spec]. Used both as the +/// R2 key prefix and (by the read interface) as the monitoring URL path +/// component. +/// +/// [spec]: https://c2sp.org/tlog-mirror +#[must_use] +pub(crate) fn origin_hash(origin: &str) -> String { + hex::encode(Sha256::digest(origin.as_bytes())) +} + +/// An [`ObjectBackend`] that stores every object under a single origin's +/// `/` key prefix in the shared public bucket. +/// +/// Constructed per request via [`load_origin_bucket`]. Callers pass bare +/// tlog-tiles paths; the prefix is applied transparently. +pub(crate) struct OriginBucket { + inner: ObjectBucket, + prefix: String, +} + +impl OriginBucket { + /// Prepend the origin prefix to a bare tlog-tiles key. + fn prefixed(&self, key: &str) -> String { + format!("{}{key}", self.prefix) + } +} + +impl ObjectBackend for OriginBucket { + async fn upload, D: Into>>( + &self, + key: S, + data: D, + opts: &UploadOptions, + ) -> Result<()> { + self.inner + .upload(self.prefixed(key.as_ref()), data, opts) + .await + } + + async fn fetch>(&self, key: S) -> Result>> { + self.inner.fetch(self.prefixed(key.as_ref())).await + } +} + +/// Build an [`OriginBucket`] for `origin`, backed by the shared public R2 +/// bucket bound as [`PUBLIC_BUCKET_BINDING`]. +/// +/// # Errors +/// +/// Returns an error if the `PUBLIC_BUCKET` binding is missing or not an +/// R2 bucket. +pub(crate) fn load_origin_bucket(env: &Env, origin: &str) -> Result { + let bucket = env.bucket(PUBLIC_BUCKET_BINDING)?; + Ok(OriginBucket { + inner: ObjectBucket::new(bucket), + prefix: format!("{}/", origin_hash(origin)), + }) +} + +#[cfg(test)] +mod tests { + use super::origin_hash; + + /// Pin the origin-hash construction against a known SHA-256 vector so + /// the storage prefix (and, later, the monitoring URL path) can't + /// drift. `echo -n "example.com/log1" | sha256sum`. + #[test] + fn origin_hash_matches_known_vector() { + assert_eq!( + origin_hash("example.com/log1"), + "82df480cc8e80fed3584d9ac8520c582266fcefbb4257d4c758a0efa6bad9c95" + ); + } +} diff --git a/crates/mirror_worker/wrangler.jsonc b/crates/mirror_worker/wrangler.jsonc index fe5de69e..75f47866 100644 --- a/crates/mirror_worker/wrangler.jsonc +++ b/crates/mirror_worker/wrangler.jsonc @@ -29,6 +29,17 @@ } ] }, + // Public R2 bucket backing the mirrored copy of every origin + // log. The mirror stores each origin under a distinct + // `/` key prefix (see `crate::storage`), so a + // single bucket serves all configured logs. Served publicly + // via the c2sp.org/tlog-tiles read interface. + "r2_buckets": [ + { + "bucket_name": "tlog-mirror-public-dev", + "binding": "PUBLIC_BUCKET" + } + ], "migrations": [ { "tag": "v1", From 69693a8f1273cb5abb892ab33ee57ac7fbe7ed83 Mon Sep 17 00:00:00 2001 From: Luke Valenta Date: Thu, 30 Jul 2026 16:58:03 -0400 Subject: [PATCH 02/11] mirror_worker: reject add-entries with too many excess entries Enforce the spec's excess_entries bound in add-entries "Processing": reject with 409 when min(upload_end, next_entry) - upload_start exceeds one package (256). Entries below the persisted frontier are re-verified but not re-saved, so without a lower bound on upload_start a client could set upload_start=0 and force the mirror to re-verify the entire persisted prefix on every request. A legitimate resume sets upload_start to the persisted frontier (or, mid-tile, the frontier rounded down to a 256 boundary), keeping excess_entries at or below the threshold. --- crates/mirror_worker/src/add_entries.rs | 78 ++++++++++++++++++++++++- 1 file changed, 77 insertions(+), 1 deletion(-) diff --git a/crates/mirror_worker/src/add_entries.rs b/crates/mirror_worker/src/add_entries.rs index 62e3909b..6e5c7956 100644 --- a/crates/mirror_worker/src/add_entries.rs +++ b/crates/mirror_worker/src/add_entries.rs @@ -140,6 +140,29 @@ pub(crate) async fn add_entries( return Ok(mirror_info_409(&env, &snapshot, &header.log_origin)); } + // Spec (add-entries "Processing"): reject when excess_entries is too + // large. These are entries in [upload_start, next_entry) that are + // already persisted, so they are re-verified (subtree consistency) but + // not re-saved. Without a bound a client could set upload_start=0 and + // force the mirror to re-verify the entire persisted prefix on every + // request (a cheap DoS). A legitimate resume sets upload_start to the + // persisted frontier, or, when the frontier is mid-tile, to the + // frontier rounded down to a 256 boundary; excess_entries is then at + // most one package (256), which is our threshold. + let excess = excess_entries( + header.upload_start, + header.upload_end, + snapshot.next_entry.size, + ); + if excess > PACKAGE_ALIGNMENT { + log::info!( + "add-entries: rejecting upload_start={us} with excess_entries {excess} > {PACKAGE_ALIGNMENT} (next_entry={ne})", + us = header.upload_start, + ne = snapshot.next_entry.size, + ); + return Ok(mirror_info_409(&env, &snapshot, &header.log_origin)); + } + let first_prefix = first_package_prefix(&env, &header, snapshot.next_entry.size).await?; verify_and_persist( @@ -313,6 +336,16 @@ async fn verify_and_persist( cosign_and_serve(env, header, target, &bucket).await } +/// The spec's `excess_entries = min(upload_end, next_entry) - +/// upload_start`: the count of already-persisted entries a request +/// re-verifies without re-saving (see add-entries "Processing"). +/// +/// Saturates at 0 so an `upload_start` above the frontier (rejected +/// separately) can't underflow. +fn excess_entries(upload_start: u64, upload_end: u64, next_entry: u64) -> u64 { + next_entry.min(upload_end).saturating_sub(upload_start) +} + /// Outcome of reading the next entry package from the buffered body. enum PackageOutcome { Ok(EntryPackage), @@ -787,11 +820,12 @@ impl HashReader for MapReader<'_> { #[cfg(test)] mod tests { - use super::{MapReader, verify_package}; + use super::{MapReader, excess_entries, verify_package}; use crate::mirror_state_do::PendingCheckpoint; use std::collections::HashMap; use tlog_core::{Hash, Subtree, stored_hash_index, stored_hashes, tree_hash}; use tlog_mirror::EntryPackage; + use tlog_mirror::PACKAGE_ALIGNMENT; /// Deterministic distinct entry bytes for leaf `i`. fn entry(i: u64) -> Vec { @@ -895,4 +929,46 @@ mod tests { pkg.entries[0] = b"not the real entry".to_vec(); assert!(verify_package(&prefix, &pkg, 256, 512, &cp).is_err()); } + + #[test] + fn excess_entries_resume_at_frontier_is_zero() { + // The common resume: upload_start == next_entry, no overlap. + assert_eq!(excess_entries(2816, 3000, 2816), 0); + } + + #[test] + fn excess_entries_mid_tile_resume_within_one_package() { + // Mid-tile frontier: client rounds upload_start down to the 256 + // boundary, so overlap is the sub-256 tail and is accepted. + let next_entry = 600; + let upload_start = 512; // 600 rounded down to a 256 boundary + assert_eq!( + excess_entries(upload_start, 1000, next_entry), + next_entry - upload_start + ); + assert!(excess_entries(upload_start, 1000, next_entry) <= PACKAGE_ALIGNMENT); + } + + #[test] + fn excess_entries_from_zero_reverifies_whole_prefix() { + // The DoS case: upload_start=0 against a large frontier forces + // re-verification of the entire persisted prefix, well over the + // one-package threshold. + let excess = excess_entries(0, 10_000, 5_000); + assert_eq!(excess, 5_000); + assert!(excess > PACKAGE_ALIGNMENT); + } + + #[test] + fn excess_entries_bounded_by_upload_end() { + // Only entries below upload_end count as already-persisted overlap. + assert_eq!(excess_entries(100, 300, 5_000), 200); + } + + #[test] + fn excess_entries_saturates_above_frontier() { + // upload_start past the frontier (rejected separately) must not + // underflow. + assert_eq!(excess_entries(5_000, 6_000, 4_000), 0); + } } From e2991b3bb5c65f2db385a87b2ba40b590313273e Mon Sep 17 00:00:00 2001 From: Luke Valenta Date: Thu, 30 Jul 2026 17:07:06 -0400 Subject: [PATCH 03/11] mirror_worker: authenticate the reloaded partial entry bundle An incremental commit that resumes from a mid-tile frontier reloads the current partial entry bundle from object storage and extends it. Those bytes were re-uploaded and re-served without ever being checked: new-leaf hashes are derived only from the freshly-uploaded entries and the authenticated edge tiles, and the recomputed root never reads the reloaded bundle, so a corrupted or tampered partial bundle went undetected. Verify the reloaded bundle decodes to exactly its [subtree_start, persisted_size) leaves with no trailing bytes, and that each decoded entry's record_hash matches the authenticated leaf hash from the edge tiles, before extending it. --- crates/mirror_worker/src/commit.rs | 128 ++++++++++++++++++++++++++++- 1 file changed, 126 insertions(+), 2 deletions(-) diff --git a/crates/mirror_worker/src/commit.rs b/crates/mirror_worker/src/commit.rs index 08ed72bc..55ceb9b1 100644 --- a/crates/mirror_worker/src/commit.rs +++ b/crates/mirror_worker/src/commit.rs @@ -34,7 +34,9 @@ use generic_log_worker::{ log_ops::{HashReaderWithOverlay, TileWithBytes, UploadOptions, read_edge_tiles}, }; use length_prefixed::{ReadLengthPrefixedBytesExt as _, WriteLengthPrefixedBytesExt as _}; -use tlog_core::{Hash, record_hash, stored_hash_index, stored_hashes_for_record_hash}; +use tlog_core::{ + Hash, HashReader as _, record_hash, stored_hash_index, stored_hashes_for_record_hash, +}; use tlog_tiles::{PathElem, TlogTile}; #[allow(clippy::wildcard_imports)] use worker::*; @@ -105,6 +107,58 @@ pub(crate) async fn read_persisted_leaves( Ok(out) } +/// Authenticate a reloaded partial entry bundle before it is extended and +/// re-served. +/// +/// The bytes come back from object storage untrusted. New-leaf hashes are +/// computed only from the freshly-uploaded entries and the authenticated +/// edge tiles, and the recomputed root never reads the reloaded bundle +/// bytes, so a corrupted or tampered partial bundle would otherwise be +/// silently re-uploaded and served. +/// +/// Confirms the bundle decodes to exactly the `[subtree_start, +/// persisted_size)` leaves as length-prefixed entries with no trailing +/// bytes, and that each decoded entry's `record_hash` matches the +/// authenticated leaf hash from `edge_tiles`. +/// +/// # Errors +/// +/// Returns an error if the bundle is truncated, carries trailing bytes, a +/// leaf hash is unavailable, or a decoded entry does not match its +/// authenticated leaf hash. +fn verify_partial_bundle( + bytes: &[u8], + subtree_start: u64, + persisted_size: u64, + edge_tiles: &HashMap, +) -> Result<()> { + let overlay = HashMap::new(); + let reader = HashReaderWithOverlay { + edge_tiles, + overlay: &overlay, + }; + let mut cur: &[u8] = bytes; + for leaf in subtree_start..persisted_size { + let entry = cur + .read_length_prefixed(2) + .map_err(|e| Error::from(format!("partial entry bundle leaf {leaf} truncated: {e}")))?; + let want = reader + .read_hashes(&[stored_hash_index(0, leaf)]) + .map_err(|e| Error::from(format!("missing persisted leaf hash {leaf}: {e:?}")))?; + if record_hash(&entry) != want[0] { + return Err(Error::from(format!( + "partial entry bundle leaf {leaf} does not match its authenticated hash" + ))); + } + } + if !cur.is_empty() { + return Err(Error::from( + "partial entry bundle has trailing bytes past the persisted frontier", + )); + } + Ok(()) +} + /// Serialize one entry into its tlog-tiles entry-bundle framing (a /// big-endian uint16 length prefix followed by the entry bytes) and /// append it to `buf`. Matches `tlog_entry`'s `to_data_tile_entry` so @@ -179,14 +233,18 @@ pub(crate) async fn persist_entries( }; // Load the current partial entry bundle so we extend rather than - // overwrite it. Only exists when the frontier is mid-tile. + // overwrite it. Only exists when the frontier is mid-tile. The reloaded + // bytes are untrusted (storage could return corrupted or tampered + // data), so authenticate them before extending and re-serving them. let mut data_tile = Vec::new(); if persisted_size > 0 && !persisted_size.is_multiple_of(TILE_WIDTH) { + let subtree_start = (persisted_size / TILE_WIDTH) * TILE_WIDTH; let partial = TlogTile::from_index(stored_hash_index(0, persisted_size - 1)) .with_data_path(PathElem::Entries); data_tile = object.fetch(partial.path()).await?.ok_or_else(|| { Error::from(format!("partial entry bundle missing: {}", partial.path())) })?; + verify_partial_bundle(&data_tile, subtree_start, persisted_size, &edge_tiles)?; } // Replay leaves, flushing entry bundles at 256-entry boundaries. @@ -371,6 +429,72 @@ mod tests { assert_eq!(root1, reference_root(800)); } + #[tokio::test] + async fn incremental_commit_rejects_tampered_partial_bundle() { + let obj = MemBackend::default(); + // First commit to a mid-tile size (300), leaving a partial bundle + // [256, 300) that the next commit reloads and extends. + let root0 = persist_entries(&obj, 0, EMPTY_HASH, 300, &leaves(0..300)) + .await + .unwrap(); + + // Corrupt a byte inside an entry of the persisted partial bundle + // while keeping its length framing valid, so it still decodes but no + // longer matches its authenticated leaf hash. + let key = TlogTile::from_index(stored_hash_index(0, 299)) + .with_data_path(PathElem::Entries) + .path(); + let mut bytes = obj + .fetch(&key) + .await + .unwrap() + .expect("partial bundle stored"); + let last = bytes.len() - 1; + bytes[last] ^= 0xff; + obj.upload(&key, bytes, &immutable_tile_opts()) + .await + .unwrap(); + + // The next incremental commit reloads the tampered bundle and must + // reject it rather than re-serve unverified bytes. + assert!( + persist_entries(&obj, 300, root0, 800, &leaves(300..800)) + .await + .is_err(), + "tampered partial bundle must be rejected" + ); + } + + #[tokio::test] + async fn incremental_commit_rejects_truncated_partial_bundle() { + let obj = MemBackend::default(); + let root0 = persist_entries(&obj, 0, EMPTY_HASH, 300, &leaves(0..300)) + .await + .unwrap(); + + // Drop the trailing bytes of the partial bundle: it no longer + // decodes to the full [256, 300) range. + let key = TlogTile::from_index(stored_hash_index(0, 299)) + .with_data_path(PathElem::Entries) + .path(); + let mut bytes = obj + .fetch(&key) + .await + .unwrap() + .expect("partial bundle stored"); + bytes.truncate(bytes.len() - 3); + obj.upload(&key, bytes, &immutable_tile_opts()) + .await + .unwrap(); + + assert!( + persist_entries(&obj, 300, root0, 800, &leaves(300..800)) + .await + .is_err(), + "truncated partial bundle must be rejected" + ); + } + #[tokio::test] async fn entry_bundles_roundtrip() { use length_prefixed::ReadLengthPrefixedBytesExt as _; From fc238ed31d441f11f405f1ee0407d2fcfb8d2135 Mon Sep 17 00:00:00 2001 From: Luke Valenta Date: Fri, 31 Jul 2026 09:57:09 -0400 Subject: [PATCH 04/11] mirror_worker: let the state DO own the served-checkpoint write cosign_and_serve wrote the served checkpoint object to R2 (an unconditional overwrite of a fixed key) before advancing the durable mirror checkpoint via the DO. Under concurrent add-entries requests a slower request could overwrite R2 with an older checkpoint after a newer request had already advanced it, rewinding the served checkpoint. Move the R2 write into the state DO's /commit, serialized with the durable advance under a commit mutex (as the sequencer serializes its init path). Because a Durable Object has a single live instance per origin, holding the lock across the durable put and the R2 write makes the two atomic with respect to concurrent commits, including across the external R2 write's await, so the served checkpoint can no longer be rewound. Durable storage is advanced before the R2 write so the served object is never ahead of committed; a failed R2 write leaves it lagging and is rewritten by the next commit. The DO resolves the origin bucket from its own name (state.id().name()). --- crates/mirror_worker/Cargo.toml | 1 + crates/mirror_worker/src/add_entries.rs | 28 ++--- crates/mirror_worker/src/mirror_state_do.rs | 122 ++++++++++++++------ 3 files changed, 101 insertions(+), 50 deletions(-) diff --git a/crates/mirror_worker/Cargo.toml b/crates/mirror_worker/Cargo.toml index 6443cc3d..74d8b39f 100644 --- a/crates/mirror_worker/Cargo.toml +++ b/crates/mirror_worker/Cargo.toml @@ -57,6 +57,7 @@ tlog_cosignature.workspace = true tlog_mirror.workspace = true tlog_tiles.workspace = true tlog_witness.workspace = true +tokio.workspace = true tower-service.workspace = true worker = { workspace = true, features = ["http", "axum"] } diff --git a/crates/mirror_worker/src/add_entries.rs b/crates/mirror_worker/src/add_entries.rs index 6e5c7956..30d8b208 100644 --- a/crates/mirror_worker/src/add_entries.rs +++ b/crates/mirror_worker/src/add_entries.rs @@ -43,7 +43,7 @@ use tlog_mirror::{ #[allow(clippy::wildcard_imports)] use worker::*; -use generic_log_worker::{ObjectBackend, util::now_millis}; +use generic_log_worker::util::now_millis; use crate::{ body, commit, @@ -333,7 +333,7 @@ async fn verify_and_persist( )); } - cosign_and_serve(env, header, target, &bucket).await + cosign_and_serve(env, header, target).await } /// The spec's `excess_entries = min(upload_end, next_entry) - @@ -371,22 +371,26 @@ fn read_package(cursor: &mut Cursor<&[u8]>, num_entries: u64) -> PackageOutcome } } -/// Cosign the target checkpoint with the mirror key, persist the served -/// checkpoint object, advance the durable mirror checkpoint via the DO, -/// and return the 200 response carrying the mirror cosignature line(s). +/// Cosign the target checkpoint with the mirror key, advance the durable +/// mirror checkpoint via the DO, and return the 200 response carrying the +/// mirror cosignature line(s). /// /// Called only once the whole upload is durably committed (spec: the /// mirror MUST NOT sign until all entries are committed). /// +/// The DO's `/commit` both advances the durable checkpoint and writes the +/// served checkpoint object to R2, serialized under its commit lock, so +/// concurrent commits cannot rewind the served checkpoint. The frontend +/// therefore does not write R2 itself. +/// /// # Errors /// /// Returns an error if the target note/checkpoint fails to parse, signing -/// fails, or a storage/DO write fails. -async fn cosign_and_serve( +/// fails, or the DO commit fails. +async fn cosign_and_serve( env: &Env, header: &AddEntriesRequestHeader, target: &PendingCheckpoint, - bucket: &O, ) -> ApiResult { // The checkpoint text comes from the log-signed pending note; the // response is the bare cosignature line(s), identical to a witness's @@ -402,13 +406,11 @@ async fn cosign_and_serve( let cosig_body = tlog_witness::serialize_add_checkpoint_response(std::slice::from_ref(¬e_sig)); - // Persist the served checkpoint (log note + mirror cosignature) before - // advancing the durable mirror checkpoint, so a monitor can never - // observe an advanced size without the matching checkpoint object. + // The served checkpoint is the log's signed note with the mirror's + // cosignature appended. The DO writes it to R2 while advancing the + // durable checkpoint under its commit lock. let mut checkpoint_obj = target.signed_note_bytes.clone(); checkpoint_obj.extend_from_slice(&cosig_body); - commit::write_checkpoint(bucket, checkpoint_obj.clone()).await?; - dispatch_commit( env, &header.log_origin, diff --git a/crates/mirror_worker/src/mirror_state_do.rs b/crates/mirror_worker/src/mirror_state_do.rs index eace124d..3b45a3f0 100644 --- a/crates/mirror_worker/src/mirror_state_do.rs +++ b/crates/mirror_worker/src/mirror_state_do.rs @@ -28,10 +28,11 @@ use serde::{Deserialize, Serialize}; use serde_with::{base64::Base64 as Base64As, serde_as}; use tlog_core::{Hash, verify_consistency_proof}; +use tokio::sync::Mutex; #[allow(clippy::wildcard_imports)] use worker::*; -use crate::MIRROR_STATE_BINDING; +use crate::{MIRROR_STATE_BINDING, commit, storage::load_origin_bucket}; const PENDING_KEY: &str = "pending"; const COMMITTED_KEY: &str = "committed"; @@ -169,6 +170,16 @@ pub struct UpdatePendingRequest { #[durable_object(fetch)] struct MirrorState { state: State, + /// Held to resolve the origin's R2 bucket when the DO writes the + /// served checkpoint on `/commit`. + env: Env, + /// Serializes `/commit` so the durable-storage advance and the R2 + /// served-checkpoint write happen atomically with respect to + /// concurrent commits, even across the external R2 write's await. + /// Without it a slower commit could overwrite the served checkpoint + /// with an older one after a concurrent commit already advanced it. + /// Mirrors the sequencer's `init_mux` (see `generic_log_worker`). + commit_mux: Mutex<()>, } // SAFETY: Durable Objects are single-threaded; the `RefUnwindSafe` bound @@ -179,7 +190,11 @@ impl std::panic::RefUnwindSafe for MirrorState {} impl DurableObject for MirrorState { fn new(state: State, env: Env) -> Self { crate::init_sentry(&env); - Self { state } + Self { + state, + env, + commit_mux: Mutex::new(()), + } } async fn fetch(&self, req: Request) -> Result { @@ -200,42 +215,8 @@ impl MirrorState { Response::from_json(&snapshot) } (Method::Post, "/commit") => { - // Compare-and-swap advance of the mirror checkpoint: - // * size > next_entry.size: commit beyond the - // persisted-entry frontier, i.e. cosigning entries - // that have not been durably written yet; 400. This - // preserves `committed.size <= next_entry.size`. Since - // `next_entry.size <= pending.size`, it also rejects - // commits beyond the accepted pending checkpoint. - // * size < committed.size: a concurrent add-entries - // already advanced past us. Spec forbids rolling - // back, so no-op and return the current state. - // * otherwise: advance committed. let body: CommitRequest = req.json().await?; - let snapshot = self.read_snapshot().await?; - if body.size > snapshot.next_entry.size { - return Response::error( - format!( - "commit beyond persisted-entry frontier: requested size {} > next_entry size {}", - body.size, snapshot.next_entry.size - ), - 400, - ); - } - if body.size < snapshot.committed.size { - // Already ahead; no-op success. - return Response::from_json(&snapshot.committed); - } - let new_committed = CommittedCheckpoint { - size: body.size, - hash: body.hash, - signed_note_bytes: body.signed_note_bytes, - }; - self.state - .storage() - .put(COMMITTED_KEY, &new_committed) - .await?; - Response::from_json(&new_committed) + self.commit(body).await } (Method::Post, "/advance-next-entry") => { let body: AdvanceNextEntryRequest = req.json().await?; @@ -308,6 +289,73 @@ impl MirrorState { } impl MirrorState { + /// Handle `/commit`: monotonically advance the mirror checkpoint and + /// write the served checkpoint object to R2. + /// + /// `commit_mux` serializes the whole read-check-advance-write + /// sequence, including the external R2 write, so concurrent commits + /// cannot interleave and rewind the served checkpoint. Compare-and-swap + /// semantics: + /// * `size > next_entry.size`: commit beyond the persisted-entry + /// frontier (cosigning entries not yet durably written); 400. This + /// preserves `committed.size <= next_entry.size`, and since + /// `next_entry.size <= pending.size` it also rejects commits beyond + /// the accepted pending checkpoint. + /// * `size < committed.size`: a concurrent add-entries already + /// advanced past us. The spec forbids rolling back, so no-op and + /// return the current committed checkpoint (whose served object the + /// concurrent commit already wrote). + /// * otherwise: advance committed in durable storage, then write the + /// served checkpoint to R2. + /// + /// Durable storage is advanced before the R2 write so the served + /// checkpoint is never ahead of committed; a failed R2 write leaves R2 + /// lagging and is rewritten by the next commit. + async fn commit(&self, body: CommitRequest) -> Result { + let _guard = self.commit_mux.lock().await; + + let snapshot = self.read_snapshot().await?; + if body.size > snapshot.next_entry.size { + return Response::error( + format!( + "commit beyond persisted-entry frontier: requested size {} > next_entry size {}", + body.size, snapshot.next_entry.size + ), + 400, + ); + } + if body.size < snapshot.committed.size { + // Already ahead; no-op success. The concurrent commit that + // advanced past us already wrote the newer served checkpoint. + return Response::from_json(&snapshot.committed); + } + + let new_committed = CommittedCheckpoint { + size: body.size, + hash: body.hash, + signed_note_bytes: body.signed_note_bytes, + }; + self.state + .storage() + .put(COMMITTED_KEY, &new_committed) + .await?; + + // Write the served checkpoint (the log's signed note plus the + // mirror's cosignature) to R2 at + // //checkpoint. The DO owns this write so + // it stays serialized with the durable advance above; the origin is + // the DO's own name. + let origin = self + .state + .id() + .name() + .ok_or_else(|| Error::from("mirror state DO missing origin name"))?; + let bucket = load_origin_bucket(&self.env, &origin)?; + commit::write_checkpoint(&bucket, new_committed.signed_note_bytes.clone()).await?; + + Response::from_json(&new_committed) + } + /// Handle `/advance-next-entry`: monotonically advance the /// persisted-entry frontier. Compare-and-swap, like `/commit`: /// From 90e540322d9232f0e204c0202be91aabdb29bd3e Mon Sep 17 00:00:00 2001 From: Luke Valenta Date: Fri, 31 Jul 2026 11:44:30 -0400 Subject: [PATCH 05/11] mirror_worker: correct the committed-checkpoint doc comment CommittedCheckpoint.signed_note_bytes stores the served checkpoint (the log's signed note with the mirror's cosignature appended), not the log's original note, so describe it that way. --- crates/mirror_worker/src/mirror_state_do.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/crates/mirror_worker/src/mirror_state_do.rs b/crates/mirror_worker/src/mirror_state_do.rs index 3b45a3f0..1d887410 100644 --- a/crates/mirror_worker/src/mirror_state_do.rs +++ b/crates/mirror_worker/src/mirror_state_do.rs @@ -75,8 +75,9 @@ pub struct CommittedCheckpoint { /// Root hash. All-zero if `size` is 0. #[serde(with = "hash_hex")] pub hash: Hash, - /// Full signed-note bytes as the log signed them, empty if `size` is - /// 0. Served with the mirror's cosignature lines appended. + /// The served checkpoint bytes: the log's signed note with the + /// mirror's cosignature line(s) appended, exactly as written to R2. + /// Empty if `size` is 0. #[serde_as(as = "Base64As")] pub signed_note_bytes: Vec, } From 7c1cfb082fe30ad7dedf1d693e132fce4604eb65 Mon Sep 17 00:00:00 2001 From: Luke Valenta Date: Mon, 3 Aug 2026 08:29:31 -0400 Subject: [PATCH 06/11] mirror_worker: authenticate reloaded persisted-leaf prefix read_persisted_leaves returned reloaded entry-bundle bytes without checking them, so a corrupted or stale bundle flowed into package verification and surfaced as a client 422 for a mirror storage fault. Authenticate the decoded leaves against the frontier root via the tlog-tiles tile-hash reader (edge tiles alone cannot cover a resume that lands below the frontier's edge tile, which the excess_entries bound permits). Addresses bonk #264 review on commit.rs read_persisted_leaves. --- crates/mirror_worker/src/add_entries.rs | 16 ++- crates/mirror_worker/src/commit.rs | 138 +++++++++++++++++++++--- 2 files changed, 136 insertions(+), 18 deletions(-) diff --git a/crates/mirror_worker/src/add_entries.rs b/crates/mirror_worker/src/add_entries.rs index 30d8b208..33439514 100644 --- a/crates/mirror_worker/src/add_entries.rs +++ b/crates/mirror_worker/src/add_entries.rs @@ -163,7 +163,13 @@ pub(crate) async fn add_entries( return Ok(mirror_info_409(&env, &snapshot, &header.log_origin)); } - let first_prefix = first_package_prefix(&env, &header, snapshot.next_entry.size).await?; + let first_prefix = first_package_prefix( + &env, + &header, + snapshot.next_entry.size, + snapshot.next_entry.hash, + ) + .await?; verify_and_persist( &env, @@ -439,14 +445,19 @@ async fn cosign_and_serve( /// next_entry.size` is enforced upstream, the requested leaves are always /// present in storage. /// +/// The prefix is authenticated against the frontier hash tiles inside +/// [`commit::read_persisted_leaves`]; `persisted_hash` is the frontier +/// root at `persisted_size`. +/// /// # Errors /// /// Returns an error if opening the origin bucket or reading the persisted -/// entry bundle fails. +/// entry bundle fails, or the reloaded leaves fail authentication. async fn first_package_prefix( env: &Env, header: &AddEntriesRequestHeader, persisted_size: u64, + persisted_hash: Hash, ) -> Result>> { let subtree_start = (header.upload_start / PACKAGE_ALIGNMENT) * PACKAGE_ALIGNMENT; if header.upload_start == subtree_start { @@ -458,6 +469,7 @@ async fn first_package_prefix( subtree_start, header.upload_start - subtree_start, persisted_size, + persisted_hash, ) .await } diff --git a/crates/mirror_worker/src/commit.rs b/crates/mirror_worker/src/commit.rs index 55ceb9b1..b8f2d4d1 100644 --- a/crates/mirror_worker/src/commit.rs +++ b/crates/mirror_worker/src/commit.rs @@ -35,9 +35,9 @@ use generic_log_worker::{ }; use length_prefixed::{ReadLengthPrefixedBytesExt as _, WriteLengthPrefixedBytesExt as _}; use tlog_core::{ - Hash, HashReader as _, record_hash, stored_hash_index, stored_hashes_for_record_hash, + Hash, HashReader as _, TlogError, record_hash, stored_hash_index, stored_hashes_for_record_hash, }; -use tlog_tiles::{PathElem, TlogTile}; +use tlog_tiles::{PathElem, PreloadedTlogTileReader, TileHashReader, TlogTile, TlogTileRecorder}; #[allow(clippy::wildcard_imports)] use worker::*; @@ -63,16 +63,25 @@ pub(crate) const CHECKPOINT_KEY: &str = "checkpoint"; /// upload_start)` are already in the log and therefore absent from the /// uploaded package. /// +/// The reloaded bytes are untrusted storage output, so each decoded entry +/// is authenticated against the hash tiles committed at the persisted +/// frontier (`persisted_hash` is the frontier root, used to authenticate +/// the edge tiles via [`read_edge_tiles`]). Without this a corrupted or +/// stale bundle would flow into package verification and surface as a +/// client "422 Unprocessable Entity" for what is actually a mirror +/// storage fault. +/// /// # Errors /// -/// Returns an error if the bundle is missing from storage or is shorter -/// than `count` entries (i.e. the requested leaves were not actually -/// persisted). +/// Returns an error if the bundle is missing from storage, is shorter than +/// `count` entries, or a decoded entry does not match its authenticated +/// leaf hash. pub(crate) async fn read_persisted_leaves( object: &impl ObjectBackend, start: u64, count: u64, persisted_size: u64, + persisted_hash: Hash, ) -> Result>> { debug_assert!( start.is_multiple_of(TILE_WIDTH), @@ -93,20 +102,81 @@ pub(crate) async fn read_persisted_leaves( .await? .ok_or_else(|| Error::from(format!("persisted entry bundle missing: {}", tile.path())))?; + // Authenticate the decoded leaves against the frontier hash tiles. + // Leaves below the frontier's edge tile are not covered by + // `read_edge_tiles`, and the `excess_entries` bound lets a resume land + // in the tile just before the edge, so authenticate against the tree + // root via the full tile-hash reader instead. + let indexes: Vec = (start..start + count) + .map(|leaf| stored_hash_index(0, leaf)) + .collect(); + let want = authenticated_leaf_hashes(object, persisted_size, persisted_hash, &indexes).await?; + let mut cur: &[u8] = &bytes; let mut out = Vec::with_capacity(usize::try_from(count).unwrap_or(0)); for i in 0..count { - let entry = cur.read_length_prefixed(2).map_err(|e| { - Error::from(format!( - "persisted bundle leaf {} truncated: {e}", - start + i - )) - })?; + let leaf = start + i; + let entry = cur + .read_length_prefixed(2) + .map_err(|e| Error::from(format!("persisted bundle leaf {leaf} truncated: {e}")))?; + let idx = usize::try_from(i).unwrap_or(usize::MAX); + if record_hash(&entry) != want[idx] { + return Err(Error::from(format!( + "persisted bundle leaf {leaf} does not match its authenticated hash" + ))); + } out.push(entry); } Ok(out) } +/// Fetch and authenticate the record hashes for `indexes` (level-0 leaf +/// stored-hash indexes) against the frontier `(tree_size, tree_hash)`. +/// +/// Runs the standard tlog-tiles two-pass [`TileHashReader`] protocol: a +/// recording pass discovers the hash tiles needed to prove the requested +/// leaves, they are fetched from storage, then a verifying pass +/// authenticates them against `tree_hash`. A storage fault (missing, +/// stale, or tampered tile) therefore surfaces as an error here rather +/// than as a spurious client rejection downstream. +/// +/// # Errors +/// +/// Returns an error if a required hash tile is missing from storage or the +/// fetched tiles do not authenticate against `tree_hash`. +async fn authenticated_leaf_hashes( + object: &impl ObjectBackend, + tree_size: u64, + tree_hash: Hash, + indexes: &[u64], +) -> Result> { + // Recording pass: `TlogTileRecorder` short-circuits `read_hashes` with + // `RecordedTilesOnly` after collecting the tiles it would need. + let recorder = TlogTileRecorder::default(); + match TileHashReader::new(tree_size, Hash::default(), &recorder).read_hashes(indexes) { + Err(TlogError::RecordedTilesOnly) => {} + other => { + return Err(Error::from(format!( + "expected RecordedTilesOnly while recording hash tiles, got {other:?}" + ))); + } + } + + let mut tiles: HashMap> = HashMap::new(); + for tile in recorder.0.into_inner() { + let bytes = object + .fetch(tile.path()) + .await? + .ok_or_else(|| Error::from(format!("persisted hash tile missing: {}", tile.path())))?; + tiles.insert(tile, bytes); + } + + let reader = PreloadedTlogTileReader(tiles); + TileHashReader::new(tree_size, tree_hash, &reader) + .read_hashes(indexes) + .map_err(|e| Error::from(format!("authenticate persisted leaf hashes: {e:?}"))) +} + /// Authenticate a reloaded partial entry bundle before it is extended and /// re-served. /// @@ -536,29 +606,65 @@ mod tests { async fn read_persisted_leaves_from_full_and_partial_bundles() { let obj = MemBackend::default(); // 300 leaves: one full bundle [0,256) + a partial bundle [256,300). - persist_entries(&obj, 0, EMPTY_HASH, 300, &leaves(0..300)) + let root = persist_entries(&obj, 0, EMPTY_HASH, 300, &leaves(0..300)) .await .unwrap(); // Prefix within the full bundle (persisted_size 300 -> stored // width 256 for tile 0). - let got = read_persisted_leaves(&obj, 0, 44, 300).await.unwrap(); + let got = read_persisted_leaves(&obj, 0, 44, 300, root).await.unwrap(); assert_eq!(got, leaves(0..44)); // Whole full bundle. - let got = read_persisted_leaves(&obj, 0, 256, 300).await.unwrap(); + let got = read_persisted_leaves(&obj, 0, 256, 300, root) + .await + .unwrap(); assert_eq!(got, leaves(0..256)); // Prefix within the trailing partial bundle (tile 1, stored width // 300 - 256 = 44). - let got = read_persisted_leaves(&obj, 256, 20, 300).await.unwrap(); + let got = read_persisted_leaves(&obj, 256, 20, 300, root) + .await + .unwrap(); assert_eq!(got, leaves(256..276)); } #[tokio::test] async fn read_persisted_leaves_missing_bundle_errors() { let obj = MemBackend::default(); - assert!(read_persisted_leaves(&obj, 0, 10, 300).await.is_err()); + assert!( + read_persisted_leaves(&obj, 0, 10, 300, EMPTY_HASH) + .await + .is_err() + ); + } + + #[tokio::test] + async fn read_persisted_leaves_rejects_tampered_bundle() { + let obj = MemBackend::default(); + let root = persist_entries(&obj, 0, EMPTY_HASH, 300, &leaves(0..300)) + .await + .unwrap(); + + // Corrupt a byte inside the full bundle [0,256) while keeping the + // length framing valid, so it still decodes but no longer matches + // its authenticated leaf hash. + let key = TlogTile::from_index(stored_hash_index(0, 255)) + .with_data_path(PathElem::Entries) + .path(); + let mut bytes = obj.fetch(&key).await.unwrap().expect("full bundle stored"); + let last = bytes.len() - 1; + bytes[last] ^= 0xff; + obj.upload(&key, bytes, &immutable_tile_opts()) + .await + .unwrap(); + + assert!( + read_persisted_leaves(&obj, 0, 256, 300, root) + .await + .is_err(), + "tampered persisted bundle must be rejected" + ); } #[tokio::test] From 09864007c1a9fff92cba161bc1abe7880f43a163 Mon Sep 17 00:00:00 2001 From: Luke Valenta Date: Mon, 3 Aug 2026 08:30:28 -0400 Subject: [PATCH 07/11] mirror_worker: discard trailing add-entries bytes instead of 400 The spec says the mirror discards any partial bytes after the last successfully authenticated entry package; the only defined 400 is when no package was authenticated at all. The trailing-byte 400 also ran after persist_entries and advance_next_entry had already committed and advanced the frontier, so it told the client nothing was saved when everything was, with no next_entry to resume from. Drop the check. Addresses bonk #264 review on add_entries.rs trailing-data handling. --- crates/mirror_worker/src/add_entries.rs | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/crates/mirror_worker/src/add_entries.rs b/crates/mirror_worker/src/add_entries.rs index 33439514..e8f59803 100644 --- a/crates/mirror_worker/src/add_entries.rs +++ b/crates/mirror_worker/src/add_entries.rs @@ -313,14 +313,12 @@ async fn verify_and_persist( )); } - // Complete upload: no trailing bytes may remain after the last package. - let pos = usize::try_from(cursor.position()).unwrap_or(usize::MAX); - if pos < cursor.get_ref().len() { - log::warn!("add-entries: trailing data after the last entry package"); - return Err(AppError::BadRequest( - "trailing data after the last entry package".to_owned(), - )); - } + // Every canonical package was received. Any bytes past the last one + // are discarded, not rejected: the spec says "the mirror discards any + // partial bytes after the last successfully authenticated entry + // package", and the only defined 400 is when no package authenticated + // at all (handled above). Rejecting here would also be dishonest, since + // the entries were already persisted and the frontier advanced. // When we persisted new entries, the recomputed tree MUST match the // pending checkpoint the packages were proven against. A mismatch means From 38928f1ae59f928542bdd9b8e278d06248c57a17 Mon Sep 17 00:00:00 2001 From: Luke Valenta Date: Mon, 3 Aug 2026 08:42:47 -0400 Subject: [PATCH 08/11] mirror_worker: return 409 when the mirror commit is skipped dispatch_commit discarded the DO response and cosign_and_serve always returned 200 with a cosignature, even when the DO refused to rewind because a concurrent add-entries had already advanced the mirror checkpoint past upload_end. The client read that as "my checkpoint is served" while the mirror was actually at a larger size. Return the DO's CommittedCheckpoint and, when its size is ahead of upload_end, respond 409 with mirror-info per the spec's "upload_end too small" case so the client resyncs. Addresses bonk #264 review on cosign_and_serve. --- crates/mirror_worker/src/add_entries.rs | 46 ++++++++++++++++++++----- 1 file changed, 37 insertions(+), 9 deletions(-) diff --git a/crates/mirror_worker/src/add_entries.rs b/crates/mirror_worker/src/add_entries.rs index e8f59803..a15ff6b0 100644 --- a/crates/mirror_worker/src/add_entries.rs +++ b/crates/mirror_worker/src/add_entries.rs @@ -50,8 +50,8 @@ use crate::{ frontend_worker::{ApiResult, AppError}, load_mirror_signer, load_ticket_sealer, log_verifiers, mirror_state_do::{ - AdvanceNextEntryRequest, CommitRequest, MirrorStateSnapshot, NextEntry, PendingCheckpoint, - state_stub, + AdvanceNextEntryRequest, CommitRequest, CommittedCheckpoint, MirrorStateSnapshot, + NextEntry, PendingCheckpoint, state_stub, }, storage::load_origin_bucket, }; @@ -337,7 +337,7 @@ async fn verify_and_persist( )); } - cosign_and_serve(env, header, target).await + cosign_and_serve(env, header, target, snapshot).await } /// The spec's `excess_entries = min(upload_end, next_entry) - @@ -387,6 +387,13 @@ fn read_package(cursor: &mut Cursor<&[u8]>, num_entries: u64) -> PackageOutcome /// concurrent commits cannot rewind the served checkpoint. The frontend /// therefore does not write R2 itself. /// +/// If the DO reports a committed size ahead of `upload_end`, a concurrent +/// `add-entries` already advanced the mirror checkpoint past this upload. +/// Per the spec's final step ("If `upload_end` was too small, the mirror +/// MUST respond with a 409 Conflict") this returns a 409 with mirror-info +/// so the client resyncs, rather than a 200 that would falsely claim the +/// client's smaller checkpoint is being served. +/// /// # Errors /// /// Returns an error if the target note/checkpoint fails to parse, signing @@ -395,6 +402,7 @@ async fn cosign_and_serve( env: &Env, header: &AddEntriesRequestHeader, target: &PendingCheckpoint, + snapshot: &MirrorStateSnapshot, ) -> ApiResult { // The checkpoint text comes from the log-signed pending note; the // response is the bare cosignature line(s), identical to a witness's @@ -415,7 +423,7 @@ async fn cosign_and_serve( // durable checkpoint under its commit lock. let mut checkpoint_obj = target.signed_note_bytes.clone(); checkpoint_obj.extend_from_slice(&cosig_body); - dispatch_commit( + let committed = dispatch_commit( env, &header.log_origin, &CommitRequest { @@ -426,6 +434,18 @@ async fn cosign_and_serve( ) .await?; + if committed.size != header.upload_end { + // The DO refused to rewind: a concurrent commit already advanced + // the mirror checkpoint past upload_end, so ours was skipped. + log::info!( + "add-entries: commit skipped, mirror checkpoint {} already past upload_end {}; \ + returning 409", + committed.size, + header.upload_end, + ); + return Ok(mirror_info_409(env, snapshot, &header.log_origin)); + } + Ok(( StatusCode::OK, [(CONTENT_TYPE, "text/plain; charset=utf-8")], @@ -473,10 +493,18 @@ async fn first_package_prefix( } /// POST the [`CommitRequest`] to the per-origin DO, advancing the mirror -/// checkpoint. A non-200 status (or a `/commit` beyond pending) is a -/// frontend/mirror bug, so it is surfaced as a transport error that the -/// handler maps to 500. -async fn dispatch_commit(env: &Env, origin: &str, commit_req: &CommitRequest) -> Result<()> { +/// checkpoint, and return the DO's resulting [`CommittedCheckpoint`]. +/// +/// The returned checkpoint may be ahead of `commit_req` when a concurrent +/// `add-entries` already advanced past it (the DO refuses to rewind); the +/// caller compares sizes to detect that skip. A non-200 status (a +/// `/commit` beyond the persisted frontier) is a frontend/mirror bug, +/// surfaced as a transport error that the handler maps to 500. +async fn dispatch_commit( + env: &Env, + origin: &str, + commit_req: &CommitRequest, +) -> Result { let stub = state_stub(env, origin)?; let mut resp = stub .fetch_with_request(Request::new_with_init( @@ -494,7 +522,7 @@ async fn dispatch_commit(env: &Env, origin: &str, commit_req: &CommitRequest) -> )?) .await?; match resp.status_code() { - 200 => Ok(()), + 200 => Ok(resp.json().await?), status => { let msg = resp.text().await.unwrap_or_default(); log::error!("add-entries: DO /commit returned {status}: {msg}"); From b41d7e9d21aa5266aa02b95ecdb44fb680db4ba4 Mon Sep 17 00:00:00 2001 From: Luke Valenta Date: Mon, 3 Aug 2026 08:51:41 -0400 Subject: [PATCH 09/11] mirror_worker: accept the mirror checkpoint size as upload_end The spec requires the mirror to accept upload_end equal to the mirror checkpoint's tree size, not just a known pending value; resolve_target_ pending only matched the current pending or a ticket-carried past pending, so a client targeting the already-committed size with no valid ticket got a spurious 409. Accept upload_end == committed.size using the committed checkpoint as the target. cosign_and_serve now skips the /commit dispatch when the upload only reaches the committed size, which also avoids redundantly rewriting R2 and appending a duplicate cosignature line. Addresses bonk #264 review on resolve_target_pending. --- crates/mirror_worker/src/add_entries.rs | 36 +++++++++++++++++++++++-- 1 file changed, 34 insertions(+), 2 deletions(-) diff --git a/crates/mirror_worker/src/add_entries.rs b/crates/mirror_worker/src/add_entries.rs index a15ff6b0..d3f3b0c1 100644 --- a/crates/mirror_worker/src/add_entries.rs +++ b/crates/mirror_worker/src/add_entries.rs @@ -418,6 +418,22 @@ async fn cosign_and_serve( let cosig_body = tlog_witness::serialize_add_checkpoint_response(std::slice::from_ref(¬e_sig)); + // When the upload only reaches the mirror checkpoint's current size, + // the checkpoint at that size is already committed and served. There + // is nothing to advance, and re-committing would redundantly rewrite + // R2 and append a duplicate cosignature line to the served note, so + // return a fresh cosignature without dispatching `/commit`. (Committed + // is monotonic, so a stale-low snapshot only skips this optimization, + // never the other way.) + if header.upload_end <= snapshot.committed.size { + return Ok(( + StatusCode::OK, + [(CONTENT_TYPE, "text/plain; charset=utf-8")], + cosig_body, + ) + .into_response()); + } + // The served checkpoint is the log's signed note with the mirror's // cosignature appended. The DO writes it to R2 while advancing the // durable checkpoint under its commit lock. @@ -590,15 +606,18 @@ async fn fetch_snapshot(env: &Env, origin: &str) -> Result } /// Resolve the target pending checkpoint that this `add-entries` -/// request is uploading toward. Either: +/// request is uploading toward. Any of: /// /// * `upload_end == snapshot.pending.size`: use the current pending. +/// * `upload_end == snapshot.committed.size`: use the mirror +/// checkpoint, which the spec requires the mirror to accept as a +/// valid `upload_end` independent of any ticket. /// * The ticket round-trips and yields a past pending whose /// embedded checkpoint has size `upload_end` and verifies against /// the trusted log keys: use that. /// /// Returns `Err(reason)` (a static `&str` describing why) when -/// neither path produces a target. The frontend turns the error into +/// none of these produce a target. The frontend turns the error into /// a 409 with `text/x.tlog.mirror-info`. fn resolve_target_pending( env: &Env, @@ -617,6 +636,19 @@ fn resolve_target_pending( return Ok(snapshot.pending.clone()); } + // Spec: the mirror MUST also accept `upload_end` equal to the mirror + // checkpoint's tree size, whatever the ticket says. Everything up to + // it is already committed and served, so the committed checkpoint is + // the target and nothing new is persisted; `cosign_and_serve` returns + // a fresh cosignature without re-advancing. + if snapshot.committed.size > 0 && header.upload_end == snapshot.committed.size { + return Ok(PendingCheckpoint { + size: snapshot.committed.size, + hash: snapshot.committed.hash, + signed_note_bytes: snapshot.committed.signed_note_bytes.clone(), + }); + } + // Try the ticket. An empty ticket can't carry a past pending, so // there's nothing to fall back to. if header.ticket.is_empty() { From 38a1cee1afddd8bc1b8d9e6d9f4c6e068942e95a Mon Sep 17 00:00:00 2001 From: Luke Valenta Date: Mon, 3 Aug 2026 09:49:39 -0400 Subject: [PATCH 10/11] mirror_worker: tighten add-entries spec compliance Enforce the spec's Content-Type MUST, advertise gzip support in responses, and use a fresh DO snapshot when a concurrent commit forces a final 409 so the mirror-info body is not stale. --- crates/mirror_worker/src/add_entries.rs | 73 ++++++++++++++++++++- crates/mirror_worker/src/frontend_worker.rs | 35 +++++++++- 2 files changed, 104 insertions(+), 4 deletions(-) diff --git a/crates/mirror_worker/src/add_entries.rs b/crates/mirror_worker/src/add_entries.rs index d3f3b0c1..040cbd29 100644 --- a/crates/mirror_worker/src/add_entries.rs +++ b/crates/mirror_worker/src/add_entries.rs @@ -68,6 +68,16 @@ pub(crate) async fn add_entries( State(env): State, req: axum::extract::Request, ) -> ApiResult { + // Spec: the add-entries request body MUST have Content-Type + // application/octet-stream. Reject anything else up front (before + // spending time reading or decoding the body). + let (parts, body) = req.into_parts(); + if !content_type_is_octet_stream(&parts.headers) { + return Err(AppError::UnsupportedMediaType( + "add-entries requires Content-Type: application/octet-stream".to_owned(), + )); + } + // No DefaultBodyLimit: Cloudflare enforces a request-body cap at the // edge (100 MB, higher on paid plans) and 413s oversized bodies there. // Clients on body-limited platforms truncate at a package boundary and @@ -78,7 +88,6 @@ pub(crate) async fn add_entries( // // The whole (decoded) body is buffered before processing; a follow-up // commit streams it instead of buffering it all. - let (parts, body) = req.into_parts(); let raw = body::read_decoded_body(&parts.headers, body).await?; let mut cursor = Cursor::new(raw.as_slice()); @@ -182,6 +191,20 @@ pub(crate) async fn add_entries( .await } +/// Return true iff the request's `Content-Type` is +/// `application/octet-stream`, ignoring any parameters (e.g. charset). +fn content_type_is_octet_stream(headers: &axum::http::HeaderMap) -> bool { + headers + .get(CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + .unwrap_or_default() + .split(';') + .next() + .unwrap_or_default() + .trim() + == "application/octet-stream" +} + /// Read, verify, and persist the entry packages for `[upload_start, /// upload_end)`, then produce the HTTP response. /// @@ -453,13 +476,25 @@ async fn cosign_and_serve( if committed.size != header.upload_end { // The DO refused to rewind: a concurrent commit already advanced // the mirror checkpoint past upload_end, so ours was skipped. + // Fetch the latest state so the 409 mirror-info body advertises + // the current pending size and next entry, not the stale snapshot + // from the start of this request. log::info!( "add-entries: commit skipped, mirror checkpoint {} already past upload_end {}; \ returning 409", committed.size, header.upload_end, ); - return Ok(mirror_info_409(env, snapshot, &header.log_origin)); + let fresh_snapshot = match fetch_snapshot(env, &header.log_origin).await { + Ok(s) => s, + Err(e) => { + log::warn!( + "add-entries: failed to fetch fresh snapshot for 409; using stale: {e:?}" + ); + snapshot.clone() + } + }; + return Ok(mirror_info_409(env, &fresh_snapshot, &header.log_origin)); } Ok(( @@ -892,7 +927,9 @@ impl HashReader for MapReader<'_> { #[cfg(test)] mod tests { - use super::{MapReader, excess_entries, verify_package}; + use super::{ + CONTENT_TYPE, MapReader, content_type_is_octet_stream, excess_entries, verify_package, + }; use crate::mirror_state_do::PendingCheckpoint; use std::collections::HashMap; use tlog_core::{Hash, Subtree, stored_hash_index, stored_hashes, tree_hash}; @@ -1043,4 +1080,34 @@ mod tests { // underflow. assert_eq!(excess_entries(5_000, 6_000, 4_000), 0); } + + #[test] + fn content_type_octet_stream_accepted() { + let mut headers = axum::http::HeaderMap::new(); + headers.insert(CONTENT_TYPE, "application/octet-stream".parse().unwrap()); + assert!(content_type_is_octet_stream(&headers)); + } + + #[test] + fn content_type_octet_stream_with_params_accepted() { + let mut headers = axum::http::HeaderMap::new(); + headers.insert( + CONTENT_TYPE, + "application/octet-stream; charset=binary".parse().unwrap(), + ); + assert!(content_type_is_octet_stream(&headers)); + } + + #[test] + fn content_type_missing_rejected() { + let headers = axum::http::HeaderMap::new(); + assert!(!content_type_is_octet_stream(&headers)); + } + + #[test] + fn content_type_non_octet_stream_rejected() { + let mut headers = axum::http::HeaderMap::new(); + headers.insert(CONTENT_TYPE, "application/json".parse().unwrap()); + assert!(!content_type_is_octet_stream(&headers)); + } } diff --git a/crates/mirror_worker/src/frontend_worker.rs b/crates/mirror_worker/src/frontend_worker.rs index b7ea7c51..c6bea375 100644 --- a/crates/mirror_worker/src/frontend_worker.rs +++ b/crates/mirror_worker/src/frontend_worker.rs @@ -29,7 +29,7 @@ use axum::{ Json, Router, body::Bytes, extract::{DefaultBodyLimit, State}, - http::{StatusCode, header}, + http::{HeaderValue, StatusCode, header}, response::IntoResponse, routing::{get, post}, }; @@ -56,6 +56,22 @@ fn start() { let _ = console_log::init_with_level(level); } +/// Middleware that adds `Accept-Encoding: gzip` to every response. +/// +/// [c2sp.org/tlog-mirror][spec] says mirrors SHOULD advertise supported +/// compression algorithms in responses so clients can compress future +/// `add-entries` request bodies. +/// +/// [spec]: https://c2sp.org/tlog-mirror#add-entries +async fn add_accept_encoding( + mut response: axum::http::Response, +) -> axum::http::Response { + response + .headers_mut() + .insert(header::ACCEPT_ENCODING, HeaderValue::from_static("gzip")); + response +} + /// Top-level `#[event(fetch)]` handler. Delegates to the axum router; /// unmatched routes return 404. #[event(fetch, respond_with_errors)] @@ -82,6 +98,7 @@ async fn fetch( .route("/add-entries", post(crate::add_entries::add_entries)) .route("/metadata", get(metadata)) .route("/", get(root)) + .layer(axum::middleware::map_response(add_accept_encoding)) .with_state(env) .call(req) .await @@ -414,3 +431,19 @@ fn tlog_size_conflict(current: &PendingCheckpoint) -> axum::response::Response { ) .into_response() } + +#[cfg(test)] +mod tests { + use super::add_accept_encoding; + use axum::http::header::ACCEPT_ENCODING; + + #[tokio::test] + async fn accept_encoding_middleware_adds_gzip() { + let response = axum::http::Response::new(axum::body::Body::empty()); + let response = add_accept_encoding(response).await; + assert_eq!( + response.headers().get(ACCEPT_ENCODING).unwrap().as_bytes(), + b"gzip" + ); + } +} From cba781126db9f1696f02edbdd6f0801dc10ef231 Mon Sep 17 00:00:00 2001 From: Luke Valenta Date: Tue, 4 Aug 2026 11:26:52 -0400 Subject: [PATCH 11/11] mirror_worker: address lbaquerofierro review on PR 264 --- crates/mirror_worker/src/add_entries.rs | 22 +++++-- crates/mirror_worker/src/commit.rs | 77 +++++++++++++++++++------ 2 files changed, 75 insertions(+), 24 deletions(-) diff --git a/crates/mirror_worker/src/add_entries.rs b/crates/mirror_worker/src/add_entries.rs index 040cbd29..4731066a 100644 --- a/crates/mirror_worker/src/add_entries.rs +++ b/crates/mirror_worker/src/add_entries.rs @@ -441,6 +441,11 @@ async fn cosign_and_serve( let cosig_body = tlog_witness::serialize_add_checkpoint_response(std::slice::from_ref(¬e_sig)); + // The served checkpoint is the log's signed note with the mirror's + // cosignature appended. Build it once so both the early-return and the + // `/commit` paths use the same bytes. + let checkpoint_obj = [target.signed_note_bytes.as_slice(), &cosig_body].concat(); + // When the upload only reaches the mirror checkpoint's current size, // the checkpoint at that size is already committed and served. There // is nothing to advance, and re-committing would redundantly rewrite @@ -448,6 +453,11 @@ async fn cosign_and_serve( // return a fresh cosignature without dispatching `/commit`. (Committed // is monotonic, so a stale-low snapshot only skips this optimization, // never the other way.) + // + // Note: this also means a previously-failed R2 checkpoint write won't + // self-heal at the same size; it heals on the next larger commit. That + // is acceptable because the durable committed state advanced before the + // R2 write, and duplicate cosignatures are worse than a lagging object. if header.upload_end <= snapshot.committed.size { return Ok(( StatusCode::OK, @@ -457,22 +467,24 @@ async fn cosign_and_serve( .into_response()); } - // The served checkpoint is the log's signed note with the mirror's - // cosignature appended. The DO writes it to R2 while advancing the + // The DO writes the cosigned checkpoint to R2 while advancing the // durable checkpoint under its commit lock. - let mut checkpoint_obj = target.signed_note_bytes.clone(); - checkpoint_obj.extend_from_slice(&cosig_body); let committed = dispatch_commit( env, &header.log_origin, &CommitRequest { size: header.upload_end, hash: target.hash, - signed_note_bytes: checkpoint_obj, + signed_note_bytes: checkpoint_obj.clone(), }, ) .await?; + debug_assert_eq!( + committed.signed_note_bytes, checkpoint_obj, + "DO returned checkpoint bytes that do not match the cosigned note we sent" + ); + if committed.size != header.upload_end { // The DO refused to rewind: a concurrent commit already advanced // the mirror checkpoint past upload_end, so ours was skipped. diff --git a/crates/mirror_worker/src/commit.rs b/crates/mirror_worker/src/commit.rs index b8f2d4d1..6f0bbf02 100644 --- a/crates/mirror_worker/src/commit.rs +++ b/crates/mirror_worker/src/commit.rs @@ -29,6 +29,10 @@ use std::collections::HashMap; +use futures_util::{ + future::try_join_all, + stream::{StreamExt as _, TryStreamExt as _, iter as stream_iter}, +}; use generic_log_worker::{ ObjectBackend, log_ops::{HashReaderWithOverlay, TileWithBytes, UploadOptions, read_edge_tiles}, @@ -44,6 +48,14 @@ use worker::*; /// tlog-tiles fixes a tile height of 8, i.e. 256 entries per full tile. const TILE_WIDTH: u64 = TlogTile::FULL_WIDTH as u64; +/// Max in-flight R2 uploads per commit. A single `add-entries` request is +/// bounded only by the request-body size, so uploading every entry bundle +/// and hash tile at once could open thousands of connections and hold all +/// their bytes in memory. Workers allows only six connections to be waiting +/// on response headers at a time, so a small bound captures the concurrency +/// win without the unbounded fan-out. +const UPLOAD_CONCURRENCY: usize = 6; + /// Object key for the (cosigned) checkpoint the mirror serves at /// `//checkpoint`. Matches /// [`generic_log_worker::log_ops::CHECKPOINT_KEY`]. @@ -162,14 +174,17 @@ async fn authenticated_leaf_hashes( } } - let mut tiles: HashMap> = HashMap::new(); - for tile in recorder.0.into_inner() { - let bytes = object - .fetch(tile.path()) - .await? - .ok_or_else(|| Error::from(format!("persisted hash tile missing: {}", tile.path())))?; - tiles.insert(tile, bytes); - } + let tile_futures = recorder.0.into_inner().into_iter().map(|tile| { + let path = tile.path(); + async move { + let bytes = object + .fetch(path.clone()) + .await? + .ok_or_else(|| Error::from(format!("persisted hash tile missing: {path}")))?; + Ok::<(TlogTile, Vec), Error>((tile, bytes)) + } + }); + let tiles: HashMap> = try_join_all(tile_futures).await?.into_iter().collect(); let reader = PreloadedTlogTileReader(tiles); TileHashReader::new(tree_size, tree_hash, &reader) @@ -317,9 +332,12 @@ pub(crate) async fn persist_entries( verify_partial_bundle(&data_tile, subtree_start, persisted_size, &edge_tiles)?; } - // Replay leaves, flushing entry bundles at 256-entry boundaries. + // Replay leaves, buffering entry-bundle uploads until the end so they + // can be issued with bounded concurrency. Bundles are immutable and + // idempotent, so overlapping them is safe. let mut overlay: HashMap = HashMap::new(); let mut n = persisted_size; + let mut bundle_uploads = Vec::new(); for entry in entries { push_tile_leaf(&mut data_tile, entry)?; let hashes = stored_hashes_for_record_hash( @@ -336,16 +354,31 @@ pub(crate) async fn persist_entries( } n += 1; if n.is_multiple_of(TILE_WIDTH) { - upload_entry_bundle(object, n, std::mem::take(&mut data_tile)).await?; + bundle_uploads.push(upload_entry_bundle( + object, + n, + std::mem::take(&mut data_tile), + )); } } debug_assert_eq!(n, target_size); // Trailing partial entry bundle. if !target_size.is_multiple_of(TILE_WIDTH) { - upload_entry_bundle(object, target_size, std::mem::take(&mut data_tile)).await?; + bundle_uploads.push(upload_entry_bundle( + object, + target_size, + std::mem::take(&mut data_tile), + )); } - - // (Re)compute and upload hash tiles. + stream_iter(bundle_uploads) + .buffer_unordered(UPLOAD_CONCURRENCY) + .try_collect::<()>() + .await?; + + // (Re)compute hash tiles, then upload them with bounded concurrency + // while keeping edge_tiles current for the final root hash. + let tile_opts = immutable_tile_opts(); + let mut hash_uploads = Vec::new(); for tile in TlogTile::new_tiles(persisted_size, target_size) { let bytes = tile .read_data(&HashReaderWithOverlay { @@ -353,13 +386,19 @@ pub(crate) async fn persist_entries( overlay: &overlay, }) .map_err(|e| Error::from(format!("couldn't build hash tile {tile:?}: {e}")))?; - object - .upload(tile.path(), bytes.clone(), &immutable_tile_opts()) - .await?; - // Keep edge_tiles current so read_data of a higher/next tile can - // still resolve persisted hashes it depends on. - edge_tiles.insert(tile.level(), TileWithBytes { tile, b: bytes }); + edge_tiles.insert( + tile.level(), + TileWithBytes { + tile, + b: bytes.clone(), + }, + ); + hash_uploads.push(object.upload(tile.path(), bytes, &tile_opts)); } + stream_iter(hash_uploads) + .buffer_unordered(UPLOAD_CONCURRENCY) + .try_collect::<()>() + .await?; // Recompute the root hash from the frontier we just built. tlog_core::tree_hash(