Fix a bug in the batcher

ipa-core/src/protocol/context/batcher.rs

@@ -114,7 +114,7 @@ impl<'a, B> Batcher<'a, B> {
             while self.batches.len() <= batch_offset {
                 let (validation_result, _) = watch::channel::<bool>(false);
                 let state = BatchState {
-                    batch: (self.batch_constructor)(self.first_batch + batch_offset),
+                    batch: (self.batch_constructor)(self.first_batch + self.batches.len()),
                     validation_result,
                     pending_count: 0,
                     pending_records: bitvec![0; self.records_per_batch],
@@ -296,6 +296,23 @@ mod tests {
         );
     }
 
+    #[test]
+    fn makes_batches_out_of_order() {
+        // Regression test for a bug where, when adding batches i..j to fill in a gap in
+        // the batch deque prior to out-of-order requested batch j, the batcher passed
+        // batch index `j` to the constructor for all of them, as opposed to the correct
+        // sequence of indices i..=j.
+
+        let batcher = Batcher::new(1, 2, Box::new(std::convert::identity));
+        let mut batcher = batcher.lock().unwrap();
+
+        batcher.get_batch(RecordId::from(1));
+        batcher.get_batch(RecordId::from(0));
+
+        assert_eq!(batcher.get_batch(RecordId::from(0)).batch, 0);
+        assert_eq!(batcher.get_batch(RecordId::from(1)).batch, 1);
+    }
+
     #[tokio::test]
     async fn validates_batches() {
         let batcher = Batcher::new(2, 4, Box::new(|_| Vec::new()));