mirror_worker: serve cut tiles when cosigning below the frontier

The frontier>upload_end case was mishandled: when a prior upload (or a
racing client with an old ticket) had already persisted past upload_end,
an add-entries whose packages were all already-persisted would cosign a
checkpoint at upload_end even though the edge tiles were only written at
the larger frontier. The narrower "cut" tiles a tree of size upload_end
requires could be missing, so the cosignature was unverifiable against
the mirror's own tiles.

The spec mandates a 200 cosignature here (upload_end >= committed, all
packages received) and that the tree at upload_end be servable first, so
synthesize the cut tiles before cosigning, matching Sunlight's
ensureCutTiles. This applies whether or not upload_end is tile-aligned:
an aligned size below the frontier still needs upper-level partial hash
tiles the frontier never wrote.

ensure_cut_tiles re-persists [base, cut_size) on top of the aligned
prefix via persist_entries, rewriting exactly the edge tiles at
upload_end; re-uploads are idempotent. The frontier-below-upload_end
guard is unchanged (still a 202 to resume).

Author: lukevalenta

crates/mirror_worker/src/add_entries.rs

@@ -350,19 +350,13 @@ async fn verify_and_persist(
         ));
     }
 
-    // Spec: the mirror updates its checkpoint to `upload_end` only once
-    // "the next entry will be greater or equal to `upload_end`", i.e. all
-    // entries up to `upload_end` are durably persisted. A request that
-    // persists nothing (e.g. an empty body, or one whose packages are all
-    // already-persisted) must not let us cosign past our frontier: without
-    // this guard `upload_end` above `next_entry` would sign a checkpoint at
-    // a size we never wrote tiles for. When the frontier has not reached
-    // `upload_end`, treat it like a truncated upload and 202 so the client
-    // resumes from the advertised next entry.
+    // Frontier below `upload_end`: not every entry up to `upload_end` is
+    // persisted yet (a truncated body, or an already-persisted request that
+    // stops short). This is the spec's "not yet received all packages" case,
+    // so 202 with the advanced frontier for the client to resume from.
     if frontier_size < header.upload_end {
         log::info!(
-            "add-entries: frontier {frontier_size} below upload_end {}; nothing to persist \
-             this request, returning 202 to resume",
+            "add-entries: frontier {frontier_size} below upload_end {}; returning 202 to resume",
             header.upload_end,
         );
         return Ok(mirror_info_202(
@@ -400,6 +394,17 @@ async fn verify_and_persist(
         ));
     }
 
+    // Frontier ahead of `upload_end` (a prior upload persisted past it, or a
+    // racing client): the edge tiles were written at the larger frontier, so
+    // the narrower "cut" tiles a tree of size `upload_end` needs may be
+    // missing and a cosignature at `upload_end` would be unverifiable against
+    // our own tiles. Spec requires the tree at `upload_end` be servable before
+    // we cosign, so synthesize those cut tiles first.
+    if frontier_size > header.upload_end {
+        let bucket = load_origin_bucket(env, &header.log_origin)?;
+        commit::ensure_cut_tiles(&bucket, header.upload_end, frontier_size, frontier_hash).await?;
+    }
+
     cosign_and_serve(env, header, target, snapshot).await
 }
 

crates/mirror_worker/src/commit.rs

@@ -39,7 +39,8 @@ use generic_log_worker::{
 };
 use length_prefixed::{ReadLengthPrefixedBytesExt as _, WriteLengthPrefixedBytesExt as _};
 use tlog_core::{
-    Hash, HashReader as _, TlogError, record_hash, stored_hash_index, stored_hashes_for_record_hash,
+    EMPTY_HASH, Hash, HashReader, TlogError, record_hash, stored_hash_index,
+    stored_hashes_for_record_hash,
 };
 use tlog_tiles::{PathElem, PreloadedTlogTileReader, TileHashReader, TlogTile, TlogTileRecorder};
 #[allow(clippy::wildcard_imports)]
@@ -162,6 +163,26 @@ async fn authenticated_leaf_hashes(
     tree_hash: Hash,
     indexes: &[u64],
 ) -> Result<Vec<Hash>> {
+    let tiles = fetch_authenticated_tiles(object, tree_size, indexes).await?;
+    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:?}")))
+}
+
+/// Fetch (from storage) the hash tiles needed to prove `indexes` against
+/// `(tree_size, tree_hash)`. The returned map feeds a [`TileHashReader`] whose
+/// verifying pass authenticates them; a missing or stale tile surfaces as an
+/// error rather than a downstream client rejection.
+///
+/// # Errors
+///
+/// Returns an error if a required hash tile is missing from storage.
+async fn fetch_authenticated_tiles(
+    object: &impl ObjectBackend,
+    tree_size: u64,
+    indexes: &[u64],
+) -> Result<HashMap<TlogTile, Vec<u8>>> {
     // Recording pass: `TlogTileRecorder` short-circuits `read_hashes` with
     // `RecordedTilesOnly` after collecting the tiles it would need.
     let recorder = TlogTileRecorder::default();
@@ -184,12 +205,7 @@ async fn authenticated_leaf_hashes(
             Ok::<(TlogTile, Vec<u8>), Error>((tile, bytes))
         }
     });
-    let tiles: HashMap<TlogTile, Vec<u8>> = try_join_all(tile_futures).await?.into_iter().collect();
-
-    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:?}")))
+    Ok(try_join_all(tile_futures).await?.into_iter().collect())
 }
 
 /// Authenticate a reloaded partial entry bundle before it is extended and
@@ -435,6 +451,89 @@ pub(crate) async fn write_checkpoint(object: &impl ObjectBackend, bytes: Vec<u8>
         .await
 }
 
+/// Ensure the partial "cut" tiles for a tree of size `cut_size` exist, so a
+/// checkpoint at `cut_size` is servable per [tlog-tiles][tiles].
+///
+/// Needed when the persisted frontier advanced past `cut_size` (a prior
+/// upload, or a racing client): the frontier wrote wider partial tiles, so the
+/// narrower ones `cut_size` requires (the level-0 data bundle when mid-tile,
+/// and upper-level partial hash tiles) may be missing.
+///
+/// Re-persists the leaves `[base, cut_size)`, `base` being the largest
+/// 256-boundary below `cut_size`, on top of the aligned prefix, which rewrites
+/// exactly the edge tiles at `cut_size`. Re-uploads are idempotent, so racing
+/// a concurrent commit is safe. `frontier_size`/`frontier_hash` locate and
+/// authenticate the wider stored bundle and the aligned prefix root.
+///
+/// # Errors
+///
+/// Returns an error on a storage fault, or if the reloaded leaves fail
+/// authentication.
+pub(crate) async fn ensure_cut_tiles(
+    object: &impl ObjectBackend,
+    cut_size: u64,
+    frontier_size: u64,
+    frontier_hash: Hash,
+) -> Result<()> {
+    if cut_size == 0 {
+        return Ok(());
+    }
+    let base = ((cut_size - 1) / TILE_WIDTH) * TILE_WIDTH;
+
+    // Reload the cut leaves from the wider stored bundle, authenticated
+    // against the frontier tiles.
+    let leaves =
+        read_persisted_leaves(object, base, cut_size - base, frontier_size, frontier_hash).await?;
+
+    // Root of the aligned prefix, the base persist_entries grows from,
+    // authenticated against the frontier tree (whose edge tiles exist).
+    let prefix_hash = if base == 0 {
+        EMPTY_HASH
+    } else {
+        subtree_prefix_hash(object, base, frontier_size, frontier_hash).await?
+    };
+
+    persist_entries(object, base, prefix_hash, cut_size, &leaves).await?;
+    Ok(())
+}
+
+/// Tree hash of the prefix `[0, prefix_size)`, where `prefix_size` is
+/// 256-aligned, read from the full hash tiles authenticated against the target
+/// `(tree_size, tree_hash)`.
+async fn subtree_prefix_hash(
+    object: &impl ObjectBackend,
+    prefix_size: u64,
+    tree_size: u64,
+    tree_hash: Hash,
+) -> Result<Hash> {
+    let indexes = tlog_core::tree_hash_indexes(prefix_size);
+    let tiles = fetch_authenticated_tiles(object, tree_size, &indexes).await?;
+    let reader = PreloadedTlogTileReader(tiles);
+    let hashes = TileHashReader::new(tree_size, tree_hash, &reader)
+        .read_hashes(&indexes)
+        .map_err(|e| Error::from(format!("read prefix hashes: {e:?}")))?;
+    tlog_core::tree_hash(prefix_size, &StaticHashReader(&indexes, &hashes))
+        .map_err(|e| Error::from(format!("compute prefix hash: {e:?}")))
+}
+
+/// A [`HashReader`] over a fixed index-to-hash mapping, for computing a tree
+/// hash from a preauthenticated set of stored hashes.
+struct StaticHashReader<'a>(&'a [u64], &'a [Hash]);
+impl HashReader for StaticHashReader<'_> {
+    fn read_hashes(&self, indexes: &[u64]) -> std::result::Result<Vec<Hash>, TlogError> {
+        indexes
+            .iter()
+            .map(|want| {
+                self.0
+                    .iter()
+                    .position(|i| i == want)
+                    .map(|p| self.1[p])
+                    .ok_or(TlogError::IndexesNotInTree)
+            })
+            .collect()
+    }
+}
+
 /// 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<u8>) -> Result<()> {
@@ -711,6 +810,95 @@ mod tests {
         );
     }
 
+    /// Assert a checkpoint at `size` is servable: the stored tiles
+    /// authenticate the last leaf against `reference_root(size)`, and the cut
+    /// data bundle decodes to the expected leaves.
+    async fn assert_servable_at(obj: &MemBackend, size: u64) {
+        use length_prefixed::ReadLengthPrefixedBytesExt as _;
+        let root = reference_root(size);
+
+        let idx = [stored_hash_index(0, size - 1)];
+        let tiles = fetch_authenticated_tiles(obj, size, &idx).await.unwrap();
+        let reader = PreloadedTlogTileReader(tiles);
+        let got = TileHashReader::new(size, root, &reader)
+            .read_hashes(&idx)
+            .expect("tiles authenticate against root");
+        assert_eq!(got[0], record_hash(&entry(size - 1)));
+
+        // For a mid-tile size, the cut data bundle must decode to exactly the
+        // trailing leaves [subtree_start, size).
+        if !size.is_multiple_of(TILE_WIDTH) {
+            let subtree_start = (size / TILE_WIDTH) * TILE_WIDTH;
+            let data_key = TlogTile::from_index(stored_hash_index(0, size - 1))
+                .with_data_path(PathElem::Entries)
+                .path();
+            let bytes = obj
+                .fetch(&data_key)
+                .await
+                .unwrap()
+                .expect("cut data bundle stored");
+            let mut cur: &[u8] = &bytes;
+            for i in subtree_start..size {
+                let got = cur.read_length_prefixed(2).unwrap();
+                assert_eq!(got, entry(i), "leaf {i} mismatch in cut bundle");
+            }
+            assert!(cur.is_empty(), "cut bundle has trailing bytes");
+        }
+    }
+
+    #[tokio::test]
+    async fn ensure_cut_tiles_first_block() {
+        let obj = MemBackend::default();
+        // Frontier at 300; a checkpoint at 100 (mid-tile, first block) needs
+        // its width-100 data bundle synthesized. The frontier only wrote the
+        // width-300 partial (a full width-256 tile 0 + width-44 tile 1).
+        let frontier = persist_entries(&obj, 0, EMPTY_HASH, 300, &leaves(0..300))
+            .await
+            .unwrap();
+        ensure_cut_tiles(&obj, 100, 300, frontier).await.unwrap();
+        assert_servable_at(&obj, 100).await;
+    }
+
+    #[tokio::test]
+    async fn ensure_cut_tiles_aligned_below_frontier() {
+        let obj = MemBackend::default();
+        // Frontier at 768; a checkpoint at a tile-aligned 256 still needs the
+        // upper-level partial hash tile (tile/1/000.p/1) synthesized: the
+        // frontier wrote tile/1/000.p/3, never the .p/1 a tree of 256 needs.
+        let frontier = persist_entries(&obj, 0, EMPTY_HASH, 768, &leaves(0..768))
+            .await
+            .unwrap();
+        ensure_cut_tiles(&obj, 256, 768, frontier).await.unwrap();
+        assert_servable_at(&obj, 256).await;
+    }
+
+    #[tokio::test]
+    async fn ensure_cut_tiles_later_block() {
+        let obj = MemBackend::default();
+        // Frontier at 1000; a checkpoint at 900 cuts a later block (base
+        // 768), exercising the prefix-hash path.
+        let frontier = persist_entries(&obj, 0, EMPTY_HASH, 1000, &leaves(0..1000))
+            .await
+            .unwrap();
+        ensure_cut_tiles(&obj, 900, 1000, frontier).await.unwrap();
+        assert_servable_at(&obj, 900).await;
+    }
+
+    #[tokio::test]
+    async fn ensure_cut_tiles_rejects_wrong_frontier_hash() {
+        let obj = MemBackend::default();
+        persist_entries(&obj, 0, EMPTY_HASH, 300, &leaves(0..300))
+            .await
+            .unwrap();
+        // A frontier hash that doesn't match storage must fail authentication
+        // rather than write mismatched tiles.
+        assert!(
+            ensure_cut_tiles(&obj, 100, 300, Hash([0xab; tlog_core::HASH_SIZE]))
+                .await
+                .is_err()
+        );
+    }
+
     #[tokio::test]
     async fn hash_tiles_authenticate_against_root() {
         let obj = MemBackend::default();