hyperactor/
config.rs

1/*
2 * Copyright (c) Meta Platforms, Inc. and affiliates.
3 * All rights reserved.
4 *
5 * This source code is licensed under the BSD-style license found in the
6 * LICENSE file in the root directory of this source tree.
7 */
8
9//! Configuration keys and I/O for hyperactor.
10//!
11//! This module declares hyperactor-specific config keys.
12
13use std::io::Cursor;
14use std::io::Read;
15use std::path::PathBuf;
16use std::time::Duration;
17
18use hyperactor_config::AttrValue;
19use hyperactor_config::CONFIG;
20use hyperactor_config::ConfigAttr;
21use hyperactor_config::attrs::declare_attrs;
22use serde::Deserialize;
23use serde::Serialize;
24use typeuri::Named;
25
26/// Stores a PEM-encoded value, either specified directly or read from a file.
27#[derive(Clone, Debug, Serialize, Named)]
28#[named("hyperactor::config::Pem")]
29pub enum Pem {
30    /// Raw PEM value stored directly.
31    Value(Vec<u8>),
32    /// Path to a file containing the PEM data.
33    File(PathBuf),
34    /// Static path string (for use in const/static contexts).
35    StaticPath(&'static str),
36}
37
38// Custom Deserialize implementation because StaticPath contains &'static str
39impl<'de> Deserialize<'de> for Pem {
40    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
41    where
42        D: serde::Deserializer<'de>,
43    {
44        #[derive(Deserialize)]
45        enum PemDeserialize {
46            Value(Vec<u8>),
47            File(PathBuf),
48            // StaticPath deserializes as a File since we can't deserialize &'static str
49            StaticPath(String),
50        }
51
52        match PemDeserialize::deserialize(deserializer)? {
53            PemDeserialize::Value(v) => Ok(Pem::Value(v)),
54            PemDeserialize::File(p) => Ok(Pem::File(p)),
55            // Convert StaticPath string to File during deserialization
56            PemDeserialize::StaticPath(s) => Ok(Pem::File(PathBuf::from(s))),
57        }
58    }
59}
60
61impl AttrValue for Pem {
62    fn display(&self) -> String {
63        match self {
64            Pem::Value(data) => String::from_utf8_lossy(data).to_string(),
65            Pem::File(path) => path.display().to_string(),
66            Pem::StaticPath(path) => path.to_string(),
67        }
68    }
69
70    fn parse(value: &str) -> Result<Self, anyhow::Error> {
71        // PEM data starts with "-----BEGIN"
72        if value.trim_start().starts_with("-----BEGIN") {
73            Ok(Pem::Value(value.as_bytes().to_vec()))
74        } else {
75            Ok(Pem::File(PathBuf::from(value)))
76        }
77    }
78}
79
80impl Pem {
81    /// Returns a reader for the PEM data.
82    pub fn reader(&self) -> std::io::Result<Box<dyn Read + '_>> {
83        match self {
84            Pem::Value(data) => Ok(Box::new(Cursor::new(data))),
85            Pem::File(path) => Ok(Box::new(std::fs::File::open(path)?)),
86            Pem::StaticPath(path) => Ok(Box::new(std::fs::File::open(path)?)),
87        }
88    }
89}
90
91/// A bundle of PEM files for TLS configuration: CA certificate, server/client certificate, and private key.
92#[derive(Clone, Debug)]
93pub struct PemBundle {
94    /// The CA certificate used to verify peer certificates.
95    pub ca: Pem,
96    /// The certificate to present to peers.
97    pub cert: Pem,
98    /// The private key for the certificate.
99    pub key: Pem,
100}
101
102// Declare hyperactor-specific configuration keys
103declare_attrs! {
104    /// Maximum frame length for codec
105    @meta(CONFIG = ConfigAttr::new(
106        Some("HYPERACTOR_CODEC_MAX_FRAME_LENGTH".to_string()),
107        Some("codec_max_frame_length".to_string()),
108    ))
109    pub attr CODEC_MAX_FRAME_LENGTH: usize = 10 * 1024 * 1024 * 1024; // 10 GiB
110
111    /// Message delivery timeout
112    @meta(CONFIG = ConfigAttr::new(
113        Some("HYPERACTOR_MESSAGE_DELIVERY_TIMEOUT".to_string()),
114        Some("message_delivery_timeout".to_string()),
115    ))
116    pub attr MESSAGE_DELIVERY_TIMEOUT: Duration = Duration::from_secs(30);
117
118    /// Maximum number of terminated actor snapshots retained per
119    /// proc for post-mortem introspection. When the limit is
120    /// exceeded, the oldest entries are evicted.
121    @meta(CONFIG = ConfigAttr::new(
122        Some("HYPERACTOR_TERMINATED_SNAPSHOT_RETENTION".to_string()),
123        Some("terminated_snapshot_retention".to_string()),
124    ))
125    pub attr TERMINATED_SNAPSHOT_RETENTION: usize = 100;
126
127    /// Timeout to stop a proc.
128    @meta(CONFIG = ConfigAttr::new(
129        Some("HYPERACTOR_PROCESS_EXIT_TIMEOUT".to_string()),
130        Some("process_exit_timeout".to_string()),
131    ))
132    pub attr PROCESS_EXIT_TIMEOUT: Duration = Duration::from_secs(10);
133
134    /// Message acknowledgment interval
135    @meta(CONFIG = ConfigAttr::new(
136        Some("HYPERACTOR_MESSAGE_ACK_TIME_INTERVAL".to_string()),
137        Some("message_ack_time_interval".to_string()),
138    ))
139    pub attr MESSAGE_ACK_TIME_INTERVAL: Duration = Duration::from_millis(500);
140
141    /// Number of messages after which to send an acknowledgment
142    @meta(CONFIG = ConfigAttr::new(
143        Some("HYPERACTOR_MESSAGE_ACK_EVERY_N_MESSAGES".to_string()),
144        Some("message_ack_every_n_messages".to_string()),
145    ))
146    pub attr MESSAGE_ACK_EVERY_N_MESSAGES: u64 = 1000;
147
148    /// Default hop Time-To-Live for message envelopes.
149    @meta(CONFIG = ConfigAttr::new(
150        Some("HYPERACTOR_MESSAGE_TTL_DEFAULT".to_string()),
151        Some("message_ttl_default".to_string()),
152    ))
153    pub attr MESSAGE_TTL_DEFAULT : u8 = 64;
154
155    /// Maximum buffer size for split port messages
156    @meta(CONFIG = ConfigAttr::new(
157        Some("HYPERACTOR_SPLIT_MAX_BUFFER_SIZE".to_string()),
158        Some("split_max_buffer_size".to_string()),
159    ))
160    pub attr SPLIT_MAX_BUFFER_SIZE: usize = 5;
161
162    /// The maximum time an update can be buffered before being reduced.
163    @meta(CONFIG = ConfigAttr::new(
164        Some("HYPERACTOR_SPLIT_MAX_BUFFER_AGE".to_string()),
165        Some("split_max_buffer_age".to_string()),
166    ))
167    pub attr SPLIT_MAX_BUFFER_AGE: Duration = Duration::from_millis(50);
168
169    /// Timeout used by proc mesh for stopping an actor.
170    @meta(CONFIG = ConfigAttr::new(
171        Some("HYPERACTOR_STOP_ACTOR_TIMEOUT".to_string()),
172        Some("stop_actor_timeout".to_string()),
173    ))
174    pub attr STOP_ACTOR_TIMEOUT: Duration = Duration::from_secs(10);
175
176    /// Timeout used by proc for running the cleanup callback on an actor.
177    /// Should be less than the timeout for STOP_ACTOR_TIMEOUT.
178    @meta(CONFIG = ConfigAttr::new(
179        Some("HYPERACTOR_CLEANUP_TIMEOUT".to_string()),
180        Some("cleanup_timeout".to_string()),
181    ))
182    pub attr CLEANUP_TIMEOUT: Duration = Duration::from_secs(3);
183
184    /// How often to check for full MPSC channel on NetRx.
185    @meta(CONFIG = ConfigAttr::new(
186        Some("HYPERACTOR_CHANNEL_NET_RX_BUFFER_FULL_CHECK_INTERVAL".to_string()),
187        Some("channel_net_rx_buffer_full_check_interval".to_string()),
188    ))
189    pub attr CHANNEL_NET_RX_BUFFER_FULL_CHECK_INTERVAL: Duration = Duration::from_secs(5);
190
191    /// Kernel TCP keepalive idle period: the gap from last activity
192    /// until the kernel sends its first probe on connections created
193    /// by hyperactor's channel layer. On a healthy idle connection
194    /// this is also the probe cadence (each ACK resets the timer).
195    /// Total peer-death detection time is `idle + probe_budget` where
196    /// `probe_budget = TCP_KEEPALIVE_INTERVAL * TCP_KEEPALIVE_RETRIES`
197    /// (private constants in `channel::net`, currently 15s). Larger
198    /// values reduce keepalive ACK traffic at the cost of slower
199    /// peer-death detection.
200    @meta(CONFIG = ConfigAttr::new(
201        Some("HYPERACTOR_CHANNEL_TCP_KEEPALIVE_IDLE".to_string()),
202        Some("channel_tcp_keepalive_idle".to_string()),
203    ))
204    pub attr CHANNEL_TCP_KEEPALIVE_IDLE: Duration = Duration::from_secs(60);
205
206    /// Maximum time `Link::next()` spends retrying a failed connect
207    /// before giving up. Pairs with TCP keepalive: keepalive surfaces
208    /// peer death as an I/O error, then the connect-retry loop quits
209    /// after this bound and the outer NetTx loop terminates the link
210    /// with `TxStatus::Closed`.
211    @meta(CONFIG = ConfigAttr::new(
212        Some("HYPERACTOR_CHANNEL_RECONNECT_TIMEOUT".to_string()),
213        Some("channel_reconnect_timeout".to_string()),
214    ))
215    pub attr CHANNEL_RECONNECT_TIMEOUT: Duration = Duration::from_secs(60);
216
217    /// Sampling rate for logging message latency
218    /// Set to 0.01 for 1% sampling, 0.1 for 10% sampling, 0.90 for 90% sampling, etc.
219    @meta(CONFIG = ConfigAttr::new(
220        Some("HYPERACTOR_MESSAGE_LATENCY_SAMPLING_RATE".to_string()),
221        Some("message_latency_sampling_rate".to_string()),
222    ))
223    pub attr MESSAGE_LATENCY_SAMPLING_RATE: f32 = 0.01;
224
225    /// Whether to enable dest actor reordering buffer.
226    @meta(CONFIG = ConfigAttr::new(
227        Some("HYPERACTOR_ENABLE_DEST_ACTOR_REORDERING_BUFFER".to_string()),
228        Some("enable_dest_actor_reordering_buffer".to_string()),
229    ))
230    pub attr ENABLE_DEST_ACTOR_REORDERING_BUFFER: bool = true;
231
232    /// Timeout for [`Host::spawn`] to await proc readiness.
233    ///
234    /// Default: 30 seconds. If set to zero, disables the timeout and
235    /// waits indefinitely.
236    @meta(CONFIG = ConfigAttr::new(
237        Some("HYPERACTOR_HOST_SPAWN_READY_TIMEOUT".to_string()),
238        Some("host_spawn_ready_timeout".to_string()),
239    ))
240    pub attr HOST_SPAWN_READY_TIMEOUT: Duration = Duration::from_secs(30);
241
242    /// Heartbeat interval for server health metrics. The server emits a
243    /// heartbeat metric at this interval to indicate it is alive.
244    @meta(CONFIG = ConfigAttr::new(
245        Some("HYPERACTOR_SERVER_HEARTBEAT_INTERVAL".to_string()),
246        Some("server_heartbeat_interval".to_string()),
247    ))
248    pub attr SERVER_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(1);
249
250    /// Timeout for best-effort forwarder flush during proc/actor
251    /// teardown. If the remote side has already torn down its
252    /// networking, acks may never arrive; this timeout prevents the
253    /// flush from hanging indefinitely.
254    @meta(CONFIG = ConfigAttr::new(
255        Some("HYPERACTOR_FORWARDER_FLUSH_TIMEOUT".to_string()),
256        Some("forwarder_flush_timeout".to_string()),
257    ))
258    pub attr FORWARDER_FLUSH_TIMEOUT: Duration = Duration::from_secs(5);
259
260    /// Path to TLS certificate file for the 'tls' transport.
261    @meta(CONFIG = ConfigAttr::new(
262        Some("HYPERACTOR_TLS_CERT".to_string()),
263        Some("hyperactor_tls_cert".to_string()),
264    ).process_local())
265    pub attr TLS_CERT: Pem = Pem::StaticPath("/etc/hyperactor/tls/tls.crt");
266
267    /// Path to TLS private key file for the 'tls' transport.
268    @meta(CONFIG = ConfigAttr::new(
269        Some("HYPERACTOR_TLS_KEY".to_string()),
270        Some("hyperactor_tls_key".to_string()),
271    ).process_local())
272    pub attr TLS_KEY: Pem = Pem::StaticPath("/etc/hyperactor/tls/tls.key");
273
274    /// Path to TLS CA certificate file for the 'tls' transport.
275    @meta(CONFIG = ConfigAttr::new(
276        Some("HYPERACTOR_TLS_CA".to_string()),
277        Some("hyperactor_tls_ca".to_string()),
278    ).process_local())
279    pub attr TLS_CA: Pem = Pem::StaticPath("/etc/hyperactor/tls/ca.crt");
280}
281
282#[cfg(test)]
283mod tests {
284    use std::collections::HashSet;
285
286    use hyperactor_config::Attrs;
287    use hyperactor_config::from_env;
288    use indoc::indoc;
289
290    use super::*;
291
292    /// Like the `logs_assert` injected by `#[traced_test]`, but without scope
293    /// filtering. Use when asserting on events emitted outside the test's span
294    /// (e.g. from spawned tasks or panic hooks).
295    fn logs_assert_unscoped(f: impl Fn(&[&str]) -> Result<(), String>) {
296        let buf = tracing_test::internal::global_buf().lock().unwrap();
297        let logs_str = std::str::from_utf8(&buf).expect("Logs contain invalid UTF8");
298        let lines: Vec<&str> = logs_str.lines().collect();
299        match f(&lines) {
300            Ok(()) => {}
301            Err(msg) => panic!("{}", msg),
302        }
303    }
304
305    const CODEC_MAX_FRAME_LENGTH_DEFAULT: usize = 10 * 1024 * 1024 * 1024;
306
307    #[test]
308    fn test_default_config() {
309        let config = Attrs::new();
310        assert_eq!(
311            config[CODEC_MAX_FRAME_LENGTH],
312            CODEC_MAX_FRAME_LENGTH_DEFAULT
313        );
314        assert_eq!(config[MESSAGE_DELIVERY_TIMEOUT], Duration::from_secs(30));
315        assert_eq!(
316            config[MESSAGE_ACK_TIME_INTERVAL],
317            Duration::from_millis(500)
318        );
319        assert_eq!(config[MESSAGE_ACK_EVERY_N_MESSAGES], 1000);
320        assert_eq!(config[SPLIT_MAX_BUFFER_SIZE], 5);
321    }
322
323    #[tracing_test::traced_test]
324    #[test]
325    // TODO: OSS: The logs_assert function returned an error: missing log lines: {"# export HYPERACTOR_DEFAULT_ENCODING=serde_multipart", ...}
326    #[cfg_attr(not(fbcode_build), ignore)]
327    fn test_from_env() {
328        // Set environment variables
329        // SAFETY: TODO: Audit that the environment access only happens in single-threaded code.
330        unsafe { std::env::set_var("HYPERACTOR_CODEC_MAX_FRAME_LENGTH", "1024") };
331        // SAFETY: TODO: Audit that the environment access only happens in single-threaded code.
332        unsafe { std::env::set_var("HYPERACTOR_MESSAGE_DELIVERY_TIMEOUT", "60s") };
333
334        let config = from_env();
335
336        assert_eq!(config[CODEC_MAX_FRAME_LENGTH], 1024);
337        assert_eq!(config[MESSAGE_DELIVERY_TIMEOUT], Duration::from_mins(1));
338        assert_eq!(
339            config[MESSAGE_ACK_TIME_INTERVAL],
340            Duration::from_millis(500)
341        ); // Default value
342
343        let expected_lines: HashSet<&str> = indoc! {"
344            # export HYPERACTOR_MESSAGE_LATENCY_SAMPLING_RATE=0.01
345            # export HYPERACTOR_CHANNEL_NET_RX_BUFFER_FULL_CHECK_INTERVAL=5s
346            # export HYPERACTOR_STOP_ACTOR_TIMEOUT=10s
347            # export HYPERACTOR_SPLIT_MAX_BUFFER_SIZE=5
348            # export HYPERACTOR_MESSAGE_TTL_DEFAULT=64
349            # export HYPERACTOR_MESSAGE_ACK_EVERY_N_MESSAGES=1000
350            # export HYPERACTOR_MESSAGE_ACK_TIME_INTERVAL=500ms
351            # export HYPERACTOR_PROCESS_EXIT_TIMEOUT=10s
352            # export HYPERACTOR_MESSAGE_DELIVERY_TIMEOUT=30s
353            export HYPERACTOR_MESSAGE_DELIVERY_TIMEOUT=1m
354            # export HYPERACTOR_CODEC_MAX_FRAME_LENGTH=10737418240
355            export HYPERACTOR_CODEC_MAX_FRAME_LENGTH=1024
356            # export HYPERACTOR_CLEANUP_TIMEOUT=3s
357            # export HYPERACTOR_SPLIT_MAX_BUFFER_AGE=50ms
358            # export HYPERACTOR_DEFAULT_ENCODING=serde_multipart
359            # export HYPERACTOR_HOST_SPAWN_READY_TIMEOUT=30s
360        "}
361        .trim_end()
362        .lines()
363        .collect();
364
365        // For some reason, logs_contaqin fails to find these lines individually
366        // (possibly to do with the fact that we have newlines in our log entries);
367        // instead, we test it manually.
368        logs_assert_unscoped(|logged_lines: &[&str]| {
369            let mut expected_lines = expected_lines.clone(); // this is an `Fn` closure
370            for logged in logged_lines {
371                expected_lines.remove(logged);
372            }
373
374            if expected_lines.is_empty() {
375                Ok(())
376            } else {
377                Err(format!("missing log lines: {:?}", expected_lines))
378            }
379        });
380
381        // Clean up
382        // SAFETY: TODO: Audit that the environment access only happens in single-threaded code.
383        unsafe { std::env::remove_var("HYPERACTOR_CODEC_MAX_FRAME_LENGTH") };
384        // SAFETY: TODO: Audit that the environment access only happens in single-threaded code.
385        unsafe { std::env::remove_var("HYPERACTOR_MESSAGE_DELIVERY_TIMEOUT_SECS") };
386    }
387
388    #[test]
389    fn test_defaults() {
390        // Test that empty config now returns defaults via get_or_default
391        let config = Attrs::new();
392
393        // Verify that the config is empty (no values explicitly set)
394        assert!(config.is_empty());
395
396        // But getters should still return the defaults from the keys
397        assert_eq!(
398            config[CODEC_MAX_FRAME_LENGTH],
399            CODEC_MAX_FRAME_LENGTH_DEFAULT
400        );
401        assert_eq!(config[MESSAGE_DELIVERY_TIMEOUT], Duration::from_secs(30));
402        assert_eq!(
403            config[MESSAGE_ACK_TIME_INTERVAL],
404            Duration::from_millis(500)
405        );
406        assert_eq!(config[MESSAGE_ACK_EVERY_N_MESSAGES], 1000);
407        assert_eq!(config[SPLIT_MAX_BUFFER_SIZE], 5);
408
409        // Verify the keys have defaults
410        assert!(CODEC_MAX_FRAME_LENGTH.has_default());
411        assert!(MESSAGE_DELIVERY_TIMEOUT.has_default());
412        assert!(MESSAGE_ACK_TIME_INTERVAL.has_default());
413        assert!(MESSAGE_ACK_EVERY_N_MESSAGES.has_default());
414        assert!(SPLIT_MAX_BUFFER_SIZE.has_default());
415
416        // Verify we can get defaults directly from keys
417        assert_eq!(
418            CODEC_MAX_FRAME_LENGTH.default(),
419            Some(&(CODEC_MAX_FRAME_LENGTH_DEFAULT))
420        );
421        assert_eq!(
422            MESSAGE_DELIVERY_TIMEOUT.default(),
423            Some(&Duration::from_secs(30))
424        );
425        assert_eq!(
426            MESSAGE_ACK_TIME_INTERVAL.default(),
427            Some(&Duration::from_millis(500))
428        );
429        assert_eq!(MESSAGE_ACK_EVERY_N_MESSAGES.default(), Some(&1000));
430        assert_eq!(SPLIT_MAX_BUFFER_SIZE.default(), Some(&5));
431    }
432
433    #[test]
434    fn test_serialization_only_includes_set_values() {
435        let mut config = Attrs::new();
436
437        // Initially empty, serialization should be empty
438        let serialized = serde_json::to_string(&config).unwrap();
439        assert_eq!(serialized, "{}");
440
441        config[CODEC_MAX_FRAME_LENGTH] = 1024;
442
443        let serialized = serde_json::to_string(&config).unwrap();
444        assert!(serialized.contains("codec_max_frame_length"));
445        assert!(!serialized.contains("message_delivery_timeout")); // Default not serialized
446
447        // Deserialize back
448        let restored_config: Attrs = serde_json::from_str(&serialized).unwrap();
449
450        // Custom value should be preserved
451        assert_eq!(restored_config[CODEC_MAX_FRAME_LENGTH], 1024);
452
453        // Defaults should still work for other values
454        assert_eq!(
455            restored_config[MESSAGE_DELIVERY_TIMEOUT],
456            Duration::from_secs(30)
457        );
458    }
459}
hyperactor/config.rs

hyperactor/
config.rs
