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    name_hint: Option<String>,
486    cluster_idx: Option<usize>,
487}
488
489impl From<Arc<dyn Host>> for TrybuildHost {
490    fn from(host: Arc<dyn Host>) -> Self {
491        Self {
492            host,
493            display_name: None,
494            rustflags: None,
495            profile: None,
496            additional_hydro_features: vec![],
497            features: vec![],
498            tracing: None,
499            build_envs: vec![],
500            env: HashMap::new(),
501            name_hint: None,
502            cluster_idx: None,
503        }
504    }
505}
506
507impl<H: Host + 'static> From<Arc<H>> for TrybuildHost {
508    fn from(host: Arc<H>) -> Self {
509        Self {
510            host,
511            display_name: None,
512            rustflags: None,
513            profile: None,
514            additional_hydro_features: vec![],
515            features: vec![],
516            tracing: None,
517            build_envs: vec![],
518            env: HashMap::new(),
519            name_hint: None,
520            cluster_idx: None,
521        }
522    }
523}
524
525#[expect(missing_docs, reason = "TODO")]
526impl TrybuildHost {
527    pub fn new(host: Arc<dyn Host>) -> Self {
528        Self {
529            host,
530            display_name: None,
531            rustflags: None,
532            profile: None,
533            additional_hydro_features: vec![],
534            features: vec![],
535            tracing: None,
536            build_envs: vec![],
537            env: HashMap::new(),
538            name_hint: None,
539            cluster_idx: None,
540        }
541    }
542
543    pub fn display_name(self, display_name: impl Into<String>) -> Self {
544        if self.display_name.is_some() {
545            panic!("{} already set", name_of!(display_name in Self));
546        }
547
548        Self {
549            display_name: Some(display_name.into()),
550            ..self
551        }
552    }
553
554    pub fn rustflags(self, rustflags: impl Into<String>) -> Self {
555        if self.rustflags.is_some() {
556            panic!("{} already set", name_of!(rustflags in Self));
557        }
558
559        Self {
560            rustflags: Some(rustflags.into()),
561            ..self
562        }
563    }
564
565    pub fn profile(self, profile: impl Into<String>) -> Self {
566        if self.profile.is_some() {
567            panic!("{} already set", name_of!(profile in Self));
568        }
569
570        Self {
571            profile: Some(profile.into()),
572            ..self
573        }
574    }
575
576    pub fn additional_hydro_features(self, additional_hydro_features: Vec<String>) -> Self {
577        Self {
578            additional_hydro_features,
579            ..self
580        }
581    }
582
583    pub fn features(self, features: Vec<String>) -> Self {
584        Self {
585            features: self.features.into_iter().chain(features).collect(),
586            ..self
587        }
588    }
589
590    pub fn tracing(self, tracing: TracingOptions) -> Self {
591        if self.tracing.is_some() {
592            panic!("{} already set", name_of!(tracing in Self));
593        }
594
595        Self {
596            tracing: Some(tracing),
597            ..self
598        }
599    }
600
601    pub fn build_env(self, key: impl Into<String>, value: impl Into<String>) -> Self {
602        Self {
603            build_envs: self
604                .build_envs
605                .into_iter()
606                .chain(std::iter::once((key.into(), value.into())))
607                .collect(),
608            ..self
609        }
610    }
611
612    pub fn env(self, key: impl Into<String>, value: impl Into<String>) -> Self {
613        let mut env = self.env;
614        env.insert(key.into(), value.into());
615        Self { env, ..self }
616    }
617}
618
619impl IntoProcessSpec<'_, HydroDeploy> for Arc<dyn Host> {
620    type ProcessSpec = TrybuildHost;
621    fn into_process_spec(self) -> TrybuildHost {
622        TrybuildHost {
623            host: self,
624            display_name: None,
625            rustflags: None,
626            profile: None,
627            additional_hydro_features: vec![],
628            features: vec![],
629            tracing: None,
630            build_envs: vec![],
631            env: HashMap::new(),
632            name_hint: None,
633            cluster_idx: None,
634        }
635    }
636}
637
638impl<H: Host + 'static> IntoProcessSpec<'_, HydroDeploy> for Arc<H> {
639    type ProcessSpec = TrybuildHost;
640    fn into_process_spec(self) -> TrybuildHost {
641        TrybuildHost {
642            host: self,
643            display_name: None,
644            rustflags: None,
645            profile: None,
646            additional_hydro_features: vec![],
647            features: vec![],
648            tracing: None,
649            build_envs: vec![],
650            env: HashMap::new(),
651            name_hint: None,
652            cluster_idx: None,
653        }
654    }
655}
656
657#[expect(missing_docs, reason = "TODO")]
658#[derive(Clone)]
659pub struct DeployExternal {
660    next_port: Rc<RefCell<usize>>,
661    host: Arc<dyn Host>,
662    underlying: Rc<RefCell<Option<Arc<CustomService>>>>,
663    client_ports: Rc<RefCell<HashMap<String, CustomClientPort>>>,
664    allocated_ports: Rc<RefCell<HashMap<ExternalPortId, String>>>,
665}
666
667impl DeployExternal {
668    pub(crate) fn raw_port(&self, external_port_id: ExternalPortId) -> CustomClientPort {
669        self.client_ports
670            .borrow()
671            .get(
672                self.allocated_ports
673                    .borrow()
674                    .get(&external_port_id)
675                    .unwrap(),
676            )
677            .unwrap()
678            .clone()
679    }
680}
681
682impl<'a> RegisterPort<'a, HydroDeploy> for DeployExternal {
683    fn register(&self, external_port_id: ExternalPortId, port: Self::Port) {
684        assert!(
685            self.allocated_ports
686                .borrow_mut()
687                .insert(external_port_id, port.clone())
688                .is_none_or(|old| old == port)
689        );
690    }
691
692    fn as_bytes_bidi(
693        &self,
694        external_port_id: ExternalPortId,
695    ) -> impl Future<
696        Output = (
697            Pin<Box<dyn Stream<Item = Result<BytesMut, Error>>>>,
698            Pin<Box<dyn Sink<Bytes, Error = Error>>>,
699        ),
700    > + 'a {
701        let port = self.raw_port(external_port_id);
702
703        async move {
704            let (source, sink) = port.connect().await.into_source_sink();
705            (
706                Box::pin(source) as Pin<Box<dyn Stream<Item = Result<BytesMut, Error>>>>,
707                Box::pin(sink) as Pin<Box<dyn Sink<Bytes, Error = Error>>>,
708            )
709        }
710    }
711
712    fn as_bincode_bidi<InT, OutT>(
713        &self,
714        external_port_id: ExternalPortId,
715    ) -> impl Future<
716        Output = (
717            Pin<Box<dyn Stream<Item = OutT>>>,
718            Pin<Box<dyn Sink<InT, Error = Error>>>,
719        ),
720    > + 'a
721    where
722        InT: Serialize + 'static,
723        OutT: DeserializeOwned + 'static,
724    {
725        let port = self.raw_port(external_port_id);
726        async move {
727            let (source, sink) = port.connect().await.into_source_sink();
728            (
729                Box::pin(source.map(|item| bincode::deserialize(&item.unwrap()).unwrap()))
730                    as Pin<Box<dyn Stream<Item = OutT>>>,
731                Box::pin(
732                    sink.with(|item| async move { Ok(bincode::serialize(&item).unwrap().into()) }),
733                ) as Pin<Box<dyn Sink<InT, Error = Error>>>,
734            )
735        }
736    }
737
738    fn as_bincode_sink<T: Serialize + 'static>(
739        &self,
740        external_port_id: ExternalPortId,
741    ) -> impl Future<Output = Pin<Box<dyn Sink<T, Error = Error>>>> + 'a {
742        let port = self.raw_port(external_port_id);
743        async move {
744            let sink = port.connect().await.into_sink();
745            Box::pin(sink.with(|item| async move { Ok(bincode::serialize(&item).unwrap().into()) }))
746                as Pin<Box<dyn Sink<T, Error = Error>>>
747        }
748    }
749
750    fn as_bincode_source<T: DeserializeOwned + 'static>(
751        &self,
752        external_port_id: ExternalPortId,
753    ) -> impl Future<Output = Pin<Box<dyn Stream<Item = T>>>> + 'a {
754        let port = self.raw_port(external_port_id);
755        async move {
756            let source = port.connect().await.into_source();
757            Box::pin(source.map(|item| bincode::deserialize(&item.unwrap()).unwrap()))
758                as Pin<Box<dyn Stream<Item = T>>>
759        }
760    }
761}
762
763impl Node for DeployExternal {
764    type Port = String;
765    /// Map from Cluster location ID to member IDs.
766    type Meta = SparseSecondaryMap<LocationKey, Vec<TaglessMemberId>>;
767    type InstantiateEnv = Deployment;
768
769    fn next_port(&self) -> Self::Port {
770        let next_port = *self.next_port.borrow();
771        *self.next_port.borrow_mut() += 1;
772
773        format!("port_{}", next_port)
774    }
775
776    fn instantiate(
777        &self,
778        env: &mut Self::InstantiateEnv,
779        _meta: &mut Self::Meta,
780        _graph: DfirGraph,
781        extra_stmts: &[syn::Stmt],
782        sidecars: &[syn::Expr],
783    ) {
784        assert!(extra_stmts.is_empty());
785        assert!(sidecars.is_empty());
786        let service = env.CustomService(self.host.clone(), vec![]);
787        *self.underlying.borrow_mut() = Some(service);
788    }
789
790    fn update_meta(&self, _meta: &Self::Meta) {}
791}
792
793impl ExternalSpec<'_, HydroDeploy> for Arc<dyn Host> {
794    fn build(self, _key: LocationKey, _name_hint: &str) -> DeployExternal {
795        DeployExternal {
796            next_port: Rc::new(RefCell::new(0)),
797            host: self,
798            underlying: Rc::new(RefCell::new(None)),
799            allocated_ports: Rc::new(RefCell::new(HashMap::new())),
800            client_ports: Rc::new(RefCell::new(HashMap::new())),
801        }
802    }
803}
804
805impl<H: Host + 'static> ExternalSpec<'_, HydroDeploy> for Arc<H> {
806    fn build(self, _key: LocationKey, _name_hint: &str) -> DeployExternal {
807        DeployExternal {
808            next_port: Rc::new(RefCell::new(0)),
809            host: self,
810            underlying: Rc::new(RefCell::new(None)),
811            allocated_ports: Rc::new(RefCell::new(HashMap::new())),
812            client_ports: Rc::new(RefCell::new(HashMap::new())),
813        }
814    }
815}
816
817pub(crate) enum CrateOrTrybuild {
818    Crate(RustCrate, Arc<dyn Host>),
819    Trybuild(TrybuildHost),
820}
821
822#[expect(missing_docs, reason = "TODO")]
823#[derive(Clone)]
824pub struct DeployNode {
825    next_port: Rc<RefCell<usize>>,
826    service_spec: Rc<RefCell<Option<CrateOrTrybuild>>>,
827    underlying: Rc<RefCell<Option<Arc<RustCrateService>>>>,
828}
829
830impl DeployCrateWrapper for DeployNode {
831    fn underlying(&self) -> Arc<RustCrateService> {
832        Arc::clone(self.underlying.borrow().as_ref().unwrap())
833    }
834}
835
836impl Node for DeployNode {
837    type Port = String;
838    /// Map from Cluster location ID to member IDs.
839    type Meta = SparseSecondaryMap<LocationKey, Vec<TaglessMemberId>>;
840    type InstantiateEnv = Deployment;
841
842    fn next_port(&self) -> String {
843        let next_port = *self.next_port.borrow();
844        *self.next_port.borrow_mut() += 1;
845
846        format!("port_{}", next_port)
847    }
848
849    fn update_meta(&self, meta: &Self::Meta) {
850        let underlying_node = self.underlying.borrow();
851        underlying_node.as_ref().unwrap().update_meta(HydroMeta {
852            clusters: meta.clone(),
853            cluster_id: None,
854        });
855    }
856
857    fn instantiate(
858        &self,
859        env: &mut Self::InstantiateEnv,
860        _meta: &mut Self::Meta,
861        graph: DfirGraph,
862        extra_stmts: &[syn::Stmt],
863        sidecars: &[syn::Expr],
864    ) {
865        let (service, host) = match self.service_spec.borrow_mut().take().unwrap() {
866            CrateOrTrybuild::Crate(c, host) => (c, host),
867            CrateOrTrybuild::Trybuild(trybuild) => {
868                // Determine linking mode based on host target type
869                let linking_mode = if !cfg!(target_os = "windows")
870                    && trybuild.host.target_type() == hydro_deploy::HostTargetType::Local
871                {
872                    // When compiling for local, prefer dynamic linking to reduce binary size
873                    // Windows is currently not supported due to https://github.com/bevyengine/bevy/pull/2016
874                    LinkingMode::Dynamic
875                } else {
876                    LinkingMode::Static
877                };
878                let (bin_name, config) = create_graph_trybuild(
879                    graph,
880                    extra_stmts,
881                    sidecars,
882                    trybuild.name_hint.as_deref(),
883                    crate::compile::trybuild::generate::DeployMode::HydroDeploy,
884                    linking_mode,
885                );
886                let host = trybuild.host.clone();
887                (
888                    create_trybuild_service(
889                        trybuild,
890                        &config.project_dir,
891                        &config.target_dir,
892                        config.features.as_deref(),
893                        &bin_name,
894                        &config.linking_mode,
895                    ),
896                    host,
897                )
898            }
899        };
900
901        *self.underlying.borrow_mut() = Some(env.add_service(service, host));
902    }
903}
904
905#[expect(missing_docs, reason = "TODO")]
906#[derive(Clone)]
907pub struct DeployClusterNode {
908    underlying: Arc<RustCrateService>,
909}
910
911impl DeployCrateWrapper for DeployClusterNode {
912    fn underlying(&self) -> Arc<RustCrateService> {
913        self.underlying.clone()
914    }
915}
916#[expect(missing_docs, reason = "TODO")]
917#[derive(Clone)]
918pub struct DeployCluster {
919    key: LocationKey,
920    next_port: Rc<RefCell<usize>>,
921    cluster_spec: Rc<RefCell<Option<Vec<CrateOrTrybuild>>>>,
922    members: Rc<RefCell<Vec<DeployClusterNode>>>,
923    name_hint: Option<String>,
924}
925
926impl DeployCluster {
927    #[expect(missing_docs, reason = "TODO")]
928    pub fn members(&self) -> Vec<DeployClusterNode> {
929        self.members.borrow().clone()
930    }
931}
932
933impl Node for DeployCluster {
934    type Port = String;
935    /// Map from Cluster location ID to member IDs.
936    type Meta = SparseSecondaryMap<LocationKey, Vec<TaglessMemberId>>;
937    type InstantiateEnv = Deployment;
938
939    fn next_port(&self) -> String {
940        let next_port = *self.next_port.borrow();
941        *self.next_port.borrow_mut() += 1;
942
943        format!("port_{}", next_port)
944    }
945
946    fn instantiate(
947        &self,
948        env: &mut Self::InstantiateEnv,
949        meta: &mut Self::Meta,
950        graph: DfirGraph,
951        extra_stmts: &[syn::Stmt],
952        sidecars: &[syn::Expr],
953    ) {
954        let has_trybuild = self
955            .cluster_spec
956            .borrow()
957            .as_ref()
958            .unwrap()
959            .iter()
960            .any(|spec| matches!(spec, CrateOrTrybuild::Trybuild { .. }));
961
962        // For clusters, use static linking if ANY host is non-local (conservative approach)
963        let linking_mode = if !cfg!(target_os = "windows")
964            && self
965                .cluster_spec
966                .borrow()
967                .as_ref()
968                .unwrap()
969                .iter()
970                .all(|spec| match spec {
971                    CrateOrTrybuild::Crate(_, _) => true, // crates handle their own linking
972                    CrateOrTrybuild::Trybuild(t) => {
973                        t.host.target_type() == hydro_deploy::HostTargetType::Local
974                    }
975                }) {
976            // See comment above for Windows exception
977            LinkingMode::Dynamic
978        } else {
979            LinkingMode::Static
980        };
981
982        let maybe_trybuild = if has_trybuild {
983            Some(create_graph_trybuild(
984                graph,
985                extra_stmts,
986                sidecars,
987                self.name_hint.as_deref(),
988                crate::compile::trybuild::generate::DeployMode::HydroDeploy,
989                linking_mode,
990            ))
991        } else {
992            None
993        };
994
995        let cluster_nodes = self
996            .cluster_spec
997            .borrow_mut()
998            .take()
999            .unwrap()
1000            .into_iter()
1001            .map(|spec| {
1002                let (service, host) = match spec {
1003                    CrateOrTrybuild::Crate(c, host) => (c, host),
1004                    CrateOrTrybuild::Trybuild(trybuild) => {
1005                        let (bin_name, config) = maybe_trybuild.as_ref().unwrap();
1006                        let host = trybuild.host.clone();
1007                        (
1008                            create_trybuild_service(
1009                                trybuild,
1010                                &config.project_dir,
1011                                &config.target_dir,
1012                                config.features.as_deref(),
1013                                bin_name,
1014                                &config.linking_mode,
1015                            ),
1016                            host,
1017                        )
1018                    }
1019                };
1020
1021                env.add_service(service, host)
1022            })
1023            .collect::<Vec<_>>();
1024        meta.insert(
1025            self.key,
1026            (0..(cluster_nodes.len() as u32))
1027                .map(TaglessMemberId::from_raw_id)
1028                .collect(),
1029        );
1030        *self.members.borrow_mut() = cluster_nodes
1031            .into_iter()
1032            .map(|n| DeployClusterNode { underlying: n })
1033            .collect();
1034    }
1035
1036    fn update_meta(&self, meta: &Self::Meta) {
1037        for (cluster_id, node) in self.members.borrow().iter().enumerate() {
1038            node.underlying.update_meta(HydroMeta {
1039                clusters: meta.clone(),
1040                cluster_id: Some(TaglessMemberId::from_raw_id(cluster_id as u32)),
1041            });
1042        }
1043    }
1044}
1045
1046#[expect(missing_docs, reason = "TODO")]
1047#[derive(Clone)]
1048pub struct DeployProcessSpec(RustCrate, Arc<dyn Host>);
1049
1050impl DeployProcessSpec {
1051    #[expect(missing_docs, reason = "TODO")]
1052    pub fn new(t: RustCrate, host: Arc<dyn Host>) -> Self {
1053        Self(t, host)
1054    }
1055}
1056
1057impl ProcessSpec<'_, HydroDeploy> for DeployProcessSpec {
1058    fn build(self, _key: LocationKey, _name_hint: &str) -> DeployNode {
1059        DeployNode {
1060            next_port: Rc::new(RefCell::new(0)),
1061            service_spec: Rc::new(RefCell::new(Some(CrateOrTrybuild::Crate(self.0, self.1)))),
1062            underlying: Rc::new(RefCell::new(None)),
1063        }
1064    }
1065}
1066
1067impl ProcessSpec<'_, HydroDeploy> for TrybuildHost {
1068    fn build(mut self, key: LocationKey, name_hint: &str) -> DeployNode {
1069        self.name_hint = Some(format!("{} (process {})", name_hint, key));
1070        DeployNode {
1071            next_port: Rc::new(RefCell::new(0)),
1072            service_spec: Rc::new(RefCell::new(Some(CrateOrTrybuild::Trybuild(self)))),
1073            underlying: Rc::new(RefCell::new(None)),
1074        }
1075    }
1076}
1077
1078#[expect(missing_docs, reason = "TODO")]
1079#[derive(Clone)]
1080pub struct DeployClusterSpec(Vec<(RustCrate, Arc<dyn Host>)>);
1081
1082impl DeployClusterSpec {
1083    #[expect(missing_docs, reason = "TODO")]
1084    pub fn new(crates: Vec<(RustCrate, Arc<dyn Host>)>) -> Self {
1085        Self(crates)
1086    }
1087}
1088
1089impl ClusterSpec<'_, HydroDeploy> for DeployClusterSpec {
1090    fn build(self, key: LocationKey, _name_hint: &str) -> DeployCluster {
1091        DeployCluster {
1092            key,
1093            next_port: Rc::new(RefCell::new(0)),
1094            cluster_spec: Rc::new(RefCell::new(Some(
1095                self.0
1096                    .into_iter()
1097                    .map(|(c, h)| CrateOrTrybuild::Crate(c, h))
1098                    .collect(),
1099            ))),
1100            members: Rc::new(RefCell::new(vec![])),
1101            name_hint: None,
1102        }
1103    }
1104}
1105
1106impl<T: Into<TrybuildHost>, I: IntoIterator<Item = T>> ClusterSpec<'_, HydroDeploy> for I {
1107    fn build(self, key: LocationKey, name_hint: &str) -> DeployCluster {
1108        let name_hint = format!("{} (cluster {})", name_hint, key);
1109        DeployCluster {
1110            key,
1111            next_port: Rc::new(RefCell::new(0)),
1112            cluster_spec: Rc::new(RefCell::new(Some(
1113                self.into_iter()
1114                    .enumerate()
1115                    .map(|(idx, b)| {
1116                        let mut b = b.into();
1117                        b.name_hint = Some(name_hint.clone());
1118                        b.cluster_idx = Some(idx);
1119                        CrateOrTrybuild::Trybuild(b)
1120                    })
1121                    .collect(),
1122            ))),
1123            members: Rc::new(RefCell::new(vec![])),
1124            name_hint: Some(name_hint),
1125        }
1126    }
1127}
1128
1129fn create_trybuild_service(
1130    trybuild: TrybuildHost,
1131    dir: &std::path::Path,
1132    target_dir: &std::path::PathBuf,
1133    features: Option<&[String]>,
1134    bin_name: &str,
1135    linking_mode: &LinkingMode,
1136) -> RustCrate {
1137    // For dynamic linking, use the dylib-examples crate; for static, use the base crate
1138    let crate_dir = match linking_mode {
1139        LinkingMode::Dynamic => dir.join("dylib-examples"),
1140        LinkingMode::Static => dir.to_path_buf(),
1141    };
1142
1143    let mut ret = RustCrate::new(&crate_dir, dir)
1144        .target_dir(target_dir)
1145        .example(bin_name)
1146        .no_default_features();
1147
1148    ret = ret.set_is_dylib(matches!(linking_mode, LinkingMode::Dynamic));
1149
1150    if let Some(display_name) = trybuild.display_name {
1151        ret = ret.display_name(display_name);
1152    } else if let Some(name_hint) = trybuild.name_hint {
1153        if let Some(cluster_idx) = trybuild.cluster_idx {
1154            ret = ret.display_name(format!("{} / {}", name_hint, cluster_idx));
1155        } else {
1156            ret = ret.display_name(name_hint);
1157        }
1158    }
1159
1160    if let Some(rustflags) = trybuild.rustflags {
1161        ret = ret.rustflags(rustflags);
1162    }
1163
1164    if let Some(profile) = trybuild.profile {
1165        ret = ret.profile(profile);
1166    }
1167
1168    if let Some(tracing) = trybuild.tracing {
1169        ret = ret.tracing(tracing);
1170    }
1171
1172    ret = ret.features(
1173        vec!["hydro___feature_deploy_integration".to_owned()]
1174            .into_iter()
1175            .chain(
1176                trybuild
1177                    .additional_hydro_features
1178                    .into_iter()
1179                    .map(|runtime_feature| {
1180                        assert!(
1181                            HYDRO_RUNTIME_FEATURES.iter().any(|f| f == &runtime_feature),
1182                            "{runtime_feature} is not a valid Hydro runtime feature"
1183                        );
1184                        format!("hydro___feature_{runtime_feature}")
1185                    }),
1186            )
1187            .chain(trybuild.features),
1188    );
1189
1190    for (key, value) in trybuild.build_envs {
1191        ret = ret.build_env(key, value);
1192    }
1193
1194    for (key, value) in trybuild.env {
1195        ret = ret.env(key, value);
1196    }
1197
1198    ret = ret.build_env("STAGELEFT_TRYBUILD_BUILD_STAGED", "1");
1199    ret = ret.config("build.incremental = false");
1200
1201    if let Some(features) = features {
1202        ret = ret.features(features);
1203    }
1204
1205    ret
1206}
hydro_lang/deploy/deploy_graph.rs

hydro_lang/deploy/
deploy_graph.rs
