feat(storage): object-storage worker with S3/GCS/R2/local backends + multi-provider e2e harness

[andersonleal]

Summary

• New `storage/` worker (binary, Rust): unified object-storage RPC + triggers across AWS S3, GCS, Cloudflare R2, and a managed local backend backed by an embedded rustfs sidecar — auto-spawned and lifecycle-managed when `provider: local` is configured.
• RPC surface: `storage::putObject`, `storage::getObject`, `storage::deleteObject`, `storage::presignUrl`. Trigger types: `storage::object-created`, `storage::object-deleted` with per-provider source pollers (SQS for S3, Pub/Sub for GCS, CF Queue for R2, loopback HTTP webhook for local/rustfs).
• TypeScript e2e harness mirroring the iii-database pattern: `run-tests.sh` + engine config + `HARNESS_DONE` sentinel; covers RPC, triggers, errors, edge cases, concurrency, and provider-specific flows. `--providers=local|all|` selects which backends to exercise; `--trigger-providers=` gates trigger fan-out so R2 (no Cloudflare Queue in CI) doesn't propagate ERROR exit codes.
• Full `workers/binary-worker.md` compliance: `build.rs`, `src/manifest.rs`, `--manifest` CLI early-exit, `WorkerMetadata` registration, `config.yaml` default, `handlers::register_all` indirection, `tests/integration.rs` spec pattern A subprocess harness.
• New CI workflow `.github/workflows/storage-e2e.yml` runs the script self-tests and the harness with all three cloud providers via docker-compose (MinIO + ElasticMQ + fake-gcs-server + minio-sqs-bridge).

```
80 files, +13,701 LOC. Net new code; no existing files touched outside CI + .gitignore.
```

Pre-landing review (from `/review`)

No CRITICAL findings. Six items surfaced for follow-up — none block landing on their own, but at least the first two should be addressed before any production-ish deploy:

1. Pollers have no graceful shutdown — `storage/src/main.rs:309,343,365`. Three `tokio::spawn` JoinHandles dropped; `loop {}` runs until process exit. SQS visibility timeout makes this at-least-once safe; rustfs webhook delivery does not. Fix: `watch::channel` shutdown signal + tracked JoinHandles, ~30-40 LOC.
2. rustfs stderr pipe never drained — `storage/src/rustfs/spawn.rs:101`. `Stdio::piped()` with no consumer can deadlock the sidecar after ~64KB of stderr output. Fix: drain stderr to `tracing` (~20 LOC) or `Stdio::null()`.
3. Soft cap bypass when `content_length` missing — `storage/src/backend/s3.rs:119`. `unwrap_or(0)` makes the inline-body cap trivially passable on S3-compatible backends that omit the header (potential issue for the rustfs-backed local provider). Fix: count bytes during `collect()` and short-circuit at cap+1.
4. Webhook receiver collapses all dispatch failures to 500 — `storage/src/triggers/pollers/rustfs_webhook.rs:88-89`. Poison messages and transient errors retry-storm identically. Fix: return 200 on poison, 500 on transient.
5. No `cargo audit` step in CI — supply-chain insurance for the AWS/GCS/CF SDK + axum/reqwest dep tree.
6. `curl … | sh` engine install — documented project install path; flagging for visibility only.

Full review notes are in `~/.gstack/projects/iii-hq-workers/featiii-storage-worker-reviews.jsonl`.

Notes

• This branch is currently 9 commits behind `main`. The 3-dot diff against the merge-base is clean (80 files, all new), but a rebase before merge will surface any conflicts with the recent `acp`/`mcp` BDD work that landed in parallel (feat(acp): session/resume, set_mode, set_config_option #86, feat(mcp): add BDD testing framework and modularize HTTP bridge #88) and the BDD-framework reorganization.
• Local backend tests don't require docker; the rustfs binary is auto-fetched (with SHA256 verification, with-WARN fallback for flaky CDN) on first run and cached at `storage/target/release/rustfs`. CI uses `--providers=all` which adds MinIO/ElasticMQ/fake-gcs-server via compose.
• ACP, MCP, skills, and iii-database changes that show up in raw `git diff origin/main` (no `...`) come from `main` advancing while this branch was open — they are NOT in this PR.

Test plan

• CI `storage E2E / Script self-tests` job passes.
• CI `storage E2E / Harness (--providers=all)` job passes (note: `--trigger-providers=local,s3` — R2 trigger ERROR is expected and gated).
• Local: `cd storage && cargo test --workspace` (lib + manifest + schemas + integration) — should be 81 passed, 3 ignored.
• Local: `cd storage && cargo fmt --all -- --check` and `cargo clippy --all-targets --all-features -- -D warnings` clean.
• Local: `cd storage/tests/e2e && ./script-tests/run.sh` (bash self-tests of the orchestrator).
• Local: `cd storage/tests/e2e && ./run-tests.sh` (default `--providers=local`, no docker required).
• Manual: `./target/release/storage --manifest | jq .` returns the five required fields and exits cleanly without an engine.
• Rebase onto latest `main` and verify the diff still applies cleanly.

Summary by CodeRabbit

• New Features

  • Storage worker supporting S3, GCS, Cloudflare R2 and a local sidecar for development; put/get/delete and presigned-URL operations with input validation and a 10MiB inline read cap.
  • Object-created/object-deleted triggers with per-bucket registration and delivery/ack semantics.
  • CLI option to emit the worker manifest.

• Documentation

  • Comprehensive storage README with examples and config templates.

• Testing

  • Full e2e harness, fixtures and automated test runner for multi-provider scenarios.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: d5dacd1a-64e2-423a-b270-a5c343300793

📥 Commits

Reviewing files that changed from the base of the PR and between 025d976 and adffc71.

📒 Files selected for processing (2)

• .github/workflows/storage-e2e.yml
• .gitignore

✅ Files skipped from review due to trivial changes (1)

• .gitignore

🚧 Files skipped from review as they are similar to previous changes (1)

• .github/workflows/storage-e2e.yml

---

📝 Walkthrough

Walkthrough

Adds a new Rust storage worker crate with S3/GCS/R2/local backends, a Backend trait, RPC handlers (put/get/delete/presign), event trigger system and pollers, rustfs sidecar lifecycle, CLI/runtime wiring and manifest, README/configs, CI workflow for E2E, and extensive unit/integration/e2e tests plus a TypeScript harness and Docker fixtures.

Changes

Storage worker end-to-end feature

Layer / File(s)	Summary
All changes (single ordered review) storage/**, storage/tests/**, .github/workflows/storage-e2e.yml, .gitignore, storage/README.md	Adds the entire storage worker crate (configs, backends, rustfs sidecar, triggers, handlers, manifest, main runtime), CI workflow, documentation, e2e harness (TS runner, cases, scripts), Docker fixtures, and many tests and fixtures. See diffs for file-level details.

Sequence Diagram(s)

sequenceDiagram
  participant Client
  participant Worker
  participant Backend
  participant Poller
  participant Engine
  Client->>Worker: storage::* RPC
  Worker->>Backend: put/get/delete/presign
  Backend-->>Worker: Result/Error
  Poller-->>Worker: Provider event
  Worker->>Engine: trigger(functionId,payload,timeout)
  Engine-->>Worker: ack/null/error

Loading

Estimated code review effort

🎯 5 (Critical) | ⏱️ ~120 minutes

Suggested reviewers

• sergiofilhowz
• ytallo

Poem

I nibble through the storage stems,
From rusted burrow to cloudly hems.
Backends dance and tests take flight,
Webhooks ping into the night.
A rabbit hops — the CI bites!

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/iii-storage-worker

[coderabbitai]

Actionable comments posted: 15

Note

Due to the large number of review comments, Critical, Major severity comments were prioritized as inline comments.

🟡 Minor comments (13)

storage/README.md-54-71 (1)

54-71: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Rust SDK example incorrectly uses TriggerRequest to invoke an RPC.

iii.trigger(TriggerRequest { function_id: "storage::putObject" ... }) dispatches a trigger event, not an RPC call. Invoking storage::putObject (an RPC handler) via the trigger API will not behave as shown. The Rust SDK equivalent of the TypeScript call('storage::putObject', ...) should use a call/invoke method, not trigger.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/README.md` around lines 54 - 71, The example incorrectly uses
iii.trigger with a TriggerRequest to call the RPC handler storage::putObject;
change the example to use the Rust SDK's RPC/call/invoke method (the counterpart
to TypeScript's call('storage::putObject', ...)) instead of
TriggerRequest/iii.trigger so the RPC is invoked properly; locate usage of
TriggerRequest and iii.trigger in the README example and replace it with the
SDK's call/invoke API, passing the function id "storage::putObject" and the same
payload and timeout parameters.

storage/src/rustfs/health.rs-20-27 (1)

20-27: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Deadline can be exceeded by a full request timeout.

The loop can run a final .send() with a 2s timeout even when almost no budget remains, so max_wait is not strictly honored. Clamp each attempt timeout to remaining time.

Suggested fix

 pub async fn wait_for_healthy(port: u16, max_wait: Duration) -> Result<(), StorageError> {
     let deadline = Instant::now() + max_wait;
     let url = format!("http://127.0.0.1:{port}/");
-    let client = reqwest::Client::builder()
-        .timeout(Duration::from_secs(2))
+    let client = reqwest::Client::builder()
         .build()
         .map_err(|e| StorageError::LocalBackendBootFailed {
             reason: format!("reqwest build: {e}"),
         })?;
     while Instant::now() < deadline {
-        if let Ok(resp) = client.get(&url).send().await {
+        let remaining = deadline.saturating_duration_since(Instant::now());
+        let req_timeout = remaining.min(Duration::from_secs(2));
+        if req_timeout.is_zero() {
+            break;
+        }
+        if let Ok(resp) = client.get(&url).timeout(req_timeout).send().await {
             if resp.status().as_u16() < 500 {
                 return Ok(());
             }
         }
         sleep(Duration::from_millis(200)).await;

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/src/rustfs/health.rs` around lines 20 - 27, The loop can exceed
max_wait because each iteration calls client.get(&url).send().await with a fixed
request timeout; clamp each attempt to the remaining time by computing let
remaining = deadline.saturating_duration_since(Instant::now()) and using either
client.get(&url).timeout(remaining.min(MAX_REQ_TIMEOUT)).send().await or
wrapping the send() in tokio::time::timeout(remaining.min(MAX_REQ_TIMEOUT),
client.get(&url).send()).await, then handle the timeout/error case the same as
other errors and continue/sleep; reference the existing symbols deadline,
Instant::now(), client.get(&url).send().await, and
sleep(Duration::from_millis(...)).await.

storage/tests/e2e/workers/harness/src/cases-concurrency.ts-17-19 (1)

17-19: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Strengthen ETag assertions to catch missing response fields.

etags.size >= 1 is too weak here and can still pass when all puts return missing/undefined etag.

Suggested patch

-      const etags = new Set(puts.map((p) => p.etag as string));
-      assertTruthy(etags.size >= 1, `expected at least one etag, got: ${JSON.stringify([...etags])}`);
+      const etags = new Set(
+        puts.map((p) => p.etag).filter((e): e is string => typeof e === 'string' && e.length > 0),
+      );
+      assertTruthy(etags.size === puts.length,
+        `expected etag on every put, got ${etags.size}/${puts.length}`);

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/tests/e2e/workers/harness/src/cases-concurrency.ts` around lines 17 -
19, The current assertion using etags.size >= 1 is too weak and can pass when
many put responses are missing etag; update the check around the etags Set
created from puts.map(...) and the assertTruthy call to verify every put has a
defined, non-empty etag (e.g., assert that puts.every(p => p.etag != null &&
p.etag !== '') and that etags.size === puts.length) so missing/undefined etag
fields fail the test; refer to the etags variable, the puts array mapping, and
the assertTruthy invocation to locate and update the assertion.

storage/tests/e2e/workers/harness/src/worker.ts-17-27 (1)

17-27: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Validate provider names instead of blind casting.

as Provider[] trusts env input. Unknown provider strings should fail fast with a clear message at parse time.

Suggested patch

 const PROVIDERS_RAW = process.env.HARNESS_PROVIDERS ?? 'local';
-const PROVIDERS = PROVIDERS_RAW.split(',').map((s) => s.trim()).filter(Boolean) as Provider[];
+const ALLOWED_PROVIDERS = new Set<Provider>(['local', 's3', 'gcs', 'r2']);
+const PROVIDERS = PROVIDERS_RAW.split(',')
+  .map((s) => s.trim())
+  .filter(Boolean)
+  .map((p) => {
+    if (!ALLOWED_PROVIDERS.has(p as Provider)) {
+      throw new Error(`Invalid provider in HARNESS_PROVIDERS: ${p}`);
+    }
+    return p as Provider;
+  });
@@
 const TRIGGER_PROVIDERS: Provider[] | undefined = TRIGGER_PROVIDERS_RAW
-  ? (TRIGGER_PROVIDERS_RAW.split(',').map((s) => s.trim()).filter(Boolean) as Provider[])
+  ? TRIGGER_PROVIDERS_RAW.split(',')
+      .map((s) => s.trim())
+      .filter(Boolean)
+      .map((p) => {
+        if (!ALLOWED_PROVIDERS.has(p as Provider)) {
+          throw new Error(`Invalid provider in HARNESS_TRIGGER_PROVIDERS: ${p}`);
+        }
+        return p as Provider;
+      })
   : undefined;

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/tests/e2e/workers/harness/src/worker.ts` around lines 17 - 27, The
code blindly casts environment strings to Provider[] via PROVIDERS_RAW/PROVIDERS
and TRIGGER_PROVIDERS_RAW/TRIGGER_PROVIDERS which can hide invalid provider
names; update parsing to validate each token against the canonical set of
allowed Provider values (e.g., an enum or array of valid provider strings)
rather than using "as Provider[]", and if any token is not recognized throw a
clear, early error (with the invalid names listed) during initialization; factor
this into a small helper like parseProviders(raw: string): Provider[] and use it
for both PROVIDERS and TRIGGER_PROVIDERS to ensure fail-fast validation.

storage/tests/e2e/workers/harness/src/cases-provider.ts-17-35 (1)

17-35: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Actually validate URL parseability in the S3 presign case.

The case name/intent says “URL is parseable,” but current assertion only checks includes(bucket).

Suggested patch

         const r = await ctx.call('storage::presignUrl', {
           bucket: ctx.bucket, key, method: 'GET', expires_in_seconds: 60,
         });
+        let parsed: URL;
+        try {
+          parsed = new URL(r.url);
+        } catch {
+          throw new Error(`presigned URL is not parseable: ${r.url}`);
+        }
@@
-        assertTruthy(r.url.includes(ctx.bucket),
-          `presigned URL should reference bucket ${ctx.bucket}: ${r.url}`);
+        assertTruthy(
+          parsed.hostname.includes(ctx.bucket) || parsed.pathname.includes(`/${ctx.bucket}/`),
+          `presigned URL should reference bucket ${ctx.bucket}: ${r.url}`,
+        );

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/tests/e2e/workers/harness/src/cases-provider.ts` around lines 17 -
35, The test's presign URL assertion only checks r.url.includes(ctx.bucket) but
must actually validate URL parseability and that the bucket appears either in
hostname or pathname; update the run(ctx) for the '[s3] presigned URL is
path-style against MinIO' case to attempt parsing r.url with the URL constructor
(ensuring it doesn't throw), then assert that either parsed.hostname or
parsed.pathname contains ctx.bucket (so both virtual-host and path-style URLs
pass), using the existing r from the storage::presignUrl call to produce clear
assertion messages if parsing fails or the bucket is not found.

storage/tests/e2e/s3.rs-20-24 (1)

20-24: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Skip early when AWS credentials are unset.

This test currently skips only on S3_TEST_BUCKET; missing AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY will fail later with less actionable auth errors.

Suggested patch

 async fn s3_round_trip() {
     let Some(bucket) = require_env("S3_TEST_BUCKET") else {
         return;
     };
+    let Some(_) = require_env("AWS_ACCESS_KEY_ID") else {
+        return;
+    };
+    let Some(_) = require_env("AWS_SECRET_ACCESS_KEY") else {
+        return;
+    };
     let region = std::env::var("S3_TEST_REGION").unwrap_or_else(|_| "us-east-1".into());

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/tests/e2e/s3.rs` around lines 20 - 24, The test currently only
early-returns when S3_TEST_BUCKET is missing, causing unclear auth failures
later; update the startup checks to also require AWS_ACCESS_KEY_ID and
AWS_SECRET_ACCESS_KEY (e.g., call require_env for "AWS_ACCESS_KEY_ID" and
"AWS_SECRET_ACCESS_KEY" along with "S3_TEST_BUCKET" before constructing
BucketConfig/S3BucketConfig) and return early if either credential is unset so
the test is skipped with a clear reason.

storage/tests/e2e/fixtures/minio-sqs-bridge/index.ts-77-81 (1)

77-81: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Guard against non-object records before dereferencing

Line 78-80 assumes each Records element is object-shaped; null or primitive values can throw and return 500. Filter/guard invalid elements before reading eventName.

Suggested fix

-    const normalized = body.Records.map((rec) => {
-      const r = rec as Record<string, unknown>
+    const normalized = body.Records
+      .filter((rec): rec is Record<string, unknown> => !!rec && typeof rec === 'object')
+      .map((r) => {
       const en = typeof r.eventName === 'string' ? r.eventName : ''
       return en.startsWith('s3:') ? { ...r, eventName: en.slice('s3:'.length) } : r
-    })
+    })

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/tests/e2e/fixtures/minio-sqs-bridge/index.ts` around lines 77 - 81,
The code mapping body.Records into normalized assumes each record is a non-null
object and directly dereferences rec.eventName; update the logic around
normalized to first filter out non-object or null records (e.g., ensure typeof
rec === 'object' && rec !== null) and only then cast to Record<string, unknown>
before reading eventName, and handle missing or non-string eventName safely so
primitives/nulls are skipped or returned unchanged; change references in this
block (body.Records and normalized) accordingly.

storage/tests/e2e/workers/harness/src/runner.ts-181-187 (1)

181-187: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

resetEvents clears waiter timers without rejecting them — any in-flight waitForEvent will hang forever.

If a case (or future case) calls ctx.resetEvents() while a waitForEvent is still pending — the most likely path is a case that does Promise.all([waitForEvent(...), somethingElse()]) and then resets in a finally — the promise loses both its resolver (we drop the entry) and its timeout-rejector (we clearTimeout it), leaving the awaiter wedged indefinitely. One-line fix: reject before dropping.

🛡️ Reject pending waiters on reset

   private resetEvents(): void {
-    // Cancel any pending waiters first; their cases either already failed
-    // or are no longer interested.
-    for (const w of this.waiters) clearTimeout(w.timer);
+    // Cancel any pending waiters first; their cases either already failed
+    // or are no longer interested. Reject so awaiters don't leak.
+    for (const w of this.waiters) {
+      clearTimeout(w.timer);
+      w.reject(new Error('waitForEvent aborted by resetEvents'));
+    }
     this.waiters = [];
     this.events = [];
   }

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/tests/e2e/workers/harness/src/runner.ts` around lines 181 - 187,
resetEvents currently clears waiter timers and drops waiter entries causing any
in-flight waitForEvent promises to hang; modify resetEvents (which manipulates
this.waiters, this.events and each waiter.timer) to first reject each pending
waiter (call its rejector with a clear Error/Cancel reason), then clear its
timer, and finally clear this.waiters and this.events so no promise is left
unresolved; ensure you reference the waiter objects used by waitForEvent so
their resolve/reject callbacks are invoked before removal.

storage/src/backend/gcs.rs-200-219 (1)

200-219: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Presign URLs ignore endpoint_url and always target storage.googleapis.com.

The build-time comment (lines 60–64) acknowledges this, but the presign method itself silently returns a real-GCS URL when the bucket is configured against a custom endpoint (e.g., fake-gcs-server for the e2e harness, or any private GCS-compatible store). Callers that hand the URL back to a client will see "host not reachable" / signature mismatch errors rather than a clear configuration message.

Consider rejecting presign upfront when an override is configured, with a documented error so harness/operators get a deterministic failure rather than a runtime 4xx from the wrong host:

🛠 Suggested guard at presign entry

// At GcsBackend::build, stash endpoint_url.is_some() into a field, then:
async fn presign(&self, req: PresignReq) -> Result<PresignResp, BackendError> {
    if self.endpoint_overridden {
        return Err(BackendError::PresignUnsupported(
            "gcs presign always signs against storage.googleapis.com; \
             not supported when endpoint_url is overridden (fake-gcs-server, etc.)".into()
        ));
    }
    // ...
}

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/src/backend/gcs.rs` around lines 200 - 219, The presign method
currently always signs against storage.googleapis.com and must early-fail when a
custom endpoint is configured; add a boolean field (e.g., endpoint_overridden)
to the GcsBackend struct during GcsBackend::build to record
endpoint_url.is_some(), then modify async fn presign(&self, req: PresignReq) to
check that flag and return Err(BackendError::PresignUnsupported("gcs presign
always signs against storage.googleapis.com; not supported when endpoint_url is
overridden (fake-gcs-server, etc.)".into())) when true, otherwise proceed with
the existing signing logic.

storage/src/triggers/pollers/sqs.rs-94-104 (1)

94-104: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

delete_message errors silently dropped.

let _ = self.client.delete_message()...send().await; discards both Ok and Err. If the delete fails (network blip, expired receipt handle, permissions), the message redelivers and you get duplicate dispatches with no signal in logs that anything went wrong. A warn! here would make that visible without changing semantics.

🛠 Proposed fix

-                    let _ = self
-                        .client
-                        .delete_message()
-                        .queue_url(&self.queue_url)
-                        .receipt_handle(rh)
-                        .send()
-                        .await;
+                    if let Err(e) = self
+                        .client
+                        .delete_message()
+                        .queue_url(&self.queue_url)
+                        .receipt_handle(rh)
+                        .send()
+                        .await
+                    {
+                        tracing::warn!(error=%e, "sqs delete_message failed; message will redeliver");
+                    }

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/src/triggers/pollers/sqs.rs` around lines 94 - 104, The
delete_message() call currently discards the Result from send().await; change it
to inspect the result and emit a warn! when it fails (include the error, the
receipt_handle value and self.queue_url for context) while keeping the same
non-panicking behavior; locate the block handling msg.receipt_handle inside the
poller (where delete_message(), .queue_url(&self.queue_url) and
.receipt_handle(rh) are invoked) and replace the `let _ = ...send().await` with
a match or if let Err(e) => warn!(...) that logs e and identifying info.

storage/tests/e2e/run-tests.sh-335-353 (1)

335-353: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Health-wait loop misclassifies services without a healthcheck as unhealthy.

docker compose ps --format '{{.Service}} {{.Health}}' emits an empty {{.Health}} field for services that don't declare a healthcheck: block. The awk filter $2 != "healthy" then matches those rows and treats them as unhealthy forever, which can hang the loop until the 60s deadline and then produce a misleading FATAL listing services that never had a healthcheck to begin with.

Either require every compose service to define a healthcheck, or treat empty as "ok":

🛠 Proposed fix

     unhealthy=$(cd "$ROOT_DIR" && \
       docker compose --profile cloud ps --format '{{.Service}} {{.Health}}' \
-      | awk '$2 != "healthy" {print $1}')
+      | awk '$2 != "" && $2 != "healthy" {print $1}')

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/tests/e2e/run-tests.sh` around lines 335 - 353, The health-wait loop
misclassifies services with no healthcheck as unhealthy; update the awk filter
in the unhealthy assignment (the line that runs docker compose --profile cloud
ps --format '{{.Service}} {{.Health}}' and pipes to awk) so it ignores empty
{{.Health}} fields (treat empty as OK) — e.g. only print services whose second
field is present and not "healthy" — leaving the rest of the loop
(health_deadline, break/exit logic, and per-service logs) unchanged.

storage/src/triggers/pollers/pubsub.rs-99-106 (1)

99-106: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Silent JSON-parse fallback can produce a permanent nack loop on malformed messages.

When pm.data isn't valid JSON, data becomes Null. normalize_gcs will then return a non-Ignored error, which the dispatcher path translates to "do not ack" → Pub/Sub redelivers → same parse failure → forever. For poison messages we want to drop (ack), not retry.

Suggest logging at warn and acking on parse failure (or returning NormalizeError::Ignored upstream so the existing ack-on-Ignored branch handles it):

🛠 Proposed fix

-            let data = serde_json::from_slice::<serde_json::Value>(&pm.data)
-                .unwrap_or(serde_json::Value::Null);
+            let data = match serde_json::from_slice::<serde_json::Value>(&pm.data) {
+                Ok(v) => v,
+                Err(e) => {
+                    tracing::warn!(
+                        message_id = %pm.message_id,
+                        error = %e,
+                        "pubsub message body not JSON; acking to drop"
+                    );
+                    if let Err(e) = received.ack().await {
+                        tracing::warn!(error = %e, "pubsub ack of poison message failed");
+                    }
+                    continue;
+                }
+            };

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/src/triggers/pollers/pubsub.rs` around lines 99 - 106, The current
silent JSON-parse fallback turns malformed pm.data into Null and causes
normalize_gcs to emit a non-Ignored error that leads to endless redelivery;
replace the unwrap_or handling so a JSON parse Err triggers a warning log
(include pm.message_id and error info) and returns NormalizeError::Ignored (so
the dispatcher will ack), e.g. change the
serde_json::from_slice::<Value>(&pm.data).unwrap_or(...) usage in pubsub.rs to a
match/if let that logs warn and returns early with NormalizeError::Ignored
before constructing the envelope; this ensures normalize_gcs/dispatcher ack
poisoned messages instead of retrying forever.

storage/src/rustfs/spawn.rs-174-181 (1)

174-181: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Serialization recommended for env-var mutations in tests.

std::env::set_var/remove_var mutate process-global state. Since discover_binary() reads RUSTFS_BIN and tests run in parallel by default, this test can race with concurrent execution. Recommend serializing with the serial_test crate or using temp_env::with_var(...) for scoped, thread-safe overrides.

Note: std::env::set_var and remove_var will become unsafe in Rust 2024 edition due to inherent unsoundness in multithreaded contexts; migration to temp_env sidesteps this future concern.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/src/rustfs/spawn.rs` around lines 174 - 181, The test function
discover_binary_reads_env_var mutates global process env with
std::env::set_var/remove_var which races when tests run in parallel; update the
test (discover_binary_reads_env_var) to scope the environment change using a
thread-safe helper such as temp_env::with_var or annotate/serialize the test
using the serial_test crate (e.g., #[serial]) so RUSTFS_BIN is set only for the
test duration and automatically restored/isolated, ensuring discover_binary()
reads the temporary value without global races.

🧹 Nitpick comments (8)

storage/build.rs (1)

4-4: 💤 Low value

Prefer expect over unwrap for a more actionable build failure message.

TARGET is always set by Cargo in build scripts, so this won't panic in practice, but expect communicates intent more clearly.

🔧 Suggested improvement

-        std::env::var("TARGET").unwrap()
+        std::env::var("TARGET").expect("Cargo always sets TARGET in build scripts")

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/build.rs` at line 4, Replace the unwrap() call on
std::env::var("TARGET") in build.rs with expect() to provide a clearer,
actionable error message; update the expression where
std::env::var("TARGET").unwrap() appears so it uses expect("TARGET must be set
by Cargo build scripts") (or similar) to document intent and improve the
build-time panic message.

.github/workflows/storage-e2e.yml (1)

64-68: ⚡ Quick win

Pin the iii engine version to prevent non-reproducible CI runs.

Installing from main means every CI run may pick up a different iii engine version. A breaking change in the engine will silently flip all e2e tests from green to red with no code change in this repo.

🔧 Suggested fix

+env:
+  CARGO_TERM_COLOR: always
+  III_VERSION: '0.x.y'   # pin to a known-good release tag
+
 ...
       - name: Install iii engine
         run: |
-          curl -fsSL --retry 3 --retry-connrefused --retry-delay 5 \
-            https://install.iii.dev/iii/main/install.sh | sh
+          curl -fsSL --retry 3 --retry-connrefused --retry-delay 5 \
+            "https://install.iii.dev/iii/${III_VERSION}/install.sh" | sh
           echo "$HOME/.local/bin" >> "$GITHUB_PATH"

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In @.github/workflows/storage-e2e.yml around lines 64 - 68, The workflow step
"Install iii engine" currently pulls install.sh from the main branch which
causes non-reproducible CI; update that step to pin the installer to a specific,
committed release/tag or version (e.g., replace the "main" URL with a
release/tag or add an explicit version parameter/ENV used by the installer) so
every run installs the same iii engine version; ensure the step name "Install
iii engine" and the curl invocation are updated accordingly and include a
comment or variable (e.g., III_VERSION) so future updates require an explicit
version bump.

storage/src/triggers/dispatcher.rs (1)

36-111: ⚖️ Poor tradeoff

LGTM — ack semantics and timeout handling line up with the iii-database contract.

Worth calling out for a future revision: the per-subscriber await is sequential, so total dispatch latency for a (bucket, kind) with N subscribers is Σ(handler_timeout_ms+1s). Pollers (SQS visibility timeout, CF Queue lease) only get one ack decision for the message, so a single slow subscriber stalls the entire group. With the current 1-subscriber-typical shape this is fine, but if/when fan-out grows, switching to futures::future::join_all over the per-sub futures would bound the wall-clock at max(...) instead of sum(...).

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/src/triggers/dispatcher.rs` around lines 36 - 111, Sequential
per-subscriber awaits in dispatch cause total latency to sum per-subscriber
time; change dispatch in storage::triggers::dispatcher.rs to fire all
per-subscriber trigger futures concurrently (e.g., map subs from
registry.subscribers_for(...) to individual futures that call
self.iii.trigger(req) wrapped with the same timeout logic using
sub.handler_timeout_ms) and then await them together with
futures::future::join_all or similar, preserving the existing ack extraction
logic (null => true, value.get("ack")... ) and keeping function_id and
timeout_ms logging behavior.

storage/src/triggers/pollers/rustfs_webhook.rs (1)

32-68: ⚡ Quick win

spawn_receiver and spawn_receiver_on are line-for-line duplicates apart from the bind argument.

Both functions build the same router, channel, and axum::serve task; only the TcpListener::bind call differs. Easier to maintain (and harder to drift) if spawn_receiver delegates through spawn_receiver_on(state, 0) — 127.0.0.1:0 already gives the ephemeral-port behavior the doc-comment promises.

♻️ Collapse to one bind helper

-pub async fn spawn_receiver(state: WebhookState) -> std::io::Result<WebhookHandle> {
-    let listener = TcpListener::bind("127.0.0.1:0").await?;
-    let addr = listener.local_addr()?;
-    let app = Router::new()
-        .route("/notify", post(notify_handler))
-        .with_state(state);
-    let (shutdown_tx, shutdown_rx) = oneshot::channel();
-    tokio::spawn(async move {
-        let _ = axum::serve(listener, app)
-            .with_graceful_shutdown(async {
-                let _ = shutdown_rx.await;
-            })
-            .await;
-    });
-    Ok(WebhookHandle { addr, shutdown_tx })
-}
-
-/// Same as `spawn_receiver` but binds on a pre-allocated port. ...
-pub async fn spawn_receiver_on(state: WebhookState, port: u16) -> std::io::Result<WebhookHandle> {
-    let listener =
-        TcpListener::bind(SocketAddr::from((std::net::Ipv4Addr::LOCALHOST, port))).await?;
-    let addr = listener.local_addr()?;
-    let app = Router::new()
-        .route("/notify", post(notify_handler))
-        .with_state(state);
-    let (shutdown_tx, shutdown_rx) = oneshot::channel();
-    tokio::spawn(async move {
-        let _ = axum::serve(listener, app)
-            .with_graceful_shutdown(async {
-                let _ = shutdown_rx.await;
-            })
-            .await;
-    });
-    Ok(WebhookHandle { addr, shutdown_tx })
-}
+pub async fn spawn_receiver(state: WebhookState) -> std::io::Result<WebhookHandle> {
+    spawn_receiver_on(state, 0).await
+}
+
+/// Bind the loopback HTTP receiver on `port` (use `0` for an ephemeral port).
+pub async fn spawn_receiver_on(state: WebhookState, port: u16) -> std::io::Result<WebhookHandle> {
+    let listener =
+        TcpListener::bind(SocketAddr::from((std::net::Ipv4Addr::LOCALHOST, port))).await?;
+    let addr = listener.local_addr()?;
+    let app = Router::new()
+        .route("/notify", post(notify_handler))
+        .with_state(state);
+    let (shutdown_tx, shutdown_rx) = oneshot::channel();
+    tokio::spawn(async move {
+        let _ = axum::serve(listener, app)
+            .with_graceful_shutdown(async {
+                let _ = shutdown_rx.await;
+            })
+            .await;
+    });
+    Ok(WebhookHandle { addr, shutdown_tx })
+}

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/src/triggers/pollers/rustfs_webhook.rs` around lines 32 - 68,
spawn_receiver duplicates spawn_receiver_on except for the bind argument; change
spawn_receiver to simply call and return spawn_receiver_on(state, 0) so the
ephemeral-port behavior is delegated to spawn_receiver_on, eliminating the
duplicated router/serve/channels code (keep the same async signature and return
type for spawn_receiver and remove the duplicate implementation).

storage/src/triggers/handler.rs (1)

15-98: 💤 Low value

Two near-identical handlers — consider a single generic over EventKind.

ObjectCreatedHandler and ObjectDeletedHandler differ only in (a) the TriggerConfig newtype they parse, (b) the EventKind they register, and (c) the log string. A single generic struct (or shared private helper that takes EventKind + a parser closure) would halve the surface area and ensure the two paths can't drift (e.g., the JSON-injection fix above has to be applied twice). Optional given chill-mode and that the file is small.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/src/triggers/handler.rs` around lines 15 - 98, The two handlers
ObjectCreatedHandler and ObjectDeletedHandler duplicate
register_trigger/unregister_trigger logic; consolidate into one generic handler
to avoid drift. Replace both with a single struct (e.g., ObjectEventHandler)
parameterized by EventKind and a parser function/enum specifying which config
type to deserialize (CreatedConfig vs DeletedConfig) so register_trigger calls
the chosen parser, validates wired_buckets, calls registry.register with the
injected EventKind, and emits a log message derived from that EventKind; keep
unregister_trigger delegating to registry.unregister(&config.id) unchanged.
Implement the parser injection as a Fn(&Value) -> Result<ParsedConfig, IIIError>
or an enum branch inside the generic handler to locate logic around
register_trigger in ObjectCreatedHandler/ObjectDeletedHandler and switch to the
unified register_trigger implementation.

storage/tests/e2e/workers/harness/src/cases.ts (2)

163-168: 💤 Low value

Confusing assertion failure message.

assertEqual(del.deleted, true, 'delete returned deleted=false') reads as if the failure mode is deleted=false, but assertEqual only fires when the value differs from true. On failure the user sees the string "delete returned deleted=false" alongside the actual mismatch — fine in this specific case but inverted phrasing for an assertEqual(..., true, ...). Recommend 'expected delete.deleted=true'.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/tests/e2e/workers/harness/src/cases.ts` around lines 163 - 168, The
assertion message for the delete check is misleading: update the assertion in
the block that calls ctx.call('storage::deleteObject') so the assertEqual on
del.deleted uses a positive expectation message (e.g., "expected
del.deleted=true" or "expected delete.deleted to be true") instead of "delete
returned deleted=false"; target the assertEqual call that compares del.deleted
to true to make the message reflect the expected state.

---

41-41: 💤 Low value

Typo in JSDoc: "an storage" → "a storage".

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/tests/e2e/workers/harness/src/cases.ts` at line 41, The JSDoc above
the call property contains a typo "an storage" — update the comment to read "a
storage" (or reword to "the storage" if more appropriate) for the call:
(functionId: string, payload: unknown) => Promise<any>; identifier so the
documentation is grammatically correct.

storage/tests/e2e/run-tests.sh (1)

253-272: 💤 Low value

SHA256 verification is best-effort even when checksums file is reachable but unparseable.

If SHA256SUMS is reachable but awk finds no line for $asset, the script logs WARN ... skipping verification and continues. For a CI flow that auto-fetches a binary over the public internet, silent fallback widens the supply-chain surface — a cache poisoning that returns a different SHA256SUMS would not block download. Consider failing closed (exit 1) when the asset is expected to be listed but isn't, gated behind a RUSTFS_REQUIRE_SHA256=1 env so local dev keeps the soft path.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@storage/tests/e2e/run-tests.sh` around lines 253 - 272, The current SHA256
verification silently skips when the SHA256SUMS file is reachable but awk finds
no entry for "$asset"; change this to fail when the caller requires strict
verification by checking an env var (RUSTFS_REQUIRE_SHA256=1): in the
run-tests.sh block that computes expected="$(awk -v a="$asset" '$2==a {print
$1}' "$tmpdir/SHA256SUMS")", if [[ -z "$expected" ]] then if
RUSTFS_REQUIRE_SHA256 is set to 1 exit with an error (return 1) and log a clear
fatal message referencing "$asset" and "$tmpdir/SHA256SUMS", otherwise keep the
existing warn-and-skip behavior; preserve the existing sha256sum/shasum branch
and subsequent checks for actual vs expected.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@storage/src/backend/factory.rs`:
- Around line 38-48: S3 path-style routing is hardcoded false in factory.rs when
calling S3Backend::build via S3BackendOpts, causing S3-compatible endpoints to
fail; add a force_path_style: bool field to S3BucketConfig
(storage/src/config.rs), read it where configs are loaded, and pass that value
into S3BackendOpts.force_path_style in factory.rs instead of the literal false
(update any call sites such as the one inside the S3Backend::build invocation
and ensure tests like build_with_endpoint_url_does_not_panic use the new config
field).

In `@storage/src/backend/local.rs`:
- Around line 46-58: The current logic treats any head_bucket() failure as
"bucket missing" and always calls create_bucket(); modify the flow so
head_bucket().send().await is matched: if Ok(_) return Ok(()); if Err(e) inspect
the error and only proceed to client.create_bucket() when the error indicates a
definite NotFound/NoSuchBucket (HTTP 404 or the provider-specific NotFound
variant); for any other error (auth, connection, etc.) immediately map and
return a BackendError::Provider with the original error info (preserve e in the
message), and keep the create_bucket() call and its error mapping unchanged so
creation only happens on explicit not-found errors.
- Line 92: The call to std::fs::create_dir_all(queue_dir) currently ignores
errors and should fail fast; update the code that constructs the queue_dir (the
create_dir_all(queue_dir) call) to handle the Result instead of discarding
it—propagate the error (use the function's error return path) or return an Err
with context so registration stops when the directory cannot be created; ensure
you use the existing error type or map the std::io::Error into the function's
error type so callers (and tests) see a clear failure instead of silently
continuing.

In `@storage/src/backend/s3.rs`:
- Around line 121-141: The code currently treats
resp.content_length.unwrap_or(0) as 0 which lets objects with unknown length
bypass req.max_inline_bytes and then be fully buffered by resp.body.collect();
change the logic so that before calling resp.body.collect() you first check if
req.max_inline_bytes.is_some() and if resp.content_length.is_none() fail closed
(return a BackendError::ObjectTooLarge or a clear Provider error) OR
alternatively enforce the cap while streaming (read resp.body incrementally and
abort with BackendError::ObjectTooLarge when the cumulative bytes exceed cap);
update the branch around resp.content_length, the req.max_inline_bytes check,
and the resp.body.collect() usage to implement one of these two approaches.

In `@storage/src/handlers/put_object.rs`:
- Line 16: The PutObject request struct only accepts body_base64 which breaks
compatibility with clients sending body; update the PutObject request field (the
struct containing pub body_base64: String, e.g., PutObjectRequest) to accept the
legacy key by adding a serde alias for "body" on that field (use #[serde(alias =
"body")] or equivalent deserialization attribute) so incoming payloads with
"body" or "body_base64" both deserialize correctly; run/update relevant tests to
cover both keys.
- Around line 37-47: Reject oversized payloads before allocating by checking the
base64 string length instead of decoding first: in put_object.rs, before calling
B64.decode(&req.body_base64) compute the maximum decoded byte length from
req.body_base64 (e.g. (req.body_base64.len() * 3) / 4 minus padding) and compare
against INLINE_BODY_CAP, returning the StorageError::BodyTooLarge via err_to_str
if it exceeds cap; only call B64.decode(&req.body_base64) afterwards and keep
existing InvalidBase64 handling for decode errors.

In `@storage/src/triggers/handler.rs`:
- Around line 26-36: The error JSON is being constructed by interpolating
Display values (serde_json::Error and cfg.bucket) into a raw string which can
produce malformed JSON; instead build the payload as a serde_json value and
serialize it so values are properly escaped — replace the raw r#"...{e}..."# and
r#"...{}`..."# constructions used when returning IIIError::Handler in the
object-created handler (where CreatedConfig is parsed and cfg.bucket is checked)
with a serde_json::json! or serde_json::Map that sets
{"code":"CONFIG_ERROR","message": <string>} and call to_string() for the error
envelope; apply the identical fix to the ObjectDeletedHandler::register_trigger
block (the duplicate at lines ~67–77) so both handlers produce safe JSON-escaped
messages.

In `@storage/src/triggers/normalize.rs`:
- Around line 18-19: The decode_s3_key function currently only percent-decodes
%XX sequences and misses S3-style '+' space encoding; update decode_s3_key to
first replace all '+' characters with space characters (or decode '+' to ' ')
before calling percent_decode_str so inputs like "reports+2026.pdf" normalize
correctly and match RPC-sourced keys; reference the decode_s3_key function and
ensure the replacement occurs prior to percent_decode_str().

In `@storage/src/triggers/pollers/cf_queue.rs`:
- Around line 191-197: probe_auth currently only errors on 401/403 but returns
Ok(()) for 404/429/5xx; change it to treat any non-2xx as a startup failure by
checking resp.status().is_success() and returning an Err when false. Keep the
existing special-case for reqwest::StatusCode::UNAUTHORIZED / FORBIDDEN
(StorageError::CfQueueAuthFailed with the current message), and for all other
non-success statuses return a suitable StorageError (e.g., a generic
CfQueueAuthFailed or a new variant) that includes the HTTP status and response
body or context so startup fails when the probe returns 404/429/5xx. Ensure you
update probe_auth to use resp.status().is_success(), reference resp and status,
and include the status details in the error message.
- Around line 115-124: The match on normalize_r2 currently drops non-ignored
errors without acking, causing permanent malformed (poison) messages to be
retried forever; update the match in the poller handling (the normalize_r2(...)
match block) so that for non-ignored/unmapped and for any other Err(_) cases you
log the error with context (payload and NormalizeError) and push lease_id into
acked_lease_ids (or otherwise dead-letter it) before continue; ensure you still
preserve the existing behavior for NormalizeError::Ignored and
NormalizeError::UnmappedBucket where acking is already done, and only add
ack+log for permanently malformed cases so the lease is removed instead of
retried.

In `@storage/src/triggers/pollers/sqs.rs`:
- Around line 71-81: The normalize_s3 call currently continues the SQS message
loop on NormalizeError::Ignored(_) and NormalizeError::UnmappedBucket(_),
leaving the SQS message un-acked and causing infinite redelivery; change those
branches to ack-and-drop the SQS message instead of continuing: mirror
pubsub.rs::dispatch_one behavior by performing the same acknowledgement (invoke
the SQS delete/ack path or return the same "ack" sentinel used by this poller)
when encountering NormalizeError::Ignored and NormalizeError::UnmappedBucket,
and keep the existing tracing/logging; leave the other Err(e) branch as-is.

In `@storage/tests/e2e/fixtures/minio-sqs-bridge/Dockerfile`:
- Around line 3-23: The Dockerfile runs the Bun service as root; add an explicit
non-root runtime user and switch to it before CMD to harden the container: in
the Dockerfile create a dedicated user/group (e.g., app or bunuser) after
setting WORKDIR, ensure ownership of /app and any copied files is changed to
that user (chown), and then set USER to that non-root account so HEALTHCHECK and
CMD ["bun","run","index.ts"] run unprivileged; reference the Dockerfile's
WORKDIR, COPY steps, HEALTHCHECK, and CMD when applying the changes.

In `@storage/tests/e2e/local.rs`:
- Around line 22-37: The test currently calls spawn::spawn(...) and stores
handle but uses expect(...) on subsequent awaits (health::wait_for_healthy and
storage::backend::local::ensure_bucket) so a panic will skip cleaning up the
child; modify the test to ensure spawn::shutdown(handle) always runs on all
failure paths by converting the sequence to return a Result and using match/if
let Err(_) or a finally-like block (e.g., let res = (async { ...
health::wait_for_healthy(...).await?;
storage::backend::local::ensure_bucket(...).await?; Ok(()) }).await; if
res.is_err() { spawn::shutdown(handle).await.expect("shutdown"); } else {
spawn::shutdown(handle).await.expect("shutdown"); } ), or wrap handle in a guard
type that calls spawn::shutdown in Drop; apply the same pattern to the other
occurrence referenced (lines ~81-82) so the child process is always shut down.

In `@storage/tests/e2e/workers/harness/src/cases.ts`:
- Around line 7-15: The harness omits the new GCS backend: update the Provider
union type (Provider), the ALL_PROVIDERS array, and the BUCKETS map to include
'gcs' and a corresponding bucket name (e.g., 'scratch-gcs') so the TypeScript
harness will exercise GCS; also add a fake-gcs service to docker-compose.yml
(configure an appropriate emulator image/ports and ensure tests can reach it) so
the harness can run GCS-backed tests.

In `@storage/tests/integration.rs`:
- Around line 58-64: When spawning the worker Child fails, the earlier
background process named iii is not cleaned up; change the worker spawn handling
in boot() from using .spawn().ok()? to explicitly handle the Result and on Err
call iii.kill().ok() (or iii.kill().await/iii.wait().ok() as appropriate) before
returning None so the iii Child is terminated and no subprocess is leaked;
reference the worker Command::new(...).spawn() call and the iii Child variable
in your fix.

---

Minor comments:
In `@storage/README.md`:
- Around line 54-71: The example incorrectly uses iii.trigger with a
TriggerRequest to call the RPC handler storage::putObject; change the example to
use the Rust SDK's RPC/call/invoke method (the counterpart to TypeScript's
call('storage::putObject', ...)) instead of TriggerRequest/iii.trigger so the
RPC is invoked properly; locate usage of TriggerRequest and iii.trigger in the
README example and replace it with the SDK's call/invoke API, passing the
function id "storage::putObject" and the same payload and timeout parameters.

In `@storage/src/backend/gcs.rs`:
- Around line 200-219: The presign method currently always signs against
storage.googleapis.com and must early-fail when a custom endpoint is configured;
add a boolean field (e.g., endpoint_overridden) to the GcsBackend struct during
GcsBackend::build to record endpoint_url.is_some(), then modify async fn
presign(&self, req: PresignReq) to check that flag and return
Err(BackendError::PresignUnsupported("gcs presign always signs against
storage.googleapis.com; not supported when endpoint_url is overridden
(fake-gcs-server, etc.)".into())) when true, otherwise proceed with the existing
signing logic.

In `@storage/src/rustfs/health.rs`:
- Around line 20-27: The loop can exceed max_wait because each iteration calls
client.get(&url).send().await with a fixed request timeout; clamp each attempt
to the remaining time by computing let remaining =
deadline.saturating_duration_since(Instant::now()) and using either
client.get(&url).timeout(remaining.min(MAX_REQ_TIMEOUT)).send().await or
wrapping the send() in tokio::time::timeout(remaining.min(MAX_REQ_TIMEOUT),
client.get(&url).send()).await, then handle the timeout/error case the same as
other errors and continue/sleep; reference the existing symbols deadline,
Instant::now(), client.get(&url).send().await, and
sleep(Duration::from_millis(...)).await.

In `@storage/src/rustfs/spawn.rs`:
- Around line 174-181: The test function discover_binary_reads_env_var mutates
global process env with std::env::set_var/remove_var which races when tests run
in parallel; update the test (discover_binary_reads_env_var) to scope the
environment change using a thread-safe helper such as temp_env::with_var or
annotate/serialize the test using the serial_test crate (e.g., #[serial]) so
RUSTFS_BIN is set only for the test duration and automatically
restored/isolated, ensuring discover_binary() reads the temporary value without
global races.

In `@storage/src/triggers/pollers/pubsub.rs`:
- Around line 99-106: The current silent JSON-parse fallback turns malformed
pm.data into Null and causes normalize_gcs to emit a non-Ignored error that
leads to endless redelivery; replace the unwrap_or handling so a JSON parse Err
triggers a warning log (include pm.message_id and error info) and returns
NormalizeError::Ignored (so the dispatcher will ack), e.g. change the
serde_json::from_slice::<Value>(&pm.data).unwrap_or(...) usage in pubsub.rs to a
match/if let that logs warn and returns early with NormalizeError::Ignored
before constructing the envelope; this ensures normalize_gcs/dispatcher ack
poisoned messages instead of retrying forever.

In `@storage/src/triggers/pollers/sqs.rs`:
- Around line 94-104: The delete_message() call currently discards the Result
from send().await; change it to inspect the result and emit a warn! when it
fails (include the error, the receipt_handle value and self.queue_url for
context) while keeping the same non-panicking behavior; locate the block
handling msg.receipt_handle inside the poller (where delete_message(),
.queue_url(&self.queue_url) and .receipt_handle(rh) are invoked) and replace the
`let _ = ...send().await` with a match or if let Err(e) => warn!(...) that logs
e and identifying info.

In `@storage/tests/e2e/fixtures/minio-sqs-bridge/index.ts`:
- Around line 77-81: The code mapping body.Records into normalized assumes each
record is a non-null object and directly dereferences rec.eventName; update the
logic around normalized to first filter out non-object or null records (e.g.,
ensure typeof rec === 'object' && rec !== null) and only then cast to
Record<string, unknown> before reading eventName, and handle missing or
non-string eventName safely so primitives/nulls are skipped or returned
unchanged; change references in this block (body.Records and normalized)
accordingly.

In `@storage/tests/e2e/run-tests.sh`:
- Around line 335-353: The health-wait loop misclassifies services with no
healthcheck as unhealthy; update the awk filter in the unhealthy assignment (the
line that runs docker compose --profile cloud ps --format '{{.Service}}
{{.Health}}' and pipes to awk) so it ignores empty {{.Health}} fields (treat
empty as OK) — e.g. only print services whose second field is present and not
"healthy" — leaving the rest of the loop (health_deadline, break/exit logic, and
per-service logs) unchanged.

In `@storage/tests/e2e/s3.rs`:
- Around line 20-24: The test currently only early-returns when S3_TEST_BUCKET
is missing, causing unclear auth failures later; update the startup checks to
also require AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY (e.g., call require_env
for "AWS_ACCESS_KEY_ID" and "AWS_SECRET_ACCESS_KEY" along with "S3_TEST_BUCKET"
before constructing BucketConfig/S3BucketConfig) and return early if either
credential is unset so the test is skipped with a clear reason.

In `@storage/tests/e2e/workers/harness/src/cases-concurrency.ts`:
- Around line 17-19: The current assertion using etags.size >= 1 is too weak and
can pass when many put responses are missing etag; update the check around the
etags Set created from puts.map(...) and the assertTruthy call to verify every
put has a defined, non-empty etag (e.g., assert that puts.every(p => p.etag !=
null && p.etag !== '') and that etags.size === puts.length) so missing/undefined
etag fields fail the test; refer to the etags variable, the puts array mapping,
and the assertTruthy invocation to locate and update the assertion.

In `@storage/tests/e2e/workers/harness/src/cases-provider.ts`:
- Around line 17-35: The test's presign URL assertion only checks
r.url.includes(ctx.bucket) but must actually validate URL parseability and that
the bucket appears either in hostname or pathname; update the run(ctx) for the
'[s3] presigned URL is path-style against MinIO' case to attempt parsing r.url
with the URL constructor (ensuring it doesn't throw), then assert that either
parsed.hostname or parsed.pathname contains ctx.bucket (so both virtual-host and
path-style URLs pass), using the existing r from the storage::presignUrl call to
produce clear assertion messages if parsing fails or the bucket is not found.

In `@storage/tests/e2e/workers/harness/src/runner.ts`:
- Around line 181-187: resetEvents currently clears waiter timers and drops
waiter entries causing any in-flight waitForEvent promises to hang; modify
resetEvents (which manipulates this.waiters, this.events and each waiter.timer)
to first reject each pending waiter (call its rejector with a clear Error/Cancel
reason), then clear its timer, and finally clear this.waiters and this.events so
no promise is left unresolved; ensure you reference the waiter objects used by
waitForEvent so their resolve/reject callbacks are invoked before removal.

In `@storage/tests/e2e/workers/harness/src/worker.ts`:
- Around line 17-27: The code blindly casts environment strings to Provider[]
via PROVIDERS_RAW/PROVIDERS and TRIGGER_PROVIDERS_RAW/TRIGGER_PROVIDERS which
can hide invalid provider names; update parsing to validate each token against
the canonical set of allowed Provider values (e.g., an enum or array of valid
provider strings) rather than using "as Provider[]", and if any token is not
recognized throw a clear, early error (with the invalid names listed) during
initialization; factor this into a small helper like parseProviders(raw:
string): Provider[] and use it for both PROVIDERS and TRIGGER_PROVIDERS to
ensure fail-fast validation.

---

Nitpick comments:
In @.github/workflows/storage-e2e.yml:
- Around line 64-68: The workflow step "Install iii engine" currently pulls
install.sh from the main branch which causes non-reproducible CI; update that
step to pin the installer to a specific, committed release/tag or version (e.g.,
replace the "main" URL with a release/tag or add an explicit version
parameter/ENV used by the installer) so every run installs the same iii engine
version; ensure the step name "Install iii engine" and the curl invocation are
updated accordingly and include a comment or variable (e.g., III_VERSION) so
future updates require an explicit version bump.

In `@storage/build.rs`:
- Line 4: Replace the unwrap() call on std::env::var("TARGET") in build.rs with
expect() to provide a clearer, actionable error message; update the expression
where std::env::var("TARGET").unwrap() appears so it uses expect("TARGET must be
set by Cargo build scripts") (or similar) to document intent and improve the
build-time panic message.

In `@storage/src/triggers/dispatcher.rs`:
- Around line 36-111: Sequential per-subscriber awaits in dispatch cause total
latency to sum per-subscriber time; change dispatch in
storage::triggers::dispatcher.rs to fire all per-subscriber trigger futures
concurrently (e.g., map subs from registry.subscribers_for(...) to individual
futures that call self.iii.trigger(req) wrapped with the same timeout logic
using sub.handler_timeout_ms) and then await them together with
futures::future::join_all or similar, preserving the existing ack extraction
logic (null => true, value.get("ack")... ) and keeping function_id and
timeout_ms logging behavior.

In `@storage/src/triggers/handler.rs`:
- Around line 15-98: The two handlers ObjectCreatedHandler and
ObjectDeletedHandler duplicate register_trigger/unregister_trigger logic;
consolidate into one generic handler to avoid drift. Replace both with a single
struct (e.g., ObjectEventHandler) parameterized by EventKind and a parser
function/enum specifying which config type to deserialize (CreatedConfig vs
DeletedConfig) so register_trigger calls the chosen parser, validates
wired_buckets, calls registry.register with the injected EventKind, and emits a
log message derived from that EventKind; keep unregister_trigger delegating to
registry.unregister(&config.id) unchanged. Implement the parser injection as a
Fn(&Value) -> Result<ParsedConfig, IIIError> or an enum branch inside the
generic handler to locate logic around register_trigger in
ObjectCreatedHandler/ObjectDeletedHandler and switch to the unified
register_trigger implementation.

In `@storage/src/triggers/pollers/rustfs_webhook.rs`:
- Around line 32-68: spawn_receiver duplicates spawn_receiver_on except for the
bind argument; change spawn_receiver to simply call and return
spawn_receiver_on(state, 0) so the ephemeral-port behavior is delegated to
spawn_receiver_on, eliminating the duplicated router/serve/channels code (keep
the same async signature and return type for spawn_receiver and remove the
duplicate implementation).

In `@storage/tests/e2e/run-tests.sh`:
- Around line 253-272: The current SHA256 verification silently skips when the
SHA256SUMS file is reachable but awk finds no entry for "$asset"; change this to
fail when the caller requires strict verification by checking an env var
(RUSTFS_REQUIRE_SHA256=1): in the run-tests.sh block that computes
expected="$(awk -v a="$asset" '$2==a {print $1}' "$tmpdir/SHA256SUMS")", if [[
-z "$expected" ]] then if RUSTFS_REQUIRE_SHA256 is set to 1 exit with an error
(return 1) and log a clear fatal message referencing "$asset" and
"$tmpdir/SHA256SUMS", otherwise keep the existing warn-and-skip behavior;
preserve the existing sha256sum/shasum branch and subsequent checks for actual
vs expected.

In `@storage/tests/e2e/workers/harness/src/cases.ts`:
- Around line 163-168: The assertion message for the delete check is misleading:
update the assertion in the block that calls ctx.call('storage::deleteObject')
so the assertEqual on del.deleted uses a positive expectation message (e.g.,
"expected del.deleted=true" or "expected delete.deleted to be true") instead of
"delete returned deleted=false"; target the assertEqual call that compares
del.deleted to true to make the message reflect the expected state.
- Line 41: The JSDoc above the call property contains a typo "an storage" —
update the comment to read "a storage" (or reword to "the storage" if more
appropriate) for the call: (functionId: string, payload: unknown) =>
Promise<any>; identifier so the documentation is grammatically correct.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes