Repair unterminated rollout files before appending (#32276)

## Why

Appending directly to a non-empty rollout that lacks a trailing newline joins the
next JSON object to the existing record, producing invalid JSONL.

## What changed

- Ensure non-empty rollout files end with a newline before opening them for
  append, including resumed, compressed, and direct append paths.
- Perform the synchronous tail inspection outside the async runtime.

## Testing

- Add a regression test covering newline repair and repeated opens.

GitOrigin-RevId: a3ca84cc1fd3b811caa9f4c5c8761a4b5b791357
This commit is contained in:
Owen Lin
2026-07-10 21:31:00 +00:00
committed by copyberry
parent c0ea3c4d0a
commit e7d0e14172
2 changed files with 55 additions and 19 deletions
+43 -19
View File
@@ -4,6 +4,10 @@ use std::collections::HashSet;
use std::fs;
use std::fs::File;
use std::io::Error as IoError;
use std::io::Read;
use std::io::Seek;
use std::io::SeekFrom;
use std::io::Write;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
@@ -811,18 +815,8 @@ impl RolloutRecorder {
(None, Some(log_file_info), path, Some(session_meta))
}
RolloutRecorderParams::Resume { path } => {
let path = compression::materialize_rollout_for_append(path.as_path()).await?;
(
Some(
tokio::fs::OpenOptions::new()
.append(true)
.open(&path)
.await?,
),
None,
path,
None,
)
let (path, file) = open_rollout_for_append(path.as_path()).await?;
(Some(file), None, path, None)
}
};
@@ -1536,10 +1530,13 @@ fn open_log_file(path: &Path) -> std::io::Result<File> {
)));
};
fs::create_dir_all(parent)?;
std::fs::OpenOptions::new()
let mut file = std::fs::OpenOptions::new()
.read(true)
.append(true)
.create(true)
.open(path)
.open(path)?;
ensure_rollout_is_newline_terminated(&mut file)?;
Ok(file)
}
/// Mutable state owned by the background rollout writer.
@@ -1788,15 +1785,42 @@ pub async fn append_rollout_item_to_path(
rollout_path: &Path,
item: &RolloutItem,
) -> std::io::Result<()> {
let rollout_path = compression::materialize_rollout_for_append(rollout_path).await?;
let file = tokio::fs::OpenOptions::new()
.append(true)
.open(rollout_path)
.await?;
let (_rollout_path, file) = open_rollout_for_append(rollout_path).await?;
let mut writer = JsonlWriter { file };
writer.write_rollout_item(item).await
}
async fn open_rollout_for_append(path: &Path) -> std::io::Result<(PathBuf, tokio::fs::File)> {
let path = compression::materialize_rollout_for_append(path).await?;
let path_for_open = path.clone();
let file = tokio::task::spawn_blocking(move || {
let mut file = File::options()
.read(true)
.append(true)
.open(path_for_open)?;
ensure_rollout_is_newline_terminated(&mut file)?;
Ok::<_, std::io::Error>(file)
})
.await
.map_err(IoError::other)??;
Ok((path, tokio::fs::File::from_std(file)))
}
fn ensure_rollout_is_newline_terminated(file: &mut File) -> std::io::Result<()> {
if file.metadata()?.len() == 0 {
return Ok(());
}
file.seek(SeekFrom::End(-1))?;
let mut final_byte = [0];
file.read_exact(&mut final_byte)?;
if final_byte[0] != b'\n' {
file.write_all(b"\n")?;
file.flush()?;
}
Ok(())
}
struct JsonlWriter {
file: tokio::fs::File,
}
+12
View File
@@ -71,6 +71,18 @@ fn write_session_file(root: &Path, ts: &str, uuid: Uuid) -> std::io::Result<Path
Ok(path)
}
#[test]
fn append_repair_terminates_nonempty_rollout_tail() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
let rollout_path = home.path().join("rollout.jsonl");
fs::write(&rollout_path, b"{\"type\":\"event_msg\"}")?;
drop(open_log_file(&rollout_path)?);
drop(open_log_file(&rollout_path)?);
assert_eq!(fs::read(&rollout_path)?, b"{\"type\":\"event_msg\"}\n");
Ok(())
}
#[tokio::test]
async fn state_db_init_backfills_before_returning() -> anyhow::Result<()> {
let home = TempDir::new().expect("temp dir");