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::{Path, PathBuf};
11use std::pin::Pin;
12use std::process::Stdio;
13use std::rc::Rc;
14
15use bytes::{Bytes, BytesMut};
16use dfir_lang::graph::{AsCodeOptions, 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::{
26    ExampleBuildConfig, LinkingMode, TrybuildConfig, compile_trybuild_example,
27    create_graph_trybuild,
28};
29use crate::location::dynamic::LocationId;
30use crate::location::member_id::TaglessMemberId;
31use crate::location::{LocationKey, MembershipEvent, NetworkHint};
32
33/// Deployment backend that targets Maelstrom for distributed systems testing.
34///
35/// This backend compiles Hydro programs to binaries that communicate via Maelstrom's
36/// stdin/stdout JSON protocol. It is restricted to programs with:
37/// - Exactly one cluster (no processes)
38/// - A single external input channel for client communication
39pub enum MaelstromDeploy {}
40
41impl<'a> Deploy<'a> for MaelstromDeploy {
42    type Meta = ();
43    type InstantiateEnv = MaelstromDeployment;
44
45    type Process = MaelstromProcess;
46    type Cluster = MaelstromCluster;
47    type External = MaelstromExternal;
48
49    fn o2o_sink_source(
50        _env: &mut Self::InstantiateEnv,
51        _p1: &Self::Process,
52        _p1_port: &<Self::Process as Node>::Port,
53        _p2: &Self::Process,
54        _p2_port: &<Self::Process as Node>::Port,
55        _name: Option<&str>,
56        _networking_info: &crate::networking::NetworkingInfo,
57        _external_types: Option<(&syn::Type, &syn::Type)>,
58    ) -> (syn::Expr, syn::Expr) {
59        panic!("Maelstrom deployment does not support processes, only clusters")
60    }
61
62    fn o2o_connect(
63        _p1: &Self::Process,
64        _p1_port: &<Self::Process as Node>::Port,
65        _p2: &Self::Process,
66        _p2_port: &<Self::Process as Node>::Port,
67    ) -> Box<dyn FnOnce()> {
68        panic!("Maelstrom deployment does not support processes, only clusters")
69    }
70
71    fn o2m_sink_source(
72        _env: &mut Self::InstantiateEnv,
73        _p1: &Self::Process,
74        _p1_port: &<Self::Process as Node>::Port,
75        _c2: &Self::Cluster,
76        _c2_port: &<Self::Cluster as Node>::Port,
77        _name: Option<&str>,
78        _networking_info: &crate::networking::NetworkingInfo,
79        _external_types: Option<(&syn::Type, &syn::Type)>,
80    ) -> (syn::Expr, syn::Expr) {
81        panic!("Maelstrom deployment does not support processes, only clusters")
82    }
83
84    fn o2m_connect(
85        _p1: &Self::Process,
86        _p1_port: &<Self::Process as Node>::Port,
87        _c2: &Self::Cluster,
88        _c2_port: &<Self::Cluster as Node>::Port,
89    ) -> Box<dyn FnOnce()> {
90        panic!("Maelstrom deployment does not support processes, only clusters")
91    }
92
93    fn m2o_sink_source(
94        _env: &mut Self::InstantiateEnv,
95        _c1: &Self::Cluster,
96        _c1_port: &<Self::Cluster as Node>::Port,
97        _p2: &Self::Process,
98        _p2_port: &<Self::Process as Node>::Port,
99        _name: Option<&str>,
100        _networking_info: &crate::networking::NetworkingInfo,
101        _external_types: Option<(&syn::Type, &syn::Type)>,
102    ) -> (syn::Expr, syn::Expr) {
103        panic!("Maelstrom deployment does not support processes, only clusters")
104    }
105
106    fn m2o_connect(
107        _c1: &Self::Cluster,
108        _c1_port: &<Self::Cluster as Node>::Port,
109        _p2: &Self::Process,
110        _p2_port: &<Self::Process as Node>::Port,
111    ) -> Box<dyn FnOnce()> {
112        panic!("Maelstrom deployment does not support processes, only clusters")
113    }
114
115    fn m2m_sink_source(
116        env: &mut Self::InstantiateEnv,
117        _c1: &Self::Cluster,
118        _c1_port: &<Self::Cluster as Node>::Port,
119        _c2: &Self::Cluster,
120        _c2_port: &<Self::Cluster as Node>::Port,
121        _name: Option<&str>,
122        networking_info: &crate::networking::NetworkingInfo,
123        _external_types: Option<(&syn::Type, &syn::Type)>,
124    ) -> (syn::Expr, syn::Expr) {
125        use crate::networking::{NetworkingInfo, TcpFault};
126        match networking_info {
127            NetworkingInfo::Tcp { fault } => match (fault, env.nemesis.as_deref()) {
128                (TcpFault::Lossy | TcpFault::LossyDelayedForever, _) => {} /* lossy/delayed are always allowed */
129                (_, None) => {} // no nemesis means any fault model is fine
130                (TcpFault::FailStop, Some("partition")) => {
131                    panic!(
132                        "Maelstrom partition nemesis requires lossy networking, but fail_stop was used. \
133                         Use `TCP.lossy().bincode()` or `TCP.lossy_delayed_forever().bincode()` instead of `TCP.fail_stop().bincode()`."
134                    );
135                }
136                (TcpFault::FailStop, Some(_)) => {} // other nemeses are fine with fail_stop
137            },
138            NetworkingInfo::Udp { .. } => {} // UDP is always lossy, which is always allowed
139            NetworkingInfo::Embedded { .. } => {
140                unreachable!("`EMBEDDED` channels are only supported by the embedded backend")
141            }
142        }
143        deploy_maelstrom_m2m(RuntimeData::new("__hydro_lang_maelstrom_meta"))
144    }
145
146    fn m2m_connect(
147        _c1: &Self::Cluster,
148        _c1_port: &<Self::Cluster as Node>::Port,
149        _c2: &Self::Cluster,
150        _c2_port: &<Self::Cluster as Node>::Port,
151    ) -> Box<dyn FnOnce()> {
152        // No runtime connection needed for Maelstrom - all routing is via stdin/stdout
153        Box::new(|| {})
154    }
155
156    fn e2o_many_source(
157        _extra_stmts: &mut Vec<syn::Stmt>,
158        _p2: &Self::Process,
159        _p2_port: &<Self::Process as Node>::Port,
160        _codec_type: &syn::Type,
161        _shared_handle: String,
162    ) -> syn::Expr {
163        panic!("Maelstrom deployment does not support processes, only clusters")
164    }
165
166    fn e2o_many_sink(_shared_handle: String) -> syn::Expr {
167        panic!("Maelstrom deployment does not support processes, only clusters")
168    }
169
170    fn e2o_source(
171        _extra_stmts: &mut Vec<syn::Stmt>,
172        _p1: &Self::External,
173        _p1_port: &<Self::External as Node>::Port,
174        _p2: &Self::Process,
175        _p2_port: &<Self::Process as Node>::Port,
176        _codec_type: &syn::Type,
177        _shared_handle: String,
178    ) -> syn::Expr {
179        panic!("Maelstrom deployment does not support processes, only clusters")
180    }
181
182    fn e2o_connect(
183        _p1: &Self::External,
184        _p1_port: &<Self::External as Node>::Port,
185        _p2: &Self::Process,
186        _p2_port: &<Self::Process as Node>::Port,
187        _many: bool,
188        _server_hint: NetworkHint,
189    ) -> Box<dyn FnOnce()> {
190        panic!("Maelstrom deployment does not support processes, only clusters")
191    }
192
193    fn o2e_sink(
194        _p1: &Self::Process,
195        _p1_port: &<Self::Process as Node>::Port,
196        _p2: &Self::External,
197        _p2_port: &<Self::External as Node>::Port,
198        _shared_handle: String,
199    ) -> syn::Expr {
200        panic!("Maelstrom deployment does not support processes, only clusters")
201    }
202
203    fn cluster_ids(
204        _of_cluster: LocationKey,
205    ) -> impl QuotedWithContext<'a, &'a [TaglessMemberId], ()> + Clone + 'a {
206        cluster_members(RuntimeData::new("__hydro_lang_maelstrom_meta"), _of_cluster)
207    }
208
209    fn cluster_self_id() -> impl QuotedWithContext<'a, TaglessMemberId, ()> + Clone + 'a {
210        cluster_self_id(RuntimeData::new("__hydro_lang_maelstrom_meta"))
211    }
212
213    fn cluster_membership_stream(
214        _env: &mut Self::InstantiateEnv,
215        _at_location: &LocationId,
216        location_id: &LocationId,
217    ) -> impl QuotedWithContext<'a, Box<dyn Stream<Item = (TaglessMemberId, MembershipEvent)> + Unpin>, ()>
218    {
219        cluster_membership_stream(location_id)
220    }
221}
222
223/// A dummy process type for Maelstrom (processes are not supported).
224#[derive(Clone)]
225pub struct MaelstromProcess {
226    _private: (),
227}
228
229impl Node for MaelstromProcess {
230    type Port = String;
231    type Meta = ();
232    type InstantiateEnv = MaelstromDeployment;
233
234    fn next_port(&self) -> Self::Port {
235        panic!("Maelstrom deployment does not support processes")
236    }
237
238    fn update_meta(&self, _meta: &Self::Meta) {}
239
240    fn instantiate(
241        &self,
242        _env: &mut Self::InstantiateEnv,
243        _meta: &mut Self::Meta,
244        _graph: DfirGraph,
245        _extra_stmts: &[syn::Stmt],
246        _sidecars: &[syn::Expr],
247        _as_code_options: &AsCodeOptions,
248    ) {
249        panic!("Maelstrom deployment does not support processes")
250    }
251}
252
253/// Represents a cluster in Maelstrom deployment.
254#[derive(Clone)]
255pub struct MaelstromCluster {
256    next_port: Rc<RefCell<usize>>,
257    name_hint: Option<String>,
258}
259
260impl Node for MaelstromCluster {
261    type Port = String;
262    type Meta = ();
263    type InstantiateEnv = MaelstromDeployment;
264
265    fn next_port(&self) -> Self::Port {
266        let next_port = *self.next_port.borrow();
267        *self.next_port.borrow_mut() += 1;
268        format!("port_{}", next_port)
269    }
270
271    fn update_meta(&self, _meta: &Self::Meta) {}
272
273    fn instantiate(
274        &self,
275        env: &mut Self::InstantiateEnv,
276        _meta: &mut Self::Meta,
277        graph: DfirGraph,
278        extra_stmts: &[syn::Stmt],
279        sidecars: &[syn::Expr],
280        as_code_options: &AsCodeOptions,
281    ) {
282        let (bin_name, config) = create_graph_trybuild(
283            graph,
284            extra_stmts,
285            sidecars,
286            as_code_options,
287            self.name_hint.as_deref(),
288            crate::compile::trybuild::generate::DeployMode::Maelstrom,
289            LinkingMode::Dynamic,
290        );
291
292        env.bin_name = Some(bin_name);
293        env.trybuild = Some(config);
294    }
295}
296
297/// Represents an external client in Maelstrom deployment.
298#[derive(Clone)]
299pub enum MaelstromExternal {}
300
301impl Node for MaelstromExternal {
302    type Port = String;
303    type Meta = ();
304    type InstantiateEnv = MaelstromDeployment;
305
306    fn next_port(&self) -> Self::Port {
307        unreachable!()
308    }
309
310    fn update_meta(&self, _meta: &Self::Meta) {}
311
312    fn instantiate(
313        &self,
314        _env: &mut Self::InstantiateEnv,
315        _meta: &mut Self::Meta,
316        _graph: DfirGraph,
317        _extra_stmts: &[syn::Stmt],
318        _sidecars: &[syn::Expr],
319        _as_code_options: &AsCodeOptions,
320    ) {
321        unreachable!()
322    }
323}
324
325impl<'a> RegisterPort<'a, MaelstromDeploy> for MaelstromExternal {
326    fn register(&self, _external_port_id: ExternalPortId, _port: Self::Port) {
327        unreachable!()
328    }
329
330    #[expect(clippy::manual_async_fn, reason = "false positive, involves lifetimes")]
331    fn as_bytes_bidi(
332        &self,
333        _external_port_id: ExternalPortId,
334    ) -> impl Future<
335        Output = (
336            Pin<Box<dyn Stream<Item = Result<BytesMut, Error>>>>,
337            Pin<Box<dyn Sink<Bytes, Error = Error>>>,
338        ),
339    > + 'a {
340        async move { unreachable!() }
341    }
342
343    #[expect(clippy::manual_async_fn, reason = "false positive, involves lifetimes")]
344    fn as_bincode_bidi<InT, OutT>(
345        &self,
346        _external_port_id: ExternalPortId,
347    ) -> impl Future<
348        Output = (
349            Pin<Box<dyn Stream<Item = OutT>>>,
350            Pin<Box<dyn Sink<InT, Error = Error>>>,
351        ),
352    > + 'a
353    where
354        InT: Serialize + 'static,
355        OutT: DeserializeOwned + 'static,
356    {
357        async move { unreachable!() }
358    }
359
360    #[expect(clippy::manual_async_fn, reason = "false positive, involves lifetimes")]
361    fn as_bincode_sink<T: Serialize + 'static>(
362        &self,
363        _external_port_id: ExternalPortId,
364    ) -> impl Future<Output = Pin<Box<dyn Sink<T, Error = Error>>>> + 'a {
365        async move { unreachable!() }
366    }
367
368    #[expect(clippy::manual_async_fn, reason = "false positive, involves lifetimes")]
369    fn as_bincode_source<T: DeserializeOwned + 'static>(
370        &self,
371        _external_port_id: ExternalPortId,
372    ) -> impl Future<Output = Pin<Box<dyn Stream<Item = T>>>> + 'a {
373        async move { unreachable!() }
374    }
375}
376
377/// Specification for building a Maelstrom cluster.
378#[derive(Clone)]
379pub struct MaelstromClusterSpec;
380
381impl<'a> ClusterSpec<'a, MaelstromDeploy> for MaelstromClusterSpec {
382    fn build(self, key: LocationKey, name_hint: &str) -> MaelstromCluster {
383        assert_eq!(
384            key,
385            LocationKey::FIRST,
386            "there should only be one location for a Maelstrom deployment"
387        );
388        MaelstromCluster {
389            next_port: Rc::new(RefCell::new(0)),
390            name_hint: Some(name_hint.to_owned()),
391        }
392    }
393}
394
395/// The Maelstrom deployment environment.
396///
397/// This holds configuration for the Maelstrom run and accumulates
398/// compilation artifacts during deployment.
399pub struct MaelstromDeployment {
400    /// Number of nodes in the cluster.
401    pub node_count: usize,
402    /// Path to the maelstrom binary.
403    pub maelstrom_path: PathBuf,
404    /// Workload to run (e.g., "echo", "broadcast", "g-counter").
405    pub workload: String,
406    /// Time limit in seconds.
407    pub time_limit: Option<u64>,
408    /// Rate of requests per second.
409    pub rate: Option<u64>,
410    /// The availability of nodes.
411    pub availability: Option<String>,
412    /// Nemesis to run during tests.
413    pub nemesis: Option<String>,
414    /// Additional maelstrom arguments.
415    pub extra_args: Vec<String>,
416
417    // Populated during deployment
418    pub(crate) bin_name: Option<String>,
419    pub(crate) trybuild: Option<TrybuildConfig>,
420}
421
422impl MaelstromDeployment {
423    /// Create a new Maelstrom deployment with the given node count.
424    pub fn new(workload: impl Into<String>) -> Self {
425        Self {
426            node_count: 1,
427            maelstrom_path: PathBuf::from("maelstrom"),
428            workload: workload.into(),
429            time_limit: None,
430            rate: None,
431            availability: None,
432            nemesis: None,
433            extra_args: vec![],
434            bin_name: None,
435            trybuild: None,
436        }
437    }
438
439    /// Set the node count.
440    #[must_use]
441    pub fn node_count(mut self, count: usize) -> Self {
442        self.node_count = count;
443        self
444    }
445
446    /// Set the path to the maelstrom binary.
447    pub fn maelstrom_path(mut self, path: impl Into<PathBuf>) -> Self {
448        self.maelstrom_path = path.into();
449        self
450    }
451
452    /// Set the time limit in seconds.
453    #[must_use]
454    pub fn time_limit(mut self, seconds: u64) -> Self {
455        self.time_limit = Some(seconds);
456        self
457    }
458
459    /// Set the request rate per second.
460    #[must_use]
461    pub fn rate(mut self, rate: u64) -> Self {
462        self.rate = Some(rate);
463        self
464    }
465
466    /// Set the availability for the test.
467    pub fn availability(mut self, availability: impl Into<String>) -> Self {
468        self.availability = Some(availability.into());
469        self
470    }
471
472    /// Set the nemesis for the test.
473    pub fn nemesis(mut self, nemesis: impl Into<String>) -> Self {
474        self.nemesis = Some(nemesis.into());
475        self
476    }
477
478    /// Add extra arguments to pass to maelstrom.
479    pub fn extra_args(mut self, args: impl IntoIterator<Item = impl Into<String>>) -> Self {
480        self.extra_args.extend(args.into_iter().map(Into::into));
481        self
482    }
483
484    /// Build the compiled binary in dev mode.
485    /// Returns the path to the compiled binary.
486    ///
487    /// This shares the same parallel-compilation machinery as the simulator: the
488    /// program is linked dynamically against a prebuilt dylib of its dependencies,
489    /// so repeated and concurrent builds only need to recompile the generated
490    /// example itself.
491    pub fn build(&self) -> Result<PathBuf, Error> {
492        let bin_name = self
493            .bin_name
494            .as_ref()
495            .expect("No binary name set - did you call deploy?");
496        let trybuild = self
497            .trybuild
498            .as_ref()
499            .expect("No trybuild config set - did you call deploy?");
500
501        compile_trybuild_example(ExampleBuildConfig {
502            trybuild: trybuild.clone(),
503            bin_name: bin_name.clone(),
504            runtime_feature: "hydro___feature_maelstrom_runtime",
505            // Maelstrom builds the generated example directly as an executable.
506            example_name: bin_name.clone(),
507            crate_type: None,
508            set_trybuild_lib_name: false,
509            allow_fuzz: false,
510        })
511        .map_err(|()| Error::other("Maelstrom binary compilation failed"))
512    }
513
514    /// Run Maelstrom with the compiled binary, return Ok(()) if all checks pass.
515    ///
516    /// This will block until Maelstrom completes.
517    pub fn run(self) -> Result<(), Error> {
518        let binary_path = self.build()?;
519
520        // Warm up the binary before handing it to Maelstrom. On macOS, the
521        // first execution of a freshly written binary triggers a Gatekeeper /
522        // XProtect (`syspolicyd`) scan that can take several seconds on loaded
523        // CI machines. Maelstrom only waits 10 seconds for each node to answer
524        // the `init` RPC, so a cold first exec (multiplied across concurrently
525        // launched nodes) can cause spurious node-startup timeouts. The warmup
526        // invocation sees EOF on stdin and exits immediately, priming the
527        // system's first-exec caches for the real run.
528        std::process::Command::new(&binary_path)
529            .stdin(Stdio::null())
530            .stdout(Stdio::null())
531            .stderr(Stdio::null())
532            .spawn()?
533            .wait()?;
534
535        // Use a unique working directory per run to avoid conflicts with concurrent tests.
536        let run_dir = tempfile::tempdir().map_err(Error::other)?;
537
538        let mut cmd = std::process::Command::new(&self.maelstrom_path);
539        cmd.arg("test")
540            .arg("-w")
541            .arg(&self.workload)
542            .arg("--bin")
543            .arg(&binary_path)
544            .arg("--node-count")
545            .arg(self.node_count.to_string())
546            .current_dir(run_dir.path())
547            .stdout(Stdio::piped());
548
549        if let Some(time_limit) = self.time_limit {
550            cmd.arg("--time-limit").arg(time_limit.to_string());
551        }
552
553        if let Some(rate) = self.rate {
554            cmd.arg("--rate").arg(rate.to_string());
555        }
556
557        if let Some(availability) = self.availability {
558            cmd.arg("--availability").arg(availability);
559        }
560
561        if let Some(nemesis) = self.nemesis {
562            cmd.arg("--nemesis").arg(nemesis);
563        }
564
565        for arg in &self.extra_args {
566            cmd.arg(arg);
567        }
568
569        let spawned = cmd.spawn()?;
570
571        for line in BufReader::new(spawned.stdout.unwrap()).lines() {
572            let line = line?;
573            eprintln!("{}", line);
574
575            if line.starts_with("Analysis invalid!") {
576                let path = run_dir.keep();
577                dump_node_logs(&path);
578                return Err(Error::other(format!(
579                    "Analysis was invalid. Maelstrom store at: {}",
580                    path.display()
581                )));
582            } else if line.starts_with("Errors occurred during analysis, but no anomalies found.")
583                || line.starts_with("Everything looks good!")
584            {
585                return Ok(());
586            }
587        }
588
589        let path = run_dir.keep();
590        dump_node_logs(&path);
591        Err(Error::other(format!(
592            "Maelstrom produced an unexpected result. Store at: {}",
593            path.display()
594        )))
595    }
596
597    /// Get the path to the compiled binary, building it if necessary.
598    pub fn binary_path(&self) -> Option<PathBuf> {
599        self.build().ok()
600    }
601}
602
603/// Print the per-node logs from a Maelstrom run directory to stderr.
604///
605/// Maelstrom truncates node stderr in its own error messages (keeping only the
606/// tail), so when a node crashes the actual panic message is often cut off.
607/// The full logs live in `store/<workload>/<timestamp>/node-logs/*.log` under
608/// the run directory; dump them so failures are debuggable in CI, where the
609/// preserved store directory is not otherwise accessible.
610fn dump_node_logs(run_dir: &Path) {
611    fn collect(dir: &Path, out: &mut Vec<PathBuf>) {
612        let Ok(entries) = std::fs::read_dir(dir) else {
613            return;
614        };
615        for entry in entries.flatten() {
616            let path = entry.path();
617            // Skip symlinks (e.g. `store/latest`) to avoid duplicates.
618            if path.is_symlink() || !path.is_dir() {
619                continue;
620            }
621            if path.file_name().is_some_and(|name| name == "node-logs") {
622                if let Ok(logs) = std::fs::read_dir(&path) {
623                    out.extend(logs.flatten().map(|e| e.path()));
624                }
625            } else {
626                collect(&path, out);
627            }
628        }
629    }
630
631    let mut log_files = Vec::new();
632    collect(&run_dir.join("store"), &mut log_files);
633    log_files.sort();
634
635    for log in log_files {
636        eprintln!("==== Maelstrom node log: {} ====", log.display());
637        match std::fs::read_to_string(&log) {
638            Ok(contents) => eprint!("{}", contents),
639            Err(e) => eprintln!("(failed to read log: {})", e),
640        }
641        eprintln!("==== end of node log: {} ====", log.display());
642    }
643}
hydro_lang/deploy/maelstrom/deploy_maelstrom.rs

hydro_lang/deploy/maelstrom/
deploy_maelstrom.rs
