hydro_lang/deploy/
deploy_graph_containerized_ecs.rs

1//! Deployment backend for ECS that generates manifests describing the binaries,
2//! ports, and service naming needed to package and orchestrate Hydro applications.
3
4use std::cell::RefCell;
5use std::collections::BTreeMap;
6use std::pin::Pin;
7use std::rc::Rc;
8
9use bytes::Bytes;
10use dfir_lang::graph::{AsCodeOptions, DfirGraph};
11use futures::{Sink, Stream};
12use proc_macro2::Span;
13use serde::{Deserialize, Serialize};
14use stageleft::QuotedWithContext;
15use syn::parse_quote;
16use tracing::{instrument, trace};
17
18/// Manifest for exporting - describes all processes, clusters, and their configuration
19#[derive(Debug, Clone, Serialize, Deserialize)]
20pub struct HydroManifest {
21    /// Process definitions (single-instance services)
22    pub processes: BTreeMap<String, ProcessManifest>,
23    /// Cluster definitions (multi-instance services)
24    pub clusters: BTreeMap<String, ClusterManifest>,
25}
26
27/// Information the build toolchain needs to compile a trybuild binary.
28#[derive(Debug, Clone, Serialize, Deserialize)]
29pub struct BuildConfig {
30    /// Path to the trybuild project directory
31    pub project_dir: String,
32    /// Path to the target directory
33    pub target_dir: String,
34    /// Example/binary name to build
35    pub bin_name: String,
36    /// Package name containing the example (for -p flag)
37    pub package_name: String,
38    /// Features to enable
39    pub features: Vec<String>,
40}
41
42/// Information about an exposed port.
43#[derive(Debug, Clone, Serialize, Deserialize)]
44#[serde(tag = "protocol", rename_all = "lowercase")]
45pub enum PortInfo {
46    /// A TCP listener.
47    Tcp {
48        /// The port number.
49        port: u16,
50    },
51}
52
53/// Manifest entry for a single process
54#[derive(Debug, Clone, Serialize, Deserialize)]
55pub struct ProcessManifest {
56    /// Build toolchain info for this binary
57    pub build: BuildConfig,
58    /// Ports that need to be exposed, keyed by external port identifier
59    pub ports: BTreeMap<String, PortInfo>,
60    /// Task family name (used for ECS service discovery)
61    pub task_family: String,
62}
63
64/// Manifest entry for a cluster (multiple instances of the same service)
65#[derive(Debug, Clone, Serialize, Deserialize)]
66pub struct ClusterManifest {
67    /// Build configuration for this cluster (same binary for all instances)
68    pub build: BuildConfig,
69    /// Ports that need to be exposed, keyed by external port identifier
70    pub ports: BTreeMap<String, PortInfo>,
71    /// Default number of instances
72    pub default_count: usize,
73    /// Task family prefix (instances will be named {prefix}-0, {prefix}-1, etc.)
74    pub task_family_prefix: String,
75}
76
77use super::deploy_runtime_containerized_ecs::*;
78use crate::compile::builder::ExternalPortId;
79use crate::compile::deploy::DeployResult;
80use crate::compile::deploy_provider::{
81    ClusterSpec, Deploy, ExternalSpec, Node, ProcessSpec, RegisterPort,
82};
83use crate::compile::trybuild::generate::create_graph_trybuild;
84use crate::location::dynamic::LocationId;
85use crate::location::member_id::TaglessMemberId;
86use crate::location::{LocationKey, MembershipEvent, NetworkHint};
87
88/// Represents a process running in an ecs deployment
89#[derive(Clone)]
90pub struct EcsDeployProcess {
91    id: LocationKey,
92    name: String,
93    next_port: Rc<RefCell<u16>>,
94
95    exposed_ports: Rc<RefCell<BTreeMap<String, PortInfo>>>,
96
97    trybuild_config:
98        Rc<RefCell<Option<(String, crate::compile::trybuild::generate::TrybuildConfig)>>>,
99}
100
101impl Node for EcsDeployProcess {
102    type Port = u16;
103    type Meta = ();
104    type InstantiateEnv = EcsDeploy;
105
106    #[instrument(level = "trace", skip_all, ret, fields(id = %self.id, name = self.name))]
107    fn next_port(&self) -> Self::Port {
108        let port = {
109            let mut borrow = self.next_port.borrow_mut();
110            let port = *borrow;
111            *borrow += 1;
112            port
113        };
114
115        port
116    }
117
118    #[instrument(level = "trace", skip_all, fields(id = %self.id, name = self.name))]
119    fn update_meta(&self, _meta: &Self::Meta) {}
120
121    #[instrument(level = "trace", skip_all, fields(id = %self.id, name = self.name, ?meta, extra_stmts = extra_stmts.len()))]
122    fn instantiate(
123        &self,
124        _env: &mut Self::InstantiateEnv,
125        meta: &mut Self::Meta,
126        graph: DfirGraph,
127        extra_stmts: &[syn::Stmt],
128        sidecars: &[syn::Expr],
129        as_code_options: &AsCodeOptions,
130    ) {
131        let (bin_name, config) = create_graph_trybuild(
132            graph,
133            extra_stmts,
134            sidecars,
135            as_code_options,
136            Some(&self.name),
137            crate::compile::trybuild::generate::DeployMode::Containerized,
138            crate::compile::trybuild::generate::LinkingMode::Static,
139        );
140
141        // Store the trybuild config for export
142        *self.trybuild_config.borrow_mut() = Some((bin_name, config));
143    }
144}
145
146impl EcsDeployProcess {
147    /// Expose a TCP port on this process for external access.
148    ///
149    /// This method records the port in the manifest's `ports` map so that
150    /// downstream tooling (CDK, deployment scripts) can configure security
151    /// groups, load balancers, and service discovery accordingly.
152    pub fn expose_port(&self, port: u16) {
153        let port_name = format!("exposed-{}", port);
154        self.exposed_ports
155            .borrow_mut()
156            .insert(port_name, PortInfo::Tcp { port });
157    }
158}
159
160/// Represents a logical cluster, which can be a variable amount of individual containers.
161#[derive(Clone)]
162pub struct EcsDeployCluster {
163    id: LocationKey,
164    name: String,
165    next_port: Rc<RefCell<u16>>,
166
167    exposed_ports: Rc<RefCell<BTreeMap<String, PortInfo>>>,
168
169    count: usize,
170
171    /// Stored trybuild config for export
172    trybuild_config:
173        Rc<RefCell<Option<(String, crate::compile::trybuild::generate::TrybuildConfig)>>>,
174}
175
176impl Node for EcsDeployCluster {
177    type Port = u16;
178    type Meta = ();
179    type InstantiateEnv = EcsDeploy;
180
181    #[instrument(level = "trace", skip_all, ret, fields(id = %self.id, name = self.name))]
182    fn next_port(&self) -> Self::Port {
183        let port = {
184            let mut borrow = self.next_port.borrow_mut();
185            let port = *borrow;
186            *borrow += 1;
187            port
188        };
189
190        port
191    }
192
193    #[instrument(level = "trace", skip_all, fields(id = %self.id, name = self.name))]
194    fn update_meta(&self, _meta: &Self::Meta) {}
195
196    #[instrument(level = "trace", skip_all, fields(id = %self.id, name = self.name, extra_stmts = extra_stmts.len()))]
197    fn instantiate(
198        &self,
199        _env: &mut Self::InstantiateEnv,
200        _meta: &mut Self::Meta,
201        graph: DfirGraph,
202        extra_stmts: &[syn::Stmt],
203        sidecars: &[syn::Expr],
204        as_code_options: &AsCodeOptions,
205    ) {
206        let (bin_name, config) = create_graph_trybuild(
207            graph,
208            extra_stmts,
209            sidecars,
210            as_code_options,
211            Some(&self.name),
212            crate::compile::trybuild::generate::DeployMode::Containerized,
213            crate::compile::trybuild::generate::LinkingMode::Static,
214        );
215
216        // Store the trybuild config for export
217        *self.trybuild_config.borrow_mut() = Some((bin_name, config));
218    }
219}
220
221/// Represents an external process, outside the control of this deployment but still with some communication into this deployment.
222#[derive(Clone, Debug)]
223pub struct EcsDeployExternal {
224    name: String,
225    next_port: Rc<RefCell<u16>>,
226}
227
228impl Node for EcsDeployExternal {
229    type Port = u16;
230    type Meta = ();
231    type InstantiateEnv = EcsDeploy;
232
233    #[instrument(level = "trace", skip_all, ret, fields(name = self.name))]
234    fn next_port(&self) -> Self::Port {
235        let port = {
236            let mut borrow = self.next_port.borrow_mut();
237            let port = *borrow;
238            *borrow += 1;
239            port
240        };
241
242        port
243    }
244
245    #[instrument(level = "trace", skip_all, fields(name = self.name))]
246    fn update_meta(&self, _meta: &Self::Meta) {}
247
248    #[instrument(level = "trace", skip_all, fields(name = self.name, ?meta, extra_stmts = extra_stmts.len(), sidecars = sidecars.len()))]
249    fn instantiate(
250        &self,
251        _env: &mut Self::InstantiateEnv,
252        meta: &mut Self::Meta,
253        graph: DfirGraph,
254        extra_stmts: &[syn::Stmt],
255        sidecars: &[syn::Expr],
256        _as_code_options: &AsCodeOptions,
257    ) {
258        trace!(name: "surface", surface = graph.surface_syntax_string());
259    }
260}
261
262impl EcsDeployCluster {
263    /// Expose a TCP port on every member of this cluster for external access.
264    ///
265    /// The binary running on this cluster must bind a `TcpListener` on this port.
266    /// This method records the port in the manifest's `ports` map so that
267    /// downstream tooling (CDK, deployment scripts) can configure security
268    /// groups, load balancers, and service discovery accordingly.
269    pub fn expose_port(&self, port: u16) {
270        let port_name = format!("exposed-{}", port);
271        self.exposed_ports
272            .borrow_mut()
273            .insert(port_name, PortInfo::Tcp { port });
274    }
275}
276
277type DynSourceSink<Out, In, InErr> = (
278    Pin<Box<dyn Stream<Item = Out>>>,
279    Pin<Box<dyn Sink<In, Error = InErr>>>,
280);
281
282impl<'a> RegisterPort<'a, EcsDeploy> for EcsDeployExternal {
283    #[instrument(level = "trace", skip_all, fields(name = self.name, %external_port_id, %port))]
284    fn register(&self, external_port_id: ExternalPortId, port: Self::Port) {}
285
286    #[expect(clippy::manual_async_fn, reason = "matches trait signature")]
287    fn as_bytes_bidi(
288        &self,
289        _external_port_id: ExternalPortId,
290    ) -> impl Future<
291        Output = DynSourceSink<Result<bytes::BytesMut, std::io::Error>, Bytes, std::io::Error>,
292    > + 'a {
293        async { unimplemented!() }
294    }
295
296    #[expect(clippy::manual_async_fn, reason = "matches trait signature")]
297    fn as_bincode_bidi<InT, OutT>(
298        &self,
299        _external_port_id: ExternalPortId,
300    ) -> impl Future<Output = DynSourceSink<OutT, InT, std::io::Error>> + 'a
301    where
302        InT: Serialize + 'static,
303        OutT: serde::de::DeserializeOwned + 'static,
304    {
305        async { unimplemented!() }
306    }
307
308    #[expect(clippy::manual_async_fn, reason = "matches trait signature")]
309    fn as_bincode_sink<T>(
310        &self,
311        _external_port_id: ExternalPortId,
312    ) -> impl Future<Output = Pin<Box<dyn Sink<T, Error = std::io::Error>>>> + 'a
313    where
314        T: Serialize + 'static,
315    {
316        async { unimplemented!() }
317    }
318
319    #[expect(clippy::manual_async_fn, reason = "matches trait signature")]
320    fn as_bincode_source<T>(
321        &self,
322        _external_port_id: ExternalPortId,
323    ) -> impl Future<Output = Pin<Box<dyn Stream<Item = T>>>> + 'a
324    where
325        T: serde::de::DeserializeOwned + 'static,
326    {
327        async { unimplemented!() }
328    }
329}
330
331/// Represents an aws ecs deployment.
332pub struct EcsDeploy;
333
334impl Default for EcsDeploy {
335    fn default() -> Self {
336        Self::new()
337    }
338}
339
340impl EcsDeploy {
341    /// Creates a new ecs deployment.
342    pub fn new() -> Self {
343        Self
344    }
345
346    /// Add an internal ecs process to the deployment.
347    pub fn add_ecs_process(&mut self) -> EcsDeployProcessSpec {
348        EcsDeployProcessSpec
349    }
350
351    /// Add an internal ecs cluster to the deployment.
352    pub fn add_ecs_cluster(&mut self, count: usize) -> EcsDeployClusterSpec {
353        EcsDeployClusterSpec { count }
354    }
355
356    /// Add an external process to the deployment.
357    pub fn add_external(&self, name: String) -> EcsDeployExternalSpec {
358        EcsDeployExternalSpec { name }
359    }
360
361    /// Export a deployment manifest describing each process and cluster: its
362    /// binary name, exposed ports, and ECS task-family naming.
363    ///
364    /// The returned [`HydroManifest`] is typically serialized to JSON and
365    /// consumed by a build script (to compile the trybuild binaries) and a
366    /// deployment tool (CDK, custom scripts, etc.) to create container images
367    /// and orchestrate services.
368    #[instrument(level = "trace", skip_all)]
369    pub fn export(&self, nodes: &DeployResult<'_, Self>) -> HydroManifest {
370        let mut manifest = HydroManifest {
371            processes: BTreeMap::new(),
372            clusters: BTreeMap::new(),
373        };
374
375        // processes
376        for (location_id, name_hint, process) in nodes.get_all_processes() {
377            let LocationId::Process(_) = location_id else {
378                unreachable!()
379            };
380
381            let (bin_name, trybuild_config) = process
382                .trybuild_config
383                .borrow()
384                .clone()
385                .expect("trybuild_config should be set after instantiate");
386
387            manifest.processes.insert(
388                name_hint.to_owned(),
389                ProcessManifest {
390                    build: build_info_from_config(&bin_name, &trybuild_config),
391                    ports: process.exposed_ports.borrow().clone(),
392                    task_family: process.name.clone(),
393                },
394            );
395        }
396
397        // clusters
398        for (location_id, name_hint, cluster) in nodes.get_all_clusters() {
399            let LocationId::Cluster(_) = location_id else {
400                unreachable!()
401            };
402
403            let (bin_name, trybuild_config) = cluster
404                .trybuild_config
405                .borrow()
406                .clone()
407                .expect("trybuild_config should be set after instantiate");
408
409            manifest.clusters.insert(
410                name_hint.to_owned(),
411                ClusterManifest {
412                    build: build_info_from_config(&bin_name, &trybuild_config),
413                    ports: cluster.exposed_ports.borrow().clone(),
414                    default_count: cluster.count,
415                    task_family_prefix: cluster.name.clone(),
416                },
417            );
418        }
419
420        manifest
421    }
422}
423
424fn build_info_from_config(
425    bin_name: &str,
426    config: &crate::compile::trybuild::generate::TrybuildConfig,
427) -> BuildConfig {
428    let mut features = vec!["hydro___feature_ecs_runtime".to_owned()];
429    if let Some(extra) = &config.features {
430        features.extend(extra.clone());
431    }
432    let crate_name = config
433        .project_dir
434        .file_name()
435        .and_then(|n| n.to_str())
436        .unwrap_or("unknown")
437        .replace("_", "-");
438
439    let package_name = format!("{}-hydro-trybuild", crate_name);
440
441    BuildConfig {
442        project_dir: config.project_dir.to_string_lossy().into_owned(),
443        target_dir: config.target_dir.to_string_lossy().into_owned(),
444        bin_name: bin_name.to_owned(),
445        package_name,
446        features,
447    }
448}
449
450impl<'a> Deploy<'a> for EcsDeploy {
451    type InstantiateEnv = Self;
452    type Process = EcsDeployProcess;
453    type Cluster = EcsDeployCluster;
454    type External = EcsDeployExternal;
455    type Meta = ();
456
457    #[instrument(level = "trace", skip_all, fields(p1 = p1.name, %p1_port, p2 = p2.name, %p2_port))]
458    fn o2o_sink_source(
459        _env: &mut Self::InstantiateEnv,
460        p1: &Self::Process,
461        p1_port: &<Self::Process as Node>::Port,
462        p2: &Self::Process,
463        p2_port: &<Self::Process as Node>::Port,
464        name: Option<&str>,
465        networking_info: &crate::networking::NetworkingInfo,
466        _external_types: Option<(&syn::Type, &syn::Type)>,
467    ) -> (syn::Expr, syn::Expr) {
468        match networking_info {
469            crate::networking::NetworkingInfo::Tcp {
470                fault: crate::networking::TcpFault::FailStop,
471            } => {}
472            _ => panic!("Unsupported networking info: {:?}", networking_info),
473        }
474
475        deploy_containerized_o2o(
476            &p2.name,
477            name.expect("channel name is required for containerized deployment"),
478        )
479    }
480
481    #[instrument(level = "trace", skip_all, fields(p1 = p1.name, %p1_port, p2 = p2.name, %p2_port))]
482    fn o2o_connect(
483        p1: &Self::Process,
484        p1_port: &<Self::Process as Node>::Port,
485        p2: &Self::Process,
486        p2_port: &<Self::Process as Node>::Port,
487    ) -> Box<dyn FnOnce()> {
488        let serialized = format!(
489            "o2o_connect {}:{p1_port:?} -> {}:{p2_port:?}",
490            p1.name, p2.name
491        );
492
493        Box::new(move || {
494            trace!(name: "o2o_connect thunk", %serialized);
495        })
496    }
497
498    #[instrument(level = "trace", skip_all, fields(p1 = p1.name, %p1_port, c2 = c2.name, %c2_port))]
499    fn o2m_sink_source(
500        _env: &mut Self::InstantiateEnv,
501        p1: &Self::Process,
502        p1_port: &<Self::Process as Node>::Port,
503        c2: &Self::Cluster,
504        c2_port: &<Self::Cluster as Node>::Port,
505        name: Option<&str>,
506        networking_info: &crate::networking::NetworkingInfo,
507        _external_types: Option<(&syn::Type, &syn::Type)>,
508    ) -> (syn::Expr, syn::Expr) {
509        match networking_info {
510            crate::networking::NetworkingInfo::Tcp {
511                fault: crate::networking::TcpFault::FailStop,
512            } => {}
513            _ => panic!("Unsupported networking info: {:?}", networking_info),
514        }
515
516        deploy_containerized_o2m(
517            name.expect("channel name is required for containerized deployment"),
518        )
519    }
520
521    #[instrument(level = "trace", skip_all, fields(p1 = p1.name, %p1_port, c2 = c2.name, %c2_port))]
522    fn o2m_connect(
523        p1: &Self::Process,
524        p1_port: &<Self::Process as Node>::Port,
525        c2: &Self::Cluster,
526        c2_port: &<Self::Cluster as Node>::Port,
527    ) -> Box<dyn FnOnce()> {
528        let serialized = format!(
529            "o2m_connect {}:{p1_port:?} -> {}:{c2_port:?}",
530            p1.name, c2.name
531        );
532
533        Box::new(move || {
534            trace!(name: "o2m_connect thunk", %serialized);
535        })
536    }
537
538    #[instrument(level = "trace", skip_all, fields(c1 = c1.name, %c1_port, p2 = p2.name, %p2_port))]
539    fn m2o_sink_source(
540        _env: &mut Self::InstantiateEnv,
541        c1: &Self::Cluster,
542        c1_port: &<Self::Cluster as Node>::Port,
543        p2: &Self::Process,
544        p2_port: &<Self::Process as Node>::Port,
545        name: Option<&str>,
546        networking_info: &crate::networking::NetworkingInfo,
547        _external_types: Option<(&syn::Type, &syn::Type)>,
548    ) -> (syn::Expr, syn::Expr) {
549        match networking_info {
550            crate::networking::NetworkingInfo::Tcp {
551                fault: crate::networking::TcpFault::FailStop,
552            } => {}
553            _ => panic!("Unsupported networking info: {:?}", networking_info),
554        }
555
556        deploy_containerized_m2o(
557            &p2.name,
558            name.expect("channel name is required for containerized deployment"),
559        )
560    }
561
562    #[instrument(level = "trace", skip_all, fields(c1 = c1.name, %c1_port, p2 = p2.name, %p2_port))]
563    fn m2o_connect(
564        c1: &Self::Cluster,
565        c1_port: &<Self::Cluster as Node>::Port,
566        p2: &Self::Process,
567        p2_port: &<Self::Process as Node>::Port,
568    ) -> Box<dyn FnOnce()> {
569        let serialized = format!(
570            "o2m_connect {}:{c1_port:?} -> {}:{p2_port:?}",
571            c1.name, p2.name
572        );
573
574        Box::new(move || {
575            trace!(name: "m2o_connect thunk", %serialized);
576        })
577    }
578
579    #[instrument(level = "trace", skip_all, fields(c1 = c1.name, %c1_port, c2 = c2.name, %c2_port))]
580    fn m2m_sink_source(
581        _env: &mut Self::InstantiateEnv,
582        c1: &Self::Cluster,
583        c1_port: &<Self::Cluster as Node>::Port,
584        c2: &Self::Cluster,
585        c2_port: &<Self::Cluster as Node>::Port,
586        name: Option<&str>,
587        networking_info: &crate::networking::NetworkingInfo,
588        _external_types: Option<(&syn::Type, &syn::Type)>,
589    ) -> (syn::Expr, syn::Expr) {
590        match networking_info {
591            crate::networking::NetworkingInfo::Tcp {
592                fault: crate::networking::TcpFault::FailStop,
593            } => {}
594            _ => panic!("Unsupported networking info: {:?}", networking_info),
595        }
596
597        deploy_containerized_m2m(
598            name.expect("channel name is required for containerized deployment"),
599        )
600    }
601
602    #[instrument(level = "trace", skip_all, fields(c1 = c1.name, %c1_port, c2 = c2.name, %c2_port))]
603    fn m2m_connect(
604        c1: &Self::Cluster,
605        c1_port: &<Self::Cluster as Node>::Port,
606        c2: &Self::Cluster,
607        c2_port: &<Self::Cluster as Node>::Port,
608    ) -> Box<dyn FnOnce()> {
609        let serialized = format!(
610            "m2m_connect {}:{c1_port:?} -> {}:{c2_port:?}",
611            c1.name, c2.name
612        );
613
614        Box::new(move || {
615            trace!(name: "m2m_connect thunk", %serialized);
616        })
617    }
618
619    #[instrument(level = "trace", skip_all, fields(p2 = p2.name, %p2_port, %shared_handle, extra_stmts = extra_stmts.len()))]
620    fn e2o_many_source(
621        extra_stmts: &mut Vec<syn::Stmt>,
622        p2: &Self::Process,
623        p2_port: &<Self::Process as Node>::Port,
624        codec_type: &syn::Type,
625        shared_handle: String,
626    ) -> syn::Expr {
627        p2.exposed_ports
628            .borrow_mut()
629            .insert(shared_handle.clone(), PortInfo::Tcp { port: *p2_port });
630
631        let socket_ident = syn::Ident::new(
632            &format!("__hydro_deploy_many_{}_socket", shared_handle),
633            Span::call_site(),
634        );
635
636        let source_ident = syn::Ident::new(
637            &format!("__hydro_deploy_many_{}_source", shared_handle),
638            Span::call_site(),
639        );
640
641        let sink_ident = syn::Ident::new(
642            &format!("__hydro_deploy_many_{}_sink", shared_handle),
643            Span::call_site(),
644        );
645
646        let membership_ident = syn::Ident::new(
647            &format!("__hydro_deploy_many_{}_membership", shared_handle),
648            Span::call_site(),
649        );
650
651        let bind_addr = format!("0.0.0.0:{}", p2_port);
652
653        extra_stmts.push(syn::parse_quote! {
654            let #socket_ident = tokio::net::TcpListener::bind(#bind_addr).await.unwrap();
655        });
656
657        let root = crate::staging_util::get_this_crate();
658
659        extra_stmts.push(syn::parse_quote! {
660            let (#source_ident, #sink_ident, #membership_ident) = #root::runtime_support::hydro_deploy_integration::multi_connection::tcp_multi_connection::<_, #codec_type>(#socket_ident);
661        });
662
663        parse_quote!(#source_ident)
664    }
665
666    #[instrument(level = "trace", skip_all, fields(%shared_handle))]
667    fn e2o_many_sink(shared_handle: String) -> syn::Expr {
668        let sink_ident = syn::Ident::new(
669            &format!("__hydro_deploy_many_{}_sink", shared_handle),
670            Span::call_site(),
671        );
672        parse_quote!(#sink_ident)
673    }
674
675    #[instrument(level = "trace", skip_all, fields(p1 = p1.name, %p1_port, p2 = p2.name, %p2_port, ?codec_type, %shared_handle))]
676    fn e2o_source(
677        extra_stmts: &mut Vec<syn::Stmt>,
678        p1: &Self::External,
679        p1_port: &<Self::External as Node>::Port,
680        p2: &Self::Process,
681        p2_port: &<Self::Process as Node>::Port,
682        codec_type: &syn::Type,
683        shared_handle: String,
684    ) -> syn::Expr {
685        // Record the port for manifest export
686        p2.exposed_ports
687            .borrow_mut()
688            .insert(shared_handle.clone(), PortInfo::Tcp { port: *p2_port });
689
690        let source_ident = syn::Ident::new(
691            &format!("__hydro_deploy_{}_source", shared_handle),
692            Span::call_site(),
693        );
694
695        let bind_addr = format!("0.0.0.0:{}", p2_port);
696
697        // Always use LazySinkSource for external connections - it creates both sink and source
698        // which is needed for bidirectional connections (unpaired: false)
699        let socket_ident = syn::Ident::new(
700            &format!("__hydro_deploy_{}_socket", shared_handle),
701            Span::call_site(),
702        );
703
704        let sink_ident = syn::Ident::new(
705            &format!("__hydro_deploy_{}_sink", shared_handle),
706            Span::call_site(),
707        );
708
709        extra_stmts.push(syn::parse_quote! {
710            let #socket_ident = tokio::net::TcpListener::bind(#bind_addr).await.unwrap();
711        });
712
713        let create_expr = deploy_containerized_external_sink_source_ident(bind_addr, socket_ident);
714
715        extra_stmts.push(syn::parse_quote! {
716            let (#sink_ident, #source_ident) = (#create_expr).split();
717        });
718
719        parse_quote!(#source_ident)
720    }
721
722    #[instrument(level = "trace", skip_all, fields(p1 = p1.name, %p1_port, p2 = p2.name, %p2_port, ?many, ?server_hint))]
723    fn e2o_connect(
724        p1: &Self::External,
725        p1_port: &<Self::External as Node>::Port,
726        p2: &Self::Process,
727        p2_port: &<Self::Process as Node>::Port,
728        many: bool,
729        server_hint: NetworkHint,
730    ) -> Box<dyn FnOnce()> {
731        let serialized = format!(
732            "e2o_connect {}:{p1_port:?} -> {}:{p2_port:?}",
733            p1.name, p2.name
734        );
735
736        Box::new(move || {
737            trace!(name: "e2o_connect thunk", %serialized);
738        })
739    }
740
741    #[instrument(level = "trace", skip_all, fields(p1 = p1.name, %p1_port, p2 = p2.name, %p2_port, %shared_handle))]
742    fn o2e_sink(
743        p1: &Self::Process,
744        p1_port: &<Self::Process as Node>::Port,
745        p2: &Self::External,
746        p2_port: &<Self::External as Node>::Port,
747        shared_handle: String,
748    ) -> syn::Expr {
749        let sink_ident = syn::Ident::new(
750            &format!("__hydro_deploy_{}_sink", shared_handle),
751            Span::call_site(),
752        );
753        parse_quote!(#sink_ident)
754    }
755
756    #[instrument(level = "trace", skip_all, fields(%of_cluster))]
757    fn cluster_ids(
758        of_cluster: LocationKey,
759    ) -> impl QuotedWithContext<'a, &'a [TaglessMemberId], ()> + Clone + 'a {
760        cluster_ids()
761    }
762
763    #[instrument(level = "trace", skip_all)]
764    fn cluster_self_id() -> impl QuotedWithContext<'a, TaglessMemberId, ()> + Clone + 'a {
765        cluster_self_id()
766    }
767
768    #[instrument(level = "trace", skip_all, fields(?location_id))]
769    fn cluster_membership_stream(
770        _env: &mut Self::InstantiateEnv,
771        _at_location: &LocationId,
772        location_id: &LocationId,
773    ) -> impl QuotedWithContext<'a, Box<dyn Stream<Item = (TaglessMemberId, MembershipEvent)> + Unpin>, ()>
774    {
775        cluster_membership_stream(location_id)
776    }
777}
778
779#[instrument(level = "trace", skip_all, ret, fields(%name_hint, %location))]
780fn get_ecs_image_name(name_hint: &str, location: LocationKey) -> String {
781    let name_hint = name_hint
782        .split("::")
783        .last()
784        .unwrap()
785        .to_ascii_lowercase()
786        .replace(".", "-")
787        .replace("_", "-")
788        .replace("::", "-");
789
790    format!("hy-{name_hint}-{location}")
791}
792
793/// Represents a Process running in an ecs deployment
794#[derive(Clone)]
795pub struct EcsDeployProcessSpec;
796
797impl<'a> ProcessSpec<'a, EcsDeploy> for EcsDeployProcessSpec {
798    #[instrument(level = "trace", skip_all, fields(%id, %name_hint))]
799    fn build(self, id: LocationKey, name_hint: &'_ str) -> <EcsDeploy as Deploy<'a>>::Process {
800        EcsDeployProcess {
801            id,
802            name: get_ecs_image_name(name_hint, id),
803            next_port: Rc::new(RefCell::new(10001)),
804            exposed_ports: Rc::new(RefCell::new(BTreeMap::new())),
805            trybuild_config: Rc::new(RefCell::new(None)),
806        }
807    }
808}
809
810/// Represents a Cluster running across `count` ecs tasks.
811#[derive(Clone)]
812pub struct EcsDeployClusterSpec {
813    count: usize,
814}
815
816impl<'a> ClusterSpec<'a, EcsDeploy> for EcsDeployClusterSpec {
817    #[instrument(level = "trace", skip_all, fields(%id, %name_hint))]
818    fn build(self, id: LocationKey, name_hint: &str) -> <EcsDeploy as Deploy<'a>>::Cluster {
819        EcsDeployCluster {
820            id,
821            name: get_ecs_image_name(name_hint, id),
822            next_port: Rc::new(RefCell::new(10001)),
823            exposed_ports: Rc::new(RefCell::new(BTreeMap::new())),
824            count: self.count,
825            trybuild_config: Rc::new(RefCell::new(None)),
826        }
827    }
828}
829
830/// Represents an external process outside of the management of hydro deploy.
831pub struct EcsDeployExternalSpec {
832    name: String,
833}
834
835impl<'a> ExternalSpec<'a, EcsDeploy> for EcsDeployExternalSpec {
836    #[instrument(level = "trace", skip_all, fields(%id, %name_hint))]
837    fn build(self, id: LocationKey, name_hint: &str) -> <EcsDeploy as Deploy<'a>>::External {
838        EcsDeployExternal {
839            name: self.name,
840            next_port: Rc::new(RefCell::new(10000)),
841        }
842    }
843}
hydro_lang/deploy/deploy_graph_containerized_ecs.rs

hydro_lang/deploy/
deploy_graph_containerized_ecs.rs
