[#265] feat: 쿼리 단위 메모리 제한 구현

* feat: add query-level memory limit (OOM killer) (#265)

* test(init): include max_query_memory_bytes in config mock expectation

* fix: apply CodeRabbit review - task-local tracker, FileSystem metadata, decode budget

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>>;
+    /// 파일의 크기(bytes)를 반환합니다. (#265)
+    /// `read_segment_rows`가 파일 전체를 메모리로 읽기 전에 예산을 확보하는 데 사용합니다.
+    async fn metadata(&self, path: &Path) -> io::Result<u64>;
 }
 
 pub struct RealFileSystem;
@@ -45,4 +48,9 @@ impl FileSystem for RealFileSystem {
     async fn read(&self, path: &Path) -> io::Result<Vec<u8>> {
         tokio::fs::read(path).await
     }
+
+    async fn metadata(&self, path: &Path) -> io::Result<u64> {
+        let metadata = tokio::fs::metadata(path).await?;
+        Ok(metadata.len())
+    }
 }

src/config/launch_config.rs

@@ -17,6 +17,20 @@ pub struct LaunchConfig {
     pub wal_directory: String,
     pub wal_segment_size: u32,
     pub wal_extension: String,
+
+    /// 쿼리 수준 메모리 상한 (bytes) (#265).
+    /// SELECT/UPDATE/DELETE 실행 중 추정 메모리가 이를 넘으면
+    /// 에러를 반환해 쿼리를 강제로 중단합니다.
+    /// `0`이면 비활성 (OOM killer 작동 안 함).
+    /// 기존 설정 파일에 이 필드가 없어도 동작하도록 serde default 적용.
+    #[serde(default = "default_max_query_memory_bytes")]
+    pub max_query_memory_bytes: u64,
+}
+
+/// 새 설정 필드가 사용자 TOML에서 빠져 있을 때의 기본값 (#265).
+/// 이 관리가 없으면 기존 rrdb.config를 가진 사용자는 업데이트 시 설정 로드가 실패합니다.
+fn default_max_query_memory_bytes() -> u64 {
+    128 * 1024 * 1024
 }
 
 #[allow(clippy::derivable_impls)]
@@ -40,6 +54,8 @@ impl std::default::Default for LaunchConfig {
                 .to_string(),
             wal_segment_size: 1024 * 1024 * 16, // 16MB 세그먼트 사이즈
             wal_extension: DEFAULT_WAL_EXTENSION.to_string(),
+            // 기본 128MB. 사용자가 메모리 퍼센트를 조절하면 TOML에서 설정.
+            max_query_memory_bytes: 128 * 1024 * 1024,
         }
     }
 }

src/engine/actions/dml/scan.rs

@@ -10,7 +10,7 @@ use crate::engine::ast::dml::plan::select::scan::IndexScanPlan;
 use crate::engine::ast::types::TableName;
 use crate::engine::encoder::schema_encoder::StorageEncoder;
 use crate::engine::row_buffer::{ROW_FRAME_LIVE, RowBufferWrite, encode_live_row_frames};
-use crate::engine::schema::row::TableDataRow;
+use crate::engine::schema::row::{ESTIMATED_FIELD_OVERHEAD, TableDataRow};
 use crate::errors;
 use crate::errors::execute_error::ExecuteError;
 
@@ -60,6 +60,17 @@ impl DBEngine {
             }
         };
 
+        // 메모리 예산 추적 (#265): 전체 세그먼트 로드 후
+        // 자신의 반환 결과(Vec<(RowLocation, TableDataRow)>) 크기를 reserve.
+        if let Some(tracker) = self.query_memory().await {
+            let bytes: u64 = rows
+                .iter()
+                .filter_map(|row| row.as_ref())
+                .map(|row| row.estimated_bytes() + ESTIMATED_FIELD_OVERHEAD)
+                .sum();
+            tracker.reserve(bytes)?;
+        }
+
         let live = rows
             .into_iter()
             .enumerate()
@@ -307,6 +318,20 @@ impl DBEngine {
         &self,
         segment_path: &Path,
     ) -> errors::Result<Vec<Option<TableDataRow>>> {
+        // 파일 전체를 메모리로 읽기 전에 예산을 미리 확보 (#265).
+        // `tokio::fs::read`가 들어올 파일 크만큼을 추적해
+        // 상한 초과 시 실제 할당 이전에 거부합니다.
+        // `self.file_system.metadata()`를 사용해 주입된 mock으로
+        // metadata 오류와 메모리 예약을 검증할 수 있게 합니다 (CodeRabbit).
+        if let Some(tracker) = self.query_memory().await {
+            let file_size = match self.file_system.metadata(segment_path).await {
+                Ok(size) => size,
+                Err(error) if error.kind() == IOErrorKind::NotFound => 0,
+                Err(error) => return Err(ExecuteError::wrap(error.to_string())),
+            };
+            tracker.reserve(file_size)?;
+        }
+
         let content = match tokio::fs::read(segment_path).await {
             Ok(content) => content,
             Err(error) if error.kind() == IOErrorKind::NotFound => return Ok(Vec::new()),
@@ -343,6 +368,13 @@ impl DBEngine {
             if tombstoned {
                 rows.push(None);
             } else {
+                // 역직렬화 전에 프레임 디코드 예산을 추가로 reserve (#265).
+                // 파일 크기는 원본 바이트인데 디코드 결과는 틀이 크게
+                // 날 수 있으므로, 보수적으로 프레임 크기만큼을
+                // 추가 예약하여 상한 초과 시 디코드 전에 거부합니다.
+                if let Some(tracker) = self.query_memory().await {
+                    tracker.reserve(frame_len as u64)?;
+                }
                 let row = encoder
                     .decode::<TableDataRow>(&content[offset..offset + frame_len])
                     .map_err(|error| {
@@ -596,7 +628,13 @@ impl DBEngine {
             })?;
 
             match all_rows.get(row_index) {
-                Some(Some(row)) => result.push((RowLocation { row_index }, row.clone())),
+                Some(Some(row)) => {
+                    // 메모리 예산 추적 (#265): 결과 행이 앱드될 때마다 크기를 reserve.
+                    if let Some(tracker) = self.query_memory().await {
+                        tracker.reserve(row.estimated_bytes() + ESTIMATED_FIELD_OVERHEAD)?;
+                    }
+                    result.push((RowLocation { row_index }, row.clone()));
+                }
                 Some(None) | None => {
                     return Err(ExecuteError::wrap(format!(
                         "index '{}' is out of sync with table data; drop and recreate the index",

src/engine/actions/dml/select.rs

@@ -634,12 +634,25 @@ mod tests {
     use crate::engine::{DBEngine, SharedWALManager};
 
     async fn build_test_engine(test_name: &str) -> (DBEngine, SharedWALManager) {
+        build_test_engine_with_memory_limit(test_name, 128 * 1024 * 1024).await
+    }
+
+    async fn build_test_engine_with_memory_limit(
+        test_name: &str,
+        max_query_memory_bytes: u64,
+    ) -> (DBEngine, SharedWALManager) {
         let base_path = PathBuf::from("target").join(test_name);
-        if base_path.exists() {
-            tokio::fs::remove_dir_all(&base_path).await.unwrap();
+        // 동기 `Path::exists()`는 metadata 오류를 `false`로 처리해 정리를 건너뛸 수
+        // 있으므로, 비동기 `tokio::fs::metadata`로 NotFound만 무시하고 다른 오류는
+        // 테스트를 실패시킵니다 (CodeRabbit).
+        match tokio::fs::metadata(&base_path).await {
+            Ok(_) => tokio::fs::remove_dir_all(&base_path).await.unwrap(),
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+            Err(error) => panic!("metadata check failed for {base_path:?}: {error}"),
         }
 
-        let config = LaunchConfig::default_for_base_path(&base_path);
+        let mut config = LaunchConfig::default_for_base_path(&base_path);
+        config.max_query_memory_bytes = max_query_memory_bytes;
         tokio::fs::create_dir_all(&config.data_directory)
             .await
             .unwrap();
@@ -821,7 +834,11 @@ mod tests {
                 .await
                 .unwrap_or_else(|error| panic!("{sql} failed: {error}"));
 
-            assert_eq!(result.rows.len(), expected, "unexpected row count for {sql}");
+            assert_eq!(
+                result.rows.len(),
+                expected,
+                "unexpected row count for {sql}"
+            );
         }
     }
 
@@ -846,7 +863,195 @@ mod tests {
             ("select id from key_value offset 1;", 2),
         ] {
             let result = execute_sql(&engine, wal.clone(), sql).await.unwrap();
-            assert_eq!(result.rows.len(), expected, "unexpected row count for {sql}");
+            assert_eq!(
+                result.rows.len(),
+                expected,
+                "unexpected row count for {sql}"
+            );
         }
     }
+
+    /// 트래킹 상한을 넘기는 크기의 문자열 값을 가진 행을 삽입합니다.
+    /// `size_bytes` 이상의 문자열이 들어가도록 패딩합니다.
+    async fn insert_big_string(engine: &DBEngine, wal: SharedWALManager, size_bytes: usize) {
+        let value = "x".repeat(size_bytes);
+        engine
+            .insert(
+                InsertQuery::builder()
+                    .set_into_table(TableName::new(
+                        Some("rrdb".to_string()),
+                        "key_value".to_string(),
+                    ))
+                    .set_columns(vec!["id".to_string()])
+                    .set_values(vec![InsertValue {
+                        list: vec![Some(SQLExpression::String(value))],
+                    }])
+                    .build(),
+                wal,
+            )
+            .await
+            .unwrap();
+    }
+
+    /// 메모리 예산을 매우 낮게 설정하면, SELECT가
+    /// 행을 메모리로 로드하는 순간 강제 중단됩니다 (#265).
+    #[tokio::test]
+    async fn select_over_memory_limit_is_killed() {
+        let (engine, wal) = build_test_engine_with_memory_limit("test_select_oom_kill", 1024).await;
+
+        execute_sql(&engine, wal.clone(), "create database rrdb;")
+            .await
+            .unwrap();
+        execute_sql(
+            &engine,
+            wal.clone(),
+            "create table key_value (id varchar(65536));",
+        )
+        .await
+        .unwrap();
+
+        // 행 하나만 로드해도 예산(1KB)를 넘는 크기(4KB 문자열).
+        insert_big_string(&engine, wal.clone(), 4096).await;
+
+        let result = execute_sql(&engine, wal, "select id from key_value;")
+            .await
+            .unwrap_err();
+
+        let message = result.to_string();
+        assert!(
+            message.contains("query memory limit exceeded"),
+            "expected memory limit error, got: {message}"
+        );
+    }
+
+    /// `max_query_memory_bytes = 0`이면 OOM killer가 비활성입니다 (#265).
+    /// 크기의 행을 로드해도 에러 없이 정상 동작합니다.
+    #[tokio::test]
+    async fn select_with_disabled_memory_limit_succeeds() {
+        let (engine, wal) =
+            build_test_engine_with_memory_limit("test_select_oom_disabled", 0).await;
+
+        execute_sql(&engine, wal.clone(), "create database rrdb;")
+            .await
+            .unwrap();
+        execute_sql(
+            &engine,
+            wal.clone(),
+            "create table key_value (id varchar(65536));",
+        )
+        .await
+        .unwrap();
+
+        insert_big_string(&engine, wal.clone(), 4096).await;
+
+        let result = execute_sql(&engine, wal, "select id from key_value;")
+            .await
+            .unwrap();
+
+        assert_eq!(result.rows.len(), 1);
+    }
+
+    /// 예산을 넘지 않는 손상된 실행은 정상 동작해야 합니다 (#265).
+    #[tokio::test]
+    async fn select_within_memory_limit_succeeds() {
+        let (engine, wal) =
+            build_test_engine_with_memory_limit("test_select_oom_within", 1024 * 1024).await;
+
+        execute_sql(&engine, wal.clone(), "create database rrdb;")
+            .await
+            .unwrap();
+        execute_sql(
+            &engine,
+            wal.clone(),
+            "create table key_value (id varchar(65536));",
+        )
+        .await
+        .unwrap();
+
+        insert_big_string(&engine, wal.clone(), 4096).await;
+
+        let result = execute_sql(&engine, wal, "select id from key_value;")
+            .await
+            .unwrap();
+
+        assert_eq!(result.rows.len(), 1);
+    }
+
+    /// 지정한 테이블에 큰 문자열 행을 삽입합니다.
+    async fn insert_big_string_into(
+        engine: &DBEngine,
+        wal: SharedWALManager,
+        table: &str,
+        size_bytes: usize,
+    ) {
+        let value = "x".repeat(size_bytes);
+        engine
+            .insert(
+                InsertQuery::builder()
+                    .set_into_table(TableName::new(Some("rrdb".to_string()), table.to_string()))
+                    .set_columns(vec!["id".to_string()])
+                    .set_values(vec![InsertValue {
+                        list: vec![Some(SQLExpression::String(value))],
+                    }])
+                    .build(),
+                wal,
+            )
+            .await
+            .unwrap();
+    }
+
+    /// 동시에 실행되는 두 쿼리는 서로의 메모리 예산을 방해하지 않아야 합니다 (#265).
+    ///
+    /// CodeRabbit #1: 이전 구현(공유 RwLock 슬롯)은 두 `process_query`가 동시에
+    /// 실행되면 한쪽이 다른 쪽의 tracker를 덮어쓰거나, 한쪽이 스캔을 마치기 전에
+    /// 슬롯을 클리어해서 나머지가 예산 없이 실행될 수 있었습니다.
+    ///
+    /// task-local로 전환한 후에는 각 쿼리가 자기 tracker를 가지므로,
+    /// 큰 쿼리(예산 초과 → 에러)와 작은 쿼리(예산 내 → 성공)를 동시에 실행해도
+    /// 서로의 결과에 영향을 주지 않아야 합니다.
+    #[tokio::test]
+    async fn concurrent_queries_do_not_interfere_with_each_others_budget() {
+        let (engine, wal) =
+            build_test_engine_with_memory_limit("test_select_oom_concurrent", 2048).await;
+
+        execute_sql(&engine, wal.clone(), "create database rrdb;")
+            .await
+            .unwrap();
+        execute_sql(
+            &engine,
+            wal.clone(),
+            "create table small_t (id varchar(65536));",
+        )
+        .await
+        .unwrap();
+        execute_sql(
+            &engine,
+            wal.clone(),
+            "create table big_t (id varchar(65536));",
+        )
+        .await
+        .unwrap();
+
+        // small_t: 예산(2KB) 안에 들어오는 작은 값
+        // big_t: 예산을 넘기는 큰 값 (4KB 문자열)
+        insert_big_string_into(&engine, wal.clone(), "small_t", 256).await;
+        insert_big_string_into(&engine, wal.clone(), "big_t", 4096).await;
+
+        // 동시 실행: big 쿼리는 실패, small 쿼리는 성공해야 함
+        let (big_result, small_result) = tokio::join!(
+            execute_sql(&engine, wal.clone(), "select id from big_t;"),
+            execute_sql(&engine, wal.clone(), "select id from small_t;"),
+        );
+
+        let big_error = big_result.unwrap_err();
+        assert!(
+            big_error
+                .to_string()
+                .contains("query memory limit exceeded"),
+            "big query should be killed, got: {big_error}"
+        );
+
+        let small_rows = small_result.unwrap();
+        assert_eq!(small_rows.rows.len(), 1, "small query should still succeed");
+    }
 }

src/engine/initialize.rs

@@ -262,6 +262,7 @@ wal_enabled = true
 wal_directory = "/var/lib/rrdb/wal"
 wal_segment_size = 16777216
 wal_extension = "log"
+max_query_memory_bytes = 134217728
 "##;
 
         struct TestCase {