[#268] fix: CodeRabbit 리뷰 반영 - torn tail truncate 및 commit marker 순서 수정

- FileSystem 트레이트에 truncate 메서드 추가 (RealFileSystem 구현)
- newest 세그먼트의 torn tail을 recovery 시 디스크에서 실제 truncate:
  stale bytes가 rotation 후 intermediate corruption으로 이어지는 것 방지
- write_entry의 header 슬롯 예약을 실제 frame_len이 아닌 0으로 수행:
  예약 시점에 commit marker가 노출되는 문제 수정
- torn tail truncate 검증 테스트 추가

Author: DPS0340

src/common/fs.rs

@@ -14,6 +14,9 @@ pub trait FileSystem {
     async fn write_file(&self, path: &str, content: &[u8]) -> io::Result<()>;
     async fn read_dir(&self, path: &str) -> io::Result<Vec<FileSystemEntry>>;
     async fn read(&self, path: &Path) -> io::Result<Vec<u8>>;
+    /// 파일을 지정한 크기로 자릅니다. (#268)
+    /// WAL 복구 시 newest segment의 torn tail을 제거하는 데 사용합니다.
+    async fn truncate(&self, path: &Path, size: u64) -> io::Result<()>;
     /// 파일의 크기(bytes)를 반환합니다. (#265)
     /// `read_segment_rows`가 파일 전체를 메모리로 읽기 전에 예산을 확보하는 데 사용합니다.
     async fn metadata(&self, path: &Path) -> io::Result<u64>;
@@ -49,6 +52,11 @@ impl FileSystem for RealFileSystem {
         tokio::fs::read(path).await
     }
 
+    async fn truncate(&self, path: &Path, size: u64) -> io::Result<()> {
+        let file = tokio::fs::OpenOptions::new().write(true).open(path).await?;
+        file.set_len(size).await
+    }
+
     async fn metadata(&self, path: &Path) -> io::Result<u64> {
         let metadata = tokio::fs::metadata(path).await?;
         Ok(metadata.len())

src/engine/wal/manager/builder.rs

@@ -110,6 +110,20 @@ impl<'a> WALBuilder<'a> {
                 // before it, following the PostgreSQL/SQLite recovery
                 // convention.
                 log::warn!("discarding torn WAL tail in {:?}: {}", path, reason);
+
+                // Truncate the torn tail on disk. Otherwise the stale bytes
+                // would survive a checkpoint and, once this segment rotates
+                // into an intermediate one, the next restart would treat
+                // them as intermediate corruption and fail to start.
+                self.file_system
+                    .truncate(&path, used_bytes as u64)
+                    .await
+                    .map_err(|e| {
+                        WALError::wrap(format!(
+                            "failed to truncate torn WAL tail in {:?}: {}",
+                            path, e
+                        ))
+                    })?;
             }
 
             let entries: Vec<WALEntry> = decoder.decode(&content[..used_bytes]).map_err(|e| {

src/engine/wal/manager/mod.rs

@@ -126,12 +126,13 @@ where
         // memcpy, the on-disk frame either has len == 0 (unwritten) or a
         // truncated body under a valid len — both are discarded on recovery.
         let header_offset = self.current_offset;
-        let header = &frame[..size_of::<u32>()];
-        // Reserve the header slot first (as zeros = "not committed"), then
-        // write the body, then patch the header last (see #268).
-        self.append_frame_to_mmap(header).await?;
+        // Reserve the header slot as zeros first: an unwritten slot can never
+        // be mistaken for a committed frame. Then write the body, and patch
+        // the real length header last (see #268).
+        self.append_frame_to_mmap(&[0u8; size_of::<u32>()]).await?;
         let body = &frame[size_of::<u32>()..];
         self.append_frame_to_mmap(body).await?;
+        let header = &frame[..size_of::<u32>()];
         self.patch_frame_header_at(header_offset, header).await?;
 
         self.unsynced_bytes += frame.len();
@@ -734,6 +735,47 @@ mod tests {
         assert_eq!(payloads, vec![b"complete".to_vec()]);
     }
 
+    /// #268: after discarding a torn tail the file itself must be truncated,
+    /// so the segment stays clean even after it later becomes an intermediate
+    /// segment (rotation) — otherwise the next restart would fail.
+    #[tokio::test]
+    async fn test_build_truncates_torn_tail_from_disk() {
+        let wal_dir = setup_test_wal_dir("truncate_torn_tail").await;
+        let config = get_test_config(&wal_dir);
+        write_wal_file(
+            &config,
+            1,
+            &vec![create_entry(EntryType::Insert, Some("complete"))],
+        )
+        .await;
+
+        let path = wal_dir.join(format!("00000001.{}", config.wal_extension));
+        let intact_len = tokio::fs::metadata(&path).await.unwrap().len();
+        {
+            let mut file = tokio::fs::OpenOptions::new()
+                .append(true)
+                .open(&path)
+                .await
+                .unwrap();
+            file.write_all(&8u32.to_le_bytes()).await.unwrap();
+            file.write_all(&[1, 2, 3]).await.unwrap();
+        }
+        assert!(tokio::fs::metadata(&path).await.unwrap().len() > intact_len);
+
+        let wal_manager = WALBuilder::new(&config)
+            .build(BincodeDecoder::new(), BincodeEncoder::new())
+            .await
+            .expect("a torn tail must not prevent startup");
+        assert_eq!(wal_manager.pending_entries().len(), 1);
+
+        // The torn bytes must be gone from the file itself.
+        assert_eq!(
+            tokio::fs::metadata(&path).await.unwrap().len(),
+            intact_len,
+            "the torn tail must be truncated from disk, not just ignored"
+        );
+    }
+
     #[tokio::test]
     async fn test_build_rejects_corrupt_intermediate_segment() {
         let wal_dir = setup_test_wal_dir("corrupt_intermediate_segment").await;