perf(log-store): reuse the release records the object store WAL recovery already read (#9645)

* perf(log-store): reuse the release records the object store WAL recovery already read

Recovery reads every object it lists to verify its header and footer: an
object of at most 64 KiB whole, a larger one as its header and the last
64 KiB. It then read the segment of release records of every indexed
object that holds one again with a range read, although the segment, the
last one before the footer, is almost always among the bytes already read.
A prefix of many dropped regions, each released in an object of its own,
paid two reads per such object.

The footer read now keeps a copy of that segment when the bytes it read
hold it, and recovery decodes the copy, or reads the segment as before when
the copy is missing, only for the objects on the selected chain. The
segment is copied so that the read window is not held until the releases
are decoded. Chain selection, the merge of the largest released entry id
per region, and the failure of recovery on a release segment that cannot
be fetched or decoded, whatever on_corrupted_segment says, are unchanged.

Tests cover the reads of a recovery whose release segments are all in the
read bytes, a release segment that starts before the tail window, and that
a damaged release segment fails recovery on the chain and is never decoded
off it. An ignored measurement test counts the reads and times recovery
over a prefix of dropped regions.

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* perf(log-store): bound the release segments object store WAL recovery keeps

Recovery keeps the segment of release records of every object it fetches
before it selects the chain, so an object off the chain could hold up to
the tail window until recovery ends. Keep a segment only up to 4 KiB, far
above the records of the few regions a release seals at once, and read a
larger one again only when its object is on the chain, as before.

Tests cover a release segment inside the tail window of an object read as
its header and its tail, which exercises the offset into the window, and a
segment above the kept size in an object read whole.

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

---------

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

src/log-store/src/object_store_wal/store.rs

@@ -123,8 +123,16 @@ const READ_PREFETCH_DEPTH: usize = 8;
 const READ_PREFETCH_BYTES: usize = 64 * 1024 * 1024;
 /// Bytes recovery reads from the end of an object in one request. The window
 /// holds the trailer and the footer of an object with up to 1364 regions of
-/// 48 bytes each, so a second request for the footer is rare.
+/// 48 bytes each, so a second request for the footer is rare, and usually the
+/// segment of release records, which is the last segment.
 const RECOVERY_TAIL_WINDOW: usize = 64 * 1024;
+/// Largest segment of release records that recovery keeps from the bytes it
+/// read for a footer. It keeps them before it knows which objects are on the
+/// chain, so the bound caps what objects off the chain hold until recovery
+/// ends; a larger segment is read again if its object is on the chain. A
+/// release seals its record at once, so a segment usually holds the records of
+/// one or a few regions, 16 bytes each.
+const MAX_KEPT_RELEASE_BYTES: u64 = 4 * 1024;
 
 /// A log store over the immutable WAL objects under one prefix.
 ///
@@ -3031,11 +3039,13 @@ struct FetchedObjects {
     vanished: Vec<(u64, Error)>,
 }
 
-/// An object as recovery fetched it: its key, header and footer.
+/// An object as recovery fetched it: its key, header and footer, and its
+/// segment of release records when the reads of the footer returned it.
 struct FetchedObject {
     object: ListedObject,
     header: Header,
     footer: Vec<FooterEntry>,
+    releases: Option<Bytes>,
 }
 
 /// Rebuilds the catalog from object headers and footers,
@@ -3048,16 +3058,25 @@ struct FetchedObject {
 /// Only the objects on the chain [`select_chain`] picks are indexed.
 async fn recover(io: &dyn WalObjectIo) -> Result<Recovered> {
     let objects = io.list().await?;
-    let mut recovered = finish_recovery(fetch_footers(io, objects, RECOVERY_CONCURRENCY).await?)?;
-    recovered.released_entry_ids = read_releases(io, &recovered.catalog).await?;
+    let mut fetched = fetch_footers(io, objects, RECOVERY_CONCURRENCY).await?;
+    let releases = fetched
+        .objects
+        .iter_mut()
+        .filter_map(|fetched| Some((fetched.object.object_seq, fetched.releases.take()?)))
+        .collect();
+    let mut recovered = finish_recovery(fetched)?;
+    recovered.released_entry_ids = read_releases(io, &recovered.catalog, releases).await?;
     Ok(recovered)
 }
 
 /// Reads the release records of the indexed objects, which seed the
-/// watermarks of their regions: one range read of the segment of
-/// [`RELEASE_REGION_ID`] per object that holds one. A segment that cannot be
-/// fetched or does not decode fails recovery, whatever `on_corrupted_segment`
-/// says: skipping it would lose the watermarks it seeds.
+/// watermarks of their regions, from the segment of [`RELEASE_REGION_ID`] of
+/// every indexed object that holds one: the bytes in `fetched` by object
+/// sequence, which the reads of the footers returned, or else one range read.
+/// The segments of objects that are not indexed are never decoded. A segment
+/// that cannot be fetched or does not decode fails recovery, whatever
+/// `on_corrupted_segment` says: skipping it would lose the watermarks it
+/// seeds.
 ///
 /// A release record needs no bookkeeping of its own. It is sealed once every
 /// entry it covers was admitted, so it lies in the object holding the last
@@ -3073,6 +3092,7 @@ async fn recover(io: &dyn WalObjectIo) -> Result<Recovered> {
 async fn read_releases(
     io: &dyn WalObjectIo,
     catalog: &ObjectCatalog,
+    mut fetched: HashMap<u64, Bytes>,
 ) -> Result<HashMap<RegionId, EntryId>> {
     let segments = catalog
         .objects_in_order()
@@ -3080,14 +3100,18 @@ async fn read_releases(
             footer
                 .iter()
                 .find(|entry| entry.region_id == RELEASE_REGION_ID)
-                .map(|entry| (object_seq, entry.clone()))
+                .map(|entry| (object_seq, entry.clone(), fetched.remove(&object_seq)))
         })
         .collect::<Vec<_>>();
     let releases = futures::stream::iter(segments)
-        .map(|(object_seq, entry)| async move {
-            let bytes = io
-                .get_range(object_seq, entry.segment_offset, entry.segment_len)
-                .await?;
+        .map(|(object_seq, entry, fetched)| async move {
+            let bytes = match fetched {
+                Some(bytes) => bytes,
+                None => {
+                    io.get_range(object_seq, entry.segment_offset, entry.segment_len)
+                        .await?
+                }
+            };
             decode_segment(&bytes, &entry)
                 .and_then(|records| decode_releases(&records[0].payload))
                 .with_context(|_| InvalidWalObjectSnafu {
@@ -3345,10 +3369,11 @@ async fn fetch_footers(
     let outcomes = futures::stream::iter(objects)
         .map(|object| async move {
             match fetch_footer(io, &object).await {
-                Ok((header, footer)) => Ok(Ok(FetchedObject {
+                Ok((header, footer, releases)) => Ok(Ok(FetchedObject {
                     object,
                     header,
                     footer,
+                    releases,
                 })),
                 Err(error) if is_not_found(&error) => Ok(Err((object.object_seq, error))),
                 Err(error) => Err(error),
@@ -3376,7 +3401,9 @@ fn is_not_found(error: &Error) -> bool {
 /// Reads the header, trailer and footer of `object` and verifies them: the
 /// header must carry the sequence of the key, the trailer must be well formed,
 /// the footer must match the checksum the trailer holds and its segments must
-/// tile the object body.
+/// tile the object body. Also returns a copy of the segment of release records
+/// when the bytes read hold it and it is at most [`MAX_KEPT_RELEASE_BYTES`],
+/// which is not decoded here.
 ///
 /// A short object is read whole. Otherwise the header and a window at the end
 /// of the object are read concurrently, and the footer is read separately only
@@ -3384,7 +3411,7 @@ fn is_not_found(error: &Error) -> bool {
 async fn fetch_footer(
     io: &dyn WalObjectIo,
     object: &ListedObject,
-) -> Result<(Header, Vec<FooterEntry>)> {
+) -> Result<(Header, Vec<FooterEntry>, Option<Bytes>)> {
     let ListedObject {
         object_seq, size, ..
     } = *object;
@@ -3442,7 +3469,18 @@ async fn fetch_footer(
     };
     let footer = decode_footer(&footer, trailer).map_err(invalid)?;
     verify_segment_ranges(&footer, footer_range.start).map_err(invalid)?;
-    Ok((header, footer))
+    // A copy, so the window is not kept until the releases are decoded.
+    let releases = footer
+        .iter()
+        .find(|entry| entry.region_id == RELEASE_REGION_ID)
+        .filter(|entry| {
+            entry.segment_offset >= tail_start as u64 && entry.segment_len <= MAX_KEPT_RELEASE_BYTES
+        })
+        .map(|entry| {
+            let start = entry.segment_offset as usize - tail_start;
+            Bytes::copy_from_slice(&tail[start..start + entry.segment_len as usize])
+        });
+    Ok((header, footer, releases))
 }
 
 /// Verifies the header and trailer of the object `object_seq` of `object_len`
@@ -3968,6 +4006,7 @@ mod tests {
                 object,
                 header: decoded.header,
                 footer: decoded.footer,
+                releases: None,
             });
         }
         finish_recovery(FetchedObjects {
@@ -9167,6 +9206,206 @@ mod tests {
         store.stop().await.unwrap();
     }
 
+    #[tokio::test]
+    async fn test_store_recovery_takes_release_records_from_the_footer_reads() {
+        let _serialized = SERIALIZED.lock().await;
+        let object_store = memory_store();
+        let (one, two, three) = (region(1), region(2), region(3));
+        let store = open(object_store.clone(), &eager()).await;
+        // Object 1 holds entries of regions one and two, object 2 the release
+        // of region one and object 3 the drop of region three; region two
+        // holds the boundary at object 1.
+        store
+            .append_batch(vec![entry(&store, one, "a1"), entry(&store, two, "b1")])
+            .await
+            .unwrap();
+        store.release_region(&provider(one), one).await.unwrap();
+        store.obsolete_all(&provider(three), three).await.unwrap();
+        store.stop().await.unwrap();
+
+        // Every object is shorter than the tail window, so recovery reads
+        // each one whole, once, and decodes the release records from those
+        // bytes without a range read.
+        let (io, reads) = RecordingIo::over(object_store);
+        let store = open_over(io, &eager()).await;
+        assert_eq!(Some(id(1, 1)), watermark(&store, one));
+        assert_eq!(None, watermark(&store, two));
+        assert_eq!(Some(EntryId::MAX), watermark(&store, three));
+        assert!(reads.lock().unwrap().is_empty());
+        store.stop().await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_store_recovery_reads_a_release_segment_outside_the_tail_window() {
+        let _serialized = SERIALIZED.lock().await;
+        let object_store = memory_store();
+        // So many release records that their segment starts before the
+        // window at the end of the object.
+        let releases = (1..=5_000)
+            .map(|number| (region(number), id(0, 1)))
+            .collect::<BTreeMap<_, _>>();
+        let record = Record {
+            region_id: RELEASE_REGION_ID,
+            entry_id: id(0, 1),
+            payload: encode_releases(&releases),
+        };
+        put_records(&object_store, 0, &[record]).await;
+        let (io, reads) = RecordingIo::over(object_store);
+        let object = io.list().await.unwrap().remove(0);
+        let (_, footer) = footer_of(&io.get(0).await.unwrap());
+        let segment = &footer[0];
+        assert!(segment.segment_offset < object.size - RECOVERY_TAIL_WINDOW as u64);
+
+        let store = open_over(io, &eager()).await;
+        for number in [1, 5_000] {
+            assert_eq!(Some(id(0, 1)), watermark(&store, region(number)));
+        }
+        let mut reads = reads.lock().unwrap().clone();
+        reads.sort_unstable();
+        assert_eq!(
+            vec![
+                (0, 0, HEADER_LEN as u64),
+                (0, segment.segment_offset, segment.segment_len),
+                (
+                    0,
+                    object.size - RECOVERY_TAIL_WINDOW as u64,
+                    RECOVERY_TAIL_WINDOW as u64
+                ),
+            ],
+            reads
+        );
+        store.stop().await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_store_recovery_takes_a_release_segment_from_the_tail_window() {
+        let _serialized = SERIALIZED.lock().await;
+        let object_store = memory_store();
+        // A segment longer than the window comes first, so the object is read
+        // as its header and its tail, and the segment of release records, the
+        // last one, lies inside the tail but not at its start.
+        let records = [
+            Record {
+                region_id: region(1),
+                entry_id: id(0, 1),
+                payload: Bytes::from(vec![0; RECOVERY_TAIL_WINDOW * 2]),
+            },
+            Record {
+                region_id: RELEASE_REGION_ID,
+                entry_id: id(0, 1),
+                payload: encode_releases(&BTreeMap::from([(region(2), 7)])),
+            },
+        ];
+        put_records(&object_store, 0, &records).await;
+        let (io, reads) = RecordingIo::over(object_store);
+        let object = io.list().await.unwrap().remove(0);
+        let tail_start = object.size - RECOVERY_TAIL_WINDOW as u64;
+        let (_, footer) = footer_of(&io.get(0).await.unwrap());
+        let segment = footer
+            .iter()
+            .find(|entry| entry.region_id == RELEASE_REGION_ID)
+            .unwrap();
+        assert!(segment.segment_offset > tail_start);
+
+        let store = open_over(io, &eager()).await;
+        assert_eq!(Some(7), watermark(&store, region(2)));
+        let mut reads = reads.lock().unwrap().clone();
+        reads.sort_unstable();
+        assert_eq!(
+            vec![
+                (0, 0, HEADER_LEN as u64),
+                (0, tail_start, RECOVERY_TAIL_WINDOW as u64)
+            ],
+            reads
+        );
+        store.stop().await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_store_recovery_reads_a_release_segment_above_the_kept_size_again() {
+        let _serialized = SERIALIZED.lock().await;
+        let object_store = memory_store();
+        // The object is read whole, but its segment of release records is
+        // larger than recovery keeps.
+        let releases = (1..=300)
+            .map(|number| (region(number), id(0, 1)))
+            .collect::<BTreeMap<_, _>>();
+        let record = Record {
+            region_id: RELEASE_REGION_ID,
+            entry_id: id(0, 1),
+            payload: encode_releases(&releases),
+        };
+        put_records(&object_store, 0, &[record]).await;
+        let (io, reads) = RecordingIo::over(object_store);
+        let object = io.list().await.unwrap().remove(0);
+        let (_, footer) = footer_of(&io.get(0).await.unwrap());
+        let segment = &footer[0];
+        assert!(object.size <= RECOVERY_TAIL_WINDOW as u64);
+        assert!(segment.segment_len > MAX_KEPT_RELEASE_BYTES);
+
+        let store = open_over(io, &eager()).await;
+        for number in [1, 300] {
+            assert_eq!(Some(id(0, 1)), watermark(&store, region(number)));
+        }
+        assert_eq!(
+            vec![(0, segment.segment_offset, segment.segment_len)],
+            *reads.lock().unwrap()
+        );
+        store.stop().await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_store_recovery_decodes_release_records_only_on_the_chain() {
+        let _serialized = SERIALIZED.lock().await;
+        let object_store = memory_store();
+        let io = ObjectStoreIo::new(object_store.clone(), PREFIX).unwrap();
+        let release_record = |object_seq, number| Record {
+            region_id: RELEASE_REGION_ID,
+            entry_id: id(object_seq, 1),
+            payload: encode_releases(&BTreeMap::from([(region(number), 1)])),
+        };
+        let damage_release_segment = |bytes: &mut Vec<u8>| {
+            let (_, footer) = footer_of(bytes);
+            let segment = footer
+                .iter()
+                .find(|entry| entry.region_id == RELEASE_REGION_ID)
+                .unwrap();
+            let last = segment.segment_offset + segment.segment_len - 1;
+            bytes[last as usize] ^= 1;
+        };
+        // Object 0 holds an entry of region one and the release of region
+        // two. Object 1 releases region three but is off the chain: it links
+        // to an object 0 of another epoch.
+        let entry_record = Record {
+            region_id: region(1),
+            entry_id: id(0, 1),
+            payload: Bytes::from_static(b"a1"),
+        };
+        put_records(&object_store, 0, &[entry_record, release_record(0, 2)]).await;
+        let orphan = Header {
+            object_seq: 1,
+            epoch: 1,
+            prev: Some(ChainLink {
+                object_seq: 0,
+                epoch: 2,
+            }),
+            reclaim_boundary: 0,
+        };
+        put_object_with_header(&io, orphan, &[release_record(1, 3)]).await;
+
+        // A damaged release segment off the chain is never decoded.
+        corrupt_object(&object_store, &io.object_path(1), damage_release_segment).await;
+        let store = open(object_store.clone(), &eager()).await;
+        assert_eq!(Some(1), watermark(&store, region(2)));
+        assert_eq!(None, watermark(&store, region(3)));
+        store.stop().await.unwrap();
+
+        // A damaged release segment on the chain fails recovery.
+        let path = io.object_path(0);
+        corrupt_object(&object_store, &path, damage_release_segment).await;
+        assert_recovery_rejects(&object_store, &path, "checksum mismatch").await;
+    }
+
     #[tokio::test]
     async fn test_store_carries_the_record_of_a_failed_release() {
         let _serialized = SERIALIZED.lock().await;
@@ -9687,6 +9926,123 @@ mod tests {
         }
     }
 
+    /// Measurements of costs that grow with the prefix, taken through the
+    /// store. They are ignored by default and print what they measure; run
+    /// them with `--run-ignored only` and `--no-capture`, preferably with
+    /// `--cargo-profile release`.
+    #[allow(clippy::print_stdout)]
+    mod measurements {
+        use super::*;
+
+        /// Object access that counts the reads of whole objects and of byte
+        /// ranges.
+        struct CountingIo {
+            inner: ObjectStoreIo,
+            gets: AtomicUsize,
+            range_gets: AtomicUsize,
+        }
+
+        #[async_trait::async_trait]
+        impl WalObjectIo for CountingIo {
+            async fn put_if_absent(&self, object_seq: u64, content: Bytes) -> Result<PutResult> {
+                self.inner.put_if_absent(object_seq, content).await
+            }
+
+            async fn get(&self, object_seq: u64) -> Result<Bytes> {
+                self.gets.fetch_add(1, Ordering::Relaxed);
+                self.inner.get(object_seq).await
+            }
+
+            async fn get_range(&self, object_seq: u64, offset: u64, len: u64) -> Result<Bytes> {
+                self.range_gets.fetch_add(1, Ordering::Relaxed);
+                self.inner.get_range(object_seq, offset, len).await
+            }
+
+            async fn delete(&self, object_seq: u64) -> Result<()> {
+                self.inner.delete(object_seq).await
+            }
+
+            async fn list(&self) -> Result<Vec<ListedObject>> {
+                self.inner.list().await
+            }
+
+            fn object_path(&self, object_seq: u64) -> String {
+                self.inner.object_path(object_seq)
+            }
+
+            fn storage(&self) -> String {
+                self.inner.storage()
+            }
+        }
+
+        /// The object store a measurement of recovery runs on: a fresh root
+        /// in the bucket the `GT_S3_*` variables name when they are set, an
+        /// in-memory store otherwise.
+        fn measured_object_store() -> (ObjectStore, &'static str) {
+            let Ok(bucket) = std::env::var("GT_S3_BUCKET") else {
+                return (memory_store(), "memory");
+            };
+            let variable = |name: &str| std::env::var(name).unwrap();
+            let mut builder = object_store::services::S3::default()
+                .root(&format!("measurements/{}", uuid::Uuid::new_v4()))
+                .bucket(&bucket)
+                .region(&variable("GT_S3_REGION"))
+                .access_key_id(&variable("GT_S3_ACCESS_KEY_ID"))
+                .secret_access_key(&variable("GT_S3_ACCESS_KEY"));
+            if let Ok(endpoint) = std::env::var("GT_S3_ENDPOINT_URL") {
+                builder = builder.endpoint(&endpoint);
+            }
+            (ObjectStore::new(builder).unwrap(), "s3")
+        }
+
+        /// Counts the reads of a recovery over a prefix where every dropped
+        /// region left its release record in an object of its own, while one
+        /// region still holds the boundary at the object of the first
+        /// entries, and times the recovery.
+        #[tokio::test]
+        #[ignore]
+        async fn measure_recovery_of_dropped_regions() {
+            let _serialized = SERIALIZED.lock().await;
+            for regions in [1_000, 4_000] {
+                let (object_store, storage) = measured_object_store();
+                let store = open(object_store.clone(), &eager()).await;
+                let entries = (1..=regions + 1)
+                    .map(|number| entry(&store, region(number), "a"))
+                    .collect();
+                store.append_batch(entries).await.unwrap();
+                for number in 1..=regions {
+                    let region_id = region(number);
+                    store
+                        .obsolete_all(&provider(region_id), region_id)
+                        .await
+                        .unwrap();
+                }
+                store.stop().await.unwrap();
+
+                let io = Arc::new(CountingIo {
+                    inner: ObjectStoreIo::new(object_store, PREFIX).unwrap(),
+                    gets: AtomicUsize::new(0),
+                    range_gets: AtomicUsize::new(0),
+                });
+                let objects = io.list().await.unwrap().len();
+                let started_at = Instant::now();
+                let store = open_over(io.clone(), &eager()).await;
+                let elapsed = started_at.elapsed();
+                for number in 1..=regions {
+                    assert_eq!(Some(EntryId::MAX), watermark(&store, region(number)));
+                }
+                assert_eq!(None, watermark(&store, region(regions + 1)));
+                println!(
+                    "recovery of dropped regions: storage={storage} regions={regions} objects={objects} gets={} range_gets={} recovery_ms={:.3}",
+                    io.gets.load(Ordering::Relaxed),
+                    io.range_gets.load(Ordering::Relaxed),
+                    elapsed.as_secs_f64() * 1e3,
+                );
+                store.stop().await.unwrap();
+            }
+        }
+    }
+
     /// Drives every outcome of a conditional create and every failure of the
     /// object store access through the store, one case per fault, and checks
     /// that every row of the *Failure matrix* has one.