fix: multiple fixes around system stability under load #1346
Conversation
WalkthroughThis change refactors and extends the synchronization, staging, and conversion logic for arrow and parquet files. It introduces asynchronous, per-stream synchronization using Tokio's Changes
Sequence Diagram(s)sequenceDiagram
participant Server
participant SyncModule
participant Streams
participant ObjectStorage
Server->>+SyncModule: sync_start()
SyncModule->>+Streams: flush_and_convert(init_signal=true, shutdown_signal=false)
Streams-->>-SyncModule: Result
SyncModule->>+ObjectStorage: sync_all_streams(joinset)
ObjectStorage->>+ObjectStorage: upload_files_from_staging(stream_name) (per stream, concurrent)
ObjectStorage-->>-SyncModule: Result (per stream)
SyncModule-->>-Server: Result
Possibly related PRs
Suggested labels
Poem
✨ Finishing Touches
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🔭 Outside diff range comments (1)
src/parseable/streams.rs (1)
345-353:⚠️ Potential issue
File::created()will panic on common Linux filesystems
std::fs::Metadata::created()is not implemented on most Unix filesystems and returns an error.
Theexpect("Creation time should be accessible")will therefore bring the whole conversion task – and potentially the server – down on virtually every Linux deployment.Replace with a fall-back that relies on the consistently available
modified()timestamp (or skip the filtering whencreated()is unsupported).- let creation = path.metadata()?.created().expect("Creation time should be accessible"); + let creation = path + .metadata()? + .created() + .or_else(|_| path.metadata()?.modified())?; // fallback
♻️ Duplicate comments (1)
src/handlers/http/modal/server.rs (1)
126-131: Startup sync duplicationSame concern as in
query_server.rs: this detachedsync_start()may overlap with the periodicsync::handlerspawned just below.
Verify that the two paths cannot run concurrently or gate them behind a single-execution guard.
🧹 Nitpick comments (6)
src/handlers/http/modal/query_server.rs (1)
131-136: Possible duplicate & unmanaged sync task
sync_start()is fired here while a separate periodic sync thread (sync::handler) is launched a few lines later.
Without a guard this can lead to two independent object-store uploads running simultaneously, wasting IO and increasing race-surface.
Consider:
- Running the startup sync inside the periodic worker before its main loop begins, or
- Protecting concurrent runs with a mutex/atomic “sync-in-progress” flag, and/or
- At least keep the
JoinHandleso it can be awaited during shutdown.src/parseable/staging/reader.rs (1)
88-103: Great improvement, but one remainingunwrapstill panicsThe new
match File::openkeeps the reader resilient – nice.
However the earlierfile.metadata().unwrap()(line 51) will still crash the entire process for an I/O error (e.g. permission issue).
Replace with a fallible path similar to this block and skip the file on failure.src/handlers/http/health_check.rs (1)
63-65: Hard-coded 5 s sleep may stall graceful shutdownA fixed
sleep(5s)delays shutdown even when no in-flight requests exist and also blocks all new requests (flag is already set).
Consider making this configurable or replacing with a loop that waits until active-request-counter drops to zero (with max timeout).src/parseable/streams.rs (2)
70-77:arrow_path_to_parquetargument order is confusing
staging_pathis expected to be a directory, yet tests pass a file path.
Consider tightening the contract:-fn arrow_path_to_parquet(staging_path: &Path, path: &Path, random_string: &str) +fn arrow_path_to_parquet(staging_dir: &Path, arrow_file: &Path, random_string: &str)and add an early
debug_assert!(staging_dir.is_dir())to prevent accidental misuse.
This will avoid silently generating paths like/some/file.arrows/<name>.parquet.
588-604: Metric update panics on I/O errors
file.metadata().unwrap()will panic on any transient I/O failure.
Preferif let Ok(meta) = file.metadata()to keep the sync path resilient under load.src/sync.rs (1)
155-161: JoinSet drain duplicates log code – factor into helperThe same success / failure logging block appears three times in this file.
Extracting a small helper (log_join_result(res, "object store sync")) will remove repetition and keep future fixes in one spot.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (7)
src/handlers/http/health_check.rs(2 hunks)src/handlers/http/modal/query_server.rs(2 hunks)src/handlers/http/modal/server.rs(2 hunks)src/parseable/staging/reader.rs(2 hunks)src/parseable/streams.rs(24 hunks)src/storage/object_storage.rs(2 hunks)src/sync.rs(7 hunks)
🧰 Additional context used
🧬 Code Graph Analysis (2)
src/handlers/http/modal/query_server.rs (1)
src/sync.rs (1)
sync_start(248-271)
src/handlers/http/modal/server.rs (1)
src/sync.rs (1)
sync_start(248-271)
⏰ Context from checks skipped due to timeout of 90000ms (10)
- GitHub Check: Build Default x86_64-pc-windows-msvc
- GitHub Check: Build Default aarch64-unknown-linux-gnu
- GitHub Check: Build Default aarch64-apple-darwin
- GitHub Check: Build Default x86_64-apple-darwin
- GitHub Check: Build Default x86_64-unknown-linux-gnu
- GitHub Check: Build Kafka x86_64-unknown-linux-gnu
- GitHub Check: Build Kafka aarch64-apple-darwin
- GitHub Check: coverage
- GitHub Check: Quest Smoke and Load Tests for Standalone deployments
- GitHub Check: Quest Smoke and Load Tests for Distributed deployments
🔇 Additional comments (2)
src/sync.rs (2)
79-82: Re-consider spawning a fresh Tokio runtime here
handleris annotated with#[tokio::main], and the crate already defines another
#[tokio::main](alert_runtime). Each call spins up a stand-alone multi-thread
runtime, which is expensive and prevents reuse of the global IO driver.If
handleris invoked from within an existing runtime (e.g. Actix), switch to
tokio::spawn+async fninstead of a second#[tokio::main].
247-271: Startup sync may run concurrently with periodic sync
sync_start()performs a full flush + upload but does not coordinate with the
long-runninglocal_sync/object_store_syncloops. If startup takes longer than
LOCAL_SYNC_INTERVAL, you can end up with overlapping conversions on the same stream.Consider signalling completion (e.g. a OnceCell flag) before enabling the periodic
tasks, or make the periodic loops wait onsync_start’s JoinHandle.
nikhilsinhaparseable
force-pushed
the
fix-conversion-in-peak
branch
from
42d27d0 to
7416b7a
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (4)
src/sync.rs (4)
131-144: Potential starvation & duplicated code aroundJoinSetloopEvery tick builds a fresh
JoinSet, firessync_all_streams, then blocks until all joins resolve.
If a large backlog exists, the loop may overrun the next tick, defeating periodic behaviour.Consider extracting the common “spawn, join, log” logic used here and in
local_syncinto a reusable helper that:
- Re-uses a single
JoinSet(or a bounded one) across ticks, or- Offloads joins into a detached task so the driving loop never stalls.
This will keep upload cadence predictable and DRY-up the nearly identical code in the two sync paths.
195-207: Duplicate flush/convert scaffold – factor out to avoid divergenceThe block mirrors lines 131-144 almost verbatim (different callback only). If a bug fix is needed later you’ll have to patch two places.
Extract into something like:
async fn run_and_log<F>(name: &str, threshold: Duration, f: F) where F: FnOnce(&mut JoinSet<()>) + Send + 'static, { … }and call it from both locations.
238-264:sync_startruns heavy IO on the main thread during startup
sync_start()performs full flush + upload synchronously before returning. On large datasets this can delay server readiness and make liveness probes fail.Introduce either:
• a “started but warming up” state exposed via health-check, or
• run this in the background and let the server accept traffic immediately (with proper back-pressure).Also consider making the 15 s threshold configurable – real-world clusters may legitimately take longer.
269-278: Minor: inconsistent log levels
log_join_result()logs success withinfo!, failures withwarn!, and join errors witherror!.
You may want to promote task failures (Ok(Err(_))) toerror!as well to ease alerting.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
src/sync.rs(6 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (10)
- GitHub Check: Build Default x86_64-apple-darwin
- GitHub Check: Build Default x86_64-pc-windows-msvc
- GitHub Check: Build Default aarch64-apple-darwin
- GitHub Check: Build Default aarch64-unknown-linux-gnu
- GitHub Check: Quest Smoke and Load Tests for Standalone deployments
- GitHub Check: Build Default x86_64-unknown-linux-gnu
- GitHub Check: Quest Smoke and Load Tests for Distributed deployments
- GitHub Check: Build Kafka aarch64-apple-darwin
- GitHub Check: Build Kafka x86_64-unknown-linux-gnu
- GitHub Check: coverage
1. perform object store sync for all streams in parallel
2. remove restriction of multi threading to utilise all available cores
3. add atomicity in conversion by -
i. each conversion task processes one minute of arrows
ii. move arrow files to inprocess folder to maintain atomicity
iii. add a init sync task to process all pending files
iv. add tokio sleep of 5 secs in shutdown task to let complete ongoing jobs
v. remove unwrap of write locks to avoid thread poisoning
nikhilsinhaparseable
force-pushed
the
fix-conversion-in-peak
branch
from
fe684f8 to
0cb81b6
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (2)
src/sync.rs (1)
131-145: Consider graceful early-exit for still-runningJoinSettasksIf
inbox_rxreceives a shutdown signal whilesync_all_streamsis in progress, any in-flight uploads insidejoinsetkeep running until completion because theJoinSetlives only inside this closure.
In practice that is fine, but you lose the option to abort quickly during shutdown or when the monitor times out.A lightweight pattern is to call
joinset.shutdown().await;before thewhile let Some(..)loop, which causes unfinished tasks to be aborted and immediately joined:- let mut joinset = JoinSet::new(); + let mut joinset = JoinSet::new(); sync_all_streams(&mut joinset); + // Abort remaining tasks if the outer future is cancelled + joinset.shutdown().await; while let Some(res) = joinset.join_next().await { log_join_result(res, "object store sync"); }This avoids dangling uploads during shutdown and keeps task lifetimes explicit.
src/parseable/streams.rs (1)
380-387: Cross-device rename can fail – fall back to copy/remove
fs::renamefails withEXDEVif the staging disk and the in-process directory reside on
different mounts (common in containerised setups with bind-mounts).
Handle that case explicitly:- if let Err(e) = fs::rename(&arrow_file_path, &new_path) { + if let Err(e) = fs::rename(&arrow_file_path, &new_path) { error!( "Failed to rename arrow file to inprocess directory: {} -> {}: {e}", arrow_file_path.display(), new_path.display() ); + if e.kind() == std::io::ErrorKind::CrossDeviceLink { + if let Err(copy_err) = fs::copy(&arrow_file_path, &new_path) { + error!("Copy fallback also failed: {copy_err}"); + } else { + let _ = remove_file(&arrow_file_path); + } }This prevents the entire batch from stalling on heterogeneous storage layouts.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (7)
src/handlers/http/health_check.rs(2 hunks)src/handlers/http/modal/query_server.rs(2 hunks)src/handlers/http/modal/server.rs(2 hunks)src/parseable/staging/reader.rs(3 hunks)src/parseable/streams.rs(22 hunks)src/storage/object_storage.rs(2 hunks)src/sync.rs(6 hunks)
🚧 Files skipped from review as they are similar to previous changes (5)
- src/handlers/http/modal/query_server.rs
- src/handlers/http/modal/server.rs
- src/handlers/http/health_check.rs
- src/parseable/staging/reader.rs
- src/storage/object_storage.rs
🧰 Additional context used
🧠 Learnings (1)
src/sync.rs (1)
Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1346
File: src/sync.rs:78-83
Timestamp: 2025-06-16T04:56:21.596Z
Learning: In Rust async code, `#[tokio::main]` is appropriate when functions are called from separate OS threads (like `thread::spawn()`), as it creates a new Tokio runtime for that thread. The "Cannot start a runtime from within a runtime" error only occurs when `#[tokio::main]` functions are called from within an existing Tokio runtime context.
🧬 Code Graph Analysis (1)
src/sync.rs (2)
src/storage/object_storage.rs (1)
sync_all_streams(878-897)src/parseable/streams.rs (3)
new(109-126)flush_and_convert(951-973)flush_and_convert(1047-1062)
⏰ Context from checks skipped due to timeout of 90000ms (7)
- GitHub Check: Quest Smoke and Load Tests for Standalone deployments
- GitHub Check: Quest Smoke and Load Tests for Distributed deployments
- GitHub Check: coverage
- GitHub Check: Build Default x86_64-unknown-linux-gnu
- GitHub Check: Build Default x86_64-pc-windows-msvc
- GitHub Check: Build Default aarch64-unknown-linux-gnu
- GitHub Check: Build Kafka x86_64-unknown-linux-gnu
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
src/parseable/streams.rs (2)
69-84:arrow_path_to_parquettrustsstream_staging_pathis a directory
parquet_path.push(filename_with_random_number)silently produces paths like
/tmp/myfile.arrows/<generated>.parquetwhenstream_staging_pathis accidentally a
file (the current unit-tests pass such a path).
At runtime this is unlikely, but if it happens the rename that follows will fail with
Not a directory.- let mut parquet_path = stream_staging_path.to_owned(); - parquet_path.push(filename_with_random_number); + // Guard against a file being passed instead of the staging directory. + let mut parquet_path = if stream_staging_path.is_dir() { + stream_staging_path.to_path_buf() + } else { + stream_staging_path + .parent() + .unwrap_or_else(|| Path::new(".")) + .to_path_buf() + }; + parquet_path.push(filename_with_random_number);This keeps the function robust and lets the unit-tests keep their current call-site.
585-593: Metrics reset only clears “arrows”/“parquet” bucketsThe helper zeroes
STAGING_FILESand bothSTORAGE_SIZEgauges, but only for the
labels"arrows"and"parquet". If a future extension introduces another label
(e.g."orc"), the metrics will become stale. Consider looping through
metrics::STORAGE_SIZE.collect()and clearing every entry for the current stream.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
src/otel/traces.rs(12 hunks)src/parseable/streams.rs(22 hunks)
✅ Files skipped from review due to trivial changes (1)
- src/otel/traces.rs
🧰 Additional context used
🧠 Learnings (1)
src/parseable/streams.rs (1)
Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1346
File: src/parseable/streams.rs:351-355
Timestamp: 2025-06-16T05:20:18.557Z
Learning: In the Parseable codebase, arrow files are expected to always have valid creation or modified timestamps as a basic system assumption. The conversion flow uses expect() on file metadata operations to enforce this invariant with fail-fast behavior rather than graceful error handling, as violations represent fundamental system issues that should cause immediate failures.
⏰ Context from checks skipped due to timeout of 90000ms (10)
- GitHub Check: Quest Smoke and Load Tests for Standalone deployments
- GitHub Check: Quest Smoke and Load Tests for Distributed deployments
- GitHub Check: coverage
- GitHub Check: Build Default aarch64-apple-darwin
- GitHub Check: Build Default x86_64-unknown-linux-gnu
- GitHub Check: Build Default x86_64-pc-windows-msvc
- GitHub Check: Build Default x86_64-apple-darwin
- GitHub Check: Build Default aarch64-unknown-linux-gnu
- GitHub Check: Build Kafka x86_64-unknown-linux-gnu
- GitHub Check: Build Kafka aarch64-apple-darwin
🔇 Additional comments (2)
src/parseable/streams.rs (2)
717-723: Hard-coded label mismatch may skew storage metricsHere the label is
ARROW_FILE_EXTENSION, whileupdate_staging_metricsuses the
literal string"arrows". If the constant ever diverges (e.g."arrow"), metrics
for deletions and additions will land in different time-series.Validate that
ARROW_FILE_EXTENSIONis exactly"arrows"or unify both call-sites
to the same constant.
140-149: 👍 Graceful recovery from poisoned mutexGreat call replacing the immediate panic with a warning and
into_inner()—
this prevents a single thread panic from killing ingestion on the whole stream.
Summary by CodeRabbit