refactor(dfir_rs): use slotmap SecondaryMap for metrics, remove custom slot_vec tags

Replace custom `SecondarySlotVec<SubgraphTag, _>` and
`SecondarySlotVec<HandoffTag, _>` in DfirMetrics with
`slotmap::SecondaryMap<GraphSubgraphId, _>` and
`SecondaryMap<GraphNodeId, _>` from dfir_lang.

This eliminates the `slotmap_raw_idx` conversion hack in codegen —
keys are now reconstructed via `slotmap::KeyData::from_ffi()` which
preserves the full key (including version), not just the raw index.

Make `dfir_lang` a non-optional dependency of `dfir_rs` (it was already
always compiled via `dfir_macro`). Re-export `dfir_lang` and `slotmap`
from `dfir_rs` for use in generated code.

Remove `HandoffTag` and `HandoffId` from `scheduled/mod.rs` (no longer
needed). Update `hydro_lang/src/telemetry/emf.rs` to use `Debug`
formatting for slotmap keys instead of `Display`.

Co-authored-by: Infinity 🤖 <infinity@hydro.run>

Author: MingweiSamuel

dfir_lang/src/graph/meta_graph.rs

@@ -822,12 +822,10 @@ impl DfirGraph {
         prefix: TokenStream,
         diagnostics: &mut Diagnostics,
     ) -> Result<TokenStream, Diagnostics> {
-        // Extract the slot index from a slotmap key for use as a runtime metrics key.
-        // Uses the low 32 bits of `KeyData::as_ffi()` (the idx, ignoring the version).
-        // TODO(cleanup): When scheduled Dfir is removed, DfirMetrics could use slotmap
-        // SecondaryMaps directly, eliminating this conversion.
-        fn slotmap_raw_idx(key: impl Key) -> usize {
-            (key.data().as_ffi() & 0xFFFF_FFFF) as usize
+        // Convert a slotmap key to its FFI representation for use in generated code.
+        // The FFI value can reconstruct the key at runtime via `KeyData::from_ffi`.
+        fn slotmap_key_ffi(key: impl Key) -> u64 {
+            key.data().as_ffi()
         }
 
         let df = Ident::new(GRAPH, Span::call_site());
@@ -974,7 +972,7 @@ impl DfirGraph {
         let mut subgraph_blocks = Vec::new();
         {
             for &(subgraph_id, subgraph_nodes) in all_subgraphs.iter() {
-                let sg_metrics_idx = slotmap_raw_idx(subgraph_id);
+                let sg_metrics_ffi = slotmap_key_ffi(subgraph_id);
                 let (recv_hoffs, send_hoffs) = &subgraph_handoffs[subgraph_id];
 
                 // Generate buffer ident helpers for this subgraph's handoffs.
@@ -1011,7 +1009,7 @@ impl DfirGraph {
                     .zip(recv_buf_idents.iter())
                     .zip(recv_hoffs.iter())
                     .map(|((port_ident, buf_ident), &hoff_id)| {
-                        let hoff_idx = slotmap_raw_idx(hoff_id);
+                        let hoff_ffi = slotmap_key_ffi(hoff_id);
                         // Use call_site span for internal identifiers to avoid
                         // hygiene issues when invoked through declarative macros
                         // (e.g. dfir_expect_warnings!). TODO(#2781): define these once.
@@ -1030,7 +1028,7 @@ impl DfirGraph {
                                     #work_done = true;
                                 }
                                 let hoff_metrics = &#metrics.handoffs[
-                                    #root::util::slot_vec::Key::<#root::scheduled::HandoffTag>::from_raw(#hoff_idx)
+                                    #root::slotmap::KeyData::from_ffi(#hoff_ffi).into()
                                 ];
                                 hoff_metrics.total_items_count.update(|x| x + hoff_len);
                                 hoff_metrics.curr_items_count.set(hoff_len);
@@ -1436,10 +1434,10 @@ impl DfirGraph {
                     .iter()
                     .zip(send_buf_idents.iter())
                     .map(|(&hoff_id, buf_ident)| {
-                        let hoff_idx = slotmap_raw_idx(hoff_id);
+                        let hoff_ffi = slotmap_key_ffi(hoff_id);
                         quote! {
                             __dfir_metrics.handoffs[
-                                #root::util::slot_vec::Key::<#root::scheduled::HandoffTag>::from_raw(#hoff_idx)
+                                #root::slotmap::KeyData::from_ffi(#hoff_ffi).into()
                             ].curr_items_count.set(#buf_ident.len());
                         }
                     })
@@ -1455,7 +1453,7 @@ impl DfirGraph {
                     };
                     {
                         let sg_metrics = &__dfir_metrics.subgraphs[
-                            #root::util::slot_vec::Key::<#root::scheduled::SubgraphTag>::from_raw(#sg_metrics_idx)
+                            #root::slotmap::KeyData::from_ffi(#sg_metrics_ffi).into()
                         ];
                         #root::scheduled::metrics::InstrumentSubgraph::new(
                             #sg_fut_ident, sg_metrics
@@ -1485,19 +1483,19 @@ impl DfirGraph {
         // Generate metrics initialization: one entry per handoff and per subgraph.
         let metrics_init_code = {
             let handoff_inits = handoff_nodes.iter().map(|&(node_id, _)| {
-                let idx = slotmap_raw_idx(node_id);
+                let ffi = slotmap_key_ffi(node_id);
                 quote! {
                     dfir_metrics.handoffs.insert(
-                        #root::util::slot_vec::Key::from_raw(#idx),
+                        #root::slotmap::KeyData::from_ffi(#ffi).into(),
                         ::std::default::Default::default(),
                     );
                 }
             });
             let subgraph_inits = all_subgraphs.iter().map(|&(sg_id, _)| {
-                let idx = slotmap_raw_idx(sg_id);
+                let ffi = slotmap_key_ffi(sg_id);
                 quote! {
                     dfir_metrics.subgraphs.insert(
-                        #root::util::slot_vec::Key::from_raw(#idx),
+                        #root::slotmap::KeyData::from_ffi(#ffi).into(),
                         ::std::default::Default::default(),
                     );
                 }

dfir_rs/Cargo.toml

@@ -14,7 +14,7 @@ workspace = true
 [features]
 default = [ "macros", "debugging", "meta" ]
 
-meta = [ "dep:dfir_lang" ]
+meta = [ ]
 macros = [ "dfir_macro" ]
 dfir_macro = [ "dep:dfir_macro" ]
 debugging = [ "dfir_lang/debugging" ]
@@ -28,7 +28,7 @@ name = "kvs_bench"
 [dependencies]
 bincode = "1.3.1"
 bytes = "1.1.0"
-dfir_lang = { path = "../dfir_lang", version = "^0.15.0", optional = true }
+dfir_lang = { path = "../dfir_lang", version = "^0.15.0" }
 dfir_macro = { optional = true, path = "../dfir_macro", version = "^0.15.0" }
 dfir_pipes = { path = "../dfir_pipes", version = "^0.0.1" }
 futures = "0.3.0"

dfir_rs/src/lib.rs

@@ -27,7 +27,9 @@ pub use ::{
 #[cfg(feature = "meta")]
 #[cfg_attr(docsrs, doc(cfg(feature = "meta")))]
 pub use dfir_lang as lang;
+pub use dfir_lang;
 pub use dfir_pipes::itertools;
+pub use slotmap;
 pub use variadics::{self, var_args, var_expr, var_type};
 
 /// `#[macro_use]` automagically brings the declarative macro export to the crate-level.

dfir_rs/src/scheduled/metrics.rs

@@ -5,12 +5,11 @@ use std::pin::Pin;
 use std::rc::Rc;
 use std::task::{Context, Poll};
 
+use dfir_lang::graph::{GraphNodeId, GraphSubgraphId};
 use pin_project_lite::pin_project;
+use slotmap::SecondaryMap;
 use web_time::{Duration, Instant};
 
-use super::{HandoffTag, SubgraphTag};
-use crate::util::slot_vec::SecondarySlotVec;
-
 /// Metrics for a [`Dfir`](super::context::Dfir) graph instance.
 ///
 /// Call [`Dfir::metrics`](super::context::Dfir::metrics) for reference-counted continually-updated metrics,
@@ -20,9 +19,9 @@ use crate::util::slot_vec::SecondarySlotVec;
 #[non_exhaustive]
 pub struct DfirMetrics {
     /// Per-subgraph metrics.
-    pub subgraphs: SecondarySlotVec<SubgraphTag, SubgraphMetrics>,
+    pub subgraphs: SecondaryMap<GraphSubgraphId, SubgraphMetrics>,
     /// Per-handoff metrics.
-    pub handoffs: SecondarySlotVec<HandoffTag, HandoffMetrics>,
+    pub handoffs: SecondaryMap<GraphNodeId, HandoffMetrics>,
 }
 
 impl DfirMetrics {
@@ -222,13 +221,18 @@ where
 
 #[cfg(test)]
 mod test {
+    use dfir_lang::graph::{GraphNodeId, GraphSubgraphId};
+    use slotmap::SlotMap;
+
     use super::*;
-    use crate::scheduled::{HandoffId, SubgraphId};
 
     #[test]
     fn test_dfir_metrics_intervals() {
-        let sg_id = SubgraphId::from_raw(0);
-        let handoff_id = HandoffId::from_raw(0);
+        // Create slotmaps to generate valid keys.
+        let mut sg_map: SlotMap<GraphSubgraphId, ()> = SlotMap::with_key();
+        let mut node_map: SlotMap<GraphNodeId, ()> = SlotMap::with_key();
+        let sg_id = sg_map.insert(());
+        let handoff_id = node_map.insert(());
 
         let mut metrics = DfirMetrics::default();
         metrics.subgraphs.insert(

dfir_rs/src/scheduled/mod.rs

@@ -11,14 +11,9 @@ pub mod ticks;
 
 /// Tag for [`SubgraphId`].
 pub enum SubgraphTag {}
-/// A subgraph's ID. Used as a key for metrics tracking.
+/// A subgraph's ID.
 pub type SubgraphId = Key<SubgraphTag>;
 
-/// Tag for [`HandoffId`].
-pub enum HandoffTag {}
-/// A handoff's ID. Used as a key for metrics tracking.
-pub type HandoffId = Key<HandoffTag>;
-
 /// Tag for [`StateId`].
 pub enum StateTag {}
 /// A state handle's ID.

hydro_lang/src/telemetry/emf.rs

@@ -184,7 +184,7 @@ where
                 ]
             },
             "LocationName": location_name,
-            "HandoffId": hoff_id.to_string(),
+            "HandoffId": format!("{:?}", hoff_id),
             "CurrItemsCount": hoff_metrics.curr_items_count(),
             "TotalItemsCount": hoff_metrics.total_items_count(),
         })
@@ -213,7 +213,7 @@ where
                 ]
             },
             "LocationName": location_name,
-            "SubgraphId": sg_id.to_string(),
+            "SubgraphId": format!("{:?}", sg_id),
             "TotalRunCount": sg_metrics.total_run_count(),
             "TotalPollDuration": sg_metrics.total_poll_duration().as_micros(),
             "TotalPollCount": sg_metrics.total_poll_count(),