Add Message::HistoricalBatch variant for efficient bulk data transfer in Historical mode

Author: 0-jake-0

wingfoil/src/channel/kanal_chan.rs

@@ -110,6 +110,12 @@ impl<T: Element + Send> ChannelSender<T> {
         self.send_message(message)
     }
 
+    #[allow(dead_code)]
+    pub fn send_historical_batch(&self, batch: Vec<ValueAt<T>>) -> SendResult {
+        let message = Message::HistoricalBatch(batch.into_boxed_slice());
+        self.send_message(message)
+    }
+
     pub fn close(&mut self) -> SendResult {
         // check if already closed
         if self.kanal_sender.is_none() {
@@ -177,6 +183,12 @@ impl<T: Element + Send> AsyncChannelSender<T> {
         self.send_message(message).await;
     }
 
+    #[allow(dead_code)]
+    pub async fn send_historical_batch(&self, batch: Vec<ValueAt<T>>) {
+        let message = Message::HistoricalBatch(batch.into_boxed_slice());
+        self.send_message(message).await;
+    }
+
     pub async fn close(&mut self) {
         let message = Message::EndOfStream;
         self.send_message(message).await;

wingfoil/src/channel/message.rs

@@ -9,7 +9,7 @@ use crate::types::Element;
 /// Message that can be sent between threads.
 #[derive(new, Debug, Clone, PartialEq, Eq)]
 pub(crate) enum Message<T: Element + Send> {
-    /// In [RunMode::HistoricalFrom], this message  
+    /// In [RunMode::HistoricalFrom], this message
     /// allows receiving graph to progress, even when
     /// the channel is ticking less frequently than
     /// other inputs.
@@ -22,6 +22,9 @@ pub(crate) enum Message<T: Element + Send> {
     HistoricalValue(ValueAt<T>),
     /// Sent in [RunMode::RealTime].  Just a value.
     RealtimeValue(T),
+    /// Used in [RunMode::HistoricalFrom] for bulk data transfer.
+    /// Contains multiple values with their timestamps.
+    HistoricalBatch(Box<[ValueAt<T>]>),
 }
 
 pub trait ReceiverMessageSource<T: Element + Send> {

wingfoil/src/nodes/async_io.rs

@@ -331,6 +331,12 @@ where
                     Message::CheckPoint(t) => {
                         time = *t;
                     },
+                    Message::HistoricalBatch(batch) => {
+                        // Update time to earliest timestamp in batch
+                        if let Some(value_at) = batch.first() {
+                            time = value_at.time;
+                        }
+                    }
                     Message::EndOfStream => {
                         finished = true;
                     },
@@ -356,6 +362,13 @@ where
                     Message::HistoricalValue(value_at) => {
                         yield (value_at.time, value_at.value)
                     },
+                    Message::HistoricalBatch(batch) => {
+                        // Convert to Vec to own the data and avoid holding reference across await
+                        let batch_vec = batch.into_vec();
+                        for value_at in batch_vec {
+                            yield (value_at.time, value_at.value)
+                        }
+                    }
                     Message::CheckPoint(_) => {},
                     Message::EndOfStream => {},
                 }

wingfoil/src/nodes/channel.rs

@@ -109,6 +109,11 @@ impl<T: Element + Send> MutableNode for ReceiverStream<T> {
                                 Message::HistoricalValue(value_at) => {
                                     values.push(value_at.value);
                                 }
+                                Message::HistoricalBatch(_) => {
+                                    return Err(anyhow!(
+                                        "received HistoricalBatch but RunMode is RealTime"
+                                    ));
+                                }
                                 Message::EndOfStream => self.finished = true,
                                 Message::CheckPoint(_) => {}
                             },
@@ -157,6 +162,29 @@ impl<T: Element + Send> MutableNode for ReceiverStream<T> {
                             self.message_time = Some(value_at.time);
                             self.queue.push_back(value_at);
                         }
+                        Message::HistoricalBatch(batch) => {
+                            if batch.is_empty() {
+                                continue;
+                            }
+
+                            // Validate: all timestamps must be >= current graph time
+                            let min_time = batch.iter().map(|va| va.time).min().unwrap();
+                            if min_time < state.time() {
+                                return Err(anyhow!(
+                                    "received HistoricalBatch with timestamp less than graph time, {} < {}",
+                                    min_time,
+                                    state.time()
+                                ));
+                            }
+
+                            // Set message_time to earliest timestamp
+                            self.message_time = Some(min_time);
+
+                            // Unpack all values into queue
+                            for value_at in batch.iter() {
+                                self.queue.push_back(value_at.clone());
+                            }
+                        }
                         Message::EndOfStream => self.finished = true,
                         Message::CheckPoint(check_point) => {
                             self.message_time = Some(check_point);