Buffer OCI registry uploads to S3 (#752)

Buffer OCI upload network frames into configurable S3 chunks
(`registry.data_store.chunk_size`, default 5 MB) to reduce S3 operations
during Docker push.

Author: epompeii

docker/bench.Dockerfile

@@ -89,7 +89,7 @@ COPY services/api/openapi.json /usr/src/bencher/services/api/openapi.json
 
 WORKDIR /usr/src/bencher
 RUN cargo bench --package bencher_adapter --no-run --message-format=json \
-    | jq -r 'select(.executable != null) | .executable' \
+    | jq -r 'select(.executable != null and .target.kind == ["bench"]) | .executable' \
     | xargs -I {} cp {} /usr/local/bin/bench-adapter
 
 FROM debian:bookworm-slim@sha256:b4aa902587c2e61ce789849cb54c332b0400fe27b1ee33af4669e1f7e7c3e22f

lib/bencher_json/src/system/config/mod.rs

@@ -29,7 +29,10 @@ pub use plus::{
         JsonOciBandwidth, JsonPublicRateLimiter, JsonRateLimiting, JsonRunnerRateLimiter,
         JsonUserRateLimiter,
     },
-    registry::{DEFAULT_MAX_BODY_SIZE, DEFAULT_UPLOAD_TIMEOUT_SECS, RegistryDataStore},
+    registry::{
+        DEFAULT_CHUNK_SIZE, DEFAULT_MAX_BODY_SIZE, DEFAULT_UPLOAD_TIMEOUT_SECS, MAX_CHUNK_SIZE,
+        RegistryDataStore,
+    },
     runners::{DEFAULT_HEARTBEAT_TIMEOUT_SECS, DEFAULT_JOB_TIMEOUT_GRACE_PERIOD_SECS},
     stats::JsonStats,
 };

lib/bencher_json/src/system/config/plus/registry.rs

@@ -9,6 +9,26 @@ pub const DEFAULT_UPLOAD_TIMEOUT_SECS: u64 = 3600;
 /// Default maximum body size: 1 GiB (1,073,741,824 bytes)
 pub const DEFAULT_MAX_BODY_SIZE: u64 = 0x4000_0000;
 
+/// Default chunk size for S3 upload buffering: 5 MB (5,242,880 bytes).
+///
+/// HTTP request bodies arrive as small network frames (typically 8–64 KB).
+/// Without batching, each frame would be stored as a separate S3 object,
+/// creating thousands of objects per layer and making both upload
+/// (~3 S3 ops per frame) and completion (~1 S3 `GetObject` per chunk)
+/// extremely slow. This value controls the minimum batch size before
+/// flushing to S3.
+///
+/// 5 MB also matches the S3 multipart upload minimum part size,
+/// so chunks stored at this size can be efficiently assembled during
+/// upload completion.
+pub const DEFAULT_CHUNK_SIZE: u64 = 5 * 1024 * 1024;
+
+/// Maximum chunk size for S3 upload buffering: 5 GB (5,368,709,120 bytes).
+///
+/// The S3 multipart upload maximum part size is 5 GiB.
+/// We use 5 GB as a practical upper bound.
+pub const MAX_CHUNK_SIZE: u64 = 5 * 1024 * 1024 * 1024;
+
 /// Container registry configuration
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "schema", derive(JsonSchema))]
@@ -59,6 +79,10 @@ pub enum RegistryDataStore {
         /// S3 Access Point ARN with optional path prefix
         /// Format: arn:aws:s3:<region>:<account-id>:accesspoint/<bucket>[/path]
         access_point: String,
+        /// Minimum chunk size in bytes for buffering upload data before storing to S3.
+        /// Valid range: 5 MB–5 GB. Defaults to 5 MB (5,242,880 bytes).
+        /// See [`DEFAULT_CHUNK_SIZE`] and [`MAX_CHUNK_SIZE`].
+        chunk_size: Option<u64>,
     },
 }
 

lib/bencher_schema/src/model/project/report/mod.rs

@@ -154,7 +154,9 @@ impl QueryReport {
                 job: new_run_job,
         } = new_run_report;
 
-        #[cfg(feature = "plus")]
+        #[cfg(all(feature = "plus", not(feature = "otel")))]
+        let _ = is_claimed;
+        #[cfg(all(feature = "plus", feature = "otel"))]
         let priority = plan_kind.priority(is_claimed);
 
         #[cfg(all(feature = "otel", feature = "plus"))]
@@ -333,7 +335,7 @@ impl QueryReport {
                 json_settings,
                 #[cfg(feature = "plus")]
                 plan_kind,
-                #[cfg(feature = "plus")]
+                #[cfg(all(feature = "plus", feature = "otel"))]
                 priority,
                 #[cfg(feature = "plus")]
                 query_project,
@@ -434,7 +436,7 @@ impl QueryReport {
         adapter: Adapter,
         settings: JsonReportSettings,
         #[cfg(feature = "plus")] plan_kind: PlanKind,
-        #[cfg(feature = "plus")] priority: bencher_json::Priority,
+        #[cfg(all(feature = "plus", feature = "otel"))] priority: bencher_json::Priority,
         #[cfg(feature = "plus")] query_project: &QueryProject,
     ) -> Result<(), HttpError> {
         #[cfg(feature = "plus")]

lib/bencher_schema/src/model/runner/job.rs

@@ -147,6 +147,7 @@ impl QueryJob {
                 query_report.adapter,
                 settings,
                 plan_kind,
+                #[cfg(feature = "otel")]
                 self.priority,
                 &query_project,
             )

plus/api_oci/src/blobs.rs

@@ -42,6 +42,7 @@ use crate::auth::{check_oci_bandwidth, record_oci_bandwidth};
 use crate::auth::{
     require_pull_access, require_push_access, resolve_project, validate_push_access,
 };
+use crate::error::storage_error;
 use crate::response::{
     APPLICATION_OCTET_STREAM, DOCKER_CONTENT_DIGEST, DOCKER_UPLOAD_UUID, oci_cors_headers,
 };
@@ -131,7 +132,7 @@ pub async fn oci_blob_exists(
     let size = storage
         .get_blob_size(&project_uuid, &digest)
         .await
-        .map_err(|e| crate::error::into_http_error(OciError::from(e)))?;
+        .map_err(storage_error)?;
 
     // Build response with OCI-compliant headers (no body for HEAD)
     let response = oci_cors_headers(
@@ -195,7 +196,7 @@ pub async fn oci_blob_get(
     let (blob_body, size) = storage
         .get_blob_stream(&project_uuid, &digest)
         .await
-        .map_err(|e| crate::error::into_http_error(OciError::from(e)))?;
+        .map_err(storage_error)?;
 
     // Record bandwidth usage
     #[cfg(feature = "plus")]
@@ -260,7 +261,7 @@ pub async fn oci_blob_delete(
     let exists = storage
         .blob_exists(&project_uuid, &digest)
         .await
-        .map_err(|e| crate::error::into_http_error(OciError::from(e)))?;
+        .map_err(storage_error)?;
     if !exists {
         return Err(crate::error::into_http_error(OciError::BlobUnknown {
             digest: digest.to_string(),
@@ -271,7 +272,7 @@ pub async fn oci_blob_delete(
     storage
         .delete_blob(&project_uuid, &digest)
         .await
-        .map_err(|e| crate::error::into_http_error(OciError::from(e)))?;
+        .map_err(storage_error)?;
 
     // OCI spec requires 202 Accepted for DELETE
     let response = oci_cors_headers(
@@ -387,7 +388,7 @@ pub async fn oci_upload_start(
     let upload_id = storage
         .start_upload(&project_uuid)
         .await
-        .map_err(|e| crate::error::into_http_error(OciError::from(e)))?;
+        .map_err(storage_error)?;
 
     // Build 202 Accepted response
     let location = format!("/v2/{project_slug}/blobs/uploads/{upload_id}");
@@ -460,15 +461,15 @@ pub async fn oci_upload_monolithic(
     let upload_id = storage
         .start_upload(&project_uuid)
         .await
-        .map_err(|e| crate::error::into_http_error(OciError::from(e)))?;
+        .map_err(storage_error)?;
 
     // Stream body to storage (storage enforces max_body_size incrementally)
     let result = async {
         let final_size = crate::uploads::stream_to_storage(body, storage, &upload_id, 0).await?;
         let digest = storage
             .complete_upload(&upload_id, &expected_digest)
             .await
-            .map_err(|e| crate::error::into_http_error(OciError::from(e)))?;
+            .map_err(storage_error)?;
         Ok::<(Digest, u64), HttpError>((digest, final_size))
     }
     .await;

plus/api_oci/src/error.rs

@@ -4,6 +4,11 @@ use bencher_json::oci::{OCI_ERROR_SIZE_INVALID, OCI_ERROR_UNKNOWN, oci_error_bod
 use bencher_oci_storage::OciError;
 use dropshot::{ClientErrorStatusCode, ErrorStatusCode, HttpError};
 
+/// Converts an `OciStorageError` into an `HttpError` via `OciError`
+pub fn storage_error(error: bencher_oci_storage::OciStorageError) -> HttpError {
+    into_http_error(OciError::from(error))
+}
+
 /// Converts an `OciError` into an `HttpError` with OCI-compliant JSON error body
 #[expect(
     clippy::needless_pass_by_value,

plus/api_oci/src/manifests.rs

@@ -21,6 +21,7 @@ use crate::auth::{check_oci_bandwidth, record_oci_bandwidth};
 use crate::auth::{
     require_pull_access, require_push_access, resolve_project, validate_push_access,
 };
+use crate::error::storage_error;
 use crate::response::{DOCKER_CONTENT_DIGEST, OCI_SUBJECT, oci_cors_headers};
 
 /// Parse a reference string, returning the correct OCI error code on failure.
@@ -63,10 +64,7 @@ async fn resolve_reference(
 ) -> Result<Digest, HttpError> {
     match reference {
         Reference::Digest(d) => Ok(d.clone()),
-        Reference::Tag(t) => storage
-            .resolve_tag(name, t)
-            .await
-            .map_err(|e| crate::error::into_http_error(OciError::from(e))),
+        Reference::Tag(t) => storage.resolve_tag(name, t).await.map_err(storage_error),
     }
 }
 
@@ -127,7 +125,7 @@ pub async fn oci_manifest_exists(
     let manifest = storage
         .get_manifest_by_digest(&project_uuid, &digest)
         .await
-        .map_err(|e| crate::error::into_http_error(OciError::from(e)))?;
+        .map_err(storage_error)?;
 
     // Determine content type from typed manifest
     let parsed = Manifest::from_bytes(&manifest).map_err(|e| {
@@ -188,7 +186,7 @@ pub async fn oci_manifest_get(
     let manifest = storage
         .get_manifest_by_digest(&project_uuid, &digest)
         .await
-        .map_err(|e| crate::error::into_http_error(OciError::from(e)))?;
+        .map_err(storage_error)?;
 
     // Record bandwidth usage
     #[cfg(feature = "plus")]
@@ -314,7 +312,7 @@ pub async fn oci_manifest_put(
             &parsed_manifest,
         )
         .await
-        .map_err(|e| crate::error::into_http_error(OciError::from(e)))?;
+        .map_err(storage_error)?;
 
     // Record bandwidth usage
     #[cfg(feature = "plus")]
@@ -396,7 +394,7 @@ async fn verify_referenced_blobs(
             let exists = storage
                 .blob_exists(repository, &digest)
                 .await
-                .map_err(|e| crate::error::into_http_error(OciError::from(e)))?;
+                .map_err(storage_error)?;
             if !exists {
                 return Err(crate::error::into_http_error(
                     OciError::ManifestBlobUnknown { digest: digest_str },
@@ -442,7 +440,7 @@ pub async fn oci_manifest_delete(
         Reference::Tag(tag) => storage
             .resolve_tag(&project_uuid, tag)
             .await
-            .map_err(|e| crate::error::into_http_error(OciError::from(e)))?,
+            .map_err(storage_error)?,
     };
 
     match &reference {
@@ -451,14 +449,14 @@ pub async fn oci_manifest_delete(
             storage
                 .delete_manifest(&project_uuid, digest)
                 .await
-                .map_err(|e| crate::error::into_http_error(OciError::from(e)))?;
+                .map_err(storage_error)?;
         },
         Reference::Tag(tag) => {
             // Delete by tag - delete the tag link only (manifest may still exist)
             storage
                 .delete_tag(&project_uuid, tag)
                 .await
-                .map_err(|e| crate::error::into_http_error(OciError::from(e)))?;
+                .map_err(storage_error)?;
         },
     }
 

plus/api_oci/src/referrers.rs

@@ -8,14 +8,15 @@
 use bencher_endpoint::{CorsResponse, Endpoint, Get};
 use bencher_json::ProjectResourceId;
 use bencher_json::oci::{OCI_IMAGE_INDEX_MEDIA_TYPE, OciImageIndex, OciManifestDescriptor};
-use bencher_oci_storage::{Digest, OciError};
+use bencher_oci_storage::Digest;
 use bencher_schema::context::ApiContext;
 use dropshot::{Body, HttpError, Path, Query, RequestContext, endpoint};
 use http::Response;
 use schemars::JsonSchema;
 use serde::Deserialize;
 
 use crate::auth::{require_pull_access, resolve_project};
+use crate::error::storage_error;
 use crate::response::{OCI_FILTERS_APPLIED, oci_cors_headers};
 
 /// Path parameters for referrers endpoint
@@ -85,7 +86,7 @@ pub async fn oci_referrers_list(
     let referrers = storage
         .list_referrers(&project_uuid, &digest, query.artifact_type.as_deref())
         .await
-        .map_err(|e| crate::error::into_http_error(OciError::from(e)))?;
+        .map_err(storage_error)?;
 
     // Build an OCI image index response
     // Per spec: returns application/vnd.oci.image.index.v1+json

plus/api_oci/src/tags.rs

@@ -12,6 +12,7 @@ use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 
 use crate::auth::{require_pull_access, resolve_project};
+use crate::error::storage_error;
 use crate::response::{APPLICATION_JSON, oci_cors_headers};
 
 /// Path parameters for tags list
@@ -106,7 +107,7 @@ pub async fn oci_tags_list(
     let result = storage
         .list_tags(&project_uuid, Some(page_size), last_tag)
         .await
-        .map_err(|e| crate::error::into_http_error(OciError::from(e)))?;
+        .map_err(storage_error)?;
 
     // Record metric
     #[cfg(feature = "otel")]