feat: add remote batching pipeline with buffered per-destination queues

Author: aaronlifton

README.md

@@ -6,24 +6,35 @@ core runtime.
 
 Current fastest speed, with default controls of `max-depth` 4, `max-links-per-page` 16, `politeness-ms` 250,
 `partition-strategy` 'wiki-prefix' (instead of 'hash'), and `duration-secs` 4 (it crawls for 4 seconds, but any enqued
-link is still awaited, so it runs for approximately 30 sec.) is **68.98 pages/sec**.
+link is still awaited, so it runs for approximately 30 sec.) is **71.95 pages/sec**.
 
 ## Metrics
 
-When running `cargo run --example wiki --features multi_thread -- --duration-secs 4 --partition wiki-prefix`:
+When running
 
 ```
---- crawl metrics (29.43s) ---
-pages fetched: 2030
-urls fetched/sec: 68.98
-urls discovered: 4844
-urls enqueued: 2026
-duplicate skips: 2818
+  cargo run --example wiki --features multi_thread -- \
+    --duration-secs 4 \
+    --partition wiki-prefix \
+    --partition-namespace \
+    --partition-buckets 26 \
+    --remote-batch-size 32
+```
+
+The crawl metrics were:
+
+```
+--- crawl metrics (26.46s) ---
+pages fetched: 1904
+urls fetched/sec: 71.95
+urls discovered: 4351
+urls enqueued: 1900
+duplicate skips: 2451
 frontier rejects: 0
 http errors: 0
 url parse errors: 0
-local shard enqueues: 9315
-remote shard links: 3039 (batches 1442)
+local shard enqueues: 7870
+remote shard links: 2718 (batches 331)
 ```
 
 ## Highlights
@@ -73,14 +84,19 @@ cargo run --example wiki --features multi_thread -- --duration-secs 60
 - Cross-shard discoveries are routed through bounded mpsc channels, so enqueue contention happens on a single consumer
   instead of every worker.
 - Pass `--partition wiki-prefix` (default: `hash`) to keep Wikipedia articles with similar prefixes on the same shard.
+- Use `--partition-buckets <n>` (default `0`, meaning shard count) to control how many alphabetical buckets feed into
+  the shards, and `--partition-namespace` to keep namespaces like `Talk:` or `Help:` on stable shards.
+- Tune `--remote-batch-size <n>` (default 8) to control how many cross-shard links get buffered before the router sends
+  them; higher values reduce channel wakeups at the cost of slightly delayed enqueues on the destination shard.
 
 ## Customizing Crawls
 
 - `CrawlControls` (exposed via CLI/env vars) manage maximum depth, per-domain filters, link-per-page caps, politeness
   delays, run duration, and more. See `src/controls.rs` for every option.
 - `UrlFilter` lets you inject arbitrary site-specific logic—`examples/wiki.rs` filters out non-article namespaces and
   query patterns.
-- Metrics live in `src/runtime.rs` and can be extended if you need additional counters or telemetry sinks.
+- Metrics live in `src/runtime.rs` and can be extended if you need additional counters or telemetry sinks. Multi-thread
+  runs also report `local shard enqueues` vs `remote shard links (batches)` so you can gauge partition efficiency.
 
 ## LLM-Oriented Next Steps
 

src/controls.rs

@@ -95,6 +95,21 @@ pub struct Cli {
     #[cfg(feature = "multi_thread")]
     #[arg(long, env = "FASTCRAWL_PARTITION", default_value = "hash")]
     pub partition: PartitionStrategyArg,
+
+    /// Number of wiki prefix buckets (0 = auto = shard count)
+    #[cfg(feature = "multi_thread")]
+    #[arg(long, env = "FASTCRAWL_PARTITION_BUCKETS", default_value_t = 0)]
+    pub partition_buckets: usize,
+
+    /// Treat namespaces (e.g., Talk:, Help:) as part of the partition key
+    #[cfg(feature = "multi_thread")]
+    #[arg(long, env = "FASTCRAWL_PARTITION_NAMESPACE", default_value_t = false)]
+    pub partition_namespace: bool,
+
+    /// Maximum remote links to buffer before flushing to another shard (0 = default)
+    #[cfg(feature = "multi_thread")]
+    #[arg(long, env = "FASTCRAWL_REMOTE_BATCH_SIZE", default_value_t = 0)]
+    pub remote_batch_size: usize,
 }
 
 impl Cli {
@@ -123,8 +138,13 @@ impl Cli {
 
     #[cfg(feature = "multi_thread")]
     /// Returns the requested sharding strategy when multi-threading is enabled.
-    pub fn partition_strategy(&self) -> PartitionStrategyArg {
-        self.partition
+    pub fn partition_settings(&self) -> PartitionSettings {
+        PartitionSettings {
+            strategy: self.partition,
+            wiki_bucket_count: (self.partition_buckets > 0).then_some(self.partition_buckets),
+            wiki_include_namespace: self.partition_namespace,
+            remote_batch_size: (self.remote_batch_size > 0).then_some(self.remote_batch_size),
+        }
     }
 }
 
@@ -137,3 +157,17 @@ pub enum PartitionStrategyArg {
     /// Use Wikipedia-style namespace/title prefixes to keep related pages together.
     WikiPrefix,
 }
+
+#[cfg(feature = "multi_thread")]
+#[derive(Copy, Clone, Debug)]
+/// Parsed partition configuration used by the runtime.
+pub struct PartitionSettings {
+    /// Selected strategy variant.
+    pub strategy: PartitionStrategyArg,
+    /// Optional bucket count override for wiki prefix strategy (defaults to shard count when `None`).
+    pub wiki_bucket_count: Option<usize>,
+    /// Whether to incorporate namespace prefixes (e.g., `Talk:`) into the partition key.
+    pub wiki_include_namespace: bool,
+    /// Optional remote batch size override.
+    pub remote_batch_size: Option<usize>,
+}