feat: add size-based operational log rotation

crates/cli/src/configuration/logging.rs

@@ -7,8 +7,9 @@ use std::path::PathBuf;
 
 use nemo_relay::error::FlowError;
 use nemo_relay::logging::{
-    DEFAULT_FILE_FLUSH_INTERVAL_MILLIS, DEFAULT_FILE_SINK_QUEUE_ENTRIES, FileLogSinkConfig,
-    LogFormat, LogLevel, LogSinkConfig, LoggingConfig, MAX_FILE_SINK_QUEUE_ENTRIES,
+    DEFAULT_FILE_FLUSH_INTERVAL_MILLIS, DEFAULT_FILE_SINK_QUEUE_ENTRIES, FileLogRotationConfig,
+    FileLogSinkConfig, LogFormat, LogLevel, LogSinkConfig, LoggingConfig,
+    MAX_FILE_SINK_QUEUE_ENTRIES,
 };
 use serde::Deserialize;
 
@@ -36,6 +37,8 @@ struct RawFileLogSinkConfig {
     /// Optional advanced: pending async queue entries per file sink (default
     /// [`DEFAULT_FILE_SINK_QUEUE_ENTRIES`]).
     queue_capacity: Option<usize>,
+    max_file_size_bytes: Option<u64>,
+    retained_files: Option<usize>,
 }
 
 pub(super) fn apply_file_logging_config(
@@ -100,11 +103,25 @@ fn parse_file_log_sink(
         Some(capacity) => capacity,
         None => DEFAULT_FILE_SINK_QUEUE_ENTRIES,
     };
+    let rotation = match (config.max_file_size_bytes, config.retained_files) {
+        (None, None) => None,
+        (Some(max_file_size_bytes), Some(retained_files)) => Some(
+            FileLogRotationConfig::new(max_file_size_bytes, retained_files)
+                .map_err(logging_parse_error)?,
+        ),
+        _ => {
+            return Err(CliError::Config(
+                "logging sink max_file_size_bytes and retained_files must be configured together"
+                    .into(),
+            ));
+        }
+    };
     Ok(LogSinkConfig::File(FileLogSinkConfig {
         path,
         level,
         format,
         queue_capacity,
+        rotation,
     }))
 }
 

crates/cli/tests/coverage/shared/config_tests.rs

@@ -3271,6 +3271,56 @@ format = "human"
     }
 }
 
+#[test]
+fn logging_rotation_cli_config_preserves_pair_and_rejects_incomplete_pair() {
+    let temp = tempfile::tempdir().unwrap();
+    let config_path = isolated_config_path(&temp);
+    let log_path = temp.path().join("relay.log.jsonl");
+    std::fs::write(
+        &config_path,
+        format!(
+            r#"
+[[logging.sinks]]
+path = {}
+max_file_size_bytes = 1024
+retained_files = 2
+"#,
+            toml_basic_string(log_path.to_string_lossy().as_ref())
+        ),
+    )
+    .unwrap();
+
+    let resolved = resolve_server_config(&GatewayOverrides {
+        config: Some(config_path),
+        ..GatewayOverrides::default()
+    })
+    .unwrap();
+    let LogSinkConfig::File(sink) = &resolved.logging.sinks[0];
+    let rotation = sink.rotation.expect("complete rotation configuration");
+    assert_eq!(rotation.max_file_size_bytes(), 1024);
+    assert_eq!(rotation.retained_files(), 2);
+
+    let incomplete_path = isolated_config_path(&temp);
+    std::fs::write(
+        &incomplete_path,
+        r#"
+[[logging.sinks]]
+path = "relay.log.jsonl"
+max_file_size_bytes = 1024
+"#,
+    )
+    .unwrap();
+    let error = resolve_server_config(&GatewayOverrides {
+        config: Some(incomplete_path),
+        ..GatewayOverrides::default()
+    })
+    .unwrap_err()
+    .to_string();
+    assert!(error.contains(
+        "logging sink max_file_size_bytes and retained_files must be configured together"
+    ));
+}
+
 #[test]
 fn logging_rejects_invalid_level_format_missing_path_and_zero_queue() {
     let temp = tempfile::tempdir().unwrap();

crates/core/src/logging/config.rs

@@ -28,6 +28,13 @@ pub const DEFAULT_FILE_FLUSH_INTERVAL_MILLIS: u64 = 1000;
 /// configuration above this bound is rejected with a config error. It cannot be raised.
 pub const MAX_FILE_SINK_QUEUE_ENTRIES: usize = 8_192;
 
+/// Fixed hard maximum number of retained backup files per rotating file sink.
+///
+/// Size-based rotation renames existing backup files on each rotation, so an unbounded value can
+/// make one log write perform excessive filesystem work. This limit counts backup files and does
+/// not include the active log file.
+pub const MAX_FILE_SINK_RETAINED_FILES: usize = 9;
+
 /// Operational logging configuration for [`LoggingRuntime::configure`](super::LoggingRuntime::configure).
 ///
 /// `level` is the process-wide **minimum severity**: call sites may emit any level, but records
@@ -219,6 +226,8 @@ pub struct FileLogSinkConfig {
     /// Maximum pending asynchronous queue entries for this file sink. Must be greater than 0 and
     /// at most [`MAX_FILE_SINK_QUEUE_ENTRIES`].
     pub queue_capacity: usize,
+    /// Optional size-based rotation and retention settings.
+    pub rotation: Option<FileLogRotationConfig>,
 }
 
 impl Default for FileLogSinkConfig {
@@ -228,10 +237,56 @@ impl Default for FileLogSinkConfig {
             level: LogLevel::Info,
             format: LogFormat::Jsonl,
             queue_capacity: DEFAULT_FILE_SINK_QUEUE_ENTRIES,
+            rotation: None,
         }
     }
 }
 
+/// Size-based rotation settings for a file log sink.
+///
+/// `retained_files` counts previous log files and excludes the active file.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct FileLogRotationConfig {
+    max_file_size_bytes: u64,
+    retained_files: usize,
+}
+
+impl FileLogRotationConfig {
+    /// Creates validated size-based rotation settings.
+    pub fn new(max_file_size_bytes: u64, retained_files: usize) -> Result<Self> {
+        if max_file_size_bytes == 0 {
+            return Err(FlowError::InvalidArgument(
+                "logging sink max_file_size_bytes must be greater than 0".into(),
+            ));
+        }
+        if retained_files == 0 {
+            return Err(FlowError::InvalidArgument(
+                "logging sink retained_files must be greater than 0".into(),
+            ));
+        }
+        if retained_files > MAX_FILE_SINK_RETAINED_FILES {
+            return Err(FlowError::InvalidArgument(format!(
+                "logging sink retained_files {retained_files} exceeds maximum \
+                 {MAX_FILE_SINK_RETAINED_FILES} backup files per sink"
+            )));
+        }
+        Ok(Self {
+            max_file_size_bytes,
+            retained_files,
+        })
+    }
+
+    /// Maximum active file size before the next record triggers rotation.
+    pub fn max_file_size_bytes(self) -> u64 {
+        self.max_file_size_bytes
+    }
+
+    /// Number of previous log files retained in addition to the active file.
+    pub fn retained_files(self) -> usize {
+        self.retained_files
+    }
+}
+
 #[derive(Debug, Deserialize)]
 struct LoggingDocument {
     logging: Option<RawLoggingConfig>,
@@ -277,6 +332,8 @@ struct RawFileLogSinkConfig {
     level: Option<String>,
     format: Option<String>,
     queue_capacity: Option<usize>,
+    max_file_size_bytes: Option<u64>,
+    retained_files: Option<usize>,
 }
 
 impl RawFileLogSinkConfig {
@@ -318,11 +375,26 @@ impl RawFileLogSinkConfig {
             None => DEFAULT_FILE_SINK_QUEUE_ENTRIES,
         };
 
+        let rotation = match (self.max_file_size_bytes, self.retained_files) {
+            (None, None) => None,
+            (Some(max_file_size_bytes), Some(retained_files)) => Some(FileLogRotationConfig::new(
+                max_file_size_bytes,
+                retained_files,
+            )?),
+            _ => {
+                return Err(FlowError::InvalidArgument(
+                    "logging sink max_file_size_bytes and retained_files must be configured \
+                     together"
+                        .into(),
+                ));
+            }
+        };
         Ok(LogSinkConfig::File(FileLogSinkConfig {
             path,
             level,
             format,
             queue_capacity,
+            rotation,
         }))
     }
 }

crates/core/src/logging/mod.rs

@@ -8,6 +8,7 @@
 
 mod config;
 mod format;
+mod rotation;
 mod sink;
 
 use std::io::{self, Write};
@@ -21,8 +22,9 @@ use uuid::Uuid;
 use crate::error::{FlowError, Result};
 
 pub use config::{
-    DEFAULT_FILE_FLUSH_INTERVAL_MILLIS, DEFAULT_FILE_SINK_QUEUE_ENTRIES, FileLogSinkConfig,
-    LogFormat, LogLevel, LogSinkConfig, LoggingConfig, MAX_FILE_SINK_QUEUE_ENTRIES,
+    DEFAULT_FILE_FLUSH_INTERVAL_MILLIS, DEFAULT_FILE_SINK_QUEUE_ENTRIES, FileLogRotationConfig,
+    FileLogSinkConfig, LogFormat, LogLevel, LogSinkConfig, LoggingConfig,
+    MAX_FILE_SINK_QUEUE_ENTRIES, MAX_FILE_SINK_RETAINED_FILES,
 };
 pub(crate) use sink::build_logger;
 use sink::log_level_filter;

crates/core/src/logging/rotation.rs

@@ -0,0 +1,148 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+//! Size-based file rotation for operational log sinks.
+
+use std::fs::{self, File, OpenOptions};
+use std::io::{self, BufWriter, Write};
+use std::path::{Path, PathBuf};
+
+pub(crate) struct SizeRotatingFileWriter {
+    base_path: PathBuf,
+    file: Option<BufWriter<File>>,
+    current_size: u64,
+    max_file_size_bytes: u64,
+    retained_files: usize,
+}
+
+impl SizeRotatingFileWriter {
+    pub(crate) fn new(
+        base_path: PathBuf,
+        max_file_size_bytes: u64,
+        retained_files: usize,
+    ) -> io::Result<Self> {
+        create_parent_directory(&base_path)?;
+        let file = open_active_file(&base_path, false)?;
+        let current_size = file.get_ref().metadata()?.len();
+
+        Ok(Self {
+            base_path,
+            file: Some(file),
+            current_size,
+            max_file_size_bytes,
+            retained_files,
+        })
+    }
+
+    fn rotate_if_needed(&mut self, incoming_bytes: usize) -> io::Result<()> {
+        if self.current_size == 0
+            || self.current_size.saturating_add(incoming_bytes as u64) <= self.max_file_size_bytes
+        {
+            return Ok(());
+        }
+
+        self.rotate()
+    }
+
+    fn rotate(&mut self) -> io::Result<()> {
+        let mut file = self
+            .file
+            .take()
+            .ok_or_else(|| io::Error::other("rotating log file is not open"))?;
+        if let Err(error) = file.flush() {
+            self.file = Some(file);
+            return Err(error);
+        }
+        drop(file);
+
+        if let Err(error) = rotate_files(&self.base_path, self.retained_files) {
+            return match self.reopen_after_failed_rotation() {
+                Ok(()) => Err(error),
+                Err(reopen_error) => Err(io::Error::new(
+                    reopen_error.kind(),
+                    format!(
+                        "log rotation failed: {error}; failed to reopen active log file: \
+                         {reopen_error}"
+                    ),
+                )),
+            };
+        }
+
+        self.file = Some(open_active_file(&self.base_path, true)?);
+        self.current_size = 0;
+        Ok(())
+    }
+
+    fn reopen_after_failed_rotation(&mut self) -> io::Result<()> {
+        let file = open_active_file(&self.base_path, false)?;
+        self.current_size = file.get_ref().metadata()?.len();
+        self.file = Some(file);
+        Ok(())
+    }
+}
+
+impl Write for SizeRotatingFileWriter {
+    fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
+        self.rotate_if_needed(buffer.len())?;
+        self.file
+            .as_mut()
+            .ok_or_else(|| io::Error::other("rotating log file is not open"))?
+            .write_all(buffer)?;
+        self.current_size = self.current_size.saturating_add(buffer.len() as u64);
+        Ok(buffer.len())
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.file
+            .as_mut()
+            .ok_or_else(|| io::Error::other("rotating log file is not open"))?
+            .flush()
+    }
+}
+
+fn create_parent_directory(path: &Path) -> io::Result<()> {
+    if let Some(parent) = path.parent()
+        && !parent.as_os_str().is_empty()
+    {
+        fs::create_dir_all(parent)?;
+    }
+    Ok(())
+}
+
+fn open_active_file(path: &Path, truncate: bool) -> io::Result<BufWriter<File>> {
+    let file = OpenOptions::new()
+        .create(true)
+        .write(true)
+        .append(!truncate)
+        .truncate(truncate)
+        .open(path)?;
+    Ok(BufWriter::new(file))
+}
+
+fn rotate_files(base_path: &Path, retained_files: usize) -> io::Result<()> {
+    for index in (1..=retained_files).rev() {
+        let source = if index == 1 {
+            base_path.to_path_buf()
+        } else {
+            rotated_log_path(base_path, index - 1)
+        };
+        if !source.exists() {
+            continue;
+        }
+
+        let destination = rotated_log_path(base_path, index);
+        fs::rename(source, destination)?;
+    }
+    Ok(())
+}
+
+pub(crate) fn rotated_log_path(base_path: &Path, index: usize) -> PathBuf {
+    let stem = base_path.file_stem().unwrap_or(base_path.as_os_str());
+    let mut file_name = stem.to_os_string();
+    file_name.push(format!("_{index}"));
+    if let Some(extension) = base_path.extension() {
+        file_name.push(".");
+        file_name.push(extension);
+    }
+    base_path.with_file_name(file_name)
+}