Skip to content

[WIP] Add node-local wakeups to ProcessorInbox #2470

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
lquerel wants to merge 18 commits into
base: main
Choose a base branch
from
Draft

[WIP] Add node-local wakeups to ProcessorInbox #2470

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
18 commits
Select commit Hold shift + click to select a range
f65efd6
Add processor-local wakeups and fix workspace checks
lquerel Mar 30, 2026
09d650f
Handle wakeups in control channel benchmark
lquerel Mar 31, 2026
44d64d4
Add batch wakeup liveness test
lquerel Apr 2, 2026
24408dc
Document processor wakeup API
lquerel Apr 2, 2026
207af00
Use indexed heap for local wakeups
lquerel Apr 2, 2026
9237dc0
Document local wakeup guarantees
lquerel Apr 2, 2026
6c05e19
Derive batch wakeup slots from signal keys
lquerel Apr 2, 2026
aeb0527
Handle durable buffer wakeup overflow
lquerel Apr 2, 2026
fc65d84
Add wakeup regression tests
lquerel Apr 3, 2026
dfc9519
Document wakeup test guarantees
lquerel Apr 3, 2026
8146d39
Document batch wakeup slot test
lquerel Apr 3, 2026
5725ef2
Refine local wakeup API and durable retry state
lquerel Apr 3, 2026
bc2f194
Fix wakeup and exporter test regressions
lquerel Apr 3, 2026
9d5fe47
Simplify durable buffer retry scheduling
lquerel Apr 4, 2026
2f8d8d2
Simplify deferred retry state and inbox docs
lquerel Apr 4, 2026
5b05cff
Make processor wakeups opt-in
lquerel Apr 4, 2026
ecb2ce5
Drain deferred retries during shutdown
lquerel Apr 4, 2026
4a0338d
Merge branch 'main' into feature/node-local-wakeup
lquerel Apr 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions rust/otap-dataflow/benchmarks/benches/control_channel/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,7 +140,8 @@ async fn consume_current_local(
NodeControlMsg::Config { .. } => observed.configs += 1,
NodeControlMsg::DrainIngress { .. }
| NodeControlMsg::Shutdown { .. }
| NodeControlMsg::DelayedData { .. } => {
| NodeControlMsg::DelayedData { .. }
| NodeControlMsg::Wakeup { .. } => {
panic!("unexpected message in benchmark current local receiver");
}
}
Expand Down Expand Up @@ -203,7 +204,8 @@ async fn consume_current_shared(
NodeControlMsg::Config { .. } => observed.configs += 1,
NodeControlMsg::DrainIngress { .. }
| NodeControlMsg::Shutdown { .. }
| NodeControlMsg::DelayedData { .. } => {
| NodeControlMsg::DelayedData { .. }
| NodeControlMsg::Wakeup { .. } => {
panic!("unexpected message in benchmark current shared receiver");
}
}
Expand Down
6 changes: 6 additions & 0 deletions rust/otap-dataflow/crates/core-nodes/src/exporters/otap_exporter/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1280,11 +1280,13 @@ mod tests {

let (batch_received_tx, mut batch_received_rx) = tokio::sync::mpsc::channel(1);
let (server_shutdown_tx, server_shutdown_rx) = tokio::sync::oneshot::channel();
let (server_ready_tx, server_ready_rx) = tokio::sync::oneshot::channel();

// Start gRPC server that returns errors
let listening_addr: SocketAddr = format!("{grpc_addr}:{grpc_port}").parse().unwrap();
let server_handle = tokio_rt.spawn(async move {
let tcp_listener = TcpListener::bind(listening_addr).await.unwrap();
let _ = server_ready_tx.send(());
let tcp_stream = TcpListenerStream::new(tcp_listener);
let error_service = ArrowLogsServiceServer::new(ArrowLogsServiceGrpcErrorMock {
sender: batch_received_tx,
Expand Down Expand Up @@ -1316,6 +1318,10 @@ mod tests {
});

tokio::join!(local_set, async {
server_ready_rx
.await
.expect("server should bind before exporter traffic starts");

// Send a batch with ACK/NACK subscription
let log_message = create_otap_batch(LOG_BATCH_ID, ArrowPayloadType::Logs);
let pdata = OtapPdata::new_default(log_message.into()).test_subscribe_to(
Expand Down
Loading
Loading