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