Parallelize SSE search and add concurrency controls

CHANGELOG.md

@@ -54,14 +54,20 @@ All notable changes to this project will be documented in this file.
     element batches, cancellation on disconnect, and max-results limit
 * Added server configuration fields to `MonocleConfig`: `server_address`,
   `server_port`, `server_max_search_batch_size`, `server_max_search_results`,
-  `server_search_timeout_secs`. All configurable via `monocle.toml` and
-  `MONOCLE_*` environment variables.
-* SSE search streaming uses sequential file processing with `Arc<AtomicBool>`
-  cancellation — no new lens method needed. The server calls existing
-  `SearchFilters` utility methods (`to_broker_items`, `to_parser`) directly.
+  `server_search_timeout_secs`, `server_max_concurrent_searches`, and shared
+  `search_concurrency`. All configurable via `monocle.toml` and `MONOCLE_*`
+  environment variables.
+* Parallelized SSE search streaming through the shared search executor while
+  preserving cancellation, bounded-channel backpressure, max-results handling,
+  timeout handling, and the single-terminal-event invariant.
 * Bounded mpsc channel (capacity 32) with backpressure: element batches are
   never dropped; progress events may be coalesced under backpressure.
 * Terminal event invariant: exactly one of `completed`, `cancelled`, or `error`.
+* Added `--concurrency` to local search and server commands. Explicit values use
+  local rayon thread pools; unset/0 keeps rayon defaults including
+  `RAYON_NUM_THREADS`.
+* Added server admission control for SSE search requests via
+  `server_max_concurrent_searches` (default 3); excess requests return HTTP 429.
 * Added Phase 2 REST API endpoints:
   - Tier 1 (stateless): `POST /time/parse`, `POST /country/lookup`,
     `POST /ip/lookup`, `GET /ip/public`

monocle.toml.example

@@ -21,9 +21,15 @@ server_max_search_batch_size = 100
 # Maximum search results per request (0 = unlimited)
 server_max_search_results = 10000
 
+# Search concurrency; 0 = auto/rayon default (default: 0)
+search_concurrency = 0
+
 # Search timeout in seconds (0 = no timeout)
 server_search_timeout_secs = 300
 
+# Maximum concurrent SSE search requests (0 = unlimited, default: 3)
+server_max_concurrent_searches = 3
+
 # =============================================================================
 # Auth (token-based, disabled by default)
 # =============================================================================

src/bin/commands/config.rs

@@ -104,7 +104,9 @@ struct ServerDefaults {
     port: u16,
     max_search_batch_size: usize,
     max_search_results: u64,
+    search_concurrency: usize,
     search_timeout_secs: u64,
+    max_concurrent_searches: usize,
     auth_enabled: bool,
 }
 
@@ -115,7 +117,9 @@ impl From<&MonocleConfig> for ServerDefaults {
             port: config.server_port,
             max_search_batch_size: config.server_max_search_batch_size,
             max_search_results: config.server_max_search_results,
+            search_concurrency: config.search_concurrency,
             search_timeout_secs: config.server_search_timeout_secs,
+            max_concurrent_searches: config.server_max_concurrent_searches,
             auth_enabled: config.server_auth_enabled,
         }
     }
@@ -371,10 +375,26 @@ fn print_config_table(info: &ConfigInfo, verbose: bool) {
         "  Search max results: {} (0 = unlimited)",
         info.server_defaults.max_search_results
     );
+    println!(
+        "  Search concurrency: {}",
+        if info.server_defaults.search_concurrency == 0 {
+            "auto".to_string()
+        } else {
+            info.server_defaults.search_concurrency.to_string()
+        }
+    );
     println!(
         "  Search timeout:    {} seconds (0 = no timeout)",
         info.server_defaults.search_timeout_secs
     );
+    println!(
+        "  Max SSE searches:  {}",
+        if info.server_defaults.max_concurrent_searches == 0 {
+            "unlimited".to_string()
+        } else {
+            info.server_defaults.max_concurrent_searches.to_string()
+        }
+    );
     println!("  Auth enabled:      {}", info.server_defaults.auth_enabled);
 
     if verbose {

src/bin/commands/search.rs

@@ -88,6 +88,10 @@ pub struct SearchArgs {
     #[clap(flatten)]
     pub filters: SearchFilters,
 
+    /// Search concurrency (0 = auto/rayon default; overrides config)
+    #[clap(long)]
+    pub concurrency: Option<usize>,
+
     /// Remote Monocle server URL (e.g., http://localhost:8080/api/v1/search/stream).
     /// When set, search runs against the remote server instead of locally.
     #[clap(long, value_name = "URL")]
@@ -554,6 +558,24 @@ fn fetch_broker_items_cached(
 }
 
 pub fn run(config: &MonocleConfig, args: SearchArgs, output_format: OutputFormat) {
+    let concurrency = args.concurrency.unwrap_or(config.search_concurrency);
+    if concurrency > 0 {
+        match rayon::ThreadPoolBuilder::new()
+            .num_threads(concurrency)
+            .build()
+        {
+            Ok(pool) => pool.install(|| run_inner(config, args, output_format)),
+            Err(e) => {
+                eprintln!("Failed to create rayon thread pool: {e}");
+                std::process::exit(1);
+            }
+        }
+    } else {
+        run_inner(config, args, output_format);
+    }
+}
+
+fn run_inner(config: &MonocleConfig, args: SearchArgs, output_format: OutputFormat) {
     let SearchArgs {
         dry_run,
         sqlite_path,
@@ -569,6 +591,7 @@ pub fn run(config: &MonocleConfig, args: SearchArgs, output_format: OutputFormat
         use_cache,
         cache_dir,
         mut filters,
+        concurrency: _,
         remote_url,
         remote_token,
     } = args;

src/bin/monocle.rs

@@ -115,10 +115,18 @@ struct ServerArgs {
     #[clap(long)]
     max_search_results: Option<u64>,
 
+    /// Search concurrency (0 = auto/rayon default, overrides config)
+    #[clap(long)]
+    concurrency: Option<usize>,
+
     /// Search timeout in seconds (0 = no timeout, overrides config)
     #[clap(long)]
     search_timeout_secs: Option<u64>,
 
+    /// Maximum concurrent SSE search requests (0 = unlimited, overrides config)
+    #[clap(long)]
+    max_concurrent_searches: Option<usize>,
+
     /// Enable token auth for /api/v1/* endpoints (overrides config)
     #[clap(long)]
     auth_enabled: Option<bool>,
@@ -203,9 +211,15 @@ fn main() {
                 if let Some(v) = args.max_search_results {
                     server_config.server_max_search_results = v;
                 }
+                if let Some(v) = args.concurrency {
+                    server_config.search_concurrency = v;
+                }
                 if let Some(v) = args.search_timeout_secs {
                     server_config.server_search_timeout_secs = v;
                 }
+                if let Some(v) = args.max_concurrent_searches {
+                    server_config.server_max_concurrent_searches = v;
+                }
                 if let Some(v) = args.auth_enabled {
                     server_config.server_auth_enabled = v;
                 }

src/config.rs

@@ -24,6 +24,12 @@ pub const DEFAULT_SERVER_MAX_SEARCH_RESULTS: u64 = 0;
 /// Default search timeout in seconds (0 = no timeout)
 pub const DEFAULT_SERVER_SEARCH_TIMEOUT_SECS: u64 = 0;
 
+/// Default search concurrency (0 = rayon default / CPU count)
+pub const DEFAULT_SEARCH_CONCURRENCY: usize = 0;
+
+/// Default maximum concurrent SSE search requests
+pub const DEFAULT_SERVER_MAX_CONCURRENT_SEARCHES: usize = 3;
+
 /// Default: auth disabled
 pub const DEFAULT_SERVER_AUTH_ENABLED: bool = false;
 
@@ -69,9 +75,15 @@ pub struct MonocleConfig {
     /// Maximum search results per request, 0 = unlimited (default: 0)
     pub server_max_search_results: u64,
 
+    /// Search concurrency, 0 = rayon default / CPU count (default: 0)
+    pub search_concurrency: usize,
+
     /// Search timeout in seconds, 0 = no timeout (default: 0)
     pub server_search_timeout_secs: u64,
 
+    /// Maximum concurrent SSE search requests, 0 = unlimited (default: 3)
+    pub server_max_concurrent_searches: usize,
+
     /// Enable token auth for /api/v1/* endpoints (default: false)
     pub server_auth_enabled: bool,
 
@@ -101,6 +113,10 @@ const EMPTY_CONFIG: &str = r#"### monocle configuration file
 ### If true, error out instead of falling back to Cloudflare when RTR fails
 # rpki_rtr_no_fallback = false
 
+### Search execution configuration
+### Search concurrency; 0 = auto/rayon default. Can also be set with MONOCLE_SEARCH_CONCURRENCY.
+# search_concurrency = 0
+
 ### HTTP service configuration
 ### These settings control the monocle HTTP/SSE server.
 # server_address = "127.0.0.1"
@@ -111,6 +127,8 @@ const EMPTY_CONFIG: &str = r#"### monocle configuration file
 # server_max_search_results = 0
 ### Search timeout in seconds (0 = no timeout)
 # server_search_timeout_secs = 0
+### Maximum concurrent SSE search requests (0 = unlimited)
+# server_max_concurrent_searches = 3
 
 ### Auth (token-based, disabled by default)
 ### When enabled, requests to /api/v1/* require Authorization: Bearer <token>
@@ -228,7 +246,9 @@ impl Default for MonocleConfig {
             server_port: DEFAULT_SERVER_PORT,
             server_max_search_batch_size: DEFAULT_SERVER_MAX_SEARCH_BATCH_SIZE,
             server_max_search_results: DEFAULT_SERVER_MAX_SEARCH_RESULTS,
+            search_concurrency: DEFAULT_SEARCH_CONCURRENCY,
             server_search_timeout_secs: DEFAULT_SERVER_SEARCH_TIMEOUT_SECS,
+            server_max_concurrent_searches: DEFAULT_SERVER_MAX_CONCURRENT_SEARCHES,
             server_auth_enabled: DEFAULT_SERVER_AUTH_ENABLED,
             server_auth_token: String::new(),
         }
@@ -359,10 +379,18 @@ impl MonocleConfig {
             .get("server_max_search_results")
             .and_then(|s| s.parse().ok())
             .unwrap_or(DEFAULT_SERVER_MAX_SEARCH_RESULTS);
+        let search_concurrency = config
+            .get("search_concurrency")
+            .and_then(|s| s.parse().ok())
+            .unwrap_or(DEFAULT_SEARCH_CONCURRENCY);
         let server_search_timeout_secs = config
             .get("server_search_timeout_secs")
             .and_then(|s| s.parse().ok())
             .unwrap_or(DEFAULT_SERVER_SEARCH_TIMEOUT_SECS);
+        let server_max_concurrent_searches = config
+            .get("server_max_concurrent_searches")
+            .and_then(|s| s.parse().ok())
+            .unwrap_or(DEFAULT_SERVER_MAX_CONCURRENT_SEARCHES);
 
         // Parse auth configuration
         let server_auth_enabled = config
@@ -385,7 +413,9 @@ impl MonocleConfig {
             server_port,
             server_max_search_batch_size,
             server_max_search_results,
+            search_concurrency,
             server_search_timeout_secs,
+            server_max_concurrent_searches,
             server_auth_enabled,
             server_auth_token,
         })
@@ -461,10 +491,26 @@ impl MonocleConfig {
             "Search Max Results: {}",
             self.server_max_search_results
         ));
+        lines.push(format!(
+            "Search Concurrency: {}",
+            if self.search_concurrency == 0 {
+                "auto".to_string()
+            } else {
+                self.search_concurrency.to_string()
+            }
+        ));
         lines.push(format!(
             "Search Timeout:     {} seconds",
             self.server_search_timeout_secs
         ));
+        lines.push(format!(
+            "Max SSE Searches:   {}",
+            if self.server_max_concurrent_searches == 0 {
+                "unlimited".to_string()
+            } else {
+                self.server_max_concurrent_searches.to_string()
+            }
+        ));
         lines.push(format!("Auth Enabled:       {}", self.server_auth_enabled));
 
         // Check if cache directories exist and show status
@@ -941,10 +987,15 @@ mod tests {
             config.server_max_search_results,
             DEFAULT_SERVER_MAX_SEARCH_RESULTS
         );
+        assert_eq!(config.search_concurrency, DEFAULT_SEARCH_CONCURRENCY);
         assert_eq!(
             config.server_search_timeout_secs,
             DEFAULT_SERVER_SEARCH_TIMEOUT_SECS
         );
+        assert_eq!(
+            config.server_max_concurrent_searches,
+            DEFAULT_SERVER_MAX_CONCURRENT_SEARCHES
+        );
         assert!(!config.server_auth_enabled);
         assert_eq!(config.server_auth_token, "");
     }