only flush after holocene

crates/derive/src/stages/batch_queue.rs

@@ -166,8 +166,10 @@ where
                 }
                 BatchValidity::Drop => {
                     // If we drop a batch, flush previous batches buffered in the BatchStream
-                    // stage.
-                    self.prev.flush();
+                    // stage if holocene is active.
+                    if self.cfg.is_holocene_active(origin.timestamp) {
+                        self.prev.flush();
+                    }
                     warn!(target: "batch-queue", "Dropping batch with parent: {}", parent.block_info);
                     continue;
                 }
@@ -265,10 +267,13 @@ where
         let validity =
             data.check_batch(&self.cfg, &self.l1_blocks, parent, &mut self.fetcher).await;
         // Post-Holocene, future batches are dropped due to prevent gaps.
-        let drop = validity.is_drop() ||
-            (self.cfg.is_holocene_active(origin.timestamp) && validity.is_future());
+        let is_holocene = self.cfg.is_holocene_active(origin.timestamp);
+        let drop = validity.is_drop() || (is_holocene && validity.is_future());
         if drop {
-            self.prev.flush();
+            // If holocene is active, we don't flush the active span / channel.
+            if is_holocene {
+                self.prev.flush();
+            }
             return Ok(());
         } else if validity.is_outdated() {
             // If the batch is outdated, we drop it without flushing the previous stage.