[#226] row append 경로의 storage lock 누락 수정

append_table_rows_with_buffer_limit가 row_storage_lock 없이 row_buffer_pool만 갱신해 update/delete의 snapshot 기반 replace_rows와 경합할 수 있었다. update/delete가 snapshot을 읽은 뒤 replace_rows로 pending append를 비우는 사이 INSERT가 끼면 새 행이 snapshot에도 pending에도 남지 않아 유실될 수 있었다.

append 경로도 update/delete와 동일하게 row_storage_lock을 잡도록 바꾸고, 버퍼 한도 초과 시에는 이미 보유한 락 안에서 flush_row_buffers_locked(false)를 호출해 재진입 deadlock을 피했다. append가 row_storage_lock을 기다리는 회귀 테스트도 추가했다.

Author: myyrakle

src/engine/actions/dml/scan.rs

@@ -69,6 +69,7 @@ impl DBEngine {
             return Ok(());
         }
 
+        let _guard = self.row_storage_lock.lock().await;
         let segment_path = self.row_segment_path(table_name)?;
         let frame = encode_row_frames(rows)?;
 
@@ -80,7 +81,7 @@ impl DBEngine {
         };
 
         if buffered_bytes >= buffer_limit_bytes {
-            self.flush_row_buffers().await?;
+            self.flush_row_buffers_locked(false).await?;
         }
 
         Ok(())
@@ -204,6 +205,7 @@ impl DBEngine {
         Ok(rows)
     }
 
+    #[cfg(test)]
     pub(crate) async fn flush_row_buffers(&self) -> errors::Result<()> {
         let _guard = self.row_storage_lock.lock().await;
         self.flush_row_buffers_locked(false).await
@@ -333,6 +335,7 @@ impl DBEngine {
 #[cfg(test)]
 mod tests {
     use std::path::PathBuf;
+    use std::sync::Arc;
 
     use crate::config::launch_config::LaunchConfig;
     use crate::engine::DBEngine;
@@ -610,4 +613,42 @@ mod tests {
 
         assert!(engine.row_buffer_pool.lock().await.is_unsynced_empty());
     }
+
+    #[tokio::test]
+    async fn append_table_rows_waits_for_row_storage_lock() {
+        let base_path = PathBuf::from(format!(
+            "target/test_row_segments/append_waits_for_storage_lock_{}",
+            std::process::id()
+        ));
+        if base_path.exists() {
+            tokio::fs::remove_dir_all(&base_path).await.unwrap();
+        }
+
+        let config = LaunchConfig::default_for_base_path(&base_path);
+        let table_name = TableName::new(Some("rrdb".to_string()), "users".to_string());
+        let rows_path = PathBuf::from(&config.data_directory)
+            .join("rrdb")
+            .join("tables")
+            .join("users")
+            .join("rows");
+        tokio::fs::create_dir_all(&rows_path).await.unwrap();
+
+        let engine = Arc::new(DBEngine::new(config));
+        let row = TableDataRow {
+            fields: vec![TableDataField {
+                table_name: table_name.clone(),
+                column_name: "id".to_string(),
+                data: TableDataFieldType::Integer(1),
+            }],
+        };
+        let _guard = engine.row_storage_lock.lock().await;
+        let append_task = {
+            let engine = engine.clone();
+            tokio::spawn(async move { engine.append_table_rows(&table_name, &[row]).await })
+        };
+
+        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+
+        assert!(!append_task.is_finished());
+    }
 }