feat(hydro_lang): remove NoTick, track consistency in the type system

Author: shadaj

docs/docs/hydro/learn/quickstart/partitioned-counter.mdx

@@ -27,16 +27,16 @@ Before partitioning the counter, you need to make the `keyed_counter_service` fu
 <CodeBlock language="rust" title={
   <Link href="https://github.com/hydro-project/hydro/tree/main/hydro_test/src/tutorials/keyed_counter.rs">src/keyed_counter.rs</Link>
 } showLineNumbers={8}>{getLines(keyedCounterSrc, 8, 14, `
-pub fn keyed_counter_service<'a, L: Location<'a> + NoTick>(
+pub fn keyed_counter_service<'a, L: Location<'a>>(
     increment_requests: KeyedStream<u32, String, L, Unbounded>,
     get_requests: KeyedStream<u32, String, L, Unbounded>,
 ) -> (
     KeyedStream<u32, String, L, Unbounded>,
-    KeyedStream<u32, (String, usize), L, Unbounded, NoOrder>,
+    KeyedStream<u32, (String, usize), L::NoConsistency, Unbounded, NoOrder>,
 ) {
 `)}</CodeBlock>
 
-The `L: Location<'a> + NoTick` bound means the function works with any location that that is outside a `Tick` (like `Process` or `Cluster`). This lets you reuse the same counter logic whether it runs on a single process or several cluster members. The rest of the function body remains unchanged, it already works with any location type.
+The `L: Location<'a>` bound means the function works with any location. This lets you reuse the same counter logic whether it runs on a single process or several cluster members. The rest of the function body remains unchanged, it already works with any location type. You'll notice that we also had to modify the location type parameter for the second output stream (the lookup responses), to be `L::NoConsistency`. In addition to tracking determinism, Hydro also tracks whether _eventual consistency_ is guaranteed across several replicas. Now that we will be scaling up the service, Hydro points out that even if the requests are the same across several replicas, the responses may be different because the service uses an asynchronous snapshot of the counter state. Hence, the responses stream is tagged with `L::NoConsistency`.
 
 ## Introducing Clusters
 

docs/docs/hydro/reference/locations/clusters.md

@@ -124,7 +124,7 @@ on_worker.send(&p2, TCP.fail_stop().bincode())
 # }));
 ```
 
-This API requires a [non-determinism guard](../live-collections/determinism.md#unsafe-operations-in-hydro), because the set of cluster members may asynchronously change over time. Depending on when we are notified of membership changes, we will broadcast to different members. Under the hood, the `broadcast` API uses a list of members of the cluster provided by the deployment system. To manually access this list, you can use the `source_cluster_members` method to get a stream of membership events (cluster members joining or leaving):
+This API requires a [non-determinism guard](../live-collections/determinism.md#unsafe-operations-in-hydro), because the set of cluster members may asynchronously change over time. Depending on when we are notified of membership changes, we will broadcast to different members. Under the hood, the `broadcast` API uses a list of members of the cluster provided by the deployment system. To manually access this list, you can use the `source_cluster_membership_stream` method to get a stream of membership events (cluster members joining or leaving):
 
 ```rust
 # use hydro_lang::prelude::*;
@@ -134,7 +134,7 @@ 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);
+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], ... }

hydro_lang/src/compile/built.rs

@@ -153,6 +153,7 @@ impl<'a> BuiltFlow<'a> {
             cluster_max_sizes: SparseSecondaryMap::new(),
             externals_port_registry: Default::default(),
             test_safety_only: false,
+            skip_consistency_assertions: false,
             _phantom: PhantomData,
         }
     }

hydro_lang/src/compile/deploy.rs

@@ -448,7 +448,7 @@ impl<'a, D: Deploy<'a>> DeployResult<'a, D> {
     }
 
     pub fn get_cluster<C>(&self, c: &Cluster<'a, C>) -> &D::Cluster {
-        let LocationId::Cluster(location_key) = c.id() else {
+        let LocationId::Cluster(location_key, _) = c.id() else {
             panic!("Cluster ID expected")
         };
         self.clusters.get(location_key).unwrap()
@@ -472,9 +472,16 @@ impl<'a, D: Deploy<'a>> DeployResult<'a, D> {
         self.location_names
             .iter()
             .filter_map(|(location_key, location_name)| {
-                self.clusters
-                    .get(location_key)
-                    .map(|cluster| (LocationId::Cluster(location_key), &**location_name, cluster))
+                self.clusters.get(location_key).map(|cluster| {
+                    (
+                        LocationId::Cluster(
+                            location_key,
+                            crate::location::dynamic::ClusterConsistency::EventuallyConsistent,
+                        ),
+                        &**location_name,
+                        cluster,
+                    )
+                })
             })
     }
 

hydro_lang/src/compile/embedded.rs

@@ -410,11 +410,11 @@ impl<'a> Deploy<'a> for EmbeddedDeploy {
     ) -> impl QuotedWithContext<'a, Box<dyn Stream<Item = (TaglessMemberId, MembershipEvent)> + Unpin>, ()>
     {
         let at_key = match at_location {
-            LocationId::Process(key) | LocationId::Cluster(key) => *key,
+            LocationId::Process(key) | LocationId::Cluster(key, _) => *key,
             _ => panic!("cluster_membership_stream must be called from a process or cluster"),
         };
         let cluster_key = match location_id {
-            LocationId::Cluster(key) => *key,
+            LocationId::Cluster(key, _) => *key,
             _ => panic!("cluster_membership_stream target must be a cluster"),
         };
         let vec = env.membership_streams.entry(at_key).unwrap().or_default();

hydro_lang/src/compile/ir/mod.rs

@@ -427,6 +427,13 @@ pub trait DfirBuilder {
         serialize: Option<&DebugExpr>,
         tag_id: usize,
     );
+
+    fn assert_is_consistent(
+        &mut self,
+        location: &LocationId,
+        in_ident: syn::Ident,
+        out_ident: &syn::Ident,
+    );
 }
 
 #[cfg(feature = "build")]
@@ -661,6 +668,22 @@ impl DfirBuilder for SecondaryMap<LocationKey, FlatGraphBuilder> {
             );
         }
     }
+
+    fn assert_is_consistent(
+        &mut self,
+        location: &LocationId,
+        in_ident: syn::Ident,
+        out_ident: &syn::Ident,
+    ) {
+        let builder = self.get_dfir_mut(location);
+        builder.add_dfir(
+            parse_quote! {
+                #out_ident = #in_ident;
+            },
+            None,
+            None,
+        );
+    }
 }
 
 #[cfg(feature = "build")]
@@ -721,7 +744,7 @@ impl HydroRoot {
         &mut self,
         extra_stmts: &mut SparseSecondaryMap<LocationKey, Vec<syn::Stmt>>,
         seen_tees: &mut SeenSharedNodes,
-        seen_cluster_members: &mut HashSet<(LocationId, LocationId)>,
+        seen_cluster_members: &mut HashSet<(LocationId, LocationKey)>,
         processes: &SparseSecondaryMap<LocationKey, D::Process>,
         clusters: &SparseSecondaryMap<LocationKey, D::Cluster>,
         externals: &SparseSecondaryMap<LocationKey, D::External>,
@@ -815,7 +838,7 @@ impl HydroRoot {
                                         )
                                     }
                                 }
-                                LocationId::Cluster(cluster_key) => {
+                                LocationId::Cluster(cluster_key, _) => {
                                     let from_node = clusters
                                         .get(*cluster_key)
                                         .unwrap_or_else(|| {
@@ -863,7 +886,7 @@ impl HydroRoot {
                         _ => panic!("Embedded output must have Stream collection kind"),
                     };
                     let location_key = match input.metadata().location_id.root() {
-                        LocationId::Process(key) | LocationId::Cluster(key) => *key,
+                        LocationId::Process(key) | LocationId::Cluster(key, _) => *key,
                         _ => panic!("Embedded output must be on a process or cluster"),
                     };
                     D::register_embedded_output(
@@ -964,7 +987,7 @@ impl HydroRoot {
                                         D::e2o_connect(&from_node, &sink_port, &to_node, &source_port, *from_many, *port_hint),
                                     )
                                 }
-                                LocationId::Cluster(cluster_key) => {
+                                LocationId::Cluster(cluster_key, _) => {
                                     let to_node = clusters
                                         .get(*cluster_key)
                                         .unwrap_or_else(|| {
@@ -1010,7 +1033,7 @@ impl HydroRoot {
                         _ => panic!("Embedded source must have Stream collection kind"),
                     };
                     let location_key = match metadata.location_id.root() {
-                        LocationId::Process(key) | LocationId::Cluster(key) => *key,
+                        LocationId::Process(key) | LocationId::Cluster(key, _) => *key,
                         _ => panic!("Embedded source must be on a process or cluster"),
                     };
                     D::register_embedded_stream_input(
@@ -1025,7 +1048,7 @@ impl HydroRoot {
                         _ => panic!("EmbeddedSingleton source must have Singleton collection kind"),
                     };
                     let location_key = match metadata.location_id.root() {
-                        LocationId::Process(key) | LocationId::Cluster(key) => *key,
+                        LocationId::Process(key) | LocationId::Cluster(key, _) => *key,
                         _ => panic!("EmbeddedSingleton source must be on a process or cluster"),
                     };
                     D::register_embedded_singleton_input(
@@ -1038,7 +1061,7 @@ impl HydroRoot {
                     match state {
                         ClusterMembersState::Uninit => {
                             let at_location = metadata.location_id.root().clone();
-                            let key = (at_location.clone(), LocationId::Cluster(location_id.key()));
+                            let key = (at_location.clone(), location_id.key());
                             if refcell_seen_cluster_members.borrow_mut().insert(key) {
                                 // First occurrence: call cluster_membership_stream and mark as Stream.
                                 let expr = stageleft::QuotedWithContext::splice_untyped_ctx(
@@ -1482,7 +1505,7 @@ fn remap_location(loc: &mut LocationId, uf: &mut HashMap<ClockId, ClockId>) {
         LocationId::Atomic(inner) => {
             remap_location(inner, uf);
         }
-        LocationId::Process(_) | LocationId::Cluster(_) => {}
+        LocationId::Process(_) | LocationId::Cluster(_, _) => {}
     }
 }
 
@@ -2086,6 +2109,11 @@ pub enum HydroNode {
         input: Box<HydroNode>,
         metadata: HydroIrMetadata,
     },
+
+    AssertIsConsistent {
+        inner: Box<HydroNode>,
+        metadata: HydroIrMetadata,
+    },
 }
 
 pub type SeenSharedNodes = HashMap<*const RefCell<HydroNode>, Rc<RefCell<HydroNode>>>;
@@ -2174,7 +2202,8 @@ impl HydroNode {
             | HydroNode::BeginAtomic { inner, .. }
             | HydroNode::EndAtomic { inner, .. }
             | HydroNode::Batch { inner, .. }
-            | HydroNode::YieldConcat { inner, .. } => {
+            | HydroNode::YieldConcat { inner, .. }
+            | HydroNode::AssertIsConsistent { inner, .. } => {
                 transform(inner.as_mut(), seen_tees);
             }
 
@@ -2250,6 +2279,10 @@ impl HydroNode {
                 trusted: *trusted,
                 metadata: metadata.clone(),
             },
+            HydroNode::AssertIsConsistent { inner, metadata } => HydroNode::AssertIsConsistent {
+                inner: Box::new(inner.deep_clone(seen_tees)),
+                metadata: metadata.clone(),
+            },
             HydroNode::Source { source, metadata } => HydroNode::Source {
                 source: source.clone(),
                 metadata: metadata.clone(),
@@ -2606,6 +2639,30 @@ impl HydroNode {
                         // input_ident stays on stack as output
                     }
 
+                    HydroNode::AssertIsConsistent { inner, .. } => {
+                        let inner_ident = ident_stack.pop().unwrap();
+
+                        let out_ident =
+                            syn::Ident::new(&format!("stream_{}", *next_stmt_id), Span::call_site());
+
+                        match builders_or_callback {
+                            BuildersOrCallback::Builders(graph_builders) => {
+                                graph_builders.assert_is_consistent(
+                                    &inner.metadata().location_id,
+                                    inner_ident,
+                                    &out_ident,
+                                );
+                            }
+                            BuildersOrCallback::Callback(_, node_callback) => {
+                                node_callback(node, next_stmt_id);
+                            }
+                        }
+
+                        *next_stmt_id += 1;
+
+                        ident_stack.push(out_ident);
+                    }
+
                     HydroNode::ObserveNonDet {
                         inner,
                         trusted,
@@ -4099,7 +4156,9 @@ impl HydroNode {
             HydroNode::Placeholder => {
                 panic!()
             }
-            HydroNode::Cast { .. } | HydroNode::ObserveNonDet { .. } => {}
+            HydroNode::Cast { .. }
+            | HydroNode::ObserveNonDet { .. }
+            | HydroNode::AssertIsConsistent { .. } => {}
             HydroNode::Source { source, .. } => match source {
                 HydroSource::Stream(expr) | HydroSource::Iter(expr) => transform(expr),
                 HydroSource::ExternalNetwork()
@@ -4185,6 +4244,7 @@ impl HydroNode {
             }
             HydroNode::Cast { metadata, .. } => metadata,
             HydroNode::ObserveNonDet { metadata, .. } => metadata,
+            HydroNode::AssertIsConsistent { metadata, .. } => metadata,
             HydroNode::Source { metadata, .. } => metadata,
             HydroNode::SingletonSource { metadata, .. } => metadata,
             HydroNode::CycleSource { metadata, .. } => metadata,
@@ -4239,6 +4299,7 @@ impl HydroNode {
             }
             HydroNode::Cast { metadata, .. } => metadata,
             HydroNode::ObserveNonDet { metadata, .. } => metadata,
+            HydroNode::AssertIsConsistent { metadata, .. } => metadata,
             HydroNode::Source { metadata, .. } => metadata,
             HydroNode::SingletonSource { metadata, .. } => metadata,
             HydroNode::CycleSource { metadata, .. } => metadata,
@@ -4301,7 +4362,8 @@ impl HydroNode {
             | HydroNode::YieldConcat { inner, .. }
             | HydroNode::BeginAtomic { inner, .. }
             | HydroNode::EndAtomic { inner, .. }
-            | HydroNode::Batch { inner, .. } => {
+            | HydroNode::Batch { inner, .. }
+            | HydroNode::AssertIsConsistent { inner, .. } => {
                 vec![inner]
             }
             HydroNode::Chain { first, second, .. } => {
@@ -4376,6 +4438,7 @@ impl HydroNode {
             }
             HydroNode::Cast { .. } => "Cast()".to_owned(),
             HydroNode::ObserveNonDet { .. } => "ObserveNonDet()".to_owned(),
+            HydroNode::AssertIsConsistent { .. } => "AssertIsConsistent()".to_owned(),
             HydroNode::Source { source, .. } => format!("Source({:?})", source),
             HydroNode::SingletonSource {
                 value,
@@ -4505,7 +4568,7 @@ where
                 D::o2o_connect(&from_node, &sink_port, &to_node, &source_port),
             )
         }
-        (&LocationId::Process(from), &LocationId::Cluster(to)) => {
+        (&LocationId::Process(from), &LocationId::Cluster(to, _)) => {
             let from_node = processes
                 .get(from)
                 .unwrap_or_else(|| {
@@ -4535,7 +4598,7 @@ where
                 D::o2m_connect(&from_node, &sink_port, &to_node, &source_port),
             )
         }
-        (&LocationId::Cluster(from), &LocationId::Process(to)) => {
+        (&LocationId::Cluster(from, _), &LocationId::Process(to)) => {
             let from_node = clusters
                 .get(from)
                 .unwrap_or_else(|| {
@@ -4565,7 +4628,7 @@ where
                 D::m2o_connect(&from_node, &sink_port, &to_node, &source_port),
             )
         }
-        (&LocationId::Cluster(from), &LocationId::Cluster(to)) => {
+        (&LocationId::Cluster(from, _), &LocationId::Cluster(to, _)) => {
             let from_node = clusters
                 .get(from)
                 .unwrap_or_else(|| {

hydro_lang/src/deploy/deploy_graph_containerized_ecs.rs

@@ -376,7 +376,7 @@ impl EcsDeploy {
         }
 
         for (location_id, name_hint, cluster) in nodes.get_all_clusters() {
-            let LocationId::Cluster(raw_id) = location_id else {
+            let LocationId::Cluster(raw_id, _) = location_id else {
                 unreachable!()
             };
             let task_family_prefix = cluster.name.clone();

hydro_lang/src/deploy/maelstrom/deploy_runtime_maelstrom.rs

@@ -21,7 +21,7 @@ use crate::live_collections::keyed_stream::KeyedStream;
 use crate::live_collections::stream::{ExactlyOnce, NoOrder, TotalOrder};
 use crate::location::dynamic::LocationId;
 use crate::location::member_id::TaglessMemberId;
-use crate::location::{Cluster, LocationKey, MembershipEvent, NoTick};
+use crate::location::{Cluster, LocationKey, MembershipEvent};
 use crate::nondet::nondet;
 
 /// Maelstrom message envelope structure.

hydro_lang/src/deploy/maelstrom/mod.rs

@@ -6,7 +6,7 @@ use serde::de::DeserializeOwned;
 use crate::forward_handle::ForwardHandle;
 use crate::live_collections::KeyedStream;
 use crate::live_collections::stream::TotalOrder;
-use crate::location::{Cluster, NoTick};
+use crate::location::Cluster;
 use crate::nondet::nondet;
 
 #[cfg(stageleft_runtime)]
@@ -39,10 +39,7 @@ pub fn maelstrom_bidi_clients<'a, C, In: DeserializeOwned, Out: Serialize>(
 ) -> (
     KeyedStream<String, In, Cluster<'a, C>>,
     ForwardHandle<'a, KeyedStream<String, Out, Cluster<'a, C>>>,
-)
-where
-    Cluster<'a, C>: NoTick,
-{
+) {
     use stageleft::q;
 
     use crate::location::Location;

hydro_lang/src/forward_handle.rs

@@ -50,6 +50,8 @@ where
 {
     type Location: Location<'a>;
 
+    fn location(&self) -> &Self::Location;
+
     fn create_source_with_initial(
         cycle_id: CycleId,
         initial: Self,