diff --git a/CHANGELOG.md b/CHANGELOG.md
index d20d4da..ea11a8d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,62 @@
# Changelog
+## 0.4.5 - 21.6.2026.
+
+### Fixed
+- Hash objects by their `__slots__`/`__dict__` state instead of the memory
+ address. Slotted value types (including `@dataclass(slots=True)`) cached by
+ identity before, so identical calls missed the cache every run and lossy
+ custom reprs could collide. Truly opaque objects now warn instead of silently
+ address-hashing.
+- Record `ResultRef` values nested inside custom-object attributes in commit
+ lineage, matching what the hasher already accounts for, so garbage collection
+ cannot evict a blob a commit still depends on.
+- Hash referenced module-global `dict`/`list`/`set` values consistently with
+ scalars, so config-style globals invalidate the cache when they change.
+- Canonicalize function source with `ast.unparse` instead of `ast.dump`, which
+ keeps the same comment/whitespace/docstring insensitivity but stays stable
+ across Python versions whose `ast.dump` field set differs.
+- Treat the `last_accessed_at` bump on a cache hit as best-effort so a locked
+ database no longer turns a successful lookup into an error.
+- Skip a post-eviction `VACUUM` that is blocked by a concurrent writer instead
+ of failing an eviction whose deletes already committed.
+- Prevent Redis tag index key collisions: presence and value indexes now use
+ distinct prefixes and the value key length-prefixes the tag key, so a `:` in
+ a key or value can no longer cross-match unrelated commits.
+- Make the Redis blob delete script idempotent against a missing reference
+ counter so it cannot drop a blob another commit still references.
+- Enforce the HTTP server request size limit on the bytes actually received,
+ closing a bypass for chunked requests that omit `Content-Length`. On
+ token-protected servers, unauthenticated requests are now rejected before any
+ body is buffered.
+- Validate handler input (returning 400) and wrap every server handler in a
+ generic 500 barrier so malformed input and internal errors no longer leak
+ stack traces; `/gc` also accepts an empty body and falls back to defaults.
+- Report commits skipped during a lossy import. A truncated archive no longer
+ looks like a clean import.
+- Exit non-zero from `cashet show` and `cashet get` when a commit is missing,
+ matching `cashet rm`.
+- Cancel in-flight sibling tasks when a parallel `submit_many` task fails,
+ instead of leaving them running after the error is surfaced.
+
+### Performance
+- Redis `find_running_by_fingerprint` uses a per-fingerprint running-claim index
+ for O(1) lookup instead of scanning a fingerprint's full commit history on
+ every submit.
+
+### Changed
+- `import_archive` (sync and async) now returns an `ImportResult(imported,
+ skipped)` named tuple instead of a bare imported count.
+- Sync `submit_many` no longer recomputes the dependency graph that the async
+ client already builds.
+
+### Notes
+- The hashing fixes change function and argument cache keys, so results cached
+ by earlier versions recompute on first access. Blobs are content-addressed and
+ remain until garbage collected.
+- The Redis tag index key scheme changed. Tag indexes written by older versions
+ are not migrated; rewrite affected commits to rebuild them.
+
## 0.4.4 — 11.5.2026.
### Fixed
diff --git a/README.md b/README.md
index d18162d..f89050d 100644
--- a/README.md
+++ b/README.md
@@ -1,18 +1,22 @@
cashet
- A Python memoization cache with Redis, async support, and an HTTP server.
- Hash functions + args into cache keys. Results stored as immutable blobs. Chain outputs with DAG resolution.
- Run a function once — get the same result instantly every time after.
+ Content-addressable compute cache for Python: persistent function memoization with Redis, async, DAG pipelines, and an HTTP server.
+ Hash a function plus its arguments into a cache key, store the result as an immutable blob, and return it instantly on every later call.
+ Think git for your function results.
+
+
+
+ Keywords: Python caching, memoization, disk cache, Redis cache, async asyncio, DAG pipeline, content-addressable, reproducible computation, joblib / lru_cache alternative.
Install ·
- Quick Start ·
- Why ·
+ Quick Start ·
+ Why cashet ·
Use Cases ·
CLI ·
- API ·
+ Python API ·
How It Works
@@ -24,9 +28,15 @@
---
+## What is cashet
+
+cashet is a caching library for expensive Python functions. You submit a function and its arguments, cashet runs it once, stores the result as a content-addressed blob, and serves that result instantly on every later call with the same code and inputs. Caches persist across process restarts, can be shared across machines through Redis, and can be inspected, diffed, and chained from the CLI or the Python API.
+
+The cache key is a SHA-256 hash of the function's AST-normalized source plus its arguments. Comments, docstrings, and formatting do not invalidate the cache; only a real change to the code or the inputs does. Identical results are deduplicated to a single blob on disk, and every result is a git-like object you can inspect and chain into pipelines.
+
## Install
-**Global CLI tool** (recommended):
+Install as a global CLI tool:
```bash
uv tool install cashet
@@ -34,13 +44,7 @@ uv tool install cashet
pipx install cashet
```
-Then use the CLI anywhere:
-
-```bash
-cashet --help
-```
-
-**In a project** (library + CLI):
+Add to a project as a library and project-local CLI:
```bash
uv add cashet
@@ -48,31 +52,15 @@ uv add cashet
pip install cashet
```
-This installs `cashet` as both an importable Python library (`from cashet import Client`) and a project-local CLI (`uv run cashet`).
-
-**With Redis backend:**
-
-```bash
-uv add "cashet[redis]"
-# or
-pip install "cashet[redis]"
-```
-
-**With HTTP server:**
-
-```bash
-uv add "cashet[server]"
-# or
-pip install "cashet[server]"
-```
-
-**All extras:**
+Optional backends:
```bash
+uv add "cashet[redis]" # shared cache via Redis
+uv add "cashet[server]" # HTTP server
uv add "cashet[redis,server]"
```
-**Develop / contribute:**
+Develop and contribute:
```bash
git clone https://github.com/jolovicdev/cashet.git
@@ -81,37 +69,27 @@ uv sync --all-extras
uv run pytest
```
-## Version Compatibility
-
-**0.2.x → 0.3.x: Hash format changed.** `hash_args` was unified with `serialize_args` to use a single canonical representation via `_stable_repr_to`. Same function + same arguments produce different cache fingerprints across the major version line. Caches created with 0.2.x **will not hit** on 0.3.x.
-
-**0.3.0 → 0.3.1:** Redis blob data keys renamed from `cashet:blob:{hash}` to `cashet:blob:data:{hash}` to fix a stats double-counting bug. Redis blob object/byte totals are maintained in `cashet:stats:blob` after a one-time backfill for existing caches. Function hashing also now includes defaults / keyword defaults in more cases, and custom object argument hashing includes the defining module. Existing 0.3.0 caches may miss after upgrading. Clear old caches before upgrading if you use Redis or rely on long-lived cache reuse across versions.
-
-**0.3.x → 0.4.0:** Added per-entry TTL and tag-based invalidation. SQLite stores auto-migrate (new `ttl_seconds` and `expires_at` columns). Redis stores encode new fields in commit JSON. Old caches are fully readable after upgrading.
-
-**Upgrading from an incompatible version:** either clear the cache (`cashet clear`) and rebuild, or point the new version at a fresh store directory.
-
## Quick Start
```python
from cashet import Client
-client = Client() # creates .cashet/ in current directory
+client = Client() # creates .cashet/ in the current directory
def expensive_transform(data, scale=1.0):
# imagine this takes 10 minutes
return [x * scale for x in data]
-# First call: runs the function
+# First call runs the function
ref = client.submit(expensive_transform, [1, 2, 3], scale=2.0)
print(ref.load()) # [2.0, 4.0, 6.0]
-# Second call with same args: instant — returns cached result
+# Same args again: instant, returns the cached result with no re-computation
ref2 = client.submit(expensive_transform, [1, 2, 3], scale=2.0)
-print(ref2.load()) # [2.0, 4.0, 6.0] — no re-computation
+print(ref2.load()) # [2.0, 4.0, 6.0]
```
-You can also use `Client` as a context manager to ensure the store connection is closed cleanly:
+`Client` works as a context manager so the store connection closes cleanly:
```python
with Client() as client:
@@ -119,50 +97,37 @@ with Client() as client:
print(ref.load())
```
-Chain tasks into a pipeline where each step's output feeds into the next:
+### Pipelines
-```python
-from cashet import Client
+Pass a result reference straight into the next task. Each step's output feeds the next, and cashet records the lineage:
+```python
client = Client()
-def load_dataset(path):
- return list(range(100))
-
-def normalize(data):
- max_val = max(data)
- return [x / max_val for x in data]
-
-def train_model(data, lr=0.01):
- return {"loss": 0.05, "lr": lr, "samples": len(data)}
-
-# Step 1: load
raw = client.submit(load_dataset, "data/train.csv")
-
-# Step 2: normalize (receives raw output as input)
-normalized = client.submit(normalize, raw)
-
-# Step 3: train (receives normalized output)
+normalized = client.submit(normalize, raw) # receives raw's output
model = client.submit(train_model, normalized, lr=0.001)
-print(model.load()) # {'loss': 0.05, 'lr': 0.001, 'samples': 100}
+print(model.load())
```
-Re-run the script — everything returns instantly from cache. Change one argument and only that step (and downstream) re-runs.
+Re-run the script and everything returns instantly from cache. Change one argument and only that step and the steps downstream of it re-run.
-**Shared cache with Redis** — multiple processes or machines share one cache:
+### Shared cache with Redis
+
+Multiple processes or machines share one cache:
```python
from cashet import Client
from cashet.redis_store import RedisStore
-client = Client(store=RedisStore("redis://localhost:6379"))
-
-# Any process on any machine using the same Redis gets cached results
+client = Client(store=RedisStore("redis://localhost:6379/0"))
ref = client.submit(expensive_transform, [1, 2, 3], scale=2.0)
```
-**Async API** — drop-in `AsyncClient` for asyncio workflows:
+### Async
+
+`AsyncClient` is a drop-in for asyncio workflows and accepts both sync and async callables:
```python
import asyncio
@@ -170,31 +135,16 @@ from cashet.async_client import AsyncClient
async def main():
client = AsyncClient()
-
- def compute(x: int) -> int:
- return x * 2
-
- ref = await client.submit(compute, 21)
- result = await ref.load()
- print(result) # 42
+ ref = await client.submit(lambda x: x * 2, 21)
+ print(await ref.load()) # 42
+ await client.close()
asyncio.run(main())
```
-**HTTP server** — expose cache metadata and registered tasks over HTTP:
-
-```bash
-python -c "from cashet import Client; Client().serve(port=8000)"
-```
-
-```python
-# With auth token
-from cashet import Client
-client = Client()
-client.serve(port=8000, require_token="my-secret-token")
-```
+### HTTP server
-Register server-side tasks in the server process:
+Expose cache metadata and registered tasks over HTTP:
```python
from cashet import Client
@@ -207,26 +157,19 @@ def double(x):
client.serve(port=8000, tasks={"double": double})
```
-Submit JSON args from another process:
-
```python
import requests
-r = requests.post("http://localhost:8000/submit", json={
- "task": "double",
- "args": [5],
-})
+requests.post("http://localhost:8000/submit", json={"task": "double", "args": [5]})
```
-Submitting Python source, dill payloads, or serializer-encoded args is disabled by default.
-For trusted internal clients only, enable the legacy remote-code path with both
-`allow_remote_code=True` and a non-empty `require_token=...`.
+By default the server only runs tasks registered in the server process. Submitting Python source, dill payloads, or serializer-encoded args is disabled unless you explicitly opt in (see [HTTP server endpoints](#http-server-endpoints)).
-## Why
+## Why cashet
-You already have caches (`functools.lru_cache`, `joblib.Memory`). Here's what's different:
+You already have `functools.lru_cache` and `joblib.Memory`. cashet adds persistence, sharing, provenance, and pipelines on top of memoization:
-| | lru_cache | joblib.Memory | **cashet** |
+| | lru_cache | joblib.Memory | cashet |
|---|---|---|---|
| AST-normalized hashing | No | No | Yes |
| DAG / pipeline chaining | No | No | Yes |
@@ -241,359 +184,112 @@ You already have caches (`functools.lru_cache`, `joblib.Memory`). Here's what's
| HTTP server | No | No | Yes |
| Persists across restarts | No | Yes | Yes |
-The core idea: **hash the function's AST-normalized source + arguments = unique cache key**. Comments, docstrings, and formatting changes don't invalidate the cache — only semantic changes do. Same function + same args = same result, stored immutably on disk. The result is a git-like blob you can inspect, diff, and chain.
+The core idea: hash the function's AST-normalized source plus its arguments into a unique cache key. Same function and same args give the same result, stored immutably on disk, that you can inspect, diff, and chain.
## Use Cases
-### 1. ML Experiment Tracking Without the Bloat
-
-You run 200 hyperparameter sweeps overnight. Half crash. You fix a bug and re-run. Without cashet, you re-process the dataset 200 times. With cashet:
+**ML experiment tracking.** Run many hyperparameter sweeps that share one expensive preprocessing step. `preprocess` runs once and every training job reuses its cached output. Use `TaskRef(index)` to wire one task's output into another within a batch:
```python
-from cashet import Client, TaskError, TaskRef
+from cashet import Client, TaskRef
client = Client()
-def preprocess(dataset_path, image_size):
- # 45 minutes of image resizing
- ...
-
-def train(data, learning_rate, dropout):
- ...
-
-# Batch submit with topological ordering
-# TaskRef(0) refers to the first task's output
results = client.submit_many([
- (preprocess, ("s3://my-bucket/images", 224)),
+ (preprocess, ("s3://bucket/images", 224)),
(train, (TaskRef(0), 0.01, 0.2)),
- (train, (TaskRef(0), 0.01, 0.5)),
(train, (TaskRef(0), 0.001, 0.2)),
- (train, (TaskRef(0), 0.001, 0.5)),
(train, (TaskRef(0), 0.0001, 0.2)),
- (train, (TaskRef(0), 0.0001, 0.5)),
])
```
-`preprocess` runs **once** — all 6 training jobs reuse its cached output. Re-run the script tomorrow and even the training results come from cache (same function + same args = instant).
-
-### 2. Data Pipeline Debugging
-
-Your ETL pipeline fails at step 5. You fix a typo. Now you need to re-run steps 5-7 but steps 1-4 are unchanged and expensive:
-
-```python
-from cashet import Client
-
-client = Client()
-
-raw = client.submit(load_s3, "s3://logs/2024-05-01/")
-clean = client.submit(remove_pii, raw)
-enriched = client.submit(join_crm, clean, "select * from users")
-report = client.submit(generate_report, enriched)
-```
-
-Fix the `join_crm` function and re-run the script. Steps 1-2 return instantly from cache. Only step 3 onward re-executes. This works because cashet tracks which function produced which output — changing a function's source code changes its hash, invalidating downstream cache entries.
-
-### 3. Reproducible Notebook Results
-
-`cashet` is designed to work in Jupyter notebooks and IPython sessions. Share a result with a colleague and they can verify exactly how it was produced:
-
-```python
-# your notebook
-ref = client.submit(generate_forecast, date="2024-01-01", model="v3")
-print(f"Result hash: {ref.hash}")
-```
-
-```bash
-# their terminal — inspect provenance
-cashet show
-
-# Output:
-# Hash: a3b4c5d6...
-# Function: generate_forecast
-# Source: def generate_forecast(date, model): ...
-# Args: (('2024-01-01',), {'model': 'v3'})
-# Created: 2024-05-01T10:32:17
-
-# Retrieve the actual result
-cashet get -o forecast.csv
-```
+**Data pipeline debugging.** Your ETL job fails at step 5. Fix the function and re-run the script. Unchanged upstream steps return from cache; only the changed step and everything after it re-executes, because changing a function's source changes its hash and invalidates downstream entries.
-### 4. Incremental Computation
+**Reproducible notebooks.** Share a result hash and a colleague can inspect exactly how it was produced from their terminal with `cashet show ` and retrieve it with `cashet get -o out.bin`.
-Process a large dataset in chunks. Already-processed chunks return instantly:
+**Incremental computation.** Process a large dataset in chunks with `client.map`. Already-processed chunks return instantly; add a new chunk and only that one runs:
```python
-from cashet import Client
-
-client = Client()
-
-def process_chunk(chunk_id, source_file):
- # expensive per-chunk processing
- ...
-
-results = []
-for chunk_id in range(100):
- ref = client.submit(process_chunk, chunk_id, "huge_file.parquet")
- results.append(ref)
+refs = client.map(process_chunk, range(100), source_file="data.parquet")
+results = [r.load() for r in refs]
```
-First run processes all 100 chunks. Second run (even after restarting Python) returns all 100 results instantly. Add a new chunk? Only that one runs.
-
## CLI
```bash
-# Show commit history
-cashet log
-
-# Filter by function name
-cashet log --func "preprocess"
-
-# Filter by tag
-cashet log --tag env=prod --tag experiment=run-1
-
-# Show full commit details (source code, args, error)
-cashet show
-
-# Retrieve a result (pretty-prints strings/dicts/lists)
-cashet get
-
-# Write a result to file
-cashet get -o output.bin
-
-# Compare two commits
-cashet diff
-
-# Show lineage of a result (same function+args over time)
-cashet history
-
-# Delete a specific commit
-cashet rm
-
-# Evict old cache entries and orphaned blobs
-cashet gc --older-than 30
-
-# Evict oldest entries until under a size limit
-cashet gc --max-size 1GB
-
-# Clear everything (alias for gc --older-than 0)
-cashet clear
-
-# Export all commits and blobs to a tar.gz archive
-cashet export backup.tar.gz
-
-# Import commits and blobs from a tar.gz archive
-cashet import backup.tar.gz
-
-# Storage statistics (includes disk size)
-cashet stats
-
-# Start the HTTP server
+cashet log # commit history
+cashet log --func preprocess # filter by function
+cashet log --tag env=prod # filter by tag
+cashet show # full commit details (source, args, error)
+cashet get # retrieve a result (pretty-prints str/dict/list)
+cashet get -o output.bin # write a result to a file
+cashet diff # compare two commits
+cashet history # lineage of one function+args over time
+cashet rm # delete a commit
+cashet invalidate -t env=prod # delete commits by tag
+cashet gc --older-than 30 # evict entries older than 30 days
+cashet gc --max-size 1GB # evict oldest entries until under a size limit
+cashet clear # remove everything
+cashet export backup.tar.gz # export commits and blobs to an archive
+cashet import backup.tar.gz # import from an archive
+cashet stats # storage statistics
cashet serve --host 127.0.0.1 --port 8000
-
-# Legacy unsafe remote-code mode requires auth
-cashet serve --require-token secret123 --allow-remote-code
-```
-
-## API
-
-### `Client`
-
-```python
-from cashet import Client
-
-client = Client(
- store_dir=".cashet", # where to store blobs + metadata (SQLiteStore)
- # falls back to $CASHET_DIR env var if set
- store=None, # or inject any Store implementation (SQLiteStore, RedisStore)
- executor=None, # or inject any Executor implementation
- serializer=None, # defaults to PickleSerializer
- max_workers=1, # max parallelism for submit_many (default: 1, sequential)
-)
```
-### `AsyncClient`
-
-```python
-import asyncio
-from cashet.async_client import AsyncClient
-
-async def main():
- client = AsyncClient(
- store_dir=".cashet", # defaults to AsyncSQLiteStore
- store=None, # or AsyncRedisStore, or any AsyncStore
- executor=None, # defaults to AsyncLocalExecutor
- serializer=None, # defaults to PickleSerializer
- max_workers=1, # max parallelism for submit_many (default: 1, sequential)
- )
-
- async def square(x: int) -> int:
- return x * x
-
- ref = await client.submit(square, 5)
- result = await ref.load() # 25
- await client.close()
-
-asyncio.run(main())
-```
-
-`AsyncClient` mirrors `Client` — `submit()`, `submit_many()`, `log()`, `show()`, `get()`, `stats()`, `gc()`, `rm()`, `clear()`, `serve()` are all `async def`. `submit()` accepts both sync and async callables, and returns `AsyncResultRef` with `async load()`. Chain tasks by passing `AsyncResultRef` as an argument.
-
-### HTTP Server
-
-```python
-# Start the server
-from cashet import Client
-client = Client()
-client.serve(host="127.0.0.1", port=8000)
-
-# With Bearer token authentication
-client.serve(host="0.0.0.0", port=8000, require_token="secret123")
-```
-
-Execute tasks over HTTP by registering callables in the server process and sending
-JSON arguments from another process:
-
-```python
-from cashet import Client
-
-client = Client()
-
-def add(x: int, y: int) -> int:
- return x + y
-
-client.serve(port=8000, tasks={"add": add})
-```
-
-```python
-import requests
-
-r = requests.post("http://localhost:8000/submit", json={"task": "add", "args": [3, 4]})
-```
-
-Endpoints:
-
-| Method | Path | Description |
-|---|---|---|
-| POST | `/submit` | Submit a registered task for execution |
-| GET | `/result/{hash}` | Fetch deserialized result |
-| GET | `/commit/{hash}` | Commit metadata |
-| GET | `/log` | List commits (query: `?func=`, `?limit=`, `?status=`) |
-| GET | `/stats` | Storage statistics |
-| POST | `/gc` | Run garbage collection |
-
-Use `AsyncClient.serve()` for the async variant with `create_async_app()`.
+`cashet show`, `cashet get`, and `cashet rm` exit with a non-zero status when the commit is not found, so they compose in scripts.
-> **Security:** `/submit` does not execute client-supplied Python by default. The legacy
-> `func_source`, `func_b64`, `args_b64`, and `kwargs_b64` payloads require
-> `allow_remote_code=True` and a non-empty Bearer token. In the CLI, this is
-> `cashet serve --require-token secret123 --allow-remote-code`. That mode deserializes
-> and executes arbitrary Python, so expose it only to trusted clients.
+## Python API
-### Redis Backend
+### Client and AsyncClient
```python
from cashet import Client
-from cashet.redis_store import RedisStore
-
-# Sync
-client = Client(store=RedisStore("redis://localhost:6379/0"))
-
-# Async
-from cashet.async_client import AsyncClient
-from cashet.redis_store import AsyncRedisStore
-
-client = AsyncClient(store=AsyncRedisStore("redis://localhost:6379/0"))
-```
-
-Redis-backed stores support all operations — cache dedup, fingerprint lookup, history, eviction, and size-based GC. Cross-process claim dedup uses per-fingerprint Redis locks (`cashet:lock:{fingerprint}`) to prevent redundant execution across machines. Last-access eviction is indexed in Redis (`cashet:index:last_accessed`), blob stats are maintained in `cashet:stats:blob`, and blob ref counts (`cashet:blob:ref:{hash}`) enable orphan cleanup without scanning all commits.
-### Pluggable Backends
-
-Everything is protocol-based. Swap the store, executor, or serializer without touching your task code:
-
-```python
-from pathlib import Path
-
-from cashet import Client, Store, Executor, Serializer
-from cashet.store import SQLiteStore
-from cashet.executor import LocalExecutor
-
-# These are equivalent (the defaults):
-client = Client(store_dir=".cashet")
-
-# Explicit injection:
client = Client(
- store=SQLiteStore(Path(".cashet")),
- executor=LocalExecutor(),
+ store_dir=".cashet", # blob + metadata directory; falls back to $CASHET_DIR
+ store=None, # or any Store (SQLiteStore, RedisStore, ...)
+ executor=None, # or any Executor
+ serializer=None, # defaults to PickleSerializer
+ max_workers=1, # parallelism for submit_many (1 = sequential)
)
```
-**Store protocol** — implement this to use RocksDB, Redis, S3, or anything else:
+`AsyncClient` (from `cashet.async_client`) mirrors `Client`: `submit`, `submit_many`, `map`, `log`, `show`, `get`, `diff`, `history`, `stats`, `gc`, `rm`, `clear`, `invalidate`, and `serve` are all `async def`. It returns `AsyncResultRef` with an `async load()`.
-```python
-from cashet.protocols import Store
-
-class RedisStore:
- def put_blob(self, data: bytes) -> ObjectRef: ...
- def get_blob(self, ref: ObjectRef) -> bytes: ...
- def put_commit(self, commit: Commit) -> None: ...
- def get_commit(self, hash: str) -> Commit | None: ...
- def find_by_fingerprint(self, fingerprint: str) -> Commit | None: ...
- def find_running_by_fingerprint(self, fingerprint: str) -> Commit | None: ...
- def list_commits(self, ...) -> list[Commit]: ...
- def get_history(self, hash: str) -> list[Commit]: ...
- def stats(self) -> dict[str, int]: ...
- def evict(self, older_than: datetime, max_size_bytes: int | None = None) -> int: ...
- def delete_commit(self, hash: str) -> bool: ...
- def close(self) -> None: ...
-
-client = Client(store=RedisStore("redis://localhost"))
-# Everything else works identically
-```
-
-**Executor protocol** — implement this for distributed execution (Celery, Kafka, RQ):
+### submit and ResultRef
```python
-from cashet.protocols import Executor
-
-class CeleryExecutor:
- def submit(self, func, args, kwargs, task_def, store, serializer):
- # Push to Celery, poll for result
- ...
-
-client = Client(
- store=RedisStore("redis://localhost"),
- executor=CeleryExecutor(),
-)
+ref = client.submit(my_func, arg1, arg2, key="value")
+ref.hash # content hash of the result blob
+ref.commit_hash # commit hash (use for show / history / rm / get)
+ref.size # size in bytes
+ref.load() # deserialize and return the result
```
-**Serializer protocol** — already covered below.
+If the same function and arguments were submitted before, `submit` returns the cached result without re-executing. `ResultRef` is generic, so `submit` infers the return type from the function's annotation. Pass a `ResultRef` as an argument (including nested inside lists, tuples, sets, frozensets, and dicts) to chain tasks; it resolves to its output before execution.
-### `client.submit(func, *args, **kwargs) -> ResultRef`
+### Per-task options
-Submit a function for execution. Returns a `ResultRef` — a lazy handle to the result.
+Each option is available per call (prefixed with `_`) and on the `@client.task` decorator:
```python
-ref = client.submit(my_func, arg1, arg2, key="value")
-ref.hash # content hash of the result blob
-ref.commit_hash # commit hash (use this for show/history/rm/get)
-ref.size # size in bytes
-ref.load() # deserialize and return the result
-```
+ref = client.submit(fetch_api, url, _retries=3, _timeout=30, _ttl=3600, _tags={"env": "prod"})
-If the same function + same arguments have been submitted before, returns the cached result **without re-executing**.
-
-### `client.clear()`
-
-Remove all cache entries and orphaned blobs. Equivalent to `client.gc(timedelta(days=0))`.
+@client.task(cache=False, retries=3, timeout=30, ttl=3600, tags={"team": "ml"})
+def fetch_api(url):
+ ...
-```python
-client.clear()
+ref = my_task(url) # decorated tasks are directly callable and return a ResultRef
```
-### `client.submit_many(tasks) -> list[ResultRef]`
+- `cache=False` runs the function every time but still records lineage (non-deterministic work).
+- `force=True` skips the cache and re-executes once.
+- `retries=N` retries failed attempts, then raises `TaskError` with the original traceback.
+- `timeout=seconds` bounds an attempt. Local timeouts are soft: cashet stops waiting and records a failure, but Python cannot kill running thread code, so keep task functions idempotent.
+- `ttl=seconds` expires a commit. Expired commits are skipped on lookup and removed at garbage collection.
+- `tags={...}` attach metadata for filtering. Tags are not part of the cache key.
-Submit a batch of tasks with automatic topological ordering. Use `TaskRef(index)` to wire outputs between tasks in the batch.
+### submit_many and map
```python
from cashet import TaskRef
@@ -601,367 +297,181 @@ from cashet import TaskRef
refs = client.submit_many([
step1_func,
(step2_func, (TaskRef(0),)),
- (step3_func, (TaskRef(1), "extra_arg")),
-], max_workers=4) # run independent tasks in parallel
-```
-
-This enables parallel fan-out and ensures each task only runs after its dependencies.
-
-> **Note:** With the default `SQLiteStore`, parallel execution serializes on the SQLite write lock — `max_workers > 1` only benefits compute-heavy tasks where execution time dominates. For true parallel fan-out, use `RedisStore`.
+ (step3_func, (TaskRef(1), "extra")),
+], max_workers=4)
-### `client.map(func, items, *args, **kwargs) -> list[ResultRef]`
-
-Map a function over an iterable with per-item caching. Each item becomes the first positional argument; additional `*args` and `**kwargs` are appended.
-
-```python
refs = client.map(process_chunk, range(100), source_file="data.parquet")
-# refs[0].load() == process_chunk(0, source_file="data.parquet")
-
-results = [r.load() for r in refs]
-```
-
-Already-computed items return instantly from cache. Add a new chunk later and only that item re-runs.
-
-**Opt out of caching:**
-
-```python
-# Per-call
-ref = client.submit(non_deterministic_func, _cache=False)
-
-# Per-function via decorator
-@client.task(cache=False)
-def random_score():
- return random.random()
-```
-
-**Force re-execution (skip cache, always run):**
-
-```python
-# Per-call
-ref = client.submit(my_func, arg, _force=True)
-
-# Per-function via decorator
-@client.task(force=True)
-def always_rerun():
- ...
-```
-
-**Tag commits:**
-
-```python
-# Per-call
-ref = client.submit(train, data, lr=0.01, _tags={"experiment": "v1"})
-
-# Per-function via decorator
-@client.task(tags={"team": "ml"})
-def preprocess(raw):
- ...
-```
-
-Tags are not part of the cache key — they are metadata for organization and filtering.
-
-**Retry flaky operations:**
-
-```python
-# Per-call
-ref = client.submit(fetch_api, url, _retries=3)
-
-# Per-function via decorator
-@client.task(retries=3)
-def fetch_api(url):
- ...
-```
-
-Retries wait briefly between attempts. When retries are exhausted, `client.submit` raises `TaskError` with the original traceback included in the message.
-
-**Task timeouts:**
-
-```python
-# Per-call (seconds)
-ref = client.submit(slow_func, _timeout=30)
-
-# Per-function via decorator
-@client.task(timeout=30)
-def slow_func():
- ...
-```
-
-Timeouts can be combined with retries — a timed-out attempt counts as a failure and will be retried. Local execution uses Python threads, so timeouts are soft: cashet stops waiting and records the attempt as failed, but Python cannot safely kill already-running thread code. Use idempotent task functions; hard cancellation belongs in a process/distributed executor such as a future Celery executor.
-
-**Per-entry TTL:**
-
-```python
-# Per-call (seconds)
-ref = client.submit(fetch_api_data, url, _ttl=3600)
-
-# Per-function via decorator
-@client.task(ttl=3600)
-def fetch_api_data(url):
- ...
```
-Commits expire automatically after the TTL. Expired commits are skipped by cache lookups and history queries but are not physically deleted until garbage collection runs.
-
-### `@client.task`
+`submit_many` orders tasks topologically and runs independent ones in parallel up to `max_workers`. With the default `SQLiteStore`, parallel execution serializes on the SQLite write lock, so `max_workers > 1` mainly helps compute-heavy tasks; use `RedisStore` for true fan-out across processes.
-Register a function with cashet metadata and make it directly callable:
+### Inspecting, diffing, and eviction
```python
-@client.task
-def my_func(x):
- return x * 2
-
-ref = my_func(5) # Returns ResultRef, same as client.submit(my_func, 5)
-ref.load() # 10
-
-@client.task(cache=False, name="custom_task_name", tags={"env": "prod"})
-def other_func(x):
- return x + 1
-```
-
-`client.submit(my_func, 5)` still works identically.
-
-### `client.log()`, `client.show()`, `client.get()`, `client.diff()`, `client.history()`, `client.rm()`, `client.gc()`
-
-```python
-# List commits
-commits = client.log(func_name="preprocess", limit=10)
-
-# Filter by status
-commits = client.log(status="failed")
+commits = client.log(func_name="preprocess", limit=10, status="failed", tags={"experiment": "v1"})
-# Filter by tags
-commits = client.log(tags={"experiment": "v1"})
-
-# Get commit details
commit = client.show(hash)
-commit.task_def.func_source # the source code
-commit.task_def.args_snapshot # the serialized args
-commit.parent_hash # previous commit for same func+args
-commit.created_at
+commit.task_def.func_source # the source code
+commit.task_def.args_snapshot # the serialized args
+commit.parent_hash # previous commit for the same func + args
-# Load a result by commit hash
result = client.get(hash)
+diff = client.diff(hash_a, hash_b) # {'func_changed': ..., 'args_changed': ..., 'output_changed': ...}
+history = client.history(hash) # all runs of the same func + args
-# Diff two commits
-diff = client.diff(hash_a, hash_b)
-# {'func_changed': True, 'args_changed': False, 'output_changed': True, ...}
-
-# Get lineage (all runs of same func+args)
-history = client.history(hash)
-
-# Evict old entries (default: 30 days)
-evicted = client.gc()
-# Evict entries older than 7 days
from datetime import timedelta
-evicted = client.gc(older_than=timedelta(days=7))
-# Evict oldest entries until under size limit
-evicted = client.gc(max_size_bytes=1024 * 1024 * 1024) # 1GB
-
-# Invalidate commits by tags
-deleted = client.invalidate(tags={"experiment": "v1"})
-# deleted = client.invalidate(tags={"experiment": None}) # any commit with 'experiment' tag
+client.gc(older_than=timedelta(days=7))
+client.gc(max_size_bytes=1024 ** 3) # evict until under 1 GB
+client.invalidate(tags={"experiment": "v1"})
+client.clear()
-# Storage stats
stats = client.stats()
-# {
-# 'total_commits': 42,
-# 'completed_commits': 40,
-# 'stored_objects': 38, # blob_objects + inline_objects
-# 'disk_bytes': 10485760, # blob_bytes + inline_bytes
-# 'blob_objects': 35,
-# 'blob_bytes': 9437184,
-# 'inline_objects': 3,
-# 'inline_bytes': 1048576,
-# }
+# {'total_commits', 'completed_commits', 'stored_objects', 'disk_bytes',
+# 'blob_objects', 'blob_bytes', 'inline_objects', 'inline_bytes'}
```
-### `client.export(path)` / `client.import_archive(path)`
+### Export and import
-Export the entire cache to a portable `.tar.gz` archive and import it elsewhere. Blobs are content-addressable, so deduplication is preserved across stores.
+Export the whole cache to a portable archive and import it elsewhere. Blobs are content-addressed, so deduplication is preserved across stores. Use this for migrations (SQLite to Redis), CI cache warm-up, or backups.
```python
-# Export
client.export("backup.tar.gz")
-# Import into a fresh store
client2 = Client(store_dir=".cashet2")
-count = client2.import_archive("backup.tar.gz")
-print(f"Imported {count} commits")
+result = client2.import_archive("backup.tar.gz")
+print(f"imported {result.imported}, skipped {result.skipped}")
```
-Use this for migrations (SQLite → Redis), CI cache warm-up, or backups. Existing commits are skipped during import.
+`import_archive` verifies every blob's content hash and returns an `ImportResult(imported, skipped)`. Commits whose blobs are missing from the archive are skipped and reported rather than silently dropped; existing commits are skipped without re-import.
-### Jupyter & Notebook Support
-
-`cashet` works seamlessly in Jupyter notebooks, IPython, and the Python REPL. It uses a tiered source-resolution strategy:
-
-1. **`inspect.getsource()`** — for normal `.py` files
-2. **`dill.source.getsource()`** — for interactive sessions with live history
-3. **`dis.Bytecode` fallback** — for any live function, even after a kernel restart
-
-This means you can define functions in a notebook cell, rerun the cell with changes, and `cashet` will correctly invalidate the cache based on the new code.
+### HTTP server endpoints
```python
-# In a notebook cell
-client = Client()
-
-def preprocess(data):
- return [x * 2 for x in data]
+from cashet import Client
-ref = client.submit(preprocess, [1, 2, 3])
+client = Client()
+client.serve(host="127.0.0.1", port=8000, require_token="secret123")
```
-Change the cell body and rerun — the cache invalidates automatically.
+| Method | Path | Description |
+|---|---|---|
+| POST | `/submit` | Run a registered task |
+| GET | `/result/{hash}` | Fetch the deserialized result |
+| GET | `/commit/{hash}` | Commit metadata |
+| GET | `/log` | List commits (`?func=`, `?limit=`, `?status=`) |
+| GET | `/stats` | Storage statistics |
+| POST | `/gc` | Run garbage collection |
-### Thread Safety
+When `require_token` is set, every request needs an `Authorization: Bearer ` header. Request bodies are size-limited (default 500 MB). Use `AsyncClient.serve()` for the native async server.
-`cashet` is safe to use from multiple threads and processes sharing the same store directory. Concurrent submissions of the same uncached task are deduplicated: the function executes **exactly once** and all callers receive the same cached result. This works across `multiprocessing.Process`, `ProcessPoolExecutor`, and multiple independent Python interpreters.
+> **Security.** `/submit` does not execute client-supplied Python by default. The legacy `func_source`, `func_b64`, `args_b64`, and `kwargs_b64` payloads require both `allow_remote_code=True` and a non-empty token (`cashet serve --require-token secret123 --allow-remote-code`). That mode deserializes and runs arbitrary Python, so expose it only to trusted clients.
-> **Note:** Cross-process dedup uses a 5-minute timeout by default. If a process dies while running a task, its claim is automatically reclaimed after that timeout so other workers are not blocked forever. You can adjust this via `LocalExecutor(running_ttl=...)`:
->
-> ```python
-> from datetime import timedelta
-> from cashet.executor import LocalExecutor
->
-> client = Client(executor=LocalExecutor(running_ttl=timedelta(minutes=10)))
-> ```
+### Serialization
```python
-import threading
-
-def worker():
- c = Client() # separate Client instance, same store
- c.submit(expensive_func, arg)
-
-threads = [threading.Thread(target=worker) for _ in range(10)]
-for t in threads:
- t.start()
-for t in threads:
- t.join()
-# expensive_func ran only once
-```
-
-### `ResultRef`
-
-A lazy reference to a stored result. Pass it as an argument to chain tasks:
+from cashet import Client, PickleSerializer, SafePickleSerializer, JsonSerializer
-```python
-step1 = client.submit(func_a, input_data)
-step2 = client.submit(func_b, step1) # step1 auto-resolves to its output
-step3 = client.submit(func_c, {"payload": step2}) # nested refs resolve too
+Client(serializer=PickleSerializer()) # default, arbitrary Python objects
+Client(serializer=SafePickleSerializer()) # restrict unpickling to an allowlist
+Client(serializer=SafePickleSerializer(extra_classes=[MyClass]))
+Client(serializer=JsonSerializer()) # JSON-safe data
```
-`ResultRef` is generic — `submit()` infers the return type from the function annotation:
+Implement the `Serializer` protocol (`dumps`/`loads`) for MessagePack or any custom format.
-```python
-ref: ResultRef[int] = client.submit(double, 5)
-result = ref.load() # typed as int
-```
+### Pluggable backends
-### Custom Serialization
+Everything is protocol-based, so you can swap the store, executor, or serializer without changing task code.
```python
-from cashet import Client, PickleSerializer, SafePickleSerializer, JsonSerializer
-
-# Default: pickle (handles arbitrary Python objects)
-client = Client(serializer=PickleSerializer())
-
-# Safe pickle: restricts deserialization to an allowlist of known types
-client = Client(serializer=SafePickleSerializer())
-
-# Allow custom classes through the allowlist
-client = Client(serializer=SafePickleSerializer(extra_classes=[MyClass]))
-
-# For JSON-safe data (dicts, lists, primitives)
-client = Client(serializer=JsonSerializer())
-
-# Or implement the Serializer protocol
-from cashet.hashing import Serializer
+from cashet import Client
+from cashet.store import SQLiteStore
+from cashet.executor import LocalExecutor
-class MySerializer:
- def dumps(self, obj) -> bytes:
- ...
- def loads(self, data: bytes):
- ...
+client = Client(store=SQLiteStore(".cashet"), executor=LocalExecutor())
```
+| Protocol | Default | Built-in alternatives | Implement for |
+|---|---|---|---|
+| `Store` | `SQLiteStore` | `RedisStore` | RocksDB, S3, Postgres |
+| `AsyncStore` | `AsyncSQLiteStore` | `AsyncRedisStore` | async variants |
+| `Executor` | `LocalExecutor` | | Celery, Kafka, RQ |
+| `AsyncExecutor` | `AsyncLocalExecutor` | | Celery, Kafka, RQ |
+| `Serializer` | `PickleSerializer` | `JsonSerializer`, `SafePickleSerializer` | MessagePack, custom |
+
## How It Works
```
client.submit(func, arg1, arg2)
│
▼
- ┌─────────────────┐
- │ Hash function │ SHA256(AST-normalized source + dep versions + referenced user helpers)
+ ┌──────────────────┐
+ │ Hash function │ SHA256(AST-normalized source + dep versions + referenced helpers)
│ Hash arguments │ SHA256(canonical repr of args/kwargs)
- └────────┬────────┘
+ └────────┬─────────┘
│
▼
- ┌─────────────────┐
+ ┌──────────────────┐
│ Fingerprint │ func_hash:args_hash
│ cache lookup │ ← Store protocol (SQLiteStore, RedisStore, ...)
- └────────┬────────┘
+ └────────┬─────────┘
│
┌─────┴─────┐
- │ │
- CACHED MISS
+ CACHED MISS
│ │
▼ ▼
- Return ref ← Executor protocol (LocalExecutor, CeleryExecutor, ...)
- Execute function
- Store result as blob → Store protocol
- Record commit with parent lineage
- Return ref
+ Return ref Execute (Executor protocol), store result as a blob,
+ record a commit with parent lineage, return ref
```
-**Architecture (protocol-based):**
-
-| Protocol | Default | Built-in alternatives | Implement for |
-|---|---|---|---|
-| `Store` | `SQLiteStore` | `RedisStore` | RocksDB, S3, Postgres |
-| `AsyncStore` | `AsyncSQLiteStore` | `AsyncRedisStore` | async variants of above |
-| `Executor` | `LocalExecutor` | — | Celery, Kafka, RQ |
-| `AsyncExecutor` | `AsyncLocalExecutor` | — | Celery, Kafka, RQ |
-| `Serializer` | `PickleSerializer` | `JsonSerializer`, `SafePickleSerializer` | MessagePack, custom |
-
-**Storage layout** (in `.cashet/`):
+Storage layout under `.cashet/`:
```
.cashet/
-├── objects/ # content-addressable blobs (like git objects)
-│ ├── a3/
-│ │ └── b4c5d6... # compressed result blob
-│ └── e7/
-│ └── f8g9h0...
-└── meta.db # SQLite: commits, fingerprints, provenance, inline_objects
+├── objects/ # content-addressable blobs, like git objects
+│ └── a3/b4c5d6... # compressed result blob
+└── meta.db # SQLite: commits, fingerprints, provenance, inline objects
```
-**Small objects** (<1KB) are stored inline in `meta.db` instead of the filesystem. This reduces inode overhead for caches with many tiny results. Larger objects are stored as compressed blobs in `objects/` as usual.
+Small results (under 1 KB) are stored inline in `meta.db` to avoid inode overhead; larger results are compressed blobs in `objects/`.
-**Key design decisions:**
+### Key design decisions
-- **Closure variables are not hashed** and emit a `ClosureWarning` if present. Function identity is source code, defaults, keyword defaults, immutable referenced globals, and referenced helper functions; not arbitrary runtime state. If you need cache invalidation based on a mutable value, pass it as an explicit argument.
-- **Referenced user-defined helper functions are hashed recursively.** If your cached function calls or references a helper from your own project (via `co_names` / `globals`), that helper's source is included in the cache key. Change the helper and the caller's cache invalidates. Builtin and stdlib functions are skipped. This behavior is automatic and invisible — no decorators or imports needed.
-- **Nested refs resolve through containers.** `ResultRef` and `AsyncResultRef` values inside lists, tuples, sets, frozensets, and dicts are loaded before execution and recorded as commit input refs.
+- **Function identity is source code, not runtime state.** The hash covers AST-normalized source, default and keyword-default values, immutable referenced globals, and referenced user-defined helper functions. Change a helper and the caller's cache invalidates. Builtins and stdlib are skipped.
+- **Closure variables are not hashed** and emit a `ClosureWarning`. To invalidate on a mutable value, pass it as an explicit argument.
+- **Arguments are hashed by value, including objects.** Custom objects are hashed by their `__dict__` and `__slots__` state plus their class module and qualname, so same-named classes from different modules do not collide.
+- **Source is canonicalized with `ast.unparse`,** so comments, docstrings, and whitespace do not invalidate the cache, and hashes stay stable across Python versions.
- **Blobs are deduplicated by content hash.** Identical results share one blob on disk.
-- **Source is hashed as an AST.** Comments, docstrings, and whitespace changes don't invalidate the cache.
-- **Custom object arguments include their class module and qualname** in the argument hash so same-named classes from different modules do not collide.
-- **Non-cached tasks get unique commit hashes** (timestamp salt) so they always re-execute but still record lineage.
-- **Parent tracking:** Each commit records the hash of the previous commit for the same function+args, forming a history chain you can traverse.
+- **Nested refs resolve through containers.** `ResultRef` values inside lists, tuples, sets, frozensets, and dicts are loaded before execution and recorded as commit inputs.
+- **Non-cached tasks get a timestamp-salted commit hash,** so they always re-execute while still recording lineage.
+
+### Concurrency
+
+cashet is safe across threads, processes, and machines that share one store. Concurrent submissions of the same uncached task are deduplicated: the function runs exactly once and all callers get the same result. Cross-process claims use file locks (SQLite) or per-fingerprint Redis locks, with a heartbeat lease so a crashed worker's claim is reclaimed (default 5 minutes, configurable via `LocalExecutor(running_ttl=...)`).
+
+### Notebooks
+
+cashet resolves function source through `inspect.getsource`, then `dill.source.getsource` for interactive sessions, then a bytecode fallback that survives a kernel restart. Edit a notebook cell, rerun it, and the cache invalidates on the new code.
## Configuration
-- **`CASHET_DIR`** — override the default `.cashet` store directory. Equivalent to passing `store_dir=`.
-- **`CASHET_LOG`** — set to `DEBUG`, `INFO`, `WARNING`, or `ERROR` to print log output to stderr with a `[cashet]` tag. Messages include task fingerprint, function name, commit hash, and duration inline — no structured logging backend needed.
+- **`CASHET_DIR`** overrides the default `.cashet` store directory, equivalent to `store_dir=`.
+- **`CASHET_LOG`** set to `DEBUG`, `INFO`, `WARNING`, or `ERROR` prints logs to stderr with a `[cashet]` tag, including fingerprint, function name, commit hash, and duration.
-## Project Status
+## Version Compatibility
+
+Upgrading across a hash-format change does not corrupt anything; old entries simply miss and recompute on first access. To start clean, run `cashet clear` or point at a fresh store directory.
-**Beta.** The core (hashing, DAG resolution, fingerprint dedup) is stable. Works reliably for single-machine, multiprocess, and multi-machine (Redis) workflows.
+- **0.4.4 to 0.4.5:** Hashing fixes (slotted-object state, referenced global containers, `ast.unparse` canonicalization) change function and argument cache keys, so results cached by earlier versions recompute on first access. The Redis tag index key scheme changed and is not migrated; rewrite affected commits to rebuild tag indexes. `import_archive` now returns `ImportResult(imported, skipped)` instead of a bare count.
+- **0.3.x to 0.4.0:** Added per-entry TTL and tag-based invalidation. SQLite auto-migrates; old caches stay readable.
+- **0.3.0 to 0.3.1:** Redis blob keys renamed to `cashet:blob:data:{hash}` and stats backfilled once. Clear Redis caches before upgrading if you rely on long-lived reuse.
+- **0.2.x to 0.3.x:** Hash format unified; caches from 0.2.x do not hit on 0.3.x.
+
+## Project Status
-Built-in: `SQLiteStore` + `AsyncSQLiteStore`, `RedisStore` + `AsyncRedisStore`, `LocalExecutor` + `AsyncLocalExecutor`, `PickleSerializer` + `JsonSerializer` + `SafePickleSerializer`, HTTP server (`client.serve()`), CLI.
+Beta. The core (hashing, DAG resolution, fingerprint dedup) is stable and works for single-machine, multiprocess, and multi-machine (Redis) workflows.
-Not yet built: RocksDB, S3 stores; Celery/Kafka executors. PRs welcome.
+Built in: `SQLiteStore` and `AsyncSQLiteStore`, `RedisStore` and `AsyncRedisStore`, `LocalExecutor` and `AsyncLocalExecutor`, `PickleSerializer`, `JsonSerializer`, and `SafePickleSerializer`, the HTTP server, and the CLI. Not yet built: RocksDB and S3 stores, Celery and Kafka executors. Pull requests welcome.
## License
diff --git a/pyproject.toml b/pyproject.toml
index 1819618..36692e5 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
[project]
name = "cashet"
-version = "0.4.4"
+version = "0.4.5"
description = "A Python memoization cache with Redis, async support, DAG pipelines, and an HTTP server"
readme = "README.md"
license = "MIT"
diff --git a/src/cashet/_batch.py b/src/cashet/_batch.py
index fff3398..9a1f790 100644
--- a/src/cashet/_batch.py
+++ b/src/cashet/_batch.py
@@ -228,16 +228,25 @@ def _schedule_ready() -> None:
tasks[task] = key
_schedule_ready()
- while tasks:
- done, _ = await asyncio.wait(tasks.keys(), return_when=asyncio.FIRST_COMPLETED)
- for task in done:
- key = tasks.pop(task)
- results[key] = task.result()
- for dependent in rev[key]:
- in_degree[dependent] -= 1
- if in_degree[dependent] == 0:
- ready.append(dependent)
- _schedule_ready()
+ try:
+ while tasks:
+ done, _ = await asyncio.wait(tasks.keys(), return_when=asyncio.FIRST_COMPLETED)
+ for task in done:
+ key = tasks.pop(task)
+ results[key] = task.result()
+ for dependent in rev[key]:
+ in_degree[dependent] -= 1
+ if in_degree[dependent] == 0:
+ ready.append(dependent)
+ _schedule_ready()
+ except BaseException:
+ # On any failure, cancel and drain in-flight siblings so they cannot keep
+ # running (and writing commits) after the error has been surfaced.
+ for pending in tasks:
+ pending.cancel()
+ if tasks:
+ await asyncio.gather(*tasks, return_exceptions=True)
+ raise
logger.info("batch completed task_count=%d", len(keys))
return results
diff --git a/src/cashet/_export.py b/src/cashet/_export.py
index 0d5a77c..e439f3a 100644
--- a/src/cashet/_export.py
+++ b/src/cashet/_export.py
@@ -8,7 +8,7 @@
from datetime import datetime, timedelta
from io import BytesIO
from pathlib import Path
-from typing import Any
+from typing import Any, NamedTuple
from cashet.models import Commit, ObjectRef, StorageTier, TaskDef, TaskStatus
from cashet.protocols import AsyncStore
@@ -164,8 +164,14 @@ def _is_safe_tar_member(name: str) -> bool:
return ".." not in name.split("/")
-async def import_store(store: AsyncStore, tar_path: Path) -> int:
+class ImportResult(NamedTuple):
+ imported: int
+ skipped: int
+
+
+async def import_store(store: AsyncStore, tar_path: Path) -> ImportResult:
imported = 0
+ skipped = 0
with tarfile.open(tar_path, "r:gz") as tar:
members: dict[str, tarfile.TarInfo] = {}
for m in tar.getmembers():
@@ -257,6 +263,10 @@ async def import_store(store: AsyncStore, tar_path: Path) -> int:
missing_blobs.add(h)
has_missing = True
if has_missing:
+ skipped += 1
+ logger.warning(
+ "skipping commit with missing blob(s) hash=%s", commit.hash[:12]
+ )
continue
commit.input_refs = [
blob_refs.get(r.hash, r) for r in commit.input_refs
@@ -267,4 +277,4 @@ async def import_store(store: AsyncStore, tar_path: Path) -> int:
commit.output_ref = blob_refs[out_h]
await store.put_commit(commit)
imported += 1
- return imported
+ return ImportResult(imported=imported, skipped=skipped)
diff --git a/src/cashet/async_client.py b/src/cashet/async_client.py
index c1c9499..deb8696 100644
--- a/src/cashet/async_client.py
+++ b/src/cashet/async_client.py
@@ -21,7 +21,7 @@
resolve_task_config,
set_task_metadata,
)
-from cashet._export import export_store, import_store
+from cashet._export import ImportResult, export_store, import_store
from cashet.async_executor import AsyncLocalExecutor
from cashet.dag import AsyncResultRef
from cashet.hashing import PickleSerializer, Serializer, build_task_def, warn_default_pickle
@@ -251,7 +251,7 @@ async def clear(self) -> int:
async def export(self, path: str | Path) -> None:
await export_store(self.store, Path(path))
- async def import_archive(self, path: str | Path) -> int:
+ async def import_archive(self, path: str | Path) -> ImportResult:
return await import_store(self.store, Path(path))
async def close(self) -> None:
diff --git a/src/cashet/cli.py b/src/cashet/cli.py
index 1c2c13c..b19f3d2 100644
--- a/src/cashet/cli.py
+++ b/src/cashet/cli.py
@@ -119,7 +119,7 @@ def show_cmd(hash: str) -> None:
raise SystemExit(1) from None
if commit is None:
console.print(f"[red]Commit {hash} not found.[/red]")
- return
+ raise SystemExit(1)
out_hash = commit.output_ref.hash[:12] if commit.output_ref else "none"
out_size = commit.output_ref.size if commit.output_ref else 0
parent = commit.parent_hash[:12] if commit.parent_hash else "none"
@@ -164,7 +164,7 @@ def get_cmd(hash: str, output: str | None) -> None:
raise SystemExit(1) from None
if commit is None or commit.output_ref is None:
console.print(f"[red]No result for {hash}[/red]")
- return
+ raise SystemExit(1)
ref = client.store.get_blob(commit.output_ref)
if output:
pathlib.Path(output).write_bytes(ref)
@@ -357,11 +357,15 @@ def import_archive(path: str) -> None:
client = _client()
try:
- count = client.import_archive(path)
+ result = client.import_archive(path)
except (OSError, tarfile.TarError, ValueError) as e:
console.print(f"[red]Import failed: {e}[/red]")
raise SystemExit(1) from None
- console.print(f"[green]Imported {count} commit(s) from {path}[/green]")
+ console.print(f"[green]Imported {result.imported} commit(s) from {path}[/green]")
+ if result.skipped:
+ console.print(
+ f"[yellow]Skipped {result.skipped} commit(s) with missing blobs[/yellow]"
+ )
if __name__ == "__main__":
diff --git a/src/cashet/client.py b/src/cashet/client.py
index 879255d..c8d5e29 100644
--- a/src/cashet/client.py
+++ b/src/cashet/client.py
@@ -9,9 +9,6 @@
from typing import Any, TypeVar, cast, overload
from cashet._batch import (
- build_deps,
- normalize_tasks,
- topological_sort,
unpack_dict_tasks,
unpack_list_tasks,
)
@@ -19,6 +16,7 @@
resolve_store_dir,
set_task_metadata,
)
+from cashet._export import ImportResult
from cashet._runner import BlockingAsyncRunner
from cashet.adapters import SyncStoreAdapter
from cashet.async_client import AsyncClient
@@ -192,15 +190,12 @@ def submit_many(
max_workers: int | None = None,
) -> list[ResultRef[Any]] | dict[str, ResultRef[Any]]:
is_dict = isinstance(tasks, dict)
+ # Only the keys are needed here to map results back; the async client owns
+ # normalization, dependency resolution, and topological ordering.
if is_dict:
- keys, raw_tasks = unpack_dict_tasks(tasks)
+ keys, _ = unpack_dict_tasks(tasks)
else:
- keys, raw_tasks = unpack_list_tasks(tasks)
-
- key_set = set(keys)
- normalized = normalize_tasks(raw_tasks, _cache, _tags, _retries, _force, _timeout, _ttl)
- deps, _task_refs = build_deps(keys, normalized, key_set)
- _order = topological_sort(deps)
+ keys, _ = unpack_list_tasks(tasks)
workers = max_workers if max_workers is not None else self._max_workers
async_results = self._runner.call(
@@ -314,7 +309,7 @@ def serve(
def export(self, path: str | Path) -> None:
self._runner.call(self._async_client.export(path))
- def import_archive(self, path: str | Path) -> int:
+ def import_archive(self, path: str | Path) -> ImportResult:
return self._runner.call(self._async_client.import_archive(path))
def close(self) -> None:
diff --git a/src/cashet/dag.py b/src/cashet/dag.py
index 1151f11..6a9711c 100644
--- a/src/cashet/dag.py
+++ b/src/cashet/dag.py
@@ -6,7 +6,7 @@
from datetime import UTC, datetime
from typing import Any, Generic, TypeVar
-from cashet.hashing import Serializer
+from cashet.hashing import Serializer, object_state
from cashet.models import Commit, ObjectRef, TaskDef, TaskStatus
from cashet.protocols import AsyncStore
@@ -36,6 +36,13 @@ def _collect_input_refs(value: Any, refs: list[ObjectRef], visited: set[int]) ->
for item in value:
_collect_input_refs(item, refs, visited)
visited.discard(value_id)
+ else:
+ state = object_state(value)
+ if state is not None:
+ visited.add(value_id)
+ for item in state.values():
+ _collect_input_refs(item, refs, visited)
+ visited.discard(value_id)
def resolve_input_refs(args: tuple[Any, ...], kwargs: dict[str, Any]) -> list[ObjectRef]:
diff --git a/src/cashet/hashing.py b/src/cashet/hashing.py
index 5cfa771..ffb893b 100644
--- a/src/cashet/hashing.py
+++ b/src/cashet/hashing.py
@@ -12,6 +12,7 @@
import types
import warnings
from datetime import timedelta
+from functools import lru_cache
from typing import Any, Protocol, runtime_checkable
from cashet.models import TaskDef
@@ -21,6 +22,10 @@ class ClosureWarning(UserWarning):
pass
+class UnhashableArgWarning(UserWarning):
+ pass
+
+
_pickle_warning_issued = False
@@ -203,10 +208,13 @@ def hash_source(source: str) -> str:
def _ast_canonical(source: str) -> str:
+ # ast.unparse normalizes whitespace and comments like ast.dump but, being
+ # source text rather than the internal AST repr, stays stable across Python
+ # versions whose ast.dump field set differs (e.g. type_params in 3.12).
try:
tree = ast.parse(source)
_strip_docstrings(tree)
- return ast.dump(tree)
+ return ast.unparse(tree)
except SyntaxError:
return source
@@ -225,6 +233,8 @@ def _strip_docstrings(node: ast.AST) -> None:
and isinstance(child.body[0].value.value, str)
):
child.body = child.body[1:]
+ if not child.body:
+ child.body = [ast.Pass()]
def _is_stdlib_or_site_path(path: str) -> bool:
@@ -292,11 +302,19 @@ def _should_hash_global_value(obj: Any, visited: set[int] | None = None) -> bool
)
visited.discard(obj_id)
return result
- if isinstance(obj, tuple | frozenset):
+ if isinstance(obj, tuple | frozenset | list | set):
visited.add(obj_id)
result = all(_should_hash_global_value(item, visited) for item in obj)
visited.discard(obj_id)
return result
+ if isinstance(obj, dict):
+ visited.add(obj_id)
+ result = all(
+ _should_hash_global_value(k, visited) and _should_hash_global_value(v, visited)
+ for k, v in obj.items()
+ )
+ visited.discard(obj_id)
+ return result
return False
@@ -376,6 +394,37 @@ def hash_function(
return h.hexdigest()
+@lru_cache(maxsize=1024)
+def _slot_names(cls: type) -> tuple[str, ...]:
+ names: list[str] = []
+ for klass in cls.__mro__:
+ slots = klass.__dict__.get("__slots__")
+ if slots is None:
+ continue
+ if isinstance(slots, str):
+ slots = (slots,)
+ for name in slots:
+ if name in ("__dict__", "__weakref__"):
+ continue
+ names.append(name)
+ return tuple(names)
+
+
+_UNSET = object()
+
+
+def object_state(obj: Any) -> dict[str, Any] | None:
+ state: dict[str, Any] = {}
+ instance_dict = getattr(obj, "__dict__", None)
+ if isinstance(instance_dict, dict):
+ state.update(instance_dict)
+ for name in _slot_names(type(obj)):
+ value = getattr(obj, name, _UNSET)
+ if value is not _UNSET:
+ state[name] = value
+ return state or None
+
+
def _stable_repr_to(
buf: io.StringIO, obj: Any, _visited: set[int] | None = None
) -> None:
@@ -453,18 +502,30 @@ def _stable_repr_to(
buf.write(f"")
elif hasattr(obj, "__cashet_ref__"):
buf.write(f"")
- elif hasattr(obj, "__dict__"):
- obj_id = id(obj)
- if obj_id in _visited:
- buf.write(f"<{type(obj).__module__}.{type(obj).__qualname__}:...>")
- return
- _visited.add(obj_id)
- buf.write(f"<{type(obj).__module__}.{type(obj).__qualname__}:")
- _stable_repr_to(buf, obj.__dict__, _visited)
- buf.write(">")
- _visited.discard(obj_id)
else:
- buf.write(repr(obj))
+ state = object_state(obj)
+ if state is not None:
+ obj_id = id(obj)
+ if obj_id in _visited:
+ buf.write(f"<{type(obj).__module__}.{type(obj).__qualname__}:...>")
+ return
+ _visited.add(obj_id)
+ buf.write(f"<{type(obj).__module__}.{type(obj).__qualname__}:")
+ _stable_repr_to(buf, state, _visited)
+ buf.write(">")
+ _visited.discard(obj_id)
+ else:
+ if type(obj).__repr__ is object.__repr__:
+ warnings.warn(
+ f"Argument of type "
+ f"{type(obj).__module__}.{type(obj).__qualname__} has no "
+ f"__dict__/__slots__ and uses the default repr; it cannot be "
+ f"hashed by value and will not cache reliably. Pass a "
+ f"value-stable representation instead.",
+ UnhashableArgWarning,
+ stacklevel=2,
+ )
+ buf.write(repr(obj))
def _stable_hash(
diff --git a/src/cashet/redis_store.py b/src/cashet/redis_store.py
index 3bb0951..334230e 100644
--- a/src/cashet/redis_store.py
+++ b/src/cashet/redis_store.py
@@ -18,6 +18,9 @@
local ref_key = KEYS[1]
local blob_key = KEYS[2]
local stats_key = KEYS[3]
+ if redis.call('EXISTS', ref_key) == 0 then
+ return 0
+ end
local count = redis.call('DECR', ref_key)
if count <= 0 then
local existed = redis.call('EXISTS', blob_key)
@@ -54,6 +57,10 @@ def _fp_key(fingerprint: str) -> str:
return f"cashet:index:fingerprint:{fingerprint}"
+def _running_key(fingerprint: str) -> str:
+ return f"cashet:index:running:{fingerprint}"
+
+
def _func_key(func_name: str) -> str:
return f"cashet:index:func:{func_name}"
@@ -63,11 +70,13 @@ def _status_key(status: str) -> str:
def _tag_key(key: str) -> str:
- return f"cashet:tag:{key}"
+ return f"cashet:tagk:{key}"
def _tag_value_key(key: str, value: str) -> str:
- return f"cashet:tag:{key}:{value}"
+ # Length-prefix the key so a ':' inside a tag key or value can never make
+ # two distinct (key, value) pairs collide onto the same set.
+ return f"cashet:tagv:{len(key)}:{key}:{value}"
def _access_key() -> str:
@@ -250,6 +259,10 @@ def _index_commit_commands(pipe: Any, commit: Commit) -> None:
for status in TaskStatus:
pipe.srem(_status_key(status.value), commit.hash)
pipe.sadd(_status_key(commit.status.value), commit.hash)
+ if commit.status == TaskStatus.RUNNING:
+ pipe.sadd(_running_key(commit.fingerprint), commit.hash)
+ else:
+ pipe.srem(_running_key(commit.fingerprint), commit.hash)
for key, val in commit.tags.items():
pipe.sadd(_tag_key(key), commit.hash)
pipe.sadd(_tag_value_key(key, val), commit.hash)
@@ -259,6 +272,7 @@ def _remove_commit_index_commands(pipe: Any, commit: Commit, resolved_hash: str)
pipe.delete(_commit_key(resolved_hash))
pipe.zrem("cashet:index:all", resolved_hash)
pipe.zrem(_fp_key(commit.fingerprint), resolved_hash)
+ pipe.srem(_running_key(commit.fingerprint), resolved_hash)
pipe.zrem(_func_key(commit.task_def.func_name), resolved_hash)
pipe.zrem(_access_key(), resolved_hash)
for status in TaskStatus:
@@ -397,7 +411,7 @@ async def find_by_fingerprint(self, fingerprint: str) -> Commit | None:
return None
async def find_running_by_fingerprint(self, fingerprint: str) -> Commit | None:
- hashes = await self._redis.zrevrange(_fp_key(fingerprint), 0, -1)
+ hashes = await self._redis.smembers(_running_key(fingerprint))
for h in hashes:
h_str = h.decode() if isinstance(h, bytes) else h
commit = await self.get_commit(h_str)
diff --git a/src/cashet/server.py b/src/cashet/server.py
index 1f35634..dec1f9c 100644
--- a/src/cashet/server.py
+++ b/src/cashet/server.py
@@ -2,6 +2,7 @@
import asyncio
import base64
+import functools
import hmac
import json
import logging
@@ -12,7 +13,6 @@
from starlette.applications import Starlette
from starlette.middleware import Middleware
-from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route
@@ -48,24 +48,64 @@ def _validate_remote_code_options(
_DEFAULT_MAX_CONTENT_LENGTH = 500 * 1024 * 1024
-def _limit_request_size(
- request: Request,
-) -> JSONResponse | None:
- max_size = getattr(request.app.state, "max_content_length", _DEFAULT_MAX_CONTENT_LENGTH)
- content_length = request.headers.get("content-length")
- if content_length is not None:
+def _too_large_response(max_size: int) -> _CustomJSONResponse:
+ return _CustomJSONResponse(
+ {"error": f"request body exceeds {max_size} bytes"}, status_code=413
+ )
+
+
+class _BadRequestError(Exception):
+ pass
+
+
+def _query_int(request: Request, name: str, default: int) -> int:
+ raw = request.query_params.get(name)
+ if raw is None:
+ return default
+ try:
+ return int(raw)
+ except ValueError as exc:
+ raise _BadRequestError(f"{name} must be an integer") from exc
+
+
+async def _json_body(request: Request) -> dict[str, Any]:
+ body = await request.body()
+ if not body:
+ return {}
+ data = json.loads(body)
+ if not isinstance(data, dict):
+ raise _BadRequestError("request body must be a JSON object")
+ return data
+
+
+def _gc_params(data: dict[str, Any]) -> tuple[float, int | None]:
+ older_than_days = data.get("older_than_days", 30)
+ if isinstance(older_than_days, bool) or not isinstance(older_than_days, int | float):
+ raise _BadRequestError("older_than_days must be a number")
+ max_size = data.get("max_size")
+ if max_size is not None and (isinstance(max_size, bool) or not isinstance(max_size, int)):
+ raise _BadRequestError("max_size must be an integer or null")
+ return older_than_days, max_size
+
+
+def _safe_handler(handler: Any) -> Any:
+ @functools.wraps(handler)
+ async def wrapper(request: Request) -> JSONResponse:
try:
- size = int(content_length)
- except ValueError:
- return _CustomJSONResponse(
- {"error": "invalid content-length"}, status_code=400
- )
- if size > max_size:
- return _CustomJSONResponse(
- {"error": f"request body exceeds {max_size} bytes"},
- status_code=413,
+ return await handler(request)
+ except _BadRequestError as exc:
+ return _CustomJSONResponse({"error": str(exc)}, status_code=400)
+ except json.JSONDecodeError:
+ return _CustomJSONResponse({"error": "invalid JSON body"}, status_code=400)
+ except Exception:
+ logger.exception(
+ "request failed method=%s path=%s status=500",
+ request.method,
+ request.url.path,
)
- return None
+ return _CustomJSONResponse({"error": "Internal server error"}, status_code=500)
+
+ return wrapper
def _reconstruct_func(data: dict[str, Any]) -> Callable[..., Any] | None:
@@ -160,13 +200,17 @@ def _decode_call(
return tuple(args), kwargs, None
+def _token_authorized(auth: str, token: str) -> bool:
+ return auth.startswith("Bearer ") and hmac.compare_digest(auth[7:], token)
+
+
def _require_token(handler: Any, token: str | None) -> Any:
if token is None:
return handler
async def wrapper(request: Request) -> JSONResponse:
auth = request.headers.get("authorization", "")
- if not auth.startswith("Bearer ") or not hmac.compare_digest(auth[7:], token):
+ if not _token_authorized(auth, token):
logger.warning(
"request unauthorized method=%s path=%s",
request.method,
@@ -295,7 +339,7 @@ async def _async_commit(request: Request) -> JSONResponse:
async def _async_log(request: Request) -> JSONResponse:
client: AsyncClient = request.app.state.client
func_name = request.query_params.get("func")
- limit = int(request.query_params.get("limit", "50"))
+ limit = _query_int(request, "limit", 50)
status = request.query_params.get("status")
start = time.perf_counter()
commits = await client.log(func_name=func_name, limit=limit, status=status)
@@ -323,9 +367,7 @@ async def _async_gc(request: Request) -> JSONResponse:
from datetime import timedelta
client: AsyncClient = request.app.state.client
- data = await request.json()
- older_than_days = data.get("older_than_days", 30)
- max_size = data.get("max_size")
+ older_than_days, max_size = _gc_params(await _json_body(request))
start = time.perf_counter()
deleted = await client.gc(timedelta(days=older_than_days), max_size_bytes=max_size)
duration = int((time.perf_counter() - start) * 1000)
@@ -346,20 +388,28 @@ def create_async_app(
) -> Starlette:
_validate_remote_code_options(allow_remote_code, require_token)
routes = [
- Route("/submit", _require_token(_async_submit, require_token), methods=["POST"]),
+ Route(
+ "/submit",
+ _require_token(_safe_handler(_async_submit), require_token),
+ methods=["POST"],
+ ),
Route(
"/result/{commit_hash}",
- _require_token(_async_result, require_token),
+ _require_token(_safe_handler(_async_result), require_token),
methods=["GET"],
),
Route(
"/commit/{commit_hash}",
- _require_token(_async_commit, require_token),
+ _require_token(_safe_handler(_async_commit), require_token),
methods=["GET"],
),
- Route("/log", _require_token(_async_log, require_token), methods=["GET"]),
- Route("/stats", _require_token(_async_stats, require_token), methods=["GET"]),
- Route("/gc", _require_token(_async_gc, require_token), methods=["POST"]),
+ Route("/log", _require_token(_safe_handler(_async_log), require_token), methods=["GET"]),
+ Route(
+ "/stats",
+ _require_token(_safe_handler(_async_stats), require_token),
+ methods=["GET"],
+ ),
+ Route("/gc", _require_token(_safe_handler(_async_gc), require_token), methods=["POST"]),
]
app = Starlette(
@@ -370,15 +420,76 @@ def create_async_app(
app.state.tasks = _server_tasks(client, tasks)
app.state.allow_remote_code = allow_remote_code
app.state.max_content_length = max_content_length
+ app.state.require_token = require_token
return app
-class _SizeLimitMiddleware(BaseHTTPMiddleware):
- async def dispatch(self, request: Request, call_next: Any) -> JSONResponse:
- error = _limit_request_size(request)
- if error is not None:
- return error
- return await call_next(request)
+class _SizeLimitMiddleware:
+ # Pure-ASGI so the body cap is enforced on bytes actually received, not on a
+ # client-supplied Content-Length that is absent for chunked requests.
+ def __init__(self, app: Any) -> None:
+ self.app = app
+
+ async def __call__(self, scope: Any, receive: Any, send: Any) -> None:
+ if scope["type"] != "http":
+ await self.app(scope, receive, send)
+ return
+ starlette_app = scope.get("app")
+ state = starlette_app.state if starlette_app is not None else None
+ max_size = getattr(state, "max_content_length", _DEFAULT_MAX_CONTENT_LENGTH)
+
+ # Reject unauthenticated requests before buffering any body, so an
+ # unauthenticated client cannot force buffering up to max_content_length.
+ require_token = getattr(state, "require_token", None)
+ if require_token is not None:
+ auth = ""
+ for name, value in scope["headers"]:
+ if name == b"authorization":
+ auth = value.decode("latin-1")
+ break
+ if not _token_authorized(auth, require_token):
+ await _CustomJSONResponse(
+ {"error": "unauthorized"}, status_code=401
+ )(scope, receive, send)
+ return
+
+ for name, value in scope["headers"]:
+ if name == b"content-length":
+ try:
+ declared = int(value)
+ except ValueError:
+ await _CustomJSONResponse(
+ {"error": "invalid content-length"}, status_code=400
+ )(scope, receive, send)
+ return
+ if declared > max_size:
+ await _too_large_response(max_size)(scope, receive, send)
+ return
+ break
+
+ body = bytearray()
+ while True:
+ message = await receive()
+ if message["type"] != "http.request":
+ break
+ body.extend(message.get("body", b""))
+ if len(body) > max_size:
+ await _too_large_response(max_size)(scope, receive, send)
+ return
+ if not message.get("more_body", False):
+ break
+
+ buffered = bytes(body)
+ replayed = False
+
+ async def replay_receive() -> Any:
+ nonlocal replayed
+ if not replayed:
+ replayed = True
+ return {"type": "http.request", "body": buffered, "more_body": False}
+ return await receive()
+
+ await self.app(scope, replay_receive, send)
# Sync handlers run sync Client operations in threads so they don't block the event loop
@@ -502,7 +613,7 @@ def _run() -> JSONResponse:
async def _log(request: Request) -> JSONResponse:
client: Client = request.app.state.client
func_name = request.query_params.get("func")
- limit = int(request.query_params.get("limit", "50"))
+ limit = _query_int(request, "limit", 50)
status = request.query_params.get("status")
start = time.perf_counter()
@@ -537,9 +648,7 @@ async def _gc(request: Request) -> JSONResponse:
from datetime import timedelta
client: Client = request.app.state.client
- data = await request.json()
- older_than_days = data.get("older_than_days", 30)
- max_size = data.get("max_size")
+ older_than_days, max_size = _gc_params(await _json_body(request))
start = time.perf_counter()
def _run() -> JSONResponse:
@@ -565,20 +674,28 @@ def create_app(
) -> Starlette:
_validate_remote_code_options(allow_remote_code, require_token)
routes = [
- Route("/submit", _require_token(_submit, require_token), methods=["POST"]),
+ Route(
+ "/submit",
+ _require_token(_safe_handler(_submit), require_token),
+ methods=["POST"],
+ ),
Route(
"/result/{commit_hash}",
- _require_token(_result, require_token),
+ _require_token(_safe_handler(_result), require_token),
methods=["GET"],
),
Route(
"/commit/{commit_hash}",
- _require_token(_commit, require_token),
+ _require_token(_safe_handler(_commit), require_token),
+ methods=["GET"],
+ ),
+ Route("/log", _require_token(_safe_handler(_log), require_token), methods=["GET"]),
+ Route(
+ "/stats",
+ _require_token(_safe_handler(_stats), require_token),
methods=["GET"],
),
- Route("/log", _require_token(_log, require_token), methods=["GET"]),
- Route("/stats", _require_token(_stats, require_token), methods=["GET"]),
- Route("/gc", _require_token(_gc, require_token), methods=["POST"]),
+ Route("/gc", _require_token(_safe_handler(_gc), require_token), methods=["POST"]),
]
app = Starlette(
routes=routes,
@@ -588,4 +705,5 @@ def create_app(
app.state.tasks = _server_tasks(client, tasks)
app.state.allow_remote_code = allow_remote_code
app.state.max_content_length = max_content_length
+ app.state.require_token = require_token
return app
diff --git a/src/cashet/store.py b/src/cashet/store.py
index a80c60a..34f63c6 100644
--- a/src/cashet/store.py
+++ b/src/cashet/store.py
@@ -341,10 +341,15 @@ def find_by_fingerprint(self, fingerprint: str) -> Commit | None:
).fetchone()
if row is None:
return None
- conn.execute(
- "UPDATE commits SET last_accessed_at = ? WHERE hash = ?",
- (now_iso, row["hash"]),
- )
+ try:
+ conn.execute(
+ "UPDATE commits SET last_accessed_at = ? WHERE hash = ?",
+ (now_iso, row["hash"]),
+ )
+ except sqlite3.OperationalError:
+ # Access-time bump only feeds LRU ordering; never fail a cache hit
+ # because a concurrent writer holds the lock past busy_timeout.
+ logger.debug("last_accessed_at bump skipped (db locked) hash=%s", row["hash"][:12])
return self._row_to_commit(row)
def find_running_by_fingerprint(self, fingerprint: str) -> Commit | None:
@@ -553,17 +558,25 @@ def evict(
deleted,
"size_limit" if max_size_bytes is not None else "ttl",
)
- c = self._connect()
- mode = c.execute("PRAGMA auto_vacuum").fetchone()[0]
- if mode == 2:
- c.execute("PRAGMA incremental_vacuum")
- else:
- c.execute("VACUUM")
+ try:
+ self._vacuum()
+ except sqlite3.OperationalError:
+ # Reclaiming space is best-effort; a concurrent writer holding the
+ # lock must not undo an eviction whose deletes already committed.
+ logger.debug("vacuum skipped (db busy) after eviction")
else:
logger.debug("eviction found no candidates")
return deleted
+ def _vacuum(self) -> None:
+ conn = self._connect()
+ mode = conn.execute("PRAGMA auto_vacuum").fetchone()[0]
+ if mode == 2:
+ conn.execute("PRAGMA incremental_vacuum")
+ else:
+ conn.execute("VACUUM")
+
def _evict_to_size(self, current_bytes: int, max_size_bytes: int) -> int:
pending_orphans: list[str] = []
deleted = 0
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 9c6ff88..c60b329 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -104,7 +104,7 @@ def greet(name: str) -> str:
def test_show_missing(self, cli_runner: CliRunner, store_dir: Path) -> None:
result = _invoke(cli_runner, ["show", "deadbeef"], store_dir)
- assert result.exit_code == 0
+ assert result.exit_code == 1
assert "not found" in result.output
def test_show_ambiguous_hash(self, cli_runner: CliRunner, store_dir: Path) -> None:
@@ -152,7 +152,7 @@ def val() -> str:
def test_get_missing(self, cli_runner: CliRunner, store_dir: Path) -> None:
result = _invoke(cli_runner, ["get", "deadbeef"], store_dir)
- assert result.exit_code == 0
+ assert result.exit_code == 1
assert "No result" in result.output
def test_get_ambiguous_hash(self, cli_runner: CliRunner, store_dir: Path) -> None:
diff --git a/tests/test_core.py b/tests/test_core.py
index 840e8f8..39b1078 100644
--- a/tests/test_core.py
+++ b/tests/test_core.py
@@ -283,6 +283,27 @@ def total(values: list[int]) -> int:
assert commit is not None
assert [ref.hash for ref in commit.input_refs] == [data_ref.hash]
+ def test_ref_in_custom_object_attribute_recorded_as_input(self, client: Client) -> None:
+ class Box:
+ __slots__ = ("payload",)
+
+ def __init__(self, payload: object) -> None:
+ self.payload = payload
+
+ def gen_data() -> int:
+ return 42
+
+ def consume(box: object) -> int:
+ return 1
+
+ data_ref = client.submit(gen_data)
+ result_ref = client.submit(consume, Box(data_ref))
+ assert result_ref.load() == 1
+
+ commit = client.show(result_ref.commit_hash)
+ assert commit is not None
+ assert data_ref.hash in [ref.hash for ref in commit.input_refs]
+
def test_chained_pipeline(self, client: Client) -> None:
def step1() -> int:
return 10
diff --git a/tests/test_export.py b/tests/test_export.py
index a6036d6..cac79df 100644
--- a/tests/test_export.py
+++ b/tests/test_export.py
@@ -40,6 +40,22 @@ def corrupt_first_blob(src: Path, dst: Path) -> None:
patched.addfile(info, BytesIO(data))
+def drop_first_blob_and_manifest(src: Path, dst: Path) -> None:
+ with tarfile.open(src, "r:gz") as original, tarfile.open(dst, "w:gz") as patched:
+ dropped = False
+ for member in original.getmembers():
+ if member.name == "cashet-export/manifest.json":
+ continue
+ if member.name.startswith("cashet-export/blobs/") and not dropped:
+ dropped = True
+ continue
+ extracted = original.extractfile(member) if member.isfile() else None
+ data = extracted.read() if extracted is not None else b""
+ info = tarfile.TarInfo(member.name)
+ info.size = len(data)
+ patched.addfile(info, BytesIO(data))
+
+
class TestSyncExport:
def test_export_import_roundtrip(self, client: Client, tmp_path: Path) -> None:
ref1 = client.submit(add, 1, 2)
@@ -52,8 +68,8 @@ def test_export_import_roundtrip(self, client: Client, tmp_path: Path) -> None:
assert archive.exists()
client2 = Client(store_dir=tmp_path / ".cashet2")
- count = client2.import_archive(archive)
- assert count == 2
+ result = client2.import_archive(archive)
+ assert result.imported == 2
imported1 = client2.show(ref1.commit_hash)
assert imported1 is not None
@@ -73,8 +89,8 @@ def test_import_skips_existing(self, client: Client, tmp_path: Path) -> None:
archive = tmp_path / "export.tar.gz"
client.export(archive)
- count = client.import_archive(archive)
- assert count == 0
+ result = client.import_archive(archive)
+ assert result.imported == 0
def test_export_empty_store(self, client: Client, tmp_path: Path) -> None:
archive = tmp_path / "empty.tar.gz"
@@ -82,8 +98,8 @@ def test_export_empty_store(self, client: Client, tmp_path: Path) -> None:
assert archive.exists()
client2 = Client(store_dir=tmp_path / ".cashet2")
- count = client2.import_archive(archive)
- assert count == 0
+ result = client2.import_archive(archive)
+ assert result.imported == 0
def test_import_rejects_corrupt_blob_payload(self, client: Client, tmp_path: Path) -> None:
ref = client.submit(add, 1, 2)
@@ -99,6 +115,22 @@ def test_import_rejects_corrupt_blob_payload(self, client: Client, tmp_path: Pat
client2.import_archive(corrupt_archive)
assert client2.show(ref.commit_hash) is None
+ def test_import_reports_skipped_commits_with_missing_blobs(
+ self, client: Client, tmp_path: Path
+ ) -> None:
+ client.submit(add, 1, 2).load()
+ client.submit(greet, "world").load()
+
+ archive = tmp_path / "export.tar.gz"
+ lossy = tmp_path / "lossy.tar.gz"
+ client.export(archive)
+ drop_first_blob_and_manifest(archive, lossy)
+
+ client2 = Client(store_dir=tmp_path / ".cashet2")
+ result = client2.import_archive(lossy)
+ assert result.imported == 1
+ assert result.skipped == 1
+
class TestAsyncExport:
async def test_export_import_roundtrip(
@@ -114,8 +146,8 @@ async def test_export_import_roundtrip(
assert archive.exists()
client2 = AsyncClient(store_dir=tmp_path / ".cashet2")
- count = await client2.import_archive(archive)
- assert count == 2
+ result = await client2.import_archive(archive)
+ assert result.imported == 2
imported1 = await client2.show(ref1.commit_hash)
assert imported1 is not None
@@ -136,5 +168,5 @@ async def test_import_skips_existing(self, async_client: AsyncClient, tmp_path:
archive = tmp_path / "export.tar.gz"
await async_client.export(archive)
- count = await async_client.import_archive(archive)
- assert count == 0
+ result = await async_client.import_archive(archive)
+ assert result.imported == 0
diff --git a/tests/test_hashing.py b/tests/test_hashing.py
index 23d2bd0..0ad23e0 100644
--- a/tests/test_hashing.py
+++ b/tests/test_hashing.py
@@ -1,11 +1,18 @@
from __future__ import annotations
import warnings
+from dataclasses import dataclass
from datetime import date
from typing import Any
from cashet import Client
-from cashet.hashing import ClosureWarning, _ast_canonical, hash_function
+from cashet.hashing import (
+ ClosureWarning,
+ UnhashableArgWarning,
+ _ast_canonical,
+ hash_args,
+ hash_function,
+)
class TestHashingEdgeCases:
@@ -123,6 +130,71 @@ def f(x: int = base) -> int:
assert ref2.load() == 2
+class TestObjectStateHashing:
+ def test_slotted_objects_with_equal_state_hash_equal(self) -> None:
+ class Point:
+ __slots__ = ("x", "y")
+
+ def __init__(self, x: int, y: int) -> None:
+ self.x = x
+ self.y = y
+
+ assert hash_args(Point(1, 2)) == hash_args(Point(1, 2))
+
+ def test_slotted_objects_with_different_state_hash_differ(self) -> None:
+ class Point:
+ __slots__ = ("x", "y")
+
+ def __init__(self, x: int, y: int) -> None:
+ self.x = x
+ self.y = y
+
+ assert hash_args(Point(1, 2)) != hash_args(Point(1, 3))
+
+ def test_dataclass_with_slots_hashes_by_value(self) -> None:
+ @dataclass(slots=True)
+ class Config:
+ name: str
+ limit: int
+
+ assert hash_args(Config("a", 1)) == hash_args(Config("a", 1))
+ assert hash_args(Config("a", 1)) != hash_args(Config("a", 2))
+
+ def test_mixed_dict_and_slots_state_both_hashed(self) -> None:
+ class Base:
+ __slots__ = ("__dict__", "a")
+
+ def __init__(self, a: int, b: int) -> None:
+ self.a = a
+ self.b = b
+
+ assert hash_args(Base(1, 2)) == hash_args(Base(1, 2))
+ assert hash_args(Base(1, 2)) != hash_args(Base(1, 9))
+
+ def test_opaque_object_warns(self) -> None:
+ class Opaque:
+ __slots__ = ()
+
+ with warnings.catch_warnings(record=True) as w:
+ warnings.simplefilter("always")
+ hash_args(Opaque())
+ unhashable = [x for x in w if issubclass(x.category, UnhashableArgWarning)]
+ assert len(unhashable) >= 1
+
+ def test_custom_repr_object_does_not_warn(self) -> None:
+ class Stable:
+ __slots__ = ()
+
+ def __repr__(self) -> str:
+ return "Stable()"
+
+ with warnings.catch_warnings(record=True) as w:
+ warnings.simplefilter("always")
+ hash_args(Stable())
+ unhashable = [x for x in w if issubclass(x.category, UnhashableArgWarning)]
+ assert len(unhashable) == 0
+
+
class TestASTNormalizedHashing:
def test_comment_stripped(self) -> None:
src1 = "def foo(x):\n # important\n return x + 1"
@@ -144,6 +216,19 @@ def test_docstring_stripped(self) -> None:
src2 = 'def foo(x):\n """updated docs"""\n return x'
assert _ast_canonical(src1) == _ast_canonical(src2)
+ def test_only_docstring_function_does_not_crash(self) -> None:
+ src = 'def f():\n """only docs"""'
+ out = _ast_canonical(src)
+ assert "only docs" not in out
+ assert _ast_canonical(src) == out
+
+ def test_canonical_form_is_parseable_source(self) -> None:
+ import ast
+
+ out = _ast_canonical("def f(x):\n # note\n return x+1")
+ # ast.unparse yields source text (round-trippable), unlike ast.dump.
+ ast.parse(out)
+
def test_different_functions_different_hash(self) -> None:
def add(x: int, y: int) -> int:
return x + y
@@ -421,6 +506,44 @@ def test_exec_function_invalidates_on_global_value_change(self, client: Client)
assert ref1.load() == 20
assert ref2.load() == 30
+ def test_exec_function_invalidates_on_dict_global_change(self, client: Client) -> None:
+ namespace: dict[str, Any] = {"CONFIG": {"factor": 2}}
+ exec("def f(x):\n return x * CONFIG['factor']", namespace)
+ func = namespace["f"]
+ ref1 = client.submit(func, 10)
+
+ namespace["CONFIG"] = {"factor": 3}
+ ref2 = client.submit(func, 10)
+
+ assert ref1.hash != ref2.hash
+ assert ref1.load() == 20
+ assert ref2.load() == 30
+
+ def test_exec_function_invalidates_on_list_global_change(self, client: Client) -> None:
+ namespace: dict[str, Any] = {"WEIGHTS": [1, 2]}
+ exec("def f():\n return sum(WEIGHTS)", namespace)
+ func = namespace["f"]
+ ref1 = client.submit(func)
+
+ namespace["WEIGHTS"] = [1, 2, 3]
+ ref2 = client.submit(func)
+
+ assert ref1.hash != ref2.hash
+ assert ref1.load() == 3
+ assert ref2.load() == 6
+
+ def test_global_container_with_unstable_member_not_hashed(self, client: Client) -> None:
+ namespace: dict[str, Any] = {"REGISTRY": {"handler": object()}}
+ exec("def f():\n return len(REGISTRY)", namespace)
+ func = namespace["f"]
+ ref1 = client.submit(func)
+
+ namespace["REGISTRY"] = {"handler": object()}
+ ref2 = client.submit(func)
+
+ assert ref1.hash == ref2.hash
+ assert ref1.load() == 1
+
def test_comprehension_invalidates_on_global_value_change(self, client: Client) -> None:
namespace: dict[str, Any] = {"MULTIPLIER": 2}
exec("def f(xs):\n return [x * MULTIPLIER for x in xs]", namespace)
diff --git a/tests/test_map.py b/tests/test_map.py
index 841e7c6..76c31c8 100644
--- a/tests/test_map.py
+++ b/tests/test_map.py
@@ -1,7 +1,10 @@
from __future__ import annotations
+import asyncio
from pathlib import Path
+from typing import Any
+import pytest
import pytest_asyncio
from cashet import Client
@@ -93,3 +96,62 @@ def counter(x: int) -> int:
async def test_map_empty(self, async_client: AsyncClient) -> None:
refs = await async_client.map(double, [])
assert refs == []
+
+
+class TestBatchCancellation:
+ async def test_failure_cancels_in_flight_siblings(self) -> None:
+ from cashet._batch import execute_batch
+ from cashet.models import Commit, TaskDef, TaskError, TaskStatus
+
+ sibling_started = asyncio.Event()
+ sibling_cancelled = asyncio.Event()
+
+ def failing() -> int:
+ return 0
+
+ def sibling() -> int:
+ return 0
+
+ fail_commit = Commit(
+ hash="f" * 64,
+ task_def=TaskDef(
+ func_hash="x", func_name="failing", func_source="",
+ args_hash="y", args_snapshot=b"",
+ ),
+ output_ref=None,
+ error="boom",
+ status=TaskStatus.FAILED,
+ )
+
+ class StubExecutor:
+ async def submit(self, func: Any, *_: Any) -> tuple[Commit, bool]:
+ if func is sibling:
+ sibling_started.set()
+ try:
+ await asyncio.sleep(30)
+ except asyncio.CancelledError:
+ sibling_cancelled.set()
+ raise
+ raise AssertionError("sibling should have been cancelled")
+ await sibling_started.wait()
+ return fail_commit, False
+
+ keys: list[Any] = ["fail", "sib"]
+ normalized: list[Any] = [
+ (failing, (), {}, True, {}, 0, False, None, None),
+ (sibling, (), {}, True, {}, 0, False, None, None),
+ ]
+ with pytest.raises(TaskError):
+ await execute_batch(
+ order=keys,
+ keys=keys,
+ normalized=normalized,
+ task_refs={},
+ deps={"fail": set(), "sib": set()},
+ executor=StubExecutor(),
+ store=None, # type: ignore[arg-type]
+ serializer=None, # type: ignore[arg-type]
+ max_workers=2,
+ )
+ assert sibling_started.is_set()
+ assert sibling_cancelled.is_set()
diff --git a/tests/test_redis_store.py b/tests/test_redis_store.py
index c0f25c7..f35683a 100644
--- a/tests/test_redis_store.py
+++ b/tests/test_redis_store.py
@@ -580,6 +580,79 @@ def test_delete_by_tags_exact_match(self, redis_store: RedisStore) -> None:
assert deleted == 1
assert redis_store.get_commit("d" * 64) is None
+ def test_find_running_tracks_running_index(self, redis_store: RedisStore) -> None:
+ task_def = TaskDef(
+ func_hash="a" * 64, func_name="f", func_source="",
+ args_hash="b" * 64, args_snapshot=b"",
+ )
+ running = Commit(hash="a" * 64, task_def=task_def, status=TaskStatus.RUNNING)
+ redis_store.put_commit(running)
+ found = redis_store.find_running_by_fingerprint(task_def.fingerprint)
+ assert found is not None and found.hash == "a" * 64
+
+ # transition out of RUNNING clears the running claim
+ running.status = TaskStatus.COMPLETED
+ redis_store.put_commit(running)
+ assert redis_store.find_running_by_fingerprint(task_def.fingerprint) is None
+
+ def test_delete_does_not_drop_blob_when_ref_counter_missing(
+ self, redis_store: RedisStore
+ ) -> None:
+ import redis as _redis
+
+ from cashet.redis_store import _blob_ref_key
+
+ shared = redis_store.put_blob(b"shared-payload")
+ for h, fh in (("a" * 64, "1" * 64), ("b" * 64, "2" * 64)):
+ redis_store.put_commit(
+ Commit(
+ hash=h,
+ task_def=TaskDef(
+ func_hash=fh, func_name="f", func_source="",
+ args_hash="b" * 64, args_snapshot=b"",
+ ),
+ output_ref=shared,
+ status=TaskStatus.COMPLETED,
+ )
+ )
+ # Simulate a lost/inconsistent ref counter for the shared blob.
+ _redis.Redis().delete(_blob_ref_key(shared.hash))
+
+ redis_store.delete_commit("a" * 64)
+
+ # Blob must survive: commit "b" still references it.
+ assert redis_store.get_blob(shared) == b"shared-payload"
+
+ def test_delete_by_tags_no_collision_with_colon_in_key(
+ self, redis_store: RedisStore
+ ) -> None:
+ # Tag key "a:b" (presence) must not collide with tag a="b" (value).
+ colon_key = Commit(
+ hash="a" * 64,
+ task_def=TaskDef(
+ func_hash="1" * 64, func_name="f", func_source="", args_hash="b" * 64,
+ args_snapshot=b"", tags={"a:b": "x"},
+ ),
+ tags={"a:b": "x"},
+ status=TaskStatus.COMPLETED,
+ )
+ value_pair = Commit(
+ hash="b" * 64,
+ task_def=TaskDef(
+ func_hash="2" * 64, func_name="f", func_source="", args_hash="b" * 64,
+ args_snapshot=b"", tags={"a": "b"},
+ ),
+ tags={"a": "b"},
+ status=TaskStatus.COMPLETED,
+ )
+ redis_store.put_commit(colon_key)
+ redis_store.put_commit(value_pair)
+
+ deleted = redis_store.delete_by_tags({"a:b": None})
+ assert deleted == 1
+ assert redis_store.get_commit("a" * 64) is None
+ assert redis_store.get_commit("b" * 64) is not None
+
def test_delete_by_tags_bare_key(self, redis_store: RedisStore) -> None:
for i, env in enumerate(["prod", "staging"]):
task_def = TaskDef(
diff --git a/tests/test_server.py b/tests/test_server.py
index e527bff..61d825f 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -1,6 +1,7 @@
from __future__ import annotations
import base64
+from collections.abc import Iterator
from pathlib import Path
import httpx
@@ -44,6 +45,118 @@ def server_client(tmp_path: Path) -> TestClient:
return TestClient(app)
+class TestServerSizeLimit:
+ def _small_app(self, tmp_path: Path) -> TestClient:
+ client = Client(store_dir=tmp_path / ".cashet")
+ app = create_app(client, tasks={"add": _add}, max_content_length=200)
+ return TestClient(app)
+
+ def test_content_length_over_limit_rejected(self, tmp_path: Path) -> None:
+ tc = self._small_app(tmp_path)
+ payload = {"task": "add", "args": [1, 2], "pad": "x" * 500}
+ response = tc.post("/submit", json=payload)
+ assert response.status_code == 413
+ assert "exceeds" in response.json()["error"]
+
+ def test_chunked_request_without_content_length_rejected(self, tmp_path: Path) -> None:
+ tc = self._small_app(tmp_path)
+
+ def body_chunks() -> Iterator[bytes]:
+ for _ in range(10):
+ yield b"x" * 100 # 1000 bytes total, no Content-Length
+
+ response = tc.post("/submit", content=body_chunks())
+ assert response.status_code == 413
+ assert "exceeds" in response.json()["error"]
+
+ def test_request_within_limit_ok(self, tmp_path: Path) -> None:
+ tc = self._small_app(tmp_path)
+ response = tc.post("/submit", json={"task": "add", "args": [3, 4], "kwargs": {}})
+ assert response.status_code == 200
+
+ def test_unauthenticated_rejected_before_buffering(self, tmp_path: Path) -> None:
+ client = Client(store_dir=tmp_path / ".cashet")
+ app = create_app(
+ client, tasks={"add": _add}, require_token="secret", max_content_length=200
+ )
+ tc = TestClient(app)
+
+ def body_chunks() -> Iterator[bytes]:
+ for _ in range(10):
+ yield b"x" * 100 # oversized, no Content-Length
+
+ # No token: rejected with 401 before the body is buffered (not 413).
+ response = tc.post("/submit", content=body_chunks())
+ assert response.status_code == 401
+
+ def test_authenticated_request_still_processed(self, tmp_path: Path) -> None:
+ client = Client(store_dir=tmp_path / ".cashet")
+ app = create_app(client, tasks={"add": _add}, require_token="secret")
+ tc = TestClient(app)
+ response = tc.post(
+ "/submit",
+ json={"task": "add", "args": [3, 4]},
+ headers={"Authorization": "Bearer secret"},
+ )
+ assert response.status_code == 200
+
+
+class TestServerValidation:
+ def test_log_bad_limit_returns_400(self, server_client: TestClient) -> None:
+ response = server_client.get("/log?limit=abc")
+ assert response.status_code == 400
+ assert "limit" in response.json()["error"]
+
+ def test_gc_bad_older_than_days_returns_400(self, server_client: TestClient) -> None:
+ response = server_client.post("/gc", json={"older_than_days": "soon"})
+ assert response.status_code == 400
+
+ def test_gc_invalid_json_returns_400(self, server_client: TestClient) -> None:
+ response = server_client.post(
+ "/gc", content=b"{not json", headers={"content-type": "application/json"}
+ )
+ assert response.status_code == 400
+
+ def test_gc_empty_body_uses_defaults(self, server_client: TestClient) -> None:
+ response = server_client.post("/gc")
+ assert response.status_code == 200
+ assert "deleted" in response.json()
+
+ def test_internal_error_returns_generic_500(
+ self, server_client: TestClient, monkeypatch: pytest.MonkeyPatch
+ ) -> None:
+ client: Client = server_client.app.state.client # type: ignore[attr-defined]
+
+ def boom() -> dict[str, int]:
+ raise RuntimeError("secret internal detail")
+
+ monkeypatch.setattr(client, "stats", boom)
+ response = server_client.get("/stats")
+ assert response.status_code == 500
+ assert response.json() == {"error": "Internal server error"}
+
+ async def test_async_log_bad_limit_returns_400(self, tmp_path: Path) -> None:
+ client = AsyncClient(store_dir=tmp_path / ".cashet")
+ app = create_async_app(client, tasks={"add": _add})
+ async with httpx.AsyncClient(
+ transport=httpx.ASGITransport(app=app), base_url="http://test"
+ ) as ac:
+ response = await ac.get("/log?limit=abc")
+ assert response.status_code == 400
+ await client.close()
+
+ async def test_async_gc_empty_body_uses_defaults(self, tmp_path: Path) -> None:
+ client = AsyncClient(store_dir=tmp_path / ".cashet")
+ app = create_async_app(client, tasks={"add": _add})
+ async with httpx.AsyncClient(
+ transport=httpx.ASGITransport(app=app), base_url="http://test"
+ ) as ac:
+ response = await ac.post("/gc")
+ assert response.status_code == 200
+ assert "deleted" in response.json()
+ await client.close()
+
+
class TestServerSubmit:
def test_submit_registered_task_with_json_args(self, server_client: TestClient) -> None:
payload = {"task": "add", "args": [3, 4], "kwargs": {}}
diff --git a/tests/test_store.py b/tests/test_store.py
index 13cee6b..576ef7a 100644
--- a/tests/test_store.py
+++ b/tests/test_store.py
@@ -623,6 +623,68 @@ def test_inline_delete_removes_orphans(self, store_dir: Path) -> None:
store.delete_commit(commit.hash)
assert store.blob_exists(ref.hash) is False
+ def test_find_by_fingerprint_survives_locked_access_bump(self, tmp_path: Path) -> None:
+ from cashet.models import Commit, TaskDef, TaskStatus
+ from cashet.store import _SQLiteStoreCore
+
+ root = tmp_path / ".cashet"
+ core = _SQLiteStoreCore(root)
+ task_def = TaskDef(
+ func_hash="a", func_name="f", func_source="", args_hash="b", args_snapshot=b""
+ )
+ commit = Commit(
+ hash="c" * 64,
+ task_def=task_def,
+ output_ref=core.put_blob(b"v"),
+ status=TaskStatus.COMPLETED,
+ )
+ core.put_commit(commit)
+
+ conn = core._connect()
+ conn.execute("PRAGMA busy_timeout=100")
+ blocker = sqlite3.connect(str(root / "meta.db"), isolation_level=None)
+ blocker.execute("PRAGMA journal_mode=WAL")
+ blocker.execute("BEGIN IMMEDIATE")
+ try:
+ found = core.find_by_fingerprint(task_def.fingerprint)
+ assert found is not None
+ assert found.hash == commit.hash
+ finally:
+ blocker.execute("ROLLBACK")
+ blocker.close()
+
+ def test_evict_survives_vacuum_failure(
+ self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+ ) -> None:
+ from cashet.models import Commit, TaskDef, TaskStatus
+ from cashet.store import _SQLiteStoreCore
+
+ root = tmp_path / ".cashet"
+ core = _SQLiteStoreCore(root)
+ task_def = TaskDef(
+ func_hash="a", func_name="f", func_source="", args_hash="b", args_snapshot=b""
+ )
+ commit = Commit(
+ hash="c" * 64,
+ task_def=task_def,
+ output_ref=core.put_blob(b"v"),
+ status=TaskStatus.COMPLETED,
+ )
+ core.put_commit(commit)
+ core._connect().execute(
+ "UPDATE commits SET last_accessed_at = ? WHERE hash = ?",
+ ("2000-01-01T00:00:00+00:00", commit.hash),
+ )
+
+ def boom() -> None:
+ raise sqlite3.OperationalError("database is locked")
+
+ monkeypatch.setattr(core, "_vacuum", boom)
+
+ deleted = core.evict(datetime.now(UTC))
+ assert deleted == 1
+ assert core.get_commit(commit.hash) is None
+
def test_inline_stats(self, store_dir: Path) -> None:
store = SQLiteStore(store_dir)
store.put_blob(b"small")
diff --git a/uv.lock b/uv.lock
index c084fac..8f80400 100644
--- a/uv.lock
+++ b/uv.lock
@@ -26,7 +26,7 @@ wheels = [
[[package]]
name = "cashet"
-version = "0.4.4"
+version = "0.4.5"
source = { editable = "." }
dependencies = [
{ name = "click" },