hydro_lang/deploy/
deploy_graph.rs

1//! Deployment backend for Hydro that uses [`hydro_deploy`] to provision and launch services.
2
3use std::cell::RefCell;
4use std::collections::HashMap;
5use std::future::Future;
6use std::io::Error;
7use std::pin::Pin;
8use std::rc::Rc;
9use std::sync::Arc;
10
11use bytes::{Bytes, BytesMut};
12use dfir_lang::graph::DfirGraph;
13use futures::{Sink, SinkExt, Stream, StreamExt};
14use hydro_deploy::custom_service::CustomClientPort;
15use hydro_deploy::rust_crate::RustCrateService;
16use hydro_deploy::rust_crate::ports::{DemuxSink, RustCrateSink, RustCrateSource, TaggedSource};
17use hydro_deploy::rust_crate::tracing_options::TracingOptions;
18use hydro_deploy::{CustomService, Deployment, Host, RustCrate};
19use hydro_deploy_integration::{ConnectedSink, ConnectedSource};
20use nameof::name_of;
21use proc_macro2::Span;
22use serde::Serialize;
23use serde::de::DeserializeOwned;
24use slotmap::SparseSecondaryMap;
25use stageleft::{QuotedWithContext, RuntimeData};
26use syn::parse_quote;
27
28use super::deploy_runtime::*;
29use crate::compile::builder::ExternalPortId;
30use crate::compile::deploy_provider::{
31    ClusterSpec, Deploy, ExternalSpec, IntoProcessSpec, Node, ProcessSpec, RegisterPort,
32};
33use crate::compile::trybuild::generate::{
34    HYDRO_RUNTIME_FEATURES, LinkingMode, create_graph_trybuild,
35};
36use crate::location::dynamic::LocationId;
37use crate::location::member_id::TaglessMemberId;
38use crate::location::{LocationKey, MembershipEvent, NetworkHint};
39use crate::staging_util::get_this_crate;
40
41/// Deployment backend that uses [`hydro_deploy`] for provisioning and launching.
42///
43/// Automatically used when you call [`crate::compile::builder::FlowBuilder::deploy`] and pass in
44/// an `&mut` reference to [`hydro_deploy::Deployment`] as the deployment context.
45pub enum HydroDeploy {}
46
47impl<'a> Deploy<'a> for HydroDeploy {
48    /// Map from Cluster location ID to member IDs.
49    type Meta = SparseSecondaryMap<LocationKey, Vec<TaglessMemberId>>;
50    type InstantiateEnv = Deployment;
51
52    type Process = DeployNode;
53    type Cluster = DeployCluster;
54    type External = DeployExternal;
55
56    fn o2o_sink_source(
57        _env: &mut Self::InstantiateEnv,
58        _p1: &Self::Process,
59        p1_port: &<Self::Process as Node>::Port,
60        _p2: &Self::Process,
61        p2_port: &<Self::Process as Node>::Port,
62        _name: Option<&str>,
63        networking_info: &crate::networking::NetworkingInfo,
64    ) -> (syn::Expr, syn::Expr) {
65        match networking_info {
66            crate::networking::NetworkingInfo::Tcp {
67                fault: crate::networking::TcpFault::FailStop,
68            } => {}
69            _ => panic!("Unsupported networking info: {:?}", networking_info),
70        }
71        let p1_port = p1_port.as_str();
72        let p2_port = p2_port.as_str();
73        deploy_o2o(
74            RuntimeData::new("__hydro_lang_trybuild_cli"),
75            p1_port,
76            p2_port,
77        )
78    }
79
80    fn o2o_connect(
81        p1: &Self::Process,
82        p1_port: &<Self::Process as Node>::Port,
83        p2: &Self::Process,
84        p2_port: &<Self::Process as Node>::Port,
85    ) -> Box<dyn FnOnce()> {
86        let p1 = p1.clone();
87        let p1_port = p1_port.clone();
88        let p2 = p2.clone();
89        let p2_port = p2_port.clone();
90
91        Box::new(move || {
92            let self_underlying_borrow = p1.underlying.borrow();
93            let self_underlying = self_underlying_borrow.as_ref().unwrap();
94            let source_port = self_underlying.get_port(p1_port.clone());
95
96            let other_underlying_borrow = p2.underlying.borrow();
97            let other_underlying = other_underlying_borrow.as_ref().unwrap();
98            let recipient_port = other_underlying.get_port(p2_port.clone());
99
100            source_port.send_to(&recipient_port)
101        })
102    }
103
104    fn o2m_sink_source(
105        _env: &mut Self::InstantiateEnv,
106        _p1: &Self::Process,
107        p1_port: &<Self::Process as Node>::Port,
108        _c2: &Self::Cluster,
109        c2_port: &<Self::Cluster as Node>::Port,
110        _name: Option<&str>,
111        networking_info: &crate::networking::NetworkingInfo,
112    ) -> (syn::Expr, syn::Expr) {
113        match networking_info {
114            crate::networking::NetworkingInfo::Tcp {
115                fault: crate::networking::TcpFault::FailStop,
116            } => {}
117            _ => panic!("Unsupported networking info: {:?}", networking_info),
118        }
119        let p1_port = p1_port.as_str();
120        let c2_port = c2_port.as_str();
121        deploy_o2m(
122            RuntimeData::new("__hydro_lang_trybuild_cli"),
123            p1_port,
124            c2_port,
125        )
126    }
127
128    fn o2m_connect(
129        p1: &Self::Process,
130        p1_port: &<Self::Process as Node>::Port,
131        c2: &Self::Cluster,
132        c2_port: &<Self::Cluster as Node>::Port,
133    ) -> Box<dyn FnOnce()> {
134        let p1 = p1.clone();
135        let p1_port = p1_port.clone();
136        let c2 = c2.clone();
137        let c2_port = c2_port.clone();
138
139        Box::new(move || {
140            let self_underlying_borrow = p1.underlying.borrow();
141            let self_underlying = self_underlying_borrow.as_ref().unwrap();
142            let source_port = self_underlying.get_port(p1_port.clone());
143
144            let recipient_port = DemuxSink {
145                demux: c2
146                    .members
147                    .borrow()
148                    .iter()
149                    .enumerate()
150                    .map(|(id, c)| {
151                        (
152                            id as u32,
153                            Arc::new(c.underlying.get_port(c2_port.clone()))
154                                as Arc<dyn RustCrateSink + 'static>,
155                        )
156                    })
157                    .collect(),
158            };
159
160            source_port.send_to(&recipient_port)
161        })
162    }
163
164    fn m2o_sink_source(
165        _env: &mut Self::InstantiateEnv,
166        _c1: &Self::Cluster,
167        c1_port: &<Self::Cluster as Node>::Port,
168        _p2: &Self::Process,
169        p2_port: &<Self::Process as Node>::Port,
170        _name: Option<&str>,
171        networking_info: &crate::networking::NetworkingInfo,
172    ) -> (syn::Expr, syn::Expr) {
173        match networking_info {
174            crate::networking::NetworkingInfo::Tcp {
175                fault: crate::networking::TcpFault::FailStop,
176            } => {}
177            _ => panic!("Unsupported networking info: {:?}", networking_info),
178        }
179        let c1_port = c1_port.as_str();
180        let p2_port = p2_port.as_str();
181        deploy_m2o(
182            RuntimeData::new("__hydro_lang_trybuild_cli"),
183            c1_port,
184            p2_port,
185        )
186    }
187
188    fn m2o_connect(
189        c1: &Self::Cluster,
190        c1_port: &<Self::Cluster as Node>::Port,
191        p2: &Self::Process,
192        p2_port: &<Self::Process as Node>::Port,
193    ) -> Box<dyn FnOnce()> {
194        let c1 = c1.clone();
195        let c1_port = c1_port.clone();
196        let p2 = p2.clone();
197        let p2_port = p2_port.clone();
198
199        Box::new(move || {
200            let other_underlying_borrow = p2.underlying.borrow();
201            let other_underlying = other_underlying_borrow.as_ref().unwrap();
202            let recipient_port = other_underlying.get_port(p2_port.clone()).merge();
203
204            for (i, node) in c1.members.borrow().iter().enumerate() {
205                let source_port = node.underlying.get_port(c1_port.clone());
206
207                TaggedSource {
208                    source: Arc::new(source_port),
209                    tag: i as u32,
210                }
211                .send_to(&recipient_port);
212            }
213        })
214    }
215
216    fn m2m_sink_source(
217        _env: &mut Self::InstantiateEnv,
218        _c1: &Self::Cluster,
219        c1_port: &<Self::Cluster as Node>::Port,
220        _c2: &Self::Cluster,
221        c2_port: &<Self::Cluster as Node>::Port,
222        _name: Option<&str>,
223        networking_info: &crate::networking::NetworkingInfo,
224    ) -> (syn::Expr, syn::Expr) {
225        match networking_info {
226            crate::networking::NetworkingInfo::Tcp {
227                fault: crate::networking::TcpFault::FailStop,
228            } => {}
229            _ => panic!("Unsupported networking info: {:?}", networking_info),
230        }
231        let c1_port = c1_port.as_str();
232        let c2_port = c2_port.as_str();
233        deploy_m2m(
234            RuntimeData::new("__hydro_lang_trybuild_cli"),
235            c1_port,
236            c2_port,
237        )
238    }
239
240    fn m2m_connect(
241        c1: &Self::Cluster,
242        c1_port: &<Self::Cluster as Node>::Port,
243        c2: &Self::Cluster,
244        c2_port: &<Self::Cluster as Node>::Port,
245    ) -> Box<dyn FnOnce()> {
246        let c1 = c1.clone();
247        let c1_port = c1_port.clone();
248        let c2 = c2.clone();
249        let c2_port = c2_port.clone();
250
251        Box::new(move || {
252            for (i, sender) in c1.members.borrow().iter().enumerate() {
253                let source_port = sender.underlying.get_port(c1_port.clone());
254
255                let recipient_port = DemuxSink {
256                    demux: c2
257                        .members
258                        .borrow()
259                        .iter()
260                        .enumerate()
261                        .map(|(id, c)| {
262                            (
263                                id as u32,
264                                Arc::new(c.underlying.get_port(c2_port.clone()).merge())
265                                    as Arc<dyn RustCrateSink + 'static>,
266                            )
267                        })
268                        .collect(),
269                };
270
271                TaggedSource {
272                    source: Arc::new(source_port),
273                    tag: i as u32,
274                }
275                .send_to(&recipient_port);
276            }
277        })
278    }
279
280    fn e2o_many_source(
281        extra_stmts: &mut Vec<syn::Stmt>,
282        _p2: &Self::Process,
283        p2_port: &<Self::Process as Node>::Port,
284        codec_type: &syn::Type,
285        shared_handle: String,
286    ) -> syn::Expr {
287        let connect_ident = syn::Ident::new(
288            &format!("__hydro_deploy_many_{}_connect", &shared_handle),
289            Span::call_site(),
290        );
291        let source_ident = syn::Ident::new(
292            &format!("__hydro_deploy_many_{}_source", &shared_handle),
293            Span::call_site(),
294        );
295        let sink_ident = syn::Ident::new(
296            &format!("__hydro_deploy_many_{}_sink", &shared_handle),
297            Span::call_site(),
298        );
299        let membership_ident = syn::Ident::new(
300            &format!("__hydro_deploy_many_{}_membership", &shared_handle),
301            Span::call_site(),
302        );
303
304        let root = get_this_crate();
305
306        extra_stmts.push(syn::parse_quote! {
307            let #connect_ident = __hydro_lang_trybuild_cli
308                .port(#p2_port)
309                .connect::<#root::runtime_support::hydro_deploy_integration::multi_connection::ConnectedMultiConnection<_, _, #codec_type>>();
310        });
311
312        extra_stmts.push(syn::parse_quote! {
313            let #source_ident = #connect_ident.source;
314        });
315
316        extra_stmts.push(syn::parse_quote! {
317            let #sink_ident = #connect_ident.sink;
318        });
319
320        extra_stmts.push(syn::parse_quote! {
321            let #membership_ident = #connect_ident.membership;
322        });
323
324        parse_quote!(#source_ident)
325    }
326
327    fn e2o_many_sink(shared_handle: String) -> syn::Expr {
328        let sink_ident = syn::Ident::new(
329            &format!("__hydro_deploy_many_{}_sink", &shared_handle),
330            Span::call_site(),
331        );
332        parse_quote!(#sink_ident)
333    }
334
335    fn e2o_source(
336        extra_stmts: &mut Vec<syn::Stmt>,
337        _p1: &Self::External,
338        _p1_port: &<Self::External as Node>::Port,
339        _p2: &Self::Process,
340        p2_port: &<Self::Process as Node>::Port,
341        codec_type: &syn::Type,
342        shared_handle: String,
343    ) -> syn::Expr {
344        let connect_ident = syn::Ident::new(
345            &format!("__hydro_deploy_{}_connect", &shared_handle),
346            Span::call_site(),
347        );
348        let source_ident = syn::Ident::new(
349            &format!("__hydro_deploy_{}_source", &shared_handle),
350            Span::call_site(),
351        );
352        let sink_ident = syn::Ident::new(
353            &format!("__hydro_deploy_{}_sink", &shared_handle),
354            Span::call_site(),
355        );
356
357        let root = get_this_crate();
358
359        extra_stmts.push(syn::parse_quote! {
360            let #connect_ident = __hydro_lang_trybuild_cli
361                .port(#p2_port)
362                .connect::<#root::runtime_support::hydro_deploy_integration::single_connection::ConnectedSingleConnection<_, _, #codec_type>>();
363        });
364
365        extra_stmts.push(syn::parse_quote! {
366            let #source_ident = #connect_ident.source;
367        });
368
369        extra_stmts.push(syn::parse_quote! {
370            let #sink_ident = #connect_ident.sink;
371        });
372
373        parse_quote!(#source_ident)
374    }
375
376    fn e2o_connect(
377        p1: &Self::External,
378        p1_port: &<Self::External as Node>::Port,
379        p2: &Self::Process,
380        p2_port: &<Self::Process as Node>::Port,
381        _many: bool,
382        server_hint: NetworkHint,
383    ) -> Box<dyn FnOnce()> {
384        let p1 = p1.clone();
385        let p1_port = p1_port.clone();
386        let p2 = p2.clone();
387        let p2_port = p2_port.clone();
388
389        Box::new(move || {
390            let self_underlying_borrow = p1.underlying.borrow();
391            let self_underlying = self_underlying_borrow.as_ref().unwrap();
392            let source_port = self_underlying.declare_many_client();
393
394            let other_underlying_borrow = p2.underlying.borrow();
395            let other_underlying = other_underlying_borrow.as_ref().unwrap();
396            let recipient_port = other_underlying.get_port_with_hint(
397                p2_port.clone(),
398                match server_hint {
399                    NetworkHint::Auto => hydro_deploy::PortNetworkHint::Auto,
400                    NetworkHint::TcpPort(p) => hydro_deploy::PortNetworkHint::TcpPort(p),
401                },
402            );
403
404            source_port.send_to(&recipient_port);
405
406            p1.client_ports
407                .borrow_mut()
408                .insert(p1_port.clone(), source_port);
409        })
410    }
411
412    fn o2e_sink(
413        _p1: &Self::Process,
414        _p1_port: &<Self::Process as Node>::Port,
415        _p2: &Self::External,
416        _p2_port: &<Self::External as Node>::Port,
417        shared_handle: String,
418    ) -> syn::Expr {
419        let sink_ident = syn::Ident::new(
420            &format!("__hydro_deploy_{}_sink", &shared_handle),
421            Span::call_site(),
422        );
423        parse_quote!(#sink_ident)
424    }
425
426    fn cluster_ids(
427        of_cluster: LocationKey,
428    ) -> impl QuotedWithContext<'a, &'a [TaglessMemberId], ()> + Clone + 'a {
429        cluster_members(RuntimeData::new("__hydro_lang_trybuild_cli"), of_cluster)
430    }
431
432    fn cluster_self_id() -> impl QuotedWithContext<'a, TaglessMemberId, ()> + Clone + 'a {
433        cluster_self_id(RuntimeData::new("__hydro_lang_trybuild_cli"))
434    }
435
436    fn cluster_membership_stream(
437        _env: &mut Self::InstantiateEnv,
438        _at_location: &LocationId,
439        location_id: &LocationId,
440    ) -> impl QuotedWithContext<'a, Box<dyn Stream<Item = (TaglessMemberId, MembershipEvent)> + Unpin>, ()>
441    {
442        cluster_membership_stream(location_id)
443    }
444}
445
446#[expect(missing_docs, reason = "TODO")]
447pub trait DeployCrateWrapper {
448    fn underlying(&self) -> Arc<RustCrateService>;
449
450    fn stdout(&self) -> tokio::sync::mpsc::UnboundedReceiver<String> {
451        self.underlying().stdout()
452    }
453
454    fn stderr(&self) -> tokio::sync::mpsc::UnboundedReceiver<String> {
455        self.underlying().stderr()
456    }
457
458    fn stdout_filter(
459        &self,
460        prefix: impl Into<String>,
461    ) -> tokio::sync::mpsc::UnboundedReceiver<String> {
462        self.underlying().stdout_filter(prefix.into())
463    }
464
465    fn stderr_filter(
466        &self,
467        prefix: impl Into<String>,
468    ) -> tokio::sync::mpsc::UnboundedReceiver<String> {
469        self.underlying().stderr_filter(prefix.into())
470    }
471}
472
473#[expect(missing_docs, reason = "TODO")]
474#[derive(Clone)]
475pub struct TrybuildHost {
476    host: Arc<dyn Host>,
477    display_name: Option<String>,
478    rustflags: Option<String>,
479    profile: Option<String>,
480    additional_hydro_features: Vec<String>,
481    features: Vec<String>,
482    tracing: Option<TracingOptions>,
483    build_envs: Vec<(String, String)>,
484    env: HashMap<String, String>,
485    pin_to_core: Option<usize>,
486    name_hint: Option<String>,
487    cluster_idx: Option<usize>,
488}
489
490impl From<Arc<dyn Host>> for TrybuildHost {
491    fn from(host: Arc<dyn Host>) -> Self {
492        Self {
493            host,
494            display_name: None,
495            rustflags: None,
496            profile: None,
497            additional_hydro_features: vec![],
498            features: vec![],
499            tracing: None,
500            build_envs: vec![],
501            env: HashMap::new(),
502            pin_to_core: None,
503            name_hint: None,
504            cluster_idx: None,
505        }
506    }
507}
508
509impl<H: Host + 'static> From<Arc<H>> for TrybuildHost {
510    fn from(host: Arc<H>) -> Self {
511        Self {
512            host,
513            display_name: None,
514            rustflags: None,
515            profile: None,
516            additional_hydro_features: vec![],
517            features: vec![],
518            tracing: None,
519            build_envs: vec![],
520            env: HashMap::new(),
521            pin_to_core: None,
522            name_hint: None,
523            cluster_idx: None,
524        }
525    }
526}
527
528#[expect(missing_docs, reason = "TODO")]
529impl TrybuildHost {
530    pub fn new(host: Arc<dyn Host>) -> Self {
531        Self {
532            host,
533            display_name: None,
534            rustflags: None,
535            profile: None,
536            additional_hydro_features: vec![],
537            features: vec![],
538            tracing: None,
539            build_envs: vec![],
540            env: HashMap::new(),
541            pin_to_core: None,
542            name_hint: None,
543            cluster_idx: None,
544        }
545    }
546
547    pub fn display_name(self, display_name: impl Into<String>) -> Self {
548        if self.display_name.is_some() {
549            panic!("{} already set", name_of!(display_name in Self));
550        }
551
552        Self {
553            display_name: Some(display_name.into()),
554            ..self
555        }
556    }
557
558    pub fn rustflags(self, rustflags: impl Into<String>) -> Self {
559        if self.rustflags.is_some() {
560            panic!("{} already set", name_of!(rustflags in Self));
561        }
562
563        Self {
564            rustflags: Some(rustflags.into()),
565            ..self
566        }
567    }
568
569    pub fn profile(self, profile: impl Into<String>) -> Self {
570        if self.profile.is_some() {
571            panic!("{} already set", name_of!(profile in Self));
572        }
573
574        Self {
575            profile: Some(profile.into()),
576            ..self
577        }
578    }
579
580    pub fn additional_hydro_features(self, additional_hydro_features: Vec<String>) -> Self {
581        Self {
582            additional_hydro_features,
583            ..self
584        }
585    }
586
587    pub fn features(self, features: Vec<String>) -> Self {
588        Self {
589            features: self.features.into_iter().chain(features).collect(),
590            ..self
591        }
592    }
593
594    pub fn tracing(self, tracing: TracingOptions) -> Self {
595        if self.tracing.is_some() {
596            panic!("{} already set", name_of!(tracing in Self));
597        }
598
599        Self {
600            tracing: Some(tracing),
601            ..self
602        }
603    }
604
605    pub fn build_env(self, key: impl Into<String>, value: impl Into<String>) -> Self {
606        Self {
607            build_envs: self
608                .build_envs
609                .into_iter()
610                .chain(std::iter::once((key.into(), value.into())))
611                .collect(),
612            ..self
613        }
614    }
615
616    pub fn env(self, key: impl Into<String>, value: impl Into<String>) -> Self {
617        let mut env = self.env;
618        env.insert(key.into(), value.into());
619        Self { env, ..self }
620    }
621
622    pub fn pin_to_core(self, core: usize) -> Self {
623        Self {
624            pin_to_core: Some(core),
625            ..self
626        }
627    }
628}
629
630impl IntoProcessSpec<'_, HydroDeploy> for Arc<dyn Host> {
631    type ProcessSpec = TrybuildHost;
632    fn into_process_spec(self) -> TrybuildHost {
633        TrybuildHost {
634            host: self,
635            display_name: None,
636            rustflags: None,
637            profile: None,
638            additional_hydro_features: vec![],
639            features: vec![],
640            tracing: None,
641            build_envs: vec![],
642            env: HashMap::new(),
643            pin_to_core: None,
644            name_hint: None,
645            cluster_idx: None,
646        }
647    }
648}
649
650impl<H: Host + 'static> IntoProcessSpec<'_, HydroDeploy> for Arc<H> {
651    type ProcessSpec = TrybuildHost;
652    fn into_process_spec(self) -> TrybuildHost {
653        TrybuildHost {
654            host: self,
655            display_name: None,
656            rustflags: None,
657            profile: None,
658            additional_hydro_features: vec![],
659            features: vec![],
660            tracing: None,
661            build_envs: vec![],
662            env: HashMap::new(),
663            pin_to_core: None,
664            name_hint: None,
665            cluster_idx: None,
666        }
667    }
668}
669
670#[expect(missing_docs, reason = "TODO")]
671#[derive(Clone)]
672pub struct DeployExternal {
673    next_port: Rc<RefCell<usize>>,
674    host: Arc<dyn Host>,
675    underlying: Rc<RefCell<Option<Arc<CustomService>>>>,
676    client_ports: Rc<RefCell<HashMap<String, CustomClientPort>>>,
677    allocated_ports: Rc<RefCell<HashMap<ExternalPortId, String>>>,
678}
679
680impl DeployExternal {
681    pub(crate) fn raw_port(&self, external_port_id: ExternalPortId) -> CustomClientPort {
682        self.client_ports
683            .borrow()
684            .get(
685                self.allocated_ports
686                    .borrow()
687                    .get(&external_port_id)
688                    .unwrap(),
689            )
690            .unwrap()
691            .clone()
692    }
693}
694
695impl<'a> RegisterPort<'a, HydroDeploy> for DeployExternal {
696    fn register(&self, external_port_id: ExternalPortId, port: Self::Port) {
697        assert!(
698            self.allocated_ports
699                .borrow_mut()
700                .insert(external_port_id, port.clone())
701                .is_none_or(|old| old == port)
702        );
703    }
704
705    fn as_bytes_bidi(
706        &self,
707        external_port_id: ExternalPortId,
708    ) -> impl Future<
709        Output = (
710            Pin<Box<dyn Stream<Item = Result<BytesMut, Error>>>>,
711            Pin<Box<dyn Sink<Bytes, Error = Error>>>,
712        ),
713    > + 'a {
714        let port = self.raw_port(external_port_id);
715
716        async move {
717            let (source, sink) = port.connect().await.into_source_sink();
718            (
719                Box::pin(source) as Pin<Box<dyn Stream<Item = Result<BytesMut, Error>>>>,
720                Box::pin(sink) as Pin<Box<dyn Sink<Bytes, Error = Error>>>,
721            )
722        }
723    }
724
725    fn as_bincode_bidi<InT, OutT>(
726        &self,
727        external_port_id: ExternalPortId,
728    ) -> impl Future<
729        Output = (
730            Pin<Box<dyn Stream<Item = OutT>>>,
731            Pin<Box<dyn Sink<InT, Error = Error>>>,
732        ),
733    > + 'a
734    where
735        InT: Serialize + 'static,
736        OutT: DeserializeOwned + 'static,
737    {
738        let port = self.raw_port(external_port_id);
739        async move {
740            let (source, sink) = port.connect().await.into_source_sink();
741            (
742                Box::pin(source.map(|item| bincode::deserialize(&item.unwrap()).unwrap()))
743                    as Pin<Box<dyn Stream<Item = OutT>>>,
744                Box::pin(
745                    sink.with(|item| async move { Ok(bincode::serialize(&item).unwrap().into()) }),
746                ) as Pin<Box<dyn Sink<InT, Error = Error>>>,
747            )
748        }
749    }
750
751    fn as_bincode_sink<T: Serialize + 'static>(
752        &self,
753        external_port_id: ExternalPortId,
754    ) -> impl Future<Output = Pin<Box<dyn Sink<T, Error = Error>>>> + 'a {
755        let port = self.raw_port(external_port_id);
756        async move {
757            let sink = port.connect().await.into_sink();
758            Box::pin(sink.with(|item| async move { Ok(bincode::serialize(&item).unwrap().into()) }))
759                as Pin<Box<dyn Sink<T, Error = Error>>>
760        }
761    }
762
763    fn as_bincode_source<T: DeserializeOwned + 'static>(
764        &self,
765        external_port_id: ExternalPortId,
766    ) -> impl Future<Output = Pin<Box<dyn Stream<Item = T>>>> + 'a {
767        let port = self.raw_port(external_port_id);
768        async move {
769            let source = port.connect().await.into_source();
770            Box::pin(source.map(|item| bincode::deserialize(&item.unwrap()).unwrap()))
771                as Pin<Box<dyn Stream<Item = T>>>
772        }
773    }
774}
775
776impl Node for DeployExternal {
777    type Port = String;
778    /// Map from Cluster location ID to member IDs.
779    type Meta = SparseSecondaryMap<LocationKey, Vec<TaglessMemberId>>;
780    type InstantiateEnv = Deployment;
781
782    fn next_port(&self) -> Self::Port {
783        let next_port = *self.next_port.borrow();
784        *self.next_port.borrow_mut() += 1;
785
786        format!("port_{}", next_port)
787    }
788
789    fn instantiate(
790        &self,
791        env: &mut Self::InstantiateEnv,
792        _meta: &mut Self::Meta,
793        _graph: DfirGraph,
794        extra_stmts: &[syn::Stmt],
795        sidecars: &[syn::Expr],
796    ) {
797        assert!(extra_stmts.is_empty());
798        assert!(sidecars.is_empty());
799        let service = env.CustomService(self.host.clone(), vec![]);
800        *self.underlying.borrow_mut() = Some(service);
801    }
802
803    fn update_meta(&self, _meta: &Self::Meta) {}
804}
805
806impl ExternalSpec<'_, HydroDeploy> for Arc<dyn Host> {
807    fn build(self, _key: LocationKey, _name_hint: &str) -> DeployExternal {
808        DeployExternal {
809            next_port: Rc::new(RefCell::new(0)),
810            host: self,
811            underlying: Rc::new(RefCell::new(None)),
812            allocated_ports: Rc::new(RefCell::new(HashMap::new())),
813            client_ports: Rc::new(RefCell::new(HashMap::new())),
814        }
815    }
816}
817
818impl<H: Host + 'static> ExternalSpec<'_, HydroDeploy> for Arc<H> {
819    fn build(self, _key: LocationKey, _name_hint: &str) -> DeployExternal {
820        DeployExternal {
821            next_port: Rc::new(RefCell::new(0)),
822            host: self,
823            underlying: Rc::new(RefCell::new(None)),
824            allocated_ports: Rc::new(RefCell::new(HashMap::new())),
825            client_ports: Rc::new(RefCell::new(HashMap::new())),
826        }
827    }
828}
829
830pub(crate) enum CrateOrTrybuild {
831    Crate(RustCrate, Arc<dyn Host>),
832    Trybuild(TrybuildHost),
833}
834
835#[expect(missing_docs, reason = "TODO")]
836#[derive(Clone)]
837pub struct DeployNode {
838    next_port: Rc<RefCell<usize>>,
839    service_spec: Rc<RefCell<Option<CrateOrTrybuild>>>,
840    underlying: Rc<RefCell<Option<Arc<RustCrateService>>>>,
841}
842
843impl DeployCrateWrapper for DeployNode {
844    fn underlying(&self) -> Arc<RustCrateService> {
845        Arc::clone(self.underlying.borrow().as_ref().unwrap())
846    }
847}
848
849impl Node for DeployNode {
850    type Port = String;
851    /// Map from Cluster location ID to member IDs.
852    type Meta = SparseSecondaryMap<LocationKey, Vec<TaglessMemberId>>;
853    type InstantiateEnv = Deployment;
854
855    fn next_port(&self) -> String {
856        let next_port = *self.next_port.borrow();
857        *self.next_port.borrow_mut() += 1;
858
859        format!("port_{}", next_port)
860    }
861
862    fn update_meta(&self, meta: &Self::Meta) {
863        let underlying_node = self.underlying.borrow();
864        underlying_node.as_ref().unwrap().update_meta(HydroMeta {
865            clusters: meta.clone(),
866            cluster_id: None,
867        });
868    }
869
870    fn instantiate(
871        &self,
872        env: &mut Self::InstantiateEnv,
873        _meta: &mut Self::Meta,
874        graph: DfirGraph,
875        extra_stmts: &[syn::Stmt],
876        sidecars: &[syn::Expr],
877    ) {
878        let (service, host) = match self.service_spec.borrow_mut().take().unwrap() {
879            CrateOrTrybuild::Crate(c, host) => (c, host),
880            CrateOrTrybuild::Trybuild(trybuild) => {
881                // Determine linking mode based on host target type
882                let linking_mode = if !cfg!(target_os = "windows")
883                    && trybuild.host.target_type() == hydro_deploy::HostTargetType::Local
884                    && trybuild.rustflags.is_none()
885                {
886                    // When compiling for local, prefer dynamic linking to reduce binary size
887                    // Windows is currently not supported due to https://github.com/bevyengine/bevy/pull/2016
888                    LinkingMode::Dynamic
889                } else {
890                    LinkingMode::Static
891                };
892                let (bin_name, config) = create_graph_trybuild(
893                    graph,
894                    extra_stmts,
895                    sidecars,
896                    trybuild.name_hint.as_deref(),
897                    crate::compile::trybuild::generate::DeployMode::HydroDeploy,
898                    linking_mode,
899                );
900                let host = trybuild.host.clone();
901                (
902                    create_trybuild_service(
903                        trybuild,
904                        &config.project_dir,
905                        &config.target_dir,
906                        config.features.as_deref(),
907                        &bin_name,
908                        &config.linking_mode,
909                    ),
910                    host,
911                )
912            }
913        };
914
915        *self.underlying.borrow_mut() = Some(env.add_service(service, host));
916    }
917}
918
919#[expect(missing_docs, reason = "TODO")]
920#[derive(Clone)]
921pub struct DeployClusterNode {
922    underlying: Arc<RustCrateService>,
923}
924
925impl DeployCrateWrapper for DeployClusterNode {
926    fn underlying(&self) -> Arc<RustCrateService> {
927        self.underlying.clone()
928    }
929}
930#[expect(missing_docs, reason = "TODO")]
931#[derive(Clone)]
932pub struct DeployCluster {
933    key: LocationKey,
934    next_port: Rc<RefCell<usize>>,
935    cluster_spec: Rc<RefCell<Option<Vec<CrateOrTrybuild>>>>,
936    members: Rc<RefCell<Vec<DeployClusterNode>>>,
937    name_hint: Option<String>,
938}
939
940impl DeployCluster {
941    #[expect(missing_docs, reason = "TODO")]
942    pub fn members(&self) -> Vec<DeployClusterNode> {
943        self.members.borrow().clone()
944    }
945}
946
947impl Node for DeployCluster {
948    type Port = String;
949    /// Map from Cluster location ID to member IDs.
950    type Meta = SparseSecondaryMap<LocationKey, Vec<TaglessMemberId>>;
951    type InstantiateEnv = Deployment;
952
953    fn next_port(&self) -> String {
954        let next_port = *self.next_port.borrow();
955        *self.next_port.borrow_mut() += 1;
956
957        format!("port_{}", next_port)
958    }
959
960    fn instantiate(
961        &self,
962        env: &mut Self::InstantiateEnv,
963        meta: &mut Self::Meta,
964        graph: DfirGraph,
965        extra_stmts: &[syn::Stmt],
966        sidecars: &[syn::Expr],
967    ) {
968        let has_trybuild = self
969            .cluster_spec
970            .borrow()
971            .as_ref()
972            .unwrap()
973            .iter()
974            .any(|spec| matches!(spec, CrateOrTrybuild::Trybuild { .. }));
975
976        // For clusters, use static linking if ANY host is non-local (conservative approach)
977        let linking_mode = if !cfg!(target_os = "windows")
978            && self
979                .cluster_spec
980                .borrow()
981                .as_ref()
982                .unwrap()
983                .iter()
984                .all(|spec| match spec {
985                    CrateOrTrybuild::Crate(_, _) => true, // crates handle their own linking
986                    CrateOrTrybuild::Trybuild(t) => {
987                        t.host.target_type() == hydro_deploy::HostTargetType::Local
988                            && t.rustflags.is_none()
989                    }
990                }) {
991            // See comment above for Windows exception
992            LinkingMode::Dynamic
993        } else {
994            LinkingMode::Static
995        };
996
997        let maybe_trybuild = if has_trybuild {
998            Some(create_graph_trybuild(
999                graph,
1000                extra_stmts,
1001                sidecars,
1002                self.name_hint.as_deref(),
1003                crate::compile::trybuild::generate::DeployMode::HydroDeploy,
1004                linking_mode,
1005            ))
1006        } else {
1007            None
1008        };
1009
1010        let cluster_nodes = self
1011            .cluster_spec
1012            .borrow_mut()
1013            .take()
1014            .unwrap()
1015            .into_iter()
1016            .map(|spec| {
1017                let (service, host) = match spec {
1018                    CrateOrTrybuild::Crate(c, host) => (c, host),
1019                    CrateOrTrybuild::Trybuild(trybuild) => {
1020                        let (bin_name, config) = maybe_trybuild.as_ref().unwrap();
1021                        let host = trybuild.host.clone();
1022                        (
1023                            create_trybuild_service(
1024                                trybuild,
1025                                &config.project_dir,
1026                                &config.target_dir,
1027                                config.features.as_deref(),
1028                                bin_name,
1029                                &config.linking_mode,
1030                            ),
1031                            host,
1032                        )
1033                    }
1034                };
1035
1036                env.add_service(service, host)
1037            })
1038            .collect::<Vec<_>>();
1039        meta.insert(
1040            self.key,
1041            (0..(cluster_nodes.len() as u32))
1042                .map(TaglessMemberId::from_raw_id)
1043                .collect(),
1044        );
1045        *self.members.borrow_mut() = cluster_nodes
1046            .into_iter()
1047            .map(|n| DeployClusterNode { underlying: n })
1048            .collect();
1049    }
1050
1051    fn update_meta(&self, meta: &Self::Meta) {
1052        for (cluster_id, node) in self.members.borrow().iter().enumerate() {
1053            node.underlying.update_meta(HydroMeta {
1054                clusters: meta.clone(),
1055                cluster_id: Some(TaglessMemberId::from_raw_id(cluster_id as u32)),
1056            });
1057        }
1058    }
1059}
1060
1061#[expect(missing_docs, reason = "TODO")]
1062#[derive(Clone)]
1063pub struct DeployProcessSpec(RustCrate, Arc<dyn Host>);
1064
1065impl DeployProcessSpec {
1066    #[expect(missing_docs, reason = "TODO")]
1067    pub fn new(t: RustCrate, host: Arc<dyn Host>) -> Self {
1068        Self(t, host)
1069    }
1070}
1071
1072impl ProcessSpec<'_, HydroDeploy> for DeployProcessSpec {
1073    fn build(self, _key: LocationKey, _name_hint: &str) -> DeployNode {
1074        DeployNode {
1075            next_port: Rc::new(RefCell::new(0)),
1076            service_spec: Rc::new(RefCell::new(Some(CrateOrTrybuild::Crate(self.0, self.1)))),
1077            underlying: Rc::new(RefCell::new(None)),
1078        }
1079    }
1080}
1081
1082impl ProcessSpec<'_, HydroDeploy> for TrybuildHost {
1083    fn build(mut self, key: LocationKey, name_hint: &str) -> DeployNode {
1084        self.name_hint = Some(format!("{} (process {})", name_hint, key));
1085        DeployNode {
1086            next_port: Rc::new(RefCell::new(0)),
1087            service_spec: Rc::new(RefCell::new(Some(CrateOrTrybuild::Trybuild(self)))),
1088            underlying: Rc::new(RefCell::new(None)),
1089        }
1090    }
1091}
1092
1093#[expect(missing_docs, reason = "TODO")]
1094#[derive(Clone)]
1095pub struct DeployClusterSpec(Vec<(RustCrate, Arc<dyn Host>)>);
1096
1097impl DeployClusterSpec {
1098    #[expect(missing_docs, reason = "TODO")]
1099    pub fn new(crates: Vec<(RustCrate, Arc<dyn Host>)>) -> Self {
1100        Self(crates)
1101    }
1102}
1103
1104impl ClusterSpec<'_, HydroDeploy> for DeployClusterSpec {
1105    fn build(self, key: LocationKey, _name_hint: &str) -> DeployCluster {
1106        DeployCluster {
1107            key,
1108            next_port: Rc::new(RefCell::new(0)),
1109            cluster_spec: Rc::new(RefCell::new(Some(
1110                self.0
1111                    .into_iter()
1112                    .map(|(c, h)| CrateOrTrybuild::Crate(c, h))
1113                    .collect(),
1114            ))),
1115            members: Rc::new(RefCell::new(vec![])),
1116            name_hint: None,
1117        }
1118    }
1119}
1120
1121impl<T: Into<TrybuildHost>, I: IntoIterator<Item = T>> ClusterSpec<'_, HydroDeploy> for I {
1122    fn build(self, key: LocationKey, name_hint: &str) -> DeployCluster {
1123        let name_hint = format!("{} (cluster {})", name_hint, key);
1124        DeployCluster {
1125            key,
1126            next_port: Rc::new(RefCell::new(0)),
1127            cluster_spec: Rc::new(RefCell::new(Some(
1128                self.into_iter()
1129                    .enumerate()
1130                    .map(|(idx, b)| {
1131                        let mut b = b.into();
1132                        b.name_hint = Some(name_hint.clone());
1133                        b.cluster_idx = Some(idx);
1134                        CrateOrTrybuild::Trybuild(b)
1135                    })
1136                    .collect(),
1137            ))),
1138            members: Rc::new(RefCell::new(vec![])),
1139            name_hint: Some(name_hint),
1140        }
1141    }
1142}
1143
1144fn create_trybuild_service(
1145    trybuild: TrybuildHost,
1146    dir: &std::path::Path,
1147    target_dir: &std::path::PathBuf,
1148    features: Option<&[String]>,
1149    bin_name: &str,
1150    linking_mode: &LinkingMode,
1151) -> RustCrate {
1152    // For dynamic linking, use the dylib-examples crate; for static, use the base crate
1153    let crate_dir = match linking_mode {
1154        LinkingMode::Dynamic => dir.join("dylib-examples"),
1155        LinkingMode::Static => dir.to_path_buf(),
1156    };
1157
1158    let mut ret = RustCrate::new(&crate_dir, dir)
1159        .target_dir(target_dir)
1160        .example(bin_name)
1161        .no_default_features();
1162
1163    ret = ret.set_is_dylib(matches!(linking_mode, LinkingMode::Dynamic));
1164
1165    if let Some(display_name) = trybuild.display_name {
1166        ret = ret.display_name(display_name);
1167    } else if let Some(name_hint) = trybuild.name_hint {
1168        if let Some(cluster_idx) = trybuild.cluster_idx {
1169            ret = ret.display_name(format!("{} / {}", name_hint, cluster_idx));
1170        } else {
1171            ret = ret.display_name(name_hint);
1172        }
1173    }
1174
1175    if let Some(rustflags) = trybuild.rustflags {
1176        ret = ret.rustflags(rustflags);
1177    }
1178
1179    if let Some(profile) = trybuild.profile {
1180        ret = ret.profile(profile);
1181    }
1182
1183    if let Some(tracing) = trybuild.tracing {
1184        ret = ret.tracing(tracing);
1185    }
1186
1187    if let Some(core) = trybuild.pin_to_core {
1188        ret = ret.pin_to_core(core);
1189    }
1190
1191    ret = ret.features(
1192        vec!["hydro___feature_deploy_integration".to_owned()]
1193            .into_iter()
1194            .chain(
1195                trybuild
1196                    .additional_hydro_features
1197                    .into_iter()
1198                    .map(|runtime_feature| {
1199                        assert!(
1200                            HYDRO_RUNTIME_FEATURES.iter().any(|f| f == &runtime_feature),
1201                            "{runtime_feature} is not a valid Hydro runtime feature"
1202                        );
1203                        format!("hydro___feature_{runtime_feature}")
1204                    }),
1205            )
1206            .chain(trybuild.features),
1207    );
1208
1209    for (key, value) in trybuild.build_envs {
1210        ret = ret.build_env(key, value);
1211    }
1212
1213    for (key, value) in trybuild.env {
1214        ret = ret.env(key, value);
1215    }
1216
1217    ret = ret.build_env("STAGELEFT_TRYBUILD_BUILD_STAGED", "1");
1218    ret = ret.config("build.incremental = false");
1219
1220    if let Some(features) = features {
1221        ret = ret.features(features);
1222    }
1223
1224    ret
1225}
hydro_lang/deploy/deploy_graph.rs

hydro_lang/deploy/
deploy_graph.rs
