hyperactor/
environment.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//! The persistent, language-independent environment carried by an actor
10//! [`Instance`](crate::proc::Instance).
11//!
12//! An [`ActorEnvironment`] is a small, immutable set of typed attributes fixed
13//! when the instance is constructed. It is inherited by local children and
14//! serialized to remote children, giving stable actor context a native home
15//! that does not depend on the actor's implementation language.
16//!
17//! `hyperactor` owns how this environment is carried through actor creation,
18//! while higher-level crates own the values it contains. Those crates declare
19//! typed [`Key`] attributes. Putting those values in a fixed struct would move
20//! higher-level concepts into `hyperactor` and require core runtime and wire
21//! changes for each new kind of inherited context. This is not an untyped
22//! general-purpose map: the API preserves each value's type, enforces fixed
23//! bounds, and requires mutable ownership for construction updates.
24//!
25//! # AENV invariant registry
26//!
27//! - **AENV-0 (value integrity):** every `ActorEnvironment` is structurally
28//!   valid, bounded, unique-keyed, and updated atomically.
29//! - **AENV-1 (instance ownership):** every [`Instance`](crate::proc::Instance)
30//!   stores exactly one immutable environment in its type-erased instance cell,
31//!   fixed at construction and exposed read-only.
32//! - **AENV-2 (local inheritance):** ordinary local child/client construction
33//!   derives its parent's exact environment from that cell.
34//! - **AENV-3 (remote inheritance):** native remote-spawn adapters transport
35//!   the caller-shaped environment; nested remote spawn derives and transports
36//!   its environment independently.
37//! - **AENV-4 (caller shaping):** spawn callers put all construction context in
38//!   one environment. When its create cast carries `CAST_POINT`, proc-mesh
39//!   construction replaces the environmental value with the point assigned to
40//!   the new actor; ordinary local child construction inherits the parent's
41//!   exact environment. [`RemoteSpawn::new`](crate::actor::RemoteSpawn) reads
42//!   that environment, and the runtime stores it unchanged.
43//! - **AENV-5 (runtime neutrality):** inheritance is implemented by native
44//!   `Instance`/spawn plumbing and does not depend on the actor implementation
45//!   language.
46
47use std::collections::HashMap;
48use std::collections::HashSet;
49use std::sync::Arc;
50
51use hyperactor_config::AttrValue;
52use hyperactor_config::Flattrs;
53use hyperactor_config::FlattrsValidationError;
54use hyperactor_config::Key;
55use serde::Deserialize;
56use serde::Deserializer;
57use serde::Serialize;
58use serde::Serializer;
59use serde::de::DeserializeOwned;
60use serde::de::Error as _;
61use serde_multipart::Part;
62
63const MAX_ACTOR_ENVIRONMENT_ENTRIES: usize = 32;
64const MAX_ACTOR_ENVIRONMENT_BYTES: usize = 16 * 1024;
65
66/// An error constructing or decoding an [`ActorEnvironment`].
67#[derive(Debug, thiserror::Error)]
68#[non_exhaustive]
69pub enum ActorEnvironmentError {
70    /// One serialized value cannot fit within the environment's byte bound.
71    #[error("serialized actor environment value is {bytes} bytes; maximum is {maximum}")]
72    ValueTooLarge {
73        /// The serialized value size in bytes.
74        bytes: usize,
75        /// The maximum accepted serialized value size.
76        maximum: usize,
77    },
78
79    /// The encoded entry count exceeds the environment's fixed bound.
80    #[error("actor environment has {entries} entries; maximum is {maximum}")]
81    TooManyEntries {
82        /// The number of encoded entries.
83        entries: usize,
84        /// The maximum accepted number of entries.
85        maximum: usize,
86    },
87
88    /// The encoded environment exceeds its fixed byte bound.
89    #[error("actor environment is {bytes} bytes; maximum is {maximum}")]
90    TooLarge {
91        /// The encoded size in bytes.
92        bytes: usize,
93        /// The maximum accepted encoded size.
94        maximum: usize,
95    },
96
97    /// A key occurs more than once in the encoded environment.
98    #[error("actor environment contains duplicate key hash {key_hash:#x}")]
99    DuplicateKey {
100        /// The repeated key hash.
101        key_hash: u64,
102    },
103
104    /// A typed value could not be serialized into the environment.
105    #[error("failed to serialize actor environment value: {0}")]
106    ValueSerialization(#[from] bincode::error::EncodeError),
107
108    /// The underlying `Flattrs` wire structure is malformed.
109    #[error("invalid actor environment encoding: {0}")]
110    InvalidEncoding(#[from] FlattrsValidationError),
111}
112
113/// The persistent environment owned by an actor instance.
114///
115/// The inner buffer is never exposed. Clones share the immutable buffer; a
116/// subsequent builder mutation creates a bounded private copy. Once placed on
117/// an [`Instance`](crate::proc::Instance), only a shared view is exposed
118/// (AENV-1).
119#[derive(Clone)]
120pub struct ActorEnvironment(Arc<Flattrs>);
121
122impl Default for ActorEnvironment {
123    fn default() -> Self {
124        Self(Arc::new(Flattrs::new()))
125    }
126}
127
128impl ActorEnvironment {
129    /// Store `value` under `key` while enforcing the environment's entry and
130    /// serialized-size bounds.
131    ///
132    /// This is a construction API. The environment stored on an `Instance` is
133    /// immutable because the runtime exposes no mutable instance accessor.
134    pub fn set<T: Serialize>(
135        &mut self,
136        key: Key<T>,
137        value: T,
138    ) -> Result<(), ActorEnvironmentError> {
139        let serialized = bincode::serde::encode_to_vec(value, bincode::config::legacy())?;
140        if serialized.len() > MAX_ACTOR_ENVIRONMENT_BYTES {
141            return Err(ActorEnvironmentError::ValueTooLarge {
142                bytes: serialized.len(),
143                maximum: MAX_ACTOR_ENVIRONMENT_BYTES,
144            });
145        }
146        let mut candidate = self.0.as_ref().clone();
147        candidate.set_serialized(key.key_hash(), &serialized);
148        validate(&candidate)?;
149        self.0 = Arc::new(candidate);
150        Ok(())
151    }
152
153    /// Read the value stored under `key`, if present.
154    pub fn get<T: AttrValue + DeserializeOwned>(&self, key: Key<T>) -> Option<T> {
155        self.0.get(key)
156    }
157
158    /// Whether the environment holds no entries.
159    pub fn is_empty(&self) -> bool {
160        self.0.is_empty()
161    }
162
163    /// The number of entries in the environment.
164    pub fn len(&self) -> usize {
165        self.0.len()
166    }
167}
168
169fn validate(attrs: &Flattrs) -> Result<(), ActorEnvironmentError> {
170    attrs.validate_wire()?;
171    let encoded_bytes = attrs.encoded_len();
172    if encoded_bytes > MAX_ACTOR_ENVIRONMENT_BYTES {
173        return Err(ActorEnvironmentError::TooLarge {
174            bytes: encoded_bytes,
175            maximum: MAX_ACTOR_ENVIRONMENT_BYTES,
176        });
177    }
178
179    let declared = attrs.len();
180    if declared > MAX_ACTOR_ENVIRONMENT_ENTRIES {
181        return Err(ActorEnvironmentError::TooManyEntries {
182            entries: declared,
183            maximum: MAX_ACTOR_ENVIRONMENT_ENTRIES,
184        });
185    }
186
187    let mut seen = HashSet::with_capacity(declared);
188    for (key_hash, _) in attrs.iter() {
189        if !seen.insert(key_hash) {
190            return Err(ActorEnvironmentError::DuplicateKey { key_hash });
191        }
192    }
193    Ok(())
194}
195
196impl Serialize for ActorEnvironment {
197    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
198    where
199        S: Serializer,
200    {
201        self.0.serialize(serializer)
202    }
203}
204
205impl<'de> Deserialize<'de> for ActorEnvironment {
206    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
207    where
208        D: Deserializer<'de>,
209    {
210        // This bound controls accepted instance state. The transport frame
211        // limit is responsible for bounding allocation before field decoding.
212        let part = Part::deserialize(deserializer)?;
213        let encoded_bytes = part.len();
214        if encoded_bytes > MAX_ACTOR_ENVIRONMENT_BYTES {
215            return Err(D::Error::custom(ActorEnvironmentError::TooLarge {
216                bytes: encoded_bytes,
217                maximum: MAX_ACTOR_ENVIRONMENT_BYTES,
218            }));
219        }
220        let attrs = Flattrs::from_part(part);
221        validate(&attrs).map_err(D::Error::custom)?;
222        Ok(Self(Arc::new(attrs)))
223    }
224}
225
226impl std::fmt::Debug for ActorEnvironment {
227    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
228        f.debug_struct("ActorEnvironment")
229            .field("len", &self.0.len())
230            .finish()
231    }
232}
233
234/// Equality over the `(key hash, serialized bytes)` set, independent of
235/// insertion order. Construction and deserialization enforce one entry per key
236/// hash.
237impl PartialEq for ActorEnvironment {
238    fn eq(&self, other: &Self) -> bool {
239        if Arc::ptr_eq(&self.0, &other.0) {
240            return true;
241        }
242        if self.0.len() != other.0.len() {
243            return false;
244        }
245        let mine: HashMap<u64, &[u8]> = self.0.iter().collect();
246        other
247            .0
248            .iter()
249            .all(|(key_hash, value)| mine.get(&key_hash) == Some(&value))
250    }
251}
252
253impl Eq for ActorEnvironment {}
254
255#[cfg(test)]
256mod tests {
257    use hyperactor_config::declare_attrs;
258
259    use super::*;
260
261    declare_attrs! {
262        attr TEST_TAG: u64;
263        attr TEST_LABEL: String;
264        attr TEST_BYTES: Vec<u8>;
265    }
266
267    fn decode_raw(raw: Vec<u8>) -> Result<ActorEnvironment, bincode::error::DecodeError> {
268        let attrs = Flattrs::from_part(Part::from(raw));
269        let encoded = bincode::serde::encode_to_vec(attrs, bincode::config::legacy())
270            .expect("serialize raw Flattrs");
271        bincode::serde::decode_from_slice(&encoded, bincode::config::legacy())
272            .map(|(value, _)| value)
273    }
274
275    // Malformed-buffer tests must encode the private layout directly because
276    // the public Flattrs builders cannot construct invalid wire shapes.
277    fn append_entry(raw: &mut Vec<u8>, key_hash: u64, value: &[u8]) {
278        raw.extend_from_slice(&key_hash.to_le_bytes());
279        raw.extend_from_slice(&(value.len() as u32).to_le_bytes());
280        raw.extend_from_slice(value);
281    }
282
283    #[test]
284    fn default_accepts_first_insertion() {
285        let mut env = ActorEnvironment::default();
286        assert!(env.is_empty());
287        env.set(TEST_TAG, 7u64).expect("insert first value");
288        assert_eq!(env.get(TEST_TAG), Some(7u64));
289        assert_eq!(env.len(), 1);
290    }
291
292    #[test]
293    fn serde_roundtrips() {
294        let mut env = ActorEnvironment::default();
295        env.set(TEST_TAG, 42u64).expect("insert tag");
296        env.set(TEST_LABEL, "root".to_string())
297            .expect("insert label");
298
299        let bytes =
300            bincode::serde::encode_to_vec(&env, bincode::config::legacy()).expect("serialize");
301        let restored: ActorEnvironment =
302            bincode::serde::decode_from_slice(&bytes, bincode::config::legacy())
303                .map(|(value, _)| value)
304                .expect("deserialize");
305
306        assert_eq!(restored.get(TEST_TAG), Some(42u64));
307        assert_eq!(restored.get(TEST_LABEL), Some("root".to_string()));
308        assert_eq!(restored, env);
309    }
310
311    #[test]
312    fn equality_is_order_independent() {
313        let mut a = ActorEnvironment::default();
314        a.set(TEST_TAG, 1u64).expect("insert tag");
315        a.set(TEST_LABEL, "x".to_string()).expect("insert label");
316
317        let mut b = ActorEnvironment::default();
318        b.set(TEST_LABEL, "x".to_string()).expect("insert label");
319        b.set(TEST_TAG, 1u64).expect("insert tag");
320
321        assert_eq!(a, b);
322
323        let mut c = ActorEnvironment::default();
324        c.set(TEST_TAG, 2u64).expect("insert tag");
325        c.set(TEST_LABEL, "x".to_string()).expect("insert label");
326        assert_ne!(a, c);
327    }
328
329    #[test]
330    fn serde_rejects_duplicate_keys() {
331        let value = bincode::serde::encode_to_vec(1u64, bincode::config::legacy())
332            .expect("serialize value");
333        let mut raw = 2u16.to_le_bytes().to_vec();
334        append_entry(&mut raw, TEST_TAG.key_hash(), &value);
335        append_entry(&mut raw, TEST_TAG.key_hash(), &value);
336
337        assert!(decode_raw(raw).is_err(), "duplicate keys must be rejected");
338    }
339
340    #[test]
341    fn serde_rejects_truncated_and_trailing_buffers() {
342        assert!(
343            decode_raw(vec![1]).is_err(),
344            "a partial count header must be rejected"
345        );
346        assert!(
347            decode_raw(1u16.to_le_bytes().to_vec()).is_err(),
348            "a declared entry without entry bytes must be rejected"
349        );
350        assert!(
351            decode_raw(vec![0, 0, 1]).is_err(),
352            "bytes after the declared entries must be rejected"
353        );
354    }
355
356    #[test]
357    fn set_rejects_oversized_values_without_mutating() {
358        let mut env = ActorEnvironment::default();
359        let err = env
360            .set(TEST_BYTES, vec![0; MAX_ACTOR_ENVIRONMENT_BYTES])
361            .expect_err("oversized value must be rejected");
362        assert!(matches!(err, ActorEnvironmentError::ValueTooLarge { .. }));
363        assert!(env.is_empty(), "a rejected insertion must be atomic");
364    }
365
366    #[test]
367    fn set_rejects_entry_and_aggregate_bounds_atomically() {
368        let mut raw = (MAX_ACTOR_ENVIRONMENT_ENTRIES as u16)
369            .to_le_bytes()
370            .to_vec();
371        for offset in 1..=MAX_ACTOR_ENVIRONMENT_ENTRIES {
372            append_entry(
373                &mut raw,
374                TEST_TAG.key_hash().wrapping_add(offset as u64),
375                &[],
376            );
377        }
378        let mut full = decode_raw(raw).expect("decode environment at the entry bound");
379        let before = full.clone();
380        let err = full
381            .set(TEST_TAG, 1u64)
382            .expect_err("a 33rd entry must be rejected");
383        assert!(matches!(err, ActorEnvironmentError::TooManyEntries { .. }));
384        assert_eq!(full, before, "a rejected insertion must be atomic");
385
386        let mut large = ActorEnvironment::default();
387        large
388            .set(TEST_BYTES, vec![0; MAX_ACTOR_ENVIRONMENT_BYTES - 64])
389            .expect("the first value fits within the aggregate bound");
390        let before = large.clone();
391        let err = large
392            .set(TEST_LABEL, "x".repeat(128))
393            .expect_err("aggregate encoded size must be enforced");
394        assert!(matches!(err, ActorEnvironmentError::TooLarge { .. }));
395        assert_eq!(large, before, "a rejected insertion must be atomic");
396    }
397
398    #[test]
399    fn serde_rejects_policy_bounds() {
400        let mut too_many = ((MAX_ACTOR_ENVIRONMENT_ENTRIES + 1) as u16)
401            .to_le_bytes()
402            .to_vec();
403        for offset in 1..=MAX_ACTOR_ENVIRONMENT_ENTRIES + 1 {
404            append_entry(
405                &mut too_many,
406                TEST_TAG.key_hash().wrapping_add(offset as u64),
407                &[],
408            );
409        }
410        assert!(
411            decode_raw(too_many).is_err(),
412            "an oversized entry set must be rejected"
413        );
414
415        let mut too_large = 1u16.to_le_bytes().to_vec();
416        append_entry(
417            &mut too_large,
418            TEST_BYTES.key_hash(),
419            &vec![0; MAX_ACTOR_ENVIRONMENT_BYTES],
420        );
421        let err = decode_raw(too_large).expect_err("oversized environment must fail");
422        assert!(
423            err.to_string().contains("maximum"),
424            "a structurally valid oversized environment must hit the byte bound: {err}"
425        );
426
427        let mut boundary = ActorEnvironment::default();
428        boundary
429            .set(TEST_BYTES, Vec::<u8>::new())
430            .expect("measure one-entry encoding overhead");
431        let payload_len = MAX_ACTOR_ENVIRONMENT_BYTES - boundary.0.encoded_len();
432        boundary
433            .set(TEST_BYTES, vec![0; payload_len])
434            .expect("the exact byte boundary must be accepted");
435        assert_eq!(boundary.0.encoded_len(), MAX_ACTOR_ENVIRONMENT_BYTES);
436        let encoded = bincode::serde::encode_to_vec(&boundary, bincode::config::legacy())
437            .expect("serialize boundary environment");
438        let restored: ActorEnvironment =
439            bincode::serde::decode_from_slice(&encoded, bincode::config::legacy())
440                .map(|(value, _)| value)
441                .expect("deserialize boundary environment");
442        assert_eq!(restored, boundary);
443    }
444}
hyperactor/environment.rs

hyperactor/
environment.rs
