: logging: move some code under test (#1677)

Summary:

it turns out that the code this diff moves is in fact a test utility. move it under `test` so it is clarified that it currently serves no other purpose.

Reviewed By: amirafzali

Differential Revision: D85604301

Author: shayne-fletcher

hyperactor_mesh/src/logging.rs

@@ -50,14 +50,10 @@ use hyperactor_telemetry::log_file_path;
 use notify::Watcher;
 use serde::Deserialize;
 use serde::Serialize;
-use tokio::fs;
 use tokio::io;
 use tokio::io::AsyncRead;
 use tokio::io::AsyncReadExt;
-use tokio::io::AsyncSeek;
-use tokio::io::AsyncSeekExt;
 use tokio::io::AsyncWriteExt;
-use tokio::io::SeekFrom;
 use tokio::sync::Mutex;
 use tokio::sync::Notify;
 use tokio::sync::RwLock;
@@ -947,113 +943,6 @@ impl StreamFwder {
     }
 }
 
-/// Result of processing file content
-#[derive(Debug)]
-struct FileProcessingResult {
-    /// Complete lines found during processing
-    lines: Vec<Vec<u8>>,
-    /// Updated position in the file after processing
-    new_position: u64,
-    /// Any remaining incomplete line data, buffered for subsequent reads
-    incomplete_line_buffer: Vec<u8>,
-}
-
-/// Process new file content from a given position, extracting complete lines
-/// This function is extracted to enable easier unit testing without file system dependencies
-async fn process_file_content<R: AsyncRead + AsyncSeek + Unpin>(
-    reader: &mut R,
-    current_position: u64,
-    file_size: u64,
-    existing_line_buffer: Vec<u8>,
-    max_buffer_size: usize,
-) -> Result<FileProcessingResult> {
-    // If position equals file size, we're at the end
-    if current_position == file_size {
-        return Ok(FileProcessingResult {
-            lines: Vec::new(),
-            new_position: current_position,
-            incomplete_line_buffer: existing_line_buffer,
-        });
-    }
-
-    // Handle potential file truncation/rotation
-    let actual_position = if current_position > file_size {
-        tracing::warn!(
-            "File appears to have been truncated (position {} > file size {}), resetting to start",
-            current_position,
-            file_size
-        );
-        reader.seek(SeekFrom::Start(0)).await?;
-        0
-    } else {
-        // current_position < file_size
-        reader.seek(SeekFrom::Start(current_position)).await?;
-        current_position
-    };
-
-    let mut buf = vec![0u8; 128 * 1024];
-    let mut line_buffer = existing_line_buffer;
-    let mut lines = Vec::with_capacity(max_buffer_size);
-    let mut processed_bytes = 0u64;
-
-    loop {
-        let bytes_read = reader.read(&mut buf).await?;
-        if bytes_read == 0 {
-            break;
-        }
-
-        let chunk = &buf[..bytes_read];
-
-        let mut start = 0;
-        while let Some(newline_pos) = chunk[start..].iter().position(|&b| b == b'\n') {
-            let absolute_pos = start + newline_pos;
-
-            line_buffer.extend_from_slice(&chunk[start..absolute_pos]);
-
-            if !line_buffer.is_empty() {
-                if line_buffer.len() > MAX_LINE_SIZE {
-                    line_buffer.truncate(MAX_LINE_SIZE);
-                    line_buffer.extend_from_slice(b"... [TRUNCATED]");
-                }
-
-                let line_data = std::mem::replace(&mut line_buffer, Vec::with_capacity(2048));
-                lines.push(line_data);
-            }
-
-            start = absolute_pos + 1;
-
-            // Check if we've reached the max buffer size after adding each line
-            if lines.len() >= max_buffer_size {
-                // We've processed up to and including the current newline
-                // The new position is where we should start reading next time
-                let new_position = actual_position + processed_bytes + start as u64;
-
-                // Don't save remaining data - we'll re-read it from the new position
-                return Ok(FileProcessingResult {
-                    lines,
-                    new_position,
-                    incomplete_line_buffer: Vec::new(),
-                });
-            }
-        }
-
-        // Only add bytes to processed_bytes if we've fully processed this chunk
-        processed_bytes += bytes_read as u64;
-
-        if start < chunk.len() {
-            line_buffer.extend_from_slice(&chunk[start..]);
-        }
-    }
-
-    let new_position = actual_position + processed_bytes;
-
-    Ok(FileProcessingResult {
-        lines,
-        new_position,
-        incomplete_line_buffer: line_buffer,
-    })
-}
-
 /// Messages that can be sent to the LogForwarder
 #[derive(
     Debug,
@@ -1554,6 +1443,7 @@ pub mod test_tap {
 
 #[cfg(test)]
 mod tests {
+
     use std::sync::Arc;
     use std::sync::Mutex;
 
@@ -1566,11 +1456,121 @@ mod tests {
     use hyperactor::mailbox::DialMailboxRouter;
     use hyperactor::mailbox::MailboxServer;
     use hyperactor::proc::Proc;
+    use tokio::io::AsyncSeek;
+    use tokio::io::AsyncSeekExt;
     use tokio::io::AsyncWriteExt;
+    use tokio::io::SeekFrom;
     use tokio::sync::mpsc;
 
     use super::*;
 
+    /// Result of processing file content
+    #[derive(Debug)]
+    struct FileProcessingResult {
+        /// Complete lines found during processing
+        lines: Vec<Vec<u8>>,
+        /// Updated position in the file after processing
+        new_position: u64,
+        /// Any remaining incomplete line data, buffered for subsequent reads
+        incomplete_line_buffer: Vec<u8>,
+    }
+
+    /// Process new file content from a given position, extracting complete lines
+    /// This function is extracted to enable easier unit testing without file system dependencies
+    async fn process_file_content<R: AsyncRead + AsyncSeek + Unpin>(
+        reader: &mut R,
+        current_position: u64,
+        file_size: u64,
+        existing_line_buffer: Vec<u8>,
+        max_buffer_size: usize,
+    ) -> Result<FileProcessingResult> {
+        // If position equals file size, we're at the end
+        if current_position == file_size {
+            return Ok(FileProcessingResult {
+                lines: Vec::new(),
+                new_position: current_position,
+                incomplete_line_buffer: existing_line_buffer,
+            });
+        }
+
+        // Handle potential file truncation/rotation
+        let actual_position = if current_position > file_size {
+            tracing::warn!(
+                "File appears to have been truncated (position {} > file size {}), resetting to start",
+                current_position,
+                file_size
+            );
+            reader.seek(SeekFrom::Start(0)).await?;
+            0
+        } else {
+            // current_position < file_size
+            reader.seek(SeekFrom::Start(current_position)).await?;
+            current_position
+        };
+
+        let mut buf = vec![0u8; 128 * 1024];
+        let mut line_buffer = existing_line_buffer;
+        let mut lines = Vec::with_capacity(max_buffer_size);
+        let mut processed_bytes = 0u64;
+
+        loop {
+            let bytes_read = reader.read(&mut buf).await?;
+            if bytes_read == 0 {
+                break;
+            }
+
+            let chunk = &buf[..bytes_read];
+
+            let mut start = 0;
+            while let Some(newline_pos) = chunk[start..].iter().position(|&b| b == b'\n') {
+                let absolute_pos = start + newline_pos;
+
+                line_buffer.extend_from_slice(&chunk[start..absolute_pos]);
+
+                if !line_buffer.is_empty() {
+                    if line_buffer.len() > MAX_LINE_SIZE {
+                        line_buffer.truncate(MAX_LINE_SIZE);
+                        line_buffer.extend_from_slice(b"... [TRUNCATED]");
+                    }
+
+                    let line_data = std::mem::replace(&mut line_buffer, Vec::with_capacity(2048));
+                    lines.push(line_data);
+                }
+
+                start = absolute_pos + 1;
+
+                // Check if we've reached the max buffer size after adding each line
+                if lines.len() >= max_buffer_size {
+                    // We've processed up to and including the current newline
+                    // The new position is where we should start reading next time
+                    let new_position = actual_position + processed_bytes + start as u64;
+
+                    // Don't save remaining data - we'll re-read it from the new position
+                    return Ok(FileProcessingResult {
+                        lines,
+                        new_position,
+                        incomplete_line_buffer: Vec::new(),
+                    });
+                }
+            }
+
+            // Only add bytes to processed_bytes if we've fully processed this chunk
+            processed_bytes += bytes_read as u64;
+
+            if start < chunk.len() {
+                line_buffer.extend_from_slice(&chunk[start..]);
+            }
+        }
+
+        let new_position = actual_position + processed_bytes;
+
+        Ok(FileProcessingResult {
+            lines,
+            new_position,
+            incomplete_line_buffer: line_buffer,
+        })
+    }
+
     #[tokio::test]
     async fn test_forwarding_log_to_client() {
         // Setup the basics