diff --git a/developer-docs/watcher.md b/developer-docs/watcher.md index aa9ada20..c390cc92 100644 --- a/developer-docs/watcher.md +++ b/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. diff --git a/uv.lock b/uv.lock index 35231ff8..0d44f617 100644 --- a/uv.lock +++ b/uv.lock @@ -416,7 +416,7 @@ requires-dist = [ [[package]] name = "data-hub-watcher" -version = "0.5.0" +version = "0.5.1" source = { editable = "watcher" } dependencies = [ { name = "click" }, diff --git a/watcher/pyproject.toml b/watcher/pyproject.toml index cc329dc0..6814b822 100644 --- a/watcher/pyproject.toml +++ b/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" diff --git a/watcher/src/data_hub_watcher/api_client.py b/watcher/src/data_hub_watcher/api_client.py index 98a9f4b6..97973a74 100644 --- a/watcher/src/data_hub_watcher/api_client.py +++ b/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"}) diff --git a/watcher/src/data_hub_watcher/constants.py b/watcher/src/data_hub_watcher/constants.py index d504c442..6fd0f951 100644 --- a/watcher/src/data_hub_watcher/constants.py +++ b/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. diff --git a/watcher/src/data_hub_watcher/uploader.py b/watcher/src/data_hub_watcher/uploader.py index 9ed5525b..75c66df4 100644 --- a/watcher/src/data_hub_watcher/uploader.py +++ b/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", }, diff --git a/watcher/tests/integration/test_auto_upload_flow.py b/watcher/tests/integration/test_auto_upload_flow.py index 4da713e9..50134136 100644 --- a/watcher/tests/integration/test_auto_upload_flow.py +++ b/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", }, diff --git a/watcher/tests/integration/test_error_paths.py b/watcher/tests/integration/test_error_paths.py index 1a2e4d0c..0053f11d 100644 --- a/watcher/tests/integration/test_error_paths.py +++ b/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( diff --git a/watcher/tests/integration/test_manual_upload_flow.py b/watcher/tests/integration/test_manual_upload_flow.py index 625c9726..f8bc6481 100644 --- a/watcher/tests/integration/test_manual_upload_flow.py +++ b/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", }, diff --git a/watcher/tests/test_uploader.py b/watcher/tests/test_uploader.py index 43f1d08a..505c0650 100644 --- a/watcher/tests/test_uploader.py +++ b/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 diff --git a/web/app/api/v1/files/[fileId]/route.ts b/web/app/api/v1/files/[fileId]/route.ts index 76063f60..7d034c30 100644 --- a/web/app/api/v1/files/[fileId]/route.ts +++ b/web/app/api/v1/files/[fileId]/route.ts @@ -5,6 +5,7 @@ import { apiError, apiErrorFromResult, CONFLICT, + INTERNAL_ERROR, NOT_FOUND, VALIDATION_ERROR, } from "@/lib/api/errors"; @@ -12,6 +13,7 @@ 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 = { 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; } diff --git a/web/app/api/v1/watchers/[watcherId]/config/route.ts b/web/app/api/v1/watchers/[watcherId]/config/route.ts index eb94a931..562e00b1 100644 --- a/web/app/api/v1/watchers/[watcherId]/config/route.ts +++ b/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) { diff --git a/web/lib/api/openapi/schemas/files.ts b/web/lib/api/openapi/schemas/files.ts index 22cce890..7d496a61 100644 --- a/web/lib/api/openapi/schemas/files.ts +++ b/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(), diff --git a/web/lib/api/watchers.ts b/web/lib/api/watchers.ts index d7258bb4..31f1f889 100644 --- a/web/lib/api/watchers.ts +++ b/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, diff --git a/web/tests/integration/files.test.ts b/web/tests/integration/files.test.ts index ddaaea2f..e5f8df54 100644 --- a/web/tests/integration/files.test.ts +++ b/web/tests/integration/files.test.ts @@ -289,16 +289,17 @@ 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, }, @@ -306,7 +307,8 @@ describe("Files API", () => { 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 () => { diff --git a/web/tests/integration/upload-request-cancellation.test.ts b/web/tests/integration/upload-request-cancellation.test.ts index ac846081..a13d87cf 100644 --- a/web/tests/integration/upload-request-cancellation.test.ts +++ b/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;