Parallelize SSE search and add concurrency controls by digizeph · Pull Request #132 · bgpkit/monocle

digizeph · 2026-07-01T20:36:51Z

Resolves #131.

Refactor search into a shared executor used by SSE and library progress search.
Parallelize SSE file processing while preserving cancellation, backpressure, max-results, timeout, and one terminal event.
Add search_concurrency, --concurrency, and SSE admission control via server_max_concurrent_searches / 429.

Tested:

cargo fmt
cargo clippy --all-features -- -D warnings
cargo test --all-features
Docker-served SSE search and concurrent-request 429 check

Copilot

Pull request overview

This PR refactors the BGP search execution logic into a shared, configurable executor and wires it into the SSE server path to enable parallel MRT processing. It also introduces concurrency controls at both the execution level (rayon thread count) and the server admission-control level (rejecting excess concurrent SSE searches with HTTP 429).

Changes:

Added a shared search executor API (SearchExecutionOptions, SearchSink, SearchOutcome/ExitReason) and migrated SearchLens::search_with_progress to use it.
Parallelized SSE /api/v1/search/stream by delegating to the shared executor while keeping bounded-channel backpressure and disconnect cancellation.
Added configuration/CLI flags for search_concurrency and server_max_concurrent_searches, plus a 429 error path.

Reviewed changes

Copilot reviewed 10 out of 10 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
src/lens/search/mod.rs	Introduces shared executor/sink abstractions and implements parallel execution with max-results/timeout/cancel controls.
src/server/search.rs	Switches SSE worker to shared executor, adds per-request concurrency selection and semaphore-based admission control.
src/server/mod.rs	Adds `Semaphore` to `ServerState` for limiting concurrent SSE searches.
src/server/http.rs	Adds `TooManyRequests` API error code and a helper constructor returning HTTP 429.
src/config.rs	Adds `search_concurrency` and `server_max_concurrent_searches` to `MonocleConfig` with defaults and parsing.
src/bin/monocle.rs	Adds server CLI overrides for `--concurrency` and `--max-concurrent-searches`.
src/bin/commands/search.rs	Adds local search `--concurrency` and installs a local rayon pool when set.
src/bin/commands/config.rs	Surfaces new search/server concurrency defaults in `config` command output.
src/server/README.md	Documents new config keys and CLI flags for concurrency and SSE admission control.
CHANGELOG.md	Updates unreleased notes to reflect parallel SSE search and new concurrency controls.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

digizeph · 2026-07-01T21:08:19Z

Addressed review comments in de872e1:

Do not report files stopped by cancellation/timeout as successful.
Avoid hot-path batch capacity loss by swapping with a preallocated Vec instead of mem::take.
Added focused executor tests for max-results reservation/truncation, batch capacity reuse, cancellation, and timeout stop conditions.

Validated with:

cargo fmt
cargo clippy --all-features -- -D warnings
cargo test --all-features

Copilot

Pull request overview

Copilot reviewed 10 out of 10 changed files in this pull request and generated 2 comments.

digizeph · 2026-07-02T02:51:19Z

Addressed the latest timeout review comments in 3729772:

Start the search timer before the broker query.
Include broker query time in duration_secs.
Check timeout immediately after broker query returns, before file processing starts.

Validated with:

cargo fmt
cargo clippy --all-features -- -D warnings
cargo test --all-features

Copilot

Pull request overview

Copilot reviewed 10 out of 10 changed files in this pull request and generated 4 comments.

+    let max_concurrent_searches = config.server_max_concurrent_searches;
    let state = ServerState {
        config: Arc::new(config),
+        search_permits: Arc::new(Semaphore::new(max_concurrent_searches)),
    };


+pub enum SearchControl {
+    Continue,
+    Stop,
+}


+        match options.concurrency {
+            Some(n) if n > 0 => {
+                let pool = rayon::ThreadPoolBuilder::new().num_threads(n).build()?;
+                pool.install(run);
            }


+    for elem in parser {
+        if search_should_stop(&state) {
+            break;
+        }
+
+        batch.push(elem);
+        if batch.len() >= state.batch_size {
+            let accepted = emit_search_batch(&state, index, &url, &collector, &mut batch);
+            file_messages += accepted;
+        }


digizeph · 2026-07-02T03:47:32Z

Addressed latest review comments in 2997feb:

Represent unlimited SSE searches explicitly with Option<Arc<Semaphore>> instead of a zero-permit semaphore.
Build and reuse a server search rayon pool at startup when search_concurrency > 0, avoiding per-request pool creation in server mode.
Document that SearchControl::Stop maps to SearchExitReason::Cancelled.
Flush batches early when the remaining max_results budget is smaller than batch_size.

Validated with:

cargo fmt
cargo clippy --all-features -- -D warnings
cargo test --all-features

Copilot

Pull request overview

Copilot reviewed 10 out of 10 changed files in this pull request and generated 2 comments.

…, update config example - Set SearchExecutionOptions::default() batch_size to 64 instead of 0 (which was clamped to 1), reducing per-element allocations in the library search path. - Suppress SearchProgress::Completed in SseSearchSink::on_progress to avoid sending a redundant progress-level completion event before the terminal SearchStreamEvent::Completed. - Add search_concurrency and server_max_concurrent_searches to monocle.toml.example to match documentation in config.rs and server/README.md.

fix: parallelize SSE search and add concurrency controls

b3229c1

digizeph requested a review from Copilot July 1, 2026 20:37

Copilot started reviewing on behalf of digizeph July 1, 2026 20:37 View session

Copilot AI reviewed Jul 1, 2026

View reviewed changes

Comment thread src/lens/search/mod.rs

Comment thread src/lens/search/mod.rs Outdated

Comment thread src/lens/search/mod.rs

test: cover search executor stop conditions

de872e1

digizeph requested a review from Copilot July 1, 2026 21:10

Copilot started reviewing on behalf of digizeph July 1, 2026 21:11 View session

Copilot AI reviewed Jul 1, 2026

View reviewed changes

Comment thread src/lens/search/mod.rs Outdated

Comment thread src/lens/search/mod.rs

fix: include broker query time in search timeout

3729772

digizeph requested a review from Copilot July 2, 2026 03:38

Copilot started reviewing on behalf of digizeph July 2, 2026 03:39 View session

Copilot AI reviewed Jul 2, 2026

View reviewed changes

fix: refine search concurrency and max result handling

2997feb

digizeph requested a review from Copilot July 2, 2026 03:51

Copilot started reviewing on behalf of digizeph July 2, 2026 03:51 View session

Copilot AI reviewed Jul 2, 2026

View reviewed changes

Comment thread src/server/search.rs

Comment thread src/server/README.md

digizeph merged commit 62d41fc into main Jul 2, 2026
1 check passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Parallelize SSE search and add concurrency controls#132

Parallelize SSE search and add concurrency controls#132
digizeph merged 5 commits into
mainfrom
fix/sse-search-concurrency-131

digizeph commented Jul 1, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

digizeph commented Jul 1, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

digizeph commented Jul 2, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

digizeph commented Jul 2, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

digizeph commented Jul 1, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

digizeph commented Jul 1, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

digizeph commented Jul 2, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

digizeph commented Jul 2, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants