src/handlers/http/modal/server.rs (1)

78-99: ⚠️ Potential issue | 🟠 Major

Reapply resource shedding to /query, /ingest, and /counts.

These registrations dropped resource_check::check_resource_utilization_middleware, so the most expensive request paths now bypass the overload gate while /logstream/{logstream} and the OTEL routes still enforce it. Under CPU/memory pressure that lets new query/ingest work in exactly when the server is trying to protect itself.

Suggested fix

                 web::scope(&base_path())
                     .service(Self::get_correlation_webscope())
-                    .service(Self::get_query_factory())
-                    .service(Self::get_ingest_factory())
+                    .service(Self::get_query_factory().wrap(from_fn(
+                        resource_check::check_resource_utilization_middleware,
+                    )))
+                    .service(Self::get_ingest_factory().wrap(from_fn(
+                        resource_check::check_resource_utilization_middleware,
+                    )))
                     .service(Self::get_liveness_factory())
                     .service(Self::get_readiness_factory())
                     .service(Self::get_about_factory())
@@
-                    .service(Self::get_counts_webscope())
+                    .service(Self::get_counts_webscope().wrap(from_fn(
+                        resource_check::check_resource_utilization_middleware,
+                    )))

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/handlers/http/modal/server.rs` around lines 78 - 99, The registrations
for get_query_factory, get_ingest_factory, and get_counts_webscope dropped the
overload gate; re-add the resource shedding middleware by wrapping each of those
webscopes with resource_check::check_resource_utilization_middleware() (e.g.,
call
.service(Self::get_query_factory().wrap(resource_check::check_resource_utilization_middleware()))
or apply the middleware on the scope returned by
get_query_factory/get_ingest_factory/get_counts_webscope) so those heavy
endpoints are subject to the same resource checks as the other routes.

src/handlers/http/modal/ingest_server.rs (1)

67-84: ⚠️ Potential issue | 🟠 Major

Restore the resource guard on the plain ingest route.

Server::get_ingest_factory() is now exposed without resource_check::check_resource_utilization_middleware, while the OTEL ingest route and /logstream/{logstream} still keep it. That makes /api/v1/ingest the one ingest entrypoint that keeps accepting writes under overload instead of shedding load like the others.

Suggested fix

                 web::scope(&base_path())
-                    .service(Server::get_ingest_factory())
+                    .service(Server::get_ingest_factory().wrap(from_fn(
+                        resource_check::check_resource_utilization_middleware,
+                    )))
                     .service(Self::logstream_api())

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/handlers/http/modal/ingest_server.rs` around lines 67 - 84, The plain
ingest route exposed by Server::get_ingest_factory is missing the resource guard
middleware; wrap Server::get_ingest_factory with the same
resource_check::check_resource_utilization_middleware used for
Server::get_ingest_otel_factory (i.e., apply
.wrap(from_fn(resource_check::check_resource_utilization_middleware)) to the
service registration for Server::get_ingest_factory) so /api/v1/ingest is
subject to the same load-shedding checks as the OTEL and /logstream/{logstream}
routes.

src/storage/object_storage.rs (1)

1109-1139: ⚠️ Potential issue | 🔴 Critical

Always clear ACTIVE_OBJECT_STORE_SYNC_FILES for failed and retried files.

Right now only successfully uploaded paths are removed from the global active set. A validation failure (manifest_file == None), task error, or task panic leaves the parquet in staging and leaves its path in ACTIVE_OBJECT_STORE_SYNC_FILES, so later sync cycles filter it out forever instead of retrying it.

Based on learnings failed parquet file uploads should remain in the staging directory for retry in the next sync cycle, while successful uploads remove their staged files immediately.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/storage/object_storage.rs` around lines 1109 - 1139, The loop only
removes successfully uploaded paths from ACTIVE_OBJECT_STORE_SYNC_FILES, leaving
failed/retried files stuck; update the join_set handling so that every branch
that represents a failed or retried upload also removes the corresponding file
path from ACTIVE_OBJECT_STORE_SYNC_FILES before continuing or returning.
Concretely: inside the join_set.join_next() match, when Ok(Ok(upload_result))
has manifest_file == None remove upload_result.file_path from
ACTIVE_OBJECT_STORE_SYNC_FILES (using the same write().await guard) instead of
leaving it; likewise ensure Ok(Err(e)) removes the associated file path from
ACTIVE_OBJECT_STORE_SYNC_FILES before returning Err(e) (or change the upload
error type to include file_path so you can remove it), and for Err(e) (task
panic) remove any known path information from ACTIVE_OBJECT_STORE_SYNC_FILES
before returning ObjectStorageError::UnhandledError; keep the existing
uploaded_files push only for successful manifest_file Some cases.

src/storage/azure_blob.rs (1)

379-418: ⚠️ Potential issue | 🟠 Major

Move multipart writer creation into the multipart branch to avoid orphaned sessions.

put_multipart() is called unconditionally at line 382 before checking file size and the enable_multipart flag. When the file is below min_multipart_size or multipart is disabled (line 393), the function takes a single-part put() path and returns without calling complete() or abort() on the multipart writer. This abandons the multipart session on the Azure backend without cleanup.

The same issue exists in src/storage/s3.rs and src/storage/gcs.rs.

Suggested fix

        let mut file = OpenOptions::new().read(true).open(path).await?;
        let location = &to_object_store_path(key);
        let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
-       let async_writer = self.client.put_multipart(location).await;
-       let mut async_writer = match async_writer {
-           Ok(writer) => writer,
-           Err(err) => {
-               return Err(err.into());
-           }
-       };
-
        let meta = file.metadata().await?;
        let total_size = meta.len() as usize;
        let min_multipart_size = PARSEABLE.options.min_multipart_size as usize;
        if total_size < min_multipart_size || !PARSEABLE.options.enable_multipart {
            let mut data = Vec::new();
@@
             }
             return Ok(());
         } else {
+            let async_writer = self.client.put_multipart(location).await;
+            let mut async_writer = match async_writer {
+                Ok(writer) => writer,
+                Err(err) => {
+                    return Err(err.into());
+                }
+            };
             let mut data = Vec::new();

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/storage/azure_blob.rs` around lines 379 - 418, The code calls
self.client.put_multipart(location) unconditionally (creating async_writer)
before deciding whether to take the single-part path, which leaves an open
multipart session when the function uses self.client.put(...) and returns; move
the put_multipart() call into the multipart branch (the branch that checks
total_size >= PARSEABLE.options.min_multipart_size and
PARSEABLE.options.enable_multipart) so the multipart writer is only created when
needed, and in that branch ensure you call async_writer.put_part(...) and
async_writer.complete() on success or async_writer.abort() on failure; apply the
same change pattern to the corresponding logic in src/storage/s3.rs and
src/storage/gcs.rs (look for put_multipart, async_writer, complete, abort, and
the single-part self.client.put paths).

src/storage/gcs.rs (1)

346-357: ⚠️ Potential issue | 🔴 Critical

Same ordering bug: multipart init happens before the disable/size check.

At Line 346, put_multipart is invoked before Line 357 decides whether multipart should be used. If multipart is disabled (or below threshold), this can still fail early and prevent the single-PUT fallback.

💡 Suggested fix

-        let async_writer = self.client.put_multipart(location).await;
-        let mut async_writer = match async_writer {
-            Ok(writer) => writer,
-            Err(err) => {
-                return Err(err.into());
-            }
-        };
-
         let meta = file.metadata().await?;
         let total_size = meta.len() as usize;
         let min_multipart_size = PARSEABLE.options.min_multipart_size as usize;
         if total_size < min_multipart_size || !PARSEABLE.options.enable_multipart {
             let mut data = Vec::new();
             file.read_to_end(&mut data).await?;
@@
             return Ok(());
         } else {
+            let async_writer = self.client.put_multipart(location).await;
+            let mut async_writer = match async_writer {
+                Ok(writer) => writer,
+                Err(err) => {
+                    return Err(err.into());
+                }
+            };
             let mut data = Vec::new();
             file.read_to_end(&mut data).await?;

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/storage/gcs.rs` around lines 346 - 357, The multipart writer is being
initialized unconditionally by calling self.client.put_multipart(...) (bound to
async_writer) before checking whether multipart should be used (based on
total_size and PARSEABLE.options.enable_multipart), which can cause failures
even when a single-PUT fallback is intended; change the logic in the upload flow
(around put_multipart, async_writer, file.metadata(), total_size,
min_multipart_size and PARSEABLE.options) to first compute meta =
file.metadata().await?, total_size and decide whether to use multipart
(total_size >= min_multipart_size && PARSEABLE.options.enable_multipart) and
only then call self.client.put_multipart(...) when multipart is chosen, leaving
the single-PUT path to proceed without having attempted multipart
initialization.

src/storage/s3.rs (1)

535-547: ⚠️ Potential issue | 🔴 Critical

Move multipart initialization below the fallback guard.

put_multipart is still called at Line 535 before the Line 546 check. So even when multipart is disabled (or file is below threshold), initialization may fail and block the intended single-PUT path, and may create unnecessary multipart sessions.

💡 Suggested fix

-        let async_writer = self.client.put_multipart(location).await;
-        let mut async_writer = match async_writer {
-            Ok(writer) => writer,
-            Err(err) => {
-                return Err(err.into());
-            }
-        };
-
         let meta = file.metadata().await?;
         let total_size = meta.len() as usize;
         let min_multipart_size = PARSEABLE.options.min_multipart_size as usize;
         if total_size < min_multipart_size || !PARSEABLE.options.enable_multipart {
             let mut data = Vec::new();
             file.read_to_end(&mut data).await?;
@@
             return Ok(());
         } else {
+            let async_writer = self.client.put_multipart(location).await;
+            let mut async_writer = match async_writer {
+                Ok(writer) => writer,
+                Err(err) => {
+                    return Err(err.into());
+                }
+            };
             let mut data = Vec::new();
             file.read_to_end(&mut data).await?;

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/storage/s3.rs` around lines 535 - 547, The code prematurely calls
self.client.put_multipart(...) and initializes async_writer before checking
multipart eligibility; move the put_multipart call (and the async_writer
creation/match on its Result) so it occurs only after computing meta =
file.metadata().await?, total_size and verifying total_size >=
PARSEABLE.options.min_multipart_size and PARSEABLE.options.enable_multipart;
keep the single-PUT path (the Vec::new() branch) untouched and only enter
multipart initialization (put_multipart and the Err/Ok handling) when those
multipart conditions are met.

src/storage/object_storage.rs (1)

1123-1166: ⚠️ Potential issue | 🟠 Major

ACTIVE_OBJECT_STORE_SYNC_FILES cleanup is delaying retries.

Line 1041-Line 1053 inserts the whole batch into the active set up front, but this cleanup only removes paths that produced a manifest, plus the first hard-failure path. Validation failures (manifest_file: None) stay blocked, and any remaining batch entries on early return stay blocked too. The fallback prune at Line 1156-Line 1166 also uses the parquet timestamp from the filename instead of insertion time, so this is not actually a 5-minute TTL in memory. Net effect: files that should retry on the next sync tick can be skipped for much longer.

Please track insertion time separately and clear the full batch on every exit path, including validation-failed uploads.

Based on learnings, failed parquet file uploads should remain in staging for retry in the next sync cycle, and validation failures should rely on the normal 30-second sync cycle rather than a bounded delay.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/storage/object_storage.rs` around lines 1123 - 1166,
ACTIVE_OBJECT_STORE_SYNC_FILES is currently keyed only by path and pruned using
parquet filename timestamps, which blocks retries; change
ACTIVE_OBJECT_STORE_SYNC_FILES to map paths to an insertion timestamp (e.g.,
HashMap<PathBuf, DateTime<Utc>>) and set the timestamp when you insert the whole
batch (refer to ACTIVE_OBJECT_STORE_SYNC_FILES and the batch insertion code). On
every exit path from the upload loop (including the Ok(Ok(upload_result)) branch
where upload_result.manifest_file is None and the Ok(Err((path, e)))
early-return), remove all paths from the map for this batch so validation-failed
files are eligible on the next sync tick (refer to uploaded_files,
upload_result.manifest_file, and the early-return branch). Replace the fallback
retain logic that calls extract_datetime_from_parquet_path_regex with a TTL
check against the stored insertion timestamp (now - insertion_ts >= 5 minutes)
so the in-memory TTL is based on insertion time, not filename. Ensure all places
that read ACTIVE_OBJECT_STORE_SYNC_FILES now account for the new value type.

src/handlers/http/modal/ingest_server.rs (1)

70-81: ⚠️ Potential issue | 🟠 Major

Confirm intentional removal of resource utilization enforcement from ingest endpoints.

The check_resource_utilization_middleware has been completely removed from the codebase—it is now defined but never called. Ingest endpoints and the logstream POST handler no longer return 503 Service Unavailable when CPU or memory exceeds configured thresholds.

The background spawn_resource_monitor continues to run and updates the RESOURCE_CHECK_ENABLED flag, but that flag is no longer consumed by any enforcement layer. This means the monitoring infrastructure persists without purpose.

This is a significant behavior change: requests will now be accepted even when resources are exhausted, increasing the risk of server instability under high load. There are no comments or documentation explaining this change.

Confirm whether this removal is:

1. Intentional (e.g., performance improvement, architectural shift)
2. Part of an incomplete refactoring
3. An oversight that should be reverted

If intentional, consider either removing the orphaned spawn_resource_monitor call or documenting the reasoning.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/handlers/http/modal/ingest_server.rs` around lines 70 - 81, The
resource-checking middleware (check_resource_utilization_middleware) was removed
while spawn_resource_monitor and RESOURCE_CHECK_ENABLED remain, leaving
monitoring orphaned; either restore enforcement by applying
check_resource_utilization_middleware to ingest-related services (e.g., wrap or
.service calls for Server::get_ingest_factory(), Self::logstream_api(),
Self::get_demo_data_webscope(), etc.) so POST logstream and ingest endpoints
return 503 when thresholds exceed, or if the change was intentional remove
spawn_resource_monitor and the RESOURCE_CHECK_ENABLED flag and delete the unused
middleware to avoid dead code and document the decision; update the service
chain where get_ingest_factory/logstream_api are registered to include the
middleware if reinstating enforcement.

src/storage/object_storage.rs (1)

1138-1148: ⚠️ Potential issue | 🟠 Major

Early return on error leaves orphaned paths in deduplication set, blocking immediate retries.

When an upload error triggers early return, the JoinSet is dropped and remaining tasks are aborted. However, paths for those aborted tasks remain in ACTIVE_OBJECT_STORE_SYNC_FILES. On the next sync cycle, these paths are filtered out (line 1047), preventing retry until the 5-minute stale cleanup runs.

Consider draining remaining tasks' paths before returning, or wrapping the entire result collection in cleanup logic:

🛡️ One possible approach: cleanup all spawned paths on error

         Ok(Err((path, e))) => {
             error!("Error uploading parquet file: {e}");
             {
                 let mut guard = ACTIVE_OBJECT_STORE_SYNC_FILES.write().await;
                 guard.remove(&path);
+                // Abort remaining tasks and clean up their paths
+                join_set.abort_all();
+                while let Some(result) = join_set.join_next().await {
+                    if let Ok(Ok(upload_result)) = result {
+                        guard.remove(&upload_result.file_path);
+                    } else if let Ok(Err((p, _))) = result {
+                        guard.remove(&p);
+                    }
+                }
             }
             return Err(e);
         }
         Err(e) => {
             error!("Task panicked: {e}");
+            // Cannot identify which path panicked, but we can clean up completed tasks
+            {
+                let mut guard = ACTIVE_OBJECT_STORE_SYNC_FILES.write().await;
+                join_set.abort_all();
+                while let Some(result) = join_set.join_next().await {
+                    if let Ok(Ok(upload_result)) = result {
+                        guard.remove(&upload_result.file_path);
+                    } else if let Ok(Err((p, _))) = result {
+                        guard.remove(&p);
+                    }
+                }
+            }
             return Err(ObjectStorageError::UnhandledError(Box::new(e)));
         }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/storage/object_storage.rs` around lines 1138 - 1148, The early returns in
the Ok(Err((path, e))) and Err(e) branches leave other spawned upload paths in
ACTIVE_OBJECT_STORE_SYNC_FILES, preventing immediate retries; before returning
from either error branch, acquire the
ACTIVE_OBJECT_STORE_SYNC_FILES.write().await lock and remove the current failing
path and all remaining spawned paths for this batch (either by iterating
remaining JoinSet handles to extract their paths or, better, by tracking
spawned_paths in a Vec when you spawn tasks and removing each of those entries).
Ensure both error branches perform this cleanup so aborted tasks do not leave
orphaned paths.

src/storage/object_storage.rs (1)

1138-1149: ⚠️ Potential issue | 🟠 Major

Release the whole reserved batch before returning on upload errors.

These branches only remove the failing path. Any earlier successes from the same process_parquet_files batch stay in ACTIVE_OBJECT_STORE_SYNC_FILES because the cleanup at Lines 1153-1166 never runs once we return here, so the next sync will skip those staged files and delay their manifest/update flow. Please clear every path reserved by this invocation on all exit paths, not just the failed one.

Based on learnings, failed parquet uploads should stay in staging for the next sync cycle, while successful uploads should be cleaned up immediately.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/storage/object_storage.rs` around lines 1138 - 1149, The error branches
in process_parquet_files only remove the single failing path from
ACTIVE_OBJECT_STORE_SYNC_FILES before returning, leaving other paths reserved by
this batch and preventing their cleanup; modify both the Ok(Err((path, e))) and
Err(e) paths to iterate over the full set of paths reserved by this invocation
(the batch variable that was used to reserve entries earlier in
process_parquet_files) and remove each from ACTIVE_OBJECT_STORE_SYNC_FILES under
the write lock before returning, reusing the same cleanup logic currently used
at the normal-exit path so that successful uploads are still removed immediately
while failed items are fully released for the next sync cycle.