hydro_lang/location/

mod.rs

//! Type definitions for distributed locations, which specify where pieces of a Hydro
//! program will be executed.
//!
//! Hydro is a **global**, **distributed** programming model. This means that the data
//! and computation in a Hydro program can be spread across multiple machines, data
//! centers, and even continents. To achieve this, Hydro uses the concept of
//! **locations** to keep track of _where_ data is located and computation is executed.
//!
//! Each live collection type (in [`crate::live_collections`]) has a type parameter `L`
//! which will always be a type that implements the [`Location`] trait (e.g. [`Process`]
//! and [`Cluster`]). To create distributed programs, Hydro provides a variety of APIs
//! to allow live collections to be _moved_ between locations via network send/receive.
//!
//! See [the Hydro docs](https://hydro.run/docs/hydro/reference/locations/) for more information.

use std::fmt::Debug;
use std::future::Future;
#[cfg(feature = "tokio")]
use std::marker::PhantomData;
use std::num::ParseIntError;
#[cfg(feature = "tokio")]
use std::time::Duration;

#[cfg(feature = "tokio")]
use bytes::{Bytes, BytesMut};
use futures::stream::Stream as FuturesStream;
use proc_macro2::Span;
use quote::quote;
#[cfg(feature = "tokio")]
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use slotmap::{Key, new_key_type};
#[cfg(feature = "tokio")]
use stageleft::quote_type;
use stageleft::runtime_support::{FreeVariableWithContextWithProps, QuoteTokens};
use stageleft::{QuotedWithContext, q};
use syn::parse_quote;
#[cfg(feature = "tokio")]
use tokio_util::codec::{Decoder, Encoder, LengthDelimitedCodec};

#[cfg(feature = "tokio")]
use crate::compile::builder::ExternalPortId;
#[cfg(feature = "tokio")]
use crate::compile::ir::DebugInstantiate;
use crate::compile::ir::{
    ClusterMembersState, HydroIrOpMetadata, HydroNode, HydroRoot, HydroSource,
};
use crate::forward_handle::ForwardRef;
#[cfg(stageleft_runtime)]
use crate::forward_handle::{CycleCollection, ForwardHandle};
use crate::live_collections::boundedness::{Bounded, Unbounded};
use crate::live_collections::keyed_stream::KeyedStream;
use crate::live_collections::singleton::Singleton;
#[cfg(feature = "sim")]
#[cfg(stageleft_runtime)]
use crate::live_collections::stream::networking::serialize_bincode;
use crate::live_collections::stream::{
    ExactlyOnce, NoOrder, Ordering, Retries, Stream, TotalOrder,
};
#[cfg(stageleft_runtime)]
use crate::location::dynamic::DynLocation;
use crate::location::dynamic::{ClusterConsistency, LocationId};
#[cfg(feature = "tokio")]
use crate::location::external_process::{
    ExternalBincodeBidi, ExternalBincodeSink, ExternalBytesPort, Many, NotMany,
};
use crate::nondet::NonDet;
#[cfg(feature = "tokio")]
use crate::properties::manual_proof;
#[cfg(feature = "sim")]
use crate::sim::SimSender;
use crate::staging_util::get_this_crate;

pub mod dynamic;

pub mod external_process;
pub use external_process::External;

pub mod process;
pub use process::Process;

pub mod cluster;
pub use cluster::Cluster;

pub mod member_id;
pub use member_id::{MemberId, TaglessMemberId};

pub mod tick;
pub use tick::{Atomic, Tick};

/// An event indicating a change in membership status of a location in a group
/// (e.g. a node in a [`Cluster`] or an external client connection).
#[derive(PartialEq, Eq, Clone, Debug, Hash, Serialize, Deserialize)]
pub enum MembershipEvent {
    /// The member has joined the group and is now active.
    Joined,
    /// The member has left the group and is no longer active.
    Left,
}

/// A hint for configuring the network transport used by an external connection.
///
/// This controls how the underlying TCP listener is set up when binding
/// external client connections via methods like [`Location::bind_single_client`]
/// or [`Location::bidi_external_many_bytes`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum NetworkHint {
    /// Automatically select the network configuration (e.g. an ephemeral port).
    Auto,
    /// Use a TCP port, optionally specifying a fixed port number.
    ///
    /// If `None`, an available port will be chosen automatically.
    /// If `Some(port)`, the given port number will be used.
    TcpPort(Option<u16>),
}

#[track_caller]
pub(crate) fn check_matching_location<'a, L: Location<'a>>(l1: &L, l2: &L) {
    assert_eq!(Location::id(l1), Location::id(l2), "locations do not match");
}

#[stageleft::export(LocationKey)]
new_key_type! {
    /// A unique identifier for a clock tick.
    pub struct LocationKey;
}

impl std::fmt::Display for LocationKey {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "loc{:?}", self.data()) // `"loc1v1"``
    }
}

/// This is used for the ECS membership stream.
/// TODO(mingwei): Make this more robust?
impl std::str::FromStr for LocationKey {
    type Err = Option<ParseIntError>;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        let nvn = s.strip_prefix("loc").ok_or(None)?;
        let (idx, ver) = nvn.split_once('v').ok_or(None)?;
        let idx: u64 = idx.parse()?;
        let ver: u64 = ver.parse()?;
        Ok(slotmap::KeyData::from_ffi((ver << 32) | idx).into())
    }
}

impl LocationKey {
    /// TODO(minwgei): Remove this and avoid magic key for simulator external.
    /// The first location key, used by the simulator as the default external location.
    pub const FIRST: Self = Self(slotmap::KeyData::from_ffi(0x0000000100000001)); // `1v1`

    /// A key for testing with index 1.
    #[cfg(test)]
    pub const TEST_KEY_1: Self = Self(slotmap::KeyData::from_ffi(0x000000FF00000001)); // `1v255`

    /// A key for testing with index 2.
    #[cfg(test)]
    pub const TEST_KEY_2: Self = Self(slotmap::KeyData::from_ffi(0x000000FF00000002)); // `2v255`
}

/// This is used within `q!` code in docker and ECS.
impl<Ctx> FreeVariableWithContextWithProps<Ctx, ()> for LocationKey {
    type O = LocationKey;

    fn to_tokens(self, _ctx: &Ctx) -> (QuoteTokens, ())
    where
        Self: Sized,
    {
        let root = get_this_crate();
        let n = Key::data(&self).as_ffi();
        (
            QuoteTokens {
                prelude: None,
                expr: Some(quote! {
                    #root::location::LocationKey::from(#root::runtime_support::slotmap::KeyData::from_ffi(#n))
                }),
            },
            (),
        )
    }
}

/// A simple enum for the type of a root location.
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Serialize)]
pub enum LocationType {
    /// A process (single node).
    Process,
    /// A cluster (multiple nodes).
    Cluster,
    /// An external client.
    External,
}

/// A top-level location (i.e. a [`Process`] or [`Cluster`]) that is outside a tick / atomic region.
pub trait TopLevel<'a>: Location<'a> {}

#[cfg(feature = "sim")]
#[cfg(stageleft_runtime)]
fn register_serialized_external_input<'a, At, L, T>(
    at: &At,
    from: &External<'_, L>,
    deserialize_fn: syn::Expr,
) -> (
    ExternalPortId,
    Stream<T, At::DropConsistency, Unbounded, TotalOrder, ExactlyOnce>,
)
where
    At: TopLevel<'a> + Sized,
{
    let (port_id, stream, sink) = at.register_serialized_single_client::<_, T, ()>(
        from,
        serialize_bincode::<()>(false),
        deserialize_fn,
    );
    sink.complete(stream.location().source_iter(q!([])));

    (port_id, stream)
}

/// A location where data can be materialized and computation can be executed.
///
/// Hydro is a **global**, **distributed** programming model. This means that the data
/// and computation in a Hydro program can be spread across multiple machines, data
/// centers, and even continents. To achieve this, Hydro uses the concept of
/// **locations** to keep track of _where_ data is located and computation is executed.
///
/// Each live collection type (in [`crate::live_collections`]) has a type parameter `L`
/// which will always be a type that implements the [`Location`] trait (e.g. [`Process`]
/// and [`Cluster`]). To create distributed programs, Hydro provides a variety of APIs
/// to allow live collections to be _moved_ between locations via network send/receive.
///
/// See [the Hydro docs](https://hydro.run/docs/hydro/reference/locations/) for more information.
#[expect(
    private_bounds,
    reason = "only internal Hydro code can define location types"
)]
pub trait Location<'a>: DynLocation {
    /// The root location type for this location.
    ///
    /// For top-level locations like [`Process`] and [`Cluster`], this is `Self`.
    /// For nested locations like [`Tick`], this is the root location that contains it.
    type Root: Location<'a>;

    /// The scope of simulator hook handles (see [`crate::sim_hooks`]) bindable to
    /// unsafe operators at this location: [`OnProcess<P>`] for a [`Process<P>`](Process)
    /// root, [`OnCluster<C>`] for a [`Cluster<C>`](Cluster) root. Operators name this in
    /// their `NonDet` hook payload, so a handle can only bind to a matching location
    /// kind.
    ///
    /// [`OnProcess<P>`]: crate::sim_hooks::OnProcess
    /// [`OnCluster<C>`]: crate::sim_hooks::OnCluster
    type SimHookScope: crate::sim_hooks::BindableHookScope;

    /// Location type with consistency guarantees dropped for the live collection on it.
    type DropConsistency: Location<'a, DropConsistency = Self::DropConsistency, SimHookScope = Self::SimHookScope>;

    /// Returns the root location for this location.
    ///
    /// For top-level locations like [`Process`] and [`Cluster`], this returns `self`.
    /// For nested locations like [`Tick`], this returns the root location that contains it.
    fn root(&self) -> Self::Root;

    /// This location but with consistency guarantees dropped for the live collection
    fn drop_consistency(&self) -> Self::DropConsistency;
    /// Gets the runtime enum variant for the current consistency level, if this is a cluster.
    fn consistency() -> Option<ClusterConsistency>;

    /// Updates the consistency guarantees to match that of the given location.
    fn with_consistency_of<L2: Location<'a, DropConsistency = Self::DropConsistency>>(&self) -> L2 {
        L2::from_drop_consistency(self.drop_consistency())
    }

    #[doc(hidden)]
    fn from_drop_consistency(l2: Self::DropConsistency) -> Self;

    /// Attempts to create a new [`Tick`] clock domain at this location.
    ///
    /// Returns `Some(Tick)` if this is a top-level location (like [`Process`] or [`Cluster`]),
    /// or `None` if this location is already inside a tick (nested ticks are not supported).
    ///
    /// Prefer using [`Location::tick`] when you know the location is top-level.
    fn try_tick(&self) -> Option<Tick<Self>> {
        if Self::is_top_level() {
            let id = if let LocationId::Atomic { .. } = self.id() {
                None
            } else {
                Some(self.flow_state().borrow_mut().next_clock_id())
            };
            Some(Tick {
                id,
                l: self.clone(),
            })
        } else {
            None
        }
    }

    /// Returns the unique identifier for this location.
    fn id(&self) -> LocationId {
        DynLocation::dyn_id(self)
    }

    /// Creates a new [`Tick`] clock domain at this location.
    ///
    /// A tick represents a logical clock that can be used to batch streaming data
    /// into discrete time steps. This is useful for implementing iterative algorithms
    /// or for synchronizing data across multiple streams.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// let inside_tick = process
    ///     .source_iter(q!(vec![1, 2, 3, 4]))
    ///     .batch(&tick, nondet!(/** test */));
    /// inside_tick.all_ticks()
    /// # }, |mut stream| async move {
    /// // 1, 2, 3, 4
    /// # for w in vec![1, 2, 3, 4] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    fn tick(&self) -> Tick<Self> {
        self.try_tick().expect("cannot create nested ticks")
    }

    /// Creates an unbounded stream that continuously emits unit values `()`.
    ///
    /// This is useful for driving computations that need to run continuously,
    /// such as polling or heartbeat mechanisms.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// process.spin()
    ///     .batch(&tick, nondet!(/** test */))
    ///     .map(q!(|_| 42))
    ///     .all_ticks()
    /// # }, |mut stream| async move {
    /// // 42, 42, 42, ...
    /// # assert_eq!(stream.next().await.unwrap(), 42);
    /// # assert_eq!(stream.next().await.unwrap(), 42);
    /// # assert_eq!(stream.next().await.unwrap(), 42);
    /// # }));
    /// # }
    /// ```
    fn spin(&self) -> Stream<(), Self, Unbounded, TotalOrder, ExactlyOnce>
    where
        Self: TopLevel<'a> + Sized,
    {
        Stream::new(
            self.clone(),
            HydroNode::Source {
                source: HydroSource::Spin(),
                metadata: self.new_node_metadata(Stream::<
                    (),
                    Self,
                    Unbounded,
                    TotalOrder,
                    ExactlyOnce,
                >::collection_kind()),
            },
        )
    }

    /// Creates a stream from an async [`FuturesStream`].
    ///
    /// This is useful for integrating with external async data sources,
    /// such as network connections or file readers.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// process.source_stream(q!(futures::stream::iter(vec![1, 2, 3])))
    /// # }, |mut stream| async move {
    /// // 1, 2, 3
    /// # for w in vec![1, 2, 3] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    fn source_stream<T, E>(
        &self,
        e: impl QuotedWithContext<'a, E, Self>,
    ) -> Stream<T, Self::DropConsistency, Unbounded, TotalOrder, ExactlyOnce>
    where
        E: FuturesStream<Item = T> + Unpin,
        Self: TopLevel<'a> + Sized,
    {
        let e = e.splice_untyped_ctx(self);

        let target_location = self.drop_consistency();
        Stream::new(
            target_location.clone(),
            HydroNode::Source {
                source: HydroSource::Stream(e.into()),
                metadata: target_location.new_node_metadata(Stream::<
                    T,
                    Self::DropConsistency,
                    Unbounded,
                    TotalOrder,
                    ExactlyOnce,
                >::collection_kind()),
            },
        )
    }

    /// Creates a bounded stream from an iterator.
    ///
    /// The iterator is evaluated once at runtime, and all elements are emitted
    /// in order. This is useful for creating streams from static data or
    /// for testing.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// process.source_iter(q!(vec![1, 2, 3, 4]))
    /// # }, |mut stream| async move {
    /// // 1, 2, 3, 4
    /// # for w in vec![1, 2, 3, 4] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    fn source_iter<T, E>(
        &self,
        e: impl QuotedWithContext<'a, E, Self>,
    ) -> Stream<T, Self::DropConsistency, Bounded, TotalOrder, ExactlyOnce>
    where
        E: IntoIterator<Item = T>,
        Self: Sized,
    {
        let e = e.splice_typed_ctx(self);

        let target_location = self.drop_consistency();
        Stream::new(
            target_location.clone(),
            HydroNode::Source {
                source: HydroSource::Iter(e.into()),
                metadata: target_location.new_node_metadata(Stream::<
                    T,
                    Self::DropConsistency,
                    Bounded,
                    TotalOrder,
                    ExactlyOnce,
                >::collection_kind()),
            },
        )
    }

    #[deprecated(note = "use .source_cluster_membership_stream(...) instead")]
    /// Creates a stream of membership events for a cluster.
    ///
    /// This stream emits [`MembershipEvent::Joined`] when a cluster member joins
    /// and [`MembershipEvent::Left`] when a cluster member leaves. The stream is
    /// keyed by the [`MemberId`] of the cluster member.
    ///
    /// This is useful for implementing protocols that need to track cluster membership,
    /// such as broadcasting to all members or detecting failures.
    ///
    /// # Non-Determinism
    /// This stream is non-deterministic because the timing of membership events, for example
    /// if a node leaves, the membership event may not be received if the node left before the
    /// stream was created.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::multi_location_test(|flow, p2| {
    /// let p1 = flow.process::<()>();
    /// let workers: Cluster<()> = flow.cluster::<()>();
    /// # // do nothing on each worker
    /// # workers.source_iter(q!(vec![])).for_each(q!(|_: ()| {}));
    /// let cluster_members = p1.source_cluster_members(&workers, nondet!(/** late joiners may miss events */));
    /// # cluster_members.entries().send(&p2, TCP.fail_stop().bincode())
    /// // if there are 4 members in the cluster, we would see a join event for each
    /// // { MemberId::<Worker>(0): [MembershipEvent::Join], MemberId::<Worker>(2): [MembershipEvent::Join], ... }
    /// # }, |mut stream| async move {
    /// # let mut results = Vec::new();
    /// # for w in 0..4 {
    /// #     results.push(format!("{:?}", stream.next().await.unwrap()));
    /// # }
    /// # results.sort();
    /// # assert_eq!(results, vec!["(MemberId::<()>(0), Joined)", "(MemberId::<()>(1), Joined)", "(MemberId::<()>(2), Joined)", "(MemberId::<()>(3), Joined)"]);
    /// # }));
    /// # }
    /// ```
    fn source_cluster_members<C: 'a>(
        &self,
        cluster: &Cluster<'a, C>,
        nondet_start: NonDet,
    ) -> KeyedStream<MemberId<C>, MembershipEvent, Self::DropConsistency, Unbounded>
    where
        Self: TopLevel<'a> + Sized,
    {
        self.source_cluster_membership_stream(cluster, nondet_start)
    }

    /// Creates a stream of membership events for a cluster.
    ///
    /// This stream emits [`MembershipEvent::Joined`] when a cluster member joins
    /// and [`MembershipEvent::Left`] when a cluster member leaves. The stream is
    /// keyed by the [`MemberId`] of the cluster member.
    ///
    /// This is useful for implementing protocols that need to track cluster membership,
    /// such as broadcasting to all members or detecting failures.
    ///
    /// # Non-Determinism
    /// This stream is non-deterministic because the timing of membership events, for example
    /// if a node leaves, the membership event may not be received if the node left before the
    /// stream was created.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::multi_location_test(|flow, p2| {
    /// let p1 = flow.process::<()>();
    /// let workers: Cluster<()> = flow.cluster::<()>();
    /// # // do nothing on each worker
    /// # workers.source_iter(q!(vec![])).for_each(q!(|_: ()| {}));
    /// let cluster_members = p1.source_cluster_membership_stream(&workers, nondet!(/** late joiners may miss events */));
    /// # cluster_members.entries().send(&p2, TCP.fail_stop().bincode())
    /// // if there are 4 members in the cluster, we would see a join event for each
    /// // { MemberId::<Worker>(0): [MembershipEvent::Join], MemberId::<Worker>(2): [MembershipEvent::Join], ... }
    /// # }, |mut stream| async move {
    /// # let mut results = Vec::new();
    /// # for w in 0..4 {
    /// #     results.push(format!("{:?}", stream.next().await.unwrap()));
    /// # }
    /// # results.sort();
    /// # assert_eq!(results, vec!["(MemberId::<()>(0), Joined)", "(MemberId::<()>(1), Joined)", "(MemberId::<()>(2), Joined)", "(MemberId::<()>(3), Joined)"]);
    /// # }));
    /// # }
    /// ```
    fn source_cluster_membership_stream<C: 'a>(
        &self,
        cluster: &Cluster<'a, C>,
        _nondet_start: NonDet,
    ) -> KeyedStream<MemberId<C>, MembershipEvent, Self::DropConsistency, Unbounded>
    where
        Self: TopLevel<'a> + Sized,
    {
        let target_consistency = self.drop_consistency();
        Stream::new(
            target_consistency.clone(),
            HydroNode::Source {
                source: HydroSource::ClusterMembers(cluster.id(), ClusterMembersState::Uninit),
                metadata: target_consistency.new_node_metadata(Stream::<
                    (TaglessMemberId, MembershipEvent),
                    Self,
                    Unbounded,
                    TotalOrder,
                    ExactlyOnce,
                >::collection_kind(
                )),
            },
        )
        .map(q!(|(k, v)| (MemberId::from_tagless(k), v)))
        .into_keyed()
    }

    /// Creates a one-way connection from an external process to receive raw bytes.
    ///
    /// Returns a port handle for the external process to connect to, and a stream
    /// of received byte buffers.
    ///
    /// For bidirectional communication or typed data, see [`Location::bind_single_client`]
    /// or [`Location::source_external_bincode`].
    #[cfg(feature = "tokio")]
    fn source_external_bytes<L>(
        &self,
        from: &External<'_, L>,
    ) -> (
        ExternalBytesPort,
        Stream<BytesMut, Self::DropConsistency, Unbounded, TotalOrder, ExactlyOnce>,
    )
    where
        Self: TopLevel<'a> + Sized,
    {
        let (port, stream, sink) =
            self.bind_single_client::<_, Bytes, LengthDelimitedCodec>(from, NetworkHint::Auto);

        sink.complete(stream.location().source_iter(q!([])));

        (port, stream)
    }

    /// Creates a one-way connection from an external process to receive bincode-serialized data.
    ///
    /// Returns a sink handle for the external process to send data to, and a stream
    /// of received values.
    ///
    /// For bidirectional communication, see [`Location::bind_single_client_bincode`].
    #[cfg(feature = "tokio")]
    fn source_external_bincode<L, T, O: Ordering, R: Retries>(
        &self,
        from: &External<'_, L>,
    ) -> (
        ExternalBincodeSink<T, NotMany, O, R>,
        Stream<T, Self::DropConsistency, Unbounded, O, R>,
    )
    where
        Self: TopLevel<'a> + Sized,
        T: Serialize + DeserializeOwned,
    {
        let (port, stream, sink) = self.bind_single_client_bincode::<_, T, ()>(from);
        sink.complete(stream.location().source_iter(q!([])));

        (
            ExternalBincodeSink {
                process_key: from.key,
                port_id: port.port_id,
                _phantom: PhantomData,
            },
            stream.weaken_ordering().weaken_retries(),
        )
    }

    /// Sets up a MessagePack-encoded simulated input port on this location for testing.
    ///
    /// Returns a handle to send messages to the location as well as a stream
    /// of received messages. Use [`Location::sim_input_with`] to select another codec.
    /// This is only available when the `sim` feature is enabled.
    #[cfg(feature = "sim")]
    fn sim_input<T, O: Ordering, R: Retries>(
        &self,
    ) -> (
        SimSender<T, O, R>,
        Stream<T, Self::DropConsistency, Unbounded, O, R>,
    )
    where
        Self: TopLevel<'a> + Sized,
        T: Serialize + DeserializeOwned,
    {
        self.sim_input_with::<crate::sim::codec::RmpCodec, T, O, R>()
    }

    /// Sets up a simulated input port using the codec `C`.
    ///
    /// Returns a handle to send messages to the location as well as a stream
    /// of received messages. The codec is a type parameter; the message, ordering and
    /// retries types are usually inferred: `location.sim_input_with::<MyCodec, _, _, _>()`.
    /// Custom codecs implement [`SimCodec`](crate::sim::codec::SimCodec), which documents
    /// where they must be defined. This is only available when the `sim` feature is enabled.
    #[cfg(feature = "sim")]
    fn sim_input_with<C: crate::sim::codec::SimCodec<T>, T, O: Ordering, R: Retries>(
        &self,
    ) -> (
        SimSender<T, O, R>,
        Stream<T, Self::DropConsistency, Unbounded, O, R>,
    )
    where
        Self: TopLevel<'a> + Sized,
    {
        let external_location: External<'a, ()> = External {
            key: LocationKey::FIRST,
            flow_state: self.flow_state().clone(),
            _phantom: PhantomData,
        };

        let (external_port_id, stream) = register_serialized_external_input(
            self,
            &external_location,
            crate::sim::codec::staged_deserialize::<T, C>(),
        );

        (
            SimSender(external_port_id, PhantomData, C::encode),
            stream.weaken_ordering().weaken_retries(),
        )
    }

    /// Creates an external input stream for embedded deployment mode.
    ///
    /// The `name` parameter specifies the name of the field in the generated `EmbeddedInputs`
    /// struct that will supply data to this stream at runtime: a
    /// [`EmbeddedInputChannel`](crate::compile::embedded_runtime::EmbeddedInputChannel) around the host's
    /// `Stream<Item = T> + Unpin`. The stream's ordering `O` and retries `R` are what the
    /// dataflow relies on, so they are what the host must promise: declare only what the
    /// dataflow relies on (for example `NoOrder` if it does not use the stream's order), and the
    /// host promises nothing more.
    fn embedded_input<T, O: Ordering, R: Retries>(
        &self,
        name: impl Into<String>,
    ) -> Stream<T, Self::DropConsistency, Unbounded, O, R>
    where
        Self: TopLevel<'a> + Sized,
    {
        let ident = syn::Ident::new(&name.into(), Span::call_site());

        let target_location = self.drop_consistency();
        Stream::new(
            target_location.clone(),
            HydroNode::Source {
                source: HydroSource::Embedded(ident),
                metadata: target_location
                    .new_node_metadata(Stream::<T, Self, Unbounded, O, R>::collection_kind()),
            },
        )
    }

    /// Creates an embedded singleton input for embedded deployment mode.
    ///
    /// The `name` parameter specifies the name of the generated function parameter
    /// that will supply data to this singleton at runtime. The generated function will
    /// accept a plain `T` parameter with this name.
    fn embedded_singleton_input<T>(
        &self,
        name: impl Into<String>,
    ) -> Singleton<T, Self::DropConsistency, Bounded>
    where
        Self: TopLevel<'a> + Sized,
    {
        let ident = syn::Ident::new(&name.into(), Span::call_site());

        let target_location = self.drop_consistency();
        Singleton::new(
            target_location.clone(),
            HydroNode::Source {
                source: HydroSource::EmbeddedSingleton(ident),
                metadata: target_location
                    .new_node_metadata(Singleton::<T, Self, Bounded>::collection_kind()),
            },
        )
    }

    /// Establishes a server on this location to receive a bidirectional connection from a single
    /// client, identified by the given `External` handle. Returns a port handle for the external
    /// process to connect to, a stream of incoming messages, and a handle to send outgoing
    /// messages.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use hydro_deploy::Deployment;
    /// # use futures::{SinkExt, StreamExt};
    /// # tokio_test::block_on(async {
    /// # use bytes::Bytes;
    /// # use hydro_lang::location::NetworkHint;
    /// # use tokio_util::codec::LengthDelimitedCodec;
    /// # let mut flow = FlowBuilder::new();
    /// let node = flow.process::<()>();
    /// let external = flow.external::<()>();
    /// let (port, incoming, outgoing) =
    ///     node.bind_single_client::<_, Bytes, LengthDelimitedCodec>(&external, NetworkHint::Auto);
    /// outgoing.complete(incoming.map(q!(|data /* : Bytes */| {
    ///     let mut resp: Vec<u8> = data.into();
    ///     resp.push(42);
    ///     resp.into() // : Bytes
    /// })));
    ///
    /// # let mut deployment = Deployment::new();
    /// let nodes = flow // ... with_process and with_external
    /// #     .with_process(&node, deployment.Localhost())
    /// #     .with_external(&external, deployment.Localhost())
    /// #     .deploy(&mut deployment);
    ///
    /// deployment.deploy().await.unwrap();
    /// deployment.start().await.unwrap();
    ///
    /// let (mut external_out, mut external_in) = nodes.connect(port).await;
    /// external_in.send(vec![1, 2, 3].into()).await.unwrap();
    /// assert_eq!(
    ///     external_out.next().await.unwrap().unwrap(),
    ///     vec![1, 2, 3, 42]
    /// );
    /// # });
    /// # }
    /// ```
    #[cfg(feature = "tokio")]
    #[expect(clippy::type_complexity, reason = "stream markers")]
    fn bind_single_client<L, T, Codec: Encoder<T> + Decoder>(
        &self,
        from: &External<'_, L>,
        port_hint: NetworkHint,
    ) -> (
        ExternalBytesPort<NotMany>,
        Stream<<Codec as Decoder>::Item, Self::DropConsistency, Unbounded, TotalOrder, ExactlyOnce>,
        ForwardHandle<'a, Stream<T, Self::DropConsistency, Unbounded, TotalOrder, ExactlyOnce>>,
    )
    where
        Self: TopLevel<'a> + Sized,
    {
        let next_external_port_id = from.flow_state.borrow_mut().next_external_port();
        let target_consistency = self.drop_consistency();

        let (fwd_ref, to_sink) = target_consistency.forward_ref::<Stream<
            T,
            Self::DropConsistency,
            Unbounded,
            TotalOrder,
            ExactlyOnce,
        >>();
        let mut flow_state_borrow = self.flow_state().borrow_mut();

        flow_state_borrow.push_root(HydroRoot::SendExternal {
            to_external_key: from.key,
            to_port_id: next_external_port_id,
            to_many: false,
            unpaired: false,
            serialize_fn: None,
            instantiate_fn: DebugInstantiate::Building,
            input: Box::new(to_sink.ir_node.replace(HydroNode::Placeholder)),
            op_metadata: HydroIrOpMetadata::new(),
        });
        drop(flow_state_borrow);

        let raw_stream: Stream<
            Result<<Codec as Decoder>::Item, <Codec as Decoder>::Error>,
            Self::DropConsistency,
            Unbounded,
            TotalOrder,
            ExactlyOnce,
        > = Stream::new(
            target_consistency.clone(),
            HydroNode::ExternalInput {
                from_external_key: from.key,
                from_port_id: next_external_port_id,
                from_many: false,
                codec_type: quote_type::<Codec>().into(),
                port_hint,
                instantiate_fn: DebugInstantiate::Building,
                deserialize_fn: None,
                metadata: target_consistency.new_node_metadata(Stream::<
                    Result<<Codec as Decoder>::Item, <Codec as Decoder>::Error>,
                    Self::DropConsistency,
                    Unbounded,
                    TotalOrder,
                    ExactlyOnce,
                >::collection_kind(
                )),
            },
        );

        (
            ExternalBytesPort {
                process_key: from.key,
                port_id: next_external_port_id,
                _phantom: PhantomData,
            },
            raw_stream.flatten_ordered(),
            fwd_ref,
        )
    }

    // TODO: Replace this staged-expression helper with codec-parameterized sink and bidi handles.
    #[doc(hidden)]
    #[cfg(feature = "tokio")]
    #[expect(clippy::type_complexity, reason = "stream markers")]
    fn register_serialized_single_client<L, InT, OutT>(
        &self,
        from: &External<'_, L>,
        serialize_fn: syn::Expr,
        deserialize_fn: syn::Expr,
    ) -> (
        ExternalPortId,
        Stream<InT, Self::DropConsistency, Unbounded, TotalOrder, ExactlyOnce>,
        ForwardHandle<'a, Stream<OutT, Self::DropConsistency, Unbounded, TotalOrder, ExactlyOnce>>,
    )
    where
        Self: TopLevel<'a> + Sized,
    {
        let next_external_port_id = from.flow_state.borrow_mut().next_external_port();

        let target_consistency = self.drop_consistency();
        let (fwd_ref, to_sink) = target_consistency.forward_ref::<Stream<
            OutT,
            Self::DropConsistency,
            Unbounded,
            TotalOrder,
            ExactlyOnce,
        >>();
        let mut flow_state_borrow = self.flow_state().borrow_mut();

        flow_state_borrow.push_root(HydroRoot::SendExternal {
            to_external_key: from.key,
            to_port_id: next_external_port_id,
            to_many: false,
            unpaired: false,
            serialize_fn: Some(serialize_fn.into()),
            instantiate_fn: DebugInstantiate::Building,
            input: Box::new(to_sink.ir_node.replace(HydroNode::Placeholder)),
            op_metadata: HydroIrOpMetadata::new(),
        });
        drop(flow_state_borrow);

        let raw_stream: Stream<InT, Self::DropConsistency, Unbounded, TotalOrder, ExactlyOnce> =
            Stream::new(
                target_consistency.clone(),
                HydroNode::ExternalInput {
                    from_external_key: from.key,
                    from_port_id: next_external_port_id,
                    from_many: false,
                    codec_type: quote_type::<LengthDelimitedCodec>().into(),
                    port_hint: NetworkHint::Auto,
                    instantiate_fn: DebugInstantiate::Building,
                    deserialize_fn: Some(deserialize_fn.into()),
                    metadata: target_consistency.new_node_metadata(Stream::<
                        InT,
                        Self::DropConsistency,
                        Unbounded,
                        TotalOrder,
                        ExactlyOnce,
                    >::collection_kind(
                    )),
                },
            );

        (next_external_port_id, raw_stream, fwd_ref)
    }

    /// Establishes a bidirectional connection from a single external client using bincode serialization.
    ///
    /// Returns a port handle for the external process to connect to, a stream of incoming messages,
    /// and a handle to send outgoing messages. This is a convenience wrapper around
    /// [`Location::bind_single_client`] that uses bincode for serialization.
    ///
    /// # Type Parameters
    /// - `InT`: The type of incoming messages (must implement [`DeserializeOwned`])
    /// - `OutT`: The type of outgoing messages (must implement [`Serialize`])
    #[cfg(feature = "tokio")]
    #[expect(clippy::type_complexity, reason = "stream markers")]
    fn bind_single_client_bincode<L, InT: DeserializeOwned, OutT: Serialize>(
        &self,
        from: &External<'_, L>,
    ) -> (
        ExternalBincodeBidi<InT, OutT, NotMany>,
        Stream<InT, Self::DropConsistency, Unbounded, TotalOrder, ExactlyOnce>,
        ForwardHandle<'a, Stream<OutT, Self::DropConsistency, Unbounded, TotalOrder, ExactlyOnce>>,
    )
    where
        Self: TopLevel<'a> + Sized,
    {
        let root = get_this_crate();

        let out_t_type = quote_type::<OutT>();
        let ser_fn: syn::Expr = syn::parse_quote! {
            #root::runtime_support::stageleft::runtime_support::fn1_type_hint::<#out_t_type, _>(
                |b| #root::runtime_support::bincode::serialize(&b).unwrap().into()
            )
        };

        let in_t_type = quote_type::<InT>();
        let deser_fn: syn::Expr = syn::parse_quote! {
            |res| {
                let b = res.unwrap();
                #root::runtime_support::bincode::deserialize::<#in_t_type>(&b).unwrap()
            }
        };

        let (port_id, raw_stream, fwd_ref) =
            self.register_serialized_single_client::<_, InT, OutT>(from, ser_fn, deser_fn);

        (
            ExternalBincodeBidi {
                process_key: from.key,
                port_id,
                _phantom: PhantomData,
            },
            raw_stream,
            fwd_ref,
        )
    }

    /// Establishes a server on this location to receive bidirectional connections from multiple
    /// external clients using raw bytes.
    ///
    /// Unlike [`Location::bind_single_client`], this method supports multiple concurrent client
    /// connections. Each client is assigned a unique `u64` identifier.
    ///
    /// Returns:
    /// - A port handle for external processes to connect to
    /// - A keyed stream of incoming messages, keyed by client ID
    /// - A keyed stream of membership events (client joins/leaves), keyed by client ID
    /// - A handle to send outgoing messages, keyed by client ID
    #[cfg(feature = "tokio")]
    #[expect(clippy::type_complexity, reason = "stream markers")]
    fn bidi_external_many_bytes<L, T, Codec: Encoder<T> + Decoder>(
        &self,
        from: &External<'_, L>,
        port_hint: NetworkHint,
    ) -> (
        ExternalBytesPort<Many>,
        KeyedStream<
            u64,
            <Codec as Decoder>::Item,
            Self::DropConsistency,
            Unbounded,
            TotalOrder,
            ExactlyOnce,
        >,
        KeyedStream<
            u64,
            MembershipEvent,
            Self::DropConsistency,
            Unbounded,
            TotalOrder,
            ExactlyOnce,
        >,
        ForwardHandle<
            'a,
            KeyedStream<u64, T, Self::DropConsistency, Unbounded, NoOrder, ExactlyOnce>,
        >,
    )
    where
        Self: TopLevel<'a> + Sized,
    {
        let next_external_port_id = from.flow_state.borrow_mut().next_external_port();

        let target_consistency = self.drop_consistency();
        let (fwd_ref, to_sink) = target_consistency.forward_ref::<KeyedStream<
            u64,
            T,
            Self::DropConsistency,
            Unbounded,
            NoOrder,
            ExactlyOnce,
        >>();
        let to_sink_input = Box::new(to_sink.entries().ir_node.replace(HydroNode::Placeholder));
        let mut flow_state_borrow = self.flow_state().borrow_mut();

        flow_state_borrow.push_root(HydroRoot::SendExternal {
            to_external_key: from.key,
            to_port_id: next_external_port_id,
            to_many: true,
            unpaired: false,
            serialize_fn: None,
            instantiate_fn: DebugInstantiate::Building,
            input: to_sink_input,
            op_metadata: HydroIrOpMetadata::new(),
        });
        drop(flow_state_borrow);

        let raw_stream: Stream<
            Result<(u64, <Codec as Decoder>::Item), <Codec as Decoder>::Error>,
            Self::DropConsistency,
            Unbounded,
            TotalOrder,
            ExactlyOnce,
        > = Stream::new(
            target_consistency.clone(),
            HydroNode::ExternalInput {
                from_external_key: from.key,
                from_port_id: next_external_port_id,
                from_many: true,
                codec_type: quote_type::<Codec>().into(),
                port_hint,
                instantiate_fn: DebugInstantiate::Building,
                deserialize_fn: None,
                metadata: target_consistency.new_node_metadata(Stream::<
                    Result<(u64, <Codec as Decoder>::Item), <Codec as Decoder>::Error>,
                    Self::DropConsistency,
                    Unbounded,
                    TotalOrder,
                    ExactlyOnce,
                >::collection_kind(
                )),
            },
        );

        let membership_stream_ident = syn::Ident::new(
            &format!(
                "__hydro_deploy_many_{}_{}_membership",
                from.key, next_external_port_id
            ),
            Span::call_site(),
        );
        let membership_stream_expr: syn::Expr = parse_quote!(#membership_stream_ident);
        let raw_membership_stream: KeyedStream<
            u64,
            bool,
            Self::DropConsistency,
            Unbounded,
            TotalOrder,
            ExactlyOnce,
        > = KeyedStream::new(
            target_consistency.clone(),
            HydroNode::Source {
                source: HydroSource::Stream(membership_stream_expr.into()),
                metadata: target_consistency.new_node_metadata(KeyedStream::<
                    u64,
                    bool,
                    Self::DropConsistency,
                    Unbounded,
                    TotalOrder,
                    ExactlyOnce,
                >::collection_kind(
                )),
            },
        );

        (
            ExternalBytesPort {
                process_key: from.key,
                port_id: next_external_port_id,
                _phantom: PhantomData,
            },
            raw_stream
                .flatten_ordered() // TODO(shadaj): this silently drops framing errors, decide on right defaults
                .into_keyed(),
            raw_membership_stream.map(q!(|join| {
                if join {
                    MembershipEvent::Joined
                } else {
                    MembershipEvent::Left
                }
            })),
            fwd_ref,
        )
    }

    /// Establishes a server on this location to receive bidirectional connections from multiple
    /// external clients using bincode serialization.
    ///
    /// Unlike [`Location::bind_single_client_bincode`], this method supports multiple concurrent
    /// client connections. Each client is assigned a unique `u64` identifier.
    ///
    /// Returns:
    /// - A port handle for external processes to connect to
    /// - A keyed stream of incoming messages, keyed by client ID
    /// - A keyed stream of membership events (client joins/leaves), keyed by client ID
    /// - A handle to send outgoing messages, keyed by client ID
    ///
    /// # Type Parameters
    /// - `InT`: The type of incoming messages (must implement [`DeserializeOwned`])
    /// - `OutT`: The type of outgoing messages (must implement [`Serialize`])
    #[cfg(feature = "tokio")]
    #[expect(clippy::type_complexity, reason = "stream markers")]
    fn bidi_external_many_bincode<L, InT: DeserializeOwned, OutT: Serialize>(
        &self,
        from: &External<'_, L>,
    ) -> (
        ExternalBincodeBidi<InT, OutT, Many>,
        KeyedStream<u64, InT, Self::DropConsistency, Unbounded, TotalOrder, ExactlyOnce>,
        KeyedStream<
            u64,
            MembershipEvent,
            Self::DropConsistency,
            Unbounded,
            TotalOrder,
            ExactlyOnce,
        >,
        ForwardHandle<
            'a,
            KeyedStream<u64, OutT, Self::DropConsistency, Unbounded, NoOrder, ExactlyOnce>,
        >,
    )
    where
        Self: TopLevel<'a> + Sized,
    {
        let next_external_port_id = from.flow_state.borrow_mut().next_external_port();

        let target_consistency = self.drop_consistency();
        let (fwd_ref, to_sink) = target_consistency.forward_ref::<KeyedStream<
            u64,
            OutT,
            Self::DropConsistency,
            Unbounded,
            NoOrder,
            ExactlyOnce,
        >>();
        let to_sink_input = Box::new(to_sink.entries().ir_node.replace(HydroNode::Placeholder));
        let mut flow_state_borrow = self.flow_state().borrow_mut();

        let root = get_this_crate();

        let out_t_type = quote_type::<OutT>();
        let ser_fn: syn::Expr = syn::parse_quote! {
            #root::runtime_support::stageleft::runtime_support::fn1_type_hint::<(u64, #out_t_type), _>(
                |(id, b)| (id, #root::runtime_support::bincode::serialize(&b).unwrap().into())
            )
        };

        flow_state_borrow.push_root(HydroRoot::SendExternal {
            to_external_key: from.key,
            to_port_id: next_external_port_id,
            to_many: true,
            unpaired: false,
            serialize_fn: Some(ser_fn.into()),
            instantiate_fn: DebugInstantiate::Building,
            input: to_sink_input,
            op_metadata: HydroIrOpMetadata::new(),
        });
        drop(flow_state_borrow);

        let in_t_type = quote_type::<InT>();

        let deser_fn: syn::Expr = syn::parse_quote! {
            |res| {
                let (id, b) = res.unwrap();
                (id, #root::runtime_support::bincode::deserialize::<#in_t_type>(&b).unwrap())
            }
        };

        let raw_stream: KeyedStream<
            u64,
            InT,
            Self::DropConsistency,
            Unbounded,
            TotalOrder,
            ExactlyOnce,
        > = KeyedStream::new(
            target_consistency.clone(),
            HydroNode::ExternalInput {
                from_external_key: from.key,
                from_port_id: next_external_port_id,
                from_many: true,
                codec_type: quote_type::<LengthDelimitedCodec>().into(),
                port_hint: NetworkHint::Auto,
                instantiate_fn: DebugInstantiate::Building,
                deserialize_fn: Some(deser_fn.into()),
                metadata: target_consistency.new_node_metadata(KeyedStream::<
                    u64,
                    InT,
                    Self::DropConsistency,
                    Unbounded,
                    TotalOrder,
                    ExactlyOnce,
                >::collection_kind(
                )),
            },
        );

        let membership_stream_ident = syn::Ident::new(
            &format!(
                "__hydro_deploy_many_{}_{}_membership",
                from.key, next_external_port_id
            ),
            Span::call_site(),
        );
        let membership_stream_expr: syn::Expr = parse_quote!(#membership_stream_ident);
        let raw_membership_stream: KeyedStream<
            u64,
            bool,
            Self::DropConsistency,
            Unbounded,
            TotalOrder,
            ExactlyOnce,
        > = KeyedStream::new(
            target_consistency.clone(),
            HydroNode::Source {
                source: HydroSource::Stream(membership_stream_expr.into()),
                metadata: target_consistency.new_node_metadata(KeyedStream::<
                    u64,
                    bool,
                    Self::DropConsistency,
                    Unbounded,
                    TotalOrder,
                    ExactlyOnce,
                >::collection_kind(
                )),
            },
        );

        (
            ExternalBincodeBidi {
                process_key: from.key,
                port_id: next_external_port_id,
                _phantom: PhantomData,
            },
            raw_stream,
            raw_membership_stream.map(q!(|join| {
                if join {
                    MembershipEvent::Joined
                } else {
                    MembershipEvent::Left
                }
            })),
            fwd_ref,
        )
    }

    /// Bridges user-owned async code to the dataflow as a **bidirectional sidecar**.
    ///
    /// The closure is called once at startup and must return a
    /// `(Stream<InT>, Sink<OutT>)` pair. The framework reads from the stream
    /// (items flowing *into* the dataflow) and writes to the sink (items flowing
    /// *out* to the sidecar). The user controls buffering, backpressure, and
    /// internal lifecycle — Hydro only sees the stream/sink interface.
    ///
    /// This will hopefully make it easy to integrate hydro with existing frameworks,
    /// for example grpc code generated service endpoints.
    ///
    /// # Returns
    /// - A `Stream<InT>` carrying items from the sidecar into the dataflow.
    /// - A [`ForwardHandle`] expecting a `Stream<OutT>` that the user completes
    ///   with items destined for the sidecar.
    ///
    /// # Example
    ///
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// // Sidecar that echoes whatever it receives back into the dataflow.
    /// let (inbound, response_handle) = process.sidecar_bidi::<String, String, _>(q!(|| {
    ///     let (to_df_tx, to_df_rx) = tokio::sync::mpsc::channel::<String>(16);
    ///     let (from_df_tx, mut from_df_rx) = tokio::sync::mpsc::channel::<String>(16);
    ///
    ///     // Spawn the sidecar: echoes items from the dataflow back into it.
    ///     tokio::spawn(async move {
    ///         while let Some(msg) = from_df_rx.recv().await {
    ///             to_df_tx.send(msg).await.ok();
    ///         }
    ///     });
    ///
    ///     // Return the framework-facing ends (concrete types, no boxing needed).
    ///     let stream = tokio_stream::wrappers::ReceiverStream::new(to_df_rx);
    ///     let sink = tokio_util::sync::PollSender::new(from_df_tx);
    ///     (stream, sink)
    /// }));
    ///
    /// // Send "hello" into the sidecar via the response channel.
    /// let input = process.source_stream(q!(futures::stream::iter(vec!["hello".to_string()])));
    /// response_handle.complete(input);
    ///
    /// // The sidecar echoes it back — assert we get "hello" out.
    /// inbound
    /// # }, |mut stream| async move {
    /// #     assert_eq!(stream.next().await.unwrap(), "hello");
    /// # }));
    /// # }
    /// ```
    fn sidecar_bidi<InT: 'static, OutT: 'static, F>(
        &self,
        sidecar: impl QuotedWithContext<'a, F, Self>,
    ) -> (
        Stream<InT, Self, Unbounded, TotalOrder, ExactlyOnce>,
        ForwardHandle<'a, Stream<OutT, Self, Unbounded, NoOrder, ExactlyOnce>>,
    )
    where
        Self: Sized + TopLevel<'a>,
    {
        let location_key = Location::id(self).key();

        let sidecar_id = self.flow_state().borrow_mut().next_sidecar_id();
        let (stream_ident, sink_ident) = sidecar_id.idents();

        let sidecar_closure: syn::Expr = sidecar.splice_untyped_ctx(self);
        self.flow_state()
            .borrow_mut()
            .sidecars
            .push(crate::compile::builder::Sidecar::Bidi {
                location_key,
                sidecar_id,
                sidecar_closure: Box::new(sidecar_closure),
            });

        // Inbound stream: reads from the stream returned by the sidecar closure
        let source_expr: syn::Expr = parse_quote! {
            #stream_ident
        };
        let inbound: Stream<InT, Self, Unbounded, TotalOrder, ExactlyOnce> = Stream::new(
            self.clone(),
            HydroNode::Source {
                source: HydroSource::Stream(source_expr.into()),
                metadata: self.new_node_metadata(Stream::<
                    InT,
                    Self,
                    Unbounded,  // TODO: maybe bounded sidecars are interesting..?
                    TotalOrder, // TODO: NoOrder..?
                    ExactlyOnce,
                >::collection_kind()),
            },
        );

        // Outbound: forward_ref cycle feeding the sink returned by the sidecar closure
        let (fwd_ref, to_sink): (
            ForwardHandle<'a, Stream<OutT, Self, Unbounded, NoOrder, ExactlyOnce>>,
            Stream<OutT, Self, Unbounded, NoOrder, ExactlyOnce>,
        ) = self.forward_ref();

        let sink_expr: syn::Expr = parse_quote! {
            #sink_ident
        };

        let sink_input_ir = to_sink.ir_node.replace(HydroNode::Placeholder);
        self.flow_state()
            .borrow_mut()
            .try_push_root(HydroRoot::DestSink {
                sink: sink_expr.into(),
                input: Box::new(sink_input_ir),
                op_metadata: HydroIrOpMetadata::new(),
            });

        (inbound, fwd_ref)
    }

    /// Constructs a [`Singleton`] materialized at this location with the given static value.
    ///
    /// See also: [`Tick::singleton`], for creating a singleton _within_ a tick, which requires
    /// `T: Clone`.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let singleton = process.singleton(q!(5));
    /// # singleton.into_stream()
    /// # }, |mut stream| async move {
    /// // 5
    /// # assert_eq!(stream.next().await.unwrap(), 5);
    /// # }));
    /// # }
    /// ```
    fn singleton<T>(
        &self,
        e: impl QuotedWithContext<'a, T, Self>,
    ) -> Singleton<T, Self::DropConsistency, Bounded>
    where
        Self: Sized,
    {
        let e = e.splice_untyped_ctx(self);

        let target_location = self.drop_consistency();
        Singleton::new(
            target_location.clone(),
            HydroNode::SingletonSource {
                value: e.into(),
                first_tick_only: false,
                metadata: target_location.new_node_metadata(Singleton::<
                    T,
                    Self::DropConsistency,
                    Bounded,
                >::collection_kind()),
            },
        )
    }

    /// Constructs a [`Singleton`] by resolving an async [`Future`] to completion.
    ///
    /// This is a convenience method equivalent to
    /// `self.singleton(future_expr).resolve_future_blocking()`, which is a common
    /// pattern when initializing a singleton from an async computation.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let singleton = process.singleton_future(q!(async { 42 }));
    /// singleton.into_stream()
    /// # }, |mut stream| async move {
    /// // 42
    /// # assert_eq!(stream.next().await.unwrap(), 42);
    /// # }));
    /// # }
    /// ```
    ///
    /// [`Future`]: std::future::Future
    fn singleton_future<F>(
        &self,
        e: impl QuotedWithContext<'a, F, Self>,
    ) -> Singleton<F::Output, Self::DropConsistency, Bounded>
    where
        F: Future,
        Self: Sized,
    {
        self.singleton(e).resolve_future_blocking()
    }

    /// Generates a stream that emits `()` at a fixed interval.
    ///
    /// The first tick completes immediately. Missed ticks will be scheduled
    /// as soon as possible.
    ///
    /// Because this only emits `()`, the non-determinism of *when* events fire
    /// is captured by the `AtLeastOnce` retry semantics downstream, so no
    /// [`NonDet`] guard is required.
    #[cfg(feature = "tokio")]
    fn source_interval(
        &self,
        interval: impl QuotedWithContext<'a, Duration, Self> + Copy + 'a,
    ) -> Stream<(), Self, Unbounded, TotalOrder, ExactlyOnce>
    where
        Self: TopLevel<'a> + Sized,
    {
        self.source_stream(q!(tokio_stream::StreamExt::map(
            tokio_stream::wrappers::IntervalStream::new(tokio::time::interval(interval)),
            |_| ()
        )))
        .assert_has_consistency_of_trusted(
            manual_proof!(/** interval does not reveal timestamps */),
        )
    }

    /// Generates a stream that emits `()` at a fixed interval, after an
    /// initial delay.
    ///
    /// Because this only emits `()`, the non-determinism of *when* events fire
    /// is captured by the `AtLeastOnce` retry semantics downstream, so no
    /// [`NonDet`] guard is required.
    #[cfg(feature = "tokio")]
    fn source_interval_delayed(
        &self,
        delay: impl QuotedWithContext<'a, Duration, Self> + Copy + 'a,
        interval: impl QuotedWithContext<'a, Duration, Self> + Copy + 'a,
    ) -> Stream<(), Self, Unbounded, TotalOrder, ExactlyOnce>
    where
        Self: TopLevel<'a> + Sized,
    {
        self.source_stream(q!(tokio_stream::StreamExt::map(
            tokio_stream::wrappers::IntervalStream::new(tokio::time::interval_at(
                tokio::time::Instant::now() + delay,
                interval,
            )),
            |_| ()
        )))
        .assert_has_consistency_of_trusted(
            manual_proof!(/** interval does not reveal timestamps */),
        )
    }

    /// Creates a forward reference, allowing a stream to be used before its source is defined.
    ///
    /// Returns a `(handle, placeholder)` pair. Use the placeholder in the dataflow graph,
    /// then call `handle.complete(actual_stream)` to wire in the real source.
    ///
    /// This is useful for mutually-dependent dataflows or when the definition order
    /// doesn't match the data flow direction. For feedback loops, prefer [`Tick::cycle`]
    /// instead, which automatically defers values by one tick.
    ///
    /// # Panics
    /// Panics if the forward reference creates a synchronous cycle (i.e., the completed
    /// stream transitively depends on the placeholder without a `defer_tick` or network
    /// hop in between).
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use hydro_lang::live_collections::stream::NoOrder;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// // Create a forward reference to define a stream that will be completed later
    /// let (complete, forward_stream) = process.forward_ref::<Stream<i32, _, _, NoOrder>>();
    ///
    /// // Use the forward reference as input to another computation
    /// let output: Stream<_, _, _, NoOrder> = forward_stream.map(q!(|x| x * 2));
    ///
    /// // Complete the forward reference with the actual source
    /// let source: Stream<_, _, Unbounded> = process.source_iter(q!([1, 2, 3])).into();
    /// complete.complete(source);
    /// output
    /// # }, |mut stream| async move {
    /// // 2, 4, 6
    /// # assert_eq!(stream.next().await.unwrap(), 2);
    /// # assert_eq!(stream.next().await.unwrap(), 4);
    /// # assert_eq!(stream.next().await.unwrap(), 6);
    /// # }));
    /// # }
    /// ```
    fn forward_ref<S>(&self) -> (ForwardHandle<'a, S>, S)
    where
        S: CycleCollection<'a, ForwardRef, Location = Self>,
    {
        let cycle_id = self.flow_state().borrow_mut().next_cycle_id();
        (
            ForwardHandle::new(cycle_id, Location::id(self)),
            S::create_source(cycle_id, self.clone()),
        )
    }
}

#[cfg(feature = "deploy")]
#[cfg(test)]
mod tests {
    use std::collections::HashSet;

    use futures::{SinkExt, StreamExt};
    use hydro_deploy::Deployment;
    use stageleft::q;
    use tokio_util::codec::LengthDelimitedCodec;

    use crate::compile::builder::FlowBuilder;
    use crate::live_collections::stream::{ExactlyOnce, TotalOrder};
    use crate::location::{Location, NetworkHint};
    use crate::nondet::nondet;

    #[tokio::test]
    async fn top_level_singleton_replay_cardinality() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let external = flow.external::<()>();

        let (in_port, input) =
            node.source_external_bincode::<_, _, TotalOrder, ExactlyOnce>(&external);
        let singleton = node.singleton(q!(123));
        let tick = node.tick();
        let out = input
            .batch(&tick, nondet!(/** test */))
            .cross_singleton(singleton.clone().snapshot(&tick, nondet!(/** test */)))
            .cross_singleton(
                singleton
                    .snapshot(&tick, nondet!(/** test */))
                    .into_stream()
                    .count(),
            )
            .all_ticks()
            .send_bincode_external(&external);

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut external_in = nodes.connect(in_port).await;
        let mut external_out = nodes.connect(out).await;

        deployment.start().await.unwrap();

        external_in.send(1).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), ((1, 123), 1));

        external_in.send(2).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), ((2, 123), 1));
    }

    #[tokio::test]
    async fn tick_singleton_replay_cardinality() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let external = flow.external::<()>();

        let (in_port, input) =
            node.source_external_bincode::<_, _, TotalOrder, ExactlyOnce>(&external);
        let tick = node.tick();
        let singleton = tick.singleton(q!(123));
        let out = input
            .batch(&tick, nondet!(/** test */))
            .cross_singleton(singleton.clone())
            .cross_singleton(singleton.into_stream().count())
            .all_ticks()
            .send_bincode_external(&external);

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut external_in = nodes.connect(in_port).await;
        let mut external_out = nodes.connect(out).await;

        deployment.start().await.unwrap();

        external_in.send(1).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), ((1, 123), 1));

        external_in.send(2).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), ((2, 123), 1));
    }

    #[tokio::test]
    async fn external_bytes() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let first_node = flow.process::<()>();
        let external = flow.external::<()>();

        let (in_port, input) = first_node.source_external_bytes(&external);
        let out = input.send_bincode_external(&external);

        let nodes = flow
            .with_process(&first_node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut external_in = nodes.connect(in_port).await.1;
        let mut external_out = nodes.connect(out).await;

        deployment.start().await.unwrap();

        external_in.send(vec![1, 2, 3].into()).await.unwrap();

        assert_eq!(external_out.next().await.unwrap(), vec![1, 2, 3]);
    }

    #[tokio::test]
    async fn multi_external_source() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let first_node = flow.process::<()>();
        let external = flow.external::<()>();

        let (in_port, input, _membership, complete_sink) =
            first_node.bidi_external_many_bincode(&external);
        let out = input.entries().send_bincode_external(&external);
        complete_sink.complete(
            first_node
                .source_iter::<(u64, ()), _>(q!([]))
                .into_keyed()
                .weaken_ordering(),
        );

        let nodes = flow
            .with_process(&first_node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let (_, mut external_in_1) = nodes.connect_bincode(in_port.clone()).await;
        let (_, mut external_in_2) = nodes.connect_bincode(in_port).await;
        let external_out = nodes.connect(out).await;

        deployment.start().await.unwrap();

        external_in_1.send(123).await.unwrap();
        external_in_2.send(456).await.unwrap();

        assert_eq!(
            external_out.take(2).collect::<HashSet<_>>().await,
            vec![(0, 123), (1, 456)].into_iter().collect()
        );
    }

    #[tokio::test]
    async fn second_connection_only_multi_source() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let first_node = flow.process::<()>();
        let external = flow.external::<()>();

        let (in_port, input, _membership, complete_sink) =
            first_node.bidi_external_many_bincode(&external);
        let out = input.entries().send_bincode_external(&external);
        complete_sink.complete(
            first_node
                .source_iter::<(u64, ()), _>(q!([]))
                .into_keyed()
                .weaken_ordering(),
        );

        let nodes = flow
            .with_process(&first_node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        // intentionally skipped to test stream waking logic
        let (_, mut _external_in_1) = nodes.connect_bincode(in_port.clone()).await;
        let (_, mut external_in_2) = nodes.connect_bincode(in_port).await;
        let mut external_out = nodes.connect(out).await;

        deployment.start().await.unwrap();

        external_in_2.send(456).await.unwrap();

        assert_eq!(external_out.next().await.unwrap(), (1, 456));
    }

    #[tokio::test]
    async fn multi_external_bytes() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let first_node = flow.process::<()>();
        let external = flow.external::<()>();

        let (in_port, input, _membership, complete_sink) = first_node
            .bidi_external_many_bytes::<_, _, LengthDelimitedCodec>(&external, NetworkHint::Auto);
        let out = input.entries().send_bincode_external(&external);
        complete_sink.complete(
            first_node
                .source_iter(q!([]))
                .into_keyed()
                .weaken_ordering(),
        );

        let nodes = flow
            .with_process(&first_node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut external_in_1 = nodes.connect(in_port.clone()).await.1;
        let mut external_in_2 = nodes.connect(in_port).await.1;
        let external_out = nodes.connect(out).await;

        deployment.start().await.unwrap();

        external_in_1.send(vec![1, 2, 3].into()).await.unwrap();
        external_in_2.send(vec![4, 5].into()).await.unwrap();

        assert_eq!(
            external_out.take(2).collect::<HashSet<_>>().await,
            vec![
                (0, (&[1u8, 2, 3] as &[u8]).into()),
                (1, (&[4u8, 5] as &[u8]).into())
            ]
            .into_iter()
            .collect()
        );
    }

    #[tokio::test]
    async fn single_client_external_bytes() {
        let mut deployment = Deployment::new();
        let mut flow = FlowBuilder::new();
        let first_node = flow.process::<()>();
        let external = flow.external::<()>();
        let (port, input, complete_sink) = first_node
            .bind_single_client::<_, _, LengthDelimitedCodec>(&external, NetworkHint::Auto);
        complete_sink.complete(input.map(q!(|data| {
            let mut resp: Vec<u8> = data.into();
            resp.push(42);
            resp.into() // : Bytes
        })));

        let nodes = flow
            .with_process(&first_node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();
        deployment.start().await.unwrap();

        let (mut external_out, mut external_in) = nodes.connect(port).await;

        external_in.send(vec![1, 2, 3].into()).await.unwrap();
        assert_eq!(
            external_out.next().await.unwrap().unwrap(),
            vec![1, 2, 3, 42]
        );
    }

    #[tokio::test]
    async fn echo_external_bytes() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let first_node = flow.process::<()>();
        let external = flow.external::<()>();

        let (port, input, _membership, complete_sink) = first_node
            .bidi_external_many_bytes::<_, _, LengthDelimitedCodec>(&external, NetworkHint::Auto);
        complete_sink
            .complete(input.map(q!(|bytes| { bytes.into_iter().map(|x| x + 1).collect() })));

        let nodes = flow
            .with_process(&first_node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let (mut external_out_1, mut external_in_1) = nodes.connect(port.clone()).await;
        let (mut external_out_2, mut external_in_2) = nodes.connect(port).await;

        deployment.start().await.unwrap();

        external_in_1.send(vec![1, 2, 3].into()).await.unwrap();
        external_in_2.send(vec![4, 5].into()).await.unwrap();

        assert_eq!(external_out_1.next().await.unwrap().unwrap(), vec![2, 3, 4]);
        assert_eq!(external_out_2.next().await.unwrap().unwrap(), vec![5, 6]);
    }

    #[tokio::test]
    async fn echo_external_bincode() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let first_node = flow.process::<()>();
        let external = flow.external::<()>();

        let (port, input, _membership, complete_sink) =
            first_node.bidi_external_many_bincode(&external);
        complete_sink.complete(input.map(q!(|text: String| { text.to_uppercase() })));

        let nodes = flow
            .with_process(&first_node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let (mut external_out_1, mut external_in_1) = nodes.connect_bincode(port.clone()).await;
        let (mut external_out_2, mut external_in_2) = nodes.connect_bincode(port).await;

        deployment.start().await.unwrap();

        external_in_1.send("hi".to_owned()).await.unwrap();
        external_in_2.send("hello".to_owned()).await.unwrap();

        assert_eq!(external_out_1.next().await.unwrap(), "HI");
        assert_eq!(external_out_2.next().await.unwrap(), "HELLO");
    }

    #[tokio::test]
    async fn closure_location_name() {
        let mut deployment = Deployment::new();
        let mut flow = FlowBuilder::new();

        enum ClosureProcess {}

        let node = flow.process::<ClosureProcess>();
        let external = flow.external::<()>();

        let (in_port, input) =
            node.source_external_bincode::<_, i32, TotalOrder, ExactlyOnce>(&external);
        let out = input.send_bincode_external(&external);

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut external_in = nodes.connect(in_port).await;
        let mut external_out = nodes.connect(out).await;

        deployment.start().await.unwrap();

        external_in.send(42).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), 42);
    }
}