Idempotent job cache with diskcache FanoutCache guarded by filelock: working versions (tested 2026-10-15)

diskcache==5.6.3
filelock==4.2.0
pytest==9.1.1

import hashlib
import json
from pathlib import Path
from typing import Any, Callable

from diskcache import FanoutCache
from filelock import FileLock

MISSING = object()


def job_key(name: str, *args: Any) -> str:
    payload = json.dumps([name, args], sort_keys=True, default=repr)
    digest = hashlib.sha256(payload.encode()).hexdigest()[:16]
    return f"{name}-{digest}"


class JobCache:
    def __init__(self, directory: str, shards: int = 4, timeout: float = 1.0, lock_timeout: float = 5.0):
        self.directory = Path(directory)
        self.directory.mkdir(parents=True, exist_ok=True)
        (self.directory / "locks").mkdir(parents=True, exist_ok=True)
        self.cache = FanoutCache(str(self.directory / "cache"), shards=shards, timeout=timeout)
        self.lock_timeout = lock_timeout
        self._locks: dict[str, FileLock] = {}

    def _lock(self, key: str) -> FileLock:
        lock = self._locks.get(key)
        if lock is None:
            lock = self._locks[key] = FileLock(str(self.directory / "locks" / f"{key}.lock"), timeout=self.lock_timeout)
        return lock

    def run_once(self, name: str, func: Callable[..., Any], *args: Any, expire: float | None = None) -> Any:
        key = job_key(name, *args)
        hit = self.cache.get(key, default=MISSING)
        if hit is not MISSING:
            return hit
        with self._lock(key):
            hit = self.cache.get(key, default=MISSING)
            if hit is not MISSING:
                return hit
            value = func(*args)
            self.cache.set(key, value, expire=expire)
            return value

    def claim(self, name: str, *args: Any) -> bool:
        return self.cache.add(job_key(name, *args), True)

    def close(self) -> None:
        self.cache.close()

import threading

import diskcache
import pytest
from filelock import Timeout as FileLockTimeout

from jobs import JobCache, job_key


@pytest.fixture
def jobs(tmp_path):
    cache = JobCache(tmp_path)
    yield cache
    cache.close()


def test_run_once_computes_once_and_reuses_result(jobs):
    calls = []

    def work(x):
        calls.append(x)
        return x * 2

    assert jobs.run_once("double", work, 21) == 42
    assert jobs.run_once("double", work, 21) == 42
    assert calls == [21]


def test_cached_none_is_a_hit_not_a_miss(jobs):
    calls = []

    def returns_none():
        calls.append(1)

    assert jobs.run_once("noop", returns_none) is None
    assert jobs.run_once("noop", returns_none) is None
    assert len(calls) == 1


def test_zero_expire_means_expire_now_not_forever(jobs):
    calls = []

    def work():
        calls.append(1)
        return "v"

    assert jobs.run_once("ttl", work, expire=0) == "v"
    assert jobs.run_once("ttl", work, expire=0) == "v"
    assert len(calls) == 2


def test_concurrent_callers_compute_once(jobs):
    barrier = threading.Barrier(2)
    calls = []
    results = {}

    def slow(x):
        calls.append(x)
        return x

    def run(tag):
        barrier.wait(timeout=5)
        results[tag] = jobs.run_once("slow", slow, "k")

    threads = [threading.Thread(target=run, args=(i,)) for i in range(2)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    assert calls == ["k"]
    assert results == {0: "k", 1: "k"}


def test_lock_contention_raises_filelock_timeout(tmp_path):
    jobs = JobCache(tmp_path, lock_timeout=0.1)
    inside = threading.Event()
    release = threading.Event()

    def slow():
        inside.set()
        release.wait(5)
        return "done"

    holder = threading.Thread(target=lambda: jobs.run_once("slow", slow))
    holder.start()
    assert inside.wait(5)

    with pytest.raises(FileLockTimeout):
        jobs.run_once("slow", lambda: "other")
    assert not issubclass(FileLockTimeout, diskcache.Timeout)

    release.set()
    holder.join()
    assert jobs.cache.get(job_key("slow")) == "done"
    jobs.close()


def test_claim_is_atomic_and_keys_are_stable(jobs):
    assert jobs.claim("build", {"a": 1, "b": 2}) is True
    assert jobs.claim("build", {"b": 2, "a": 1}) is False
    assert job_key("f", {"a": 1, "b": 2}) == job_key("f", {"b": 2, "a": 1})
    assert job_key("f", 1) != job_key("f", 2)

[pytest]
pythonpath = .