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::{
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    ) -> (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    ) -> (syn::Expr, syn::Expr) {
79        panic!("Maelstrom deployment does not support processes, only clusters")
80    }
81
82    fn o2m_connect(
83        _p1: &Self::Process,
84        _p1_port: &<Self::Process as Node>::Port,
85        _c2: &Self::Cluster,
86        _c2_port: &<Self::Cluster as Node>::Port,
87    ) -> Box<dyn FnOnce()> {
88        panic!("Maelstrom deployment does not support processes, only clusters")
89    }
90
91    fn m2o_sink_source(
92        _env: &mut Self::InstantiateEnv,
93        _c1: &Self::Cluster,
94        _c1_port: &<Self::Cluster as Node>::Port,
95        _p2: &Self::Process,
96        _p2_port: &<Self::Process as Node>::Port,
97        _name: Option<&str>,
98        _networking_info: &crate::networking::NetworkingInfo,
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    ) -> (syn::Expr, syn::Expr) {
121        use crate::networking::{NetworkingInfo, TcpFault};
122        match networking_info {
123            NetworkingInfo::Tcp { fault } => match (fault, env.nemesis.as_deref()) {
124                (TcpFault::Lossy | TcpFault::LossyDelayedForever, _) => {} /* lossy/delayed are always allowed */
125                (_, None) => {} // no nemesis means any fault model is fine
126                (TcpFault::FailStop, Some("partition")) => {
127                    panic!(
128                        "Maelstrom partition nemesis requires lossy networking, but fail_stop was used. \
129                         Use `TCP.lossy().bincode()` or `TCP.lossy_delayed_forever().bincode()` instead of `TCP.fail_stop().bincode()`."
130                    );
131                }
132                (TcpFault::FailStop, Some(_)) => {} // other nemeses are fine with fail_stop
133            },
134        }
135        deploy_maelstrom_m2m(RuntimeData::new("__hydro_lang_maelstrom_meta"))
136    }
137
138    fn m2m_connect(
139        _c1: &Self::Cluster,
140        _c1_port: &<Self::Cluster as Node>::Port,
141        _c2: &Self::Cluster,
142        _c2_port: &<Self::Cluster as Node>::Port,
143    ) -> Box<dyn FnOnce()> {
144        // No runtime connection needed for Maelstrom - all routing is via stdin/stdout
145        Box::new(|| {})
146    }
147
148    fn e2o_many_source(
149        _extra_stmts: &mut Vec<syn::Stmt>,
150        _p2: &Self::Process,
151        _p2_port: &<Self::Process as Node>::Port,
152        _codec_type: &syn::Type,
153        _shared_handle: String,
154    ) -> syn::Expr {
155        panic!("Maelstrom deployment does not support processes, only clusters")
156    }
157
158    fn e2o_many_sink(_shared_handle: String) -> syn::Expr {
159        panic!("Maelstrom deployment does not support processes, only clusters")
160    }
161
162    fn e2o_source(
163        _extra_stmts: &mut Vec<syn::Stmt>,
164        _p1: &Self::External,
165        _p1_port: &<Self::External as Node>::Port,
166        _p2: &Self::Process,
167        _p2_port: &<Self::Process as Node>::Port,
168        _codec_type: &syn::Type,
169        _shared_handle: String,
170    ) -> syn::Expr {
171        panic!("Maelstrom deployment does not support processes, only clusters")
172    }
173
174    fn e2o_connect(
175        _p1: &Self::External,
176        _p1_port: &<Self::External as Node>::Port,
177        _p2: &Self::Process,
178        _p2_port: &<Self::Process as Node>::Port,
179        _many: bool,
180        _server_hint: NetworkHint,
181    ) -> Box<dyn FnOnce()> {
182        panic!("Maelstrom deployment does not support processes, only clusters")
183    }
184
185    fn o2e_sink(
186        _p1: &Self::Process,
187        _p1_port: &<Self::Process as Node>::Port,
188        _p2: &Self::External,
189        _p2_port: &<Self::External as Node>::Port,
190        _shared_handle: String,
191    ) -> syn::Expr {
192        panic!("Maelstrom deployment does not support processes, only clusters")
193    }
194
195    fn cluster_ids(
196        _of_cluster: LocationKey,
197    ) -> impl QuotedWithContext<'a, &'a [TaglessMemberId], ()> + Clone + 'a {
198        cluster_members(RuntimeData::new("__hydro_lang_maelstrom_meta"), _of_cluster)
199    }
200
201    fn cluster_self_id() -> impl QuotedWithContext<'a, TaglessMemberId, ()> + Clone + 'a {
202        cluster_self_id(RuntimeData::new("__hydro_lang_maelstrom_meta"))
203    }
204
205    fn cluster_membership_stream(
206        _env: &mut Self::InstantiateEnv,
207        _at_location: &LocationId,
208        location_id: &LocationId,
209    ) -> impl QuotedWithContext<'a, Box<dyn Stream<Item = (TaglessMemberId, MembershipEvent)> + Unpin>, ()>
210    {
211        cluster_membership_stream(location_id)
212    }
213}
214
215/// A dummy process type for Maelstrom (processes are not supported).
216#[derive(Clone)]
217pub struct MaelstromProcess {
218    _private: (),
219}
220
221impl Node for MaelstromProcess {
222    type Port = String;
223    type Meta = ();
224    type InstantiateEnv = MaelstromDeployment;
225
226    fn next_port(&self) -> Self::Port {
227        panic!("Maelstrom deployment does not support processes")
228    }
229
230    fn update_meta(&self, _meta: &Self::Meta) {}
231
232    fn instantiate(
233        &self,
234        _env: &mut Self::InstantiateEnv,
235        _meta: &mut Self::Meta,
236        _graph: DfirGraph,
237        _extra_stmts: &[syn::Stmt],
238        _sidecars: &[syn::Expr],
239    ) {
240        panic!("Maelstrom deployment does not support processes")
241    }
242}
243
244/// Represents a cluster in Maelstrom deployment.
245#[derive(Clone)]
246pub struct MaelstromCluster {
247    next_port: Rc<RefCell<usize>>,
248    name_hint: Option<String>,
249}
250
251impl Node for MaelstromCluster {
252    type Port = String;
253    type Meta = ();
254    type InstantiateEnv = MaelstromDeployment;
255
256    fn next_port(&self) -> Self::Port {
257        let next_port = *self.next_port.borrow();
258        *self.next_port.borrow_mut() += 1;
259        format!("port_{}", next_port)
260    }
261
262    fn update_meta(&self, _meta: &Self::Meta) {}
263
264    fn instantiate(
265        &self,
266        env: &mut Self::InstantiateEnv,
267        _meta: &mut Self::Meta,
268        graph: DfirGraph,
269        extra_stmts: &[syn::Stmt],
270        sidecars: &[syn::Expr],
271    ) {
272        let (bin_name, config) = create_graph_trybuild(
273            graph,
274            extra_stmts,
275            sidecars,
276            self.name_hint.as_deref(),
277            crate::compile::trybuild::generate::DeployMode::Maelstrom,
278            LinkingMode::Dynamic,
279        );
280
281        env.bin_name = Some(bin_name);
282        env.trybuild = Some(config);
283    }
284}
285
286/// Represents an external client in Maelstrom deployment.
287#[derive(Clone)]
288pub enum MaelstromExternal {}
289
290impl Node for MaelstromExternal {
291    type Port = String;
292    type Meta = ();
293    type InstantiateEnv = MaelstromDeployment;
294
295    fn next_port(&self) -> Self::Port {
296        unreachable!()
297    }
298
299    fn update_meta(&self, _meta: &Self::Meta) {}
300
301    fn instantiate(
302        &self,
303        _env: &mut Self::InstantiateEnv,
304        _meta: &mut Self::Meta,
305        _graph: DfirGraph,
306        _extra_stmts: &[syn::Stmt],
307        _sidecars: &[syn::Expr],
308    ) {
309        unreachable!()
310    }
311}
312
313impl<'a> RegisterPort<'a, MaelstromDeploy> for MaelstromExternal {
314    fn register(&self, _external_port_id: ExternalPortId, _port: Self::Port) {
315        unreachable!()
316    }
317
318    #[expect(clippy::manual_async_fn, reason = "false positive, involves lifetimes")]
319    fn as_bytes_bidi(
320        &self,
321        _external_port_id: ExternalPortId,
322    ) -> impl Future<
323        Output = (
324            Pin<Box<dyn Stream<Item = Result<BytesMut, Error>>>>,
325            Pin<Box<dyn Sink<Bytes, Error = Error>>>,
326        ),
327    > + 'a {
328        async move { unreachable!() }
329    }
330
331    #[expect(clippy::manual_async_fn, reason = "false positive, involves lifetimes")]
332    fn as_bincode_bidi<InT, OutT>(
333        &self,
334        _external_port_id: ExternalPortId,
335    ) -> impl Future<
336        Output = (
337            Pin<Box<dyn Stream<Item = OutT>>>,
338            Pin<Box<dyn Sink<InT, Error = Error>>>,
339        ),
340    > + 'a
341    where
342        InT: Serialize + 'static,
343        OutT: DeserializeOwned + 'static,
344    {
345        async move { unreachable!() }
346    }
347
348    #[expect(clippy::manual_async_fn, reason = "false positive, involves lifetimes")]
349    fn as_bincode_sink<T: Serialize + 'static>(
350        &self,
351        _external_port_id: ExternalPortId,
352    ) -> impl Future<Output = Pin<Box<dyn Sink<T, Error = Error>>>> + 'a {
353        async move { unreachable!() }
354    }
355
356    #[expect(clippy::manual_async_fn, reason = "false positive, involves lifetimes")]
357    fn as_bincode_source<T: DeserializeOwned + 'static>(
358        &self,
359        _external_port_id: ExternalPortId,
360    ) -> impl Future<Output = Pin<Box<dyn Stream<Item = T>>>> + 'a {
361        async move { unreachable!() }
362    }
363}
364
365/// Specification for building a Maelstrom cluster.
366#[derive(Clone)]
367pub struct MaelstromClusterSpec;
368
369impl<'a> ClusterSpec<'a, MaelstromDeploy> for MaelstromClusterSpec {
370    fn build(self, key: LocationKey, name_hint: &str) -> MaelstromCluster {
371        assert_eq!(
372            key,
373            LocationKey::FIRST,
374            "there should only be one location for a Maelstrom deployment"
375        );
376        MaelstromCluster {
377            next_port: Rc::new(RefCell::new(0)),
378            name_hint: Some(name_hint.to_owned()),
379        }
380    }
381}
382
383/// The Maelstrom deployment environment.
384///
385/// This holds configuration for the Maelstrom run and accumulates
386/// compilation artifacts during deployment.
387pub struct MaelstromDeployment {
388    /// Number of nodes in the cluster.
389    pub node_count: usize,
390    /// Path to the maelstrom binary.
391    pub maelstrom_path: PathBuf,
392    /// Workload to run (e.g., "echo", "broadcast", "g-counter").
393    pub workload: String,
394    /// Time limit in seconds.
395    pub time_limit: Option<u64>,
396    /// Rate of requests per second.
397    pub rate: Option<u64>,
398    /// The availability of nodes.
399    pub availability: Option<String>,
400    /// Nemesis to run during tests.
401    pub nemesis: Option<String>,
402    /// Additional maelstrom arguments.
403    pub extra_args: Vec<String>,
404
405    // Populated during deployment
406    pub(crate) bin_name: Option<String>,
407    pub(crate) trybuild: Option<TrybuildConfig>,
408}
409
410impl MaelstromDeployment {
411    /// Create a new Maelstrom deployment with the given node count.
412    pub fn new(workload: impl Into<String>) -> Self {
413        Self {
414            node_count: 1,
415            maelstrom_path: PathBuf::from("maelstrom"),
416            workload: workload.into(),
417            time_limit: None,
418            rate: None,
419            availability: None,
420            nemesis: None,
421            extra_args: vec![],
422            bin_name: None,
423            trybuild: None,
424        }
425    }
426
427    /// Set the node count.
428    pub fn node_count(mut self, count: usize) -> Self {
429        self.node_count = count;
430        self
431    }
432
433    /// Set the path to the maelstrom binary.
434    pub fn maelstrom_path(mut self, path: impl Into<PathBuf>) -> Self {
435        self.maelstrom_path = path.into();
436        self
437    }
438
439    /// Set the time limit in seconds.
440    pub fn time_limit(mut self, seconds: u64) -> Self {
441        self.time_limit = Some(seconds);
442        self
443    }
444
445    /// Set the request rate per second.
446    pub fn rate(mut self, rate: u64) -> Self {
447        self.rate = Some(rate);
448        self
449    }
450
451    /// Set the availability for the test.
452    pub fn availability(mut self, availability: impl Into<String>) -> Self {
453        self.availability = Some(availability.into());
454        self
455    }
456
457    /// Set the nemesis for the test.
458    pub fn nemesis(mut self, nemesis: impl Into<String>) -> Self {
459        self.nemesis = Some(nemesis.into());
460        self
461    }
462
463    /// Add extra arguments to pass to maelstrom.
464    pub fn extra_args(mut self, args: impl IntoIterator<Item = impl Into<String>>) -> Self {
465        self.extra_args.extend(args.into_iter().map(Into::into));
466        self
467    }
468
469    /// Build the compiled binary in dev mode.
470    /// Returns the path to the compiled binary.
471    ///
472    /// This shares the same parallel-compilation machinery as the simulator: the
473    /// program is linked dynamically against a prebuilt dylib of its dependencies,
474    /// so repeated and concurrent builds only need to recompile the generated
475    /// example itself.
476    pub fn build(&self) -> Result<PathBuf, Error> {
477        let bin_name = self
478            .bin_name
479            .as_ref()
480            .expect("No binary name set - did you call deploy?");
481        let trybuild = self
482            .trybuild
483            .as_ref()
484            .expect("No trybuild config set - did you call deploy?");
485
486        let out = compile_trybuild_example(ExampleBuildConfig {
487            trybuild: trybuild.clone(),
488            bin_name: bin_name.clone(),
489            runtime_feature: "hydro___feature_maelstrom_runtime",
490            // Maelstrom builds the generated example directly as an executable.
491            example_name: bin_name.clone(),
492            crate_type: None,
493            set_trybuild_lib_name: false,
494            allow_fuzz: false,
495        })
496        .map_err(|()| Error::other("Maelstrom binary compilation failed"))?;
497
498        // Persist the built executable so it survives past the temporary build guards.
499        out.keep().map_err(|e| Error::other(e.to_string()))
500    }
501
502    /// Run Maelstrom with the compiled binary, return Ok(()) if all checks pass.
503    ///
504    /// This will block until Maelstrom completes.
505    pub fn run(self) -> Result<(), Error> {
506        let binary_path = self.build()?;
507
508        // Use a unique working directory per run to avoid conflicts with concurrent tests.
509        let run_dir = tempfile::tempdir().map_err(Error::other)?;
510
511        let mut cmd = std::process::Command::new(&self.maelstrom_path);
512        cmd.arg("test")
513            .arg("-w")
514            .arg(&self.workload)
515            .arg("--bin")
516            .arg(&binary_path)
517            .arg("--node-count")
518            .arg(self.node_count.to_string())
519            .current_dir(run_dir.path())
520            .stdout(Stdio::piped());
521
522        if let Some(time_limit) = self.time_limit {
523            cmd.arg("--time-limit").arg(time_limit.to_string());
524        }
525
526        if let Some(rate) = self.rate {
527            cmd.arg("--rate").arg(rate.to_string());
528        }
529
530        if let Some(availability) = self.availability {
531            cmd.arg("--availability").arg(availability);
532        }
533
534        if let Some(nemesis) = self.nemesis {
535            cmd.arg("--nemesis").arg(nemesis);
536        }
537
538        for arg in &self.extra_args {
539            cmd.arg(arg);
540        }
541
542        let spawned = cmd.spawn()?;
543
544        for line in BufReader::new(spawned.stdout.unwrap()).lines() {
545            let line = line?;
546            eprintln!("{}", &line);
547
548            if line.starts_with("Analysis invalid!") {
549                let path = run_dir.keep();
550                return Err(Error::other(format!(
551                    "Analysis was invalid. Maelstrom store at: {}",
552                    path.display()
553                )));
554            } else if line.starts_with("Errors occurred during analysis, but no anomalies found.")
555                || line.starts_with("Everything looks good!")
556            {
557                return Ok(());
558            }
559        }
560
561        let path = run_dir.keep();
562        Err(Error::other(format!(
563            "Maelstrom produced an unexpected result. Store at: {}",
564            path.display()
565        )))
566    }
567
568    /// Get the path to the compiled binary, building it if necessary.
569    pub fn binary_path(&self) -> Option<PathBuf> {
570        self.build().ok()
571    }
572}
hydro_lang/deploy/maelstrom/deploy_maelstrom.rs

hydro_lang/deploy/maelstrom/
deploy_maelstrom.rs
