Configuration used: CodeRabbit UI

Review profile: ASSERTIVE

Plan: Pro

Learnt from: junhaoliao
Repo: y-scope/clp PR: 0
File: :0-0
Timestamp: 2025-10-22T21:02:31.113Z
Learning: Repository y-scope/clp: Maintain deterministic CI/builds for Rust; add a check to verify Cargo.lock is in sync with Cargo.toml without updating dependencies (non-mutating verification in clp-rust-checks workflow).

Learnt from: hoophalab
Repo: y-scope/clp PR: 1535
File: components/clp-rust-utils/src/clp_config/package/config.rs:47-61
Timestamp: 2025-11-03T16:17:40.223Z
Learning: In the y-scope/clp repository, the `ApiServer` struct in `components/clp-rust-utils/src/clp_config/package/config.rs` is a Rust-native configuration type and does not mirror any Python code, unlike other structs in the same file (Config, Database, ResultsCache, Package) which are mirrors of Python definitions.

Learnt from: junhaoliao
Repo: y-scope/clp PR: 1575
File: components/clp-py-utils/clp_py_utils/clp_config.py:602-607
Timestamp: 2025-11-10T05:19:56.600Z
Learning: In the y-scope/clp repository, the `ApiServer` class in `components/clp-py-utils/clp_py_utils/clp_config.py` does not need a `transform_for_container()` method because no other containerized service depends on the API server - it's only accessed from the host, so no docker-network communication is expected.

components/api-server/src/client.rs (1)

183-192: File ordering is non-deterministic.

tokio::fs::read_dir does not guarantee directory entry order. If query results are split across multiple files and order matters for result correctness or user experience, results may be returned in unpredictable order across different platforms or filesystems.

This was flagged in past review comments. If ordering matters, apply this approach:

             let stream = stream! {
                 let dir_path = dir_path;
-                let mut entries = tokio::fs::read_dir(dir_path).await?;
-                while let Some(entry) = entries.next_entry().await? {
-                    let file_path = entry.path();
+                let mut entries = tokio::fs::read_dir(&dir_path).await?;
+                let mut paths = Vec::new();
+                while let Some(entry) = entries.next_entry().await? {
+                    paths.push(entry.path());
+                }
+                // Sort by filename to ensure deterministic order
+                paths.sort();
+                
+                for file_path in paths {
                     let reader = std::fs::File::open(file_path)?;
                     let mut deserializer = rmp_serde::Deserializer::new(reader);
                     while let Ok(e) = Deserialize::deserialize(&mut deserializer) {
                         let e: (i64, String, String, String, i64) = e;
                         yield Ok(e.1);
                     }
                 }
             };

Configuration used: CodeRabbit UI

Review profile: ASSERTIVE

Plan: Pro

Learnt from: junhaoliao
Repo: y-scope/clp PR: 0
File: :0-0
Timestamp: 2025-10-22T21:01:31.391Z
Learning: In y-scope/clp, for the clp-rust-checks workflow (GitHub Actions) and the Taskfile target deps:lock:check-rust, we should verify Rust Cargo.lock is in sync with Cargo.toml using a non-mutating method (e.g., cargo metadata --locked / cargo check --locked) to keep CI deterministic and avoid updating dependencies during validation.

components/api-server/src/client.rs (3)

127-151: Update doc comment to reflect actual return type.

As noted in previous review comments, the doc comment states the stream returns "parsed JSON values", but both streaming paths (filesystem and MongoDB) actually return String messages.

Update the doc comment to accurately describe both streaming paths:

 /// Asynchronously fetches the results of a completed search job.
 ///
 /// # Returns
 ///
-/// A stream of the job's results on success. Each item in the stream is a [`Result`] that:
+/// A stream of String messages representing the job's results on success.
+/// Results are streamed from either the filesystem (when `write_to_file` is true)
+/// or MongoDB (when `write_to_file` is false). Each item in the stream is a [`Result`] that:
 ///
 /// ## Returns
 ///
-/// A parsed JSON value representing a search result on success.
+/// A String message representing a search result on success.

---

188-191: Hardcoded tuple deserialization remains unaddressed.

This concern was previously raised but not yet addressed. The hardcoded tuple (i64, String, String, String, i64) is fragile and not self-documenting. If the serialisation format changes, this code will fail silently or produce incorrect results.

As suggested in the previous review, define a named struct:

#[derive(Deserialize)]
struct QueryResultRecord {
    timestamp: i64,
    message: String,
    file_path: String,
    orig_file_id: String,
    line_num: i64,
}

Then update the deserialization:

-                    while let Ok(e) = Deserialize::deserialize(&mut deserializer) {
-                        let e: (i64, String, String, String, i64) = e;
-                        yield Ok(e.1);
+                    while let Ok(record) = QueryResultRecord::deserialize(&mut deserializer) {
+                        yield Ok(record.message);

---

269-271: Use safe projection instead of map_unchecked_mut.

This was previously flagged with a safe alternative, but the unsafe code remains unaddressed. While the safety reasoning is sound, eliminating unsafe improves code maintainability and auditability.

Apply the safe alternative suggested in the previous review:

     fn poll_next(
         mut self: Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
     ) -> std::task::Poll<Option<Self::Item>> {
-        // SAFETY: The async runtime guarantees that `poll_next` is called exclusively. We only poll
-        // `self.0` and never move it out.
-        let inner = unsafe { self.map_unchecked_mut(|s| &mut s.0) };
-
-        let poll = inner.poll_next(cx);
+        let poll = self.0.as_mut().poll_next(cx);
         if let std::task::Poll::Ready(Some(Err(err))) = &poll {
             tracing::error!("An error occurred when streaming results: {}", err);
         }

Configuration used: CodeRabbit UI

Review profile: ASSERTIVE

Plan: Pro

Learnt from: junhaoliao
Repo: y-scope/clp PR: 0
File: :0-0
Timestamp: 2025-10-22T21:01:31.391Z
Learning: In y-scope/clp, for the clp-rust-checks workflow (GitHub Actions) and the Taskfile target deps:lock:check-rust, we should verify Rust Cargo.lock is in sync with Cargo.toml using a non-mutating method (e.g., cargo metadata --locked / cargo check --locked) to keep CI deterministic and avoid updating dependencies during validation.

components/api-server/src/client.rs (2)

128-162: fetch_results return type and docs are out of sync with actual behaviour.

The new return type Result<SearchResultStream<impl Stream<Item = Result<String, ClientError>>, _>, ClientError> makes the API return a stream of String messages (from Mongo or FS), but the doc comment still talks about “A parsed JSON value representing a search result on success” and mentions serde_json::from_str, which is no longer used here. Please update the docblock to describe that this returns a SearchResultStream of message Strings and adjust the error bullets to match the current implementation.

---

192-216: Replace todo!() for S3 with a proper error and consider FS streaming edge cases.

• The todo!() when StreamOutputStorage::S3 is selected will still panic in production if that configuration is ever enabled. This should instead return a ClientError variant (e.g., UnsupportedOperation / UnsupportedOutputStorage) so callers get a structured failure rather than a crash.
• The FS streaming loop correctly deserialises the tuple and yields message, but tokio::fs::read_dir does not guarantee ordering of entries. If file naming encodes ordering, you may want to collect and sort the paths before reading to keep result order deterministic across platforms.
• The while let Ok(event) = Deserialize::deserialize(&mut deserializer) pattern will treat any decode error as end-of-stream. If corruption of a result file is possible, consider distinguishing expected EOF from malformed data (e.g., via your IsMalformedData helper) so truly malformed files surface as ClientError::MalformedData instead of silently truncating the stream.

The first point is a correctness / robustness blocker; the latter two are more about predictable behaviour and diagnosability.

Configuration used: CodeRabbit UI

Review profile: ASSERTIVE

Plan: Pro

Learnt from: junhaoliao
Repo: y-scope/clp PR: 0
File: :0-0
Timestamp: 2025-10-22T21:01:31.391Z
Learning: In y-scope/clp, for the clp-rust-checks workflow (GitHub Actions) and the Taskfile target deps:lock:check-rust, we should verify Rust Cargo.lock is in sync with Cargo.toml using a non-mutating method (e.g., cargo metadata --locked / cargo check --locked) to keep CI deterministic and avoid updating dependencies during validation.

Learnt from: AVMatthews
Repo: y-scope/clp PR: 543
File: components/core/src/clp_s/clp-s.cpp:196-265
Timestamp: 2024-10-08T15:52:50.753Z
Learning: In `clp-s.cpp`, the `run_serializer` function interleaves serialization and writing of IR files, making it difficult to restructure it into separate functions.

components/api-server/src/client.rs (2)

128-162: Update fetch_results doc-comment to reflect String-based, FS-or-Mongo streaming

The doc-comment still refers to “parsed JSON value” results and Mongo/serde_json::from_str errors, but fetch_results now returns a SearchResultStream<…> yielding Result<String, ClientError> from either Mongo or the filesystem. This mismatch will mislead API consumers.

Please update the comment to describe:

• that the stream yields raw log/message Strings, not parsed JSON; and
• that results may come from Mongo or FS, with corresponding IO / decode errors rather than serde_json.

---

196-201: Replace todo!() for S3 storage with a proper error path

If stream_output.storage is ever configured as StreamOutputStorage::S3, this todo!() will panic the API server at runtime, which is not acceptable for a configuration-driven feature.

Given that this is an unsupported backend rather than a programmer error, it should return a ClientError instead of panicking. Because fetch_results_from_file currently cannot return a Result, you likely want to change its signature and propagate the error from fetch_results:

-    fn fetch_results_from_file(
-        &self,
-        search_job_id: u64,
-    ) -> impl Stream<Item = Result<String, ClientError>> + use<> {
-        let StreamOutputStorage::Fs { directory } = &self.config.stream_output.storage else {
-            todo!("Outputting query results to S3 is not supported for now.");
-        };
+    fn fetch_results_from_file(
+        &self,
+        search_job_id: u64,
+    ) -> Result<impl Stream<Item = Result<String, ClientError>> + use<>, ClientError> {
+        let StreamOutputStorage::Fs { directory } = &self.config.stream_output.storage else {
+            // Prefer a dedicated variant like `UnsupportedOperation` or `UnsupportedOutputStorage`.
+            return Err(ClientError::UnsupportedOperation(
+                "S3-based query result streaming is not yet supported".to_string(),
+            ));
+        };
@@
-        let stream = /* existing stream! / try_stream! body */;
-        stream
+        let stream = /* existing stream! / try_stream! body */;
+        Ok(stream)
     }

and in fetch_results:

-        if job_config.write_to_file {
-            Ok(SearchResultStream::File {
-                inner: self.fetch_results_from_file(search_job_id),
-            })
-        } else {
-            self.fetch_results_from_mongo(search_job_id)
-                .await
-                .map(|s| SearchResultStream::Mongo { inner: s })
-        }
+        if job_config.write_to_file {
+            Ok(SearchResultStream::File {
+                inner: self.fetch_results_from_file(search_job_id)?,
+            })
+        } else {
+            self.fetch_results_from_mongo(search_job_id)
+                .await
+                .map(|s| SearchResultStream::Mongo { inner: s })
+        }

Adjust the concrete ClientError variant name to match your existing enum.

Configuration used: CodeRabbit UI

Review profile: ASSERTIVE

Plan: Pro

Learnt from: junhaoliao
Repo: y-scope/clp PR: 0
File: :0-0
Timestamp: 2025-10-22T21:01:31.391Z
Learning: In y-scope/clp, for the clp-rust-checks workflow (GitHub Actions) and the Taskfile target deps:lock:check-rust, we should verify Rust Cargo.lock is in sync with Cargo.toml using a non-mutating method (e.g., cargo metadata --locked / cargo check --locked) to keep CI deterministic and avoid updating dependencies during validation.

Learnt from: AVMatthews
Repo: y-scope/clp PR: 543
File: components/core/src/clp_s/clp-s.cpp:196-265
Timestamp: 2024-10-08T15:52:50.753Z
Learning: In `clp-s.cpp`, the `run_serializer` function interleaves serialization and writing of IR files, making it difficult to restructure it into separate functions.

components/api-server/src/client.rs (5)

128-162: Update fetch_results docstring to match SearchResultStream and String messages.

The docstring still talks about “parsed JSON values” and Mongo-only error paths, but fetch_results now returns a SearchResultStream<..> of Result<String, ClientError> that can be backed by either files or Mongo. Please update the docs to describe the two backends (File vs Mongo) and the fact that each item is a log message String, and move Mongo/FS-specific error details into the respective helper docs.

---

222-235: Use try_stream! for fallible FS streams and surface decode errors.

This stream body uses ? on fallible ops (read_dir, next_entry, File::open) and yields Result<String, ClientError> items, but it is wrapped in async_stream::stream!. Per async-stream’s design, try_stream! is the macro intended for fallible streams using ? and Item = Result<T, E>; using stream! here is likely a compile-time error or, at best, wrong error propagation. The decode loop also treats any rmp_serde decode error as EOF and silently truncates the stream.

Consider:

• Switching to try_stream! so ? produces Err(ClientError) items instead of panics/compile failures.
• Matching on Deserialize::deserialize(&mut deserializer) and distinguishing normal EOF from malformed data, using your existing ClientError::MalformedData / IsMalformedData plumbing so corrupted files yield an error item instead of silently ending the stream.

Example (partial) change:

- use async_stream::stream;
+ use async_stream::try_stream;
@@
-        let stream = stream! {
+        let stream = try_stream! {
             let mut entries = tokio::fs::read_dir(search_job_output_dir).await?;
             while let Some(entry) = entries.next_entry().await? {
                 let search_result_path = entry.path();
                 let reader = std::fs::File::open(search_result_path)?;
                 let mut deserializer = rmp_serde::Deserializer::new(reader);
-                while let Ok(event) = Deserialize::deserialize(&mut deserializer) {
-                    let event: (i64, String, String, String, i64) = event;
-                    let message = event.1;
-                    yield Ok(message);
-                }
+                // TODO: handle EOF vs malformed data explicitly and map non-EOF errors to ClientError.
+                while let Ok(event) = Deserialize::deserialize(&mut deserializer) {
+                    let event: (i64, String, String, String, i64) = event;
+                    let message = event.1;
+                    yield Ok(message);
+                }
             }
         };

Confirm in the async-stream (v0.3.x) docs that `try_stream!` is the macro intended for fallible streams with `Item = Result<T, E>` and use of the `?` operator, and that `stream!` is for infallible streams that do not propagate `?` as `Err` items.

---

223-227: Blocking FS I/O and non-deterministic directory ordering are acceptable for now but may bite at scale.

This path uses std::fs::File::open (blocking) on the Tokio worker thread and iterates tokio::fs::read_dir without sorting entries, so:

• Large result files could block the reactor if deserialisation is slow.
• Result ordering may differ between platforms/FS implementations.

If you expect large files or care about strict ordering, consider offloading deserialisation to spawn_blocking and/or collecting and sorting paths before reading. Not a correctness bug for small result sets, just something to keep in mind.

---

228-231: Tuple-based deserialisation is still brittle; consider a named struct or at least documenting the layout.

Deserialising into (i64, String, String, String, i64) and then using event.1 is hard to read and tightly couples this code to the exact positional layout of the MessagePack tuple. Even if you must keep the tuple format for on-the-wire efficiency, wrapping it in a small Event struct (via #[serde(from)]) or adding a clear comment describing each element would improve maintainability and make future format changes less error-prone.

---

238-265: fetch_results_from_mongo docstring is out of date (no JSON parsing anymore).

The implementation now maps the "message" BSON field directly to a String and does not call serde_json::from_str, yet the docstring still claims it returns “parsed JSON values” and forwards serde_json::from_str errors. Please update the docs to say that each item is a log message String and list only the actual error cases: ClientError::MalformedData for missing/non-string "message" and forwarded mongodb::error::Error from the cursor and Collection::find.

Configuration used: CodeRabbit UI

Review profile: ASSERTIVE

Plan: Pro

Learnt from: junhaoliao
Repo: y-scope/clp PR: 0
File: :0-0
Timestamp: 2025-10-22T21:01:31.391Z
Learning: In y-scope/clp, for the clp-rust-checks workflow (GitHub Actions) and the Taskfile target deps:lock:check-rust, we should verify Rust Cargo.lock is in sync with Cargo.toml using a non-mutating method (e.g., cargo metadata --locked / cargo check --locked) to keep CI deterministic and avoid updating dependencies during validation.

Learnt from: AVMatthews
Repo: y-scope/clp PR: 543
File: components/core/src/clp_s/clp-s.cpp:196-265
Timestamp: 2024-10-08T15:52:50.753Z
Learning: In `clp-s.cpp`, the `run_serializer` function interleaves serialization and writing of IR files, making it difficult to restructure it into separate functions.

components/api-server/src/client.rs (3)

3-3: Import try_stream! instead of stream! for Result-yielding streams.

The import of async_stream::stream should be async_stream::try_stream, as the file-based stream at line 222 uses ? and yields Result items. Per the async-stream documentation, try_stream! is designed for fallible streams with Item = Result<T, E> and supports the ? operator, while stream! does not.

Apply this diff:

-use async_stream::stream;
+use async_stream::try_stream;

---

216-218: Replace todo!() with proper error handling for S3 storage.

The todo!() macro will cause a panic at runtime if StreamOutputStorage::S3 is configured—this is a production blocker. This issue was previously flagged but remains unaddressed.

You must add a new error variant to ClientError (e.g., UnsupportedOperation(String)) in components/api-server/src/error.rs and return it here.

First, add to components/api-server/src/error.rs:

 pub enum ClientError {
     // ... existing variants ...
     #[error("Malformed data")]
     MalformedData,
+    
+    #[error("{0}")]
+    UnsupportedOperation(String),
 }

Then in this file:

         let StreamOutputStorage::Fs { directory } = &self.config.stream_output.storage else {
-            todo!("Outputting query results to S3 is not supported for now.");
+            panic!("S3-based query result streaming is not yet supported");
         };

Note: Using panic! with a clear message is acceptable here given the # Panics section in the docstring at line 209. Alternatively, restructure to return Result and use the new error variant.

---

222-234: Use try_stream! macro and properly handle decode errors.

Two critical issues in the stream implementation:

1. Using stream! with ? and Result items: The body uses ? on async/IO operations and yields Ok(message), but the macro is stream!. Per async-stream documentation, try_stream! is required for fallible streams with Item = Result<T, E> that use the ? operator. With stream!, errors will not propagate correctly.

2. Decode errors silently truncate results: The while let Ok(event) = Deserialize::deserialize(&mut deserializer) loop breaks on any rmp_serde::decode::Error, including genuinely malformed data, without signalling an error. This can silently truncate results.

Both issues were previously flagged but remain unaddressed.

Apply this diff:

-        let stream = stream! {
+        let stream = try_stream! {
             let mut entries = tokio::fs::read_dir(search_job_output_dir).await?;
             while let Some(entry) = entries.next_entry().await? {
                 let search_result_path = entry.path();
                 let reader = std::fs::File::open(search_result_path)?;
                 let mut deserializer = rmp_serde::Deserializer::new(reader);
-                while let Ok(event) = Deserialize::deserialize(&mut deserializer) {
-                    let event: (i64, String, String, String, i64) = event;
-                    let message = event.1;
-                    yield Ok(message);
+                loop {
+                    match Deserialize::deserialize(&mut deserializer) {
+                        Ok(event) => {
+                            let event: (i64, String, String, String, i64) = event;
+                            let message = event.1;
+                            yield message;
+                        }
+                        Err(err) if err.is_eof() => break,
+                        Err(err) => Err(err)?,
+                    }
                 }
             }
         };

Note: With try_stream!, you yield the unwrapped value directly (not Ok(message)), and errors are propagated via ?.

Configuration used: CodeRabbit UI

Review profile: ASSERTIVE

Plan: Pro

Learnt from: junhaoliao
Repo: y-scope/clp PR: 0
File: :0-0
Timestamp: 2025-10-22T21:01:31.391Z
Learning: In y-scope/clp, for the clp-rust-checks workflow (GitHub Actions) and the Taskfile target deps:lock:check-rust, we should verify Rust Cargo.lock is in sync with Cargo.toml using a non-mutating method (e.g., cargo metadata --locked / cargo check --locked) to keep CI deterministic and avoid updating dependencies during validation.

Learnt from: AVMatthews
Repo: y-scope/clp PR: 543
File: components/core/src/clp_s/clp-s.cpp:196-265
Timestamp: 2024-10-08T15:52:50.753Z
Learning: In `clp-s.cpp`, the `run_serializer` function interleaves serialization and writing of IR files, making it difficult to restructure it into separate functions.

components/api-server/src/client.rs (3)

244-246: [DUPLICATE] Critical: Replace todo!() with proper error handling for S3.

This issue has been extensively flagged in previous reviews. The todo!() will panic at runtime if StreamOutputStorage::S3 is configured. You must add an appropriate ClientError variant (e.g., UnsupportedOperation or UnsupportedStreamOutputStorage) and return that error instead of panicking.

---

250-263: [DUPLICATE] Critical: Use try_stream! for fallible streams with ? operator.

This issue has been extensively flagged in previous reviews. The stream! macro does not support the ? operator for error propagation in fallible streams. You must change to try_stream! to properly propagate IO errors as Err(ClientError) stream items.

Additionally, the deserialization loop at lines 256-260 silently ends the stream on any decode error, which can truncate results without signaling corruption.

As noted in previous reviews, apply this fix:

-use async_stream::stream;
+use async_stream::try_stream;

And at line 250:

-        let stream = stream! {
+        let stream = try_stream! {
             let mut entries = tokio::fs::read_dir(search_job_output_dir).await?;
             while let Some(entry) = entries.next_entry().await? {
                 let search_result_path = entry.path();
                 let reader = std::fs::File::open(search_result_path)?;
                 let mut deserializer = rmp_serde::Deserializer::new(reader);
-                while let Ok(event_tuple) = Deserialize::deserialize(&mut deserializer) {
+                loop {
+                    match Deserialize::deserialize(&mut deserializer) {
+                        Ok(event_tuple) => {
-                    let event_tuple: EventTuple = event_tuple;
-                    let event: Event = event_tuple.into();
-                    yield Ok(event.message);
+                            let event_tuple: EventTuple = event_tuple;
+                            let event: Event = event_tuple.into();
+                            yield event.message;
+                        }
+                        Err(err) if err.is_eof() => break,
+                        Err(err) => yield Err(err.into()),
+                    }
                 }
             }
         };

---

266-266: [DUPLICATE] Minor: Add backticks around MongoDB to fix clippy doc-markdown error.

This has been flagged in previous reviews and is causing a pipeline failure.

-    /// Asynchronously fetches results of a completed search job from MongoDB.
+    /// Asynchronously fetches results of a completed search job from `MongoDB`.

Configuration used: CodeRabbit UI

Review profile: ASSERTIVE

Plan: Pro

Learnt from: junhaoliao
Repo: y-scope/clp PR: 0
File: :0-0
Timestamp: 2025-10-22T21:01:31.391Z
Learning: In y-scope/clp, for the clp-rust-checks workflow (GitHub Actions) and the Taskfile target deps:lock:check-rust, we should verify Rust Cargo.lock is in sync with Cargo.toml using a non-mutating method (e.g., cargo metadata --locked / cargo check --locked) to keep CI deterministic and avoid updating dependencies during validation.

Learnt from: AVMatthews
Repo: y-scope/clp PR: 543
File: components/core/src/clp_s/clp-s.cpp:196-265
Timestamp: 2024-10-08T15:52:50.753Z
Learning: In `clp-s.cpp`, the `run_serializer` function interleaves serialization and writing of IR files, making it difficult to restructure it into separate functions.