From 14540b2d4141f83513e337622065a98e36590e8a Mon Sep 17 00:00:00 2001 From: Defnull <879658+define-null@users.noreply.github.com> Date: Tue, 12 May 2026 15:53:57 +0200 Subject: [PATCH 1/5] chore: When the response is empty return query bound as a last-block In order to be able to detouch chunk ids from the data they reference the worker code should not parse the chunk id format to obtain the last_block. It may not match with the actual data in the chunk, and that prohibits chunk-id format changes. --- src/controller/p2p.rs | 3 ++- src/controller/worker.rs | 42 ++++++++++++++-------------------------- 2 files changed, 17 insertions(+), 28 deletions(-) diff --git a/src/controller/p2p.rs b/src/controller/p2p.rs index bf8ee90d..de224c6b 100644 --- a/src/controller/p2p.rs +++ b/src/controller/p2p.rs @@ -495,7 +495,8 @@ impl + Send + 'static> P2PController, + block_range: (u64, u64), chunk_id: &str, client_id: Option, query_type: QueryType, @@ -112,22 +109,17 @@ impl Worker { &self, query_str: &str, dataset: Dataset, - block_range: Option<(u64, u64)>, + block_range: (u64, u64), chunk_id: &str, ) -> QueryResult { - let Ok(chunk) = chunk_id.parse::() else { - return Err(QueryError::BadRequest(format!( - "Can't parse chunk id '{chunk_id}'" - ))); - }; let mut query = sqd_query::Query::from_json_bytes(query_str.as_bytes()) .map_err(|e| QueryError::BadRequest(format!("Couldn't parse query: {e:?}")))?; - if let Some((from_block, to_block)) = block_range { - query.set_first_block(from_block); - query.set_last_block(Some(to_block)); - } + let (from_block, to_block) = block_range; + + query.set_first_block(from_block); + query.set_last_block(Some(to_block)); - let Some(chunk_guard) = self.state_manager.clone().get_chunk(dataset, chunk.clone()) else { + let Some(chunk_guard) = self.state_manager.clone().get_chunk(dataset, chunk_id) else { return Err(QueryError::NotFound); }; @@ -148,11 +140,12 @@ impl Worker { writer.write_blocks(&mut blocks)?; blocks.last_block() } else { - if let Some(last_query_block) = query.last_block() { - std::cmp::min(last_query_block, chunk.last_block.into()) - } else { - chunk.last_block.into() - } + // No matching rows in this chunk. We used to fall back to + // the chunk's last_block (parsed from the chunk_id) so the + // portal could see progress. After NET-385 the worker + // treats chunk_id as opaque, so we report the query's + // upper bound + to_block }; let bytes = writer.finish()?; let serialization_duration = serialization_timer.elapsed(); @@ -188,11 +181,6 @@ impl Worker { dataset: Dataset, chunk_id: &str, ) -> QueryResult { - let Ok(chunk) = chunk_id.parse::() else { - return Err(QueryError::BadRequest(format!( - "Can't parse chunk id '{chunk_id}'" - ))); - }; let Ok(query_bytes) = base64.decode(query_str) else { return Err(QueryError::BadRequest(format!( "Can't decode plan '{query_str}'" @@ -207,7 +195,7 @@ impl Worker { let Some(chunk_guard) = self .state_manager .clone() - .get_chunk(dataset.clone(), chunk.clone()) + .get_chunk(dataset.clone(), chunk_id) else { return Err(QueryError::NotFound); }; From 3e3a80e97b687fd4f710f073f3650c68a96a9e8d Mon Sep 17 00:00:00 2001 From: Defnull <879658+define-null@users.noreply.github.com> Date: Wed, 13 May 2026 14:58:46 +0200 Subject: [PATCH 2/5] feat(NET-385): treat chunk id as opaque ChunkRef.chunk becomes an opaque Arc; DataChunk is reduced to a layout parser. The on-disk format gains an optional trailing suffix so multiple chunks can cover the same block range. --- Cargo.lock | 12 ++- Cargo.toml | 12 +-- src/storage/datasets_index.rs | 68 ++++++--------- src/storage/downloader.rs | 15 +--- src/storage/layout.rs | 154 +++++++++++++++++++--------------- src/storage/manager.rs | 63 +++++++------- src/storage/mod.rs | 26 +----- src/storage/state.rs | 23 ++--- src/storage/tests.rs | 112 +++++++++++++++++++++++++ src/types/state.rs | 7 +- 10 files changed, 289 insertions(+), 203 deletions(-) create mode 100644 src/storage/tests.rs diff --git a/Cargo.lock b/Cargo.lock index 06b2a63f..4983b788 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7216,17 +7216,21 @@ dependencies = [ [[package]] name = "sqd-assignments" version = "0.1.0" -source = "git+https://github.com/subsquid/sqd-network.git?rev=a1b685d#a1b685d9b0b78417a3e152303f0e2876285171da" +source = "git+https://github.com/subsquid/sqd-network.git?rev=938ffe1#938ffe1f59e7db93bf704a3301c03f611c0c5747" dependencies = [ "anyhow", + "base64 0.22.1", "crypto_box", + "curve25519-dalek", "flatbuffers 25.2.10", + "hmac", "libp2p-identity", "ouroboros", "serde", "serde_json", "sha2", "tracing", + "url", ] [[package]] @@ -7240,7 +7244,7 @@ dependencies = [ [[package]] name = "sqd-contract-client" version = "1.3.0" -source = "git+https://github.com/subsquid/sqd-network.git?rev=a1b685d#a1b685d9b0b78417a3e152303f0e2876285171da" +source = "git+https://github.com/subsquid/sqd-network.git?rev=938ffe1#938ffe1f59e7db93bf704a3301c03f611c0c5747" dependencies = [ "async-trait", "clap", @@ -7260,7 +7264,7 @@ dependencies = [ [[package]] name = "sqd-messages" version = "2.1.0" -source = "git+https://github.com/subsquid/sqd-network.git?rev=a1b685d#a1b685d9b0b78417a3e152303f0e2876285171da" +source = "git+https://github.com/subsquid/sqd-network.git?rev=938ffe1#938ffe1f59e7db93bf704a3301c03f611c0c5747" dependencies = [ "bytemuck", "flate2", @@ -7276,7 +7280,7 @@ dependencies = [ [[package]] name = "sqd-network-transport" version = "3.0.0" -source = "git+https://github.com/subsquid/sqd-network.git?rev=a1b685d#a1b685d9b0b78417a3e152303f0e2876285171da" +source = "git+https://github.com/subsquid/sqd-network.git?rev=938ffe1#938ffe1f59e7db93bf704a3301c03f611c0c5747" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index 7b2a01d1..3ca8b603 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,7 +7,7 @@ edition = "2021" [features] query-tracing = ["sqd-query/max_level_trace"] -[dependencies] +[dependencies] anyhow = "1.0" async-compression = { version = "0.4.27", features = ["gzip", "tokio"] } async-stream = "0.3.5" @@ -59,11 +59,11 @@ tracing-subscriber = { version = "0.3.18", features = ["env-filter"] } url = "2.5.2" walkdir = "2.5.0" zstd = "0.13" - -sqd-assignments = { git = "https://github.com/subsquid/sqd-network.git", rev = "a1b685d", features = ["reader"] } -sqd-contract-client = { git = "https://github.com/subsquid/sqd-network.git", rev = "a1b685d", version = "1.2.1" } -sqd-messages = { git = "https://github.com/subsquid/sqd-network.git", rev = "a1b685d", version = "2.0.2", features = ["bitstring"] } -sqd-network-transport = { git = "https://github.com/subsquid/sqd-network.git", rev = "a1b685d", version = "3.0.0", features = ["worker", "metrics"] } + +sqd-assignments = { git = "https://github.com/subsquid/sqd-network.git", rev = "938ffe1", features = ["reader", "builder"] } +sqd-contract-client = { git = "https://github.com/subsquid/sqd-network.git", rev = "938ffe1", version = "1.2.1" } +sqd-messages = { git = "https://github.com/subsquid/sqd-network.git", rev = "938ffe1", version = "2.0.2", features = ["bitstring"] } +sqd-network-transport = { git = "https://github.com/subsquid/sqd-network.git", rev = "938ffe1", version = "3.0.0", features = ["worker", "metrics"] } sqd-query = { git = "https://github.com/subsquid/data.git", rev = "9db54d3", features = ["parquet"] } sqd-polars = { git = "https://github.com/subsquid/data.git", rev = "9db54d3" } diff --git a/src/storage/datasets_index.rs b/src/storage/datasets_index.rs index c1b4ab84..465b0d50 100644 --- a/src/storage/datasets_index.rs +++ b/src/storage/datasets_index.rs @@ -1,22 +1,19 @@ use std::{collections::HashMap, str::FromStr, sync::Arc}; use reqwest::Url; -use sqd_network_transport::{Keypair, PeerId}; +use sqd_network_transport::Keypair; use tracing::error; -use crate::types::{ - dataset::Dataset, - state::{ChunkRef, ChunkSet}, -}; - -use super::layout::DataChunk; +use crate::types::state::ChunkRef; +use sqd_assignments::ChunkRef as ChunkAssignmentRef; pub struct DatasetsIndex { assignment: sqd_assignments::Assignment, assignment_id: String, - peer_id: PeerId, status: sqd_assignments::WorkerStatus, http_headers: reqwest::header::HeaderMap, + // chunks assigned to this worker + chunks: HashMap, } #[derive(Debug, PartialEq, Eq)] @@ -26,11 +23,12 @@ pub struct RemoteFile { } impl DatasetsIndex { - pub fn list_files(&self, dataset: &Dataset, chunk: &DataChunk) -> Option> { - let chunk = self - .assignment - .find_chunk(dataset, *chunk.first_block) - .ok()?; + /// Returns the remote files (URL + filename) associated with the given + /// chunk, or `None` if the chunk is not in the assignment or any URL + /// fails to parse. + pub fn list_files(&self, chunk: &ChunkRef) -> Option> { + let chunk_ref = self.chunks.get(chunk)?; + let chunk = self.assignment.get_chunk(*chunk_ref)?; let base_url = Url::from_str(&chunk.dataset_base_url()) .inspect_err(|e| { tracing::warn!( @@ -81,42 +79,25 @@ impl DatasetsIndex { }) .collect(); + let mut chunks = HashMap::new(); + let mut pool = StringPool::default(); + for (chunk_ref, chunk) in worker.iter_chunks_with_ref() { + let key = ChunkRef { + dataset: pool.get(chunk.dataset_id()), + chunk: Arc::from(chunk.id().to_string()), + }; + chunks.insert(key, chunk_ref); + } + Ok(Self { status: worker.status(), assignment, assignment_id: id.into(), - peer_id, http_headers, + chunks, }) } - pub fn create_chunks_set(&self) -> ChunkSet { - let mut chunk_set = ChunkSet::new(); - let Some(worker) = self.assignment.get_worker(&self.peer_id) else { - return chunk_set; - }; - let mut pool = StringPool::default(); - for chunk in worker.iter_chunks() { - match DataChunk::from_str(chunk.id()) { - Ok(id) => { - let chunk = ChunkRef { - dataset: pool.get(chunk.dataset_id()), - chunk: id, - }; - if let Some(last) = chunk_set.last() { - debug_assert!( - last < &chunk, - "Assigned chunks are not sorted: {last} >= {chunk}" - ); - } - chunk_set.insert(chunk); - } - Err(e) => tracing::warn!("Couldn't parse chunk id {}: {e}", chunk.id()), - } - } - chunk_set - } - pub fn status(&self) -> sqd_assignments::WorkerStatus { self.status } @@ -128,6 +109,10 @@ impl DatasetsIndex { pub fn assignment_id(&self) -> &str { &self.assignment_id } + + pub fn chunks(&self) -> &HashMap { + &self.chunks + } } #[derive(Default)] @@ -159,3 +144,4 @@ fn test_url_joining() { .unwrap(); assert_eq!(url.as_str(), "https://eclipse-testnet-2.sqd-datasets.io/0086800000/0089600001-0089800000-cg1JNYDM/blocks.parquet"); } + diff --git a/src/storage/downloader.rs b/src/storage/downloader.rs index 82ba3ae9..bd4a1c5c 100644 --- a/src/storage/downloader.rs +++ b/src/storage/downloader.rs @@ -11,11 +11,7 @@ use tracing::instrument; use crate::{cli, types::state::ChunkRef}; -use super::{ - datasets_index::{DatasetsIndex, RemoteFile}, - guard::FsGuard, - local_fs::add_temp_prefix, -}; +use super::{datasets_index::RemoteFile, guard::FsGuard, local_fs::add_temp_prefix}; const START_DELAY: Duration = Duration::from_millis(100); @@ -51,7 +47,8 @@ impl ChunkDownloader { &mut self, chunk: ChunkRef, dst: PathBuf, - datasets_index: &DatasetsIndex, + files: Vec, + headers: reqwest::header::HeaderMap, ) { let cancel_token = CancellationToken::new(); @@ -62,13 +59,7 @@ impl ChunkDownloader { panic!("Chunk {chunk} is already being downloaded"); } - let files = datasets_index - .list_files(&chunk.dataset, &chunk.chunk) - .unwrap_or_else(|| { - panic!("Dataset {} not found", chunk.dataset); - }); let num_files = files.len(); - let headers = datasets_index.get_headers().clone(); let client = self.reqwest_client.clone(); let current_delay = self.current_delay; let s3_timeout = self.args.s3_timeout; diff --git a/src/storage/layout.rs b/src/storage/layout.rs index 50cd771d..4fbd956a 100644 --- a/src/storage/layout.rs +++ b/src/storage/layout.rs @@ -58,43 +58,36 @@ impl Deref for BlockNumber { } } -#[derive(Default, PartialEq, Eq, Clone, Hash)] +/// On-disk chunk identity. +/// +/// The original chunk path format was `/--`, e.g. +/// `0000001000/0000001024-0000002047-0xabcdef`. It has been extended with an +/// optional trailing suffix so multiple chunks can cover the same block +/// range, e.g. `0000001000/0000001024-0000002047-0xabcdef`. +#[derive(PartialEq, Eq, PartialOrd, Ord, Clone, Hash)] pub struct DataChunk { - pub last_block: BlockNumber, + pub id: String, pub first_block: BlockNumber, - pub last_hash: String, - pub top: BlockNumber, + pub last_block: BlockNumber, } impl DataChunk { - pub fn path(&self) -> String { - format!( - "{}/{}-{}-{}", - self.top, self.first_block, self.last_block, self.last_hash - ) - } - // TODO: synchronize with other language implementations pub fn from_path(dirname: &str) -> Result { lazy_static! { - static ref RE: Regex = Regex::new(r"(\d{10})/(\d{10})-(\d{10})-(\w{5,8})$").unwrap(); + static ref RE: Regex = + Regex::new(r"((?:\d{10})/(\d{10})-(\d{10})-(?:\w{5,8}).*)$").unwrap(); } - let (top, beg, end, hash) = RE + let captures = RE .captures(dirname) - .and_then( - |cap| match (cap.get(1), cap.get(2), cap.get(3), cap.get(4)) { - (Some(top), Some(beg), Some(end), Some(hash)) => { - Some((top.as_str(), beg.as_str(), end.as_str(), hash.as_str())) - } - _ => None, - }, - ) .ok_or_else(|| anyhow!("Could not parse chunk dirname '{dirname}'"))?; + let id = captures.get(1).unwrap().as_str().to_owned(); + let first_block = BlockNumber::try_from(captures.get(2).unwrap().as_str())?; + let last_block = BlockNumber::try_from(captures.get(3).unwrap().as_str())?; Ok(Self { - first_block: BlockNumber::try_from(beg)?, - last_block: BlockNumber::try_from(end)?, - last_hash: hash.into(), - top: BlockNumber::try_from(top)?, + id, + first_block, + last_block, }) } } @@ -109,7 +102,7 @@ impl FromStr for DataChunk { impl std::fmt::Display for DataChunk { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "{}", self.path()) + write!(f, "{}", self.id) } } @@ -119,18 +112,6 @@ impl std::fmt::Debug for DataChunk { } } -impl Ord for DataChunk { - fn cmp(&self, other: &Self) -> std::cmp::Ordering { - self.last_block.cmp(&other.last_block) - } -} - -impl PartialOrd for DataChunk { - fn partial_cmp(&self, other: &Self) -> Option { - Some(self.cmp(other)) - } -} - #[instrument(skip_all, level = "debug")] async fn list_top_dirs(fs: &impl Filesystem) -> Result> { let mut entries: Vec<_> = fs @@ -189,7 +170,11 @@ pub async fn read_all_chunks(fs: &impl Filesystem) -> Result> { } } for (cur, next) in chunks.iter().tuple_windows() { - if cur.last_block >= next.first_block { + // Two chunks sharing the exact same range are allowed — + // suffix-distinguished forks. Anything else overlapping bails. + let same_range = cur.first_block == next.first_block + && cur.last_block == next.last_block; + if !same_range && cur.last_block >= next.first_block { bail!("Overlapping ranges: {} and {}", cur, next); } } @@ -243,25 +228,17 @@ mod tests { #[test] fn test_data_chunk() { - let chunk0 = DataChunk { - first_block: 1024.into(), - last_block: 2047.into(), - last_hash: "0xabcdef".into(), - top: 1000.into(), - }; let path = "0000001000/0000001024-0000002047-0xabcdef"; - assert_eq!(chunk0.path(), path); - assert_eq!(DataChunk::from_path(&path).unwrap(), chunk0); - - let chunk1 = DataChunk { - first_block: 221000000.into(), - last_block: 221000649.into(), - last_hash: "9QgFD".into(), - top: 221000000.into(), - }; + let chunk0 = DataChunk::from_path(path).unwrap(); + assert_eq!(&*chunk0.id, path); + assert_eq!(chunk0.first_block, 1024.into()); + assert_eq!(chunk0.last_block, 2047.into()); + let path = "0221000000/0221000000-0221000649-9QgFD"; - assert_eq!(chunk1.path(), path); - assert_eq!(DataChunk::from_path(&path).unwrap(), chunk1); + let chunk1 = DataChunk::from_path(path).unwrap(); + assert_eq!(&*chunk1.id, path); + assert_eq!(chunk1.first_block, 221000000.into()); + assert_eq!(chunk1.last_block, 221000649.into()); } #[tokio::test] @@ -290,34 +267,29 @@ mod tests { chunks, vec![ DataChunk { - top: 1000.into(), + id: "0000001000/0000001000-0000001999-0xabcdef".to_owned(), first_block: 1000.into(), last_block: 1999.into(), - last_hash: "0xabcdef".to_owned() }, DataChunk { - top: 1000.into(), + id: "0000001000/0000002000-0000002999-0x191919".to_owned(), first_block: 2000.into(), last_block: 2999.into(), - last_hash: "0x191919".to_owned() }, DataChunk { - top: 1000.into(), + id: "0000001000/0000003000-0000003999-0xdedede".to_owned(), first_block: 3000.into(), last_block: 3999.into(), - last_hash: "0xdedede".to_owned() }, DataChunk { - top: 4000.into(), + id: "0000004000/0000004000-0000004999-0xaaaaaa".to_owned(), first_block: 4000.into(), last_block: 4999.into(), - last_hash: "0xaaaaaa".to_owned() }, DataChunk { - top: 4000.into(), + id: "0000004000/1000000000-1000999999-0xbbbbbb".to_owned(), first_block: 1000000000.into(), last_block: 1000999999.into(), - last_hash: "0xbbbbbb".to_owned() }, ] ); @@ -332,4 +304,54 @@ mod tests { vec![DataChunk::from_path("0017881390/0017881390-0017882786-32ee9457").unwrap()] ); } + + #[tokio::test] + async fn test_chunks_with_same_block_range() { + let chunk_a_id = "0000000000/0000000000-0000001000-abcdef12"; + let chunk_b_id = "0000000000/0000000000-0000001000-abcdef12-fork"; + + let fs = TestFilesystem { + files: HashMap::from([( + "0000000000".into(), + vec![chunk_a_id.into(), chunk_b_id.into()], + )]), + }; + + let chunks = read_all_chunks(&fs) + .await + .expect("layout should accept both chunks"); + + assert_eq!(chunks.len(), 2); + let ids: std::collections::HashSet<&str> = + chunks.iter().map(|c| c.id.as_str()).collect(); + assert!(ids.contains(chunk_a_id)); + assert!(ids.contains(chunk_b_id)); + + for chunk in &chunks { + assert_eq!(chunk.first_block, 0u64.into()); + assert_eq!(chunk.last_block, 1000u64.into()); + } + } + + #[tokio::test] + async fn test_chunks_with_partial_overlap_rejected() { + // Ranges share blocks but aren't identical — the suffix exception + // shouldn't apply. + let fs = TestFilesystem { + files: HashMap::from([( + "0000000000".into(), + vec![ + "0000000000/0000000000-0000001000-abcdef12".into(), + "0000000000/0000000500-0000001500-bbbbbbbb".into(), + ], + )]), + }; + let err = read_all_chunks(&fs) + .await + .expect_err("partial overlap should be rejected"); + assert!( + err.to_string().contains("Overlapping ranges"), + "unexpected error: {err}" + ); + } } diff --git a/src/storage/manager.rs b/src/storage/manager.rs index aef71feb..26e42f8a 100644 --- a/src/storage/manager.rs +++ b/src/storage/manager.rs @@ -105,14 +105,18 @@ impl StateManager { } let guard = self.datasets_index.lock(); - let Some(index) = guard.as_ref() else { + let Some(dataset_index) = guard.as_ref() else { continue; }; while downloader.download_count() < self.concurrent_downloads { - if let Some(chunk) = self.state.lock().take_next_download() { - info!("Downloading chunk {chunk}"); - let dst = self.chunk_path(&chunk); - downloader.start_download(chunk, dst, &index); + if let Some(chunk_ref) = self.state.lock().take_next_download() { + info!("Downloading chunk {chunk_ref}"); + let dst = self.chunk_path(&chunk_ref); + let files = dataset_index + .list_files(&chunk_ref) + .unwrap_or_else(|| panic!("Dataset {} not found", chunk_ref.dataset)); + let headers = dataset_index.get_headers().clone(); + downloader.start_download(chunk_ref, dst, files, headers); } else { break; } @@ -161,21 +165,6 @@ impl StateManager { } } - // TODO: prevent accidental massive removals - #[instrument(skip_all)] - fn set_desired_chunks(&self, desired_chunks: ChunkSet, datasets_index: DatasetsIndex) { - let mut index = self.datasets_index.lock(); - let mut state = self.state.lock(); - match state.set_desired_chunks(desired_chunks) { - UpdateStatus::Unchanged => {} - UpdateStatus::Updated => { - info!("Got new assignment"); - self.notify.notify_one(); - } - } - *index = Some(datasets_index); - } - pub fn set_assignment( &self, assignment: sqd_assignments::Assignment, @@ -191,8 +180,19 @@ impl StateManager { } }; let status = datasets_index.status(); - let chunks = datasets_index.create_chunks_set(); - self.set_desired_chunks(chunks, datasets_index); + let chunks: ChunkSet = datasets_index.chunks().keys().cloned().collect(); + + let mut index = self.datasets_index.lock(); + let mut state = self.state.lock(); + + match state.set_desired_chunks(chunks) { + UpdateStatus::Unchanged => {} + UpdateStatus::Updated => { + info!("Got new assignment"); + self.notify.notify_one(); + } + } + *index = Some(datasets_index); match status { sqd_assignments::WorkerStatus::Ok => { @@ -223,17 +223,20 @@ impl StateManager { } } + /// Returns the on-disk path to a locally available chunk, or `None` if + /// the chunk isn't present. The chunk is reference-counted for the + /// lifetime of the returned guard — it won't be evicted by the state + /// manager until every guard for it is dropped. pub fn get_chunk( self: Arc, dataset: Dataset, - chunk: DataChunk, + chunk_id: &str, ) -> Option> { - let encoded_dataset = dataset::encode_dataset(&dataset); let chunk = self .state .lock() - .get_and_lock_chunk(Arc::new(dataset), chunk)?; - let path = self.fs.root.join(encoded_dataset).join(chunk.chunk.path()); + .get_and_lock_chunk(Arc::new(dataset), Arc::from(chunk_id.to_string()))?; + let path = self.chunk_path(&chunk); let guard = scopeguard::guard(path, move |_| self.state.lock().unlock_chunk(&chunk)); Some(guard) } @@ -248,11 +251,11 @@ impl StateManager { Ok(()) } - fn chunk_path(&self, chunk: &ChunkRef) -> PathBuf { + fn chunk_path(&self, chunk_ref: &ChunkRef) -> PathBuf { self.fs .root - .join(dataset::encode_dataset(&chunk.dataset)) - .join(chunk.chunk.path()) + .join(dataset::encode_dataset(&chunk_ref.dataset)) + .join(&chunk_ref.chunk.as_str()) } } @@ -289,7 +292,7 @@ async fn load_state(fs: &LocalFs) -> Result { for chunk in chunks { result.insert(ChunkRef { dataset: dataset.clone(), - chunk, + chunk: Arc::from(chunk.id), }); } } else { diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 7f509a7c..31d520a3 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -20,28 +20,4 @@ pub trait Filesystem { } #[cfg(test)] -pub mod tests { - use camino::{Utf8Path as Path, Utf8PathBuf as PathBuf}; - use std::collections::HashMap; - - use anyhow::Context; - - use super::Filesystem; - - pub struct TestFilesystem { - pub files: HashMap>, - } - - impl Filesystem for TestFilesystem { - async fn ls_root(&self) -> anyhow::Result> { - Ok(self.files.keys().cloned().collect()) - } - - async fn ls(&self, path: impl AsRef) -> anyhow::Result> { - self.files - .get(path.as_ref()) - .cloned() - .with_context(|| format!("Couldn't find top dir {}", path.as_ref())) - } - } -} +pub mod tests; diff --git a/src/storage/state.rs b/src/storage/state.rs index ee53fa34..4820cc7e 100644 --- a/src/storage/state.rs +++ b/src/storage/state.rs @@ -1,14 +1,10 @@ use itertools::Itertools; -use std::{collections::BTreeMap, sync::Arc}; +use std::collections::HashMap; use tracing::{info, instrument}; -use super::layout::DataChunk; use crate::{ metrics, - types::{ - dataset::Dataset, - state::{ChunkRef, ChunkSet}, - }, + types::state::{ChunkId, ChunkRef, ChunkSet, DatasetId}, }; #[derive(Debug, Default)] @@ -17,7 +13,7 @@ pub struct State { downloading: ChunkSet, // available and downloading don't intersect desired: ChunkSet, to_download: ChunkSet, // to_download is always equal to desired.diff(available).diff(downloading) - locks: BTreeMap, // stores ref count for each chunk + locks: HashMap, // stores ref count for each chunk } #[derive(Debug)] @@ -120,11 +116,7 @@ impl State { } } - pub fn get_and_lock_chunk( - &mut self, - dataset: Arc, - chunk: DataChunk, - ) -> Option { + pub fn get_and_lock_chunk(&mut self, dataset: DatasetId, chunk: ChunkId) -> Option { let chunk_ref = self.available.get(&ChunkRef { dataset, chunk }).cloned(); if let Some(chunk_ref) = chunk_ref.as_ref() { @@ -183,7 +175,7 @@ mod tests { use itertools::Itertools; - use crate::{storage::layout::DataChunk, types::state::ChunkRef}; + use crate::types::state::ChunkRef; use super::State; @@ -192,12 +184,11 @@ mod tests { let ds = Arc::new("ds".to_owned()); let chunk_ref = |x| ChunkRef { dataset: ds.clone(), - chunk: DataChunk::from_path(&format!( + chunk: Arc::new(format!( "0000000000/000000000{}-000000000{}-00000000", x, x + 1 - )) - .unwrap(), + )), }; let a = chunk_ref(0); let b = chunk_ref(1); diff --git a/src/storage/tests.rs b/src/storage/tests.rs new file mode 100644 index 00000000..bc62c34c --- /dev/null +++ b/src/storage/tests.rs @@ -0,0 +1,112 @@ +use camino::{Utf8Path as Path, Utf8PathBuf as PathBuf}; +use std::collections::HashMap; + +use anyhow::Context; + +use super::Filesystem; + +pub struct TestFilesystem { + pub files: HashMap>, +} + +impl Filesystem for TestFilesystem { + async fn ls_root(&self) -> anyhow::Result> { + Ok(self.files.keys().cloned().collect()) + } + + async fn ls(&self, path: impl AsRef) -> anyhow::Result> { + self.files + .get(path.as_ref()) + .cloned() + .with_context(|| format!("Couldn't find top dir {}", path.as_ref())) + } +} + +#[test] +fn test_chunks_with_same_block_range() { + use sqd_assignments::AssignmentBuilder; + use sqd_network_transport::Keypair; + + use super::datasets_index::DatasetsIndex; + + let chunk_a_id = "0000000000/0000000000-0000001000-abcdef12"; + let chunk_b_id = "0000000000/0000000000-0000001000-abcdef12-fork"; + + let mut builder = AssignmentBuilder::new("test-secret").check_continuity(false); + + builder + .new_chunk() + .id(chunk_a_id) + .dataset_id("test-dataset") + .dataset_base_url("https://example.com/") + .block_range(0..=1000) + .size(1) + .worker_indexes(&[0]) + .files(&["blocks.parquet".to_owned()]) + .finish() + .unwrap(); + + // Same block range as the first chunk, distinguished only by the trailing + // suffix. `add_chunk` returns Err on duplicate range, but with + // `check_continuity(false)` the chunk is still added to the buffer. + let _ = builder + .new_chunk() + .id(chunk_b_id) + .dataset_id("test-dataset") + .dataset_base_url("https://example.com/") + .block_range(0..=1000) + .size(1) + .worker_indexes(&[0]) + .files(&["blocks.parquet".to_owned()]) + .finish(); + + builder.finish_dataset(); + + let keypair = Keypair::generate_ed25519(); + let peer_id = keypair.public().to_peer_id(); + builder.add_worker(peer_id, sqd_assignments::WorkerStatus::Ok, &[0, 1]); + + let bytes = builder.finish(); + let assignment = sqd_assignments::Assignment::from_owned(bytes).unwrap(); + + let index = DatasetsIndex::new(assignment, "test-asgn", &keypair).unwrap(); + + assert_eq!( + index.chunks().len(), + 2, + "both suffix-distinguished chunks should be present in the index" + ); + + let chunk_a = index + .chunks() + .keys() + .find(|cr| cr.chunk.as_str() == chunk_a_id) + .cloned() + .expect("chunk A should be in the index"); + let chunk_b = index + .chunks() + .keys() + .find(|cr| cr.chunk.as_str() == chunk_b_id) + .cloned() + .expect("chunk B should be in the index"); + + let files_a = index + .list_files(&chunk_a) + .expect("list_files for chunk A should succeed"); + let files_b = index + .list_files(&chunk_b) + .expect("list_files for chunk B should succeed"); + + assert_eq!(files_a.len(), 1); + assert_eq!(files_b.len(), 1); + assert_eq!(files_a[0].name, "blocks.parquet"); + assert_eq!(files_b[0].name, "blocks.parquet"); + assert_eq!( + files_a[0].url.as_str(), + "https://example.com/0000000000/0000000000-0000001000-abcdef12/blocks.parquet" + ); + assert_eq!( + files_b[0].url.as_str(), + "https://example.com/0000000000/0000000000-0000001000-abcdef12-fork/blocks.parquet" + ); +} diff --git a/src/types/state.rs b/src/types/state.rs index a195918e..43c945e3 100644 --- a/src/types/state.rs +++ b/src/types/state.rs @@ -1,13 +1,14 @@ use super::dataset::Dataset; -use crate::storage::layout::DataChunk; use std::{collections::BTreeSet, sync::Arc}; pub type ChunkSet = BTreeSet; +pub type DatasetId = Arc; +pub type ChunkId = Arc; #[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] pub struct ChunkRef { - pub dataset: Arc, - pub chunk: DataChunk, + pub dataset: DatasetId, + pub chunk: ChunkId, } impl std::fmt::Debug for ChunkRef { From 413988cbcf6ca96be214d87e0855636f46cedd23 Mon Sep 17 00:00:00 2001 From: Defnull <879658+define-null@users.noreply.github.com> Date: Wed, 13 May 2026 15:16:48 +0200 Subject: [PATCH 3/5] Fix formatting --- src/controller/worker.rs | 4 ++-- src/storage/datasets_index.rs | 1 - src/storage/layout.rs | 7 +++---- 3 files changed, 5 insertions(+), 7 deletions(-) diff --git a/src/controller/worker.rs b/src/controller/worker.rs index bb5e18ac..4ce284c0 100644 --- a/src/controller/worker.rs +++ b/src/controller/worker.rs @@ -115,7 +115,7 @@ impl Worker { let mut query = sqd_query::Query::from_json_bytes(query_str.as_bytes()) .map_err(|e| QueryError::BadRequest(format!("Couldn't parse query: {e:?}")))?; let (from_block, to_block) = block_range; - + query.set_first_block(from_block); query.set_last_block(Some(to_block)); @@ -144,7 +144,7 @@ impl Worker { // the chunk's last_block (parsed from the chunk_id) so the // portal could see progress. After NET-385 the worker // treats chunk_id as opaque, so we report the query's - // upper bound + // upper bound to_block }; let bytes = writer.finish()?; diff --git a/src/storage/datasets_index.rs b/src/storage/datasets_index.rs index 465b0d50..a45685e9 100644 --- a/src/storage/datasets_index.rs +++ b/src/storage/datasets_index.rs @@ -144,4 +144,3 @@ fn test_url_joining() { .unwrap(); assert_eq!(url.as_str(), "https://eclipse-testnet-2.sqd-datasets.io/0086800000/0089600001-0089800000-cg1JNYDM/blocks.parquet"); } - diff --git a/src/storage/layout.rs b/src/storage/layout.rs index 4fbd956a..e7f21de2 100644 --- a/src/storage/layout.rs +++ b/src/storage/layout.rs @@ -172,8 +172,8 @@ pub async fn read_all_chunks(fs: &impl Filesystem) -> Result> { for (cur, next) in chunks.iter().tuple_windows() { // Two chunks sharing the exact same range are allowed — // suffix-distinguished forks. Anything else overlapping bails. - let same_range = cur.first_block == next.first_block - && cur.last_block == next.last_block; + let same_range = + cur.first_block == next.first_block && cur.last_block == next.last_block; if !same_range && cur.last_block >= next.first_block { bail!("Overlapping ranges: {} and {}", cur, next); } @@ -322,8 +322,7 @@ mod tests { .expect("layout should accept both chunks"); assert_eq!(chunks.len(), 2); - let ids: std::collections::HashSet<&str> = - chunks.iter().map(|c| c.id.as_str()).collect(); + let ids: std::collections::HashSet<&str> = chunks.iter().map(|c| c.id.as_str()).collect(); assert!(ids.contains(chunk_a_id)); assert!(ids.contains(chunk_b_id)); From 2f95e5b7184314af11296c060950e15d989e4c24 Mon Sep 17 00:00:00 2001 From: Defnull <879658+define-null@users.noreply.github.com> Date: Mon, 18 May 2026 11:03:17 +0200 Subject: [PATCH 4/5] Address PR comments --- Cargo.toml | 5 ++++- src/controller/p2p.rs | 31 ++++++++++++++++++------------- 2 files changed, 22 insertions(+), 14 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 3ca8b603..22c0c798 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -60,7 +60,7 @@ url = "2.5.2" walkdir = "2.5.0" zstd = "0.13" -sqd-assignments = { git = "https://github.com/subsquid/sqd-network.git", rev = "938ffe1", features = ["reader", "builder"] } +sqd-assignments = { git = "https://github.com/subsquid/sqd-network.git", rev = "938ffe1", features = ["reader"] } sqd-contract-client = { git = "https://github.com/subsquid/sqd-network.git", rev = "938ffe1", version = "1.2.1" } sqd-messages = { git = "https://github.com/subsquid/sqd-network.git", rev = "938ffe1", version = "2.0.2", features = ["bitstring"] } sqd-network-transport = { git = "https://github.com/subsquid/sqd-network.git", rev = "938ffe1", version = "3.0.0", features = ["worker", "metrics"] } @@ -70,6 +70,9 @@ sqd-polars = { git = "https://github.com/subsquid/data.git", rev = "9db54d3" } sql_query_plan = {git = "https://github.com/subsquid/qplan.git", rev = "658f88f" } +[dev-dependencies] +sqd-assignments = { git = "https://github.com/subsquid/sqd-network.git", rev = "938ffe1", features = ["builder"] } + [profile.release] debug = true opt-level = 3 diff --git a/src/controller/p2p.rs b/src/controller/p2p.rs index de224c6b..2aa41787 100644 --- a/src/controller/p2p.rs +++ b/src/controller/p2p.rs @@ -465,18 +465,27 @@ impl + Send + 'static> P2PController() { - if let Some(range) = query.block_range { - let active_len = std::cmp::min(chunk.last_block.into(), range.end) - .saturating_sub(std::cmp::max(chunk.first_block.into(), range.begin)) - .max(1); - let chunk_len = Into::::into(chunk.last_block) - .saturating_sub(chunk.first_block.into()) - .max(1); - allocation_chip = active_len as f32 / chunk_len as f32; - } + let (begin, end) = block_range; + let active_len = std::cmp::min(chunk.last_block.into(), end) + .saturating_sub(std::cmp::max(chunk.first_block.into(), begin)) + .max(1); + let chunk_len = Into::::into(chunk.last_block) + .saturating_sub(chunk.first_block.into()) + .max(1); + allocation_chip = active_len as f32 / chunk_len as f32; }; // We claim 1. allocation first and refund unused allocation later. It's done to prevent burst overloading with small requests. @@ -493,10 +502,6 @@ impl + Send + 'static> P2PController Date: Mon, 18 May 2026 11:33:03 +0200 Subject: [PATCH 5/5] Address PR comments --- src/storage/datasets_index.rs | 2 +- src/storage/layout.rs | 16 +++++++++++++++- src/storage/manager.rs | 2 +- src/storage/state.rs | 2 +- src/storage/tests.rs | 4 ++-- src/types/state.rs | 2 +- 6 files changed, 21 insertions(+), 7 deletions(-) diff --git a/src/storage/datasets_index.rs b/src/storage/datasets_index.rs index a45685e9..e766cc41 100644 --- a/src/storage/datasets_index.rs +++ b/src/storage/datasets_index.rs @@ -84,7 +84,7 @@ impl DatasetsIndex { for (chunk_ref, chunk) in worker.iter_chunks_with_ref() { let key = ChunkRef { dataset: pool.get(chunk.dataset_id()), - chunk: Arc::from(chunk.id().to_string()), + chunk: Arc::from(chunk.id()), }; chunks.insert(key, chunk_ref); } diff --git a/src/storage/layout.rs b/src/storage/layout.rs index e7f21de2..fa1a971e 100644 --- a/src/storage/layout.rs +++ b/src/storage/layout.rs @@ -64,13 +64,27 @@ impl Deref for BlockNumber { /// `0000001000/0000001024-0000002047-0xabcdef`. It has been extended with an /// optional trailing suffix so multiple chunks can cover the same block /// range, e.g. `0000001000/0000001024-0000002047-0xabcdef`. -#[derive(PartialEq, Eq, PartialOrd, Ord, Clone, Hash)] +#[derive(PartialEq, Eq, Clone, Hash)] pub struct DataChunk { pub id: String, pub first_block: BlockNumber, pub last_block: BlockNumber, } +impl Ord for DataChunk { + fn cmp(&self, other: &Self) -> std::cmp::Ordering { + self.last_block + .cmp(&other.last_block) + .then_with(|| self.id.cmp(&other.id)) + } +} + +impl PartialOrd for DataChunk { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + impl DataChunk { // TODO: synchronize with other language implementations pub fn from_path(dirname: &str) -> Result { diff --git a/src/storage/manager.rs b/src/storage/manager.rs index 26e42f8a..bca4b742 100644 --- a/src/storage/manager.rs +++ b/src/storage/manager.rs @@ -255,7 +255,7 @@ impl StateManager { self.fs .root .join(dataset::encode_dataset(&chunk_ref.dataset)) - .join(&chunk_ref.chunk.as_str()) + .join(chunk_ref.chunk.as_ref()) } } diff --git a/src/storage/state.rs b/src/storage/state.rs index 4820cc7e..6e2e3783 100644 --- a/src/storage/state.rs +++ b/src/storage/state.rs @@ -184,7 +184,7 @@ mod tests { let ds = Arc::new("ds".to_owned()); let chunk_ref = |x| ChunkRef { dataset: ds.clone(), - chunk: Arc::new(format!( + chunk: Arc::from(format!( "0000000000/000000000{}-000000000{}-00000000", x, x + 1 diff --git a/src/storage/tests.rs b/src/storage/tests.rs index bc62c34c..899e8b05 100644 --- a/src/storage/tests.rs +++ b/src/storage/tests.rs @@ -80,13 +80,13 @@ fn test_chunks_with_same_block_range() { let chunk_a = index .chunks() .keys() - .find(|cr| cr.chunk.as_str() == chunk_a_id) + .find(|cr| cr.chunk.as_ref() == chunk_a_id) .cloned() .expect("chunk A should be in the index"); let chunk_b = index .chunks() .keys() - .find(|cr| cr.chunk.as_str() == chunk_b_id) + .find(|cr| cr.chunk.as_ref() == chunk_b_id) .cloned() .expect("chunk B should be in the index"); diff --git a/src/types/state.rs b/src/types/state.rs index 43c945e3..96d1d4fe 100644 --- a/src/types/state.rs +++ b/src/types/state.rs @@ -3,7 +3,7 @@ use std::{collections::BTreeSet, sync::Arc}; pub type ChunkSet = BTreeSet; pub type DatasetId = Arc; -pub type ChunkId = Arc; +pub type ChunkId = Arc; #[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] pub struct ChunkRef {