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                    && trybuild.rustflags.is_none()
872                {
873                    // When compiling for local, prefer dynamic linking to reduce binary size
874                    // Windows is currently not supported due to https://github.com/bevyengine/bevy/pull/2016
875                    LinkingMode::Dynamic
876                } else {
877                    LinkingMode::Static
878                };
879                let (bin_name, config) = create_graph_trybuild(
880                    graph,
881                    extra_stmts,
882                    sidecars,
883                    trybuild.name_hint.as_deref(),
884                    crate::compile::trybuild::generate::DeployMode::HydroDeploy,
885                    linking_mode,
886                );
887                let host = trybuild.host.clone();
888                (
889                    create_trybuild_service(
890                        trybuild,
891                        &config.project_dir,
892                        &config.target_dir,
893                        config.features.as_deref(),
894                        &bin_name,
895                        &config.linking_mode,
896                    ),
897                    host,
898                )
899            }
900        };
901
902        *self.underlying.borrow_mut() = Some(env.add_service(service, host));
903    }
904}
905
906#[expect(missing_docs, reason = "TODO")]
907#[derive(Clone)]
908pub struct DeployClusterNode {
909    underlying: Arc<RustCrateService>,
910}
911
912impl DeployCrateWrapper for DeployClusterNode {
913    fn underlying(&self) -> Arc<RustCrateService> {
914        self.underlying.clone()
915    }
916}
917#[expect(missing_docs, reason = "TODO")]
918#[derive(Clone)]
919pub struct DeployCluster {
920    key: LocationKey,
921    next_port: Rc<RefCell<usize>>,
922    cluster_spec: Rc<RefCell<Option<Vec<CrateOrTrybuild>>>>,
923    members: Rc<RefCell<Vec<DeployClusterNode>>>,
924    name_hint: Option<String>,
925}
926
927impl DeployCluster {
928    #[expect(missing_docs, reason = "TODO")]
929    pub fn members(&self) -> Vec<DeployClusterNode> {
930        self.members.borrow().clone()
931    }
932}
933
934impl Node for DeployCluster {
935    type Port = String;
936    /// Map from Cluster location ID to member IDs.
937    type Meta = SparseSecondaryMap<LocationKey, Vec<TaglessMemberId>>;
938    type InstantiateEnv = Deployment;
939
940    fn next_port(&self) -> String {
941        let next_port = *self.next_port.borrow();
942        *self.next_port.borrow_mut() += 1;
943
944        format!("port_{}", next_port)
945    }
946
947    fn instantiate(
948        &self,
949        env: &mut Self::InstantiateEnv,
950        meta: &mut Self::Meta,
951        graph: DfirGraph,
952        extra_stmts: &[syn::Stmt],
953        sidecars: &[syn::Expr],
954    ) {
955        let has_trybuild = self
956            .cluster_spec
957            .borrow()
958            .as_ref()
959            .unwrap()
960            .iter()
961            .any(|spec| matches!(spec, CrateOrTrybuild::Trybuild { .. }));
962
963        // For clusters, use static linking if ANY host is non-local (conservative approach)
964        let linking_mode = if !cfg!(target_os = "windows")
965            && self
966                .cluster_spec
967                .borrow()
968                .as_ref()
969                .unwrap()
970                .iter()
971                .all(|spec| match spec {
972                    CrateOrTrybuild::Crate(_, _) => true, // crates handle their own linking
973                    CrateOrTrybuild::Trybuild(t) => {
974                        t.host.target_type() == hydro_deploy::HostTargetType::Local
975                            && t.rustflags.is_none()
976                    }
977                }) {
978            // See comment above for Windows exception
979            LinkingMode::Dynamic
980        } else {
981            LinkingMode::Static
982        };
983
984        let maybe_trybuild = if has_trybuild {
985            Some(create_graph_trybuild(
986                graph,
987                extra_stmts,
988                sidecars,
989                self.name_hint.as_deref(),
990                crate::compile::trybuild::generate::DeployMode::HydroDeploy,
991                linking_mode,
992            ))
993        } else {
994            None
995        };
996
997        let cluster_nodes = self
998            .cluster_spec
999            .borrow_mut()
1000            .take()
1001            .unwrap()
1002            .into_iter()
1003            .map(|spec| {
1004                let (service, host) = match spec {
1005                    CrateOrTrybuild::Crate(c, host) => (c, host),
1006                    CrateOrTrybuild::Trybuild(trybuild) => {
1007                        let (bin_name, config) = maybe_trybuild.as_ref().unwrap();
1008                        let host = trybuild.host.clone();
1009                        (
1010                            create_trybuild_service(
1011                                trybuild,
1012                                &config.project_dir,
1013                                &config.target_dir,
1014                                config.features.as_deref(),
1015                                bin_name,
1016                                &config.linking_mode,
1017                            ),
1018                            host,
1019                        )
1020                    }
1021                };
1022
1023                env.add_service(service, host)
1024            })
1025            .collect::<Vec<_>>();
1026        meta.insert(
1027            self.key,
1028            (0..(cluster_nodes.len() as u32))
1029                .map(TaglessMemberId::from_raw_id)
1030                .collect(),
1031        );
1032        *self.members.borrow_mut() = cluster_nodes
1033            .into_iter()
1034            .map(|n| DeployClusterNode { underlying: n })
1035            .collect();
1036    }
1037
1038    fn update_meta(&self, meta: &Self::Meta) {
1039        for (cluster_id, node) in self.members.borrow().iter().enumerate() {
1040            node.underlying.update_meta(HydroMeta {
1041                clusters: meta.clone(),
1042                cluster_id: Some(TaglessMemberId::from_raw_id(cluster_id as u32)),
1043            });
1044        }
1045    }
1046}
1047
1048#[expect(missing_docs, reason = "TODO")]
1049#[derive(Clone)]
1050pub struct DeployProcessSpec(RustCrate, Arc<dyn Host>);
1051
1052impl DeployProcessSpec {
1053    #[expect(missing_docs, reason = "TODO")]
1054    pub fn new(t: RustCrate, host: Arc<dyn Host>) -> Self {
1055        Self(t, host)
1056    }
1057}
1058
1059impl ProcessSpec<'_, HydroDeploy> for DeployProcessSpec {
1060    fn build(self, _key: LocationKey, _name_hint: &str) -> DeployNode {
1061        DeployNode {
1062            next_port: Rc::new(RefCell::new(0)),
1063            service_spec: Rc::new(RefCell::new(Some(CrateOrTrybuild::Crate(self.0, self.1)))),
1064            underlying: Rc::new(RefCell::new(None)),
1065        }
1066    }
1067}
1068
1069impl ProcessSpec<'_, HydroDeploy> for TrybuildHost {
1070    fn build(mut self, key: LocationKey, name_hint: &str) -> DeployNode {
1071        self.name_hint = Some(format!("{} (process {})", name_hint, key));
1072        DeployNode {
1073            next_port: Rc::new(RefCell::new(0)),
1074            service_spec: Rc::new(RefCell::new(Some(CrateOrTrybuild::Trybuild(self)))),
1075            underlying: Rc::new(RefCell::new(None)),
1076        }
1077    }
1078}
1079
1080#[expect(missing_docs, reason = "TODO")]
1081#[derive(Clone)]
1082pub struct DeployClusterSpec(Vec<(RustCrate, Arc<dyn Host>)>);
1083
1084impl DeployClusterSpec {
1085    #[expect(missing_docs, reason = "TODO")]
1086    pub fn new(crates: Vec<(RustCrate, Arc<dyn Host>)>) -> Self {
1087        Self(crates)
1088    }
1089}
1090
1091impl ClusterSpec<'_, HydroDeploy> for DeployClusterSpec {
1092    fn build(self, key: LocationKey, _name_hint: &str) -> DeployCluster {
1093        DeployCluster {
1094            key,
1095            next_port: Rc::new(RefCell::new(0)),
1096            cluster_spec: Rc::new(RefCell::new(Some(
1097                self.0
1098                    .into_iter()
1099                    .map(|(c, h)| CrateOrTrybuild::Crate(c, h))
1100                    .collect(),
1101            ))),
1102            members: Rc::new(RefCell::new(vec![])),
1103            name_hint: None,
1104        }
1105    }
1106}
1107
1108impl<T: Into<TrybuildHost>, I: IntoIterator<Item = T>> ClusterSpec<'_, HydroDeploy> for I {
1109    fn build(self, key: LocationKey, name_hint: &str) -> DeployCluster {
1110        let name_hint = format!("{} (cluster {})", name_hint, key);
1111        DeployCluster {
1112            key,
1113            next_port: Rc::new(RefCell::new(0)),
1114            cluster_spec: Rc::new(RefCell::new(Some(
1115                self.into_iter()
1116                    .enumerate()
1117                    .map(|(idx, b)| {
1118                        let mut b = b.into();
1119                        b.name_hint = Some(name_hint.clone());
1120                        b.cluster_idx = Some(idx);
1121                        CrateOrTrybuild::Trybuild(b)
1122                    })
1123                    .collect(),
1124            ))),
1125            members: Rc::new(RefCell::new(vec![])),
1126            name_hint: Some(name_hint),
1127        }
1128    }
1129}
1130
1131fn create_trybuild_service(
1132    trybuild: TrybuildHost,
1133    dir: &std::path::Path,
1134    target_dir: &std::path::PathBuf,
1135    features: Option<&[String]>,
1136    bin_name: &str,
1137    linking_mode: &LinkingMode,
1138) -> RustCrate {
1139    // For dynamic linking, use the dylib-examples crate; for static, use the base crate
1140    let crate_dir = match linking_mode {
1141        LinkingMode::Dynamic => dir.join("dylib-examples"),
1142        LinkingMode::Static => dir.to_path_buf(),
1143    };
1144
1145    let mut ret = RustCrate::new(&crate_dir, dir)
1146        .target_dir(target_dir)
1147        .example(bin_name)
1148        .no_default_features();
1149
1150    ret = ret.set_is_dylib(matches!(linking_mode, LinkingMode::Dynamic));
1151
1152    if let Some(display_name) = trybuild.display_name {
1153        ret = ret.display_name(display_name);
1154    } else if let Some(name_hint) = trybuild.name_hint {
1155        if let Some(cluster_idx) = trybuild.cluster_idx {
1156            ret = ret.display_name(format!("{} / {}", name_hint, cluster_idx));
1157        } else {
1158            ret = ret.display_name(name_hint);
1159        }
1160    }
1161
1162    if let Some(rustflags) = trybuild.rustflags {
1163        ret = ret.rustflags(rustflags);
1164    }
1165
1166    if let Some(profile) = trybuild.profile {
1167        ret = ret.profile(profile);
1168    }
1169
1170    if let Some(tracing) = trybuild.tracing {
1171        ret = ret.tracing(tracing);
1172    }
1173
1174    ret = ret.features(
1175        vec!["hydro___feature_deploy_integration".to_owned()]
1176            .into_iter()
1177            .chain(
1178                trybuild
1179                    .additional_hydro_features
1180                    .into_iter()
1181                    .map(|runtime_feature| {
1182                        assert!(
1183                            HYDRO_RUNTIME_FEATURES.iter().any(|f| f == &runtime_feature),
1184                            "{runtime_feature} is not a valid Hydro runtime feature"
1185                        );
1186                        format!("hydro___feature_{runtime_feature}")
1187                    }),
1188            )
1189            .chain(trybuild.features),
1190    );
1191
1192    for (key, value) in trybuild.build_envs {
1193        ret = ret.build_env(key, value);
1194    }
1195
1196    for (key, value) in trybuild.env {
1197        ret = ret.env(key, value);
1198    }
1199
1200    ret = ret.build_env("STAGELEFT_TRYBUILD_BUILD_STAGED", "1");
1201    ret = ret.config("build.incremental = false");
1202
1203    if let Some(features) = features {
1204        ret = ret.features(features);
1205    }
1206
1207    ret
1208}
hydro_lang/deploy/deploy_graph.rs

hydro_lang/deploy/
deploy_graph.rs
