hydro_lang/deploy/maelstrom/
deploy_maelstrom.rs

1//! Deployment backend for Hydro that targets Maelstrom for distributed systems testing.
2//!
3//! Maelstrom is a workbench for learning distributed systems by writing your own.
4//! This backend compiles Hydro programs to binaries that communicate via Maelstrom's
5//! stdin/stdout JSON protocol.
6
7use std::cell::RefCell;
8use std::future::Future;
9use std::io::{BufRead, BufReader, Error};
10use std::path::PathBuf;
11use std::pin::Pin;
12use std::process::Stdio;
13use std::rc::Rc;
14
15use bytes::{Bytes, BytesMut};
16use dfir_lang::graph::DfirGraph;
17use futures::{Sink, Stream};
18use serde::Serialize;
19use serde::de::DeserializeOwned;
20use stageleft::{QuotedWithContext, RuntimeData};
21
22use super::deploy_runtime_maelstrom::*;
23use crate::compile::builder::ExternalPortId;
24use crate::compile::deploy_provider::{ClusterSpec, Deploy, Node, RegisterPort};
25use crate::compile::trybuild::generate::{LinkingMode, create_graph_trybuild};
26use crate::location::dynamic::LocationId;
27use crate::location::member_id::TaglessMemberId;
28use crate::location::{LocationKey, MembershipEvent, NetworkHint};
29
30/// Deployment backend that targets Maelstrom for distributed systems testing.
31///
32/// This backend compiles Hydro programs to binaries that communicate via Maelstrom's
33/// stdin/stdout JSON protocol. It is restricted to programs with:
34/// - Exactly one cluster (no processes)
35/// - A single external input channel for client communication
36pub enum MaelstromDeploy {}
37
38impl<'a> Deploy<'a> for MaelstromDeploy {
39    type Meta = ();
40    type InstantiateEnv = MaelstromDeployment;
41
42    type Process = MaelstromProcess;
43    type Cluster = MaelstromCluster;
44    type External = MaelstromExternal;
45
46    fn o2o_sink_source(
47        _env: &mut Self::InstantiateEnv,
48        _p1: &Self::Process,
49        _p1_port: &<Self::Process as Node>::Port,
50        _p2: &Self::Process,
51        _p2_port: &<Self::Process as Node>::Port,
52        _name: Option<&str>,
53        _networking_info: &crate::networking::NetworkingInfo,
54        _external_types: Option<(&syn::Type, &syn::Type)>,
55    ) -> (syn::Expr, syn::Expr) {
56        panic!("Maelstrom deployment does not support processes, only clusters")
57    }
58
59    fn o2o_connect(
60        _p1: &Self::Process,
61        _p1_port: &<Self::Process as Node>::Port,
62        _p2: &Self::Process,
63        _p2_port: &<Self::Process as Node>::Port,
64    ) -> Box<dyn FnOnce()> {
65        panic!("Maelstrom deployment does not support processes, only clusters")
66    }
67
68    fn o2m_sink_source(
69        _env: &mut Self::InstantiateEnv,
70        _p1: &Self::Process,
71        _p1_port: &<Self::Process as Node>::Port,
72        _c2: &Self::Cluster,
73        _c2_port: &<Self::Cluster as Node>::Port,
74        _name: Option<&str>,
75        _networking_info: &crate::networking::NetworkingInfo,
76        _external_types: Option<(&syn::Type, &syn::Type)>,
77    ) -> (syn::Expr, syn::Expr) {
78        panic!("Maelstrom deployment does not support processes, only clusters")
79    }
80
81    fn o2m_connect(
82        _p1: &Self::Process,
83        _p1_port: &<Self::Process as Node>::Port,
84        _c2: &Self::Cluster,
85        _c2_port: &<Self::Cluster as Node>::Port,
86    ) -> Box<dyn FnOnce()> {
87        panic!("Maelstrom deployment does not support processes, only clusters")
88    }
89
90    fn m2o_sink_source(
91        _env: &mut Self::InstantiateEnv,
92        _c1: &Self::Cluster,
93        _c1_port: &<Self::Cluster as Node>::Port,
94        _p2: &Self::Process,
95        _p2_port: &<Self::Process as Node>::Port,
96        _name: Option<&str>,
97        _networking_info: &crate::networking::NetworkingInfo,
98        _external_types: Option<(&syn::Type, &syn::Type)>,
99    ) -> (syn::Expr, syn::Expr) {
100        panic!("Maelstrom deployment does not support processes, only clusters")
101    }
102
103    fn m2o_connect(
104        _c1: &Self::Cluster,
105        _c1_port: &<Self::Cluster as Node>::Port,
106        _p2: &Self::Process,
107        _p2_port: &<Self::Process as Node>::Port,
108    ) -> Box<dyn FnOnce()> {
109        panic!("Maelstrom deployment does not support processes, only clusters")
110    }
111
112    fn m2m_sink_source(
113        env: &mut Self::InstantiateEnv,
114        _c1: &Self::Cluster,
115        _c1_port: &<Self::Cluster as Node>::Port,
116        _c2: &Self::Cluster,
117        _c2_port: &<Self::Cluster as Node>::Port,
118        _name: Option<&str>,
119        networking_info: &crate::networking::NetworkingInfo,
120        _external_types: Option<(&syn::Type, &syn::Type)>,
121    ) -> (syn::Expr, syn::Expr) {
122        use crate::networking::{NetworkingInfo, TcpFault};
123        match networking_info {
124            NetworkingInfo::Tcp { fault } => match (fault, env.nemesis.as_deref()) {
125                (TcpFault::Lossy | TcpFault::LossyDelayedForever, _) => {} /* lossy/delayed are always allowed */
126                (_, None) => {} // no nemesis means any fault model is fine
127                (TcpFault::FailStop, Some("partition")) => {
128                    panic!(
129                        "Maelstrom partition nemesis requires lossy networking, but fail_stop was used. \
130                         Use `TCP.lossy().bincode()` or `TCP.lossy_delayed_forever().bincode()` instead of `TCP.fail_stop().bincode()`."
131                    );
132                }
133                (TcpFault::FailStop, Some(_)) => {} // other nemeses are fine with fail_stop
134            },
135        }
136        deploy_maelstrom_m2m(RuntimeData::new("__hydro_lang_maelstrom_meta"))
137    }
138
139    fn m2m_connect(
140        _c1: &Self::Cluster,
141        _c1_port: &<Self::Cluster as Node>::Port,
142        _c2: &Self::Cluster,
143        _c2_port: &<Self::Cluster as Node>::Port,
144    ) -> Box<dyn FnOnce()> {
145        // No runtime connection needed for Maelstrom - all routing is via stdin/stdout
146        Box::new(|| {})
147    }
148
149    fn e2o_many_source(
150        _extra_stmts: &mut Vec<syn::Stmt>,
151        _p2: &Self::Process,
152        _p2_port: &<Self::Process as Node>::Port,
153        _codec_type: &syn::Type,
154        _shared_handle: String,
155    ) -> syn::Expr {
156        panic!("Maelstrom deployment does not support processes, only clusters")
157    }
158
159    fn e2o_many_sink(_shared_handle: String) -> syn::Expr {
160        panic!("Maelstrom deployment does not support processes, only clusters")
161    }
162
163    fn e2o_source(
164        _extra_stmts: &mut Vec<syn::Stmt>,
165        _p1: &Self::External,
166        _p1_port: &<Self::External as Node>::Port,
167        _p2: &Self::Process,
168        _p2_port: &<Self::Process as Node>::Port,
169        _codec_type: &syn::Type,
170        _shared_handle: String,
171    ) -> syn::Expr {
172        panic!("Maelstrom deployment does not support processes, only clusters")
173    }
174
175    fn e2o_connect(
176        _p1: &Self::External,
177        _p1_port: &<Self::External as Node>::Port,
178        _p2: &Self::Process,
179        _p2_port: &<Self::Process as Node>::Port,
180        _many: bool,
181        _server_hint: NetworkHint,
182    ) -> Box<dyn FnOnce()> {
183        panic!("Maelstrom deployment does not support processes, only clusters")
184    }
185
186    fn o2e_sink(
187        _p1: &Self::Process,
188        _p1_port: &<Self::Process as Node>::Port,
189        _p2: &Self::External,
190        _p2_port: &<Self::External as Node>::Port,
191        _shared_handle: String,
192    ) -> syn::Expr {
193        panic!("Maelstrom deployment does not support processes, only clusters")
194    }
195
196    fn cluster_ids(
197        _of_cluster: LocationKey,
198    ) -> impl QuotedWithContext<'a, &'a [TaglessMemberId], ()> + Clone + 'a {
199        cluster_members(RuntimeData::new("__hydro_lang_maelstrom_meta"), _of_cluster)
200    }
201
202    fn cluster_self_id() -> impl QuotedWithContext<'a, TaglessMemberId, ()> + Clone + 'a {
203        cluster_self_id(RuntimeData::new("__hydro_lang_maelstrom_meta"))
204    }
205
206    fn cluster_membership_stream(
207        _env: &mut Self::InstantiateEnv,
208        _at_location: &LocationId,
209        location_id: &LocationId,
210    ) -> impl QuotedWithContext<'a, Box<dyn Stream<Item = (TaglessMemberId, MembershipEvent)> + Unpin>, ()>
211    {
212        cluster_membership_stream(location_id)
213    }
214}
215
216/// A dummy process type for Maelstrom (processes are not supported).
217#[derive(Clone)]
218pub struct MaelstromProcess {
219    _private: (),
220}
221
222impl Node for MaelstromProcess {
223    type Port = String;
224    type Meta = ();
225    type InstantiateEnv = MaelstromDeployment;
226
227    fn next_port(&self) -> Self::Port {
228        panic!("Maelstrom deployment does not support processes")
229    }
230
231    fn update_meta(&self, _meta: &Self::Meta) {}
232
233    fn instantiate(
234        &self,
235        _env: &mut Self::InstantiateEnv,
236        _meta: &mut Self::Meta,
237        _graph: DfirGraph,
238        _extra_stmts: &[syn::Stmt],
239        _sidecars: &[syn::Expr],
240    ) {
241        panic!("Maelstrom deployment does not support processes")
242    }
243}
244
245/// Represents a cluster in Maelstrom deployment.
246#[derive(Clone)]
247pub struct MaelstromCluster {
248    next_port: Rc<RefCell<usize>>,
249    name_hint: Option<String>,
250}
251
252impl Node for MaelstromCluster {
253    type Port = String;
254    type Meta = ();
255    type InstantiateEnv = MaelstromDeployment;
256
257    fn next_port(&self) -> Self::Port {
258        let next_port = *self.next_port.borrow();
259        *self.next_port.borrow_mut() += 1;
260        format!("port_{}", next_port)
261    }
262
263    fn update_meta(&self, _meta: &Self::Meta) {}
264
265    fn instantiate(
266        &self,
267        env: &mut Self::InstantiateEnv,
268        _meta: &mut Self::Meta,
269        graph: DfirGraph,
270        extra_stmts: &[syn::Stmt],
271        sidecars: &[syn::Expr],
272    ) {
273        let (bin_name, config) = create_graph_trybuild(
274            graph,
275            extra_stmts,
276            sidecars,
277            self.name_hint.as_deref(),
278            crate::compile::trybuild::generate::DeployMode::Maelstrom,
279            LinkingMode::Static,
280        );
281
282        env.bin_name = Some(bin_name);
283        env.project_dir = Some(config.project_dir);
284        env.target_dir = Some(config.target_dir);
285        env.features = config.features;
286    }
287}
288
289/// Represents an external client in Maelstrom deployment.
290#[derive(Clone)]
291pub enum MaelstromExternal {}
292
293impl Node for MaelstromExternal {
294    type Port = String;
295    type Meta = ();
296    type InstantiateEnv = MaelstromDeployment;
297
298    fn next_port(&self) -> Self::Port {
299        unreachable!()
300    }
301
302    fn update_meta(&self, _meta: &Self::Meta) {}
303
304    fn instantiate(
305        &self,
306        _env: &mut Self::InstantiateEnv,
307        _meta: &mut Self::Meta,
308        _graph: DfirGraph,
309        _extra_stmts: &[syn::Stmt],
310        _sidecars: &[syn::Expr],
311    ) {
312        unreachable!()
313    }
314}
315
316impl<'a> RegisterPort<'a, MaelstromDeploy> for MaelstromExternal {
317    fn register(&self, _external_port_id: ExternalPortId, _port: Self::Port) {
318        unreachable!()
319    }
320
321    #[expect(clippy::manual_async_fn, reason = "false positive, involves lifetimes")]
322    fn as_bytes_bidi(
323        &self,
324        _external_port_id: ExternalPortId,
325    ) -> impl Future<
326        Output = (
327            Pin<Box<dyn Stream<Item = Result<BytesMut, Error>>>>,
328            Pin<Box<dyn Sink<Bytes, Error = Error>>>,
329        ),
330    > + 'a {
331        async move { unreachable!() }
332    }
333
334    #[expect(clippy::manual_async_fn, reason = "false positive, involves lifetimes")]
335    fn as_bincode_bidi<InT, OutT>(
336        &self,
337        _external_port_id: ExternalPortId,
338    ) -> impl Future<
339        Output = (
340            Pin<Box<dyn Stream<Item = OutT>>>,
341            Pin<Box<dyn Sink<InT, Error = Error>>>,
342        ),
343    > + 'a
344    where
345        InT: Serialize + 'static,
346        OutT: DeserializeOwned + 'static,
347    {
348        async move { unreachable!() }
349    }
350
351    #[expect(clippy::manual_async_fn, reason = "false positive, involves lifetimes")]
352    fn as_bincode_sink<T: Serialize + 'static>(
353        &self,
354        _external_port_id: ExternalPortId,
355    ) -> impl Future<Output = Pin<Box<dyn Sink<T, Error = Error>>>> + 'a {
356        async move { unreachable!() }
357    }
358
359    #[expect(clippy::manual_async_fn, reason = "false positive, involves lifetimes")]
360    fn as_bincode_source<T: DeserializeOwned + 'static>(
361        &self,
362        _external_port_id: ExternalPortId,
363    ) -> impl Future<Output = Pin<Box<dyn Stream<Item = T>>>> + 'a {
364        async move { unreachable!() }
365    }
366}
367
368/// Specification for building a Maelstrom cluster.
369#[derive(Clone)]
370pub struct MaelstromClusterSpec;
371
372impl<'a> ClusterSpec<'a, MaelstromDeploy> for MaelstromClusterSpec {
373    fn build(self, key: LocationKey, name_hint: &str) -> MaelstromCluster {
374        assert_eq!(
375            key,
376            LocationKey::FIRST,
377            "there should only be one location for a Maelstrom deployment"
378        );
379        MaelstromCluster {
380            next_port: Rc::new(RefCell::new(0)),
381            name_hint: Some(name_hint.to_owned()),
382        }
383    }
384}
385
386/// The Maelstrom deployment environment.
387///
388/// This holds configuration for the Maelstrom run and accumulates
389/// compilation artifacts during deployment.
390pub struct MaelstromDeployment {
391    /// Number of nodes in the cluster.
392    pub node_count: usize,
393    /// Path to the maelstrom binary.
394    pub maelstrom_path: PathBuf,
395    /// Workload to run (e.g., "echo", "broadcast", "g-counter").
396    pub workload: String,
397    /// Time limit in seconds.
398    pub time_limit: Option<u64>,
399    /// Rate of requests per second.
400    pub rate: Option<u64>,
401    /// The availability of nodes.
402    pub availability: Option<String>,
403    /// Nemesis to run during tests.
404    pub nemesis: Option<String>,
405    /// Additional maelstrom arguments.
406    pub extra_args: Vec<String>,
407
408    // Populated during deployment
409    pub(crate) bin_name: Option<String>,
410    pub(crate) project_dir: Option<PathBuf>,
411    pub(crate) target_dir: Option<PathBuf>,
412    pub(crate) features: Option<Vec<String>>,
413}
414
415impl MaelstromDeployment {
416    /// Create a new Maelstrom deployment with the given node count.
417    pub fn new(workload: impl Into<String>) -> Self {
418        Self {
419            node_count: 1,
420            maelstrom_path: PathBuf::from("maelstrom"),
421            workload: workload.into(),
422            time_limit: None,
423            rate: None,
424            availability: None,
425            nemesis: None,
426            extra_args: vec![],
427            bin_name: None,
428            project_dir: None,
429            target_dir: None,
430            features: None,
431        }
432    }
433
434    /// Set the node count.
435    pub fn node_count(mut self, count: usize) -> Self {
436        self.node_count = count;
437        self
438    }
439
440    /// Set the path to the maelstrom binary.
441    pub fn maelstrom_path(mut self, path: impl Into<PathBuf>) -> Self {
442        self.maelstrom_path = path.into();
443        self
444    }
445
446    /// Set the time limit in seconds.
447    pub fn time_limit(mut self, seconds: u64) -> Self {
448        self.time_limit = Some(seconds);
449        self
450    }
451
452    /// Set the request rate per second.
453    pub fn rate(mut self, rate: u64) -> Self {
454        self.rate = Some(rate);
455        self
456    }
457
458    /// Set the availability for the test.
459    pub fn availability(mut self, availability: impl Into<String>) -> Self {
460        self.availability = Some(availability.into());
461        self
462    }
463
464    /// Set the nemesis for the test.
465    pub fn nemesis(mut self, nemesis: impl Into<String>) -> Self {
466        self.nemesis = Some(nemesis.into());
467        self
468    }
469
470    /// Add extra arguments to pass to maelstrom.
471    pub fn extra_args(mut self, args: impl IntoIterator<Item = impl Into<String>>) -> Self {
472        self.extra_args.extend(args.into_iter().map(Into::into));
473        self
474    }
475
476    /// Build the compiled binary in dev mode.
477    /// Returns the path to the compiled binary.
478    pub fn build(&self) -> Result<PathBuf, Error> {
479        let bin_name = self
480            .bin_name
481            .as_ref()
482            .expect("No binary name set - did you call deploy?");
483        let project_dir = self.project_dir.as_ref().expect("No project dir set");
484        let target_dir = self.target_dir.as_ref().expect("No target dir set");
485
486        let mut cmd = std::process::Command::new("cargo");
487        cmd.arg("build")
488            .arg("--example")
489            .arg(bin_name)
490            .arg("--no-default-features")
491            .current_dir(project_dir)
492            .env("CARGO_TARGET_DIR", target_dir)
493            .env("STAGELEFT_TRYBUILD_BUILD_STAGED", "1");
494
495        // Always include maelstrom_runtime feature for runtime support
496        let mut all_features = vec!["hydro___feature_maelstrom_runtime".to_owned()];
497        if let Some(features) = &self.features {
498            all_features.extend(features.iter().cloned());
499        }
500        if !all_features.is_empty() {
501            cmd.arg("--features").arg(all_features.join(","));
502        }
503
504        let status = cmd.status()?;
505        if !status.success() {
506            return Err(Error::other(format!(
507                "cargo build failed with status: {}",
508                status
509            )));
510        }
511
512        Ok(target_dir.join("debug").join("examples").join(bin_name))
513    }
514
515    /// Run Maelstrom with the compiled binary, return Ok(()) if all checks pass.
516    ///
517    /// This will block until Maelstrom completes.
518    pub fn run(self) -> Result<(), Error> {
519        let binary_path = self.build()?;
520
521        // Use a unique working directory per run to avoid conflicts with concurrent tests.
522        let run_dir = tempfile::tempdir().map_err(Error::other)?;
523
524        let mut cmd = std::process::Command::new(&self.maelstrom_path);
525        cmd.arg("test")
526            .arg("-w")
527            .arg(&self.workload)
528            .arg("--bin")
529            .arg(&binary_path)
530            .arg("--node-count")
531            .arg(self.node_count.to_string())
532            .current_dir(run_dir.path())
533            .stdout(Stdio::piped());
534
535        if let Some(time_limit) = self.time_limit {
536            cmd.arg("--time-limit").arg(time_limit.to_string());
537        }
538
539        if let Some(rate) = self.rate {
540            cmd.arg("--rate").arg(rate.to_string());
541        }
542
543        if let Some(availability) = self.availability {
544            cmd.arg("--availability").arg(availability);
545        }
546
547        if let Some(nemesis) = self.nemesis {
548            cmd.arg("--nemesis").arg(nemesis);
549        }
550
551        for arg in &self.extra_args {
552            cmd.arg(arg);
553        }
554
555        let spawned = cmd.spawn()?;
556
557        for line in BufReader::new(spawned.stdout.unwrap()).lines() {
558            let line = line?;
559            eprintln!("{}", &line);
560
561            if line.starts_with("Analysis invalid!") {
562                let path = run_dir.keep();
563                return Err(Error::other(format!(
564                    "Analysis was invalid. Maelstrom store at: {}",
565                    path.display()
566                )));
567            } else if line.starts_with("Errors occurred during analysis, but no anomalies found.")
568                || line.starts_with("Everything looks good!")
569            {
570                return Ok(());
571            }
572        }
573
574        let path = run_dir.keep();
575        Err(Error::other(format!(
576            "Maelstrom produced an unexpected result. Store at: {}",
577            path.display()
578        )))
579    }
580
581    /// Get the path to the compiled binary (after building).
582    pub fn binary_path(&self) -> Option<PathBuf> {
583        let bin_name = self.bin_name.as_ref()?;
584        let target_dir = self.target_dir.as_ref()?;
585        Some(target_dir.join("debug").join("examples").join(bin_name))
586    }
587}
hydro_lang/deploy/maelstrom/deploy_maelstrom.rs

hydro_lang/deploy/maelstrom/
deploy_maelstrom.rs
