Add opt-in join-the-flight ByteStream write dedup

N concurrent ByteStream Write streams for the same digest are N full
uploads over the wire and N full drains through the store stack. With the
new opt-in ByteStreamConfig.experimental_write_dedup, exactly one
concurrent upload per digest is elected the leading upload; concurrent
duplicates drain-and-discard their stream while waiting and are acked
with the REAPI early WriteResponse only once the leading upload durably
commits. Uploads of already-durable digests complete early after a single
existence check. If the leading upload fails or is cancelled, waiters
receive a retryable ABORTED via a send-on-drop guard and one retry
becomes the new leader. Resumed streams bypass dedup to preserve resume
semantics.

Waiters must drain their request stream: unread bytes would pin the
HTTP/2 connection flow-control window and deadlock a leading upload
sharing the connection (found by benchmark).

Measured (real gRPC stack, byte-verified, dedup off -> on): 228x256KiB
simultaneous same-digest uploads: server-received bytes 59.9MB -> 17.1MB
(3.5x), store updates 228 -> 1; 64x2MiB staggered burst: 136.5MB ->
8.8MB (15.5x), wall 42ms -> 3ms. Metrics group write_dedup:
flights_joined, early_completes, leader_failures, bytes_saved.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UXVtatcR9YMecBiu9RjwpC

Author: erneestoc

nativelink-config/src/backcompat.rs

@@ -102,6 +102,7 @@ where
                         max_bytes_per_stream: old_config.max_bytes_per_stream,
                         persist_stream_on_disconnect_timeout_s: old_config
                             .persist_stream_on_disconnect_timeout_s,
+                        experimental_write_dedup: false,
                     },
                 })
                 .collect();

nativelink-config/src/cas_server.rs

@@ -349,6 +349,24 @@ pub struct ByteStreamConfig {
         alias = "persist_stream_on_disconnect_timeout"
     )]
     pub persist_stream_on_disconnect_timeout_s: usize,
+
+    /// Deduplicate concurrent uploads of the same digest ("join the flight").
+    /// When multiple clients upload a blob with the same digest at the same
+    /// time, only the first upload is streamed to the store; the others wait
+    /// for it to durably commit and then complete early without transferring
+    /// their payload, as permitted by the REAPI specification. Uploads of
+    /// blobs that already exist in the store also complete early after a
+    /// single existence check.
+    ///
+    /// If the leading upload fails, waiting uploads receive a retryable
+    /// ABORTED error and one of the retrying clients becomes the new leader.
+    ///
+    /// This saves upload bandwidth and store work when many actions produce
+    /// identical outputs, at the cost of one existence check per new upload.
+    ///
+    /// Default: false (disabled)
+    #[serde(default, skip_serializing_if = "is_default")]
+    pub experimental_write_dedup: bool,
 }
 
 // Older bytestream config. All fields are as per the newer docs, but this requires