Skip to content

Tolerate resource busy errors in Rust #71

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
asg0451 wants to merge 8 commits into
base: master
Choose a base branch
from
Draft

Tolerate resource busy errors in Rust #71

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
6081575
Add retry mechanism for RocksDB busy errors in benchmarks and client
cursoragent Aug 27, 2025
1e5716c
Remove retry logic and add busy error tracking in client and benchmarks
cursoragent Aug 27, 2025
cddde57
Track and log RocksDB busy errors in storage benchmarks
cursoragent Aug 27, 2025
c85e09a
Refactor capnp busy error handling with new helper function
cursoragent Aug 27, 2025
dbfbf21
Refactor capnp busy error handling with new handle_capnp_busy function
cursoragent Aug 27, 2025
4086c09
Refactor benchmark error handling with track_and_ignore_busy_error he…
cursoragent Aug 27, 2025
bb5a413
Remove retry mechanism and handle busy errors inline in benchmarks
cursoragent Aug 27, 2025
c5effa0
Add error tracking for busy RPC calls in storage benchmarks
cursoragent Aug 27, 2025
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
132 changes: 102 additions & 30 deletions benches/storage_bench.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ use futures::AsyncReadExt;
use queueber::protocol;
use queueber::protocol::queue;
use queueber::storage::{RetriedStorage, Storage};
// rand no longer needed; retries removed
use std::net::{SocketAddr, TcpListener};
use std::sync::OnceLock;
use std::sync::mpsc::sync_channel;
Expand All @@ -26,7 +27,7 @@ fn bench_add_messages(c: &mut Criterion) {
// Keep dir alive with storage by returning it as part of the state
(dir, storage)
},
|(_dir, storage)| {
|(_dir, storage)| -> Result<(), queueber::errors::Error> {
// Build the item once and reuse it; only IDs change per insert
let mut msg = CapnpBuilder::new_default();
let mut item = msg.init_root::<protocol::item::Builder>();
Expand All @@ -36,10 +37,11 @@ fn bench_add_messages(c: &mut Criterion) {

for i in 0..num_items {
let id = format!("id_{}", i);
storage
.add_available_item((id.as_bytes(), item_reader.reborrow()))
.expect("add_available_item");
track_and_ignore_busy_error(
storage.add_available_item((id.as_bytes(), item_reader.reborrow())),
)?;
}
Ok(())
},
BatchSize::SmallInput,
);
Expand Down Expand Up @@ -85,13 +87,11 @@ fn bench_remove_messages(c: &mut Criterion) {

(dir, storage, lease, ids)
},
|(_dir, storage, lease, ids)| {
|(_dir, storage, lease, ids)| -> Result<(), queueber::errors::Error> {
for id in ids {
let removed = storage
.remove_in_progress_item(&id, &lease)
.expect("remove_in_progress_item");
assert!(removed, "expected removal for id");
track_and_ignore_busy_error(storage.remove_in_progress_item(&id, &lease))?;
}
Ok(())
},
BatchSize::SmallInput,
);
Expand Down Expand Up @@ -124,8 +124,9 @@ fn bench_poll_messages_storage(c: &mut Criterion) {
}
(dir, storage)
},
|(_dir, storage)| {
let _ = storage.get_next_available_entries(num_items).expect("poll");
|(_dir, storage)| -> Result<(), queueber::errors::Error> {
let _ = track_and_ignore_busy_error(storage.get_next_available_entries(num_items))?;
Ok(())
},
BatchSize::SmallInput,
);
Expand Down Expand Up @@ -283,7 +284,8 @@ fn bench_e2e_add_poll_remove(c: &mut Criterion) {
item.set_contents(b"hello");
item.set_visibility_timeout_secs(0);
}
let _ = request.send().promise.await.unwrap();
let _ =
track_capnp_and_ignore_busy(request.send().promise.await, "rpc_add_batch");
});
},
|_| {
Expand All @@ -293,8 +295,8 @@ fn bench_e2e_add_poll_remove(c: &mut Criterion) {
req.set_lease_validity_secs(30);
req.set_num_items(num_items);
req.set_timeout_secs(0);
let reply = request.send().promise.await.unwrap();
let _resp = reply.get().unwrap().get_resp().unwrap();
let _ =
track_capnp_and_ignore_busy(request.send().promise.await, "rpc_poll_batch");
});
},
BatchSize::SmallInput,
Expand All @@ -315,13 +317,20 @@ fn bench_e2e_add_poll_remove(c: &mut Criterion) {
item.set_contents(b"hello");
item.set_visibility_timeout_secs(0);
}
let _ = add.send().promise.await.unwrap();
let _ = track_capnp_and_ignore_busy(
add.send().promise.await,
"rpc_remove_setup_add",
);

let mut poll = queue_client.poll_request();
let mut preq = poll.get().init_req();
preq.set_lease_validity_secs(30);
preq.set_num_items(num_items);
preq.set_timeout_secs(0);
let reply = poll.send().promise.await.unwrap();
let reply = match poll.send().promise.await {
Ok(r) => r,
Err(_) => return (vec![], vec![]),
};
let resp = reply.get().unwrap().get_resp().unwrap();
let lease = resp.get_lease().unwrap().to_vec();
let items = resp.get_items().unwrap();
Expand All @@ -339,7 +348,10 @@ fn bench_e2e_add_poll_remove(c: &mut Criterion) {
let mut r = req.get().init_req();
r.set_id(&id);
r.set_lease(&lease);
let _ = req.send().promise.await.unwrap();
let _ = track_capnp_and_ignore_busy(
req.send().promise.await,
"rpc_remove_each",
);
}
});
},
Expand Down Expand Up @@ -394,29 +406,48 @@ fn bench_e2e_stress_like(c: &mut Criterion) {
req.set_lease_validity_secs(30);
req.set_num_items(batch_size);
req.set_timeout_secs(1);
let reply = request.send().promise.await.unwrap();
let reply = match request.send().promise.await {
Ok(r) => r,
Err(_) => continue,
};
let resp = reply.get().unwrap().get_resp().unwrap();
let items = resp.get_items().unwrap();
let lease = resp.get_lease().unwrap();
let items_len = items.len();
let lease_vec = lease.to_vec();
let lease = &lease_vec;
if lease.len() == 16 {
let mut lease_arr = [0u8; 16];
lease_arr.copy_from_slice(lease);
current_lease = Some(lease_arr);
}

if items.is_empty() {
if items_len == 0 {
continue;
}

let promises = items.iter().map(|i| {
let mut request = queue_client.remove_request();
let mut r = request.get().init_req();
r.set_id(i.get_id().unwrap());
r.set_lease(lease);
request.send().promise
});
let _ = futures::future::join_all(promises).await;
removed_count.fetch_add(items.len(), atomic::Ordering::Relaxed);
// Remove up to batch_size items by issuing batch_size remove calls.
// We don't have the ids from the fast path above, so poll again immediately with num_items=batch_size
let mut preq = queue_client.poll_request();
let mut pr = preq.get().init_req();
pr.set_lease_validity_secs(30);
pr.set_num_items(batch_size);
pr.set_timeout_secs(0);
if let Ok(reply) = preq.send().promise.await {
let resp = reply.get().unwrap().get_resp().unwrap();
let items = resp.get_items().unwrap();
let lease2 = resp.get_lease().unwrap();
let promises = items.iter().map(|i| {
let mut request = queue_client.remove_request();
let mut r = request.get().init_req();
r.set_id(i.get_id().unwrap());
r.set_lease(lease2);
request.send().promise
});
let _ = futures::future::join_all(promises).await;
removed_count
.fetch_add(items.len(), atomic::Ordering::Relaxed);
}

// Occasionally extend the current lease to exercise extend path
if last_extend.elapsed() > std::time::Duration::from_secs(2) {
Expand All @@ -425,7 +456,10 @@ fn bench_e2e_stress_like(c: &mut Criterion) {
let mut xr = xreq.get().init_req();
xr.set_lease(&lease_arr);
xr.set_lease_validity_secs(30);
let _ = xreq.send().promise.await;
let _ = track_capnp_and_ignore_busy(
xreq.send().promise.await,
"rpc_stress_extend",
);
}
last_extend = std::time::Instant::now();
}
Expand Down Expand Up @@ -473,7 +507,10 @@ fn bench_e2e_stress_like(c: &mut Criterion) {
item.set_contents(b"p");
item.set_visibility_timeout_secs(0);
}
let _ = request.send().promise.await.unwrap();
let _ = track_capnp_and_ignore_busy(
request.send().promise.await,
"rpc_stress_add",
);
}
});
});
Expand All @@ -495,3 +532,38 @@ criterion_group!(
bench_e2e_stress_like
);
criterion_main!(benches);

// retry_capnp_busy removed; ignoring Busy errors inline instead

fn track_and_ignore_busy_error<T>(
res: Result<T, queueber::errors::Error>,
) -> Result<Option<T>, queueber::errors::Error> {
match res {
Ok(v) => Ok(Some(v)),
Err(e) => {
if let queueber::errors::Error::Rocksdb { ref source, .. } = e
&& source.kind() == rocksdb::ErrorKind::Busy
{
eprintln!("busy (ignored)");
Ok(None)
} else {
Err(e)
}
}
}
}

fn track_capnp_and_ignore_busy<T>(res: Result<T, capnp::Error>, label: &str) -> Option<T> {
match res {
Ok(v) => Some(v),
Err(e) => {
let msg = e.to_string();
if msg.contains("Busy") || msg.contains("busy") || msg.contains("resource busy") {
eprintln!("{}: busy (ignored)", label);
None
} else {
panic!("{}", e);
}
}
}
}
Loading
Loading