Squashed commit of the following:

commit 14f92b8
Author: Dmitrii Kostyrev <dkostyrev@joom.com>
Date:   Sun Jan 25 12:07:20 2026 +0000

    Introduce batchInterval and batchDebounce.

commit 12d839c
Author: Dmitrii Kostyrev <dkostyrev@joom.com>
Date:   Sun Jan 25 12:06:45 2026 +0000

    Introduce batch notify and assign actions.

commit 7393a00
Author: Dmitrii Kostyrev <dkostyrev@joom.com>
Date:   Sun Jan 25 12:06:12 2026 +0000

    Use RWLock instead of single Mutext in MemoryAwaitedActionDb.

commit 8cfeade
Author: Dmitrii Kostyrev <dkostyrev@joom.com>
Date:   Sun Jan 25 12:05:37 2026 +0000

    Fix batch matching by allowing same worker accept multiple jobs.

commit 7cd29c9
Author: Dmitrii Kostyrev <dkostyrev@joom.com>
Date:   Fri Jan 23 12:15:40 2026 +0000

    Introduce batch worker matching.

Author: dkostyrev

nativelink-config/src/schedulers.rs

@@ -82,6 +82,18 @@ const fn default_worker_match_logging_interval_s() -> i64 {
     10
 }
 
+/// Default batch interval in milliseconds (100ms).
+/// This is the maximum time between batch matching cycles.
+const fn default_batch_interval_ms() -> u64 {
+    100
+}
+
+/// Default debounce window in milliseconds (20ms).
+/// After a trigger, wait this long to collect more changes before running.
+const fn default_batch_debounce_ms() -> u64 {
+    20
+}
+
 #[derive(Deserialize, Serialize, Debug, Default)]
 #[serde(deny_unknown_fields)]
 #[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
@@ -166,6 +178,38 @@ pub struct SimpleSpec {
         deserialize_with = "convert_duration_with_shellexpand_and_negative"
     )]
     pub worker_match_logging_interval_s: i64,
+
+    /// Enable batch worker matching optimization.
+    /// When enabled, the scheduler will collect queued actions and match them
+    /// to workers in a single batch operation, reducing lock contention.
+    /// This can significantly improve throughput when there are many queued
+    /// actions and workers.
+    /// Default: false
+    #[serde(default)]
+    pub enable_batch_worker_matching: bool,
+
+    /// Maximum interval between batch matching cycles (milliseconds).
+    /// Even without triggers, matching runs at least this often.
+    /// Only used when `enable_batch_worker_matching` is true.
+    /// Default: 100ms
+    #[serde(
+        default = "default_batch_interval_ms",
+        deserialize_with = "convert_numeric_with_shellexpand"
+    )]
+    pub batch_interval_ms: u64,
+
+    /// Debounce window after first trigger (milliseconds).
+    /// When a task or worker change notification is received, wait this long
+    /// to collect additional changes before running batch match.
+    /// This improves batching efficiency under bursty load.
+    /// 0 = immediate (no debounce).
+    /// Only used when `enable_batch_worker_matching` is true.
+    /// Default: 20ms
+    #[serde(
+        default = "default_batch_debounce_ms",
+        deserialize_with = "convert_numeric_with_shellexpand"
+    )]
+    pub batch_debounce_ms: u64,
 }
 
 #[derive(Deserialize, Serialize, Debug)]

nativelink-metric/src/lib.rs

@@ -458,6 +458,18 @@ impl<T: MetricsComponent> MetricsComponent for async_lock::Mutex<T> {
     }
 }
 
+impl<T: MetricsComponent> MetricsComponent for async_lock::RwLock<T> {
+    fn publish(
+        &self,
+        kind: MetricKind,
+        field_metadata: MetricFieldData,
+    ) -> Result<MetricPublishKnownKindData, Error> {
+        // It is safe to block in the publishing thread.
+        let lock = self.read_blocking();
+        lock.publish(kind, field_metadata)
+    }
+}
+
 impl<T: MetricsComponent> MetricsComponent for parking_lot::Mutex<T> {
     fn publish(
         &self,