diff --git a/docs/internals/split-format.md b/docs/internals/split-format.md
index 7c41f9befc7..bd85ae8dbe9 100644
--- a/docs/internals/split-format.md
+++ b/docs/internals/split-format.md
@@ -9,22 +9,45 @@ with:
 - the Tantivy index files (`.idx`, `.pos`, `.term`...)
 - a Quickwit specific file with the list of fields, including those indexed as part of a JSON type. 
 It contains the field name, type and capabilities.
+- a versioned protobuf entry named `split_recovery_metadata` containing the immutable split
+  metadata and direct lineage needed to reconstruct its metastore record.
 
 The split file data layout looks like this:
 - concatenation all of the files in the split
 - a footer
 
-The footer follows the following format.
+The footer follows this format:
 
-- a json object called `BundleStorageFileOffsets` containing the `[start, end)` byte-offsets
+- a JSON object called `BundleFileRanges` containing the `[start, end)` byte ranges
 of all files.
-- the length of this json (8 bytes little endian)
+- the length of this json (`u32`, little endian)
 - a hotcache, a small static cache that contains some important file sections.
-- the length of this hotcache (8 bytes little endian)
+- the length of this hotcache (`u32`, little endian)
+- optionally, a fixed-size footer trailer:
+  - the inclusive footer start offset (`u64`, little endian)
+  - the trailer format version (`u32`, little endian)
+  - the four-byte magic value `QWFT`
 
-This footer plays a key role a very important role in quickwit.
+This footer plays a key role in Quickwit.
 It packs in one read all of the information required to open a split.
 
-When opening a file from a distant storage,  Quickwit's metastore stores the byte offsets of this footer to make this read possible.
+When opening a file from remote storage, Quickwit's metastore normally supplies the byte offsets
+of this footer. A reader without metastore metadata can instead use the object size and the final
+16 bytes to locate a footer trailer.
 
-If this footer offset information is not available, for instance if the split is just a file on the filesystem, it is still possible to open it by reading the last 8 bytes of the split (encoding the length of the hotcache), deducing the position of the meta information and unpacking this in turn.
+Legacy splits without a trailer remain self-discoverable by reading the final four-byte hotcache
+length, then walking backward to the four-byte bundle-metadata length.
+
+## Footer trailer rollout
+
+Readers accept splits both with and without the trailer. Writers keep producing the legacy format
+by default because older Quickwit readers interpret the final four bytes as the hotcache length and
+cannot open a split with an appended trailer.
+
+The trailer is enabled for writers with `QW_ENABLE_SPLIT_FOOTER_TRAILER=true`. Roll it out in two
+phases:
+
+1. Upgrade every split reader (searchers, indexers, compactors, and split-inspection tooling) while
+   leaving the environment variable unset.
+2. After every reader is upgraded, set the environment variable on split writers, primarily
+   indexers and compactors.
diff --git a/quickwit/quickwit-cli/src/tool.rs b/quickwit/quickwit-cli/src/tool.rs
index 2dad9520bca..7553ed3d7ce 100644
--- a/quickwit/quickwit-cli/src/tool.rs
+++ b/quickwit/quickwit-cli/src/tool.rs
@@ -813,12 +813,9 @@ async fn extract_split_cli(args: ExtractSplitArgs) -> anyhow::Result<()> {
         .deserialize_index_metadata()?;
     let index_storage = storage_resolver.resolve(index_metadata.index_uri()).await?;
     let split_file = PathBuf::from(format!("{}.split", args.split_id));
-    let split_data = index_storage.get_all(split_file.as_path()).await?;
-    let (_hotcache_bytes, bundle_storage) = BundleStorage::open_from_split_data_with_owned_bytes(
-        index_storage,
-        split_file,
-        split_data,
-    )?;
+    let split_bytes = index_storage.get_all(split_file.as_path()).await?;
+    let (bundle_storage, _hotcache_bytes) =
+        BundleStorage::open_from_split_bytes(index_storage, split_file, split_bytes)?;
     std::fs::create_dir_all(&args.target_dir)?;
     for path in bundle_storage.iter_files() {
         let mut out_path = args.target_dir.to_owned();
diff --git a/quickwit/quickwit-common/src/shared_consts.rs b/quickwit/quickwit-common/src/shared_consts.rs
index 1d72203919b..5cb5601fe4c 100644
--- a/quickwit/quickwit-common/src/shared_consts.rs
+++ b/quickwit/quickwit-common/src/shared_consts.rs
@@ -76,6 +76,9 @@ pub const INGESTER_CAPACITY_SCORE_PREFIX: &str = "ingester.capacity_score:";
 /// File name for the encoded list of fields in the split
 pub const SPLIT_FIELDS_FILE_NAME: &str = "split_fields";
 
+/// Name of the recovery metadata entry embedded in split bundles.
+pub const SPLIT_RECOVERY_METADATA_FILE_NAME: &str = "split_recovery_metadata";
+
 /// More or less the indexing throughput of a core
 /// i.e. PIPELINE_THROUGHPUT / PIPELINE_FULL_CAPACITY
 pub const DEFAULT_SHARD_THROUGHPUT_LIMIT: ByteSize = ByteSize::mib(5);
diff --git a/quickwit/quickwit-directories/src/bundle_directory.rs b/quickwit/quickwit-directories/src/bundle_directory.rs
index 797a55ca146..3990e9a013a 100644
--- a/quickwit/quickwit-directories/src/bundle_directory.rs
+++ b/quickwit/quickwit-directories/src/bundle_directory.rs
@@ -18,23 +18,62 @@ use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::{fmt, io};
 
-use quickwit_storage::{BundleStorageFileOffsets, OwnedBytes, Storage, StorageResult};
+use quickwit_storage::{
+    BundleFileRanges, OwnedBytes, Storage, StorageErrorKind, StorageResult,
+    locate_split_footer_range, strip_split_footer_trailer,
+};
 use tantivy::directory::error::OpenReadError;
 use tantivy::directory::{FileHandle, FileSlice};
 use tantivy::{Directory, HasLen};
 
-/// BundleDirectory is a read-only directory that makes it possible to
-/// open a split and serve the file it contains via tantivy's `Directory`.
+/// `BundleDirectory` is a read-only directory that opens a "split bundle" and serves its files
+/// through Tantivy's [`Directory`] interface.
 ///
 /// It is the `Directory` equivalent of `BundleStorage`.
 ///
-/// Split Format:
-/// `[Files][FilesMetadata][FilesMetadata length 8 byte Little endian][Hotcache][Hotcache length 8
-/// byte Little endian]`
+/// A split has the following layout (all integer fields are little-endian):
+///
+/// ```text
+/// Split
+/// ├── Bundle
+/// │   ├── Files
+/// │   │   ├── Tantivy index files
+/// │   │   ├── split_fields
+/// │   │   └── split_recovery_metadata (optional)
+/// │   └── Bundle footer
+/// │       ├── Bundle metadata (`BundleFileRanges` serialized as versioned JSON)
+/// │       └── Bundle metadata length (u32)
+/// ├── Hotcache
+/// ├── Hotcache length (u32)
+/// └── Split footer trailer (optional)
+///     ├── Split footer start (u64)
+///     ├── Trailer version (u32)
+///     └── Magic bytes (`QWFT`)
+/// ```
+///
+/// In compact form:
+///
+/// ```text
+/// Bundle = [files][bundle metadata][bundle metadata length: 4-byte u32 LE]
+/// Split  = [bundle][hotcache][hotcache length: 4-byte u32 LE][optional trailer]
+/// Trailer = [split footer start: 8-byte u64 LE][version: 4-byte u32 LE][QWFT]
+/// ```
+///
+/// The split footer starts at the bundle footer, so it contains the bundle footer, hotcache, and
+/// hotcache length. The optional trailer points to that start offset but is not part of the split
+/// footer returned by [`read_split_footer`].
 #[derive(Clone)]
 pub struct BundleDirectory {
     file: FileSlice,
-    file_offsets: BundleStorageFileOffsets,
+    file_ranges: BundleFileRanges,
+}
+
+fn read_u32_le(bytes: &[u8]) -> u32 {
+    u32::from_le_bytes(
+        bytes
+            .try_into()
+            .expect("slice should be exactly 4-byte long"),
+    )
 }
 
 impl Debug for BundleDirectory {
@@ -43,47 +82,58 @@ impl Debug for BundleDirectory {
     }
 }
 
-/// Loads the split footer from a storage and path.
+/// Loads the split footer and its nested bundle footer from storage.
+///
+/// The first returned slice is the complete split footer:
 ///
-/// Returns (SplitFooter, BundleFooter)
-/// SplitFooter [BundleMetadata, BundleMetadata Len, Hotcache, Hotcache len]
-/// BundleFooter [BundleMetadata, BundleMetadata Len]
+/// ```text
+/// [bundle metadata (`BundleFileRanges`)][bundle metadata length][hotcache][hotcache length]
+/// ```
+///
+/// The second returned slice is its bundle footer:
+///
+/// ```text
+/// [bundle metadata (`BundleFileRanges`)][bundle metadata length]
+/// ```
 pub async fn read_split_footer(
     storage: Arc<dyn Storage>,
     path: &Path,
 ) -> StorageResult<(OwnedBytes, OwnedBytes)> {
-    let file_len = storage.file_num_bytes(path).await? as usize;
-
-    let hotcache_len_bytes = storage.get_slice(path, file_len - 8..file_len).await?;
-    let hotcache_len = u64::from_le_bytes(hotcache_len_bytes.as_ref().try_into().unwrap()) as usize;
-
-    let second_footer_start = file_len - 8 - hotcache_len - 8;
-    let second_footer_bytes = storage
-        .get_slice(path, second_footer_start..second_footer_start + 8)
+    let split_len = storage.file_num_bytes(path).await?;
+    let footer_range = locate_split_footer_range(storage.as_ref(), path, split_len)
+        .await
+        .map_err(|error| StorageErrorKind::Internal.with_error(error))?;
+    let split_footer_with_trailer = storage
+        .get_slice(path, footer_range.start as usize..footer_range.end as usize)
         .await?;
-    let second_footer_len =
-        u64::from_le_bytes(second_footer_bytes.as_ref().try_into().unwrap()) as usize;
-
-    let split_footer = storage
-        .get_slice(path, second_footer_start - second_footer_len..file_len)
-        .await?;
-    let only_bundle_footer = split_footer.slice(0..second_footer_len + 8);
-
-    Ok((split_footer, only_bundle_footer))
+    let split_footer =
+        strip_split_footer_trailer(FileSlice::new(Arc::new(split_footer_with_trailer)))
+            .and_then(|footer| footer.read_bytes().map_err(anyhow::Error::from))
+            .map_err(|error| StorageErrorKind::Internal.with_error(error))?;
+
+    let hotcache_len = read_u32_le(&split_footer[split_footer.len() - 4..]) as usize;
+    let metadata_len_offset = split_footer.len() - 4 - hotcache_len - 4;
+    let metadata_len =
+        read_u32_le(&split_footer[metadata_len_offset..metadata_len_offset + 4]) as usize;
+    let bundle_footer = split_footer.slice(0..metadata_len + 4);
+
+    Ok((split_footer, bundle_footer))
 }
 
-/// Return two slices for given split: `[body and bundle meta data] [hotcache]`
-fn split_footer(file_slice: FileSlice) -> io::Result<(FileSlice, FileSlice)> {
-    let (body_and_footer_slice, footer_len_slice) = file_slice.split_from_end(4);
+/// Splits a complete split into `[bundle][hotcache]`, discarding the hotcache length and optional
+/// split footer trailer.
+fn split_footer(split_slice: FileSlice) -> io::Result<(FileSlice, FileSlice)> {
+    let split_footer_slice = strip_split_footer_trailer(split_slice).map_err(io::Error::other)?;
+    let (body_and_footer_slice, footer_len_slice) = split_footer_slice.split_from_end(4);
     let footer_len_bytes = footer_len_slice.read_bytes()?;
-    let footer_len = u32::from_le_bytes(footer_len_bytes.as_slice().try_into().unwrap());
+    let footer_len = read_u32_le(footer_len_bytes.as_slice());
     Ok(body_and_footer_slice.split_from_end(footer_len as usize))
 }
 
-/// Return two slices for given split: `[body and bundle meta data] [hotcache]`
-pub fn get_hotcache_from_split(data: OwnedBytes) -> io::Result<OwnedBytes> {
-    let split_file = FileSlice::new(Arc::new(data));
-    let (_, hotcache) = split_footer(split_file)?;
+/// Extracts the hotcache from a complete split.
+pub fn get_hotcache_from_split(split_bytes: OwnedBytes) -> io::Result<OwnedBytes> {
+    let split_slice = FileSlice::new(Arc::new(split_bytes));
+    let (_, hotcache) = split_footer(split_slice)?;
     hotcache.read_bytes()
 }
 
@@ -92,21 +142,21 @@ impl BundleDirectory {
     pub fn get_stats_split(data: OwnedBytes) -> anyhow::Result<Vec<(PathBuf, u64)>> {
         let split_file = FileSlice::new(Arc::new(data));
         let (body_and_bundle_metadata, hot_cache) = split_footer(split_file)?;
-        let file_offsets = BundleStorageFileOffsets::open(body_and_bundle_metadata)?;
+        let file_ranges = BundleFileRanges::open(body_and_bundle_metadata)?;
 
-        let mut files_and_size: Vec<(_, _)> = file_offsets
+        let mut files_and_sizes: Vec<(_, _)> = file_ranges
             .files
             .into_iter()
             .map(|(file, range)| (file, range.end - range.start))
             .collect();
 
-        files_and_size.push((
+        files_and_sizes.push((
             PathBuf::from("hotcache".to_string()),
             hot_cache.len() as u64,
         ));
 
-        files_and_size.sort();
-        Ok(files_and_size)
+        files_and_sizes.sort();
+        Ok(files_and_sizes)
     }
 
     /// Opens a split file.
@@ -118,8 +168,8 @@ impl BundleDirectory {
 
     /// Opens a BundleDirectory, given a file containing the bundle data.
     pub fn open_bundle(file: FileSlice) -> anyhow::Result<BundleDirectory> {
-        let file_offsets = BundleStorageFileOffsets::open(file.clone())?;
-        Ok(BundleDirectory { file, file_offsets })
+        let file_ranges = BundleFileRanges::open(file.clone())?;
+        Ok(BundleDirectory { file, file_ranges })
     }
 }
 
@@ -131,7 +181,7 @@ impl Directory for BundleDirectory {
 
     fn open_read(&self, path: &Path) -> Result<FileSlice, OpenReadError> {
         let byte_range = self
-            .file_offsets
+            .file_ranges
             .get(path)
             .ok_or_else(|| OpenReadError::FileDoesNotExist(path.to_path_buf()))?;
         Ok(self
@@ -148,7 +198,7 @@ impl Directory for BundleDirectory {
     }
 
     fn exists(&self, path: &Path) -> Result<bool, OpenReadError> {
-        Ok(self.file_offsets.exists(path))
+        Ok(self.file_ranges.exists(path))
     }
 
     crate::read_only_directory!();
@@ -179,6 +229,7 @@ mod tests {
         let split_streamer = SplitPayloadBuilder::get_split_payload(
             &[test_filepath1.clone(), test_filepath2.clone()],
             &[],
+            None,
             &[
                 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18,
             ],
@@ -211,6 +262,7 @@ mod tests {
         let split_streamer = SplitPayloadBuilder::get_split_payload(
             &[test_filepath1.clone(), test_filepath2.clone()],
             &[],
+            None,
             &[
                 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18,
             ],
@@ -218,9 +270,9 @@ mod tests {
 
         let buffer = split_streamer.read_all().await?;
 
-        let bundle_file_slice = FileSlice::from(buffer.to_vec());
+        let bundle_slice = FileSlice::from(buffer.to_vec());
 
-        let bundle_dir = BundleDirectory::open_split(bundle_file_slice)?;
+        let bundle_dir = BundleDirectory::open_split(bundle_slice)?;
 
         assert!(bundle_dir.exists(Path::new("f1")).unwrap());
         assert!(bundle_dir.exists(Path::new("f2")).unwrap());
@@ -250,12 +302,15 @@ mod tests {
         let split_streamer = SplitPayloadBuilder::get_split_payload(
             &[test_filepath1.clone(), test_filepath2.clone()],
             &[5, 5, 5],
+            None,
             &[1, 2, 3],
         )?;
+        let footer_start = split_streamer.footer_range.start;
 
-        let data = split_streamer.read_all().await?;
+        let split_bytes = split_streamer.read_all().await?;
+        let split_slice = FileSlice::new(Arc::new(split_bytes));
 
-        let bundle_dir = BundleDirectory::open_split(FileSlice::from(data.to_vec()))?;
+        let bundle_dir = BundleDirectory::open_split(split_slice.clone())?;
 
         let field_data = bundle_dir.atomic_read(Path::new(SPLIT_FIELDS_FILE_NAME))?;
         assert_eq!(&*field_data, &[5, 5, 5]);
@@ -266,6 +321,19 @@ mod tests {
         let f2_data = bundle_dir.atomic_read(Path::new("f2"))?;
         assert_eq!(&f2_data[..], &[99, 55, 44]);
 
+        // Phase-one readers must also accept the trailer format before writers enable it.
+        let split_footer = strip_split_footer_trailer(split_slice)?.read_bytes()?;
+        let mut split_footer_with_trailer = split_footer.to_vec();
+        split_footer_with_trailer.extend_from_slice(&footer_start.to_le_bytes());
+        split_footer_with_trailer.extend_from_slice(&1u32.to_le_bytes());
+        split_footer_with_trailer.extend_from_slice(b"QWFT");
+
+        let bundle_dir_with_trailer =
+            BundleDirectory::open_split(FileSlice::from(split_footer_with_trailer))?;
+        assert_eq!(
+            bundle_dir_with_trailer.atomic_read(Path::new(SPLIT_FIELDS_FILE_NAME))?,
+            [5, 5, 5]
+        );
         Ok(())
     }
 }
diff --git a/quickwit/quickwit-indexing/src/actors/merge_split_downloader.rs b/quickwit/quickwit-indexing/src/actors/merge_split_downloader.rs
index be13b70789a..a48ea7e86c5 100644
--- a/quickwit/quickwit-indexing/src/actors/merge_split_downloader.rs
+++ b/quickwit/quickwit-indexing/src/actors/merge_split_downloader.rs
@@ -154,7 +154,7 @@ mod tests {
         let split_store = {
             let mut storage_builder = RamStorageBuilder::default();
             for split in &splits_to_merge {
-                let buffer = SplitPayloadBuilder::get_split_payload(&[], &[], &[1, 2, 3])?
+                let buffer = SplitPayloadBuilder::get_split_payload(&[], &[], None, &[1, 2, 3])?
                     .read_all()
                     .await?;
                 storage_builder = storage_builder.put(&split_file(split.split_id()), &buffer);
diff --git a/quickwit/quickwit-indexing/src/actors/uploader.rs b/quickwit/quickwit-indexing/src/actors/uploader.rs
index f3b9f85e126..9ee80f35610 100644
--- a/quickwit/quickwit-indexing/src/actors/uploader.rs
+++ b/quickwit/quickwit-indexing/src/actors/uploader.rs
@@ -27,16 +27,18 @@ use quickwit_common::pubsub::EventBroker;
 use quickwit_common::spawn_named_task;
 use quickwit_config::RetentionPolicy;
 use quickwit_metastore::checkpoint::IndexCheckpointDelta;
-use quickwit_metastore::{SplitMetadata, StageSplitsRequestExt};
+use quickwit_metastore::{SplitMaturity, SplitMetadata, StageSplitsRequestExt};
 use quickwit_metrics::{gauge, label_values};
-use quickwit_proto::metastore::{MetastoreService, MetastoreServiceClient, StageSplitsRequest};
+use quickwit_proto::metastore::{
+    MetastoreService, MetastoreServiceClient, SplitRecoveryMetadata, StageSplitsRequest,
+};
 use quickwit_proto::search::{ReportSplit, ReportSplitsRequest};
 use quickwit_proto::types::IndexUid;
-use quickwit_storage::SplitPayloadBuilder;
+use quickwit_storage::{SplitPayload, SplitPayloadBuilder};
 use serde::Serialize;
 use tokio::sync::oneshot::Sender;
 use tokio::sync::{Semaphore, SemaphorePermit, oneshot};
-use tracing::{Instrument, Span, debug, info, instrument, warn};
+use tracing::{Instrument, Span, debug, error, info, instrument, warn};
 
 use crate::actors::Publisher;
 use crate::actors::sequencer::{Sequencer, SequencerCommand};
@@ -129,7 +131,7 @@ impl SplitsUpdateSender {
         if let SplitsUpdateSender::Sequencer(split_uploader_tx) = self
             && split_uploader_tx.send(SequencerCommand::Discard).is_err()
         {
-            bail!("failed to send cancel command to sequencer. the sequencer is probably dead");
+            bail!("failed to send cancel command to sequencer: it is probably dead");
         }
         Ok(())
     }
@@ -304,85 +306,90 @@ impl Handler<PackagedSplitBatch> for Uploader {
                 fail_point!("uploader:intask:before");
 
                 let mut split_metadata_list = Vec::with_capacity(batch.splits.len());
+                let mut split_payloads = Vec::with_capacity(batch.splits.len());
                 let mut report_splits: Vec<ReportSplit> = Vec::with_capacity(batch.splits.len());
 
                 for packaged_split in batch.splits.iter() {
                     if batch.publish_lock.is_dead() {
                         // TODO: Remove the junk right away?
                         info!("splits' publish lock is dead");
-                        if let Err(e) = split_update_sender.discard() {
-                            warn!(cause=?e, "could not discard split");
+
+                        if let Err(error) = split_update_sender.discard() {
+                            error!(?error, "failed to discard split");
                         }
                         return;
                     }
 
-                    let split_streamer = match SplitPayloadBuilder::get_split_payload(
-                        &packaged_split.split_files,
-                        &packaged_split.serialized_split_fields,
-                        &packaged_split.hotcache_bytes,
+                    let (split_metadata, split_payload) = match prepare_split_for_upload(
+                        packaged_split,
+                        &merge_policy,
+                        retention_policy.as_ref(),
                     ) {
-                        Ok(split_streamer) => split_streamer,
-                        Err(e) => {
-                            warn!(cause=?e, split_id=packaged_split.split_id_str(), "could not create split streamer");
+                        Ok(prepared_split) => prepared_split,
+                        Err(error) => {
+                            error!(
+                                ?error,
+                                split_id = packaged_split.split_id(),
+                                "failed to prepare split for upload"
+                            );
                             return;
                         }
                     };
-                    let split_metadata = create_split_metadata(
-                        &merge_policy,
-                        retention_policy.as_ref(),
-                        &packaged_split.split_attrs,
-                        packaged_split.tags.clone(),
-                        split_streamer.footer_range.start..split_streamer.footer_range.end,
-                    );
 
                     report_splits.push(ReportSplit {
                         storage_uri: split_store.remote_uri().to_string(),
-                        split_id: packaged_split.split_id_str().to_string(),
+                        split_id: packaged_split.split_id().to_string(),
                     });
 
                     split_metadata_list.push(split_metadata);
-
+                    split_payloads.push(split_payload);
                 }
 
-                let stage_splits_request = match StageSplitsRequest::try_from_splits_metadata(index_uid.clone(), split_metadata_list.clone()) {
+                let stage_splits_request = match StageSplitsRequest::try_from_splits_metadata(
+                    index_uid.clone(),
+                    split_metadata_list.clone(),
+                ) {
                     Ok(stage_splits_request) => stage_splits_request,
-                    Err(e) => {
-                        warn!(cause=?e, "could not create stage splits request");
+                    Err(error) => {
+                        error!(?error, "failed to create stage splits request");
                         return;
                     }
                 };
-                if let Err(e) = metastore
-                    .clone()
-                    .stage_splits(stage_splits_request)
-                    .await
-                {
-                    warn!(cause=?e, "failed to stage splits");
+                if let Err(error) = metastore.stage_splits(stage_splits_request).await {
+                    error!(?error, "failed to stage splits");
                     return;
                 };
 
-                counters.num_staged_splits.fetch_add(split_metadata_list.len() as u64, Ordering::SeqCst);
+                counters
+                    .num_staged_splits
+                    .fetch_add(split_metadata_list.len() as u64, Ordering::Relaxed);
 
                 let mut packaged_splits_and_metadata = Vec::with_capacity(batch.splits.len());
 
                 event_broker.publish(ReportSplitsRequest { report_splits });
 
-                for (packaged_split, metadata) in batch.splits.into_iter().zip(split_metadata_list) {
+                for ((packaged_split, metadata), split_payload) in batch
+                    .splits
+                    .into_iter()
+                    .zip(split_metadata_list)
+                    .zip(split_payloads)
+                {
                     let upload_result = upload_split(
                         &packaged_split,
                         &metadata,
+                        split_payload,
                         &split_store,
                         counters.clone(),
                     )
                     .await;
 
-                    if let Err(cause) = upload_result {
-                        kill_switch.kill_with_fault(cause.context(format!(
-                            "Uploader failed to upload split {}",
-                            packaged_split.split_id_str()
+                    if let Err(error) = upload_result {
+                        kill_switch.kill_with_fault(error.context(format!(
+                            "failed to upload split `{}`",
+                            packaged_split.split_id()
                         )));
                         return;
                     }
-
                     packaged_splits_and_metadata.push((packaged_split, metadata));
                 }
 
@@ -399,8 +406,8 @@ impl Handler<PackagedSplitBatch> for Uploader {
                     SplitsUpdateSender::Sequencer(_) => "sequencer",
                     SplitsUpdateSender::Publisher(_) => "publisher",
                 };
-                if let Err(e) = split_update_sender.send(splits_update, &ctx_clone).await {
-                    warn!(cause=?e, target, "failed to send uploaded split");
+                if let Err(error) = split_update_sender.send(splits_update, &ctx_clone).await {
+                    error!(?error, "failed to send splits update to {target}");
                     return;
                 }
                 // We explicitly drop it in order to force move the permit guard into the async
@@ -408,13 +415,88 @@ impl Handler<PackagedSplitBatch> for Uploader {
                 mem::drop(permit_guard);
             }
             .instrument(Span::current()),
-            "upload_single_task"
+            "upload_single_task",
         );
         fail_point!("uploader:intask:after");
         Ok(())
     }
 }
 
+fn create_split_recovery_metadata(
+    split_metadata: &SplitMetadata,
+    parent_split_ids: &[quickwit_proto::types::SplitId],
+) -> SplitRecoveryMetadata {
+    let time_range = split_metadata.time_range.as_ref();
+    let time_range_start_inclusive = time_range.map(|range| *range.start());
+    let time_range_end_inclusive = time_range.map(|range| *range.end());
+
+    let parent_split_ids = parent_split_ids
+        .iter()
+        .map(|split_id| split_id.to_string())
+        .collect();
+
+    let maturation_period_millis = match split_metadata.maturity {
+        SplitMaturity::Mature => None,
+        SplitMaturity::Immature { maturation_period } => Some(
+            maturation_period
+                .as_millis()
+                .try_into()
+                .expect("maturation period should fit in u64 milliseconds"),
+        ),
+    };
+    SplitRecoveryMetadata {
+        split_id: split_metadata.split_id.to_string(),
+        index_uid: Some(split_metadata.index_uid.clone()),
+        source_id: split_metadata.source_id.clone(),
+        node_id: split_metadata.node_id.clone(),
+        doc_mapping_uid: Some(split_metadata.doc_mapping_uid),
+        partition_id: split_metadata.partition_id,
+        num_docs: split_metadata.num_docs as u64,
+        uncompressed_docs_size_bytes: split_metadata.uncompressed_docs_size_in_bytes,
+        time_range_start_inclusive,
+        time_range_end_inclusive,
+        create_timestamp: split_metadata.create_timestamp,
+        tags: split_metadata.tags.iter().cloned().collect(),
+        delete_opstamp: split_metadata.delete_opstamp,
+        num_merge_ops: split_metadata.num_merge_ops as u64,
+        parent_split_ids,
+        maturation_period_millis,
+    }
+}
+
+fn prepare_split_for_upload(
+    packaged_split: &PackagedSplit,
+    merge_policy: &Arc<dyn MergePolicy>,
+    retention_policy: Option<&RetentionPolicy>,
+) -> anyhow::Result<(SplitMetadata, SplitPayload)> {
+    // Footer offsets are unknown at this point, so we use default values.
+    let footer_offsets = Default::default();
+
+    let split_metadata = create_split_metadata(
+        merge_policy,
+        retention_policy,
+        &packaged_split.split_attrs,
+        packaged_split.tags.clone(),
+        footer_offsets,
+    );
+    let recovery_metadata = create_split_recovery_metadata(
+        &split_metadata,
+        &packaged_split.split_attrs.replaced_split_ids,
+    );
+    let serialized_recovery_metadata = recovery_metadata.serialize();
+    let split_payload = SplitPayloadBuilder::get_split_payload(
+        &packaged_split.split_files,
+        &packaged_split.serialized_split_fields,
+        Some(&serialized_recovery_metadata),
+        &packaged_split.hotcache_bytes,
+    )?;
+    let split_metadata = SplitMetadata {
+        footer_offsets: split_payload.footer_range.clone(),
+        ..split_metadata
+    };
+    Ok((split_metadata, split_payload))
+}
+
 #[async_trait]
 impl Handler<EmptySplit> for Uploader {
     type Reply = ();
@@ -484,20 +566,15 @@ fn make_publish_operation(
 async fn upload_split(
     packaged_split: &PackagedSplit,
     split_metadata: &SplitMetadata,
+    split_payload: SplitPayload,
     split_store: &IndexingSplitStore,
     counters: UploaderCounters,
 ) -> anyhow::Result<()> {
-    let split_streamer = SplitPayloadBuilder::get_split_payload(
-        &packaged_split.split_files,
-        &packaged_split.serialized_split_fields,
-        &packaged_split.hotcache_bytes,
-    )?;
-
     split_store
         .store_split(
             split_metadata,
             packaged_split.split_scratch_directory.path(),
-            Box::new(split_streamer),
+            Box::new(split_payload),
         )
         .await?;
     counters.num_uploaded_splits.fetch_add(1, Ordering::SeqCst);
@@ -523,6 +600,30 @@ mod tests {
     use crate::merge_policy::{NopMergePolicy, default_merge_policy};
     use crate::models::{SplitAttrs, SplitsUpdate};
 
+    #[test]
+    fn test_split_recovery_metadata_preserves_maturity() {
+        let split_metadata = SplitMetadata {
+            maturity: SplitMaturity::Immature {
+                maturation_period: Duration::from_millis(1_500),
+            },
+            ..Default::default()
+        };
+
+        let recovery_metadata = create_split_recovery_metadata(&split_metadata, &[]);
+        assert_eq!(recovery_metadata.maturation_period_millis, Some(1_500));
+
+        let (recovered_metadata, _parent_split_ids) =
+            SplitMetadata::try_from_recovery_metadata(recovery_metadata, 1..2).unwrap();
+        assert_eq!(recovered_metadata.maturity, split_metadata.maturity);
+
+        let mature_split_metadata = SplitMetadata::default();
+        let mature_recovery_metadata = create_split_recovery_metadata(&mature_split_metadata, &[]);
+        assert_eq!(mature_recovery_metadata.maturation_period_millis, None);
+        let (recovered_mature_metadata, _parent_split_ids) =
+            SplitMetadata::try_from_recovery_metadata(mature_recovery_metadata, 1..2).unwrap();
+        assert_eq!(recovered_mature_metadata.maturity, SplitMaturity::Mature);
+    }
+
     #[tokio::test]
     async fn test_uploader_with_sequencer() -> anyhow::Result<()> {
         quickwit_common::setup_logging_for_tests();
diff --git a/quickwit/quickwit-indexing/src/lib.rs b/quickwit/quickwit-indexing/src/lib.rs
index cf57643117f..d6351867094 100644
--- a/quickwit/quickwit-indexing/src/lib.rs
+++ b/quickwit/quickwit-indexing/src/lib.rs
@@ -46,6 +46,8 @@ pub mod merge_policy;
 mod metrics;
 pub mod models;
 pub mod source;
+#[cfg(test)]
+mod split_recovery_tests;
 mod split_store;
 #[cfg(any(test, feature = "testsuite"))]
 mod test_utils;
diff --git a/quickwit/quickwit-indexing/src/models/packaged_split.rs b/quickwit/quickwit-indexing/src/models/packaged_split.rs
index 92c0c14fdd6..27c078ed182 100644
--- a/quickwit/quickwit-indexing/src/models/packaged_split.rs
+++ b/quickwit/quickwit-indexing/src/models/packaged_split.rs
@@ -39,7 +39,7 @@ impl PackagedSplit {
         &self.split_attrs.index_uid
     }
 
-    pub fn split_id_str(&self) -> &str {
+    pub fn split_id(&self) -> &str {
         self.split_attrs.split_id.as_str()
     }
 }
diff --git a/quickwit/quickwit-indexing/src/split_recovery_tests.rs b/quickwit/quickwit-indexing/src/split_recovery_tests.rs
new file mode 100644
index 00000000000..15854c475a1
--- /dev/null
+++ b/quickwit/quickwit-indexing/src/split_recovery_tests.rs
@@ -0,0 +1,258 @@
+// Copyright 2021-Present Datadog, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::path::Path;
+use std::sync::Arc;
+use std::time::Duration;
+
+use quickwit_common::shared_consts::SPLIT_RECOVERY_METADATA_FILE_NAME;
+use quickwit_common::test_utils::wait_until_predicate;
+use quickwit_config::IndexConfig;
+use quickwit_metastore::{
+    CreateIndexRequestExt, FileBackedMetastore, ListSplitsQuery, ListSplitsRequestExt,
+    MetastoreServiceStreamSplitsExt, SplitMetadata, SplitState, StageSplitsRequestExt,
+};
+use quickwit_proto::metastore::{
+    CreateIndexRequest, ListSplitsRequest, MetastoreService, PublishSplitsRequest,
+    SplitRecoveryMetadata, StageSplitsRequest,
+};
+use quickwit_proto::types::SplitId;
+use quickwit_storage::{BundleStorage, RamStorage, Storage};
+
+use crate::test_utils::TestSandbox;
+
+async fn recover_and_publish_split(
+    test_sandbox: &TestSandbox,
+    index_id: &str,
+    original_metadata: &SplitMetadata,
+) -> anyhow::Result<Vec<SplitId>> {
+    // Load the split bundle from object storage and read its embedded recovery entry.
+    let split_filename = quickwit_common::split_file(original_metadata.split_id());
+    let split_path = Path::new(&split_filename);
+    let storage = test_sandbox.storage();
+    let (split_bundle, _hotcache, footer_offsets) =
+        BundleStorage::open_from_storage(storage, split_path.to_path_buf()).await?;
+    let serialized_recovery_metadata = split_bundle
+        .get_all(Path::new(SPLIT_RECOVERY_METADATA_FILE_NAME))
+        .await?;
+    let recovery_metadata =
+        SplitRecoveryMetadata::deserialize(serialized_recovery_metadata.as_ref())?;
+    let (mut recovered_metadata, parent_split_ids) =
+        SplitMetadata::try_from_recovery_metadata(recovery_metadata, footer_offsets)?;
+
+    assert_eq!(&recovered_metadata, original_metadata);
+
+    // Simulate a lost metastore by creating the index in a new one. Index creation assigns a
+    // new incarnation UID, so the importer explicitly remaps the recovered split to it.
+    let fresh_metastore =
+        FileBackedMetastore::try_new(Arc::new(RamStorage::default()), None).await?;
+    let index_config = IndexConfig::for_test(index_id, "ram:///recovered-index");
+    let recovered_index_uid = fresh_metastore
+        .create_index(CreateIndexRequest::try_from_index_config(&index_config)?)
+        .await?
+        .index_uid()
+        .clone();
+    recovered_metadata.index_uid = recovered_index_uid.clone();
+
+    fresh_metastore
+        .stage_splits(StageSplitsRequest::try_from_split_metadata(
+            recovered_index_uid.clone(),
+            &recovered_metadata,
+        )?)
+        .await?;
+    fresh_metastore
+        .publish_splits(PublishSplitsRequest {
+            index_uid: Some(recovered_index_uid.clone()),
+            staged_split_ids: vec![recovered_metadata.split_id.to_string()],
+            ..Default::default()
+        })
+        .await?;
+
+    let published_splits = fresh_metastore
+        .list_splits(ListSplitsRequest::try_from_index_uid(recovered_index_uid)?)
+        .await?
+        .collect_splits()
+        .await?;
+    assert_eq!(published_splits.len(), 1);
+    assert_eq!(published_splits[0].split_state, SplitState::Published);
+    assert_eq!(published_splits[0].split_metadata, recovered_metadata);
+
+    Ok(parent_split_ids)
+}
+#[tokio::test]
+async fn test_recover_split_from_bundle_and_publish_it() {
+    let index_id = quickwit_common::rand::append_random_suffix("split-recovery");
+    let test_sandbox = TestSandbox::create(
+        &index_id,
+        r#"
+            timestamp_field: timestamp
+            tag_fields:
+              - tenant
+            field_mappings:
+              - name: timestamp
+                type: datetime
+                fast: true
+              - name: body
+                type: text
+              - name: tenant
+                type: text
+                tokenizer: raw
+        "#,
+        "{}",
+        &["body"],
+    )
+    .await
+    .unwrap();
+    test_sandbox
+        .add_documents([serde_json::json!({
+            "timestamp": 1_700_000_000,
+            "body": "recover me",
+            "tenant": "acme"
+        })])
+        .await
+        .unwrap();
+
+    let original_splits = test_sandbox
+        .metastore()
+        .list_splits(ListSplitsRequest::try_from_index_uid(test_sandbox.index_uid()).unwrap())
+        .await
+        .unwrap()
+        .collect_splits()
+        .await
+        .unwrap();
+    assert_eq!(original_splits.len(), 1);
+    let original_metadata = &original_splits[0].split_metadata;
+    assert_eq!(
+        original_metadata
+            .tags
+            .iter()
+            .map(String::as_str)
+            .collect::<Vec<_>>(),
+        ["tenant!", "tenant:acme"]
+    );
+
+    let parent_split_ids = recover_and_publish_split(&test_sandbox, &index_id, original_metadata)
+        .await
+        .unwrap();
+    assert!(parent_split_ids.is_empty());
+
+    test_sandbox.assert_quit().await;
+}
+
+#[tokio::test]
+async fn test_recover_merged_split_from_bundle_and_publish_it() {
+    let index_id = quickwit_common::rand::append_random_suffix("merged-split-recovery");
+    let test_sandbox = TestSandbox::create(
+        &index_id,
+        r#"
+            timestamp_field: timestamp
+            field_mappings:
+              - name: timestamp
+                type: datetime
+                fast: true
+              - name: body
+                type: text
+        "#,
+        r#"
+            split_num_docs_target: 1000
+            merge_policy:
+              type: stable_log
+              merge_factor: 2
+              max_merge_factor: 2
+        "#,
+        &["body"],
+    )
+    .await
+    .unwrap();
+    test_sandbox
+        .add_documents([serde_json::json!({
+            "timestamp": 1_700_000_000,
+            "body": "first parent"
+        })])
+        .await
+        .unwrap();
+    test_sandbox
+        .add_documents([serde_json::json!({
+            "timestamp": 1_700_000_001,
+            "body": "second parent"
+        })])
+        .await
+        .unwrap();
+
+    wait_until_predicate(
+        || {
+            let metastore = test_sandbox.metastore();
+            let index_uid = test_sandbox.index_uid();
+            async move {
+                let query =
+                    ListSplitsQuery::for_index(index_uid).with_split_state(SplitState::Published);
+                let Ok(request) = ListSplitsRequest::try_from_list_splits_query(&query) else {
+                    return false;
+                };
+                let Ok(split_stream) = metastore.list_splits(request).await else {
+                    return false;
+                };
+                let Ok(splits) = split_stream.collect_splits().await else {
+                    return false;
+                };
+                splits.len() == 1 && splits[0].split_metadata.num_merge_ops == 1
+            }
+        },
+        Duration::from_secs(10),
+        Duration::from_millis(25),
+    )
+    .await
+    .unwrap();
+
+    let published_query = ListSplitsQuery::for_index(test_sandbox.index_uid())
+        .with_split_state(SplitState::Published);
+    let published_splits = test_sandbox
+        .metastore()
+        .list_splits(ListSplitsRequest::try_from_list_splits_query(&published_query).unwrap())
+        .await
+        .unwrap()
+        .collect_splits()
+        .await
+        .unwrap();
+    assert_eq!(published_splits.len(), 1);
+    let merged_metadata = &published_splits[0].split_metadata;
+    assert_eq!(merged_metadata.num_docs, 2);
+    assert_eq!(merged_metadata.num_merge_ops, 1);
+
+    let replaced_query = ListSplitsQuery::for_index(test_sandbox.index_uid())
+        .with_split_state(SplitState::MarkedForDeletion);
+    let replaced_splits = test_sandbox
+        .metastore()
+        .list_splits(ListSplitsRequest::try_from_list_splits_query(&replaced_query).unwrap())
+        .await
+        .unwrap()
+        .collect_splits()
+        .await
+        .unwrap();
+    assert_eq!(replaced_splits.len(), 2);
+    let mut expected_parent_ids: Vec<SplitId> = replaced_splits
+        .iter()
+        .map(|split| split.split_metadata.split_id.clone())
+        .collect();
+    expected_parent_ids.sort();
+
+    let mut recovered_parent_ids =
+        recover_and_publish_split(&test_sandbox, &index_id, merged_metadata)
+            .await
+            .unwrap();
+    recovered_parent_ids.sort();
+    assert_eq!(recovered_parent_ids, expected_parent_ids);
+
+    test_sandbox.assert_quit().await;
+}
diff --git a/quickwit/quickwit-indexing/src/split_store/indexing_split_cache.rs b/quickwit/quickwit-indexing/src/split_store/indexing_split_cache.rs
index 53700ffbbf6..f5c755b113b 100644
--- a/quickwit/quickwit-indexing/src/split_store/indexing_split_cache.rs
+++ b/quickwit/quickwit-indexing/src/split_store/indexing_split_cache.rs
@@ -812,6 +812,7 @@ mod tests {
         let split_streamer = SplitPayloadBuilder::get_split_payload(
             &[test_filepath1, test_filepath2],
             &[],
+            None,
             b"hotcache",
         )
         .unwrap();
diff --git a/quickwit/quickwit-indexing/src/split_store/indexing_split_store.rs b/quickwit/quickwit-indexing/src/split_store/indexing_split_store.rs
index b43009682ae..3b34439c2b6 100644
--- a/quickwit/quickwit-indexing/src/split_store/indexing_split_store.rs
+++ b/quickwit/quickwit-indexing/src/split_store/indexing_split_store.rs
@@ -349,9 +349,9 @@ mod tests {
         let split_store = IndexingSplitStore::new(remote_storage, Arc::new(split_cache));
 
         let split_id1 = SplitId::new();
-        let split_payload1 = SplitPayloadBuilder::get_split_payload(&[], &[], &[5, 5, 5])?;
+        let split_payload1 = SplitPayloadBuilder::get_split_payload(&[], &[], None, &[5, 5, 5])?;
         let split_id2 = SplitId::new();
-        let split_payload2 = SplitPayloadBuilder::get_split_payload(&[], &[], &[5, 5, 5, 5])?;
+        let split_payload2 = SplitPayloadBuilder::get_split_payload(&[], &[], None, &[5, 5, 5, 5])?;
 
         {
             let split_path = temp_dir.path().join(split_id1.as_str());
diff --git a/quickwit/quickwit-metastore/src/split_metadata.rs b/quickwit/quickwit-metastore/src/split_metadata.rs
index 29b00ed4032..918ad5a28f8 100644
--- a/quickwit/quickwit-metastore/src/split_metadata.rs
+++ b/quickwit/quickwit-metastore/src/split_metadata.rs
@@ -19,7 +19,9 @@ use std::path::PathBuf;
 use std::str::FromStr;
 use std::time::Duration;
 
+use anyhow::{bail, ensure};
 use bytesize::ByteSize;
+use quickwit_proto::metastore::SplitRecoveryMetadata;
 use quickwit_proto::types::{DocMappingUid, IndexUid, SourceId, SplitId};
 use serde::{Deserialize, Serialize};
 use serde_with::{DurationMilliSeconds, serde_as};
@@ -181,6 +183,70 @@ impl fmt::Debug for SplitMetadata {
 }
 
 impl SplitMetadata {
+    /// Reconstructs metastore split metadata and its direct parents from metadata embedded in a
+    /// split bundle.
+    pub fn try_from_recovery_metadata(
+        recovery_metadata: SplitRecoveryMetadata,
+        footer_offsets: Range<u64>,
+    ) -> anyhow::Result<(Self, Vec<SplitId>)> {
+        let SplitRecoveryMetadata {
+            split_id,
+            index_uid,
+            source_id,
+            node_id,
+            doc_mapping_uid,
+            partition_id,
+            num_docs,
+            uncompressed_docs_size_bytes,
+            time_range_start_inclusive,
+            time_range_end_inclusive,
+            create_timestamp,
+            tags,
+            delete_opstamp,
+            num_merge_ops,
+            parent_split_ids,
+            maturation_period_millis,
+        } = recovery_metadata;
+        let time_range = match (time_range_start_inclusive, time_range_end_inclusive) {
+            (Some(start), Some(end)) if start <= end => Some(start..=end),
+            (None, None) => None,
+            (Some(start), Some(end)) => {
+                bail!("invalid recovery time range: start {start} is after end {end}")
+            }
+            _ => bail!("recovery time range must contain both start and end"),
+        };
+        ensure!(
+            !footer_offsets.is_empty(),
+            "invalid recovery footer offsets"
+        );
+        let maturity = match maturation_period_millis {
+            Some(maturation_period_millis) => SplitMaturity::Immature {
+                maturation_period: Duration::from_millis(maturation_period_millis),
+            },
+            None => SplitMaturity::Mature,
+        };
+        let split_metadata = Self {
+            split_id: split_id.into(),
+            index_uid: index_uid.ok_or_else(|| anyhow::anyhow!("missing recovery index UID"))?,
+            partition_id,
+            source_id,
+            node_id,
+            num_docs: num_docs.try_into()?,
+            uncompressed_docs_size_in_bytes: uncompressed_docs_size_bytes,
+            time_range,
+            create_timestamp,
+            maturity,
+            tags: tags.into_iter().collect(),
+            footer_offsets,
+            delete_opstamp,
+            num_merge_ops: num_merge_ops.try_into()?,
+            doc_mapping_uid: doc_mapping_uid
+                .ok_or_else(|| anyhow::anyhow!("missing recovery doc mapping UID"))?,
+        };
+        let parent_split_ids = parent_split_ids.into_iter().map(SplitId::from).collect();
+        Ok((split_metadata, parent_split_ids))
+    }
+
     /// Creates a new instance of split metadata.
     pub fn new(
         split_id: SplitId,
diff --git a/quickwit/quickwit-proto/protos/quickwit/metastore.proto b/quickwit/quickwit-proto/protos/quickwit/metastore.proto
index df49edb3989..2a8c9b40347 100644
--- a/quickwit/quickwit-proto/protos/quickwit/metastore.proto
+++ b/quickwit/quickwit-proto/protos/quickwit/metastore.proto
@@ -39,6 +39,31 @@ enum SourceType {
   SOURCE_TYPE_STDIN = 13;
 }
 
+// Immutable metadata embedded directly in a split so that the metastore split metadata can be
+// reconstructed if the metastore database is lost.
+message SplitRecoveryMetadata {
+  string split_id = 1;
+  quickwit.common.IndexUid index_uid = 2;
+  string source_id = 3;
+  string node_id = 4;
+  quickwit.common.DocMappingUid doc_mapping_uid = 5;
+  uint64 partition_id = 6;
+  uint64 num_docs = 7;
+  uint64 uncompressed_docs_size_bytes = 8;
+  optional int64 time_range_start_inclusive = 9;
+  optional int64 time_range_end_inclusive = 10;
+  int64 create_timestamp = 11;
+  repeated string tags = 12;
+  uint64 delete_opstamp = 13;
+  uint64 num_merge_ops = 14;
+
+  // Split IDs directly replaced when this split was published.
+  repeated string parent_split_ids = 15;
+
+  // None means the split is mature; otherwise, this is its maturation period in milliseconds.
+  optional uint64 maturation_period_millis = 16;
+}
+
 // Metastore meant to manage Quickwit's indexes, their splits and delete tasks.
 //
 // I. Index and splits management.
diff --git a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.metastore.rs b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.metastore.rs
index 3d281f6b7d2..9a44e8a166a 100644
--- a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.metastore.rs
+++ b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.metastore.rs
@@ -1,4 +1,44 @@
 // This file is @generated by prost-build.
+/// Immutable metadata embedded directly in a split so that the metastore split metadata can be
+/// reconstructed if the metastore database is lost.
+#[derive(serde::Serialize, serde::Deserialize, utoipa::ToSchema)]
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct SplitRecoveryMetadata {
+    #[prost(string, tag = "1")]
+    pub split_id: ::prost::alloc::string::String,
+    #[prost(message, optional, tag = "2")]
+    pub index_uid: ::core::option::Option<crate::types::IndexUid>,
+    #[prost(string, tag = "3")]
+    pub source_id: ::prost::alloc::string::String,
+    #[prost(string, tag = "4")]
+    pub node_id: ::prost::alloc::string::String,
+    #[prost(message, optional, tag = "5")]
+    pub doc_mapping_uid: ::core::option::Option<crate::types::DocMappingUid>,
+    #[prost(uint64, tag = "6")]
+    pub partition_id: u64,
+    #[prost(uint64, tag = "7")]
+    pub num_docs: u64,
+    #[prost(uint64, tag = "8")]
+    pub uncompressed_docs_size_bytes: u64,
+    #[prost(int64, optional, tag = "9")]
+    pub time_range_start_inclusive: ::core::option::Option<i64>,
+    #[prost(int64, optional, tag = "10")]
+    pub time_range_end_inclusive: ::core::option::Option<i64>,
+    #[prost(int64, tag = "11")]
+    pub create_timestamp: i64,
+    #[prost(string, repeated, tag = "12")]
+    pub tags: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
+    #[prost(uint64, tag = "13")]
+    pub delete_opstamp: u64,
+    #[prost(uint64, tag = "14")]
+    pub num_merge_ops: u64,
+    /// Split IDs directly replaced when this split was published.
+    #[prost(string, repeated, tag = "15")]
+    pub parent_split_ids: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
+    /// None means the split is mature; otherwise, this is its maturation period in milliseconds.
+    #[prost(uint64, optional, tag = "16")]
+    pub maturation_period_millis: ::core::option::Option<u64>,
+}
 #[derive(serde::Serialize, serde::Deserialize, utoipa::ToSchema)]
 #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct EmptyResponse {}
diff --git a/quickwit/quickwit-proto/src/metastore/mod.rs b/quickwit/quickwit-proto/src/metastore/mod.rs
index 41dfe7c78f2..34a23537520 100644
--- a/quickwit/quickwit-proto/src/metastore/mod.rs
+++ b/quickwit/quickwit-proto/src/metastore/mod.rs
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::fmt;
+use std::{fmt, io};
 
 use quickwit_common::rate_limited_error;
 use quickwit_common::retry::Retryable;
@@ -31,6 +31,91 @@ pub const METASTORE_FILE_DESCRIPTOR_SET: &[u8] =
 
 pub type MetastoreResult<T> = Result<T, MetastoreError>;
 
+const SPLIT_RECOVERY_METADATA_MAGIC: &[u8; 4] = b"QWSM";
+const SPLIT_RECOVERY_METADATA_FORMAT_VERSION: u8 = 1;
+const SPLIT_RECOVERY_METADATA_HEADER_LEN: usize = SPLIT_RECOVERY_METADATA_MAGIC.len()
+    + std::mem::size_of_val(&SPLIT_RECOVERY_METADATA_FORMAT_VERSION);
+
+impl SplitRecoveryMetadata {
+    /// Serializes the recovery metadata with a small container header followed by protobuf bytes.
+    /// The container version only covers framing; compatible protobuf fields can be added without
+    /// changing it.
+    pub fn serialize(&self) -> Vec<u8> {
+        use prost::Message;
+
+        let mut output =
+            Vec::with_capacity(SPLIT_RECOVERY_METADATA_HEADER_LEN + self.encoded_len());
+        output.extend_from_slice(SPLIT_RECOVERY_METADATA_MAGIC);
+        output.push(SPLIT_RECOVERY_METADATA_FORMAT_VERSION);
+        self.encode(&mut output)
+            .expect("encoding a protobuf into a Vec should not fail");
+        output
+    }
+
+    /// Deserializes split recovery metadata embedded in a split bundle.
+    pub fn deserialize(mut bytes: &[u8]) -> io::Result<Self> {
+        use prost::Message;
+
+        if bytes.len() < SPLIT_RECOVERY_METADATA_HEADER_LEN
+            || &bytes[..SPLIT_RECOVERY_METADATA_MAGIC.len()] != SPLIT_RECOVERY_METADATA_MAGIC
+        {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "invalid split recovery metadata magic number",
+            ));
+        }
+        let version = bytes[SPLIT_RECOVERY_METADATA_MAGIC.len()];
+        if version != SPLIT_RECOVERY_METADATA_FORMAT_VERSION {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                format!("unsupported split recovery metadata format version: {version}"),
+            ));
+        }
+        bytes = &bytes[SPLIT_RECOVERY_METADATA_HEADER_LEN..];
+        Self::decode(bytes).map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))
+    }
+}
+
+#[cfg(test)]
+mod split_recovery_metadata_tests {
+    use super::SplitRecoveryMetadata;
+    use crate::types::{DocMappingUid, IndexUid};
+
+    #[test]
+    fn test_split_recovery_metadata_roundtrip_and_unknown_fields() {
+        let metadata = SplitRecoveryMetadata {
+            split_id: "split-a".to_string(),
+            index_uid: Some(IndexUid::for_test("index-a", 1)),
+            source_id: "source-a".to_string(),
+            node_id: "node-a".to_string(),
+            doc_mapping_uid: Some(DocMappingUid::for_test(2)),
+            partition_id: 3,
+            num_docs: 4,
+            uncompressed_docs_size_bytes: 5,
+            time_range_start_inclusive: Some(10),
+            time_range_end_inclusive: Some(20),
+            create_timestamp: 30,
+            tags: vec!["tenant!".to_string(), "tenant:acme".to_string()],
+            delete_opstamp: 6,
+            num_merge_ops: 7,
+            parent_split_ids: vec!["parent-a".to_string(), "parent-b".to_string()],
+            maturation_period_millis: Some(40_500),
+        };
+        let mut serialized = metadata.serialize();
+        // An unknown protobuf varint field must be ignored by older readers.
+        serialized.extend_from_slice(&[0x98, 0x06, 0x01]);
+
+        let decoded = SplitRecoveryMetadata::deserialize(&serialized).unwrap();
+        assert_eq!(decoded, metadata);
+    }
+
+    #[test]
+    fn test_split_recovery_metadata_rejects_invalid_container_header() {
+        let error = SplitRecoveryMetadata::deserialize(b"not-a-manifest").unwrap_err();
+        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
+    }
+}
+
 /// Lists the object types stored and managed by the metastore.
 #[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
diff --git a/quickwit/quickwit-search/src/leaf.rs b/quickwit/quickwit-search/src/leaf.rs
index 99ff06c312b..043da090a27 100644
--- a/quickwit/quickwit-search/src/leaf.rs
+++ b/quickwit/quickwit-search/src/leaf.rs
@@ -50,7 +50,6 @@ use quickwit_storage::{
 use tantivy::aggregation::AggContextParams;
 use tantivy::aggregation::agg_req::{AggregationVariants, Aggregations};
 use tantivy::collector::Collector;
-use tantivy::directory::FileSlice;
 use tantivy::fastfield::FastFieldReaders;
 use tantivy::index::SegmentId;
 use tantivy::schema::Field;
@@ -166,7 +165,7 @@ pub(crate) async fn open_split_bundle(
     searcher_context: &SearcherContext,
     index_storage: Arc<dyn Storage>,
     split_and_footer_offsets: &SplitIdAndFooterOffsets,
-) -> anyhow::Result<(FileSlice, BundleStorage)> {
+) -> anyhow::Result<(OwnedBytes, BundleStorage)> {
     let split_file = PathBuf::from(format!("{}.split", split_and_footer_offsets.split_id));
     let footer_data = get_split_footer_from_cache_or_fetch(
         index_storage.clone(),
@@ -184,10 +183,10 @@ pub(crate) async fn open_split_bundle(
             index_storage.clone()
         };
 
-    let (hotcache_bytes, bundle_storage) = BundleStorage::open_from_split_data(
+    let (bundle_storage, hotcache_bytes) = BundleStorage::open_from_split_bytes(
         index_storage_with_split_cache,
         split_file,
-        FileSlice::new(Arc::new(footer_data)),
+        footer_data,
     )?;
 
     Ok((hotcache_bytes, bundle_storage))
@@ -242,9 +241,9 @@ pub(crate) async fn open_index_with_caches(
 
     let hot_directory = if let Some(cache) = ephemeral_unbounded_cache {
         let caching_directory = CachingDirectory::new(Arc::new(directory), cache);
-        HotDirectory::open(caching_directory, hotcache_bytes.read_bytes()?)?
+        HotDirectory::open(caching_directory, hotcache_bytes)?
     } else {
-        HotDirectory::open(directory, hotcache_bytes.read_bytes()?)?
+        HotDirectory::open(directory, hotcache_bytes)?
     };
 
     let mut index = Index::open(hot_directory.clone())?;
diff --git a/quickwit/quickwit-storage/src/bundle_storage.rs b/quickwit/quickwit-storage/src/bundle_storage.rs
index ea073c830ba..99a17d47a3f 100644
--- a/quickwit/quickwit-storage/src/bundle_storage.rs
+++ b/quickwit/quickwit-storage/src/bundle_storage.rs
@@ -20,8 +20,9 @@ use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::{fmt, io};
 
-use anyhow::Context;
+use anyhow::{Context, bail, ensure};
 use async_trait::async_trait;
+use bytes::{Buf, BufMut};
 use quickwit_common::chunk_range;
 use quickwit_common::uri::Uri;
 use serde::{Deserialize, Serialize};
@@ -36,74 +37,180 @@ use crate::{
 };
 
 /// BundleStorage bundles together multiple files into a single file.
-/// with some metadata
 pub struct BundleStorage {
     storage: Arc<dyn Storage>,
     /// The file path of the bundle in the storage.
     bundle_filepath: PathBuf,
-    metadata: BundleStorageFileOffsets,
+    file_ranges: BundleFileRanges,
 }
 
 impl BundleStorage {
-    /// Opens a BundleStorage.
-    ///
-    /// The provided data must include the footer_bytes at the end of the slice, but it can have
-    /// more up front.
+    /// Opens a split bundle by locating its footer directly from object storage.
     ///
-    /// Returns (Hotcache, Self)
-    pub fn open_from_split_data_with_owned_bytes(
+    /// New splits expose the footer start in a fixed-size trailer. Legacy splits are supported by
+    /// walking backward through the hotcache and bundle-metadata length fields.
+    pub async fn open_from_storage(
         storage: Arc<dyn Storage>,
         bundle_filepath: PathBuf,
-        split_data: OwnedBytes,
-    ) -> anyhow::Result<(FileSlice, Self)> {
-        Self::open_from_split_data(
-            storage,
-            bundle_filepath,
-            FileSlice::new(Arc::new(split_data)),
-        )
+    ) -> anyhow::Result<(Self, OwnedBytes, Range<u64>)> {
+        let split_len = storage.file_num_bytes(&bundle_filepath).await?;
+        let split_footer_range =
+            locate_split_footer_range(storage.as_ref(), &bundle_filepath, split_len).await?;
+        let split_footer_start = usize::try_from(split_footer_range.start)?;
+        let split_footer_end = usize::try_from(split_footer_range.end)?;
+        let split_footer_range_usize = split_footer_start..split_footer_end;
+        let split_footer_bytes = storage
+            .get_slice(&bundle_filepath, split_footer_range_usize)
+            .await?;
+        let (bundle_storage, hotcache) =
+            Self::open_from_split_bytes(storage, bundle_filepath, split_footer_bytes)?;
+        Ok((bundle_storage, hotcache, split_footer_range))
     }
+
     /// Opens a BundleStorage.
     ///
-    /// The provided data must include the footer_bytes at the end of the slice, but it can have
+    /// The provided bytes must include the footer bytes at the end of the slice, but they can have
     /// more up front.
     ///
-    /// Returns (Hotcache, Self)
-    pub fn open_from_split_data(
+    /// Returns (Self, Hotcache)
+    pub fn open_from_split_bytes(
         storage: Arc<dyn Storage>,
         bundle_filepath: PathBuf,
-        split_data: FileSlice,
-    ) -> anyhow::Result<(FileSlice, Self)> {
-        let (hotcache, metadata) = BundleStorageFileOffsets::open_from_split_data(split_data)?;
+        split_bytes: OwnedBytes,
+    ) -> anyhow::Result<(Self, OwnedBytes)> {
+        let (file_ranges, hotcache) = BundleFileRanges::open_from_split_bytes(split_bytes)?;
         Ok((
-            hotcache,
             BundleStorage {
                 storage,
                 bundle_filepath,
-                metadata,
+                file_ranges,
             },
+            hotcache,
         ))
     }
 
     /// Returns Iterator over files contained in the bundle.
     pub fn iter_files(&self) -> impl Iterator<Item = &PathBuf> {
-        self.metadata.files.keys()
+        self.file_ranges.files.keys()
+    }
+}
+
+const HOTCACHE_LEN_NUM_BYTES: usize = std::mem::size_of::<u32>();
+const BUNDLE_METADATA_LEN_NUM_BYTES: usize = std::mem::size_of::<u32>();
+const SPLIT_FOOTER_TRAILER_MAGIC: &[u8; 4] = b"QWFT";
+const SPLIT_FOOTER_TRAILER_VERSION: u32 = 1;
+const SPLIT_FOOTER_START_NUM_BYTES: usize = std::mem::size_of::<u64>();
+const SPLIT_FOOTER_TRAILER_VERSION_NUM_BYTES: usize = std::mem::size_of::<u32>();
+pub(crate) const SPLIT_FOOTER_TRAILER_NUM_BYTES: usize = SPLIT_FOOTER_START_NUM_BYTES
+    + SPLIT_FOOTER_TRAILER_VERSION_NUM_BYTES
+    + SPLIT_FOOTER_TRAILER_MAGIC.len();
+
+pub(crate) fn serialize_split_footer_trailer(
+    footer_start_inclusive: u64,
+) -> [u8; SPLIT_FOOTER_TRAILER_NUM_BYTES] {
+    let mut trailer = [0u8; SPLIT_FOOTER_TRAILER_NUM_BYTES];
+    let mut writer = trailer.as_mut_slice();
+    writer.put_u64_le(footer_start_inclusive);
+    writer.put_u32_le(SPLIT_FOOTER_TRAILER_VERSION);
+    writer.put_slice(SPLIT_FOOTER_TRAILER_MAGIC);
+    debug_assert!(!writer.has_remaining_mut());
+    trailer
+}
+
+fn deserialize_split_footer_trailer(trailer: &[u8]) -> anyhow::Result<Option<u64>> {
+    if trailer.len() != SPLIT_FOOTER_TRAILER_NUM_BYTES {
+        return Ok(None);
+    }
+    let mut reader = trailer;
+    let footer_start_inclusive = reader.get_u64_le();
+    let version = reader.get_u32_le();
+
+    if reader != SPLIT_FOOTER_TRAILER_MAGIC {
+        return Ok(None);
+    }
+    ensure!(
+        version == SPLIT_FOOTER_TRAILER_VERSION,
+        "unsupported split footer trailer version {version}"
+    );
+    Ok(Some(footer_start_inclusive))
+}
+
+/// Locates a split footer range using its fixed trailer, with support for legacy split layouts.
+pub async fn locate_split_footer_range(
+    storage: &dyn Storage,
+    split_path: &Path,
+    split_len: u64,
+) -> anyhow::Result<Range<u64>> {
+    ensure!(
+        split_len >= SPLIT_FOOTER_TRAILER_NUM_BYTES as u64,
+        "split is too short to contain a footer"
+    );
+    let trailer_start = split_len - SPLIT_FOOTER_TRAILER_NUM_BYTES as u64;
+    let trailer = storage
+        .get_slice(
+            split_path,
+            usize::try_from(trailer_start)?..usize::try_from(split_len)?,
+        )
+        .await?;
+    if let Some(footer_start_inclusive) = deserialize_split_footer_trailer(&trailer)? {
+        ensure!(
+            footer_start_inclusive <= trailer_start,
+            "split footer starts after its trailer"
+        );
+        return Ok(footer_start_inclusive..split_len);
     }
+
+    // Legacy split layout:
+    // [body][bundle metadata][metadata len][hotcache][hotcache len]
+    let hotcache_len = u32::from_le_bytes(trailer[12..].try_into().unwrap()) as u64;
+    let bundle_metadata_len_offset = split_len
+        .checked_sub(HOTCACHE_LEN_NUM_BYTES as u64)
+        .and_then(|offset| offset.checked_sub(hotcache_len))
+        .and_then(|offset| offset.checked_sub(BUNDLE_METADATA_LEN_NUM_BYTES as u64))
+        .ok_or_else(|| anyhow::anyhow!("invalid legacy split footer lengths"))?;
+    let bundle_metadata_len_bytes = storage
+        .get_slice(
+            split_path,
+            usize::try_from(bundle_metadata_len_offset)?
+                ..usize::try_from(
+                    bundle_metadata_len_offset + BUNDLE_METADATA_LEN_NUM_BYTES as u64,
+                )?,
+        )
+        .await?;
+    let bundle_metadata_len =
+        u32::from_le_bytes(bundle_metadata_len_bytes.as_ref().try_into().unwrap()) as u64;
+    let footer_start_inclusive = bundle_metadata_len_offset
+        .checked_sub(bundle_metadata_len)
+        .ok_or_else(|| anyhow::anyhow!("invalid legacy split metadata length"))?;
+    Ok(footer_start_inclusive..split_len)
 }
 
-const SPLIT_HOTBYTES_FOOTER_LENGTH_NUM_BYTES: usize = std::mem::size_of::<u32>();
-const BUNDLE_METADATA_LENGTH_NUM_BYTES: usize = std::mem::size_of::<u32>();
+/// Removes the fixed split footer trailer when it is present.
+pub fn strip_split_footer_trailer(split_slice: FileSlice) -> anyhow::Result<FileSlice> {
+    if split_slice.len() < SPLIT_FOOTER_TRAILER_NUM_BYTES {
+        return Ok(split_slice);
+    }
+    let (split_slice_without_trailer, trailer) = split_slice
+        .clone()
+        .split_from_end(SPLIT_FOOTER_TRAILER_NUM_BYTES);
+    if deserialize_split_footer_trailer(trailer.read_bytes()?.as_ref())?.is_some() {
+        Ok(split_slice_without_trailer)
+    } else {
+        Ok(split_slice)
+    }
+}
 
 #[derive(Copy, Clone, Default)]
 #[repr(u32)]
-pub enum BundleStorageFileOffsetsVersions {
+pub enum BundleFileRangesVersions {
     #[default]
     V1 = 1,
 }
 
-impl VersionedComponent for BundleStorageFileOffsetsVersions {
+impl VersionedComponent for BundleFileRangesVersions {
     const MAGIC_NUMBER: u32 = 403_881_646u32;
 
-    type Component = BundleStorageFileOffsets;
+    type Component = BundleFileRanges;
 
     fn to_version_code(self) -> u32 {
         self as u32
@@ -116,69 +223,65 @@ impl VersionedComponent for BundleStorageFileOffsetsVersions {
         }
     }
 
-    fn serialize_impl(component: &BundleStorageFileOffsets, output: &mut Vec<u8>) {
-        let metadata_json = serde_json::to_string(component).unwrap();
-        output.extend_from_slice(metadata_json.as_bytes());
+    fn serialize_impl(component: &BundleFileRanges, output: &mut Vec<u8>) {
+        let file_ranges_json = serde_json::to_string(component).unwrap();
+        output.extend_from_slice(file_ranges_json.as_bytes());
     }
 
     fn deserialize_impl(&self, bytes: &mut OwnedBytes) -> anyhow::Result<Self::Component> {
-        serde_json::from_reader(bytes).context("deserializing bundle storage file offsets failed")
+        serde_json::from_reader(bytes).context("deserializing bundle file ranges failed")
     }
 }
 
-/// Returns the file offsets in the file bundle.
+/// Maps files in a bundle to their byte ranges.
 #[derive(Debug, Default, Serialize, Deserialize, Clone)]
-pub struct BundleStorageFileOffsets {
-    /// The files and their offsets in the body
+pub struct BundleFileRanges {
+    /// The files and their byte ranges in the bundle body.
     pub files: HashMap<PathBuf, Range<u64>>,
 }
 
-impl BundleStorageFileOffsets {
+impl BundleFileRanges {
     /// File need to include split data (with hotcache at the end).
     /// See docs/internals/split-format.md
-    /// [Files, FileMetadata, FileMetadata Len, HotCache, HotCache Len]
-    /// Returns (Hotcache, Self)
-    fn open_from_split_data(file: FileSlice) -> anyhow::Result<(FileSlice, Self)> {
-        let (bundle_and_hotcache_bytes, hotcache_num_bytes_data) =
-            file.split_from_end(SPLIT_HOTBYTES_FOOTER_LENGTH_NUM_BYTES);
-        let hotcache_num_bytes: u32 = u32::from_le_bytes(
-            hotcache_num_bytes_data
-                .read_bytes()?
-                .as_ref()
-                .try_into()
-                .unwrap(),
-        );
-        let (bundle, hotcache) =
-            bundle_and_hotcache_bytes.split_from_end(hotcache_num_bytes as usize);
-        Ok((hotcache, Self::open(bundle)?))
+    /// [Files, BundleFileRanges, BundleMetadata Len, HotCache, HotCache Len, Split Footer Trailer]
+    /// Returns (Self, Hotcache)
+    fn open_from_split_bytes(split_bytes: OwnedBytes) -> anyhow::Result<(Self, OwnedBytes)> {
+        let split_slice = FileSlice::new(Arc::new(split_bytes));
+        let split_slice = strip_split_footer_trailer(split_slice)?;
+        let (bundle_and_hotcache_bytes, hotcache_len_data) =
+            split_slice.split_from_end(HOTCACHE_LEN_NUM_BYTES);
+        let hotcache_len: u32 =
+            u32::from_le_bytes(hotcache_len_data.read_bytes()?.as_ref().try_into().unwrap());
+        let (bundle, hotcache) = bundle_and_hotcache_bytes.split_from_end(hotcache_len as usize);
+        Ok((Self::open(bundle)?, hotcache.read_bytes()?))
     }
 
     /// FileSlice needs to end with the bundle (without hotcache from the split at the end).
     /// See docs/internals/split-format.md
-    /// [Files, FileMetadata, FileMetadata Len]
+    /// [Files, BundleFileRanges, BundleMetadata Len]
     pub fn open(file: FileSlice) -> anyhow::Result<Self> {
-        let (tantivy_files_data, num_bytes_file_metadata) =
-            file.split_from_end(BUNDLE_METADATA_LENGTH_NUM_BYTES);
-        let footer_num_bytes: u32 = u32::from_le_bytes(
-            num_bytes_file_metadata
+        let (bundle_and_metadata, bundle_metadata_len_data) =
+            file.split_from_end(BUNDLE_METADATA_LEN_NUM_BYTES);
+        let bundle_metadata_len: u32 = u32::from_le_bytes(
+            bundle_metadata_len_data
                 .read_bytes()?
                 .as_slice()
                 .try_into()
                 .unwrap(),
         );
 
-        let mut bundle_storage_file_offsets_data = tantivy_files_data
-            .slice_from_end(footer_num_bytes as usize)
+        let mut bundle_metadata_data = bundle_and_metadata
+            .slice_from_end(bundle_metadata_len as usize)
             .read_bytes()?;
-        BundleStorageFileOffsetsVersions::try_read_component(&mut bundle_storage_file_offsets_data)
+        BundleFileRangesVersions::try_read_component(&mut bundle_metadata_data)
     }
 
-    /// Returns file offsets for given path.
+    /// Returns the byte range for a given path.
     pub fn get(&self, path: &Path) -> Option<Range<u64>> {
         self.files.get(path).cloned()
     }
 
-    /// Returns whether file exists in metadata.
+    /// Returns whether the bundle contains a file at the given path.
     pub fn exists(&self, path: &Path) -> bool {
         self.files.contains_key(path)
     }
@@ -193,7 +296,7 @@ impl Storage for BundleStorage {
             .await
             .unwrap_or(false)
         {
-            anyhow::bail!("`{}` not found in storage", self.bundle_filepath.display())
+            bail!("`{}` not found in storage", self.bundle_filepath.display())
         }
         Ok(())
     }
@@ -211,9 +314,9 @@ impl Storage for BundleStorage {
         path: &Path,
         output: &mut dyn SendableAsync,
     ) -> crate::StorageResult<()> {
-        let file_num_bytes = self.file_num_bytes(path).await? as usize;
+        let file_len = self.file_num_bytes(path).await? as usize;
         let block_size = 100_000_000;
-        for block in chunk_range(0..file_num_bytes, block_size) {
+        for block in chunk_range(0..file_len, block_size) {
             let file_content = self.get_slice(path, block).await?;
             output.write_all(&file_content).await?;
         }
@@ -226,12 +329,12 @@ impl Storage for BundleStorage {
         path: &Path,
         range: Range<usize>,
     ) -> crate::StorageResult<OwnedBytes> {
-        let file_offsets = self.metadata.get(path).ok_or_else(|| {
+        let file_range = self.file_ranges.get(path).ok_or_else(|| {
             crate::StorageErrorKind::NotFound
                 .with_error(anyhow::anyhow!("missing file `{}`", path.display()))
         })?;
         let new_range =
-            file_offsets.start as usize + range.start..file_offsets.start as usize + range.end;
+            file_range.start as usize + range.start..file_range.start as usize + range.end;
         self.storage
             .get_slice(&self.bundle_filepath, new_range)
             .await
@@ -246,14 +349,14 @@ impl Storage for BundleStorage {
     }
 
     async fn get_all(&self, path: &Path) -> crate::StorageResult<OwnedBytes> {
-        let file_offsets = self.metadata.get(path).ok_or_else(|| {
+        let file_range = self.file_ranges.get(path).ok_or_else(|| {
             crate::StorageErrorKind::NotFound
                 .with_error(anyhow::anyhow!("missing file `{}`", path.display()))
         })?;
         self.storage
             .get_slice(
                 &self.bundle_filepath,
-                file_offsets.start as usize..file_offsets.end as usize,
+                file_range.start as usize..file_range.end as usize,
             )
             .await
     }
@@ -271,11 +374,11 @@ impl Storage for BundleStorage {
 
     async fn exists(&self, path: &Path) -> crate::StorageResult<bool> {
         // also check if self.bundle_file_name exists ?
-        Ok(self.metadata.exists(path))
+        Ok(self.file_ranges.exists(path))
     }
 
     async fn file_num_bytes(&self, path: &Path) -> StorageResult<u64> {
-        let file_range = self.metadata.get(path).ok_or_else(|| {
+        let file_range = self.file_ranges.get(path).ok_or_else(|| {
             crate::StorageErrorKind::NotFound
                 .with_error(anyhow::anyhow!("missing file `{}`", path.display()))
         })?;
@@ -298,7 +401,7 @@ impl fmt::Debug for BundleStorage {
         write!(
             f,
             "BundleStorage({:?}, files={:?})",
-            &self.bundle_filepath, self.metadata
+            &self.bundle_filepath, self.file_ranges
         )
     }
 }
@@ -318,7 +421,54 @@ mod tests {
     use crate::{PutPayload, RamStorageBuilder, SplitPayloadBuilder};
 
     #[tokio::test]
-    async fn bundle_storage_file_offsets() -> anyhow::Result<()> {
+    async fn bundle_storage_locates_footer_from_object_storage() {
+        let mut split_payload_builder = SplitPayloadBuilder::default();
+        split_payload_builder.add_payload("fields".to_string(), Box::new(b"fields".to_vec()));
+        let split_payload = split_payload_builder
+            .finalize_with_footer_trailer(b"hotcache", true)
+            .unwrap();
+        let expected_footer_range = split_payload.footer_range.clone();
+        let split_bytes = split_payload.read_all().await.unwrap();
+        let split_path = PathBuf::from("split");
+        let storage = Arc::new(
+            RamStorageBuilder::default()
+                .put(&split_path.to_string_lossy(), &split_bytes)
+                .build(),
+        );
+
+        let (_bundle_storage, hotcache, footer_range) =
+            BundleStorage::open_from_storage(storage, split_path)
+                .await
+                .unwrap();
+
+        assert_eq!(hotcache.as_ref(), b"hotcache");
+        assert_eq!(footer_range, expected_footer_range);
+    }
+
+    #[tokio::test]
+    async fn bundle_storage_locates_legacy_footer_from_object_storage() {
+        let split_payload =
+            SplitPayloadBuilder::get_split_payload(&[], b"fields", None, b"hotcache").unwrap();
+        let expected_footer_range = split_payload.footer_range.clone();
+        let split_bytes = split_payload.read_all().await.unwrap();
+        let split_path = PathBuf::from("legacy-split");
+        let storage = Arc::new(
+            RamStorageBuilder::default()
+                .put(&split_path.to_string_lossy(), &split_bytes)
+                .build(),
+        );
+
+        let (_bundle_storage, hotcache, footer_range) =
+            BundleStorage::open_from_storage(storage, split_path)
+                .await
+                .unwrap();
+
+        assert_eq!(hotcache.as_ref(), b"hotcache");
+        assert_eq!(footer_range, expected_footer_range);
+    }
+
+    #[tokio::test]
+    async fn bundle_file_ranges() -> anyhow::Result<()> {
         let temp_dir = tempfile::tempdir()?;
         let test_filepath1 = temp_dir.path().join("f1");
         let test_filepath2 = temp_dir.path().join("f2");
@@ -332,22 +482,21 @@ mod tests {
         let buffer = SplitPayloadBuilder::get_split_payload(
             &[test_filepath1.clone(), test_filepath2.clone()],
             &[],
+            None,
             &[5, 5, 5],
         )?
         .read_all()
         .await?;
 
         let bundle_filepath = Path::new("bundle");
-        let bundle_file_slice = FileSlice::new(Arc::new(buffer.clone()));
-        let (hotcache, metadata) =
-            BundleStorageFileOffsets::open_from_split_data(bundle_file_slice)?;
-        assert_eq!(hotcache.read_bytes().unwrap().as_ref(), &[5, 5, 5]);
+        let (file_ranges, hotcache) = BundleFileRanges::open_from_split_bytes(buffer.clone())?;
+        assert_eq!(hotcache.as_ref(), &[5, 5, 5]);
         let ram_storage = RamStorageBuilder::default()
             .put(&bundle_filepath.to_string_lossy(), &buffer)
             .build();
 
         let bundle_storage = BundleStorage {
-            metadata,
+            file_ranges,
             bundle_filepath: bundle_filepath.to_path_buf(),
             storage: Arc::new(ram_storage),
         };
@@ -374,14 +523,14 @@ mod tests {
         let buffer = SplitPayloadBuilder::get_split_payload(
             &[test_filepath1.clone(), test_filepath2.clone()],
             &[],
+            None,
             &[1, 3, 3, 7],
         )?
         .read_all()
         .await?;
 
-        let (hotcache, metadata) =
-            BundleStorageFileOffsets::open_from_split_data(FileSlice::from(buffer.to_vec()))?;
-        assert_eq!(hotcache.read_bytes().unwrap().as_ref(), &[1, 3, 3, 7]);
+        let (file_ranges, hotcache) = BundleFileRanges::open_from_split_bytes(buffer.clone())?;
+        assert_eq!(hotcache.as_ref(), &[1, 3, 3, 7]);
 
         let bundle_filepath = Path::new("bundle");
         let ram_storage = RamStorageBuilder::default()
@@ -389,7 +538,7 @@ mod tests {
             .build();
 
         let bundle_storage = BundleStorage {
-            metadata,
+            file_ranges,
             bundle_filepath: bundle_filepath.to_path_buf(),
             storage: Arc::new(ram_storage),
         };
@@ -411,19 +560,18 @@ mod tests {
 
     #[tokio::test]
     async fn bundlestorage_test_empty() -> anyhow::Result<()> {
-        let buffer = SplitPayloadBuilder::get_split_payload(&[], &[], &[])?
+        let buffer = SplitPayloadBuilder::get_split_payload(&[], &[], None, &[])?
             .read_all()
             .await?;
 
-        let (_hotcache, metadata) =
-            BundleStorageFileOffsets::open_from_split_data(FileSlice::from(buffer.to_vec()))?;
+        let (file_ranges, _hotcache) = BundleFileRanges::open_from_split_bytes(buffer.clone())?;
 
         let bundle_filepath = PathBuf::from("bundle");
         let ram_storage = RamStorageBuilder::default()
             .put(&bundle_filepath.to_string_lossy(), &buffer)
             .build();
         let bundle_storage = BundleStorage {
-            metadata,
+            file_ranges,
             bundle_filepath,
             storage: Arc::new(ram_storage),
         };
diff --git a/quickwit/quickwit-storage/src/lib.rs b/quickwit/quickwit-storage/src/lib.rs
index a2d0a20a39b..73ef173eee2 100644
--- a/quickwit/quickwit-storage/src/lib.rs
+++ b/quickwit/quickwit-storage/src/lib.rs
@@ -60,7 +60,9 @@ pub use split_cache::SearchSplitCache;
 pub use tantivy::directory::OwnedBytes;
 pub use versioned_component::VersionedComponent;
 
-pub use self::bundle_storage::{BundleStorage, BundleStorageFileOffsets};
+pub use self::bundle_storage::{
+    BundleFileRanges, BundleStorage, locate_split_footer_range, strip_split_footer_trailer,
+};
 #[cfg(any(test, feature = "testsuite"))]
 pub use self::cache::MockStorageCache;
 pub use self::cache::{
diff --git a/quickwit/quickwit-storage/src/split.rs b/quickwit/quickwit-storage/src/split.rs
index e1479c24a87..9625b81d170 100644
--- a/quickwit/quickwit-storage/src/split.rs
+++ b/quickwit/quickwit-storage/src/split.rs
@@ -23,10 +23,10 @@ use aws_sdk_s3::primitives::{ByteStream, FsBuilder, Length, SdkBody};
 use futures::{Stream, StreamExt, stream};
 use hyper::body::{Bytes, Frame};
 use pin_project::pin_project;
-use quickwit_common::shared_consts::SPLIT_FIELDS_FILE_NAME;
+use quickwit_common::shared_consts::{SPLIT_FIELDS_FILE_NAME, SPLIT_RECOVERY_METADATA_FILE_NAME};
 
-use crate::bundle_storage::BundleStorageFileOffsetsVersions;
-use crate::{BundleStorageFileOffsets, PutPayload, VersionedComponent};
+use crate::bundle_storage::{BundleFileRangesVersions, serialize_split_footer_trailer};
+use crate::{BundleFileRanges, PutPayload, VersionedComponent};
 
 /// Payload of a split which builds the split bundle and hotcache on the fly and streams it to the
 /// storage.
@@ -143,10 +143,11 @@ pub struct SplitPayloadBuilder {
 }
 
 impl SplitPayloadBuilder {
-    /// Creates a new SplitPayloadBuilder for given files and hotcache.
+    /// Creates a new SplitPayloadBuilder for given files, recovery metadata, and hotcache.
     pub fn get_split_payload(
         split_files: &[PathBuf],
         serialized_split_fields: &[u8],
+        serialized_recovery_metadata: Option<&[u8]>,
         hotcache: &[u8],
     ) -> anyhow::Result<SplitPayload> {
         let mut split_payload_builder = SplitPayloadBuilder::default();
@@ -157,6 +158,12 @@ impl SplitPayloadBuilder {
             SPLIT_FIELDS_FILE_NAME.to_string(),
             Box::new(serialized_split_fields.to_vec()),
         );
+        if let Some(serialized_recovery_metadata) = serialized_recovery_metadata {
+            split_payload_builder.add_payload(
+                SPLIT_RECOVERY_METADATA_FILE_NAME.to_string(),
+                Box::new(serialized_recovery_metadata.to_vec()),
+            );
+        }
         let offsets = split_payload_builder.finalize(hotcache)?;
         Ok(offsets)
     }
@@ -192,12 +199,20 @@ impl SplitPayloadBuilder {
         Ok(())
     }
 
-    /// Writes the bundle file offsets metadata at the end of the bundle file,
-    /// and returns the byte-range of this metadata information.
+    /// Writes the bundle file ranges at the end of the bundle file.
     pub fn finalize(self, hotcache: &[u8]) -> anyhow::Result<SplitPayload> {
-        // Add the fields metadata to the bundle metadata.
+        let enable_footer_trailer =
+            quickwit_common::get_bool_from_env_cached!("QW_ENABLE_SPLIT_FOOTER_TRAILER", false);
+        self.finalize_with_footer_trailer(hotcache, enable_footer_trailer)
+    }
+
+    pub(crate) fn finalize_with_footer_trailer(
+        self,
+        hotcache: &[u8],
+        enable_footer_trailer: bool,
+    ) -> anyhow::Result<SplitPayload> {
         // Build the footer.
-        let metadata_with_fixed_paths = self
+        let file_ranges = self
             .payloads
             .iter()
             .map(|(file_name, _, range)| {
@@ -206,19 +221,19 @@ impl SplitPayloadBuilder {
             })
             .collect::<Result<HashMap<_, _>, anyhow::Error>>()?;
 
-        let bundle_storage_file_offsets = BundleStorageFileOffsets {
-            files: metadata_with_fixed_paths,
-        };
-        let metadata_json =
-            BundleStorageFileOffsetsVersions::serialize(&bundle_storage_file_offsets);
+        let bundle_file_ranges = BundleFileRanges { files: file_ranges };
+        let bundle_metadata = BundleFileRangesVersions::serialize(&bundle_file_ranges);
 
-        // The hotcache needs to be the next to the metadata in order to be able to read both
+        // The hotcache needs to be next to the bundle metadata in order to read both
         // in one continuous read.
         let mut footer_bytes = Vec::new();
-        footer_bytes.extend(&metadata_json);
-        footer_bytes.extend((metadata_json.len() as u32).to_le_bytes());
+        footer_bytes.extend(&bundle_metadata);
+        footer_bytes.extend((bundle_metadata.len() as u32).to_le_bytes());
         footer_bytes.extend(hotcache);
         footer_bytes.extend((hotcache.len() as u32).to_le_bytes());
+        if enable_footer_trailer {
+            footer_bytes.extend(serialize_split_footer_trailer(self.current_offset as u64));
+        }
 
         let mut payloads: Vec<Box<dyn PutPayload>> = self
             .payloads
@@ -280,6 +295,8 @@ mod tests {
     use std::fs::File;
     use std::io::Write;
 
+    use tantivy::directory::FileSlice;
+
     use super::*;
 
     #[tokio::test]
@@ -294,14 +311,37 @@ mod tests {
         let mut file2 = File::create(&test_filepath2)?;
         file2.write_all(b"world")?;
 
-        let split_payload =
-            SplitPayloadBuilder::get_split_payload(&[test_filepath1, test_filepath2], &[], b"abc")?;
+        let split_payload = SplitPayloadBuilder::get_split_payload(
+            &[test_filepath1, test_filepath2],
+            &[],
+            None,
+            b"abc",
+        )?;
 
         assert_eq!(split_payload.len(), 128);
 
         Ok(())
     }
 
+    #[tokio::test]
+    async fn test_split_payload_embeds_recovery_metadata() {
+        let recovery_metadata = b"recovery-protobuf";
+        let split_payload = SplitPayloadBuilder::get_split_payload(
+            &[],
+            b"fields",
+            Some(recovery_metadata),
+            b"hotcache",
+        )
+        .unwrap();
+
+        let recovery_range =
+            b"fields".len() as u64..(b"fields".len() + recovery_metadata.len()) as u64;
+        assert_eq!(
+            fetch_data(&split_payload, recovery_range).await.unwrap(),
+            recovery_metadata
+        );
+    }
+
     #[cfg(test)]
     async fn fetch_data(
         split_streamer: &SplitPayload,
@@ -431,6 +471,7 @@ mod tests {
         let split_streamer = SplitPayloadBuilder::get_split_payload(
             &[test_filepath1.clone(), test_filepath2.clone()],
             &[],
+            None,
             &[1, 2, 3],
         )?;
 
@@ -471,8 +512,12 @@ mod tests {
         let total_len = split_streamer.len();
         let all_data = fetch_data(&split_streamer, 0..total_len).await?;
 
-        // last 8 bytes are the length of the hotcache bytes
-        assert_eq!(all_data[all_data.len() - 4..], 3_u32.to_le_bytes());
+        let split_without_trailer =
+            crate::strip_split_footer_trailer(FileSlice::from(all_data))?.read_bytes()?;
+        assert_eq!(
+            split_without_trailer[split_without_trailer.len() - 4..],
+            3_u32.to_le_bytes()
+        );
         Ok(())
     }
 }
diff --git a/quickwit/scripts/inspect_split.py b/quickwit/scripts/inspect_split.py
index 8cc1f2486de..0f6063ffdb2 100755
--- a/quickwit/scripts/inspect_split.py
+++ b/quickwit/scripts/inspect_split.py
@@ -3,13 +3,14 @@
 
 A split is laid out from the tail as:
     [ ... file data ... ]
-    [ 8 bytes: magic + version (BundleStorageFileOffsetsVersions) ]
+    [ 8 bytes: magic + version (BundleFileRangesVersions) ]
     [ JSON: {"files": {path: {"start": .., "end": ..}, ...}} ]
     [ 4 bytes LE u32: bundle metadata length (covers 8B header + JSON) ]
     [ ... hotcache bytes ... ]
     [ 4 bytes LE u32: hotcache length ]
+    [ optional 16-byte split footer trailer: footer start (u64), version (u32), QWFT ]
 
-See `BundleStorage::open_from_split_data` in
+See `BundleStorage::open_from_split_bytes` in
 `quickwit/quickwit-storage/src/bundle_storage.rs` for the authoritative reader.
 """
 
@@ -21,20 +22,51 @@
 
 FOOTER_LEN_BYTES = 4
 BUNDLE_HEADER_BYTES = 8  # magic (u32 LE) + version (u32 LE)
+SPLIT_FOOTER_TRAILER_BYTES = 16
+SPLIT_FOOTER_TRAILER_MAGIC = b"QWFT"
+SPLIT_FOOTER_TRAILER_VERSION = 1
 
 
 def inspect(path: Path) -> dict:
     with path.open("rb") as f:
-        # Read hotcache length (last 4 bytes).
-        f.seek(-FOOTER_LEN_BYTES, 2)
+        f.seek(0, 2)
+        split_size = f.tell()
+        footer_end = split_size
+
+        if split_size >= SPLIT_FOOTER_TRAILER_BYTES:
+            f.seek(-SPLIT_FOOTER_TRAILER_BYTES, 2)
+            trailer = f.read(SPLIT_FOOTER_TRAILER_BYTES)
+            footer_start, trailer_version, trailer_magic = struct.unpack("<QI4s", trailer)
+            if trailer_magic == SPLIT_FOOTER_TRAILER_MAGIC:
+                if trailer_version != SPLIT_FOOTER_TRAILER_VERSION:
+                    raise RuntimeError(
+                        f"unsupported split footer trailer version ({trailer_version})"
+                    )
+                footer_end -= SPLIT_FOOTER_TRAILER_BYTES
+                if footer_start > footer_end:
+                    raise RuntimeError(
+                        f"invalid split footer start ({footer_start} > {footer_end})"
+                    )
+
+        if footer_end < FOOTER_LEN_BYTES:
+            raise RuntimeError("split is too short to contain a hotcache length")
+
+        # Read hotcache length immediately before the optional trailer.
+        f.seek(footer_end - FOOTER_LEN_BYTES)
         (hot_len,) = struct.unpack("<I", f.read(FOOTER_LEN_BYTES))
 
         # Read bundle metadata length (4 bytes preceding the hotcache).
-        f.seek(-(FOOTER_LEN_BYTES + hot_len + FOOTER_LEN_BYTES), 2)
+        meta_len_offset = footer_end - FOOTER_LEN_BYTES - hot_len - FOOTER_LEN_BYTES
+        if meta_len_offset < 0:
+            raise RuntimeError("invalid hotcache length")
+        f.seek(meta_len_offset)
         (meta_len,) = struct.unpack("<I", f.read(FOOTER_LEN_BYTES))
 
         # Read bundle metadata (the 8-byte magic+version header followed by JSON).
-        f.seek(-(FOOTER_LEN_BYTES + hot_len + FOOTER_LEN_BYTES + meta_len), 2)
+        meta_offset = meta_len_offset - meta_len
+        if meta_offset < 0:
+            raise RuntimeError("invalid bundle metadata length")
+        f.seek(meta_offset)
         meta = f.read(meta_len)
 
     if len(meta) < BUNDLE_HEADER_BYTES: