From 1dfbd27cce9784a1bb7e2f85441473c1506647d6 Mon Sep 17 00:00:00 2001 From: Lewis Date: Sun, 12 Apr 2026 17:57:33 +0300 Subject: [PATCH] fix(postgres): semaphore on car endpoint & more efficient query Lewis: May this revision serve well! --- Cargo.lock | 44 ++--- Cargo.toml | 2 +- crates/tranquil-config/src/lib.rs | 4 + crates/tranquil-pds/src/comms/service.rs | 29 ++- crates/tranquil-pds/src/scheduled.rs | 27 +-- crates/tranquil-pds/src/state.rs | 4 + .../tests/sync_resource_limits.rs | 147 +++++++++++++++ crates/tranquil-store/benches/recovery.rs | 4 +- .../src/blockstore/compaction.rs | 16 +- .../src/blockstore/group_commit.rs | 2 +- crates/tranquil-store/src/blockstore/hint.rs | 174 +++++++----------- crates/tranquil-store/tests/sim_blockstore.rs | 4 +- crates/tranquil-sync/src/repo.rs | 40 ++-- 13 files changed, 312 insertions(+), 185 deletions(-) create mode 100644 crates/tranquil-pds/tests/sync_resource_limits.rs diff --git a/Cargo.lock b/Cargo.lock index 48fb813..fd1a0ff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7405,7 +7405,7 @@ dependencies = [ [[package]] name = "tranquil-api" -version = "0.5.1" +version = "0.5.2" dependencies = [ "anyhow", "axum", @@ -7456,7 +7456,7 @@ dependencies = [ [[package]] name = "tranquil-auth" -version = "0.5.1" +version = "0.5.2" dependencies = [ "anyhow", "base32", @@ -7479,7 +7479,7 @@ dependencies = [ [[package]] name = "tranquil-cache" -version = "0.5.1" +version = "0.5.2" dependencies = [ "async-trait", "base64 0.22.1", @@ -7493,7 +7493,7 @@ dependencies = [ [[package]] name = "tranquil-comms" -version = "0.5.1" +version = "0.5.2" dependencies = [ "async-trait", "base64 0.22.1", @@ -7511,7 +7511,7 @@ dependencies = [ [[package]] name = "tranquil-config" -version = "0.5.1" +version = "0.5.2" dependencies = [ "confique", "serde", @@ -7519,7 +7519,7 @@ dependencies = [ [[package]] name = "tranquil-crypto" -version = "0.5.1" +version = "0.5.2" dependencies = [ "aes-gcm", "base64 0.22.1", @@ -7535,7 +7535,7 @@ dependencies = [ [[package]] name = "tranquil-db" -version = "0.5.1" +version = "0.5.2" dependencies = [ "async-trait", "chrono", @@ -7552,7 +7552,7 @@ dependencies = [ [[package]] name = "tranquil-db-traits" -version = "0.5.1" +version = "0.5.2" dependencies = [ "async-trait", "base64 0.22.1", @@ -7568,7 +7568,7 @@ dependencies = [ [[package]] name = "tranquil-infra" -version = "0.5.1" +version = "0.5.2" dependencies = [ "async-trait", "bytes", @@ -7579,7 +7579,7 @@ dependencies = [ [[package]] name = "tranquil-lexicon" -version = "0.5.1" +version = "0.5.2" dependencies = [ "chrono", "hickory-resolver", @@ -7597,7 +7597,7 @@ dependencies = [ [[package]] name = "tranquil-oauth" -version = "0.5.1" +version = "0.5.2" dependencies = [ "anyhow", "axum", @@ -7620,7 +7620,7 @@ dependencies = [ [[package]] name = "tranquil-oauth-server" -version = "0.5.1" +version = "0.5.2" dependencies = [ "axum", "base64 0.22.1", @@ -7653,7 +7653,7 @@ dependencies = [ [[package]] name = "tranquil-pds" -version = "0.5.1" +version = "0.5.2" dependencies = [ "aes-gcm", "anyhow", @@ -7745,7 +7745,7 @@ dependencies = [ [[package]] name = "tranquil-repo" -version = "0.5.1" +version = "0.5.2" dependencies = [ "bytes", "cid", @@ -7757,7 +7757,7 @@ dependencies = [ [[package]] name = "tranquil-ripple" -version = "0.5.1" +version = "0.5.2" dependencies = [ "async-trait", "backon", @@ -7782,7 +7782,7 @@ dependencies = [ [[package]] name = "tranquil-scopes" -version = "0.5.1" +version = "0.5.2" dependencies = [ "axum", "futures", @@ -7798,7 +7798,7 @@ dependencies = [ [[package]] name = "tranquil-server" -version = "0.5.1" +version = "0.5.2" dependencies = [ "axum", "clap", @@ -7819,7 +7819,7 @@ dependencies = [ [[package]] name = "tranquil-signal" -version = "0.5.1" +version = "0.5.2" dependencies = [ "async-trait", "chrono", @@ -7842,7 +7842,7 @@ dependencies = [ [[package]] name = "tranquil-storage" -version = "0.5.1" +version = "0.5.2" dependencies = [ "async-trait", "aws-config", @@ -7859,7 +7859,7 @@ dependencies = [ [[package]] name = "tranquil-store" -version = "0.5.1" +version = "0.5.2" dependencies = [ "async-trait", "bytes", @@ -7905,7 +7905,7 @@ dependencies = [ [[package]] name = "tranquil-sync" -version = "0.5.1" +version = "0.5.2" dependencies = [ "anyhow", "axum", @@ -7927,7 +7927,7 @@ dependencies = [ [[package]] name = "tranquil-types" -version = "0.5.1" +version = "0.5.2" dependencies = [ "chrono", "cid", diff --git a/Cargo.toml b/Cargo.toml index 0839019..b2493e0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,7 +26,7 @@ members = [ ] [workspace.package] -version = "0.5.1" +version = "0.5.2" edition = "2024" license = "AGPL-3.0-or-later" diff --git a/crates/tranquil-config/src/lib.rs b/crates/tranquil-config/src/lib.rs index b686b89..8c06d6d 100644 --- a/crates/tranquil-config/src/lib.rs +++ b/crates/tranquil-config/src/lib.rs @@ -720,6 +720,10 @@ pub struct FirehoseConfig { #[config(env = "FIREHOSE_MAX_LAG", default = 5000)] pub max_lag: u64, + /// Maximum concurrent full-repo exports, eg. getRepo without `since`. + #[config(env = "MAX_CONCURRENT_REPO_EXPORTS", default = 4)] + pub max_concurrent_repo_exports: usize, + /// List of relay / crawler notification URLs. #[config(env = "CRAWLERS", parse_env = split_comma_list)] pub crawlers: Option>, diff --git a/crates/tranquil-pds/src/comms/service.rs b/crates/tranquil-pds/src/comms/service.rs index cb7640c..a983c02 100644 --- a/crates/tranquil-pds/src/comms/service.rs +++ b/crates/tranquil-pds/src/comms/service.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use std::time::Duration; use chrono::Utc; -use tokio::time::interval; + use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; use tranquil_comms::{ @@ -75,17 +75,28 @@ impl CommsService { ); } info!( - poll_interval_secs = self.poll_interval.as_secs(), + poll_interval_ms = self.poll_interval.as_millis() as u64, batch_size = self.batch_size, channels = ?self.senders.keys().collect::>(), "Starting comms service" ); - let mut ticker = interval(self.poll_interval); + let base = self.poll_interval; + let max_backoff = Duration::from_secs(30); + let mut current_delay = base; loop { tokio::select! { - _ = ticker.tick() => { - if let Err(e) = self.process_batch().await { - error!(error = %e, "Failed to process comms batch"); + _ = tokio::time::sleep(current_delay) => { + match self.process_batch().await { + Ok(had_work) => { + current_delay = match had_work { + true => base, + false => max_backoff.min(current_delay.saturating_mul(2)), + }; + } + Err(e) => { + error!(error = %e, "Failed to process comms batch"); + current_delay = max_backoff.min(current_delay.saturating_mul(2)); + } } } _ = shutdown.cancelled() => { @@ -96,14 +107,14 @@ impl CommsService { } } - async fn process_batch(&self) -> Result<(), tranquil_db_traits::DbError> { + async fn process_batch(&self) -> Result { let items = self.fetch_pending().await?; if items.is_empty() { - return Ok(()); + return Ok(false); } debug!(count = items.len(), "Processing comms batch"); futures::future::join_all(items.into_iter().map(|item| self.process_item(item))).await; - Ok(()) + Ok(true) } async fn fetch_pending(&self) -> Result, tranquil_db_traits::DbError> { diff --git a/crates/tranquil-pds/src/scheduled.rs b/crates/tranquil-pds/src/scheduled.rs index f4d6aa5..bd1b46a 100644 --- a/crates/tranquil-pds/src/scheduled.rs +++ b/crates/tranquil-pds/src/scheduled.rs @@ -667,6 +667,8 @@ async fn delete_account_data( Ok(()) } +const CAR_BLOCK_BATCH_SIZE: usize = 500; + pub async fn generate_repo_car( block_store: &AnyBlockStore, head_cid: &Cid, @@ -683,21 +685,20 @@ pub async fn generate_repo_car( }) .collect(); - let car_bytes = encode_car_header(head_cid).context("Failed to encode CAR header")?; + let mut car_bytes = encode_car_header(head_cid).context("Failed to encode CAR header")?; - let blocks = block_store - .get_many(&block_cids) - .await - .context("Failed to fetch blocks")?; + for chunk in block_cids.chunks(CAR_BLOCK_BATCH_SIZE) { + let blocks = block_store + .get_many(chunk) + .await + .context("Failed to fetch blocks")?; - let car_bytes = block_cids - .iter() - .zip(blocks.iter()) - .filter_map(|(cid, block_opt)| block_opt.as_ref().map(|block| (cid, block))) - .fold(car_bytes, |mut acc, (cid, block)| { - acc.extend(encode_car_block(cid, block)); - acc - }); + chunk + .iter() + .zip(blocks.iter()) + .filter_map(|(cid, block_opt)| block_opt.as_ref().map(|block| (cid, block))) + .for_each(|(cid, block)| car_bytes.extend(encode_car_block(cid, block))); + } Ok(car_bytes) } diff --git a/crates/tranquil-pds/src/state.rs b/crates/tranquil-pds/src/state.rs index bd2bc40..71b63c6 100644 --- a/crates/tranquil-pds/src/state.rs +++ b/crates/tranquil-pds/src/state.rs @@ -50,6 +50,7 @@ pub struct AppState { pub signal_sender: Option>, pub signal_store_provider: Option>, pub eventlog_segments_dir: Option, + pub repo_export_semaphore: Arc, } #[derive(Debug, Clone, Copy)] @@ -394,6 +395,9 @@ impl AppState { signal_sender: None, signal_store_provider, eventlog_segments_dir, + repo_export_semaphore: Arc::new(tokio::sync::Semaphore::new( + cfg.firehose.max_concurrent_repo_exports, + )), } } diff --git a/crates/tranquil-pds/tests/sync_resource_limits.rs b/crates/tranquil-pds/tests/sync_resource_limits.rs new file mode 100644 index 0000000..9a25803 --- /dev/null +++ b/crates/tranquil-pds/tests/sync_resource_limits.rs @@ -0,0 +1,147 @@ +mod common; +mod helpers; +use common::*; +use helpers::*; +use reqwest::StatusCode; +use std::sync::Once; + +static SET_SEMAPHORE: Once = Once::new(); + +fn ensure_low_semaphore() { + SET_SEMAPHORE.call_once(|| unsafe { + std::env::set_var("MAX_CONCURRENT_REPO_EXPORTS", "1"); + }); +} + +#[tokio::test] +async fn test_get_repo_succeeds_with_many_records() { + ensure_low_semaphore(); + let client = client(); + let (did, jwt) = setup_new_user("sync-batched-car").await; + + let create_futures = (0..20).map(|i| { + let client = &client; + let did = &did; + let jwt = &jwt; + async move { + create_post(client, did, jwt, &format!("Batch test post {}", i)).await; + } + }); + futures::future::join_all(create_futures).await; + + let res = client + .get(format!( + "{}/xrpc/com.atproto.sync.getRepo", + base_url().await + )) + .query(&[("did", did.as_str())]) + .send() + .await + .expect("Failed to send getRepo request"); + + assert_eq!(res.status(), StatusCode::OK); + assert_eq!( + res.headers() + .get("content-type") + .and_then(|h| h.to_str().ok()), + Some("application/vnd.ipld.car") + ); + let car_bytes = res.bytes().await.expect("Failed to read response body"); + assert!( + car_bytes.len() > 200, + "CAR with 20 records should have substantial data, got {} bytes", + car_bytes.len() + ); +} + +#[tokio::test] +async fn test_get_repo_semaphore_rejects_excess_concurrency() { + ensure_low_semaphore(); + let client = client(); + let (did, jwt) = setup_new_user("sync-semaphore").await; + + for i in 0..50 { + create_post(&client, &did, &jwt, &format!("Padding post {}", i)).await; + } + + let base = base_url().await; + let concurrent_requests = 10; + + let request_futures = (0..concurrent_requests).map(|_| { + let client = client.clone(); + let did = did.clone(); + async move { + client + .get(format!("{}/xrpc/com.atproto.sync.getRepo", base)) + .query(&[("did", did.as_str())]) + .send() + .await + .expect("Failed to send request") + .status() + } + }); + + let statuses: Vec = futures::future::join_all(request_futures).await; + let ok_count = statuses.iter().filter(|s| **s == StatusCode::OK).count(); + let rejected_count = statuses + .iter() + .filter(|s| **s == StatusCode::SERVICE_UNAVAILABLE) + .count(); + + assert!(ok_count >= 1, "at least one request should succeed"); + assert!( + rejected_count > 0, + "semaphore=1 with {} concurrent requests, expected some 503 rejections", + concurrent_requests + ); + assert!( + ok_count + rejected_count == statuses.len(), + "expected only 200 or 503 responses: {:?}", + statuses + ); +} + +#[tokio::test] +async fn test_get_repo_since_not_affected_by_semaphore() { + ensure_low_semaphore(); + let client = client(); + let (did, jwt) = setup_new_user("sync-since-no-sem").await; + create_post(&client, &did, &jwt, "First post").await; + + let latest_res = client + .get(format!( + "{}/xrpc/com.atproto.sync.getLatestCommit", + base_url().await + )) + .query(&[("did", did.as_str())]) + .send() + .await + .expect("Failed to get latest commit"); + let body: serde_json::Value = latest_res.json().await.unwrap(); + let rev = body["rev"].as_str().unwrap(); + + create_post(&client, &did, &jwt, "Second post").await; + + let base = base_url().await; + let request_futures = (0..10).map(|_| { + let client = client.clone(); + let did = did.clone(); + let rev = rev.to_string(); + async move { + client + .get(format!("{}/xrpc/com.atproto.sync.getRepo", base)) + .query(&[("did", did.as_str()), ("since", rev.as_str())]) + .send() + .await + .expect("Failed to send request") + .status() + } + }); + + let statuses: Vec = futures::future::join_all(request_futures).await; + assert!( + statuses.iter().all(|s| *s == StatusCode::OK), + "getRepo with since should bypass semaphore, got: {:?}", + statuses + ); +} diff --git a/crates/tranquil-store/benches/recovery.rs b/crates/tranquil-store/benches/recovery.rs index f9fb278..fd9b29d 100644 --- a/crates/tranquil-store/benches/recovery.rs +++ b/crates/tranquil-store/benches/recovery.rs @@ -100,9 +100,7 @@ impl<'a> DirectSeeder<'a> { } let loc = self.data_writer.append_block(cid, data).unwrap(); - self.hint_writer - .append_hint(cid, loc.file_id, loc.offset, loc.length) - .unwrap(); + self.hint_writer.append_hint(cid, &loc).unwrap(); self.blocks_in_file += 1; if self.blocks_in_file.is_multiple_of(10_000) { diff --git a/crates/tranquil-store/src/blockstore/compaction.rs b/crates/tranquil-store/src/blockstore/compaction.rs index 0a21ea8..33fc21b 100644 --- a/crates/tranquil-store/src/blockstore/compaction.rs +++ b/crates/tranquil-store/src/blockstore/compaction.rs @@ -162,13 +162,7 @@ fn stream_compact( } => match index.get(&cid_bytes) { Some(e) if e.location.file_id == source_file_id && !e.refcount.is_zero() => { let loc = writer.append_block(&cid_bytes, &data)?; - hint_writer.append_relocate( - &cid_bytes, - loc.file_id, - loc.offset, - loc.length, - e.refcount.raw(), - )?; + hint_writer.append_relocate(&cid_bytes, &loc, e.refcount.raw())?; relocations.push((cid_bytes, loc)); live_count = live_count.saturating_add(1); } @@ -188,13 +182,7 @@ fn stream_compact( } false => { let loc = writer.append_block(&cid_bytes, &data)?; - hint_writer.append_relocate( - &cid_bytes, - loc.file_id, - loc.offset, - loc.length, - e.refcount.raw(), - )?; + hint_writer.append_relocate(&cid_bytes, &loc, e.refcount.raw())?; relocations.push((cid_bytes, loc)); live_count = live_count.saturating_add(1); } diff --git a/crates/tranquil-store/src/blockstore/group_commit.rs b/crates/tranquil-store/src/blockstore/group_commit.rs index feb49aa..2c9012d 100644 --- a/crates/tranquil-store/src/blockstore/group_commit.rs +++ b/crates/tranquil-store/src/blockstore/group_commit.rs @@ -1136,7 +1136,7 @@ fn process_batch( } let loc = data_writer.append_block(cid_bytes, data)?; - hint_writer.append_hint(cid_bytes, loc.file_id, loc.offset, loc.length)?; + hint_writer.append_hint(cid_bytes, &loc)?; block_bytes = block_bytes.saturating_add(data.len() as u64); block_count = block_count.saturating_add(1); diff --git a/crates/tranquil-store/src/blockstore/hint.rs b/crates/tranquil-store/src/blockstore/hint.rs index a035763..e07150c 100644 --- a/crates/tranquil-store/src/blockstore/hint.rs +++ b/crates/tranquil-store/src/blockstore/hint.rs @@ -55,22 +55,26 @@ fn write_hint_record( io.write_all_at(fd, write_offset.raw(), record) } +fn encode_location_fields(record: &mut [u8; HINT_RECORD_SIZE], loc: &BlockLocation) { + record[FIELD_A_OFFSET..FIELD_A_OFFSET + 4].copy_from_slice(&loc.file_id.raw().to_le_bytes()); + record[FIELD_A_OFFSET + 4..FIELD_A_OFFSET + 8] + .copy_from_slice(&loc.length.raw().to_le_bytes()); + record[FIELD_B_OFFSET..FIELD_B_OFFSET + 8] + .copy_from_slice(&loc.offset.raw().to_le_bytes()); +} + pub(crate) fn encode_hint_record( io: &S, fd: FileId, write_offset: HintOffset, cid_bytes: &[u8; CID_SIZE], - file_id: DataFileId, - block_offset: BlockOffset, - length: BlockLength, + loc: &BlockLocation, ) -> io::Result<()> { let mut record = [0u8; HINT_RECORD_SIZE]; record[TYPE_OFFSET] = RECORD_TYPE_PUT; record[VERSION_OFFSET] = HINT_FORMAT_VERSION; record[CID_OFFSET..CID_OFFSET + CID_SIZE].copy_from_slice(cid_bytes); - record[FIELD_A_OFFSET..FIELD_A_OFFSET + 4].copy_from_slice(&file_id.raw().to_le_bytes()); - record[FIELD_A_OFFSET + 4..FIELD_A_OFFSET + 8].copy_from_slice(&length.raw().to_le_bytes()); - record[FIELD_B_OFFSET..FIELD_B_OFFSET + 8].copy_from_slice(&block_offset.raw().to_le_bytes()); + encode_location_fields(&mut record, loc); let checksum = hint_checksum(&record[..HINT_PAYLOAD_SIZE]); record[CHECKSUM_OFFSET..].copy_from_slice(&checksum.to_le_bytes()); @@ -85,9 +89,7 @@ pub(crate) fn encode_relocate_record( fd: FileId, write_offset: HintOffset, cid_bytes: &[u8; CID_SIZE], - file_id: DataFileId, - block_offset: BlockOffset, - length: BlockLength, + loc: &BlockLocation, refcount: u32, ) -> io::Result<()> { let mut record = [0u8; HINT_RECORD_SIZE]; @@ -96,9 +98,7 @@ pub(crate) fn encode_relocate_record( let rc16 = u16::try_from(refcount).unwrap_or(u16::MAX); record[REFCOUNT_OFFSET..REFCOUNT_OFFSET + 2].copy_from_slice(&rc16.to_le_bytes()); record[CID_OFFSET..CID_OFFSET + CID_SIZE].copy_from_slice(cid_bytes); - record[FIELD_A_OFFSET..FIELD_A_OFFSET + 4].copy_from_slice(&file_id.raw().to_le_bytes()); - record[FIELD_A_OFFSET + 4..FIELD_A_OFFSET + 8].copy_from_slice(&length.raw().to_le_bytes()); - record[FIELD_B_OFFSET..FIELD_B_OFFSET + 8].copy_from_slice(&block_offset.raw().to_le_bytes()); + encode_location_fields(&mut record, loc); let checksum = hint_checksum(&record[..HINT_PAYLOAD_SIZE]); record[CHECKSUM_OFFSET..].copy_from_slice(&checksum.to_le_bytes()); @@ -323,19 +323,9 @@ impl<'a, S: StorageIO> HintFileWriter<'a, S> { pub fn append_hint( &mut self, cid_bytes: &[u8; CID_SIZE], - file_id: DataFileId, - offset: BlockOffset, - length: BlockLength, + loc: &BlockLocation, ) -> io::Result<()> { - encode_hint_record( - self.io, - self.fd, - self.position, - cid_bytes, - file_id, - offset, - length, - )?; + encode_hint_record(self.io, self.fd, self.position, cid_bytes, loc)?; self.position = self.position.advance(HINT_RECORD_SIZE as u64); Ok(()) } @@ -354,21 +344,10 @@ impl<'a, S: StorageIO> HintFileWriter<'a, S> { pub fn append_relocate( &mut self, cid_bytes: &[u8; CID_SIZE], - file_id: DataFileId, - offset: BlockOffset, - length: BlockLength, + loc: &BlockLocation, refcount: u32, ) -> io::Result<()> { - encode_relocate_record( - self.io, - self.fd, - self.position, - cid_bytes, - file_id, - offset, - length, - refcount, - )?; + encode_relocate_record(self.io, self.fd, self.position, cid_bytes, loc, refcount)?; self.position = self.position.advance(HINT_RECORD_SIZE as u64); Ok(()) } @@ -862,7 +841,8 @@ mod tests { let offset = BlockOffset::new(1024); let length = BlockLength::new(256); - encode_hint_record(&sim, fd, HintOffset::new(0), &cid, file_id, offset, length).unwrap(); + let loc = BlockLocation { file_id, offset, length }; + encode_hint_record(&sim, fd, HintOffset::new(0), &cid, &loc).unwrap(); let file_size = sim.file_size(fd).unwrap(); let record = decode_hint_record(&sim, fd, HintOffset::new(0), file_size) @@ -920,16 +900,12 @@ mod tests { (0u8..5).for_each(|i| { let cid = test_cid(i); let write_offset = HintOffset::new(i as u64 * HINT_RECORD_SIZE as u64); - encode_hint_record( - &sim, - fd, - write_offset, - &cid, - DataFileId::new(i as u32), - BlockOffset::new(i as u64 * 100), - BlockLength::new(50 + i as u32), - ) - .unwrap(); + let loc = BlockLocation { + file_id: DataFileId::new(i as u32), + offset: BlockOffset::new(i as u64 * 100), + length: BlockLength::new(50 + i as u32), + }; + encode_hint_record(&sim, fd, write_offset, &cid, &loc).unwrap(); }); let file_size = sim.file_size(fd).unwrap(); @@ -971,16 +947,12 @@ mod tests { fn detects_corrupted_hint() { let (sim, fd) = setup(); let cid = test_cid(1); - encode_hint_record( - &sim, - fd, - HintOffset::new(0), - &cid, - DataFileId::new(0), - BlockOffset::new(0), - BlockLength::new(100), - ) - .unwrap(); + let loc = BlockLocation { + file_id: DataFileId::new(0), + offset: BlockOffset::new(0), + length: BlockLength::new(100), + }; + encode_hint_record(&sim, fd, HintOffset::new(0), &cid, &loc).unwrap(); sim.write_all_at(fd, 10, &[0xFF]).unwrap(); @@ -1006,16 +978,12 @@ mod tests { fn oversized_length_treated_as_corrupted() { let (sim, fd) = setup(); let cid = test_cid(1); - encode_hint_record( - &sim, - fd, - HintOffset::new(0), - &cid, - DataFileId::new(0), - BlockOffset::new(0), - BlockLength::new(100), - ) - .unwrap(); + let loc = BlockLocation { + file_id: DataFileId::new(0), + offset: BlockOffset::new(0), + length: BlockLength::new(100), + }; + encode_hint_record(&sim, fd, HintOffset::new(0), &cid, &loc).unwrap(); let length_offset = FIELD_A_OFFSET as u64 + 4; let oversized = (MAX_BLOCK_SIZE + 1).to_le_bytes(); @@ -1040,14 +1008,12 @@ mod tests { let mut writer = HintFileWriter::new(&sim, fd); (0u8..5).for_each(|i| { - writer - .append_hint( - &test_cid(i), - DataFileId::new(0), - BlockOffset::new(i as u64 * 100), - BlockLength::new(50 + i as u32), - ) - .unwrap(); + let loc = BlockLocation { + file_id: DataFileId::new(0), + offset: BlockOffset::new(i as u64 * 100), + length: BlockLength::new(50 + i as u32), + }; + writer.append_hint(&test_cid(i), &loc).unwrap(); }); assert_eq!( @@ -1074,25 +1040,21 @@ mod tests { fn hint_writer_resume_continues_at_position() { let (sim, fd) = setup(); let mut writer = HintFileWriter::new(&sim, fd); - writer - .append_hint( - &test_cid(0), - DataFileId::new(0), - BlockOffset::new(0), - BlockLength::new(100), - ) - .unwrap(); + let loc0 = BlockLocation { + file_id: DataFileId::new(0), + offset: BlockOffset::new(0), + length: BlockLength::new(100), + }; + writer.append_hint(&test_cid(0), &loc0).unwrap(); let pos = writer.position(); let mut writer2 = HintFileWriter::resume(&sim, fd, pos); - writer2 - .append_hint( - &test_cid(1), - DataFileId::new(0), - BlockOffset::new(100), - BlockLength::new(200), - ) - .unwrap(); + let loc1 = BlockLocation { + file_id: DataFileId::new(0), + offset: BlockOffset::new(100), + length: BlockLength::new(200), + }; + writer2.append_hint(&test_cid(1), &loc1).unwrap(); let reader = HintFileReader::open(&sim, fd).unwrap(); let valid_count = reader @@ -1115,14 +1077,12 @@ mod tests { fn hint_reader_stops_on_truncated() { let (sim, fd) = setup(); let mut writer = HintFileWriter::new(&sim, fd); - writer - .append_hint( - &test_cid(0), - DataFileId::new(0), - BlockOffset::new(0), - BlockLength::new(100), - ) - .unwrap(); + let loc = BlockLocation { + file_id: DataFileId::new(0), + offset: BlockOffset::new(0), + length: BlockLength::new(100), + }; + writer.append_hint(&test_cid(0), &loc).unwrap(); sim.write_all_at(fd, writer.position().raw(), &[0u8; HINT_RECORD_SIZE - 1]) .unwrap(); @@ -1140,14 +1100,12 @@ mod tests { let mut writer = HintFileWriter::new(&sim, fd); (0u8..3).for_each(|i| { - writer - .append_hint( - &test_cid(i), - DataFileId::new(0), - BlockOffset::new(i as u64 * 100), - BlockLength::new(50), - ) - .unwrap(); + let loc = BlockLocation { + file_id: DataFileId::new(0), + offset: BlockOffset::new(i as u64 * 100), + length: BlockLength::new(50), + }; + writer.append_hint(&test_cid(i), &loc).unwrap(); }); sim.write_all_at(fd, HINT_RECORD_SIZE as u64 + 5, &[0xFF]) diff --git a/crates/tranquil-store/tests/sim_blockstore.rs b/crates/tranquil-store/tests/sim_blockstore.rs index bc8e4d4..7852e4a 100644 --- a/crates/tranquil-store/tests/sim_blockstore.rs +++ b/crates/tranquil-store/tests/sim_blockstore.rs @@ -75,7 +75,7 @@ impl SimHarness { let data = vec![seed as u8; data_size]; let loc = writer.append_block(&cid, &data).unwrap(); hint_writer - .append_hint(&cid, loc.file_id, loc.offset, loc.length) + .append_hint(&cid, &loc) .unwrap(); (cid, loc) }) @@ -539,7 +539,7 @@ fn sim_aggressive_faults_data_integrity() { let data = vec![i as u8; 64]; let loc = writer.append_block(&cid, &data).ok()?; hint_writer - .append_hint(&cid, loc.file_id, loc.offset, loc.length) + .append_hint(&cid, &loc) .ok()?; Some(()) })?; diff --git a/crates/tranquil-sync/src/repo.rs b/crates/tranquil-sync/src/repo.rs index 322a706..c3aa692 100644 --- a/crates/tranquil-sync/src/repo.rs +++ b/crates/tranquil-sync/src/repo.rs @@ -142,6 +142,17 @@ pub async fn get_repo( return get_repo_since(&state, &did, &head_cid, since).await; } + let _permit = match state.repo_export_semaphore.try_acquire() { + Ok(permit) => permit, + Err(_) => { + return ( + StatusCode::SERVICE_UNAVAILABLE, + "Too many concurrent repo exports", + ) + .into_response(); + } + }; + let car_bytes = match generate_repo_car_from_user_blocks( state.repos.repo.as_ref(), &state.block_store, @@ -213,19 +224,24 @@ async fn get_repo_since(state: &AppState, did: &Did, head_cid: &Cid, since: &str .into_response(); } - let blocks = match state.block_store.get_many(&block_cids).await { - Ok(b) => b, - Err(e) => { - error!("Block store error in get_repo_since: {:?}", e); - return ApiError::InternalError(Some("Failed to get blocks".into())).into_response(); - } - }; + for chunk_start in (0..block_cids.len()).step_by(500) { + let chunk_end = (chunk_start + 500).min(block_cids.len()); + let chunk = &block_cids[chunk_start..chunk_end]; + let blocks = match state.block_store.get_many(chunk).await { + Ok(b) => b, + Err(e) => { + error!("Block store error in get_repo_since: {:?}", e); + return ApiError::InternalError(Some("Failed to get blocks".into())) + .into_response(); + } + }; - blocks - .into_iter() - .enumerate() - .filter_map(|(i, block_opt)| block_opt.map(|block| (block_cids[i], block))) - .for_each(|(cid, block)| car_bytes.extend_from_slice(&encode_car_block(&cid, &block))); + chunk + .iter() + .zip(blocks.into_iter()) + .filter_map(|(cid, block_opt)| block_opt.map(|block| (*cid, block))) + .for_each(|(cid, block)| car_bytes.extend_from_slice(&encode_car_block(&cid, &block))); + } ( StatusCode::OK,