From 33f77372e440d570f111c0efec60c35d77f348ee Mon Sep 17 00:00:00 2001
From: Paulo Cabral Sanz <paulo@railway.app>
Date: Sat, 15 Aug 2026 01:29:43 -0300
Subject: [PATCH 1/2] test: footer body_len inflate must not collapse data_end
 to 0

body_len sits outside the CRC. scan_since trusted it, so
body_len=records_end made data_end_offset return 0.
---
 crates/engine/src/log/file.rs | 29 +++++++++++++++++++++++++++++
 1 file changed, 29 insertions(+)

diff --git a/crates/engine/src/log/file.rs b/crates/engine/src/log/file.rs
index cde2fcb..b684fc2 100644
--- a/crates/engine/src/log/file.rs
+++ b/crates/engine/src/log/file.rs
@@ -598,4 +598,33 @@ mod enospc_tests {
             );
         });
     }
+
+    /// `body_len` sits outside the footer CRC. Trusting it blinds
+    /// `scan_since`: `body_len = records_end` → `data_end_offset = 0`.
+    #[test]
+    fn footer_body_len_inflate_must_not_blind_data_end() {
+        run(async {
+            let dir = TempDir::new().unwrap();
+            let path = dir.path().join("data-0000000000.log");
+            let f = LogFile::open_rw(path.clone(), 0).await.unwrap();
+            let (_, _) = f.append(b"keep-record-bytes".to_vec()).await.unwrap();
+            let records_end = f.write_offset();
+            f.write_footer(&[]).await.unwrap();
+            let total = f.write_offset();
+            assert!(total > records_end);
+
+            // Flip the trailer body_len (first 8 bytes of the 24-byte trailer)
+            // to claim the whole file is footer. Magic stays intact.
+            let mut bytes = std::fs::read(&path).unwrap();
+            let blen_at = bytes.len() - FOOTER_TRAILER_LEN as usize;
+            bytes[blen_at..blen_at + 8].copy_from_slice(&records_end.to_le_bytes());
+            std::fs::write(&path, &bytes).unwrap();
+
+            let end = f.data_end_offset().await;
+            assert_eq!(
+                end, total,
+                "unauthenticated body_len must not collapse data_end to 0 (got {end}, total {total}, records_end {records_end})"
+            );
+        });
+    }
 }

From 4b44153eaf69fa2c208e5207cd352ee691ae170a Mon Sep 17 00:00:00 2001
From: Paulo Cabral Sanz <paulo@railway.app>
Date: Sat, 15 Aug 2026 01:30:45 -0300
Subject: [PATCH 2/2] fix: data_end_offset must not trust unauthenticated
 footer body_len

body_len sits outside the CRC. scan_since used it raw, so a flip
to records_end collapsed the watch window to 0. Reuse read_footer's
range+CRC check; on any miss scan to EOF.
---
 crates/engine/src/log/file.rs | 24 +++++++++++++-----------
 1 file changed, 13 insertions(+), 11 deletions(-)

diff --git a/crates/engine/src/log/file.rs b/crates/engine/src/log/file.rs
index b684fc2..f08caaf 100644
--- a/crates/engine/src/log/file.rs
+++ b/crates/engine/src/log/file.rs
@@ -326,23 +326,25 @@ impl LogFile {
 
     /// Returns the byte offset where record data ends — stops before the footer
     /// in sealed files so that `scan_since` doesn't misparse footer bytes as records.
+    ///
+    /// `body_len` in the trailer is outside the CRC. Trusting it unauthenticated
+    /// lets a flip collapse this to 0 and blind watch catch-up. Reuse
+    /// [`read_footer`]'s range+CRC check; on any miss scan to EOF.
     pub async fn data_end_offset(&self) -> u64 {
         let total = self.write_offset.get();
         if total < FOOTER_TRAILER_LEN {
             return total;
         }
-        let Ok(magic_bytes) = self.read_exact(total - 8, 8).await else {
-            return total;
-        };
-        let magic = u64::from_le_bytes(<[u8; 8]>::try_from(&magic_bytes[..]).unwrap_or([0u8; 8]));
-        if magic != FOOTER_MAGIC {
-            return total;
+        match self.read_footer().await {
+            Ok(Some(entries)) => {
+                let body_len: u64 = entries.iter().map(|e| e.encoded_size() as u64).sum();
+                match FOOTER_TRAILER_LEN.checked_add(body_len) {
+                    Some(n) => total.saturating_sub(n),
+                    None => total,
+                }
+            }
+            Ok(None) | Err(_) => total,
         }
-        let Ok(blen_bytes) = self.read_exact(total - FOOTER_TRAILER_LEN, 8).await else {
-            return total;
-        };
-        let body_len = u64::from_le_bytes(<[u8; 8]>::try_from(&blen_bytes[..]).unwrap_or([0u8; 8]));
-        total.saturating_sub(FOOTER_TRAILER_LEN + body_len)
     }
 
     /// Append a buffer at an offset reserved atomically *before* awaiting the