Cache issue

[lpatiny]

/submit cache never fires: deployed image predates the feature, plus three bugs that will keep it from working after a rebuild

Submitting the same spectrum twice always starts a new run. Investigated against the
live deployment at elucidation.cheminfo.org; there are two independent problems.

1. The deployed API image predates the cache check

The cache check was added in 246a7b4 (2026-07-06 12:58). adrianmirza/elucidation:latest
was built 2026-07-05 14:38, about 22 h earlier. Extracting app.py from the image
layers confirms the running /submit goes straight from short_vector_hash to
send_task:

# adrianmirza/elucidation:latest, /app/app.py
@app.post("/submit")
async def submit_spectrum_job(request: GenerateRequest):
    """Submit a spectrum processing job to the Celery queue"""
    try:
        spectrum_as_array = np.array(request.spectrum["y"])
        job_id = short_vector_hash(spectrum_as_array)

        request_data = request.model_dump()          # <-- no result_file.exists() check
        request_data["spectra_hash"] = job_id

        task = celery_app.send_task(...)

grep "exists()\|cached" over the deployed app.py returns nothing; result_file
appears there only in /result and DELETE /cache.

This is not a deployment-config problem. Both images are WORKDIR /app with
COPY ./config.py ./config.py, so CACHE_DIR = /app/cache, which is exactly what
./cache:/app/cache mounts — and /result successfully reads a worker-written file
from that path, which proves the volume is shared and readable. config.py also reads
no environment variable, so nothing in .env can affect it.

The API image just needs rebuilding and pushing. Docker Hub only has latest and
v2 (2025-08-18), so there is no newer tag to pin.

2. Three bugs in main that will surface once the image is rebuilt

a. GA snapshots occupy the final result path

gafuncs.py:68 writes CACHE_DIR/<hash>.json on every batch evaluation, so
result_file.exists() becomes true as soon as a run starts. /submit then answers
"cached" for a job that has not finished, and the follow-up /result returns
400 Job not completed. The same path also holds two different shapes — a bare
top-128 array while running, {results, metadata} when done — so clients must handle
both.

b. /result gates on Celery state instead of the file

With result_expires = 3600, an hour after completion AsyncResult(...).state is no
longer SUCCESS and /result returns 400 "Job not completed. Current status: PENDING"
even though the file is on disk. After 24 h the job_id -> task_id mapping (ex=86400)
expires and it 404s instead. A finished result becomes permanently unreachable.

c. /submit unconditionally overwrites the Redis mapping

Resubmitting a spectrum whose run already completed repoints job_id -> task_id at the
new task, so /result for the finished job returns 400 until the new run ends.

Reproduction

POST /submit {mf: "C4H8O", spectrum: {...10000 pts...}}
  -> {"job_id":"d2cc7dd964d5defb38819a3717c33875","task_id":"643e46e1-...","status":"submitted"}
     (12 min run)

GET  /jobs/d2cc7dd964d5defb38819a3717c33875/result
  -> 200, 512 candidates

POST /submit   (byte-identical spectrum)
  -> {"job_id":"d2cc7dd964d5defb38819a3717c33875","task_id":"6de5b678-...","status":"submitted"}
     (not cached; a second 12 min run is queued)

GET  /jobs/d2cc7dd964d5defb38819a3717c33875/result
  -> 400 {"detail":"Job not completed. Current status: PROGRESS"}
     (the completed result is now unreachable)

A third submission returned yet another task_id while a task for the same job_id
had been running for 20 minutes — and the GA rewrites <hash>.json on every batch
during that window, so the file was certainly present.

Suggested patch

Three files, +36/-20, all py_compile-clean. It

• writes GA snapshots to <hash>.partial.json, so <hash>.json existing means
  complete and /submit's check becomes meaningful;
• makes /result serve the stored file whenever it exists, consulting Celery only to
  explain a miss, which removes both the 1 h and the 24 h cliff;
• writes the final payload via temp file + replace(), since /result may read it
  concurrently and an in-place write can be observed half-finished.

Not addressed, because the right behaviour is not obvious: /submit could return the
in-flight task_id instead of queuing a duplicate, but Celery reports PENDING both
for queued tasks and for unknown ids, so telling "queued" from "expired" needs a
separate marker.

Patch

diff --git a/app.py b/app.py
index 7ff87db..437f5a8 100644
--- a/app.py
+++ b/app.py
@@ -119,25 +119,30 @@ def get_job_status(job_id: str):
 @app.get("/jobs/{job_id}/result")
 def get_job_result(job_id: str):
     """Get the result of a completed job"""
-    # --- MODIFIED: Get task_id from Redis ---
-    task_id = get_task_id_from_job_id(job_id)
-    task_result = AsyncResult(task_id, app=celery_app)
+    # The stored file is the source of truth. Celery drops the result after
+    # result_expires (1 h) and the job_id -> task_id mapping expires after 24 h, so
+    # gating on task state made a result that is still on disk unreachable.
+    result_file = CACHE_DIR / f"{job_id}.json"
+    if result_file.exists():
+        try:
+            with result_file.open("r") as f:
+                return json.load(f)
+        except json.JSONDecodeError as e:
+            raise HTTPException(  # noqa: B904
+                status_code=500, detail=f"Result file is not valid JSON: {e!s}"
+            )
+        except Exception as e:
+            raise HTTPException(status_code=500, detail=f"Error loading results: {e!s}")  # noqa: B904
 
-    if task_result.state != "SUCCESS":
-        raise HTTPException(
-            status_code=400,
-            detail=f"Job not completed. Current status: {task_result.state}",
-        )
+    # No stored result: use the task, when we still know it, to explain why.
+    task_id = redis_client.get(job_id)
+    if not task_id:
+        raise HTTPException(status_code=404, detail="Job not found")
 
-    result_file = CACHE_DIR / f"{job_id}.json"
-    try:
-        with result_file.open("r") as f:
-            data = json.load(f)
-        return data
-    except FileNotFoundError:
-        raise HTTPException(status_code=404, detail="Result file not found")  # noqa: B904
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=f"Error loading results: {e!s}")  # noqa: B904
+    state = AsyncResult(task_id, app=celery_app).state
+    raise HTTPException(
+        status_code=400, detail=f"Job not completed. Current status: {state}"
+    )
 
 
 @app.delete("/jobs/{job_id}")
diff --git a/celery_config.py b/celery_config.py
index 85a2691..7f29a3d 100644
--- a/celery_config.py
+++ b/celery_config.py
@@ -98,10 +98,18 @@ def process_spectrum(self, job_id: str, request_data: dict):
             },
         }
 
-        # Save the result to a file for persistence/backup
+        # Save the result to a file for persistence/backup. Written to a temporary
+        # file and renamed, because /jobs/{id}/result may read it concurrently and an
+        # in-place write can be observed half-finished.
         result_file = CACHE_DIR / f"{job_id}.json"
-        with result_file.open("w") as f:
+        temp_file = CACHE_DIR / f"{job_id}.json.tmp"
+        with temp_file.open("w") as f:
             json.dump(final_payload, f)
+        temp_file.replace(result_file)
+
+        # The in-progress snapshot is superseded by the final payload.
+        partial_file = CACHE_DIR / f"{job_id}.partial.json"
+        partial_file.unlink(missing_ok=True)
 
         logger.info(f"Job {job_id} completed in {processing_time:.2f} seconds")
 
diff --git a/gafuncs.py b/gafuncs.py
index 95472a3..c460d08 100644
--- a/gafuncs.py
+++ b/gafuncs.py
@@ -65,7 +65,10 @@ class CachedFunction:
         # Take the top 512 items from the sorted list
         top_128_items = sorted_items[:128]
 
-        with open(f"{CACHE_DIR!s}/{self.spectra_hash}.json", "w") as f:
+        # Written to a separate path: the presence of "<hash>.json" is what /submit
+        # uses to decide a run is finished, so in-progress snapshots must not create it.
+        partial_file = f"{CACHE_DIR!s}/{self.spectra_hash}.partial.json"
+        with open(partial_file, "w") as f:
             json.dump(top_128_items, f)
 
         return [self.cache[x] for x in inputs]

---