feat(derive): wire in flushing

crates/derive/src/stages/batch_queue.rs

@@ -150,6 +150,9 @@ where
                     remaining.push(batch.clone());
                 }
                 BatchValidity::Drop => {
+                    // If we drop a batch, flush previous batches buffered in the BatchStream
+                    // stage.
+                    self.prev.flush();
                     warn!(target: "batch-queue", "Dropping batch with parent: {}", parent.block_info);
                     continue;
                 }