1use std::future::Future;
2use std::sync::Arc;
3use std::time::Duration;
4
5use bytes::Bytes;
6use futures_util::{stream, StreamExt};
7use n0_future::time::Instant;
8use serde_json::Value;
9use unb_core::{Envelope, ErrorCode, Kind, Resolution, DEFAULT_HOPS};
10use unb_runtime::WireBody;
11
12use crate::handler::HandlerError;
13use crate::layer::Origin;
14use crate::node::{Node, NodeSnapshot};
15
16pub(crate) const CALL_TIMEOUT: Duration = Duration::from_secs(30);
17
18pub trait IntoBody: Send {
19 fn into_body(self) -> Bytes;
20}
21
22impl IntoBody for Bytes {
23 fn into_body(self) -> Bytes {
24 self
25 }
26}
27
28impl IntoBody for Vec<u8> {
29 fn into_body(self) -> Bytes {
30 self.into()
31 }
32}
33
34impl IntoBody for String {
35 fn into_body(self) -> Bytes {
36 self.into()
37 }
38}
39
40impl IntoBody for &str {
41 fn into_body(self) -> Bytes {
42 Bytes::copy_from_slice(self.as_bytes())
43 }
44}
45
46impl IntoBody for Value {
47 fn into_body(self) -> Bytes {
48 Envelope::encode_payload(&self)
49 }
50}
51
52impl IntoBody for () {
53 fn into_body(self) -> Bytes {
54 Bytes::new()
55 }
56}
57
58pub trait Destination: Send + Sync {
59 fn send(
60 &self,
61 request: http::Request<Bytes>,
62 ) -> impl Future<Output = Result<http::Response<Bytes>, HandlerError>> + Send;
63}
64
65impl Destination for Arc<Node> {
66 async fn send(
67 &self,
68 request: http::Request<Bytes>,
69 ) -> Result<http::Response<Bytes>, HandlerError> {
70 let response = self.fetch(request).await?;
71 let (parts, body) = response.into_parts();
72 match body {
73 crate::layer::ServiceBody::Unary(payload) => {
74 Ok(http::Response::from_parts(parts, payload))
75 }
76 crate::layer::ServiceBody::Stream(_) => Err(HandlerError::new(
77 ErrorCode::InvalidInput,
78 "subscribe is not available over send; use Node::subscribe",
79 )),
80 }
81 }
82}
83
84impl<D: Destination + ?Sized> Destination for &D {
85 async fn send(
86 &self,
87 request: http::Request<Bytes>,
88 ) -> Result<http::Response<Bytes>, HandlerError> {
89 D::send(self, request).await
90 }
91}
92
93#[cfg(feature = "hosting")]
94impl Destination for &str {
95 async fn send(
96 &self,
97 request: http::Request<Bytes>,
98 ) -> Result<http::Response<Bytes>, HandlerError> {
99 n0_future::time::timeout(CALL_TIMEOUT, one_shot_http(self, request))
100 .await
101 .map_err(|_| {
102 HandlerError::new(
103 ErrorCode::PeerUnreachable,
104 format!("{self:?} did not answer within the call timeout"),
105 )
106 })?
107 }
108}
109
110#[cfg(feature = "hosting")]
111impl Destination for String {
112 async fn send(
113 &self,
114 request: http::Request<Bytes>,
115 ) -> Result<http::Response<Bytes>, HandlerError> {
116 self.as_str().send(request).await
117 }
118}
119
120pub trait SendExt<T> {
121 fn send<D: Destination>(
122 self,
123 destination: D,
124 ) -> impl Future<Output = Result<http::Response<Bytes>, HandlerError>> + Send;
125}
126
127impl<T: IntoBody> SendExt<T> for http::Request<T> {
128 async fn send<D: Destination>(
129 self,
130 destination: D,
131 ) -> Result<http::Response<Bytes>, HandlerError> {
132 let (mut parts, body) = self.into_parts();
133 if parts.method == http::Method::GET {
134 parts.method = http::Method::POST;
135 }
136 destination
137 .send(http::Request::from_parts(parts, body.into_body()))
138 .await
139 }
140}
141
142impl<T: IntoBody> SendExt<T> for Result<http::Request<T>, http::Error> {
143 async fn send<D: Destination>(
144 self,
145 destination: D,
146 ) -> Result<http::Response<Bytes>, HandlerError> {
147 match self {
148 Ok(request) => request.send(destination).await,
149 Err(error) => Err(HandlerError::new(
150 ErrorCode::InvalidInput,
151 error.to_string(),
152 )),
153 }
154 }
155}
156
157#[cfg(feature = "hosting")]
158async fn one_shot_http(
159 address: &str,
160 request: http::Request<Bytes>,
161) -> Result<http::Response<Bytes>, HandlerError> {
162 let (tls, remainder) = if let Some(rest) = address.strip_prefix("https://") {
163 (true, rest)
164 } else if let Some(rest) = address.strip_prefix("wss://") {
165 (true, rest)
166 } else if let Some(rest) = address.strip_prefix("http://") {
167 (false, rest)
168 } else if let Some(rest) = address.strip_prefix("ws://") {
169 (false, rest)
170 } else {
171 (false, address)
172 };
173 let authority = remainder
174 .split('/')
175 .next()
176 .filter(|authority| !authority.is_empty())
177 .ok_or_else(|| {
178 HandlerError::new(
179 ErrorCode::InvalidInput,
180 format!("{address:?} names no host to send to"),
181 )
182 })?;
183 let has_port = match authority.rfind(']') {
184 Some(bracket) => authority[bracket + 1..].contains(':'),
185 None => authority.contains(':'),
186 };
187 let authority = if has_port {
188 authority.to_string()
189 } else {
190 format!("{authority}:{}", if tls { 443 } else { 80 })
191 };
192 let unreachable = |error: String| HandlerError::new(ErrorCode::PeerUnreachable, error);
193 let stream = tokio::net::TcpStream::connect(&authority)
194 .await
195 .map_err(|error| unreachable(error.to_string()))?;
196 if tls {
197 let host = authority
198 .rsplit_once(':')
199 .map(|(host, _)| host)
200 .unwrap_or(&authority)
201 .trim_start_matches('[')
202 .trim_end_matches(']');
203 let server_name = rustls_pki_types::ServerName::try_from(host.to_string())
204 .map_err(|error| HandlerError::new(ErrorCode::InvalidInput, error.to_string()))?;
205 let config = unb_transport::ws::tls_client_config()
206 .map_err(|error| unreachable(error.to_string()))?;
207 let stream = tokio_rustls::TlsConnector::from(config)
208 .connect(server_name, stream)
209 .await
210 .map_err(|error| unreachable(error.to_string()))?;
211 exchange_http1(stream, &authority, request).await
212 } else {
213 exchange_http1(stream, &authority, request).await
214 }
215}
216
217#[cfg(feature = "hosting")]
218async fn exchange_http1<T>(
219 stream: T,
220 authority: &str,
221 request: http::Request<Bytes>,
222) -> Result<http::Response<Bytes>, HandlerError>
223where
224 T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
225{
226 let target = request
227 .uri()
228 .path_and_query()
229 .map(|target| target.as_str())
230 .filter(|target| !target.is_empty())
231 .unwrap_or("/");
232 let unreachable = |error: String| HandlerError::new(ErrorCode::PeerUnreachable, error);
233 let (mut sender, connection) =
234 hyper::client::conn::http1::handshake(hyper_util::rt::TokioIo::new(stream))
235 .await
236 .map_err(|error| unreachable(error.to_string()))?;
237 tokio::spawn(async move {
238 let _ = connection.await;
239 });
240 let mut outbound = http::Request::builder()
241 .method(http::Method::POST)
242 .uri(target)
243 .header(http::header::HOST, authority);
244 for (name, value) in request.headers() {
245 if matches!(
246 *name,
247 http::header::HOST
248 | http::header::CONTENT_LENGTH
249 | http::header::TRANSFER_ENCODING
250 | http::header::CONNECTION
251 ) {
252 continue;
253 }
254 outbound = outbound.header(name, value);
255 }
256 let outbound = outbound
257 .body(http_body_util::Full::new(request.into_body()))
258 .map_err(|error| HandlerError::new(ErrorCode::InvalidInput, error.to_string()))?;
259 let response = sender
260 .send_request(outbound)
261 .await
262 .map_err(|error| unreachable(error.to_string()))?;
263 let (parts, body) = response.into_parts();
264 let body = http_body_util::Limited::new(body, unb_transport::DEFAULT_MAX_FRAME_SIZE);
265 let body = http_body_util::BodyExt::collect(body)
266 .await
267 .map_err(|error| HandlerError::new(ErrorCode::Protocol, error.to_string()))?
268 .to_bytes();
269 let mut projected = http::Response::builder().status(parts.status);
270 for (name, value) in &parts.headers {
271 if matches!(
272 *name,
273 http::header::CONNECTION
274 | http::header::CONTENT_LENGTH
275 | http::header::TRANSFER_ENCODING
276 | http::header::DATE
277 ) {
278 continue;
279 }
280 projected = projected.header(name, value);
281 }
282 projected
283 .body(body)
284 .map_err(|error| HandlerError::new(ErrorCode::Protocol, error.to_string()))
285}
286
287impl Node {
288 pub async fn fetch_body(
289 self: &Arc<Self>,
290 request: http::Request<WireBody>,
291 ) -> Result<http::Response<WireBody>, HandlerError> {
292 self.fetch_body_until(request, Instant::now() + CALL_TIMEOUT)
293 .await
294 }
295
296 pub(crate) async fn fetch_body_until(
297 self: &Arc<Self>,
298 request: http::Request<WireBody>,
299 deadline: Instant,
300 ) -> Result<http::Response<WireBody>, HandlerError> {
301 if let Some(name) = request
302 .headers()
303 .keys()
304 .find(|name| name.as_str().starts_with("unb-"))
305 {
306 return Err(HandlerError::new(
307 ErrorCode::InvalidInput,
308 format!("{name}: unb-* headers are reserved for framing metadata"),
309 ));
310 }
311 let (parts, body) = request.into_parts();
312 let subject = Envelope::subject_of(&parts.uri);
313 let (snapshot, resolution) = self.resolve_unary_until(&subject, deadline).await?;
314 match resolution {
315 Resolution::Local => {
316 let (payload, streaming_body) = match body {
317 WireBody::Bytes(payload) => (payload, None),
318 WireBody::Stream(body) => (Bytes::new(), Some(body)),
319 };
320 let envelope = Envelope::from_request(http::Request::from_parts(parts, payload))
321 .map_err(|error| {
322 HandlerError::new(ErrorCode::InvalidInput, error.to_string())
323 })?;
324 let mut request = Self::inbound_request(&envelope)?;
325 if let Some(body) = streaming_body {
326 request
327 .extensions_mut()
328 .insert(crate::service::StreamingBody(Arc::new(
329 std::sync::Mutex::new(Some(body)),
330 )));
331 }
332 match self
333 .run_service(snapshot.clone(), request, Origin::Local)
334 .await
335 {
336 Some(outcome) => {
337 let response = outcome?;
338 let (parts, body) = response.into_parts();
339 let body = match body {
340 crate::layer::ServiceBody::Unary(payload) => WireBody::Bytes(payload),
341 crate::layer::ServiceBody::Stream(body) => {
342 WireBody::Stream(Box::pin(body.map(|item| {
343 item.map_err(|error| {
344 unb_core::CoreError::Malformed(error.to_string())
345 })
346 })))
347 }
348 };
349 Ok(http::Response::from_parts(parts, body))
350 }
351 None => Err(Self::teach_unknown_subject(&snapshot, &subject)),
352 }
353 }
354 Resolution::Route(peer_name) => {
355 let link = self.route_link(&peer_name).await?;
356 let remaining = Self::remaining_unary_time(deadline)?;
357 let mut response = link
358 .wire
359 .client_session()
360 .fetch_body(http::Request::from_parts(parts, body), remaining)
361 .await
362 .map_err(Self::client_error)?;
363 let reserved = response
364 .headers()
365 .keys()
366 .filter(|name| {
367 name.as_str().starts_with("unb-") && name.as_str() != unb_core::UNB_CODE
368 })
369 .cloned()
370 .collect::<Vec<_>>();
371 for name in reserved {
372 response.headers_mut().remove(name);
373 }
374 Ok(response)
375 }
376 Resolution::Conflicted { owners } => Err(HandlerError::new(
377 ErrorCode::Conflict,
378 format!(
379 "subject {subject:?} is claimed by multiple live owners: {}",
380 owners.join(", ")
381 ),
382 )),
383 Resolution::Unknown => Err(Self::teach_unknown_subject(&snapshot, &subject)),
384 }
385 }
386
387 pub async fn fetch(
388 self: &Arc<Self>,
389 request: http::Request<bytes::Bytes>,
390 ) -> Result<http::Response<crate::layer::ServiceBody>, HandlerError> {
391 self.fetch_until(request, Instant::now() + CALL_TIMEOUT)
392 .await
393 }
394
395 pub(crate) async fn fetch_until(
396 self: &Arc<Self>,
397 request: http::Request<bytes::Bytes>,
398 deadline: Instant,
399 ) -> Result<http::Response<crate::layer::ServiceBody>, HandlerError> {
400 let (parts, body) = request.into_parts();
401 let response = self
402 .fetch_body_until(
403 http::Request::from_parts(parts, WireBody::Bytes(body)),
404 deadline,
405 )
406 .await?;
407 let (parts, body) = response.into_parts();
408 let body = match body {
409 WireBody::Bytes(payload) => crate::layer::ServiceBody::Unary(payload),
410 WireBody::Stream(body) => {
411 crate::layer::ServiceBody::Stream(Box::pin(body.map(|item| {
412 item.map_err(|error| HandlerError::new(ErrorCode::Protocol, error.to_string()))
413 })))
414 }
415 };
416 Ok(http::Response::from_parts(parts, body))
417 }
418
419 pub async fn subscribe(
420 self: &Arc<Self>,
421 subject: &str,
422 payload: Value,
423 ) -> Result<crate::EventStream, HandlerError> {
424 self.subscribe_with(subject, payload, serde_json::Map::new())
425 .await
426 }
427
428 pub async fn subscribe_with(
429 self: &Arc<Self>,
430 subject: &str,
431 payload: Value,
432 headers: serde_json::Map<String, Value>,
433 ) -> Result<crate::EventStream, HandlerError> {
434 self.subscribe_bytes(subject, Envelope::encode_payload(&payload), headers)
435 .await
436 }
437
438 pub async fn subscribe_bytes(
439 self: &Arc<Self>,
440 subject: &str,
441 payload: Bytes,
442 headers: serde_json::Map<String, Value>,
443 ) -> Result<crate::EventStream, HandlerError> {
444 let subject = &subject.trim_start_matches('/').replace('/', ".");
445 let snapshot = self.snapshot.load_full();
446 match snapshot.node_core.resolve(subject) {
447 Resolution::Local => {
448 let request =
449 Self::local_request(Kind::Subscribe, subject, payload, headers.clone())?;
450 match self
451 .run_service(snapshot.clone(), request, Origin::Local)
452 .await
453 {
454 Some(Ok(response)) => match response.into_body() {
455 crate::layer::ServiceBody::Stream(stream) => Ok(stream),
456 crate::layer::ServiceBody::Unary(_) => Err(HandlerError::new(
457 ErrorCode::Internal,
458 "a streaming operation produced a unary response",
459 )),
460 },
461 Some(Err(error)) => Err(error),
462 None => Err(Self::teach_unknown_subject(&snapshot, subject)),
463 }
464 }
465 Resolution::Route(peer_name) => {
466 let link = self.peer(&peer_name).await.ok_or_else(|| {
467 HandlerError::new(
468 ErrorCode::PeerUnreachable,
469 format!("no live connection to peer {peer_name:?}"),
470 )
471 })?;
472 let stream = link
473 .wire
474 .client_session()
475 .start(
476 subject,
477 Kind::Subscribe,
478 payload,
479 Some(DEFAULT_HOPS),
480 headers,
481 )
482 .await
483 .map_err(|error| {
484 HandlerError::new(ErrorCode::PeerUnreachable, error.to_string())
485 })?;
486 Ok(Box::pin(stream::unfold(stream, |mut stream| async move {
487 let item = match stream.next().await {
488 Ok(Some(envelope)) if envelope.kind == Kind::Event => {
489 Some(Ok(envelope.payload))
490 }
491 Ok(Some(envelope)) if envelope.kind == Kind::Response => None,
492 Ok(Some(_)) => Some(Err(HandlerError::new(
493 ErrorCode::Protocol,
494 "unexpected frame in subscription",
495 ))),
496 Ok(None) => None,
497 Err(error) => Some(Err(Node::client_error(error))),
498 };
499 item.map(|item| (item, stream))
500 })))
501 }
502 Resolution::Conflicted { owners } => Err(HandlerError::new(
503 ErrorCode::Conflict,
504 format!(
505 "subject {subject:?} is claimed by multiple live owners: {}",
506 owners.join(", ")
507 ),
508 )),
509 Resolution::Unknown => Err(Self::teach_unknown_subject(&snapshot, subject)),
510 }
511 }
512
513 pub(crate) async fn call_nested(
514 self: &Arc<Self>,
515 subject: &str,
516 payload: Value,
517 headers: serde_json::Map<String, Value>,
518 ) -> Result<Value, HandlerError> {
519 self.call_with_origin(subject, payload, headers, Origin::Nested)
520 .await
521 }
522
523 async fn call_with_origin(
524 self: &Arc<Self>,
525 subject: &str,
526 payload: Value,
527 headers: serde_json::Map<String, Value>,
528 origin: Origin,
529 ) -> Result<Value, HandlerError> {
530 let subject = &subject.trim_start_matches('/').replace('/', ".");
531 let deadline = Instant::now() + CALL_TIMEOUT;
532 let (snapshot, resolution) = self.resolve_unary_until(subject, deadline).await?;
533 match resolution {
534 Resolution::Local => {
535 let request = Self::local_request(
536 Kind::Request,
537 subject,
538 Envelope::encode_payload(&payload),
539 headers,
540 )?;
541 let outcome = self.run_service(snapshot.clone(), request, origin).await;
542 match outcome {
543 Some(outcome) => match outcome?.into_body() {
544 crate::layer::ServiceBody::Unary(payload) => {
545 Self::json_profile_payload(&payload)
546 }
547 crate::layer::ServiceBody::Stream(_) => Err(HandlerError::new(
548 ErrorCode::Internal,
549 "a unary operation produced a stream",
550 )),
551 },
552 None => Err(Self::teach_unknown_subject(&snapshot, subject)),
553 }
554 }
555 Resolution::Route(peer_name) => {
556 let link = self.route_link(&peer_name).await?;
557 self.call_peer(link, subject, payload, headers, DEFAULT_HOPS, deadline)
558 .await
559 }
560 Resolution::Conflicted { owners } => Err(HandlerError::new(
561 ErrorCode::Conflict,
562 format!(
563 "subject {subject:?} is claimed by multiple live owners: {}",
564 owners.join(", ")
565 ),
566 )),
567 Resolution::Unknown => Err(Self::teach_unknown_subject(&snapshot, subject)),
568 }
569 }
570
571 async fn route_link(&self, peer_name: &str) -> Result<crate::node::PeerLink, HandlerError> {
572 self.peer(peer_name).await.ok_or_else(|| {
573 HandlerError::new(
574 ErrorCode::PeerUnreachable,
575 format!("no live connection to peer {peer_name:?}"),
576 )
577 })
578 }
579
580 async fn call_peer(
581 &self,
582 link: crate::node::PeerLink,
583 subject: &str,
584 payload: Value,
585 headers: serde_json::Map<String, Value>,
586 hops: u8,
587 deadline: Instant,
588 ) -> Result<Value, HandlerError> {
589 let reply = self
590 .call_peer_envelope(
591 link,
592 subject,
593 Envelope::encode_payload(&payload),
594 headers,
595 hops,
596 deadline,
597 )
598 .await?;
599 Self::json_profile_payload(&reply.payload)
600 }
601
602 async fn call_peer_envelope(
603 &self,
604 link: crate::node::PeerLink,
605 subject: &str,
606 payload: bytes::Bytes,
607 headers: serde_json::Map<String, Value>,
608 hops: u8,
609 deadline: Instant,
610 ) -> Result<Envelope, HandlerError> {
611 let remaining = Self::remaining_unary_time(deadline)?;
612 let operation = async move {
613 let mut stream = link
614 .wire
615 .client_session()
616 .start(subject, Kind::Request, payload, Some(hops), headers)
617 .await
618 .map_err(|error| {
619 HandlerError::new(ErrorCode::PeerUnreachable, error.to_string())
620 })?;
621 match stream.next().await {
622 Ok(Some(envelope)) if envelope.kind == Kind::Response => Ok(envelope),
623 Err(error) => Err(Self::client_error(error)),
624 Ok(Some(_)) => Err(HandlerError::new(
625 ErrorCode::Protocol,
626 "downstream call returned an unexpected frame",
627 )),
628 Ok(None) => Err(HandlerError::new(
629 ErrorCode::PeerUnreachable,
630 "downstream call did not complete",
631 )),
632 }
633 };
634 match n0_future::time::timeout(remaining, operation).await {
635 Ok(result) => result,
636 Err(_) => Err(HandlerError::new(
637 ErrorCode::PeerUnreachable,
638 "downstream call did not complete before its deadline",
639 )),
640 }
641 }
642
643 pub(crate) async fn resolve_unary_until(
644 &self,
645 subject: &str,
646 deadline: Instant,
647 ) -> Result<(Arc<NodeSnapshot>, Resolution), HandlerError> {
648 let snapshot = self.snapshot.load_full();
649 let resolution = snapshot.node_core.resolve(subject);
650 if !matches!(resolution, Resolution::Unknown) {
651 return Ok((snapshot, resolution));
652 }
653 let reconnects = self.reconnects_for_subject(subject);
654 if reconnects.is_empty() {
655 return Ok((snapshot, Resolution::Unknown));
656 }
657 let mut waiting = stream::FuturesUnordered::new();
658 for reconnect in reconnects {
659 waiting.push(reconnect.wait());
660 }
661 let mut restored = false;
662 while !waiting.is_empty() {
663 let remaining = Self::remaining_unary_time(deadline)?;
664 let result = n0_future::time::timeout(remaining, waiting.next())
665 .await
666 .map_err(|_| {
667 HandlerError::new(
668 ErrorCode::PeerUnreachable,
669 format!("reconnect did not restore subject {subject:?} before the request deadline"),
670 )
671 })?;
672 if result.is_some_and(|result| result.is_ok()) {
673 restored = true;
674 }
675 let snapshot = self.snapshot.load_full();
676 let resolution = snapshot.node_core.resolve(subject);
677 if !matches!(resolution, Resolution::Unknown) {
678 return Ok((snapshot, resolution));
679 }
680 if waiting.is_empty() {
681 if restored {
682 return Ok((snapshot, Resolution::Unknown));
683 }
684 return Err(HandlerError::new(
685 ErrorCode::PeerUnreachable,
686 format!("reconnect did not restore subject {subject:?}"),
687 ));
688 }
689 }
690 Ok((snapshot, Resolution::Unknown))
691 }
692
693 fn remaining_unary_time(deadline: Instant) -> Result<Duration, HandlerError> {
694 let remaining = deadline.saturating_duration_since(Instant::now());
695 if remaining.is_zero() {
696 Err(HandlerError::new(
697 ErrorCode::PeerUnreachable,
698 "unary request deadline elapsed",
699 ))
700 } else {
701 Ok(remaining)
702 }
703 }
704
705 fn json_profile_payload(payload: &Bytes) -> Result<Value, HandlerError> {
706 if payload.is_empty() {
707 return Ok(Value::Null);
708 }
709 serde_json::from_slice(payload)
710 .map_err(|error| HandlerError::new(ErrorCode::Protocol, error.to_string()))
711 }
712
713 fn client_error(error: unb_runtime::ClientError) -> HandlerError {
714 match error {
715 unb_runtime::ClientError::Protocol { code, message, .. } => {
716 HandlerError::new(code, message)
717 }
718 unb_runtime::ClientError::Cancelled(_) => {
719 HandlerError::new(ErrorCode::Cancelled, error.to_string())
720 }
721 unb_runtime::ClientError::Invalid(message) => {
722 HandlerError::new(ErrorCode::InvalidInput, message)
723 }
724 _ => HandlerError::new(ErrorCode::PeerUnreachable, error.to_string()),
725 }
726 }
727}