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

hydro_lang/deploy/maelstrom/
deploy_maelstrom.rs
