Move Bytestream to array config

Author: palfrey

Cargo.lock

@@ -53,11 +53,10 @@ A very basic configuration that's a pure in-memory store is:
       "capabilities": [{
         "instance_name": "main"
       }],
-      "bytestream": {
-        "cas_stores": {
-          "main": "CAS_MAIN_STORE",
-        }
-      }
+      "bytestream": [{
+        "instance_name": "main",
+        "cas_store": "CAS_MAIN_STORE",
+      }]
     }
   }]
 }

nativelink-config/README.md

@@ -291,11 +291,10 @@ The `public` server consists of a `listener` object and a `services` object. The
           "scheduler": "MAIN_SCHEDULER",
         }
       }],
-      "bytestream": {
-        "cas_stores": {
-          "main": "WORKER_FAST_SLOW_STORE",
-        }
-      }
+      "bytestream": [{
+        "instance_name": "main",
+        "cas_store": "WORKER_FAST_SLOW_STORE",
+      }]
     },
   },{
     "name": "private_workers_servers"
@@ -342,11 +341,10 @@ The `private` server consists of a `listener` object and a `services` object. Th
           "scheduler": "MAIN_SCHEDULER",
         }
       }],
-      "bytestream": {
-        "cas_stores": {
-          "main": "WORKER_FAST_SLOW_STORE",
-        }
-      }
+      "bytestream": [{
+        "instance_name": "main",
+        "cas_store": "WORKER_FAST_SLOW_STORE",
+      }]
     },
   },{
     "name": "private_workers_servers",
@@ -519,11 +517,10 @@ Below, you will find a fully tested example that you can also find in [basic_cas
           "scheduler": "MAIN_SCHEDULER",
         }
       }],
-      "bytestream": {
-        "cas_stores": {
-          "main": "WORKER_FAST_SLOW_STORE",
-        }
-      }
+      "bytestream": [{
+        "instance_name": "main",
+        "cas_store": "WORKER_FAST_SLOW_STORE",
+      }]
     }
   }, {
     "name": "private_workers_servers",

nativelink-config/examples/README.md

@@ -156,11 +156,12 @@
             },
           },
         ],
-        bytestream: {
-          cas_stores: {
-            main: "WORKER_FAST_SLOW_STORE",
+        bytestream: [
+          {
+            instance_name: "main",
+            cas_store: "WORKER_FAST_SLOW_STORE",
           },
-        },
+        ],
       },
     },
     {

nativelink-config/examples/basic_cas.json5

@@ -0,0 +1,42 @@
+{
+  stores: [],
+  servers: [
+    {
+      listener: {
+        http: {
+          socket_address: "0.0.0.0:50051",
+        },
+      },
+      services: {
+        cas: {
+          "": {
+            cas_store: "CAS_MAIN_STORE",
+          },
+        },
+        ac: {
+          "": {
+            ac_store: "AC_MAIN_STORE",
+          },
+        },
+        execution: {
+          "": {
+            cas_store: "WORKER_FAST_SLOW_STORE",
+            scheduler: "MAIN_SCHEDULER",
+          },
+        },
+        capabilities: {
+          "": {
+            remote_execution: {
+              scheduler: "MAIN_SCHEDULER",
+            },
+          },
+        },
+        bytestream: {
+          cas_stores: {
+            main: "WORKER_FAST_SLOW_STORE",
+          },
+        },
+      },
+    },
+  ],
+}

nativelink-config/examples/legacy_service_config.json5

@@ -3,7 +3,7 @@ use std::collections::HashMap;
 use serde::{Deserialize, Deserializer, Serialize};
 use tracing::warn;
 
-use crate::cas_server::WithInstanceName;
+use crate::cas_server::{ByteStreamConfig, OldByteStreamConfig, WithInstanceName};
 
 #[derive(Debug, Deserialize)]
 #[serde(untagged)]
@@ -12,7 +12,19 @@ enum WithInstanceNameBackCompat<T> {
     Vec(Vec<WithInstanceName<T>>),
 }
 
-/// Use `#[serde(default, deserialize_with = "backcompat::opt_vec_named_config")]` for backwards
+fn deprecated(old_map: &String, new_map: &String) {
+    warn!(
+        r"WARNING: Using deprecated map format for services. Please migrate to the new array format:
+// Old:
+{}
+// New:
+{}
+",
+        old_map, new_map
+    );
+}
+
+/// Use `#[serde(default, deserialize_with = "backcompat::opt_vec_with_instance_name")]` for backwards
 /// compatibility with map-based access. A deprecation warning will be written to stderr if the
 /// old format is used.
 pub(crate) fn opt_vec_with_instance_name<'de, D, T>(
@@ -38,24 +50,73 @@ where
                     config,
                 })
                 .collect();
-            warn!(
-                r"WARNING: Using deprecated map format for services. Please migrate to the new array format:
-// Old:
-{}
-// New:
-{}
-",
-                serde_map,
+            deprecated(
+                &serde_map,
                 // TODO(palfrey): ideally this would be serde_json5::to_string_pretty but that doesn't exist
                 // JSON is close enough to be workable for now
-                serde_json::to_string_pretty(&vec).expect("valid new map")
+                &serde_json::to_string_pretty(&vec).expect("valid new map"),
             );
             Ok(Some(vec))
         }
         WithInstanceNameBackCompat::Vec(vec) => Ok(Some(vec)),
     }
 }
 
+#[derive(Debug, Deserialize)]
+#[serde(untagged)]
+enum ByteStreamKind {
+    Old(OldByteStreamConfig),
+    New(Vec<WithInstanceName<ByteStreamConfig>>),
+}
+
+/// Use `#[serde(default, deserialize_with = "backcompat::opt_bytestream")]` for backwards
+/// compatibility with older bytestream config . A deprecation warning will be written to stderr if the
+/// old format is used.
+pub(crate) fn opt_bytestream<'de, D>(
+    deserializer: D,
+) -> Result<Option<Vec<WithInstanceName<ByteStreamConfig>>>, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let Some(back_compat) = Option::deserialize(deserializer)? else {
+        return Ok(None);
+    };
+
+    match back_compat {
+        ByteStreamKind::Old(old_config) => {
+            if old_config.max_decoding_message_size != 0 {
+                warn!(
+                    "WARNING: max_decoding_message_size is ignored on Bytestream now. Please set on the HTTP listener instead"
+                );
+            }
+            // TODO(palfrey): ideally this would be serde_json5::to_string_pretty but that doesn't exist
+            // JSON is close enough to be workable for now
+            let serde_map = serde_json::to_string_pretty(&old_config).expect("valid map");
+            let names = old_config.cas_stores;
+            let vec: Vec<WithInstanceName<_>> = names
+                .iter()
+                .map(|(instance_name, cas_store)| WithInstanceName {
+                    instance_name: instance_name.clone(),
+                    config: ByteStreamConfig {
+                        cas_store: cas_store.clone(),
+                        max_bytes_per_stream: old_config.max_bytes_per_stream,
+                        persist_stream_on_disconnect_timeout: old_config
+                            .persist_stream_on_disconnect_timeout,
+                    },
+                })
+                .collect();
+            deprecated(
+                &serde_map,
+                // TODO(palfrey): ideally this would be serde_json5::to_string_pretty but that doesn't exist
+                // JSON is close enough to be workable for now
+                &serde_json::to_string_pretty(&vec).expect("valid new map"),
+            );
+            Ok(Some(vec))
+        }
+        ByteStreamKind::New(vec) => Ok(Some(vec)),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use serde_json::json;

nativelink-config/src/backcompat.rs

@@ -180,7 +180,7 @@ pub struct PushConfig {
 #[serde(deny_unknown_fields)]
 pub struct ByteStreamConfig {
     /// Name of the store in the "stores" configuration.
-    pub cas_stores: HashMap<InstanceName, StoreRefName>,
+    pub cas_store: StoreRefName,
 
     /// Max number of bytes to send on each grpc stream chunk.
     /// According to <https://github.com/grpc/grpc.github.io/issues/371>
@@ -191,11 +191,6 @@ pub struct ByteStreamConfig {
     #[serde(default, deserialize_with = "convert_data_size_with_shellexpand")]
     pub max_bytes_per_stream: usize,
 
-    /// Maximum number of bytes to decode on each grpc stream chunk.
-    /// Default: 4 MiB
-    #[serde(default, deserialize_with = "convert_data_size_with_shellexpand")]
-    pub max_decoding_message_size: usize,
-
     /// In the event a client disconnects while uploading a blob, we will hold
     /// the internal stream open for this many seconds before closing it.
     /// This allows clients that disconnect to reconnect and continue uploading
@@ -206,6 +201,21 @@ pub struct ByteStreamConfig {
     pub persist_stream_on_disconnect_timeout: usize,
 }
 
+// Older bytestream config. All fields are as per the newer docs, but this requires
+// the hashed cas_stores v.s. the WithInstanceName approach. This should _not_ be updated
+// with newer fields, and eventually dropped
+#[derive(Deserialize, Serialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
+pub struct OldByteStreamConfig {
+    pub cas_stores: HashMap<InstanceName, StoreRefName>,
+    #[serde(default, deserialize_with = "convert_data_size_with_shellexpand")]
+    pub max_bytes_per_stream: usize,
+    #[serde(default, deserialize_with = "convert_data_size_with_shellexpand")]
+    pub max_decoding_message_size: usize,
+    #[serde(default, deserialize_with = "convert_duration_with_shellexpand")]
+    pub persist_stream_on_disconnect_timeout: usize,
+}
+
 #[derive(Deserialize, Serialize, Debug)]
 #[serde(deny_unknown_fields)]
 pub struct WorkerApiConfig {
@@ -322,7 +332,8 @@ pub struct ServicesConfig {
     /// This is the service used to stream data to and from the CAS.
     /// Bazel's protocol strongly encourages users to use this streaming
     /// interface to interact with the CAS when the data is large.
-    pub bytestream: Option<ByteStreamConfig>,
+    #[serde(default, deserialize_with = "super::backcompat::opt_bytestream")]
+    pub bytestream: Option<Vec<WithInstanceName<ByteStreamConfig>>>,
 
     /// These two are collectively the Remote Asset protocol, but it's
     /// defined as two separate services
@@ -462,7 +473,7 @@ pub enum ListenerConfig {
     Http(HttpListener),
 }
 
-#[derive(Deserialize, Serialize, Debug)]
+#[derive(Deserialize, Serialize, Debug, Default)]
 #[serde(deny_unknown_fields)]
 pub struct HttpListener {
     /// Address to listen on. Example: `127.0.0.1:8080` or `:8080` to listen
@@ -478,6 +489,11 @@ pub struct HttpListener {
     #[serde(default)]
     pub advanced_http: HttpServerConfig,
 
+    /// Maximum number of bytes to decode on each grpc stream chunk.
+    /// Default: 4 MiB
+    #[serde(default, deserialize_with = "convert_data_size_with_shellexpand")]
+    pub max_decoding_message_size: usize,
+
     /// Tls Configuration for this server.
     /// If not set, the server will not use TLS.
     ///

nativelink-config/src/cas_server.rs

@@ -85,7 +85,6 @@ rust_test_suite(
         "@crates//:http-body-util",
         "@crates//:hyper-1.7.0",
         "@crates//:hyper-util",
-        "@crates//:maplit",
         "@crates//:pretty_assertions",
         "@crates//:prost",
         "@crates//:prost-types",
@@ -112,7 +111,6 @@ rust_test(
         "@crates//:async-lock",
         "@crates//:hyper",
         "@crates//:hyper-util",
-        "@crates//:maplit",
         "@crates//:pretty_assertions",
         "@crates//:prost-types",
     ],

nativelink-service/BUILD.bazel

@@ -63,7 +63,6 @@ async-trait = "0.1.88"
 hex = { version = "0.4.3", default-features = false }
 hyper = "1.6.0"
 hyper-util = "0.1.11"
-maplit = "1.0.2"
 pretty_assertions = { version = "1.4.1", features = ["std"] }
 prost-types = { version = "0.13.5", default-features = false }
 sha2 = { version = "0.10.8", default-features = false }