Add a new event type to notify nodes about dropped inputs

apis/c++/node/src/lib.rs

@@ -142,6 +142,7 @@ fn event_as_input(event: Box<DoraEvent>) -> eyre::Result<ffi::DoraInput> {
  id,
  metadata: _,
  data,
+ ..
  }) = event.0
  else {
  bail!("not an input event");

apis/rust/node/src/event_stream/event.rs

@@ -16,17 +16,33 @@ pub enum Event {
  Reload {
  operator_id: Option<OperatorId>,
  },
+ #[non_exhaustive]
  Input {
  id: DataId,
  metadata: Metadata,
  data: ArrowData,
+ /// Number of dropped inputs of this ID.
+ ///
+ /// Specifies the number of inputs of this ID that were dropped _before_ this input.
+ dropped: usize,
  },
  InputClosed {
  id: DataId,
  },
  Error(String),
 }
 
+impl Event {
+ pub fn new_input(id: DataId, metadata: Metadata, data: ArrowData) -> Event {
+ Event::Input {
+ id,
+ metadata,
+ data,
+ dropped: 0,
+ }
+ }
+}
+
 pub enum RawData {
  Empty,
  Vec(AVec<u8, ConstAlign<128>>),

apis/rust/node/src/event_stream/mod.rs

@@ -136,7 +136,12 @@ impl EventStream {
  NodeEvent::Stop => Event::Stop,
  NodeEvent::Reload { operator_id } => Event::Reload { operator_id },
  NodeEvent::InputClosed { id } => Event::InputClosed { id },
- NodeEvent::Input { id, metadata, data } => {
+ NodeEvent::Input {
+ id,
+ metadata,
+ data,
+ dropped,
+ } => {
  let data = match data {
  None => Ok(None),
  Some(daemon_messages::DataMessage::Vec(v)) => Ok(Some(RawData::Vec(v))),
@@ -164,6 +169,7 @@ impl EventStream {
  id,
  metadata,
  data: data.into(),
+ dropped,
  },
  Err(err) => Event::Error(format!("{err:?}")),
  }

binaries/cli/src/template/rust/node/main-template.rs

@@ -10,6 +10,7 @@ fn main() -> Result<(), Box<dyn Error>> {
  id,
  metadata,
  data: _,
+ ..
  } => match id.as_str() {
  other => eprintln!("Received input `{other}`"),
  },

binaries/daemon/src/lib.rs

@@ -984,6 +984,7 @@ impl Daemon {
  id: input_id.clone(),
  metadata: metadata.clone(),
  data: None,
+ dropped: 0,
  },
  &self.clock,
  );
@@ -1031,6 +1032,7 @@ impl Daemon {
  id: input_id.clone(),
  metadata: metadata.clone(),
  data: Some(message.clone()),
+ dropped: 0,
  },
  &self.clock,
  );
@@ -1162,6 +1164,7 @@ async fn send_output_to_local_receivers(
  id: input_id.clone(),
  metadata: metadata.clone(),
  data: data.clone(),
+ dropped: 0,
  };
  match channel.send(Timestamped {
  inner: item,

binaries/daemon/src/node_communication/mod.rs

@@ -11,7 +11,7 @@ use eyre::{eyre, Context};
 use futures::{future, task, Future};
 use shared_memory_server::{ShmemConf, ShmemServer};
 use std::{
- collections::{BTreeMap, VecDeque},
+ collections::{BTreeMap, HashMap, VecDeque},
  mem,
  net::Ipv4Addr,
  sync::Arc,
@@ -151,6 +151,7 @@ struct Listener {
  queue: VecDeque<Box<Option<Timestamped<NodeEvent>>>>,
  queue_sizes: BTreeMap<DataId, usize>,
  clock: Arc<uhlc::HLC>,
+ dropped_inputs: HashMap<DataId, usize>,
 }
 
 impl Listener {
@@ -211,6 +212,7 @@ impl Listener {
  queue_sizes,
  queue: VecDeque::new(),
  clock: hlc.clone(),
+ dropped_inputs: HashMap::new(),
  };
  match listener
  .run_inner(connection)
@@ -281,7 +283,10 @@ impl Listener {
 
  async fn handle_events(&mut self) -> eyre::Result<()> {
  if let Some(events) = &mut self.subscribed_events {
- while let Ok(event) = events.try_recv() {
+ while let Ok(mut event) = events.try_recv() {
+ if let NodeEvent::Input { id, dropped, .. } = &mut event.inner {
+ *dropped += self.dropped_inputs.remove(id).unwrap_or_default();
+ }
  self.queue.push_back(Box::new(Some(event)));
  }
 
@@ -294,21 +299,24 @@ impl Listener {
  #[tracing::instrument(skip(self), fields(%self.node_id), level = "trace")]
  async fn drop_oldest_inputs(&mut self) -> Result<(), eyre::ErrReport> {
  let mut queue_size_remaining = self.queue_sizes.clone();
- let mut dropped = 0;
  let mut drop_tokens = Vec::new();
 
  // iterate over queued events, newest first
  for event in self.queue.iter_mut().rev() {
  let Some(Timestamped {
- inner: NodeEvent::Input { id, data, .. },
+ inner:
+ NodeEvent::Input {
+ id, data, dropped, ..
+ },
  ..
  }) = event.as_mut()
  else {
  continue;
  };
  match queue_size_remaining.get_mut(id) {
  Some(0) => {
- dropped += 1;
+ *self.dropped_inputs.entry(id.clone()).or_default() += *dropped + 1;
+
  if let Some(drop_token) = data.as_ref().and_then(|d| d.drop_token()) {
  drop_tokens.push(drop_token);
  }
@@ -324,9 +332,6 @@ impl Listener {
  }
  self.report_drop_tokens(drop_tokens).await?;
 
- if dropped > 0 {
- tracing::debug!("dropped {dropped} inputs because event queue was too full");
- }
  Ok(())
  }
 

binaries/runtime/src/lib.rs

@@ -248,7 +248,9 @@ async fn run(
  RuntimeEvent::Event(Event::Reload { operator_id: None }) => {
  tracing::warn!("Reloading runtime nodes is not supported");
  }
- RuntimeEvent::Event(Event::Input { id, metadata, data }) => {
+ RuntimeEvent::Event(Event::Input {
+ id, metadata, data, ..
+ }) => {
  let Some((operator_id, input_id)) = id.as_str().split_once('/') else {
  tracing::warn!("received non-operator input {id}");
  continue;
@@ -261,11 +263,7 @@ async fn run(
  };
 
  if let Err(err) = operator_channel
- .send_async(Event::Input {
- id: input_id.clone(),
- metadata,
- data,
- })
+ .send_async(Event::new_input(input_id.clone(), metadata, data))
  .await
  .wrap_err_with(|| {
  format!("failed to send input `{input_id}` to operator `{operator_id}`")

binaries/runtime/src/operator/shared_lib.rs

@@ -191,6 +191,7 @@ impl<'lib> SharedLibraryOperator<'lib> {
  id: input_id,
  metadata,
  data,
+ ..
  } => {
  let (data_array, schema) = arrow::ffi::to_ffi(&data.to_data())?;
 

examples/benchmark/sink/src/main.rs

@@ -20,7 +20,9 @@ fn main() -> eyre::Result<()> {
 
  while let Some(event) = events.recv() {
  match event {
- Event::Input { id, metadata, data } => {
+ Event::Input {
+ id, metadata, data, ..
+ } => {
  // check if new size bracket
  let data_len = data.len();
  if data_len != current_size {

examples/multiple-daemons/node/src/main.rs

@@ -18,6 +18,7 @@ fn main() -> eyre::Result<()> {
  id,
  metadata,
  data: _,
+ ..
  } => match id.as_str() {
  "tick" => {
  let random: u64 = rand::random();

examples/multiple-daemons/sink/src/main.rs

@@ -10,6 +10,7 @@ fn main() -> eyre::Result<()> {
  id,
  metadata: _,
  data,
+ ..
  } => match id.as_str() {
  "message" => {
  let received_string: &str =

examples/rust-dataflow/node/src/main.rs

@@ -18,6 +18,7 @@ fn main() -> eyre::Result<()> {
  id,
  metadata,
  data: _,
+ ..
  } => match id.as_str() {
  "tick" => {
  let random: u64 = rand::random();

examples/rust-dataflow/sink/src/main.rs

@@ -10,6 +10,7 @@ fn main() -> eyre::Result<()> {
  id,
  metadata: _,
  data,
+ ..
  } => match id.as_str() {
  "message" => {
  let received_string: &str =

examples/rust-dataflow/status-node/src/main.rs

@@ -10,7 +10,9 @@ fn main() -> eyre::Result<()> {
  let mut ticks = 0;
  while let Some(event) = events.recv() {
  match event {
- Event::Input { id, metadata, data } => match id.as_ref() {
+ Event::Input {
+ id, metadata, data, ..
+ } => match id.as_ref() {
  "tick" => {
  ticks += 1;
  }

libraries/core/src/daemon_messages.rs

@@ -155,6 +155,10 @@ pub enum NodeEvent {
  id: DataId,
  metadata: Metadata,
  data: Option<DataMessage>,
+ /// Number of dropped inputs of this ID.
+ ///
+ /// Specifies the number of inputs of this ID that were dropped _before_ this input.
+ dropped: usize,
  },
  InputClosed {
  id: DataId,

tool_nodes/dora-record/src/main.rs

@@ -25,7 +25,9 @@ async fn main() -> eyre::Result<()> {
 
  while let Some(event) = events.recv() {
  match event {
- Event::Input { id, data, metadata } => {
+ Event::Input {
+ id, data, metadata, ..
+ } => {
  match writers.get(&id) {
  None => {
  let field_uhlc = Field::new("timestamp_uhlc", DataType::UInt64, false);

tool_nodes/dora-rerun/src/main.rs

@@ -43,6 +43,7 @@ fn main() -> Result<()> {
  id,
  data,
  metadata: _,
+ ..
  } = event
  {
  if id.as_str().contains("image") {