python-async-job-disk-memory-ordering

Answer

The bug was an ordering inversion in gitreins_mcp/server.py: the worker thread set the in-memory job record to complete before save_job()'s fsync landed, so a fresh server instance (or restarted process) that only reads the disk store could observe a stale running after the owning server had already reported complete. The fix centralizes terminal-state recording in a _finish() helper that persists to disk first, then flips memory — establishing the invariant "memory says complete ⟹ disk says complete."

1. Durable save_job() (fsync file + dir, atomic replace)

# gitreins_mcp/server.py
def save_job(self, job: JobRecord) -> None:
    """Durably persist one job record.

    Temp-file write + fsync(file) + fsync(dir) + atomic os.replace().
    Returns only once the record is durable, so a fresh process reading
    the store after this call is guaranteed to see it.
    """
    path = os.path.join(self.job_store_dir, f"{job.job_id}.json")
    tmp = os.path.join(self.job_store_dir, f".{job.job_id}.{os.getpid()}.tmp")
    try:
        with open(tmp, "w", encoding="utf-8") as fh:
            json.dump(asdict(job), fh, sort_keys=True)
            fh.flush()
            os.fsync(fh.fileno())
        dfd = os.open(self.job_store_dir, os.O_RDONLY)
        try:
            os.fsync(dfd)               # make the rename itself durable
        finally:
            os.close(dfd)
        os.replace(tmp, path)           # atomic: readers never see partial JSON
    finally:
        try:
            os.unlink(tmp)
        except FileNotFoundError:
            pass

2. The _finish() helper — disk first, memory second

JOB_TERMINAL = {"complete", "failed"}

def _finish(self, job_id: str, *, status: str,
            result: Any = None, error: str | None = None) -> None:
    """Record a job's terminal state (DF-006 fix).

    Ordering contract: persist to disk FIRST, then update the in-memory
    record.  Once this returns, any fresh server instance (or restarted
    process) reading the job store sees the terminal status — it can
    never observe a stale 'running'.

    Invariant: memory says complete  ==>  disk says complete.
    """
    assert status in JOB_TERMINAL, f"status {status!r} is not terminal"
    with self._jobs_lock:
        job = self._jobs[job_id]
        if job.status in JOB_TERMINAL:
            return  # idempotent — never clobber an already-terminal record
        terminal = replace(
            job,
            status=status,
            result=result,
            error=error,
            finished_at=datetime.now(timezone.utc).isoformat(),
        )
        self.save_job(terminal)         # 1. disk first  (durable, fsync'd)
        self._jobs[job_id] = terminal   # 2. memory second

3. _start_job_thread() — both terminal paths go through _finish()

def _start_job_thread(self, job_id: str) -> None:
    """Run a judge job on a worker thread (async API)."""
    def run() -> None:
        try:
            result = self._judge(job_id)          # blocking judge logic
        except Exception as exc:                   # noqa: BLE001
            self._finish(job_id, status="failed",
                         error=f"{type(exc).__name__}: {exc}")
        else:
            self._finish(job_id, status="complete", result=result)

    threading.Thread(target=run, name=f"judge-{job_id}", daemon=True).start()

Why this closes the race: in the old code the thread did self._jobs[job_id].status = "complete" and only then called save_job() — a reader on another process could sample the store between those two steps. Now the terminal record (status/result/error/finished_at) is durably on disk before memory ever advertises completion, so a fresh instance can never see running after the owner reported complete. The lock also makes _finish idempotent, so a double-finish (e.g., timeout handler racing the worker) cannot clobber a terminal record.