Derive file S3 location server-side on upload

developer-docs/watcher.md

@@ -215,7 +215,7 @@ Upgrading an existing watcher is unaffected: the environment's database already
 - **`auto`**: Files are uploaded to S3 immediately after run detection.
 - **`manual`**: Runs are reported to the API without uploading. The server decides which files to upload via a queue, polled by the upload worker thread every 60 seconds. Useful when uploads need human approval.
 
-Queued files are resolved against the current `watch_directory` (each queue entry carries a `relative_path` anchored to the root that was active when the file was detected). Two safeguards keep a stale queue entry from erroring forever (ENG-1397):
+Queued files are resolved against the current `watch_directory` (each queue entry carries a `relative_path` anchored to the root that was active when the file was detected). Two safeguards keep a stale queue entry from erroring forever:
 
 - **On `watch_directory` change**: the server reverts every pending upload request for that instrument back to `detected` (clearing `upload_requested_at`) as soon as the new config is pushed, so the queue drains immediately. The reverted files remain re-requestable detections; an operator can queue them again from their new location.
 - **Per-file 3-try cap (`MAX_QUEUE_FILE_ATTEMPTS`)**: a queued file that keeps failing — missing on disk or failing to upload — is retried on at most three upload-queue polls. After that the watcher cancels the request server-side (revert to `detected`) so the file leaves the queue instead of re-erroring each poll. The attempt count resets on watcher restart, so a transient outage longer than three polls is recovered on the next start.

watcher/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "data-hub-watcher"
-version = "0.5.0"
+version = "0.5.1"
 description = "File-watcher agent for lab instrument PCs that ingests data into Data Hub."
 readme = "README.md"
 requires-python = ">=3.12"

watcher/src/data_hub_watcher/api_client.py

@@ -232,8 +232,8 @@ def request_upload_url(
         )
         return PresignedUploadResponse.model_validate(resp.json())
 
-    def mark_file_uploaded(self, file_id: int, s3_info: dict[str, Any]) -> FileResponse:
-        resp = self._request("PATCH", f"/files/{file_id}", json=s3_info)
+    def mark_file_uploaded(self, file_id: int, updates: dict[str, Any]) -> FileResponse:
+        resp = self._request("PATCH", f"/files/{file_id}", json=updates)
         return FileResponse.model_validate(resp.json())
 
     def cancel_upload_request(self, file_id: int) -> FileResponse:
@@ -242,7 +242,7 @@ def cancel_upload_request(self, file_id: int) -> FileResponse:
         Called after the watcher gives up on a queued file (missing on disk
         or persistently failing to upload) so the server stops serving it in
         the upload queue and the watcher stops re-erroring on it every
-        heartbeat poll (ENG-1397). The file stays a re-requestable detection
+        heartbeat poll. The file stays a re-requestable detection
         rather than being deleted, so an operator can queue it again later.
         """
         resp = self._request("PATCH", f"/files/{file_id}", json={"status": "detected"})

watcher/src/data_hub_watcher/constants.py

@@ -120,7 +120,7 @@ def _resolve_watcher_log_dir() -> Path:
 # (missing or failing to upload) before the watcher gives up and cancels the
 # request server-side. Distinct from `UPLOAD_RETRY_MAX` (per-upload S3 PUT
 # retries). ~3 min at the 60s heartbeat: long enough to ride out a blip,
-# short enough that a stale entry from a dir change self-clears (ENG-1397).
+# short enough that a stale entry from a dir change self-clears.
 MAX_QUEUE_FILE_ATTEMPTS = 3
 # Upload records older than this are pruned from the local state DB
 # to prevent unbounded growth on long-running watcher instances.

watcher/src/data_hub_watcher/uploader.py

@@ -574,13 +574,12 @@ def _upload_single(self, path: Path, run_id: str) -> bool:
             return False
 
         # Notify API — treat a failed PATCH as an upload failure so the file
-        # is not recorded in the dedup DB and will be retried next time.
+        # is not recorded in the dedup DB and will be retried next time. The
+        # server derives the S3 location itself, so we only send status/type.
         try:
             self._client.mark_file_uploaded(
                 file_id,
                 {
-                    "s3_bucket": s3_bucket,
-                    "s3_key": s3_key,
                     "content_type": content_type,
                     "status": "uploaded",
                 },

watcher/tests/integration/test_auto_upload_flow.py

@@ -222,15 +222,14 @@ def test_mark_file_uploaded_transitions_detected_to_uploaded(
         result = client.mark_file_uploaded(
             file_id,
             {
-                "s3_bucket": "test-bucket",
-                "s3_key": f"{instrument_id}/EXP-001/data_001.csv",
                 "content_type": "text/csv",
                 "status": "uploaded",
             },
         )
         assert result.status == "uploaded"
-        assert result.s3_bucket == "test-bucket"
+        # The server derives the location from the run's key and the filename.
         assert result.s3_key == f"{instrument_id}/EXP-001/data_001.csv"
+        assert result.s3_bucket
         assert result.content_type == "text/csv"
         assert result.uploaded_at is not None
 
@@ -277,8 +276,6 @@ def test_request_upload_url_already_uploaded(
         client.mark_file_uploaded(
             file_id,
             {
-                "s3_bucket": "test-bucket",
-                "s3_key": f"{instrument_id}/EXP-001/data_001.csv",
                 "content_type": "text/csv",
                 "status": "uploaded",
             },
@@ -326,17 +323,18 @@ def test_queued_file_gets_presigned_url_and_marks_uploaded(
         assert presigned.file_id == file_id
         assert presigned.already_uploaded is False
 
-        # After uploading to S3, the watcher notifies the API.
+        # After uploading to S3, the watcher notifies the API. The server
+        # derives the S3 location itself, so the watcher only sends status/type.
         result = client.mark_file_uploaded(
             presigned.file_id,
             {
-                "s3_bucket": presigned.s3_bucket,
-                "s3_key": presigned.s3_key,
                 "content_type": "text/csv",
                 "status": "uploaded",
             },
         )
         assert result.status == "uploaded"
+        assert result.s3_bucket == presigned.s3_bucket
+        assert result.s3_key == presigned.s3_key
 
 
 # ------------------------------------------------------------------
@@ -377,8 +375,6 @@ def test_full_auto_mode_lifecycle(
         uploaded = client.mark_file_uploaded(
             file_id,
             {
-                "s3_bucket": "test-bucket",
-                "s3_key": f"{instrument_id}/LIFECYCLE-001/raw.csv",
                 "content_type": "text/csv",
                 "status": "uploaded",
             },

watcher/tests/integration/test_error_paths.py

@@ -44,8 +44,6 @@ def test_mark_uploaded_nonexistent_file_404(self, client: DataHubClient) -> None
             client.mark_file_uploaded(
                 99999,
                 {
-                    "s3_bucket": "b",
-                    "s3_key": "k",
                     "content_type": "text/csv",
                     "status": "uploaded",
                 },
@@ -144,16 +142,14 @@ def test_mark_uploaded_invalid_transition_409(
         )
         file_id = _get_file_id(integration_env.db_dsn, "EXP-CONFLICT", "f.csv")
 
-        s3_info = {
-            "s3_bucket": "b",
-            "s3_key": "k",
+        updates = {
             "content_type": "text/csv",
             "status": "uploaded",
         }
-        client.mark_file_uploaded(file_id, s3_info)
+        client.mark_file_uploaded(file_id, updates)
 
         with pytest.raises(ApiError) as exc_info:
-            client.mark_file_uploaded(file_id, s3_info)
+            client.mark_file_uploaded(file_id, updates)
         assert exc_info.value.status_code == 409
 
     def test_update_deleted_run_409(

watcher/tests/integration/test_manual_upload_flow.py

@@ -105,8 +105,6 @@ def test_uploaded_file_leaves_queue(
         client.mark_file_uploaded(
             file_id,
             {
-                "s3_bucket": "test-bucket",
-                "s3_key": f"{instrument_id}/MANUAL-001/data.csv",
                 "content_type": "text/csv",
                 "status": "uploaded",
             },

watcher/tests/test_uploader.py

@@ -79,11 +79,11 @@ def test_successful_upload(
 
         assert result is True
         mock_put.assert_called_once()
+        # The server derives the S3 location itself, so the watcher only
+        # reports the status and content type.
         mock_client.mark_file_uploaded.assert_called_once_with(
             42,
             {
-                "s3_bucket": "test-bucket",
-                "s3_key": "test-instrument/RUN-001/test_data.csv",
                 "content_type": "text/csv",
                 "status": "uploaded",
             },
@@ -603,7 +603,7 @@ class TestPollUploadQueueAttemptCap:
     a file that keeps failing to upload) would otherwise re-error forever.
     The watcher surfaces the visible error once, retries up to
     ``MAX_QUEUE_FILE_ATTEMPTS`` polls, then cancels the request server-side
-    so it leaves the queue (ENG-1397).
+    so it leaves the queue.
     """
 
     @staticmethod

web/app/api/v1/files/[fileId]/route.ts

@@ -5,13 +5,15 @@ import {
   apiError,
   apiErrorFromResult,
   CONFLICT,
+  INTERNAL_ERROR,
   NOT_FOUND,
   VALIDATION_ERROR,
 } from "@/lib/api/errors";
 import { dismissFile } from "@/lib/api/files";
 import { patchFileBody, readJsonBody } from "@/lib/api/openapi";
 import { db } from "@/lib/db";
 import { files, instrumentRuns } from "@/lib/db/schema";
+import { getS3RawDataBucket } from "@/lib/s3";
 
 interface RouteContext {
   params: Promise<{ fileId: string }>;
@@ -23,7 +25,7 @@ interface RouteContext {
 //   Reprocessing:   completed|failed → processing → completed|failed
 //   Cancel request: upload_requested → detected (watcher gave up locating
 //                   the local file after repeated polls; clears the queue
-//                   entry — see ENG-1397)
+//                   entry)
 const VALID_TRANSITIONS: Record<string, string[]> = {
   detected: ["uploaded"],
   upload_requested: ["uploaded", "detected"],
@@ -70,9 +72,14 @@ export async function PATCH(request: NextRequest, { params }: RouteContext) {
     return apiError(409, CONFLICT, "Cannot update a soft-deleted file");
   }
 
-  // Verify the parent run is not soft-deleted.
+  // Verify the parent run is not soft-deleted. Its natural key also feeds the
+  // server-derived S3 location on the `uploaded` transition below.
   const [parentRun] = await db
-    .select({ deletedAt: instrumentRuns.deletedAt })
+    .select({
+      deletedAt: instrumentRuns.deletedAt,
+      instrumentId: instrumentRuns.instrumentId,
+      runId: instrumentRuns.runId,
+    })
     .from(instrumentRuns)
     .where(eq(instrumentRuns.id, file.instrumentRunId))
     .limit(1);
@@ -109,6 +116,25 @@ export async function PATCH(request: NextRequest, { params }: RouteContext) {
 
     if (body.status === "uploaded") {
       updates.uploadedAt = now;
+
+      // Derive the S3 location from trusted DB state, not the request: the
+      // watcher only echoed back what `request-upload-url` already computed,
+      // and accepting it let any caller repoint a file at an arbitrary object.
+      if (!parentRun) {
+        return apiError(404, NOT_FOUND, `File '${fileId}' not found`);
+      }
+      let bucket: string;
+      try {
+        bucket = getS3RawDataBucket();
+      } catch {
+        return apiError(
+          500,
+          INTERNAL_ERROR,
+          "S3 bucket configuration is missing"
+        );
+      }
+      updates.s3Bucket = bucket;
+      updates.s3Key = `${parentRun.instrumentId}/${parentRun.runId}/${file.filename}`;
     }
     // Reverting a pending request back to detected (watcher cancel): clear
     // upload_requested_at so the row leaves the upload queue, whose filter
@@ -128,13 +154,6 @@ export async function PATCH(request: NextRequest, { params }: RouteContext) {
     }
   }
 
-  // S3 info — set when transitioning to "uploaded" (watcher path).
-  if (body.s3_bucket !== undefined) {
-    updates.s3Bucket = body.s3_bucket;
-  }
-  if (body.s3_key !== undefined) {
-    updates.s3Key = body.s3_key;
-  }
   if (body.content_type !== undefined) {
     updates.contentType = body.content_type;
   }

web/app/api/v1/watchers/[watcherId]/config/route.ts

@@ -49,7 +49,7 @@ export async function PUT(
 
   // A watch_directory change orphans every pending request: each carries a
   // relative_path under the old root that no longer resolves. Revert them to
-  // `detected` to drain the queue instead of erroring each poll (ENG-1397).
+  // `detected` to drain the queue instead of erroring each poll.
   // Gated on a known previous dir so first-push / unrelated edits don't
   // revert spuriously.
   if (previousWatchDir && nextWatchDir && previousWatchDir !== nextWatchDir) {

web/lib/api/openapi/schemas/files.ts

@@ -10,10 +10,11 @@ export const createFileBody = z.object({
   category: fileCategorySchema.optional(),
 });
 
+// No `s3_bucket` / `s3_key`: the server derives the canonical S3 location on
+// the `uploaded` transition, so accepting them from the client only let a
+// caller repoint a file at an arbitrary object.
 export const patchFileBody = z.object({
   status: fileStatusSchema.optional(),
-  s3_bucket: z.string().optional(),
-  s3_key: z.string().optional(),
   content_type: z.string().optional(),
   size_bytes: z.number().optional(),
   metadata: z.record(z.string(), z.unknown()).optional(),

web/lib/api/watchers.ts

@@ -53,8 +53,8 @@ export function extractWatchDirectory(
  *
  * Called when a watcher's `watch_directory` changes: every queued file's
  * `relative_path` was anchored to the old root, so none resolve under the
- * new one and the watcher would otherwise re-error on each heartbeat poll
- * (ENG-1397). Returns the reverted file ids (for event reporting).
+ * new one and the watcher would otherwise re-error on each heartbeat poll.
+ * Returns the reverted file ids (for event reporting).
  */
 export async function revertPendingUploadRequests(
   instrumentId: string,

web/tests/integration/files.test.ts

@@ -289,24 +289,26 @@ describe("Files API", () => {
   // PATCH /api/v1/files/:fileId — watcher path (detected → uploaded)
   // -------------------------------------------------------------------------
 
-  // Watcher path: after the watcher uploads the file to S3, it calls PATCH
-  // to transition detected → uploaded and attach the S3 coordinates.
-  it("PATCH transitions detected → uploaded with S3 info", async () => {
+  // Watcher path: detected → uploaded. The S3 location is derived server-side,
+  // so the hostile s3_bucket / s3_key sent here must be ignored and the
+  // canonical location rebuilt from DB state.
+  it("PATCH transitions detected → uploaded and derives the S3 location", async () => {
     const res = await api(`/api/v1/files/${fileId}`, {
       method: "PATCH",
       token,
       body: {
         status: "uploaded",
-        s3_bucket: "test-bucket",
-        s3_key: `${instrumentId}/${runId}/sample.csv`,
+        s3_bucket: "attacker-controlled-bucket",
+        s3_key: "someones-private-data/secret-export.csv",
         content_type: "text/csv",
         size_bytes: 512,
       },
     });
     expect(res.status).toBe(200);
     const data = await res.json();
     expect(data.status).toBe("uploaded");
-    expect(data.s3_bucket).toBe("test-bucket");
+    expect(data.s3_bucket).toBe("test-raw-data-bucket");
+    expect(data.s3_key).toBe(`${instrumentId}/${runId}/sample.csv`);
     expect(data.uploaded_at).toBeTruthy();
     fileDetail.parse(data);
   });
@@ -383,7 +385,7 @@ describe("Files API", () => {
     expect(res.status).toBe(404);
   });
 
-  // Watcher cancel path (ENG-1397): after giving up on a queued file the
+  // Watcher cancel path: after giving up on a queued file the
   // watcher reverts it upload_requested → detected, which must clear
   // upload_requested_at so the row leaves the upload queue.
   it("PATCH transitions upload_requested → detected and clears upload_requested_at", async () => {

web/tests/integration/upload-request-cancellation.test.ts

@@ -14,7 +14,7 @@ import {
 // request points at a relative path anchored to the old root and can no
 // longer be resolved by the watcher. The config PUT handler reverts those
 // files to `detected` (clearing upload_requested_at) so they drop out of the
-// upload queue instead of erroring on every heartbeat poll (ENG-1397).
+// upload queue instead of erroring on every heartbeat poll.
 describe("Upload request cancellation on watch-directory change", () => {
   let token: string;
   let watcherId: string;