From ac718b0d260038f5891bc7e17fbf00751e6855ee Mon Sep 17 00:00:00 2001 From: Lewis John McGibbney Date: Sun, 4 Oct 2026 05:40:39 -0700 Subject: [PATCH 1/5] feat(ingest): accept uploads and queue them as jobs POST/GET /ingest return 202 with a job id instead of holding the request open while every archive member is processed, which caused nginx 504s on large zips. Poll GET /ingest/jobs/{job_id} for status. One worker, started by docker-entrypoint.sh beside Gunicorn, runs queued jobs from the new ingest_job table. Clients that read the old 200 Ingest200 body must poll the job instead. --- openapi.yaml | 158 +++++++++--- services/fswatch/post.sh | 41 ++- services/postgis/tagbase_schema_tables.sql | 15 ++ tagbase_server/Dockerfile | 4 +- tagbase_server/docker-entrypoint.sh | 31 +++ .../controllers/ingest_controller.py | 173 +++++-------- tagbase_server/tagbase_server/ingest_jobs.py | 190 ++++++++++++++ .../tagbase_server/ingest_worker.py | 82 ++++++ .../tagbase_server/models/__init__.py | 4 +- .../tagbase_server/models/ingest200.py | 118 --------- .../tagbase_server/models/ingest_accepted.py | 113 +++++++++ .../tagbase_server/models/ingest_job.py | 237 ++++++++++++++++++ .../tagbase_server/openapi/openapi.yaml | 158 +++++++++--- .../tagbase_server/test/conftest.py | 8 + tagbase_server/tagbase_server/test/helpers.py | 19 ++ .../test/integration/test_ingest_api.py | 22 +- .../test/integration/test_tags_api.py | 8 +- .../test/stack/test_fswatch_contract.py | 17 +- .../tagbase_server/test/test_ingest_jobs.py | 71 ++++++ .../tagbase_server/test/test_sonar_cleanup.py | 51 ++-- .../tagbase_server/utils/processing_utils.py | 43 ++-- 21 files changed, 1207 insertions(+), 356 deletions(-) create mode 100755 tagbase_server/docker-entrypoint.sh create mode 100644 tagbase_server/tagbase_server/ingest_jobs.py create mode 100644 tagbase_server/tagbase_server/ingest_worker.py delete mode 100644 tagbase_server/tagbase_server/models/ingest200.py create mode 100644 tagbase_server/tagbase_server/models/ingest_accepted.py create mode 100644 tagbase_server/tagbase_server/models/ingest_job.py create mode 100644 tagbase_server/tagbase_server/test/test_ingest_jobs.py diff --git a/openapi.yaml b/openapi.yaml index f3d31598..9a838575 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -27,7 +27,7 @@ tags: paths: /ingest: get: - description: Get network accessible file and execute ingestion + description: Get a network accessible file, store it, and queue ingestion operationId: ingest_get parameters: - description: Location of a network accessible (file, ftp, http, https) @@ -46,22 +46,22 @@ paths: - $ref: "#/components/parameters/type" - $ref: "#/components/parameters/version" responses: - "200": + "202": content: application/json: schema: - $ref: "#/components/schemas/Ingest200" - description: A success message confirming ingestion. + $ref: "#/components/schemas/IngestAccepted" + description: The file was stored and an ingest job was queued. "400": $ref: "#/components/responses/BadRequest" "500": $ref: "#/components/responses/InternalError" - summary: Get network accessible file and execute ingestion + summary: Get network accessible file and queue ingestion tags: - ingest x-openapi-router-controller: tagbase_server.controllers.ingest_controller post: - description: Post a local file and perform a ingest operation + description: Store a local file and queue ingestion operationId: ingest_post parameters: - $ref: "#/components/parameters/filename" @@ -87,21 +87,53 @@ paths: required: true x-body-name: body responses: - "200": + "202": content: application/json: schema: - $ref: "#/components/schemas/Ingest200" - description: A success message confirming ingestion. + $ref: "#/components/schemas/IngestAccepted" + description: The file was stored and an ingest job was queued. "400": $ref: "#/components/responses/BadRequest" "500": $ref: "#/components/responses/InternalError" - summary: Post a local file and perform a ingest operation + summary: Post a local file and queue ingestion tags: - ingest x-codegen-request-body-name: body x-openapi-router-controller: tagbase_server.controllers.ingest_controller + /ingest/jobs/{job_id}: + get: + description: Status of one ingest job + operationId: get_ingest_job + parameters: + - description: Job id returned when the file was accepted + explode: true + in: path + name: job_id + required: true + schema: + type: string + format: uuid + minLength: 36 + maxLength: 36 + pattern: ^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$ + style: simple + responses: + "200": + content: + application/json: + schema: + $ref: "#/components/schemas/IngestJob" + description: The job is queued, running, or finished. + "404": + $ref: "#/components/responses/NotFound" + "500": + $ref: "#/components/responses/InternalError" + summary: Get ingest job status + tags: + - ingest + x-openapi-router-controller: tagbase_server.controllers.ingest_controller /tags: delete: description: Delete all tags @@ -205,6 +237,12 @@ components: application/problem+json: schema: $ref: "#/components/schemas/Problem" + NotFound: + description: The ingest job id is not in the table. + content: + application/problem+json: + schema: + $ref: "#/components/schemas/Problem" InternalError: description: Unexpected server error. Contact admin detailed in openapi.yaml. content: @@ -353,38 +391,88 @@ components: maxLength: 36 pattern: ^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$ example: 123e4567-e89b-12d3-a456-426614174000 - Ingest200: - description: HTTP 200 success response + IngestAccepted: + description: HTTP 202 response after an ingest file is stored example: - code: "200" - elapsed: "0:00:06.506691" - message: - Data file eTUFF-sailfish-117259.txt successfully ingested into Tagbase - DB. + id: 123e4567-e89b-12d3-a456-426614174000 + status: queued properties: - code: - description: HTTP status code - example: "200" + id: + description: Job id + example: 123e4567-e89b-12d3-a456-426614174000 + format: uuid + maxLength: 36 + minLength: 36 + pattern: ^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$ type: string - minLength: 3 - maxLength: 3 - pattern: ^[1-5][0-9]{2}$ - elapsed: - description: Elapsed time for the operation + status: + description: Job status at accept time + enum: + - queued + example: queued + maxLength: 6 + minLength: 6 + pattern: ^queued$ type: string - minLength: 1 - maxLength: 64 - pattern: ^(?!\s*$).+$ - message: - description: A string detailing specifics of an HTTP operation - example: - Data file eTUFF-sailfish-117259.txt successfully ingested into - Tagbase DB. + required: + - id + - status + title: IngestAccepted + type: object + IngestJob: + description: Status of one ingest job + example: + id: 123e4567-e89b-12d3-a456-426614174000 + status: succeeded + files_total: 2 + files_done: 2 + properties: + id: + description: Job id + example: 123e4567-e89b-12d3-a456-426614174000 + format: uuid + maxLength: 36 + minLength: 36 + pattern: ^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$ type: string - minLength: 1 + status: + description: queued, running, succeeded, or failed + enum: + - queued + - running + - succeeded + - failed + example: succeeded + maxLength: 9 + minLength: 6 + pattern: ^(queued|running|succeeded|failed)$ + type: string + error: + description: Present when the job failed + example: restarted before ingest finished maxLength: 10000 + minLength: 1 pattern: ^(?!\s*$).+$ - title: Ingest200 + type: string + files_total: + description: Members in the archive, once unpacking has started + example: 2 + format: int32 + maximum: 2147483647 + minimum: 0 + type: integer + files_done: + description: Members ingested so far + example: 2 + format: int32 + maximum: 2147483647 + minimum: 0 + type: integer + required: + - id + - status + - files_done + title: IngestJob type: object Metadata: description: Contains the ingested tag metadata consistent with the eTUFF specification diff --git a/services/fswatch/post.sh b/services/fswatch/post.sh index ade0c3f9..52dd5e27 100755 --- a/services/fswatch/post.sh +++ b/services/fswatch/post.sh @@ -106,17 +106,46 @@ ingest_file() { case "$filename" in *.txt) ctype=text/plain ;; esac - if ! curl -sS -f -X POST \ + if ! body=$(curl -sS -f -X POST \ -H "accept: application/json" \ -H "Content-Type: ${ctype}" \ -T "$line" \ - "${TAGBASE_INGEST_BASE}/ingest?filename=${enc}&type=etuff"; then + "${TAGBASE_INGEST_BASE}/ingest?filename=${enc}&type=etuff"); then echo "Ingest failed for $filename (see tagbase_server logs)" - else - printf '%s\n%s\n' "$line" "$size" >"$DEDUPE_STATE" - echo - echo "Ingest OK: $filename" + return 0 + fi + job_id=$(printf '%s' "$body" | sed -n 's/.*"id"[[:space:]]*:[[:space:]]*"\([^"]*\)".*/\1/p') + if [ -z "$job_id" ]; then + echo "Ingest accept for $filename did not return a job id" + return 0 + fi + if ! wait_for_job "$job_id"; then + echo "Ingest failed for $filename (see tagbase_server logs)" + return 0 fi + printf '%s\n%s\n' "$line" "$size" >"$DEDUPE_STATE" + echo + echo "Ingest OK: $filename" +} + +wait_for_job() { + job_id=$1 + while true; do + if ! status_body=$(curl -sS -f \ + -H "accept: application/json" \ + "${TAGBASE_INGEST_BASE}/ingest/jobs/${job_id}"); then + echo "Job status request failed for $job_id" + return 1 + fi + if printf '%s' "$status_body" | grep -Eq '"status"[[:space:]]*:[[:space:]]*"succeeded"'; then + return 0 + fi + if printf '%s' "$status_body" | grep -Eq '"status"[[:space:]]*:[[:space:]]*"failed"'; then + echo "Job $job_id failed: $status_body" + return 1 + fi + sleep 2 + done } scan_existing() { diff --git a/services/postgis/tagbase_schema_tables.sql b/services/postgis/tagbase_schema_tables.sql index aca845e1..88da28e6 100644 --- a/services/postgis/tagbase_schema_tables.sql +++ b/services/postgis/tagbase_schema_tables.sql @@ -1399,3 +1399,18 @@ AS $BODY$ WHERE position_date_time IS NULL; $BODY$; + +CREATE TABLE ingest_job ( + id uuid PRIMARY KEY, + status text NOT NULL + CHECK (status IN ('queued', 'running', 'succeeded', 'failed')), + source_path text NOT NULL, + filename text, + version text, + notes text, + error text, + files_total integer, + files_done integer NOT NULL DEFAULT 0, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now() +); diff --git a/tagbase_server/Dockerfile b/tagbase_server/Dockerfile index 152e1eed..8067d98a 100644 --- a/tagbase_server/Dockerfile +++ b/tagbase_server/Dockerfile @@ -18,6 +18,6 @@ COPY . /usr/src/app EXPOSE 5433 -ENTRYPOINT ["gunicorn"] +RUN chmod 0555 /usr/src/app/docker-entrypoint.sh -CMD ["-c", "gunicorn.conf.py", "tagbase_server.__main__:app"] +ENTRYPOINT ["/usr/src/app/docker-entrypoint.sh"] diff --git a/tagbase_server/docker-entrypoint.sh b/tagbase_server/docker-entrypoint.sh new file mode 100755 index 00000000..ad5cf985 --- /dev/null +++ b/tagbase_server/docker-entrypoint.sh @@ -0,0 +1,31 @@ +#!/bin/sh +# Gunicorn accepts uploads. One ingest process beside it runs the queue. +set -eu + +python -m tagbase_server.ingest_worker & +worker_pid=$! + +gunicorn -c gunicorn.conf.py tagbase_server.__main__:app & +gunicorn_pid=$! + +stop() { + kill "$worker_pid" "$gunicorn_pid" 2>/dev/null || true +} +trap stop INT TERM + +status=0 +while true; do + if ! kill -0 "$gunicorn_pid" 2>/dev/null; then + wait "$gunicorn_pid" || status=$? + break + fi + if ! kill -0 "$worker_pid" 2>/dev/null; then + wait "$worker_pid" || status=$? + break + fi + sleep 1 +done +stop +wait "$worker_pid" 2>/dev/null || true +wait "$gunicorn_pid" 2>/dev/null || true +exit "$status" diff --git a/tagbase_server/tagbase_server/controllers/ingest_controller.py b/tagbase_server/tagbase_server/controllers/ingest_controller.py index 0667bf26..8fa62b6e 100644 --- a/tagbase_server/tagbase_server/controllers/ingest_controller.py +++ b/tagbase_server/tagbase_server/controllers/ingest_controller.py @@ -1,20 +1,16 @@ import logging +import os import time -from functools import partial -from multiprocessing import cpu_count +import uuid -import parmap -from tqdm import tqdm as std_tqdm - -from tagbase_server.models.ingest200 import Ingest200 # noqa: E501 -from tagbase_server.problem import TagbaseClientError, as_json -from tagbase_server.telemetry import get_tracer, record_ingest_request +from tagbase_server.ingest_jobs import IngestJobError, enqueue_job, fetch_job +from tagbase_server.problem import TYPE_HTTP, TagbaseClientError, as_json +from tagbase_server.telemetry import record_ingest_request from tagbase_server.utils.io_utils import ( + TEMP_DIR, process_get_input_data, process_post_input_data, - unpack_compressed_binary, ) -from tagbase_server.utils.processing_utils import process_etuff_file logger = logging.getLogger(__name__) @@ -30,91 +26,54 @@ def _resolve_ingest_file_type(type): return ingest_file_type -def _ingest_progress_bar(data_file): - """Create a standard stdout progress bar for ingestio]n.""" - return partial(std_tqdm, desc=f"Ingesting: {data_file}") - - -def _run_ingest_queue(etuff_files, version, notes, data_file): - parmap.map( - process_etuff_file, - etuff_files, - version=version, - notes=notes, - pm_parallel=False, - pm_processes=cpu_count(), - pm_pbar=_ingest_progress_bar(data_file), +def _isolate_upload(source_path): + """Keep a POST body off the caller-supplied name so a later post cannot overwrite it.""" + unique = os.path.join( + TEMP_DIR, f"{uuid.uuid4().hex}-{os.path.basename(source_path)}" ) + os.replace(source_path, unique) + return unique -def _ingest_with_telemetry(operation_name, work): - tracer = get_tracer() - start = time.perf_counter() - outcome = "success" +def _accept(source_path, filename, version, notes, started): try: - with tracer.start_as_current_span("ingest.handle") as span: - span.set_attribute("ingest.operation", operation_name) - return work() - except TagbaseClientError: - outcome = "client_error" - raise - except ValueError as exc: - outcome = "client_error" - raise TagbaseClientError(str(exc)) from exc - except Exception: - outcome = "server_error" + job_id = enqueue_job( + source_path=source_path, + filename=filename, + version=version, + notes=notes, + ) + except IngestJobError: + logger.exception("Failed to queue ingest for %s", filename) raise - finally: - record_ingest_request(outcome, time.perf_counter() - start) + record_ingest_request("accepted", round(time.perf_counter() - started, 2)) + return as_json({"id": job_id, "status": "queued"}, 202) def ingest_get(file, notes=None, type=None, type_=None, version=None): # noqa: E501 """Get network accessible file and execute ingestion - Get network accessible file and execute ingestion # noqa: E501 - - :param file: Location of a network accessible (file, ftp, http, https) file e.g. 'file:///usr/src/app/data/eTUFF-sailfish-117259.txt'. + :param file: Location of a network accessible (file, ftp, http, https) file :type file: str - :param notes: Free-form text field where details of submitted eTUFF file for ingest can be provided e.g. submitter name, etuff data contents (tag metadata and measurements + primary position data, or just secondary solution-positional meta/data) + :param notes: Free-form text field :type notes: str - :param type: Type of file to be ingested, defaults to 'etuff' + :param type: Type of file to be ingested, defaults to 'etuff' :type type: str :param type_: Connexion pythonic alias for OpenAPI parameter ``type`` :type type_: str :param version: Version identifier for the eTUFF tag data file ingested :type version: str - :rtype: Union[Ingest200, Tuple[Ingest200, int], Tuple[Ingest200, int, Dict[str, str]] + :rtype: tuple """ - - def _work(): - ingest_file_type = _resolve_ingest_file_type( - type_ if type_ is not None else type - ) - logger.info("Ingest file type: %s", ingest_file_type) - start = time.perf_counter() + started = time.perf_counter() + ingest_file_type = _resolve_ingest_file_type(type_ if type_ is not None else type) + logger.info("Ingest file type: %s", ingest_file_type) + try: data_file = process_get_input_data(file) - etuff_files = [] - if not data_file.endswith(".txt"): - etuff_files = unpack_compressed_binary(data_file) - else: - etuff_files.append(data_file) - logger.info("eTUFF ingestion queue: %s", etuff_files) - _run_ingest_queue(etuff_files, version, notes, data_file) - finish = time.perf_counter() - elapsed = round(finish - start, 2) - return as_json( - Ingest200.from_dict( - { - "code": "200", - "elapsed": elapsed, - "message": f"Processing %s file(s) - {etuff_files}" - % len(etuff_files), - } - ) - ) - - return _ingest_with_telemetry("get", _work) + except ValueError as exc: + raise TagbaseClientError(str(exc)) from exc + return _accept(data_file, os.path.basename(data_file), version, notes, started) def ingest_post( @@ -122,49 +81,47 @@ def ingest_post( ): # noqa: E501 """Post a local file and perform a ingest operation - Post a local file and perform a ingest operation # noqa: E501 - - :param filename: Free-form text field to explicitly define the name of the file to be persisted + :param filename: Name of the file to be persisted :type filename: str :param body: :type body: str - :param notes: Free-form text field where details of submitted eTUFF file for ingest can be provided e.g. submitter name, etuff data contents (tag metadata and measurements + primary position data, or just secondary solution-positional meta/data) + :param notes: Free-form text field :type notes: str - :param type: Type of file to be ingested, defaults to 'etuff' + :param type: Type of file to be ingested, defaults to 'etuff' :type type: str :param type_: Connexion pythonic alias for OpenAPI parameter ``type`` :type type_: str :param version: Version identifier for the eTUFF tag data file ingested :type version: str - :rtype: Union[Ingest200, Tuple[Ingest200, int], Tuple[Ingest200, int, Dict[str, str]] + :rtype: tuple """ + started = time.perf_counter() + ingest_file_type = _resolve_ingest_file_type(type_ if type_ is not None else type) + logger.info("Ingest file type: %s", ingest_file_type) + data_file = process_post_input_data(filename, body) + stored = _isolate_upload(data_file) + return _accept(stored, filename, version, notes, started) - def _work(): - ingest_file_type = _resolve_ingest_file_type( - type_ if type_ is not None else type - ) - logger.info("Ingest file type: %s", ingest_file_type) - start = time.perf_counter() - data_file = process_post_input_data(filename, body) - etuff_files = [] - if not data_file.endswith(".txt"): - etuff_files = unpack_compressed_binary(data_file) - else: - etuff_files.append(data_file) - logger.info("eTUFF ingestion queue: %s", etuff_files) - _run_ingest_queue(etuff_files, version, notes, data_file) - finish = time.perf_counter() - elapsed = round(finish - start, 2) - return as_json( - Ingest200.from_dict( - { - "code": "200", - "elapsed": elapsed, - "message": f"Processing %s file(s) - {etuff_files}." - % len(etuff_files), - } - ) - ) - return _ingest_with_telemetry("post", _work) +def get_ingest_job(job_id): # noqa: E501 + """Return one ingest job. + + :param job_id: Job id returned by POST or GET /ingest + :type job_id: str + + :rtype: tuple + """ + try: + job = fetch_job(job_id) + except IngestJobError: + logger.exception("Failed to read ingest job %s", job_id) + raise + if job is None: + raise TagbaseClientError( + f"Ingest job '{job_id}' was not found.", + title="Not Found", + type_=TYPE_HTTP, + status=404, + ) + return as_json(job) diff --git a/tagbase_server/tagbase_server/ingest_jobs.py b/tagbase_server/tagbase_server/ingest_jobs.py new file mode 100644 index 00000000..c38de275 --- /dev/null +++ b/tagbase_server/tagbase_server/ingest_jobs.py @@ -0,0 +1,190 @@ +"""Queued ingest jobs. + +A request stores the upload and inserts one row. The ingest process claims +the oldest queued row and updates it until the job is terminal. +""" + +from __future__ import annotations + +import os +import uuid +from typing import Any + +from psycopg2.extras import RealDictCursor + +from tagbase_server.utils.db_utils import connect + +RESTART_ERROR = "restarted before ingest finished" +MISSING_FILE_ERROR = "upload file missing after restart" + +_CREATE_TABLE = """ +CREATE TABLE IF NOT EXISTS ingest_job ( + id uuid PRIMARY KEY, + status text NOT NULL + CHECK (status IN ('queued', 'running', 'succeeded', 'failed')), + source_path text NOT NULL, + filename text, + version text, + notes text, + error text, + files_total integer, + files_done integer NOT NULL DEFAULT 0, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now() +) +""" + + +class IngestJobError(Exception): + """The job table could not be read or written.""" + + +def stale_failures(jobs: list[dict[str, Any]], path_exists) -> list[tuple[str, str]]: + """Jobs a restart must fail. + + A running job was interrupted. A queued job whose upload is gone cannot + run. A queued job whose file is still present stays queued. + """ + failed: list[tuple[str, str]] = [] + for job in jobs: + status = job["status"] + job_id = str(job["id"]) + if status == "running": + failed.append((job_id, RESTART_ERROR)) + elif status == "queued" and not path_exists(job["source_path"]): + failed.append((job_id, MISSING_FILE_ERROR)) + return failed + + +def public_job(row: dict[str, Any]) -> dict[str, Any]: + job: dict[str, Any] = { + "id": str(row["id"]), + "status": row["status"], + "files_done": row["files_done"], + } + if row["error"] is not None: + job["error"] = row["error"] + if row["files_total"] is not None: + job["files_total"] = row["files_total"] + return job + + +def enqueue_job( + *, + source_path: str, + filename: str | None, + version: str | None, + notes: str | None, +) -> str: + job_id = str(uuid.uuid4()) + with _connection() as conn: + _ensure_table(conn) + with conn.cursor() as cur: + cur.execute( + """ + INSERT INTO ingest_job (id, status, source_path, filename, version, notes) + VALUES (%s, 'queued', %s, %s, %s, %s) + """, + (job_id, source_path, filename, version, notes), + ) + conn.commit() + return job_id + + +def fetch_job(job_id: str) -> dict[str, Any] | None: + with _connection() as conn: + _ensure_table(conn) + with conn.cursor(cursor_factory=RealDictCursor) as cur: + cur.execute("SELECT * FROM ingest_job WHERE id = %s", (job_id,)) + row = cur.fetchone() + conn.commit() + if row is None: + return None + return public_job(dict(row)) + + +def recover_stale_jobs() -> None: + with _connection() as conn: + _ensure_table(conn) + with conn.cursor(cursor_factory=RealDictCursor) as cur: + cur.execute(""" + SELECT id, status, source_path + FROM ingest_job + WHERE status IN ('queued', 'running') + """) + rows = [dict(row) for row in cur.fetchall()] + for job_id, error in stale_failures(rows, os.path.isfile): + cur.execute( + """ + UPDATE ingest_job + SET status = 'failed', error = %s, updated_at = now() + WHERE id = %s AND status IN ('queued', 'running') + """, + (error, job_id), + ) + conn.commit() + + +def claim_next_job() -> dict[str, Any] | None: + with _connection() as conn: + _ensure_table(conn) + with conn.cursor(cursor_factory=RealDictCursor) as cur: + cur.execute(""" + UPDATE ingest_job + SET status = 'running', updated_at = now() + WHERE id = ( + SELECT id + FROM ingest_job + WHERE status = 'queued' + ORDER BY created_at + FOR UPDATE SKIP LOCKED + LIMIT 1 + ) + RETURNING * + """) + row = cur.fetchone() + conn.commit() + if row is None: + return None + return dict(row) + + +def set_progress(job_id: str, *, files_total: int, files_done: int) -> None: + with _connection() as conn: + with conn.cursor() as cur: + cur.execute( + """ + UPDATE ingest_job + SET files_total = %s, files_done = %s, updated_at = now() + WHERE id = %s + """, + (files_total, files_done, job_id), + ) + conn.commit() + + +def finish_job(job_id: str, *, status: str, error: str | None) -> None: + with _connection() as conn: + with conn.cursor() as cur: + cur.execute( + """ + UPDATE ingest_job + SET status = %s, error = %s, updated_at = now() + WHERE id = %s + """, + (status, error, job_id), + ) + conn.commit() + + +def _connection(): + conn = connect() + if not hasattr(conn, "cursor"): + raise IngestJobError("database unavailable") + conn.autocommit = False + return conn + + +def _ensure_table(conn) -> None: + with conn.cursor() as cur: + cur.execute(_CREATE_TABLE) diff --git a/tagbase_server/tagbase_server/ingest_worker.py b/tagbase_server/tagbase_server/ingest_worker.py new file mode 100644 index 00000000..7a33551b --- /dev/null +++ b/tagbase_server/tagbase_server/ingest_worker.py @@ -0,0 +1,82 @@ +"""One ingest process. It runs a single zip at a time.""" + +from __future__ import annotations + +import logging +import time + +from tagbase_server.ingest_jobs import ( + claim_next_job, + finish_job, + recover_stale_jobs, + set_progress, +) +from tagbase_server.utils.io_utils import unpack_compressed_binary +from tagbase_server.utils.processing_utils import process_etuff_file + +logger = logging.getLogger(__name__) + +_POLL_SECONDS = 1 + + +def members_for(source_path: str) -> list[str]: + if source_path.endswith(".txt"): + return [source_path] + unpacked = unpack_compressed_binary(source_path) + return list(unpacked or []) + + +def run_job(job: dict) -> None: + """Ingest every member. The first error fails the job and stops the queue.""" + job_id = str(job["id"]) + members = members_for(job["source_path"]) + if not members: + raise RuntimeError("archive contained no eTUFF files") + set_progress(job_id, files_total=len(members), files_done=0) + for done, member in enumerate(members, start=1): + process_etuff_file( + member, + version=job.get("version"), + notes=job.get("notes"), + count_request=False, + ) + set_progress(job_id, files_total=len(members), files_done=done) + finish_job(job_id, status="succeeded", error=None) + + +def process_next() -> bool: + """Claim and run the oldest queued job. Return False when the queue is empty.""" + job = claim_next_job() + if job is None: + return False + job_id = str(job["id"]) + try: + run_job(job) + except Exception as exc: + logger.exception("Ingest job %s failed", job_id) + finish_job(job_id, status="failed", error=str(exc)) + return True + + +def process_available() -> int: + """Run every job that is already queued. Used by tests in this process.""" + recover_stale_jobs() + ran = 0 + while process_next(): + ran += 1 + return ran + + +def main() -> None: + logging.basicConfig( + level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" + ) + logger.info("Ingest worker starting") + recover_stale_jobs() + while True: + if not process_next(): + time.sleep(_POLL_SECONDS) + + +if __name__ == "__main__": + main() diff --git a/tagbase_server/tagbase_server/models/__init__.py b/tagbase_server/tagbase_server/models/__init__.py index 65a21d02..5234c0e7 100644 --- a/tagbase_server/tagbase_server/models/__init__.py +++ b/tagbase_server/tagbase_server/models/__init__.py @@ -3,9 +3,9 @@ # flake8: noqa from __future__ import absolute_import -from tagbase_server.models.ingest200 import Ingest200 - # import models into model package +from tagbase_server.models.ingest_accepted import IngestAccepted +from tagbase_server.models.ingest_job import IngestJob from tagbase_server.models.problem import Problem from tagbase_server.models.tag200 import Tag200 from tagbase_server.models.tag200_collection import Tag200Collection diff --git a/tagbase_server/tagbase_server/models/ingest200.py b/tagbase_server/tagbase_server/models/ingest200.py deleted file mode 100644 index 369fb75f..00000000 --- a/tagbase_server/tagbase_server/models/ingest200.py +++ /dev/null @@ -1,118 +0,0 @@ -# coding: utf-8 - -from __future__ import absolute_import - -from datetime import date, datetime # noqa: F401 -from typing import Dict, List # noqa: F401 - -from tagbase_server import util -from tagbase_server.models.base_model_ import Model - - -class Ingest200(Model): - """NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). - - Do not edit the class manually. - """ - - def __init__(self, code=None, elapsed=None, message=None): # noqa: E501 - """Ingest200 - a model defined in OpenAPI - - :param code: The code of this Ingest200. # noqa: E501 - :type code: str - :param elapsed: The elapsed of this Ingest200. # noqa: E501 - :type elapsed: str - :param message: The message of this Ingest200. # noqa: E501 - :type message: str - """ - self.openapi_types = {"code": str, "elapsed": str, "message": str} - - self.attribute_map = { - "code": "code", - "elapsed": "elapsed", - "message": "message", - } - - self._code = code - self._elapsed = elapsed - self._message = message - - @classmethod - def from_dict(cls, dikt) -> "Ingest200": - """Returns the dict as a model - - :param dikt: A dict. - :type: dict - :return: The Ingest200 of this Ingest200. # noqa: E501 - :rtype: Ingest200 - """ - return util.deserialize_model(dikt, cls) - - @property - def code(self): - """Gets the code of this Ingest200. - - HTTP status code # noqa: E501 - - :return: The code of this Ingest200. - :rtype: str - """ - return self._code - - @code.setter - def code(self, code): - """Sets the code of this Ingest200. - - HTTP status code # noqa: E501 - - :param code: The code of this Ingest200. - :type code: str - """ - - self._code = code - - @property - def elapsed(self): - """Gets the elapsed of this Ingest200. - - Elapsed time for the operation # noqa: E501 - - :return: The elapsed of this Ingest200. - :rtype: str - """ - return self._elapsed - - @elapsed.setter - def elapsed(self, elapsed): - """Sets the elapsed of this Ingest200. - - Elapsed time for the operation # noqa: E501 - - :param elapsed: The elapsed of this Ingest200. - :type elapsed: str - """ - - self._elapsed = elapsed - - @property - def message(self): - """Gets the message of this Ingest200. - - A string detailing specifics of an HTTP operation # noqa: E501 - - :return: The message of this Ingest200. - :rtype: str - """ - return self._message - - @message.setter - def message(self, message): - """Sets the message of this Ingest200. - - A string detailing specifics of an HTTP operation # noqa: E501 - - :param message: The message of this Ingest200. - :type message: str - """ - - self._message = message diff --git a/tagbase_server/tagbase_server/models/ingest_accepted.py b/tagbase_server/tagbase_server/models/ingest_accepted.py new file mode 100644 index 00000000..f08c82b1 --- /dev/null +++ b/tagbase_server/tagbase_server/models/ingest_accepted.py @@ -0,0 +1,113 @@ +# coding: utf-8 + +from __future__ import absolute_import + +import re # noqa: E501 +from datetime import date, datetime # noqa: F401 +from typing import Dict, List # noqa: F401 + +from tagbase_server import util +from tagbase_server.models.base_model_ import Model + + +class IngestAccepted(Model): + """NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). + + Do not edit the class manually. + """ + + def __init__(self, id=None, status=None): # noqa: E501 + """IngestAccepted - a model defined in OpenAPI + + :param id: The id of this IngestAccepted. # noqa: E501 + :type id: str + :param status: The status of this IngestAccepted. # noqa: E501 + :type status: str + """ + self.openapi_types = {"id": str, "status": str} + + self.attribute_map = {"id": "id", "status": "status"} + + self._id = id + self._status = status + + @classmethod + def from_dict(cls, dikt) -> "IngestAccepted": + """Returns the dict as a model + + :param dikt: A dict. + :type: dict + :return: The IngestAccepted of this IngestAccepted. # noqa: E501 + :rtype: IngestAccepted + """ + return util.deserialize_model(dikt, cls) + + @property + def id(self) -> str: + """Gets the id of this IngestAccepted. + + Job id # noqa: E501 + + :return: The id of this IngestAccepted. + :rtype: str + """ + return self._id + + @id.setter + def id(self, id: str): + """Sets the id of this IngestAccepted. + + Job id # noqa: E501 + + :param id: The id of this IngestAccepted. + :type id: str + """ + if id is None: + raise ValueError("Invalid value for `id`, must not be `None`") # noqa: E501 + if id is not None and len(id) > 36: + raise ValueError( + "Invalid value for `id`, length must be less than or equal to `36`" + ) # noqa: E501 + if id is not None and len(id) < 36: + raise ValueError( + "Invalid value for `id`, length must be greater than or equal to `36`" + ) # noqa: E501 + if id is not None and not re.search( + r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$", + id, + ): # noqa: E501 + raise ValueError( + r"Invalid value for `id`, must be a follow pattern or equal to `/^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$/`" + ) # noqa: E501 + + self._id = id + + @property + def status(self) -> str: + """Gets the status of this IngestAccepted. + + Job status at accept time # noqa: E501 + + :return: The status of this IngestAccepted. + :rtype: str + """ + return self._status + + @status.setter + def status(self, status: str): + """Sets the status of this IngestAccepted. + + Job status at accept time # noqa: E501 + + :param status: The status of this IngestAccepted. + :type status: str + """ + allowed_values = ["queued"] # noqa: E501 + if status not in allowed_values: + raise ValueError( + "Invalid value for `status` ({0}), must be one of {1}".format( + status, allowed_values + ) + ) + + self._status = status diff --git a/tagbase_server/tagbase_server/models/ingest_job.py b/tagbase_server/tagbase_server/models/ingest_job.py new file mode 100644 index 00000000..64fd5563 --- /dev/null +++ b/tagbase_server/tagbase_server/models/ingest_job.py @@ -0,0 +1,237 @@ +# coding: utf-8 + +from __future__ import absolute_import + +import re # noqa: E501 +from datetime import date, datetime # noqa: F401 +from typing import Dict, List # noqa: F401 + +from tagbase_server import util +from tagbase_server.models.base_model_ import Model + + +class IngestJob(Model): + """NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). + + Do not edit the class manually. + """ + + def __init__( + self, id=None, status=None, error=None, files_total=None, files_done=None + ): # noqa: E501 + """IngestJob - a model defined in OpenAPI + + :param id: The id of this IngestJob. # noqa: E501 + :type id: str + :param status: The status of this IngestJob. # noqa: E501 + :type status: str + :param error: The error of this IngestJob. # noqa: E501 + :type error: str + :param files_total: The files_total of this IngestJob. # noqa: E501 + :type files_total: int + :param files_done: The files_done of this IngestJob. # noqa: E501 + :type files_done: int + """ + self.openapi_types = { + "id": str, + "status": str, + "error": str, + "files_total": int, + "files_done": int, + } + + self.attribute_map = { + "id": "id", + "status": "status", + "error": "error", + "files_total": "files_total", + "files_done": "files_done", + } + + self._id = id + self._status = status + self._error = error + self._files_total = files_total + self._files_done = files_done + + @classmethod + def from_dict(cls, dikt) -> "IngestJob": + """Returns the dict as a model + + :param dikt: A dict. + :type: dict + :return: The IngestJob of this IngestJob. # noqa: E501 + :rtype: IngestJob + """ + return util.deserialize_model(dikt, cls) + + @property + def id(self) -> str: + """Gets the id of this IngestJob. + + Job id # noqa: E501 + + :return: The id of this IngestJob. + :rtype: str + """ + return self._id + + @id.setter + def id(self, id: str): + """Sets the id of this IngestJob. + + Job id # noqa: E501 + + :param id: The id of this IngestJob. + :type id: str + """ + if id is None: + raise ValueError("Invalid value for `id`, must not be `None`") # noqa: E501 + if id is not None and len(id) > 36: + raise ValueError( + "Invalid value for `id`, length must be less than or equal to `36`" + ) # noqa: E501 + if id is not None and len(id) < 36: + raise ValueError( + "Invalid value for `id`, length must be greater than or equal to `36`" + ) # noqa: E501 + if id is not None and not re.search( + r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$", + id, + ): # noqa: E501 + raise ValueError( + r"Invalid value for `id`, must be a follow pattern or equal to `/^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$/`" + ) # noqa: E501 + + self._id = id + + @property + def status(self) -> str: + """Gets the status of this IngestJob. + + queued, running, succeeded, or failed # noqa: E501 + + :return: The status of this IngestJob. + :rtype: str + """ + return self._status + + @status.setter + def status(self, status: str): + """Sets the status of this IngestJob. + + queued, running, succeeded, or failed # noqa: E501 + + :param status: The status of this IngestJob. + :type status: str + """ + allowed_values = ["queued", "running", "succeeded", "failed"] # noqa: E501 + if status not in allowed_values: + raise ValueError( + "Invalid value for `status` ({0}), must be one of {1}".format( + status, allowed_values + ) + ) + + self._status = status + + @property + def error(self) -> str: + """Gets the error of this IngestJob. + + Present when the job failed # noqa: E501 + + :return: The error of this IngestJob. + :rtype: str + """ + return self._error + + @error.setter + def error(self, error: str): + """Sets the error of this IngestJob. + + Present when the job failed # noqa: E501 + + :param error: The error of this IngestJob. + :type error: str + """ + if error is not None and len(error) > 10000: + raise ValueError( + "Invalid value for `error`, length must be less than or equal to `10000`" + ) # noqa: E501 + if error is not None and len(error) < 1: + raise ValueError( + "Invalid value for `error`, length must be greater than or equal to `1`" + ) # noqa: E501 + if error is not None and not re.search(r"^(?!\s*$).+$", error): # noqa: E501 + raise ValueError( + r"Invalid value for `error`, must be a follow pattern or equal to `/^(?!\s*$).+$/`" + ) # noqa: E501 + + self._error = error + + @property + def files_total(self) -> int: + """Gets the files_total of this IngestJob. + + Members in the archive, once unpacking has started # noqa: E501 + + :return: The files_total of this IngestJob. + :rtype: int + """ + return self._files_total + + @files_total.setter + def files_total(self, files_total: int): + """Sets the files_total of this IngestJob. + + Members in the archive, once unpacking has started # noqa: E501 + + :param files_total: The files_total of this IngestJob. + :type files_total: int + """ + if files_total is not None and files_total > 2147483647: # noqa: E501 + raise ValueError( + "Invalid value for `files_total`, must be a value less than or equal to `2147483647`" + ) # noqa: E501 + if files_total is not None and files_total < 0: # noqa: E501 + raise ValueError( + "Invalid value for `files_total`, must be a value greater than or equal to `0`" + ) # noqa: E501 + + self._files_total = files_total + + @property + def files_done(self) -> int: + """Gets the files_done of this IngestJob. + + Members ingested so far # noqa: E501 + + :return: The files_done of this IngestJob. + :rtype: int + """ + return self._files_done + + @files_done.setter + def files_done(self, files_done: int): + """Sets the files_done of this IngestJob. + + Members ingested so far # noqa: E501 + + :param files_done: The files_done of this IngestJob. + :type files_done: int + """ + if files_done is None: + raise ValueError( + "Invalid value for `files_done`, must not be `None`" + ) # noqa: E501 + if files_done is not None and files_done > 2147483647: # noqa: E501 + raise ValueError( + "Invalid value for `files_done`, must be a value less than or equal to `2147483647`" + ) # noqa: E501 + if files_done is not None and files_done < 0: # noqa: E501 + raise ValueError( + "Invalid value for `files_done`, must be a value greater than or equal to `0`" + ) # noqa: E501 + + self._files_done = files_done diff --git a/tagbase_server/tagbase_server/openapi/openapi.yaml b/tagbase_server/tagbase_server/openapi/openapi.yaml index f3d31598..9a838575 100644 --- a/tagbase_server/tagbase_server/openapi/openapi.yaml +++ b/tagbase_server/tagbase_server/openapi/openapi.yaml @@ -27,7 +27,7 @@ tags: paths: /ingest: get: - description: Get network accessible file and execute ingestion + description: Get a network accessible file, store it, and queue ingestion operationId: ingest_get parameters: - description: Location of a network accessible (file, ftp, http, https) @@ -46,22 +46,22 @@ paths: - $ref: "#/components/parameters/type" - $ref: "#/components/parameters/version" responses: - "200": + "202": content: application/json: schema: - $ref: "#/components/schemas/Ingest200" - description: A success message confirming ingestion. + $ref: "#/components/schemas/IngestAccepted" + description: The file was stored and an ingest job was queued. "400": $ref: "#/components/responses/BadRequest" "500": $ref: "#/components/responses/InternalError" - summary: Get network accessible file and execute ingestion + summary: Get network accessible file and queue ingestion tags: - ingest x-openapi-router-controller: tagbase_server.controllers.ingest_controller post: - description: Post a local file and perform a ingest operation + description: Store a local file and queue ingestion operationId: ingest_post parameters: - $ref: "#/components/parameters/filename" @@ -87,21 +87,53 @@ paths: required: true x-body-name: body responses: - "200": + "202": content: application/json: schema: - $ref: "#/components/schemas/Ingest200" - description: A success message confirming ingestion. + $ref: "#/components/schemas/IngestAccepted" + description: The file was stored and an ingest job was queued. "400": $ref: "#/components/responses/BadRequest" "500": $ref: "#/components/responses/InternalError" - summary: Post a local file and perform a ingest operation + summary: Post a local file and queue ingestion tags: - ingest x-codegen-request-body-name: body x-openapi-router-controller: tagbase_server.controllers.ingest_controller + /ingest/jobs/{job_id}: + get: + description: Status of one ingest job + operationId: get_ingest_job + parameters: + - description: Job id returned when the file was accepted + explode: true + in: path + name: job_id + required: true + schema: + type: string + format: uuid + minLength: 36 + maxLength: 36 + pattern: ^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$ + style: simple + responses: + "200": + content: + application/json: + schema: + $ref: "#/components/schemas/IngestJob" + description: The job is queued, running, or finished. + "404": + $ref: "#/components/responses/NotFound" + "500": + $ref: "#/components/responses/InternalError" + summary: Get ingest job status + tags: + - ingest + x-openapi-router-controller: tagbase_server.controllers.ingest_controller /tags: delete: description: Delete all tags @@ -205,6 +237,12 @@ components: application/problem+json: schema: $ref: "#/components/schemas/Problem" + NotFound: + description: The ingest job id is not in the table. + content: + application/problem+json: + schema: + $ref: "#/components/schemas/Problem" InternalError: description: Unexpected server error. Contact admin detailed in openapi.yaml. content: @@ -353,38 +391,88 @@ components: maxLength: 36 pattern: ^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$ example: 123e4567-e89b-12d3-a456-426614174000 - Ingest200: - description: HTTP 200 success response + IngestAccepted: + description: HTTP 202 response after an ingest file is stored example: - code: "200" - elapsed: "0:00:06.506691" - message: - Data file eTUFF-sailfish-117259.txt successfully ingested into Tagbase - DB. + id: 123e4567-e89b-12d3-a456-426614174000 + status: queued properties: - code: - description: HTTP status code - example: "200" + id: + description: Job id + example: 123e4567-e89b-12d3-a456-426614174000 + format: uuid + maxLength: 36 + minLength: 36 + pattern: ^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$ type: string - minLength: 3 - maxLength: 3 - pattern: ^[1-5][0-9]{2}$ - elapsed: - description: Elapsed time for the operation + status: + description: Job status at accept time + enum: + - queued + example: queued + maxLength: 6 + minLength: 6 + pattern: ^queued$ type: string - minLength: 1 - maxLength: 64 - pattern: ^(?!\s*$).+$ - message: - description: A string detailing specifics of an HTTP operation - example: - Data file eTUFF-sailfish-117259.txt successfully ingested into - Tagbase DB. + required: + - id + - status + title: IngestAccepted + type: object + IngestJob: + description: Status of one ingest job + example: + id: 123e4567-e89b-12d3-a456-426614174000 + status: succeeded + files_total: 2 + files_done: 2 + properties: + id: + description: Job id + example: 123e4567-e89b-12d3-a456-426614174000 + format: uuid + maxLength: 36 + minLength: 36 + pattern: ^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$ type: string - minLength: 1 + status: + description: queued, running, succeeded, or failed + enum: + - queued + - running + - succeeded + - failed + example: succeeded + maxLength: 9 + minLength: 6 + pattern: ^(queued|running|succeeded|failed)$ + type: string + error: + description: Present when the job failed + example: restarted before ingest finished maxLength: 10000 + minLength: 1 pattern: ^(?!\s*$).+$ - title: Ingest200 + type: string + files_total: + description: Members in the archive, once unpacking has started + example: 2 + format: int32 + maximum: 2147483647 + minimum: 0 + type: integer + files_done: + description: Members ingested so far + example: 2 + format: int32 + maximum: 2147483647 + minimum: 0 + type: integer + required: + - id + - status + - files_done + title: IngestJob type: object Metadata: description: Contains the ingested tag metadata consistent with the eTUFF specification diff --git a/tagbase_server/tagbase_server/test/conftest.py b/tagbase_server/tagbase_server/test/conftest.py index 701ac5d5..f000b7a6 100644 --- a/tagbase_server/tagbase_server/test/conftest.py +++ b/tagbase_server/tagbase_server/test/conftest.py @@ -117,6 +117,14 @@ def clean_db(db_conn): with db_conn.cursor() as cur: cur.execute("TRUNCATE submission CASCADE") cur.execute("TRUNCATE dataset CASCADE") + cur.execute(""" + DO $$ + BEGIN + IF to_regclass('public.ingest_job') IS NOT NULL THEN + TRUNCATE ingest_job; + END IF; + END $$; + """) for attribute_id, category, name, description, necessity in METADATA_TYPE_SEED: cur.execute( """ diff --git a/tagbase_server/tagbase_server/test/helpers.py b/tagbase_server/tagbase_server/test/helpers.py index 7280e7a1..80ac2f49 100644 --- a/tagbase_server/tagbase_server/test/helpers.py +++ b/tagbase_server/tagbase_server/test/helpers.py @@ -20,3 +20,22 @@ def response_json(response): return json.loads(response.content) + + +def finish_accepted_ingest(client, response): + """Assert the upload was accepted, then run queued jobs in this process.""" + from tagbase_server.ingest_worker import process_available + + assert response.status_code == 202 + accepted = response_json(response) + assert accepted["status"] == "queued" + assert accepted["id"] + process_available() + job = client.get( + f"{API_PREFIX}/ingest/jobs/{accepted['id']}", + headers={"Accept": "application/json"}, + ) + assert job.status_code == 200 + body = response_json(job) + assert body["status"] == "succeeded", body + return body diff --git a/tagbase_server/tagbase_server/test/integration/test_ingest_api.py b/tagbase_server/tagbase_server/test/integration/test_ingest_api.py index 70f25d22..f5b6a27a 100644 --- a/tagbase_server/tagbase_server/test/integration/test_ingest_api.py +++ b/tagbase_server/tagbase_server/test/integration/test_ingest_api.py @@ -5,7 +5,12 @@ import pytest -from tagbase_server.test.helpers import API_PREFIX, ETUFF_FIXTURE, response_json +from tagbase_server.test.helpers import ( + API_PREFIX, + ETUFF_FIXTURE, + finish_accepted_ingest, + response_json, +) pytestmark = pytest.mark.integration @@ -34,10 +39,7 @@ def test_post_ingest_etuff_makes_tag_listable_and_gettable( }, data=etuff_bytes, ) - assert response.status_code == 200 - body = response_json(response) - assert body["code"] == "200" - assert "minimal-etuff.txt" in body["message"] + finish_accepted_ingest(client, response) listed = client.get(f"{API_PREFIX}/tags", headers={"Accept": "application/json"}) assert listed.status_code == 200 @@ -70,8 +72,8 @@ def test_post_ingest_duplicate_etuff_keeps_single_submission( second = client.post( f"{API_PREFIX}/ingest", params=params, headers=headers, data=etuff_bytes ) - assert first.status_code == 200 - assert second.status_code == 200 + finish_accepted_ingest(client, first) + finish_accepted_ingest(client, second) listed = response_json( client.get(f"{API_PREFIX}/tags", headers={"Accept": "application/json"}) @@ -97,7 +99,7 @@ def test_post_ingest_zip_etuff_makes_tag_retrievable(client, clean_db, etuff_zip }, data=etuff_zip_bytes, ) - assert response.status_code == 200 + finish_accepted_ingest(client, response) listed = response_json( client.get(f"{API_PREFIX}/tags", headers={"Accept": "application/json"}) ) @@ -114,7 +116,7 @@ def test_get_ingest_file_url_makes_tag_listable(client, clean_db, short_etuff_fi }, headers={"Accept": "application/json"}, ) - assert response.status_code == 200 + finish_accepted_ingest(client, response) listed = response_json( client.get(f"{API_PREFIX}/tags", headers={"Accept": "application/json"}) ) @@ -162,7 +164,7 @@ def test_ingest_defaults_omitted_type_to_etuff(client, clean_db, etuff_bytes): }, data=etuff_bytes, ) - assert response.status_code == 200 + finish_accepted_ingest(client, response) listed = response_json( client.get(f"{API_PREFIX}/tags", headers={"Accept": "application/json"}) ) diff --git a/tagbase_server/tagbase_server/test/integration/test_tags_api.py b/tagbase_server/tagbase_server/test/integration/test_tags_api.py index 18037d70..eb1be37c 100644 --- a/tagbase_server/tagbase_server/test/integration/test_tags_api.py +++ b/tagbase_server/tagbase_server/test/integration/test_tags_api.py @@ -2,7 +2,11 @@ import pytest -from tagbase_server.test.helpers import API_PREFIX, response_json +from tagbase_server.test.helpers import ( + API_PREFIX, + finish_accepted_ingest, + response_json, +) pytestmark = pytest.mark.integration @@ -21,7 +25,7 @@ def _ingest(client, etuff_bytes, notes="notes", version="1"): }, data=etuff_bytes, ) - assert response.status_code == 200 + finish_accepted_ingest(client, response) listed = response_json( client.get(f"{API_PREFIX}/tags", headers={"Accept": "application/json"}) ) diff --git a/tagbase_server/tagbase_server/test/stack/test_fswatch_contract.py b/tagbase_server/tagbase_server/test/stack/test_fswatch_contract.py index 4fd6d5bc..b6a96606 100644 --- a/tagbase_server/tagbase_server/test/stack/test_fswatch_contract.py +++ b/tagbase_server/tagbase_server/test/stack/test_fswatch_contract.py @@ -81,10 +81,25 @@ def do_POST(self): # noqa: N802 "body": body, } ) + self.send_response(202) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write( + b'{"id":"11111111-1111-1111-1111-111111111111","status":"queued"}' + ) + + def do_GET(self): # noqa: N802 + parsed = urlparse(self.path) + if not parsed.path.endswith("/11111111-1111-1111-1111-111111111111"): + self.send_response(404) + self.end_headers() + return self.send_response(200) self.send_header("Content-Type", "application/json") self.end_headers() - self.wfile.write(b'{"code":"200","message":"ok","elapsed":"0"}') + self.wfile.write( + b'{"id":"11111111-1111-1111-1111-111111111111","status":"succeeded","error":null,"files_total":1,"files_done":1}' + ) def log_message(self, format, *args): # noqa: A003 return diff --git a/tagbase_server/tagbase_server/test/test_ingest_jobs.py b/tagbase_server/tagbase_server/test/test_ingest_jobs.py new file mode 100644 index 00000000..f96f89ae --- /dev/null +++ b/tagbase_server/tagbase_server/test/test_ingest_jobs.py @@ -0,0 +1,71 @@ +# coding: utf-8 + +from unittest.mock import patch + +import pytest + +from tagbase_server.ingest_jobs import MISSING_FILE_ERROR, RESTART_ERROR, stale_failures +from tagbase_server.ingest_worker import process_next, run_job + + +def test_stale_failures_marks_running_and_missing_queued_files(): + jobs = [ + {"id": "running-1", "status": "running", "source_path": "/tmp/a.zip"}, + {"id": "queued-gone", "status": "queued", "source_path": "/tmp/missing.zip"}, + {"id": "queued-kept", "status": "queued", "source_path": "/tmp/kept.zip"}, + {"id": "done", "status": "succeeded", "source_path": "/tmp/done.zip"}, + ] + + failed = stale_failures(jobs, lambda path: path == "/tmp/kept.zip") + + assert failed == [ + ("running-1", RESTART_ERROR), + ("queued-gone", MISSING_FILE_ERROR), + ] + + +def test_run_job_stops_on_the_first_member_error(): + job = { + "id": "job-1", + "source_path": "/tmp/archive.zip", + "version": "1", + "notes": None, + } + seen = [] + + def _ingest(path, version=None, notes=None, count_request=True): + seen.append((path, count_request)) + if path.endswith("b.txt"): + raise RuntimeError("bad member") + + with ( + patch( + "tagbase_server.ingest_worker.members_for", + return_value=["/tmp/a.txt", "/tmp/b.txt", "/tmp/c.txt"], + ), + patch("tagbase_server.ingest_worker.process_etuff_file", side_effect=_ingest), + patch("tagbase_server.ingest_worker.set_progress") as progress, + patch("tagbase_server.ingest_worker.finish_job") as finish, + ): + with pytest.raises(RuntimeError, match="bad member"): + run_job(job) + + assert [path for path, _counted in seen] == ["/tmp/a.txt", "/tmp/b.txt"] + assert all(counted is False for _path, counted in seen) + finish.assert_not_called() + assert progress.call_args_list[-1].kwargs["files_done"] == 1 + + +def test_process_next_fails_the_job_and_keeps_the_queue_moving(): + job = {"id": "job-2", "source_path": "/tmp/one.txt", "version": None, "notes": None} + with ( + patch("tagbase_server.ingest_worker.claim_next_job", return_value=job), + patch( + "tagbase_server.ingest_worker.run_job", + side_effect=RuntimeError("stopped"), + ), + patch("tagbase_server.ingest_worker.finish_job") as finish, + ): + assert process_next() is True + + finish.assert_called_once_with("job-2", status="failed", error="stopped") diff --git a/tagbase_server/tagbase_server/test/test_sonar_cleanup.py b/tagbase_server/tagbase_server/test/test_sonar_cleanup.py index 6822d80f..ac886575 100644 --- a/tagbase_server/tagbase_server/test/test_sonar_cleanup.py +++ b/tagbase_server/tagbase_server/test/test_sonar_cleanup.py @@ -12,7 +12,7 @@ from tagbase_server.controllers import ingest_controller from tagbase_server.controllers.ingest_controller import _resolve_ingest_file_type from tagbase_server.models.base_model_ import Model -from tagbase_server.models.ingest200 import Ingest200 +from tagbase_server.models.tag_put200 import TagPut200 from tagbase_server.utils import io_utils from tagbase_server.utils import processing_utils as pu @@ -125,34 +125,45 @@ def test_resolve_ingest_file_type_defaults_and_rejects(): _resolve_ingest_file_type("csv") -@mock.patch("tagbase_server.controllers.ingest_controller.parmap.map") +@mock.patch("tagbase_server.controllers.ingest_controller.record_ingest_request") +@mock.patch( + "tagbase_server.controllers.ingest_controller.enqueue_job", return_value="job-1" +) @mock.patch("tagbase_server.controllers.ingest_controller.process_get_input_data") -def test_ingest_get_happy_path(mock_get, mock_map): +def test_ingest_get_happy_path(mock_get, mock_enqueue, mock_record): mock_get.return_value = "/data/file.txt" - mock_map.return_value = [0] result, status, headers = ingest_controller.ingest_get( "file:///data/file.txt", type="etuff" ) - assert status == 200 + assert status == 202 assert headers["Content-Type"] == "application/json" - assert result.code == "200" - mock_map.assert_called_once() - - -@mock.patch("tagbase_server.controllers.ingest_controller.parmap.map") -@mock.patch("tagbase_server.controllers.ingest_controller.unpack_compressed_binary") + assert result == {"id": "job-1", "status": "queued"} + mock_enqueue.assert_called_once() + mock_record.assert_called_once() + assert mock_record.call_args.args[0] == "accepted" + + +@mock.patch("tagbase_server.controllers.ingest_controller.record_ingest_request") +@mock.patch( + "tagbase_server.controllers.ingest_controller.enqueue_job", return_value="job-2" +) +@mock.patch( + "tagbase_server.controllers.ingest_controller._isolate_upload", + return_value="/tmp/unique-bundle.zip", +) @mock.patch("tagbase_server.controllers.ingest_controller.process_post_input_data") -def test_ingest_post_archive_path(mock_post, mock_unpack, mock_map): - mock_post.return_value = "/data/bundle.zip" - mock_unpack.return_value = ["/data/a.txt"] - mock_map.return_value = [0] +def test_ingest_post_archive_path(mock_post, mock_isolate, mock_enqueue, mock_record): + mock_post.return_value = "/tmp/bundle.zip" result, status, headers = ingest_controller.ingest_post( "bundle.zip", b"x", type=None ) - assert status == 200 + assert status == 202 assert headers["Content-Type"] == "application/json" - assert result.code == "200" - mock_unpack.assert_called_once() + assert result["status"] == "queued" + mock_isolate.assert_called_once_with("/tmp/bundle.zip") + mock_enqueue.assert_called_once() + assert mock_enqueue.call_args.kwargs["source_path"] == "/tmp/unique-bundle.zip" + mock_record.assert_called_once() def test_processing_utils_has_no_pytz(): @@ -335,7 +346,7 @@ def __init__(self): assert as_dict["items"][0]["name"] == "a" assert as_dict["nested"]["name"] == "b" assert as_dict["mapping"]["k"]["name"] == "c" - a = Ingest200.from_dict({"code": "200", "elapsed": 1.0, "message": "a"}) - b = Ingest200.from_dict({"code": "200", "elapsed": 1.0, "message": "b"}) + a = TagPut200.from_dict({"code": "200", "message": "a"}) + b = TagPut200.from_dict({"code": "200", "message": "b"}) assert a != b assert not (a != a) diff --git a/tagbase_server/tagbase_server/utils/processing_utils.py b/tagbase_server/tagbase_server/utils/processing_utils.py index 8c880fff..e7564a9c 100644 --- a/tagbase_server/tagbase_server/utils/processing_utils.py +++ b/tagbase_server/tagbase_server/utils/processing_utils.py @@ -534,9 +534,20 @@ def _compute_submission_hashes(submission_filename, file_content, metadata_conte ) -def process_etuff_file(file, version=None, notes=None): +def process_etuff_file(file, version=None, notes=None, count_request=True): start = time.perf_counter() submission_filename = file + + def _record(outcome, duration): + if count_request: + record_ingest_request(outcome, duration) + + def _soft_error(outcome, duration, message): + _record(outcome, duration) + if not count_request: + raise RuntimeError(message) + return 1 + tracer = get_tracer() logger.info( "Processing etuff file: %s", @@ -546,8 +557,11 @@ def process_etuff_file(file, version=None, notes=None): conn = connect() if not hasattr(conn, "cursor"): record_db_error("connect") - record_ingest_request("error", round(time.perf_counter() - start, 2)) - return 1 + return _soft_error( + "error", + round(time.perf_counter() - start, 2), + "database unavailable", + ) conn.autocommit = True with tracer.start_as_current_span("ingest.parse_etuff"): @@ -571,7 +585,7 @@ def process_etuff_file(file, version=None, notes=None): ) except Exception: logger.exception("Error parsing etuff file: %s", submission_filename) - record_ingest_request("error", round(time.perf_counter() - start, 2)) + _record("error", round(time.perf_counter() - start, 2)) raise with conn: @@ -584,9 +598,7 @@ def process_etuff_file(file, version=None, notes=None): submission_filename, entire_file_hash, ) - record_ingest_request( - "duplicate", round(time.perf_counter() - start, 2) - ) + _record("duplicate", round(time.perf_counter() - start, 2)) return 1 dataset_id = get_dataset_id( @@ -634,9 +646,7 @@ def process_etuff_file(file, version=None, notes=None): dataset_id, metadata_hash, ) - record_ingest_request( - "metadata_only", round(time.perf_counter() - start, 2) - ) + _record("metadata_only", round(time.perf_counter() - start, 2)) return 1 proc_obs, len_proc_obs = _build_proc_observations( @@ -654,10 +664,11 @@ def process_etuff_file(file, version=None, notes=None): ) if sub_elapsed is False: record_db_error("migrate_proc_observations") - record_ingest_request( - "error", round(time.perf_counter() - start, 2) + return _soft_error( + "error", + round(time.perf_counter() - start, 2), + "failed to migrate proc observations", ) - return 1 record_rows_written(len_proc_obs) logger.info( "Successful migration of %s 'proc_observations'. Elapsed time: %s second(s).", @@ -666,9 +677,7 @@ def process_etuff_file(file, version=None, notes=None): ) except Exception: record_db_error("persist") - record_ingest_request( - "error", round(time.perf_counter() - start, 2) - ) + _record("error", round(time.perf_counter() - start, 2)) raise conn.commit() @@ -678,7 +687,7 @@ def process_etuff_file(file, version=None, notes=None): finish = time.perf_counter() elapsed = round(finish - start, 2) - record_ingest_request("success", elapsed) + _record("success", elapsed) logger.info( "Data file %s successfully ingested into Tagbase DB. Total time: %s second(s)", submission_filename, From 7257b0d341ce22524bed5880ad68e28ba9ae0540 Mon Sep 17 00:00:00 2001 From: Lewis John McGibbney Date: Mon, 12 Oct 2026 13:59:10 -0700 Subject: [PATCH 2/5] fix: clear the Sonar reliability findings on ingest uploads The job upload opened its staging file inside an async function, and repeated error strings were duplicated literals. --- tagbase_server/tagbase_server/archives.py | 23 +++++----- .../controllers/ingest_controller.py | 4 +- .../tagbase_server/upload_stream.py | 42 +++++++++++-------- 3 files changed, 40 insertions(+), 29 deletions(-) diff --git a/tagbase_server/tagbase_server/archives.py b/tagbase_server/tagbase_server/archives.py index 2325a894..4cd46ab0 100644 --- a/tagbase_server/tagbase_server/archives.py +++ b/tagbase_server/tagbase_server/archives.py @@ -22,6 +22,9 @@ _DEFAULT_MAX_MEMBERS = 10_000 _DEFAULT_MAX_UNCOMPRESSED = 10 * 1024 * 1024 * 1024 _READ_CHUNK = 1024 * 1024 +_INVALID_FILENAME = "invalid filename" +_NO_ETUFF_MEMBERS = "archive contained no eTUFF files" +_CORRUPT_ARCHIVE = "archive is corrupt or unreadable" class ArchiveError(ValueError): @@ -66,13 +69,13 @@ def archive_limits() -> tuple[int, int]: def logical_filename(name: str) -> str: """Reject blank names and any path, keeping a single basename.""" if name is None: - raise ArchiveError("invalid filename") + raise ArchiveError(_INVALID_FILENAME) text = str(name).strip() if not text or text in {".", ".."} or "/" in text or "\\" in text: - raise ArchiveError("invalid filename") + raise ArchiveError(_INVALID_FILENAME) base = os.path.basename(text) if base != text: - raise ArchiveError("invalid filename") + raise ArchiveError(_INVALID_FILENAME) return base @@ -80,7 +83,7 @@ def classify(path: str, filename: str) -> str: """Return etuff, zip, or tar after the name and magic bytes agree.""" logical = logical_filename(filename) lowered = logical.lower() - if lowered.endswith(".tar.gz") or lowered.endswith(".tgz"): + if lowered.endswith((".tar.gz", ".tgz")): expected = "tar" elif lowered.endswith(".zip"): expected = "zip" @@ -132,7 +135,7 @@ def materialize(path: str, filename: str) -> PreparedIngest: raise if not members: shutil.rmtree(workspace, ignore_errors=True) - raise ArchiveError("archive contained no eTUFF files") + raise ArchiveError(_NO_ETUFF_MEMBERS) return PreparedIngest( files=[member_path for member_path, _name in members], workspace=workspace, @@ -187,9 +190,9 @@ def _inspect_zip(path: str) -> None: txts += 1 _limits_exceeded(index, declared) except zipfile.BadZipFile as exc: - raise ArchiveError("archive is corrupt or unreadable") from exc + raise ArchiveError(_CORRUPT_ARCHIVE) from exc if txts == 0: - raise ArchiveError("archive contained no eTUFF files") + raise ArchiveError(_NO_ETUFF_MEMBERS) def _reject_zip_member(info: zipfile.ZipInfo) -> None: @@ -242,9 +245,9 @@ def _inspect_tar(path: str) -> None: txts += 1 _limits_exceeded(count, declared) except (tarfile.TarError, OSError) as exc: - raise ArchiveError("archive is corrupt or unreadable") from exc + raise ArchiveError(_CORRUPT_ARCHIVE) from exc if txts == 0: - raise ArchiveError("archive contained no eTUFF files") + raise ArchiveError(_NO_ETUFF_MEMBERS) def _reject_tar_member(member: tarfile.TarInfo) -> None: @@ -267,7 +270,7 @@ def _extract_tar(path: str, workspace: str) -> list[tuple[str, str]]: os.makedirs(os.path.dirname(target), exist_ok=True) source = archive.extractfile(member) if source is None: - raise ArchiveError("archive is corrupt or unreadable") + raise ArchiveError(_CORRUPT_ARCHIVE) with source, open(target, "wb") as dest: while True: chunk = source.read(_READ_CHUNK) diff --git a/tagbase_server/tagbase_server/controllers/ingest_controller.py b/tagbase_server/tagbase_server/controllers/ingest_controller.py index e1fe4efc..1f3f3122 100644 --- a/tagbase_server/tagbase_server/controllers/ingest_controller.py +++ b/tagbase_server/tagbase_server/controllers/ingest_controller.py @@ -192,7 +192,7 @@ def _work(): return _ingest_with_telemetry("post", _work) -def _stage_job_body(filename, body) -> str: +def _stage_job_body(body) -> str: staged = staged_upload.get() if staged: return staged @@ -238,7 +238,7 @@ def create_ingest_job( stored = None try: _resolve_ingest_file_type(type_ if type_ is not None else type) - staged = _stage_job_body(filename, body) + staged = _stage_job_body(body) try: preflight(staged, filename) except ArchiveError as exc: diff --git a/tagbase_server/tagbase_server/upload_stream.py b/tagbase_server/tagbase_server/upload_stream.py index 56542013..4f325d5c 100644 --- a/tagbase_server/tagbase_server/upload_stream.py +++ b/tagbase_server/tagbase_server/upload_stream.py @@ -20,6 +20,10 @@ JOB_MAX_BYTES = 2 * 1024 * 1024 * 1024 _STAGED_BODY = b"staged" _CHUNK = 1024 * 1024 +_HTTP_REQUEST = "http.request" +_TITLE_BAD_REQUEST = "Bad Request" +_TITLE_TOO_LARGE = "Payload Too Large" +_EMPTY_BODY = "Request body is empty." staged_upload: contextvars.ContextVar[str | None] = contextvars.ContextVar( "staged_upload", @@ -74,13 +78,13 @@ async def _guard_sync(self, scope, receive, send): length = _content_length(scope) limit = sync_max_bytes() if length == 0: - await _send_problem(send, 400, "Bad Request", "Request body is empty.") + await _send_problem(send, 400, _TITLE_BAD_REQUEST, _EMPTY_BODY) return if length is not None and length > limit: await _send_problem( send, 413, - "Payload Too Large", + _TITLE_TOO_LARGE, "Synchronous ingest accepts at most 100 MiB. Use POST /ingest/jobs.", ) return @@ -93,14 +97,14 @@ async def _guard_sync(self, scope, receive, send): await _send_problem( send, 413, - "Payload Too Large", + _TITLE_TOO_LARGE, "Synchronous ingest accepts at most 100 MiB. Use POST /ingest/jobs.", ) return except _ClientGone: return if not body: - await _send_problem(send, 400, "Bad Request", "Request body is empty.") + await _send_problem(send, 400, _TITLE_BAD_REQUEST, _EMPTY_BODY) return await self.app( _with_content_length(scope, len(body)), @@ -112,21 +116,17 @@ async def _stream_job(self, scope, receive, send): length = _content_length(scope) limit = job_max_bytes() if length == 0: - await _send_problem(send, 400, "Bad Request", "Request body is empty.") + await _send_problem(send, 400, _TITLE_BAD_REQUEST, _EMPTY_BODY) return if length is not None and length > limit: await _send_problem( send, 413, - "Payload Too Large", + _TITLE_TOO_LARGE, "Ingest jobs accept at most 2 GiB.", ) return - handle = tempfile.NamedTemporaryFile( - dir=TEMP_DIR, - prefix="tagbase-upload-", - delete=False, - ) + handle = _create_staged_file() path = handle.name try: total = await _write_bounded(receive, handle, limit) @@ -136,7 +136,7 @@ async def _stream_job(self, scope, receive, send): await _send_problem( send, 413, - "Payload Too Large", + _TITLE_TOO_LARGE, "Ingest jobs accept at most 2 GiB.", ) return @@ -151,7 +151,7 @@ async def _stream_job(self, scope, receive, send): handle.close() if total == 0: _remove(path) - await _send_problem(send, 400, "Bad Request", "Request body is empty.") + await _send_problem(send, 400, _TITLE_BAD_REQUEST, _EMPTY_BODY) return token = staged_upload.set(path) try: @@ -164,6 +164,14 @@ async def _stream_job(self, scope, receive, send): staged_upload.reset(token) +def _create_staged_file(): + return tempfile.NamedTemporaryFile( + dir=TEMP_DIR, + prefix="tagbase-upload-", + delete=False, + ) + + def _content_length(scope) -> int | None: for key, value in scope.get("headers", []): if key.lower() == b"content-length": @@ -192,9 +200,9 @@ def _replay(body: bytes): async def receive(): nonlocal sent if sent: - return {"type": "http.request", "body": b"", "more_body": False} + return {"type": _HTTP_REQUEST, "body": b"", "more_body": False} sent = True - return {"type": "http.request", "body": body, "more_body": False} + return {"type": _HTTP_REQUEST, "body": body, "more_body": False} return receive @@ -206,7 +214,7 @@ async def _read_bounded(receive, limit: int) -> bytes: message = await receive() if message["type"] == "http.disconnect": raise _ClientGone() - if message["type"] != "http.request": + if message["type"] != _HTTP_REQUEST: continue piece = message.get("body") or b"" if piece: @@ -225,7 +233,7 @@ async def _write_bounded(receive, handle, limit: int) -> int: message = await receive() if message["type"] == "http.disconnect": raise _ClientGone() - if message["type"] != "http.request": + if message["type"] != _HTTP_REQUEST: continue piece = message.get("body") or b"" if piece: From 79813abd2804bbf82aaf6fe743af6bb343f20caa Mon Sep 17 00:00:00 2001 From: Lewis John McGibbney Date: Mon, 12 Oct 2026 14:14:10 -0700 Subject: [PATCH 3/5] fix: keep the staging watcher moving through every dropped file bash 5 treated a zero-second read as another copy of the previous path, so later drops never reached the job API. The watcher also exited on startup because its state volume was not writable. --- services/fswatch/Dockerfile | 4 +++- services/fswatch/post.sh | 3 ++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/services/fswatch/Dockerfile b/services/fswatch/Dockerfile index fa38ddb4..f094d17d 100644 --- a/services/fswatch/Dockerfile +++ b/services/fswatch/Dockerfile @@ -7,7 +7,9 @@ RUN apt-get update \ install -y curl fswatch iputils-ping \ # Remove package lists for smaller image sizes && rm -rf /var/lib/apt/lists/* \ - && useradd -ms /bin/bash tagbase + && useradd -ms /bin/bash tagbase \ + && mkdir -p /var/lib/tagbase-fswatch \ + && chown tagbase:tagbase /var/lib/tagbase-fswatch # Copy owned by root and grant the runtime user read+execute only, so the # non-root user running the container cannot modify the script it executes. diff --git a/services/fswatch/post.sh b/services/fswatch/post.sh index 63fee0ab..8f194242 100755 --- a/services/fswatch/post.sh +++ b/services/fswatch/post.sh @@ -212,7 +212,8 @@ while true; do reconcile_pending if IFS= read -r -t "$RECONCILE_SECONDS" -u 3 line; then submit_file "$line" - while IFS= read -r -t 0 -u 3 line; do + # bash 5 `read -t 0` repeats the previous line and never consumes the fifo. + while IFS= read -r -t 1 -u 3 line; do submit_file "$line" done fi From 3946ff16d7854f0c089c4df83983775bfdb74c74 Mon Sep 17 00:00:00 2001 From: Lewis John McGibbney Date: Mon, 12 Oct 2026 15:01:24 -0700 Subject: [PATCH 4/5] fix: wait for both staging jobs to be recorded as pending The contract test read the pending list as soon as the second POST returned, before the watcher finished writing that job. --- .../test/stack/test_fswatch_contract.py | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/tagbase_server/tagbase_server/test/stack/test_fswatch_contract.py b/tagbase_server/tagbase_server/test/stack/test_fswatch_contract.py index c16598d1..f607082b 100644 --- a/tagbase_server/tagbase_server/test/stack/test_fswatch_contract.py +++ b/tagbase_server/tagbase_server/test/stack/test_fswatch_contract.py @@ -332,21 +332,28 @@ def test_fswatch_keeps_submitting_while_a_job_is_running(mock_ingest_server): time.sleep(0.4) (staging / "first.txt").write_bytes(b"one") (staging / "second.txt").write_bytes(b"two") + # record_pending runs after the POST returns, so wait for the file too. deadline = time.time() + 30 + pending_path = work / "pending.tsv" + pending = "" while time.time() < deadline: names = { (r["query"].get("filename") or [None])[0] for r in handler.requests } - if names >= {"first.txt", "second.txt"}: + if pending_path.is_file(): + pending = pending_path.read_text(encoding="utf-8") + if ( + names >= {"first.txt", "second.txt"} + and "first.txt" in pending + and "second.txt" in pending + ): break time.sleep(0.25) else: pytest.fail( - f"expected both posts while running. log:\n{_read_log(log_path)}" + "expected both posts to stay pending while running. " + f"pending:\n{pending}\nlog:\n{_read_log(log_path)}" ) - pending = (work / "pending.tsv").read_text(encoding="utf-8") - assert "first.txt" in pending - assert "second.txt" in pending finally: _stop_fswatch(proc) shutil.rmtree(work, ignore_errors=True) From 14d50ab5579df0f9aa1787e706915f6b8f8ca1fc Mon Sep 17 00:00:00 2001 From: Lewis John McGibbney Date: Mon, 12 Oct 2026 19:12:19 -0700 Subject: [PATCH 5/5] fix: rebuild the time-series date index after large eTUFF loads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Maintaining data_time_series_date_time_index on every row held a large load to about 29 MB/min of heap once existing data already covered 2011–2026. Submissions of at least one million time-series rows now drop that index for the insert and create it once, in the same transaction, after positions are linked. --- services/postgis/tagbase_schema.sql | 23 ++++++++++++++++++++++ services/postgis/tagbase_schema_tables.sql | 23 ++++++++++++++++++++++ 2 files changed, 46 insertions(+) diff --git a/services/postgis/tagbase_schema.sql b/services/postgis/tagbase_schema.sql index e48d8cc6..f6600b48 100644 --- a/services/postgis/tagbase_schema.sql +++ b/services/postgis/tagbase_schema.sql @@ -1163,8 +1163,28 @@ $$; CREATE PROCEDURE sp_execute_data_migration(submission_id_param integer, is_reference_track_param boolean) LANGUAGE plpgsql AS $BODY$ +DECLARE + ts_rows bigint; + defer_date_index boolean; BEGIN --\connect tagbase + -- Once older submissions already cover this time range, maintaining the + -- global date_time btree is a mid-index insert per row (~29 MB/min of heap). + -- One million rows is about 100 MB of heap. Below that, rebuilding the + -- whole index costs more than leaving it in place. + SELECT count(*) INTO ts_rows + FROM proc_observations a + JOIN observation_types c ON a.variable_id = c.variable_id + WHERE a.submission_id = submission_id_param + AND c.variable_name IN ('datetime', + 'depth', + 'temperature', + 'light', + 'internal temperature'); + defer_date_index := ts_rows >= 1000000; + IF defer_date_index THEN + DROP INDEX data_time_series_date_time_index; + END IF; -- data_time_series WITH moved_rows AS ( DELETE @@ -1417,6 +1437,9 @@ BEGIN AND t.submission_id = m.submission_id AND t.tag_id = m.tag_id AND t.date_time = m.date_time; + IF defer_date_index THEN + CREATE INDEX data_time_series_date_time_index ON data_time_series USING btree (date_time); + END IF; END; $BODY$; diff --git a/services/postgis/tagbase_schema_tables.sql b/services/postgis/tagbase_schema_tables.sql index 46a556ed..a4ffcdf9 100644 --- a/services/postgis/tagbase_schema_tables.sql +++ b/services/postgis/tagbase_schema_tables.sql @@ -1158,8 +1158,28 @@ $$; CREATE PROCEDURE sp_execute_data_migration(submission_id_param integer, is_reference_track_param boolean) LANGUAGE plpgsql AS $BODY$ +DECLARE + ts_rows bigint; + defer_date_index boolean; BEGIN --\connect tagbase + -- Once older submissions already cover this time range, maintaining the + -- global date_time btree is a mid-index insert per row (~29 MB/min of heap). + -- One million rows is about 100 MB of heap. Below that, rebuilding the + -- whole index costs more than leaving it in place. + SELECT count(*) INTO ts_rows + FROM proc_observations a + JOIN observation_types c ON a.variable_id = c.variable_id + WHERE a.submission_id = submission_id_param + AND c.variable_name IN ('datetime', + 'depth', + 'temperature', + 'light', + 'internal temperature'); + defer_date_index := ts_rows >= 1000000; + IF defer_date_index THEN + DROP INDEX data_time_series_date_time_index; + END IF; -- data_time_series WITH moved_rows AS ( DELETE @@ -1412,6 +1432,9 @@ BEGIN AND t.submission_id = m.submission_id AND t.tag_id = m.tag_id AND t.date_time = m.date_time; + IF defer_date_index THEN + CREATE INDEX data_time_series_date_time_index ON data_time_series USING btree (date_time); + END IF; END; $BODY$;