Migrate the load contract from positional to key-addressed

Positional Vec -> Vec output cannot detect reordering and misdelivers
results silently. A key-addressed map makes violations detectable:
a missing key is addressable (Error::MissingOutput), an unknown key
fails the whole batch (Error::ContractViolation).

Author: nerjs

examples/quickstart.rs

@@ -1,8 +1,11 @@
-// Smoke demo: implement BatchCollector, read the default config, see that equal
-// inputs hash to one key. Run: cargo run --example quickstart
+// Smoke demo: implement BatchCollector, read the default config,
+// see that equal inputs map to the same key.
+// Run: cargo run --example quickstart
 // Expected output:
 //   config: window=30ms max_batch=1024 timeout=30s concurrency=None max_waiting=None
-//   key(7) = 7, key(7) = 7 -> one key, one shared ride
+//   key(7) = 7, key(7) = 7 -> equal inputs share one batch entry
+
+use std::collections::HashMap;
 
 use carpool::{BatchCollector, BatchLoaderConfig};
 
@@ -18,8 +21,8 @@ impl BatchCollector for SquareLoader {
         *input
     }
 
-    async fn load(&self, inputs: Vec<u64>) -> Result<Vec<u64>, Self::Error> {
-        Ok(inputs.iter().map(|n| n * n).collect())
+    async fn load(&self, batch: HashMap<u64, u64>) -> Result<HashMap<u64, u64>, Self::Error> {
+        Ok(batch.into_iter().map(|(k, n)| (k, n * n)).collect())
     }
 }
 
@@ -36,5 +39,5 @@ fn main() {
 
     let loader = SquareLoader;
     let (a, b) = (loader.key(&7), loader.key(&7));
-    println!("key(7) = {a}, key(7) = {b} -> one key, one shared ride");
+    println!("key(7) = {a}, key(7) = {b} -> equal inputs share one batch entry");
 }

src/collector.rs

@@ -1,3 +1,4 @@
+use std::collections::HashMap;
 use std::future::Future;
 use std::hash::Hash;
 
@@ -9,11 +10,17 @@ pub trait BatchCollector: Send + Sync + 'static {
 
     fn key(&self, input: &Self::Input) -> Self::Key;
 
-    // inputs are deduplicated by key; output must be position-aligned to them
+    // batch arrives deduplicated: inputs sharing a key are interchangeable,
+    // only one representative reaches load.
+    // strict contract: every requested key must get a value;
+    // a missing key is Error::MissingOutput for its waiters,
+    // an unknown key in the response is Error::ContractViolation
+    // for the whole batch (it takes precedence when both occur).
+    // absence semantics live in the implementor's Output type.
     fn load(
         &self,
-        inputs: Vec<Self::Input>,
-    ) -> impl Future<Output = Result<Vec<Self::Output>, Self::Error>> + Send;
+        batch: HashMap<Self::Key, Self::Input>,
+    ) -> impl Future<Output = Result<HashMap<Self::Key, Self::Output>, Self::Error>> + Send;
 }
 
 #[cfg(test)]
@@ -32,21 +39,22 @@ mod tests {
             *input
         }
 
-        async fn load(&self, inputs: Vec<u64>) -> Result<Vec<u64>, Self::Error> {
-            Ok(inputs.iter().map(|x| x * x).collect())
+        async fn load(&self, batch: HashMap<u64, u64>) -> Result<HashMap<u64, u64>, Self::Error> {
+            Ok(batch.into_iter().map(|(k, x)| (k, x * x)).collect())
         }
     }
 
     // Spawning the load future proves it is `Send` (the whole point of the
     // RPITIT + Send form) and that a plain `async fn` satisfies the bound.
     #[tokio::test]
-    async fn load_future_is_send_and_position_aligned() {
+    async fn load_future_is_send_and_key_addressed() {
         let loader = SquareLoader;
-        let out = tokio::spawn(async move { loader.load(vec![2, 3, 4]).await })
+        let batch = HashMap::from([(2, 2), (3, 3), (4, 4)]);
+        let out = tokio::spawn(async move { loader.load(batch).await })
             .await
             .expect("task joins")
             .expect("load succeeds");
 
-        assert_eq!(out, vec![4, 9, 16]);
+        assert_eq!(out, HashMap::from([(2, 4), (3, 9), (4, 16)]));
     }
 }

src/error.rs

@@ -4,7 +4,14 @@ use std::fmt;
 #[non_exhaustive]
 pub enum Error<E> {
     Collector(E),
-    ContractViolation { expected: usize, got: usize },
+    // implementor bug: the response carries keys that were not in the batch;
+    // not attributable to any single caller, so the whole batch fails
+    // TODO: consider #[non_exhaustive] on this variant before 1.0 -
+    // enum-level #[non_exhaustive] does not stop exhaustive field destructuring
+    ContractViolation { unknown_keys: usize },
+    // implementor bug: no output under a requested key; not a domain "not found" -
+    // absence semantics belong to the implementor's Output type
+    MissingOutput,
     Timeout,
     WaitingTimeout,
 }
@@ -13,10 +20,11 @@ impl<E: fmt::Display> fmt::Display for Error<E> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             Error::Collector(e) => write!(f, "collector failed: {e}"),
-            Error::ContractViolation { expected, got } => write!(
+            Error::ContractViolation { unknown_keys } => write!(
                 f,
-                "collector broke the position-aligned contract: expected {expected} outputs, got {got}"
+                "collector broke the key-addressed contract: {unknown_keys} unknown key(s) in the response"
             ),
+            Error::MissingOutput => f.write_str("collector returned no output for a requested key"),
             Error::Timeout => f.write_str("batch timed out"),
             Error::WaitingTimeout => f.write_str("timed out waiting for a concurrency slot"),
         }
@@ -27,7 +35,10 @@ impl<E: std::error::Error + 'static> std::error::Error for Error<E> {
     fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
         match self {
             Error::Collector(e) => Some(e),
-            Error::ContractViolation { .. } | Error::Timeout | Error::WaitingTimeout => None,
+            Error::ContractViolation { .. }
+            | Error::MissingOutput
+            | Error::Timeout
+            | Error::WaitingTimeout => None,
         }
     }
 }
@@ -54,4 +65,23 @@ mod tests {
 
         assert_eq!(source.to_string(), "downstream: boom");
     }
+
+    // Contract-bug variants wrap nothing downstream:
+    // no source, and their Display stands on its own without touching E.
+    #[test]
+    fn contract_bug_variants_have_no_source_and_render() {
+        let missing: Error<DownstreamError> = Error::MissingOutput;
+        let violation: Error<DownstreamError> = Error::ContractViolation { unknown_keys: 2 };
+
+        assert!(std::error::Error::source(&missing).is_none());
+        assert!(std::error::Error::source(&violation).is_none());
+        assert_eq!(
+            missing.to_string(),
+            "collector returned no output for a requested key"
+        );
+        assert_eq!(
+            violation.to_string(),
+            "collector broke the key-addressed contract: 2 unknown key(s) in the response"
+        );
+    }
 }