diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8f3183b..fa78f73 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -42,13 +42,14 @@ jobs: - "3.14" django-version: - "6.0" + - "6.1" client-library-version: - "latest" include: # Pre-release client libraries (redis-py master, valkey-py main) - python-version: "3.14" - django-version: "6.0" + django-version: "6.1" client-library-version: "dev" # Pre-release Django @@ -84,6 +85,10 @@ jobs: - name: Run backend tests run: uv run pytest tests/cache/ -n auto + env: + # Guard against pyc write races on first concurrent import + # (intermittent xdist collection ImportError on fresh checkouts). + PYTHONDONTWRITEBYTECODE: "1" - name: Upload coverage to Codecov if: matrix.python-version == '3.14' && matrix.django-version == '6.0' && matrix.client-library-version == 'latest' @@ -94,6 +99,10 @@ jobs: test-admin: runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + django-version: ["6.0", "6.1"] steps: - uses: actions/checkout@v7 @@ -106,11 +115,37 @@ jobs: - name: Install dependencies run: uv sync --group dev - - name: Install Django - run: uv pip install "Django~=6.0.0" + - name: Install Django ${{ matrix.django-version }} + run: uv pip install "Django~=${{ matrix.django-version }}.0" - name: Run admin tests run: uv run pytest tests/admin/ -n auto + env: + PYTHONDONTWRITEBYTECODE: "1" + + test-glide-adapter: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + + - name: Install uv + uses: astral-sh/setup-uv@v7 + + # glide ships no free-threaded wheel, so this runs on GIL cp314 only. + - name: Set up Python + run: uv python install 3.14 + + - name: Install dependencies + run: uv sync --group dev --extra valkey-glide + + # The cache fixtures skip an adapter whose client library is missing + # (_adapter_library_available in tests/fixtures/cache.py), and no other + # job installs the glide extra, so this is the only place the shared + # suite reaches the glide adapter. + - name: Run backend tests with the glide adapter + run: uv run pytest tests/cache/ -n auto + env: + PYTHONDONTWRITEBYTECODE: "1" build-pure-wheel: name: Build pure-Python wheel (django-cachex) @@ -163,10 +198,11 @@ jobs: with: package-dir: crates/django-cachex-redis-rs env: + # No CIBW_ENABLE: cibuildwheel 4.0 dropped the cpython-freethreading + # group and builds 3.14t whenever CIBW_BUILD asks for it. CIBW_BUILD: ${{ matrix.cibw_build }} - CIBW_ENABLE: cpython-freethreading CIBW_ARCHS: ${{ matrix.cibw_archs }} - # Linux builds run inside manylinux containers — install Rust there. + # Linux builds run inside manylinux containers, so install Rust there. # macOS and Windows runners have rustup preinstalled. CIBW_BEFORE_ALL_LINUX: | curl --proto =https --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y --default-toolchain stable --profile minimal @@ -210,6 +246,8 @@ jobs: - name: Run cache tests against built wheels run: uv run --no-sync pytest tests/cache/ -n auto + env: + PYTHONDONTWRITEBYTECODE: "1" smoke-test-wheel-freethreaded: name: Smoke-test cp314t rust wheel @@ -239,7 +277,7 @@ jobs: uv pip install --no-deps wheelhouse/django_cachex_redis_rs-*-cp314-cp314t-*.whl - name: Verify FT wheel imports - run: uv run --no-sync python -c "import sys; from django_cachex.adapters import _redis_rs; assert not sys._is_gil_enabled(), 'GIL is enabled — not running on free-threaded build'; print('OK', _redis_rs, sys.version)" + run: uv run --no-sync python -c "import sys; from django_cachex.adapters import _redis_rs; assert not sys._is_gil_enabled(), 'GIL is enabled, not running on free-threaded build'; print('OK', _redis_rs, sys.version)" smoke-test-pure-only: name: Smoke-test opt-out (pure wheel only, no rust package) diff --git a/.github/workflows/dependabot-automerge.yml b/.github/workflows/dependabot-automerge.yml index bc15d5a..ccff4ed 100644 --- a/.github/workflows/dependabot-automerge.yml +++ b/.github/workflows/dependabot-automerge.yml @@ -10,9 +10,41 @@ jobs: automerge: runs-on: ubuntu-latest if: github.actor == 'dependabot[bot]' + timeout-minutes: 30 steps: - - name: Enable auto-merge - run: gh pr merge --auto --squash "$PR_URL" + # Without required status checks on main, `gh pr merge --auto` merges + # immediately, red CI included. Wait for every workflow run on this PR + # head instead, and merge only once they have all succeeded. + # + # Poll check suites, not check runs: a `needs:`-gated job gets no check + # run until its dependencies finish, so a poll landing in that gap sees + # nothing pending and merges before the wheel tests have even started. + # A suite stays in progress across the gap. This job's own suite is + # excluded so it does not wait on itself forever. + - name: Wait for CI + run: | + sha=$(gh pr view "$PR_URL" --json headRefOid --jq .headRefOid) + own=$(gh api "repos/$GITHUB_REPOSITORY/actions/runs/$GITHUB_RUN_ID" --jq .check_suite_id) + while true; do + suites=$(gh api "repos/$GITHUB_REPOSITORY/commits/$sha/check-suites?per_page=100" \ + --jq "[.check_suites[] | select(.id != $own) | {status, conclusion}]" || true) + [ -n "$suites" ] || suites='[]' + bad=$(jq -c '[.[] | select(.conclusion != null and ([.conclusion] | inside(["success", "neutral", "skipped"]) | not))]' <<<"$suites") + if [ "$(jq 'length' <<<"$bad")" -gt 0 ]; then + echo "CI did not pass: $bad" + exit 1 + fi + if [ "$(jq 'length' <<<"$suites")" -gt 0 ] && [ "$(jq '[.[] | select(.status != "completed")] | length' <<<"$suites")" -eq 0 ]; then + break + fi + sleep 30 + done + env: + PR_URL: ${{ github.event.pull_request.html_url }} + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + + - name: Merge + run: gh pr merge --squash "$PR_URL" env: PR_URL: ${{ github.event.pull_request.html_url }} GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index c6466e6..84daf52 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -66,14 +66,15 @@ jobs: with: package-dir: crates/django-cachex-redis-rs env: + # No CIBW_ENABLE: cibuildwheel 4.0 dropped the cpython-freethreading + # group and builds 3.14t whenever CIBW_BUILD asks for it. CIBW_BUILD: ${{ matrix.cibw_build }} - CIBW_ENABLE: cpython-freethreading CIBW_ARCHS: ${{ matrix.cibw_archs }} CIBW_BEFORE_ALL_LINUX: | curl --proto =https --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y --default-toolchain stable --profile minimal CIBW_ENVIRONMENT_LINUX: PATH="$HOME/.cargo/bin:$PATH" - # sdist is a single artifact — only build it once on the Linux x86_64 leg. + # sdist is a single artifact, so only build it once on the Linux x86_64 leg. - name: Build Rust sdist if: matrix.label == 'linux-x86_64' uses: PyO3/maturin-action@v1 diff --git a/.github/workflows/tag.yml b/.github/workflows/tag.yml index 61f1bde..7c73156 100644 --- a/.github/workflows/tag.yml +++ b/.github/workflows/tag.yml @@ -23,7 +23,7 @@ jobs: if curl -s "https://pypi.org/pypi/django-cachex/$CURRENT_VERSION/json" | grep -q '"version"'; then echo "should_release=false" >> $GITHUB_OUTPUT - echo "Version $CURRENT_VERSION already on PyPI — nothing to do" + echo "Version $CURRENT_VERSION already on PyPI, nothing to do" exit 0 fi @@ -33,7 +33,7 @@ jobs: git fetch --tags if git rev-parse "v$CURRENT_VERSION" >/dev/null 2>&1; then echo "should_release=false" >> $GITHUB_OUTPUT - echo "Tag v$CURRENT_VERSION already exists — delete it manually to re-release" + echo "Tag v$CURRENT_VERSION already exists; delete it manually to re-release" exit 0 fi @@ -70,6 +70,10 @@ jobs: - name: Run tests run: uv run pytest tests/cache/ -n auto + env: + # Guard against pyc write races on first concurrent import + # (intermittent xdist collection ImportError on fresh checkouts). + PYTHONDONTWRITEBYTECODE: "1" tag-and-trigger: needs: [check-version, test] @@ -90,8 +94,8 @@ jobs: git tag "v$VERSION" git push origin "v$VERSION" - # GITHUB_TOKEN-pushed tags don't trigger downstream workflows — explicit - # dispatch is required. + # GITHUB_TOKEN-pushed tags don't trigger downstream workflows, so an + # explicit dispatch is required. - name: Trigger publish workflow run: gh workflow run publish.yml -f version=$VERSION -R ${{ github.repository }} env: diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 873f050..41a22bf 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -9,7 +9,7 @@ default_language_version: repos: - repo: https://github.com/pre-commit/pre-commit-hooks - rev: v5.0.0 + rev: v6.0.0 hooks: - id: check-ast - id: check-case-conflict @@ -54,6 +54,15 @@ repos: entry: uv run ruff format language: system pass_filenames: false + # `-` makes djLint take its file list from `files` in [tool.djlint]; it + # ignores anything pre-commit would append, so don't pass filenames. + # PYTHON_GIL=1 because djLint's compiled module initialises unsafely on + # the free-threaded build and dies with a NameError on _LINE_PATTERN. + - id: djlint + name: djlint + entry: env PYTHON_GIL=1 uv run djlint - --lint + language: system + pass_filenames: false - repo: local hooks: diff --git a/README.md b/README.md index e70150a..9cbebe8 100644 --- a/README.md +++ b/README.md @@ -36,8 +36,8 @@ CACHES = { - Cache stampede prevention (TTL-based XFetch). - Two composite backends: `StreamCache` (cross-pod stream-synchronized in-memory cache) and `TieredCache` (L1/L2 with TTL propagation). - Django `LocMemCache` and `DatabaseCache` extensions with the same data-structure ops and admin support. -- Optional Rust I/O driver (PyO3 + tokio + redis-rs) under the same `RespCache` API. Free-threaded CPython (3.14t) supported. -- Optional `valkey-glide` adapter: Valkey's official Rust-cored client, exposed as `ValkeyGlideCache`. +- Optional Rust I/O driver (PyO3 + tokio + redis-rs) under the same `RespCache` API. Free-threaded CPython (3.14t) supported. Experimental. +- Optional `valkey-glide` adapter: Valkey's official Rust-cored client, exposed as `ValkeyGlideCache`. Experimental. - Django admin UI for browsing keys, inspecting values, editing, and flushing. See below. ## Cache Admin @@ -78,15 +78,18 @@ Full documentation at [oliverhaas.github.io/django-cachex](https://oliverhaas.gi - Valkey 7.0+ or Redis 6.0+ on the server (the admin's compare-and-swap edits use `SET ... KEEPTTL`, which lands in Redis 6.0) -The Rust I/O driver is optional. To opt in, install with the `redis-rs` +The Rust I/O driver is optional and experimental: interfaces and +behavior may still change, and it has seen less production testing than +the redis-py/valkey-py paths. To opt in, install with the `redis-rs` extra (`pip install django-cachex[redis-rs]`); this pulls in the `django-cachex-redis-rs` companion package. Prebuilt wheels are published for Linux x86_64, Linux aarch64, macOS arm64, and Windows amd64, on both cp314 and cp314t (free-threaded). Without the extra, the `RedisRsCache` backends are unavailable but everything else works. -The `valkey-glide` adapter is also optional. Install with the -`valkey-glide` extra (`pip install django-cachex[valkey-glide]`) to enable +The `valkey-glide` adapter is also optional and experimental, with the +same caveats. Install with the `valkey-glide` extra +(`pip install django-cachex[valkey-glide]`) to enable `ValkeyGlideCache`; it pulls in `valkey-glide-sync` and `valkey-glide`, the official Rust-cored Valkey client. cp314 GIL only; no free-threaded wheels yet. Cluster is supported via `ValkeyGlideClusterCache`; Sentinel diff --git a/benchmarks/README.md b/benchmarks/README.md index 3c81a16..2e29492 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -3,46 +3,46 @@ Throughput and memory comparison across cache adapter/parser/serializer/compressor combos. -Not part of the regular test suite — runs separately because it spins up its -own Redis and Valkey containers and is slow on purpose (timing accuracy +Not part of the regular test suite. It runs separately because it spins up +its own Redis and Valkey containers and is slow on purpose (timing accuracy depends on letting workloads run). ## What gets compared **Adapters** (with default pickle serializer): -- `redis-py` — pure-Python parser -- `redis-py+hiredis` — C parser -- `valkey-py` — pure-Python parser -- `valkey-py+libvalkey` — C parser -- `redis-rs` — our Rust extension adapter (PyO3 bindings around the +- `redis-py`: pure-Python parser +- `redis-py+hiredis`: C parser +- `valkey-py`: pure-Python parser +- `valkey-py+libvalkey`: C parser +- `redis-rs`: our Rust extension adapter (PyO3 bindings around the `redis-rs` crate, multiplexed Tokio transport) -- `valkey-glide` — `valkey/valkey-glide` Python wheels (also Rust-cored, +- `valkey-glide`: `valkey/valkey-glide` Python wheels (also Rust-cored, but with a thread-pool transport instead of multiplexed). Only available - on cp314 GIL — no cp314t (free-threaded) wheels yet. -- `django (builtin)` — Django's official built-in `django.core.cache.backends.redis.RedisCache` - (since 4.0). Not the third-party `jazzband/django-redis` package — that one is + on cp314 GIL. No cp314t (free-threaded) wheels yet. +- `django (builtin)`: Django's official built-in `django.core.cache.backends.redis.RedisCache` + (since 4.0). Not the third-party `jazzband/django-redis` package, which is unrelated. Included as an external reference point. **Serializers** (with `redis-rs` adapter, since adapter overhead is smallest there): -- `pickle` — stdlib default -- `json` — Django's `DjangoJSONEncoder` -- `msgpack` — pure-Python `msgpack` -- `orjson` — Rust-backed JSON -- `ormsgpack` — Rust-backed MessagePack +- `pickle`: stdlib default +- `json`: Django's `DjangoJSONEncoder` +- `msgpack`: pure-Python `msgpack` +- `orjson`: Rust-backed JSON +- `ormsgpack`: Rust-backed MessagePack -**Compressors** — two views, both with `redis-rs` + `pickle`: +**Compressors** get two views, both with `redis-rs` + `pickle`: -- *Macro* (`test_compressors_macro`) — end-to-end Django cache ops on a +- *Macro* (`test_compressors_macro`): end-to-end Django cache ops on a ~14 KiB queryset-shaped payload. Captures the cost of compress/decompress against the savings from sending fewer bytes over the wire. -- *Micro* (`test_compressors_micro`) — pure compress/decompress in a tight +- *Micro* (`test_compressors_micro`): pure compress/decompress in a tight loop. Reports output ratio and MB/s. No adapter, no container. Compressor candidates: `none`, `zlib`, `gzip`, `lzma`, `lz4`, `zstd`. -**Request cycle** (`test_adapters_request_cycle`) — same workload as +**Request cycle** (`test_adapters_request_cycle`) runs the same workload as `test_adapters_sync`, but each cache op runs inside a real Django request cycle: `Client().get(url)` → URL resolve → `CommonMiddleware` → view function → response → `request_finished` signal. The view in [urls.py](urls.py) does @@ -51,21 +51,21 @@ exactly one cache op per request, so ops/sec is on the same scale as ids are suffixed with `#req` in the final summary so the request-cycle rows sit next to their direct counterparts. -**ASGI** (`test_adapters_asgi`) — full-stack benchmark in the shape of +**ASGI** (`test_adapters_asgi`) is a full-stack benchmark in the shape of [`django-vcache`'s `bench_compare.py`](https://gitlab.com/glitchtip/django-vcache/-/blob/main/bench_compare.py): - Spawns a real **`granian`** ASGI server (4 workers) per adapter - Drives load with **`httpx.AsyncClient`** (100 concurrent connections, 20 second duration by default; bump `ASGI_CONCURRENCY` / `ASGI_DURATION_S` in `test_throughput.py` for hero numbers) -- Each request hits `/bench/mixed/`, which does six async cache ops — +- Each request hits `/bench/mixed/`, which does six async cache ops: `aget`, `aget_many`(3 keys), `aset`, `aset` (large, ~2.5 KiB to trigger compression), `aincr`, `aget` (large) - Samples server RSS and Valkey/Redis `connected_clients` every 5 seconds during the run; reports init / peak / final / settled (post-cooldown) This is the only benchmark that reliably surfaces connection-pool growth -under realistic load — sync direct, async direct, and request-cycle tests +under realistic load. The sync direct, async direct, and request-cycle tests all show stable connection counts because the workload is too well-behaved to stress the pool. The ASGI benchmark hits the pool from four worker processes simultaneously, which is enough to expose any per-call client @@ -78,13 +78,13 @@ inside a Docker container with `--cap-add NET_ADMIN` and apply interface. Without latency the directional ranking is the same; with it, the magnitude grows dramatically. -**Async** — two views via `aget` / `aset` / `aget_many` / etc.: +**Async** gets two views via `aget` / `aset` / `aget_many` / etc.: -- *Serial* (`test_adapters_async_serial`) — `await cache.aget(...)` one op at +- *Serial* (`test_adapters_async_serial`): `await cache.aget(...)` one op at a time. Direct comparison with sync; the gap reveals asyncio-loop overhead and, for backends without native async, the cost of Django's `sync_to_async` fallback. Ids suffixed with `#async`. -- *Concurrent* (`test_adapters_async_concurrent`) — `asyncio.gather` of +- *Concurrent* (`test_adapters_async_concurrent`): `asyncio.gather` of `ASYNC_CONCURRENCY` (default 50) ops in flight. Stresses the connection pool: peak connections jump to roughly the concurrency level for backends with native async + per-op pool checkout. The intended use is also to @@ -95,9 +95,9 @@ the magnitude grows dramatically. ## What gets measured Adapter / serializer / compressor-macro / request-cycle tests run a -seven-phase workload — `get`, `get-miss`, `set`, `mget` (10-key batch), -`mset` (10-key batch), `incr`, `delete` — `N_OPS=1000` operations per phase, -repeated `K_RUNS=10` times. +seven-phase workload: `get`, `get-miss`, `set`, `mget` (10-key batch), +`mset` (10-key batch), `incr`, `delete`. Each phase runs `N_OPS=1000` +operations, repeated `K_RUNS=10` times. Per-phase timings are reported as median ms and ops/sec across runs. Per-run metrics include Python peak memory (`tracemalloc`) and server memory delta @@ -131,8 +131,9 @@ uv run pytest benchmarks/test_throughput.py::test_adapters_asgi -c b uv run pytest 'benchmarks/test_throughput.py::test_adapters_sync[redis-rs]' -c benchmarks/pytest.ini ``` -`test_compressors_micro` is the only test that doesn't need Docker — useful -for quick algorithm comparisons on a laptop without containers running. +`test_compressors_micro` is the only test that doesn't need Docker, which +makes it useful for quick algorithm comparisons on a laptop without +containers running. A summary table prints at the end of the session. @@ -142,7 +143,7 @@ A summary table prints at the end of the session. - **Redis vs Valkey.** Each adapter is paired with its natural server (redis-py → redis, valkey-py / redis-rs / valkey-glide → valkey, django (builtin) → redis). Cross-pairings are intentionally not in the - matrix — both servers speak the same protocol, so the comparison is + matrix: both servers speak the same protocol, so the comparison is mostly a wash. - **Warmup.** Each phase runs an untimed pass before the timed runs to prime connections, server keyspace, and lazy serializer state. @@ -155,7 +156,7 @@ Snapshot of the adapter matrix on a Ryzen 9 5950X / 32 GiB / Linux 6.17, cp314 GIL, Django 6.0, all servers in local Docker. Numbers shift run to run; ordering is what matters. -### Sync direct (`test_adapters_sync`) — ops/sec +### Sync direct (`test_adapters_sync`) in ops/sec | Adapter | get | get-miss | set | mget | mset | incr | delete | py-mem KiB | | -------------------- | -----: | -------: | -----: | ----: | ----: | -----: | -----: | ---------: | @@ -167,7 +168,7 @@ run; ordering is what matters. | **valkey-glide** | 6,982 | 8,783 | 6,982 | 2,030 | 1,916 | 8,901 | 3,994 | 29 | | django (builtin) | 2,307 | 2,459 | 2,269 | 1,501 | 1,347 | 1,947 | 1,199 | 51 | -### Django request cycle (`test_adapters_request_cycle`, `#req`) — ops/sec +### Django request cycle (`test_adapters_request_cycle`, `#req`) in ops/sec One cache op per request through the full middleware/URL/view path. @@ -181,7 +182,7 @@ One cache op per request through the full middleware/URL/view path. | **valkey-glide** | 1,237 | 1,863 | 1,753 | 1,023 | 994 | 1,816 | 1,838 | | django (builtin) | 861 | 1,177 | 1,133 | 859 | 806 | 1,016 | 1,169 | -### Async serial (`test_adapters_async_serial`, `#async`) — ops/sec +### Async serial (`test_adapters_async_serial`, `#async`) in ops/sec One `await` at a time. @@ -195,9 +196,9 @@ One `await` at a time. | **valkey-glide** | 3,300 | 3,738 | 3,297 | 1,725 | 1,678 | 3,693 | 1,735 | | django (builtin) | 1,948 | 1,995 | 1,929 | 199 | 197 | 989 | 992 | -### Async concurrent at 50 (`test_adapters_async_concurrent`, `#async50`) — ops/sec +### Async concurrent at 50 (`test_adapters_async_concurrent`, `#async50`) in ops/sec -`asyncio.gather` of 50 ops in flight — the workload most Django ASGI apps +`asyncio.gather` of 50 ops in flight, the workload most Django ASGI apps actually generate. | Adapter | get | get-miss | set | mget | mset | incr | delete | conns peak | @@ -213,7 +214,7 @@ actually generate. ### ASGI full-stack (`test_adapters_asgi`) `granian` (4 workers) + `httpx` (100 concurrent, 20 s). Each request runs -six async cache ops — the shape closest to real production load. +six async cache ops, the shape closest to real production load. | Adapter | req/s | avg ms | p99 ms | RSS peak (MiB) | conns peak | conns settled | | -------------------- | ----: | -----: | -----: | -------------: | ---------: | ------------: | @@ -227,12 +228,12 @@ six async cache ops — the shape closest to real production load. ### Takeaways -- **Sync direct.** `redis-rs` leads on every phase — **~10k get / 11k set +- **Sync direct.** `redis-rs` leads on every phase at **~10k get / 11k set / 13k incr / 6k delete** ops/sec, ~4× the fastest pure-Python adapter. `valkey-glide` is second on single-key ops (`get`/`set`/`incr` ~7–9k) but trails `redis-rs` on multi-key and `delete`. Both Rust adapters use ~4× less Python memory than the pure-Python adapters. -- **Django request cycle.** Both Rust adapters lead the table — +- **Django request cycle.** Both Rust adapters lead the table: `redis-rs` ~1.4–2.1k ops/sec, `valkey-glide` ~1.2–1.9k. Earlier versions of these adapters failed this benchmark entirely because ``close()`` was tearing down the connection on every @@ -241,16 +242,16 @@ six async cache ops — the shape closest to real production load. Python adapters on every phase (6.9k get, 8.3k incr). `valkey-glide` is also faster than Python (~1.5–1.8×) but trails `redis-rs`. - **Async concurrent (50 in flight).** `redis-rs` peaks at **22k get / - 36k get-miss / 39k incr** ops/sec — **9–18×** the fastest Python + 36k get-miss / 39k incr** ops/sec, which is **9–18×** the fastest Python adapter. `valkey-glide` is half that rate but still **~4×** ahead of the Python adapters. - **ASGI full-stack** (granian × 4 workers, httpx × 100 concurrent). - Both Rust adapters lead on req/s — `redis-rs` 607 (164 ms avg, 1.15 s - p99), `valkey-glide` 553 — and use **fewer connections** than every - Python adapter (`redis-rs` 115, `valkey-glide` 172, vs Python ~210). + Both Rust adapters lead on req/s: `redis-rs` 607 (164 ms avg, 1.15 s + p99) and `valkey-glide` 553. They also use **fewer connections** than + every Python adapter (`redis-rs` 115, `valkey-glide` 172, vs Python ~210). Process-wide client sharing (added after this benchmark first surfaced 1,111 / 4,577 conn counts) brought the Rust adapters in line with the multiplexed-transport claim. - **Connection stability.** Across every shape the cachex path keeps - `Δ` at 0 between phases — no per-phase connection leaks on any - adapter. + `Δ` at 0 between phases, so there are no per-phase connection leaks on + any adapter. diff --git a/benchmarks/asgi.py b/benchmarks/asgi.py index d05bab9..9b9fe7c 100644 --- a/benchmarks/asgi.py +++ b/benchmarks/asgi.py @@ -1,4 +1,4 @@ -"""ASGI entrypoint for the ASGI benchmark — granian imports this.""" +"""ASGI entrypoint for the ASGI benchmark, imported by granian.""" import os diff --git a/benchmarks/asgi_settings.py b/benchmarks/asgi_settings.py index 4a318e5..17d6805 100644 --- a/benchmarks/asgi_settings.py +++ b/benchmarks/asgi_settings.py @@ -3,9 +3,9 @@ Reads CACHES from environment variables so the parent benchmark process can parametrize the backend without writing per-adapter settings modules: -- ``BENCH_CACHE_BACKEND`` — dotted path of the cache backend class. -- ``BENCH_CACHE_LOCATION`` — Redis/Valkey URL. -- ``BENCH_CACHE_OPTIONS_JSON`` — optional, JSON-encoded ``OPTIONS`` dict. +- ``BENCH_CACHE_BACKEND``: dotted path of the cache backend class. +- ``BENCH_CACHE_LOCATION``: Redis/Valkey URL. +- ``BENCH_CACHE_OPTIONS_JSON``: optional, JSON-encoded ``OPTIONS`` dict. """ import json diff --git a/benchmarks/configs.py b/benchmarks/configs.py index 69686ab..035a2be 100644 --- a/benchmarks/configs.py +++ b/benchmarks/configs.py @@ -13,7 +13,7 @@ class AdapterConfig: id: str backend: str options: dict - server: str # "redis" or "valkey" — picks which container URL to use + server: str # either "redis" or "valkey", picking which container URL to use @dataclass(frozen=True) @@ -29,7 +29,7 @@ class CompressorConfig: # Adapters we want to compare. The "server" field decides which container URL -# the runner connects to — we keep redis-py paired with redis-server and +# the runner connects to. We keep redis-py paired with redis-server and # valkey-py paired with valkey-server because that's the natural pairing. ADAPTER_CONFIGS: tuple[AdapterConfig, ...] = ( AdapterConfig( @@ -82,7 +82,7 @@ class CompressorConfig: ), # Django's official built-in Redis cache backend (added in Django 4.0, # `django.core.cache.backends.redis.RedisCache`). Not to be confused with - # the third-party `jazzband/django-redis` package — that one ships under + # the third-party `jazzband/django-redis` package, which ships under # `django_redis.cache.RedisCache` and is unrelated. # # Useful as an external reference point: its `get_client()` instantiates diff --git a/benchmarks/runner.py b/benchmarks/runner.py index 7547c7b..625777c 100644 --- a/benchmarks/runner.py +++ b/benchmarks/runner.py @@ -28,7 +28,7 @@ from collections.abc import Callable, Iterable -# Workload sizing — kept here so all phases share the knob. +# Workload sizing, kept here so all phases share the knob. N_OPS = 1000 K_RUNS = 10 WARMUP_KEYS = 100 @@ -127,7 +127,7 @@ def rss_growth_mb(self) -> float: @dataclass class MicroResult: - """Pure compress/decompress numbers — no adapter, no network, no Django.""" + """Pure compress/decompress numbers: no adapter, no network, no Django.""" compressor_id: str input_bytes: int @@ -177,7 +177,7 @@ def build_caches( def _build_payload() -> dict[str, Any]: - # ~150 bytes pickled — small enough that serializer cost dominates over network. + # ~150 bytes pickled, small enough that serializer cost dominates over network. return { "id": 12345, "name": "benchmark-item", @@ -189,7 +189,7 @@ def _build_payload() -> dict[str, Any]: def _build_payload_large() -> list[dict[str, Any]]: - # ~14 KiB pickled — queryset-shaped, well above the 256 B compression + # ~14 KiB pickled and queryset-shaped, well above the 256 B compression # threshold. Used for compressor benchmarks where a small payload would # bypass compression entirely. return [ @@ -335,7 +335,8 @@ def _open_info_client(location: str) -> redis.Redis: Used regardless of the cache backend under test, so backends without an ``info()`` method (Django's built-in ``RedisCache``) still report memory - and connection metrics. Works against Valkey too — same RESP protocol. + and connection metrics. Works against Valkey too, which speaks the same + RESP protocol. """ return redis.Redis.from_url(location) @@ -513,7 +514,7 @@ def run_request_cycle_benchmark( ``request_started`` / ``request_finished`` signals). Throughput is reported in ops/sec on the same scale as ``run_benchmark``, - so the two are directly comparable — the gap is the per-request overhead + so the two are directly comparable. The gap is the per-request overhead Django itself adds when cache work happens inside a view. """ from benchmarks import urls as benchmark_urls @@ -601,7 +602,7 @@ def _total_rss_kb(parent_pid: int) -> float: if ps is None: return total try: - result = subprocess.run( # noqa: S603 — args are constants, no user input + result = subprocess.run( # noqa: S603 (args are constants, no user input) [ps, "--ppid", str(parent_pid), "-o", "rss="], capture_output=True, text=True, @@ -629,7 +630,7 @@ def _wait_for_port(host: str, port: int, timeout_s: float = 15.0) -> bool: return False -def run_asgi_benchmark( # noqa: C901, PLR0915 — orchestrates many phases in one function for readability +def run_asgi_benchmark( # noqa: C901, PLR0915 (orchestrates many phases in one function for readability) adapter: AdapterConfig, serializer: SerializerConfig, location: str, @@ -647,12 +648,12 @@ def run_asgi_benchmark( # noqa: C901, PLR0915 — orchestrates many phases in o ASGI server, hit it with a configurable number of concurrent HTTP clients for a fixed duration, sample peak server RSS and Valkey/Redis ``connected_clients`` along the way. The view does six async cache ops - per request — get / aget_many / aset / aset (large) / aincr / aget - (large) — to exercise the full backend surface in one workload. + per request: get / aget_many / aset / aset (large) / aincr / aget + (large). That exercises the full backend surface in one workload. Latency simulation (e.g. ``tc qdisc add dev eth0 root netem delay 1ms``) is the missing ingredient versus django-vcache's claims about - `RedisCache` connection growth — without RTT, sync_to_async threads + `RedisCache` connection growth. Without RTT, sync_to_async threads finish too quickly to pile up. Run this benchmark inside a Docker container with ``--cap-add NET_ADMIN`` and apply ``netem`` against the Valkey/Redis interface to reproduce vcache's numbers. @@ -672,7 +673,7 @@ def run_asgi_benchmark( # noqa: C901, PLR0915 — orchestrates many phases in o info_client = _open_info_client(location) info_client.flushdb() - proc = subprocess.Popen( # noqa: S603 — args are sys.executable + constants + proc = subprocess.Popen( # noqa: S603 (args are sys.executable + constants) [ sys.executable, "-m", @@ -891,7 +892,7 @@ def run_async_benchmark( ) -> BenchmarkResult: """Async equivalent of ``run_benchmark``. - ``concurrency=1`` is serial async — every ``aget`` is awaited before the + ``concurrency=1`` is serial async: every ``aget`` is awaited before the next is issued, giving a one-to-one comparison with the sync benchmark (the gap reveals overhead of the async path: native ``aget`` vs ``sync_to_async`` fallback). @@ -925,7 +926,7 @@ def run_compressor_micro( ) -> MicroResult | None: """Pure compress/decompress throughput on the large benchmark payload. - No adapter, no Django, no network — just the algorithm against a fixed + No adapter, no Django, and no network: just the algorithm against a fixed blob. Returns None for the no-compression baseline. """ if compressor.dotted_path is None: @@ -1061,7 +1062,7 @@ def format_asgi_table(results: Iterable[AsgiResult]) -> str: def format_micro_table(results: Iterable[MicroResult]) -> str: - """Table of compressor micro results — absolute MB/s plus output ratio.""" + """Table of compressor micro results: absolute MB/s plus output ratio.""" results = list(results) if not results: return "(no micro results)" diff --git a/benchmarks/test_throughput.py b/benchmarks/test_throughput.py index 4eaab5f..318832b 100644 --- a/benchmarks/test_throughput.py +++ b/benchmarks/test_throughput.py @@ -2,16 +2,16 @@ Parametrized tests: -- ``test_adapters_sync`` — fixed pickle serializer, varies the adapter. +- ``test_adapters_sync`` fixes the pickle serializer and varies the adapter. Isolates the adapter/parser/connection stack. -- ``test_serializers`` — fixed redis-rs adapter, varies the serializer. +- ``test_serializers`` fixes the redis-rs adapter and varies the serializer. Isolates serializer cost (adapter overhead is minimal at that point). -- ``test_compressors_macro`` — fixed redis-rs + pickle, varies the +- ``test_compressors_macro`` fixes redis-rs + pickle and varies the compressor on a large payload. End-to-end ops/sec showing the compress cost vs network savings tradeoff in real cache calls. -- ``test_compressors_micro`` — pure compress/decompress in-process, no - adapter or container. Reports ratio and MB/s for each compressor. -- ``test_adapters_request_cycle`` — same shape as ``test_adapters_sync`` but +- ``test_compressors_micro`` runs pure compress/decompress in-process, with + no adapter or container. Reports ratio and MB/s for each compressor. +- ``test_adapters_request_cycle`` has the same shape as ``test_adapters_sync`` but every cache op is wrapped in a real Django request cycle (URL resolve, middleware, view dispatch, signals). Direct comparison reveals the per-request overhead Django adds on top of the cache call itself. @@ -41,7 +41,7 @@ ASYNC_CONCURRENCY = 50 -# ASGI benchmark knobs — kept short by default so the suite stays runnable +# ASGI benchmark knobs, kept short by default so the suite stays runnable # in CI; bump these manually for hero numbers. ASGI_DURATION_S = 20 ASGI_CONCURRENCY = 100 @@ -145,7 +145,7 @@ def test_adapters_request_cycle(adapter, server_url, results, capsys) -> None: @pytest.mark.parametrize("adapter", ADAPTER_CONFIGS, ids=lambda c: c.id) def test_adapters_asgi(adapter, server_url, asgi_results, capsys) -> None: - """Full-stack ASGI benchmark — granian + httpx + 6 cache ops per request. + """Full-stack ASGI benchmark with granian + httpx and 6 cache ops per request. Mirrors django-vcache's ``bench_compare.py`` shape so numbers are directly comparable. To reproduce vcache's connection-leak claim diff --git a/benchmarks/urls.py b/benchmarks/urls.py index f73ac33..058a162 100644 --- a/benchmarks/urls.py +++ b/benchmarks/urls.py @@ -2,8 +2,8 @@ Each view does exactly one cache operation matching one of the seven benchmark phases. The runner drives them via ``django.test.Client``, which -exercises the full WSGI handler — middleware, URL resolution, request/response -construction, ``request_started`` / ``request_finished`` signals — so the +exercises the full WSGI handler (middleware, URL resolution, request/response +construction, ``request_started`` / ``request_finished`` signals), so the numbers reflect what a real Django request paying for the same cache work looks like, not just the cache call in isolation. """ @@ -21,7 +21,7 @@ def set_payload_kind(kind: str) -> None: """Set the payload used by ``set`` / ``mset`` views. Called once per benchmark run.""" - global _PAYLOAD # noqa: PLW0603 — module-level state intentional for benchmark setup + global _PAYLOAD # noqa: PLW0603 (module-level state intentional for benchmark setup) _PAYLOAD = _build_payload_large() if kind == "large" else _build_payload() @@ -87,7 +87,7 @@ async def bench_seed(_request: Any) -> HttpResponse: async def bench_mixed(_request: Any) -> HttpResponse: - """Six async cache ops per request — matches django-vcache's workload.""" + """Six async cache ops per request, matching django-vcache's workload.""" await cache.aget("bench:s1") # 1. small get await cache.aget_many(["bench:s1", "bench:s2", "bench:s3"]) # 2. batch get await cache.aset("bench:s1", _BENCH_SMALL, 300) # 3. small set diff --git a/crates/django-cachex-redis-rs/README.md b/crates/django-cachex-redis-rs/README.md index cbb589d..a9ae730 100644 --- a/crates/django-cachex-redis-rs/README.md +++ b/crates/django-cachex-redis-rs/README.md @@ -1,6 +1,6 @@ # django-cachex-redis-rs -Rust adapter for [django-cachex] — built on PyO3 + tokio + [redis-rs]. +Rust adapter for [django-cachex], built on PyO3 + tokio + [redis-rs]. This is a binary-only companion package that ships the compiled `_redis_rs` extension module into the `django_cachex.adapters` @@ -10,9 +10,10 @@ namespace. Install via the `redis-rs` extra on the main package: pip install django-cachex[redis-rs] ``` -Prebuilt wheels are published for Linux x86_64 (cp314, cp314t). On other -platforms there's no wheel — pip will try to build from source (requires -the Rust toolchain) or fail clearly. +Prebuilt wheels are published for Linux x86_64, Linux aarch64, macOS +arm64, and Windows amd64, on both cp314 and cp314t (free-threaded). On +other platforms there's no wheel, so pip will try to build from source +(requires the Rust toolchain) or fail clearly. The pure-Python `django-cachex` package is fully usable on its own; this binary just unlocks the `RedisRsCache` family of diff --git a/crates/django-cachex-redis-rs/src/adapter.rs b/crates/django-cachex-redis-rs/src/adapter.rs index 24fc75d..24d81ae 100644 --- a/crates/django-cachex-redis-rs/src/adapter.rs +++ b/crates/django-cachex-redis-rs/src/adapter.rs @@ -13,13 +13,13 @@ use pyo3::prelude::*; use pyo3::types::{PyDateTime, PyDelta, PyDeltaAccess, PyDict}; // ========================================================================= -// Argument types — names mirror redis-py's ``redis.typing`` for vocabulary +// Argument types. Names mirror redis-py's ``redis.typing`` for vocabulary // consistency. Each derives ``FromPyObject`` so PyO3 raises ``TypeError`` // for invalid input shapes, replacing the old runtime-polymorphism helpers // (``value_to_bytes`` / ``to_seconds`` / ``to_unix`` / ...). // ========================================================================= -/// ``int | timedelta`` — TTL inputs to EXPIRE / PEXPIRE / TOUCH / SET ... EX. +/// ``int | timedelta``: TTL inputs to EXPIRE / PEXPIRE / TOUCH / SET ... EX. /// Matches redis-py's ``ExpiryT``; cachex narrows the int side to a signed /// integer (negative timeouts collapse to "delete immediately" elsewhere). #[derive(FromPyObject)] @@ -56,7 +56,7 @@ impl ExpiryT<'_> { } } -/// ``int | datetime`` — absolute expiry inputs to EXPIREAT / PEXPIREAT. +/// ``int | datetime``: absolute expiry inputs to EXPIREAT / PEXPIREAT. /// Matches redis-py's ``AbsExpiryT``. #[derive(FromPyObject)] pub(crate) enum AbsExpiryT<'py> { @@ -89,7 +89,7 @@ impl AbsExpiryT<'_> { } } -/// ``bytes | int`` — values written to / matched against Redis. Matches +/// ``bytes | int``: values written to / matched against Redis. Matches /// redis-py's ``EncodableT`` *vocabulary*; cachex narrows the actual set /// of accepted Python types because the cache layer's serializer always /// produces ``bytes`` and ``INCR`` expects ints. The int branch encodes @@ -111,7 +111,7 @@ impl EncodableT { } } -/// Sync command — resolve connection (cached after first call), release the +/// Sync command: resolve connection (cached after first call), release the /// GIL, block on the tokio runtime. Returns the raw redis-rs result; caller /// uses `.map_err(crate::client::to_py_err)` to convert. macro_rules! adapter_sync { @@ -123,13 +123,13 @@ macro_rules! adapter_sync { }}; } -/// Async command — resolve connection (cached after first call), spawn the +/// Async command: resolve connection (cached after first call), spawn the /// op on the tokio runtime, return an awaitable. Body must produce a /// `RawResult` (use `.into_raw_result()` from `crate::client::IntoRawResult`). /// /// Two forms: -/// - `adapter_async!(slf, conn, body)` — no post-await transform. -/// - `adapter_async!(slf, conn, body; Transform::Variant)` — apply transform +/// - `adapter_async!(slf, conn, body)`: no post-await transform. +/// - `adapter_async!(slf, conn, body; Transform::Variant)`: apply transform /// to the resolved value (e.g., `ToBool`, `NormalizeTtl`) before delivery. /// Replaces the older Python `_async_helpers` coroutine wrap pattern. macro_rules! adapter_async { @@ -152,7 +152,7 @@ macro_rules! adapter_async { } // ========================================================================= -// Helpers — Python value conversions used by adapter methods +// Helpers for Python value conversions used by adapter methods // ========================================================================= /// Coerce ``KeyT`` (str | bytes | int) to a Rust ``String``. @@ -434,7 +434,7 @@ pub(crate) fn parse_xinfo_pairs_list<'py>(py: Python<'py>, raw: &Bound<'py, PyAn Ok(out.into_any().unbind()) } -/// Coerce a Python value to ``i64`` via the builtin ``int(...)`` — tolerates +/// Coerce a Python value to ``i64`` via the builtin ``int(...)``, which tolerates /// ``int``, ``bytes`` (b"123"), and ``str``. Used by the XPENDING decoders /// where Redis returns numeric fields as bulk strings depending on protocol. fn to_py_int(value: &Bound<'_, PyAny>) -> PyResult { @@ -566,7 +566,7 @@ fn await_constant( } // ========================================================================= -// Stampede helpers — call into ``django_cachex.stampede`` +// Stampede helpers that call into ``django_cachex.stampede`` // ========================================================================= // // Stampede math (XFetch decision, config merging, buffer arithmetic) lives @@ -634,13 +634,29 @@ fn call_should_recompute<'py>( /// Convert ``Option`` timeout to the driver's ``Option`` ttl arg /// (clamping negatives to ``Some(0)`` for parity with Python's redis-py -/// behavior — ``timeout <= 0`` means "delete"). +/// behavior, where ``timeout <= 0`` means "delete"). fn timeout_to_ttl(timeout: Option) -> Option { timeout.map(|t| if t < 0 { 0 } else { t as u64 }) } +/// Whether a flagged ``SET`` actually wrote the key: with GET the old value +/// tells the story (NX writes on Nil, XX on non-Nil), without it OK vs Nil. +fn set_with_flags_executed(reply: &redis::Value, nx: bool, xx: bool, get: bool) -> bool { + if get { + if nx { + matches!(reply, redis::Value::Nil) + } else if xx { + !matches!(reply, redis::Value::Nil) + } else { + true + } + } else { + matches!(reply, redis::Value::Okay | redis::Value::SimpleString(_)) + } +} + // ========================================================================= -// RedisRsAdapter — high-level cachex adapter, subclassable from Python +// RedisRsAdapter: high-level cachex adapter, subclassable from Python // ========================================================================= /// Captured connection config so the adapter can re-connect on demand @@ -740,7 +756,7 @@ impl AdapterConnConfig { /// Resolve a ``Conn`` for this config, sharing one across all adapter /// instances in the process (per-config). Django ASGI under granian - /// instantiates a fresh adapter per asyncio task — without this cache + /// instantiates a fresh adapter per asyncio task. Without this cache, /// every task would build its own multiplexed transport, exploding /// the upstream connection count. async fn connect_cached(&self) -> Result { @@ -787,7 +803,7 @@ fn standard_config_from_options( }) } -/// Build the adapter-side ``Cluster`` config — every server in the list +/// Build the adapter-side ``Cluster`` config. Every server in the list /// is a cluster node URL. fn cluster_config_from_options( py: Python<'_>, @@ -969,7 +985,7 @@ impl RedisRsAdapter { /// Cross-driver convenience: ``cache.get_client()`` returns "the /// underlying client object" for tests / debugging. For the redis-rs - /// adapter, that's the adapter itself — its command surface mirrors + /// adapter, that's the adapter itself. Its command surface mirrors /// the redis-py client closely enough for shared cross-driver tests. /// ``key`` and ``write`` are accepted for signature parity with the /// other adapters (sentinel/cluster route by key) but ignored here. @@ -1022,7 +1038,7 @@ impl RedisRsAdapter { // ``apipeline`` is async by Protocol contract; return a pre-resolved // awaitable that delivers the constructed pipeline wrapper. Construction - // itself is sync — buffering the commands needs no I/O. + // itself is sync, since buffering the commands needs no I/O. #[pyo3(signature = (*, transaction = true))] fn apipeline( slf: &Bound<'_, Self>, @@ -1044,7 +1060,7 @@ impl RedisRsAdapter { // // ``pipeline_exec`` / ``apipeline_exec`` execute a buffered list of // commands in one round trip. Called from the ``RedisRsPipelineAdapter`` - // pyclass (see ``pipeline.rs``) — the pipeline holds an adapter + // pyclass (see ``pipeline.rs``). The pipeline holds an adapter // reference and dispatches to these methods on ``execute()``. // ===================================================================== @@ -1077,7 +1093,7 @@ impl RedisRsAdapter { // ===================================================================== // Lock primitives // - // Distributed lock built on Lua scripts — atomic acquire / release / + // Distributed lock built on Lua scripts: atomic acquire / release / // extend keyed by an adapter-owned token. Called from the Python // ``Lock`` / ``AsyncLock`` wrappers (see ``django_cachex.lock``); the // wrappers handle blocking, retry, and context-manager semantics on @@ -1166,7 +1182,7 @@ impl RedisRsAdapter { // ===================================================================== /// Resolve a per-call ``stampede_prevention`` override against the - /// instance config — returns the effective ``StampedeConfig`` or ``None``. + /// instance config, returning the effective ``StampedeConfig`` or ``None``. /// Used by :class:`RespCache` to decide whether the per-call buffer applies. #[pyo3(signature = (stampede_prevention=None))] fn resolve_stampede( @@ -1203,7 +1219,7 @@ impl RedisRsAdapter { let cfg = &slf.borrow().stampede_config; let actual = call_get_timeout_with_buffer(py, timeout, cfg, stampede_prevention)?; if actual == Some(0) { - // ``timeout=0`` is "set then immediately delete" — used by Django + // ``timeout=0`` is "set then immediately delete", used by Django // backends to express "keys with non-positive timeout expire now." let set: bool = adapter_sync!(slf, conn, conn.set_nx(&key, nvalue, None).await) .map_err(crate::client::to_py_err)?; @@ -1402,17 +1418,17 @@ impl RedisRsAdapter { let nvalue = value.into_bytes(); let cfg = &slf.borrow().stampede_config; let actual = call_get_timeout_with_buffer(py, timeout, cfg, stampede_prevention)?; - if actual == Some(0) { - // ``timeout=0`` short-circuit: GET branch returns None (no value - // to fetch); NX/XX branch returns False. - return Ok(if get { py.None() } else { false.into_pyobject(py)?.to_owned().into_any().unbind() }); - } - let ttl = timeout_to_ttl(actual); - let r: Result = adapter_sync!( - slf, - conn, - conn.set_with_flags(&key, nvalue, ttl, nx, xx, get).await - ); + // timeout=0 means expire immediately: run the SET unexpired so the + // nx/xx/get semantics still apply, then delete when it executed. + let zero = actual == Some(0); + let ttl = if zero { None } else { timeout_to_ttl(actual) }; + let r: Result = adapter_sync!(slf, conn, { + let reply = conn.set_with_flags(&key, nvalue, ttl, nx, xx, get).await?; + if zero && set_with_flags_executed(&reply, nx, xx, get) { + conn.del(&key).await?; + } + Ok(reply) + }); let result = r.map_err(crate::client::to_py_err)?; if get { // SET ... GET: ``Nil`` → None, otherwise the prior value bytes. @@ -1446,15 +1462,8 @@ impl RedisRsAdapter { let nvalue = value.into_bytes(); let cfg = &slf.borrow().stampede_config; let actual = call_get_timeout_with_buffer(py, timeout, cfg, stampede_prevention)?; - if actual == Some(0) { - let value: Py = if get { - py.None() - } else { - false.into_pyobject(py)?.to_owned().into_any().unbind() - }; - return await_constant(py, value); - } - let ttl = timeout_to_ttl(actual); + let zero = actual == Some(0); + let ttl = if zero { None } else { timeout_to_ttl(actual) }; let transform = if get { crate::async_bridge::AwaitTransform::SetWithFlagsGet } else { @@ -1464,7 +1473,15 @@ impl RedisRsAdapter { slf, conn, { use crate::client::IntoRawResult; - conn.set_with_flags(&key, nvalue, ttl, nx, xx, get).await.into_raw_result() + let result: redis::RedisResult = async { + let reply = conn.set_with_flags(&key, nvalue, ttl, nx, xx, get).await?; + if zero && set_with_flags_executed(&reply, nx, xx, get) { + conn.del(&key).await?; + } + Ok(reply) + } + .await; + result.into_raw_result() }; transform ) @@ -1494,7 +1511,7 @@ impl RedisRsAdapter { let resolved = call_resolve_stampede(py, cfg, stampede_prevention)?; if !resolved.is_none() && !out.is_empty() { // All present entries are bytes (MGET semantics); apply stampede - // check to each. Per-key TTL roundtrip — same shape as before. + // check to each. Per-key TTL roundtrip, same shape as before. let present_keys: Vec = out .keys() .iter() @@ -1629,7 +1646,7 @@ impl RedisRsAdapter { // // All methods dispatch through the driver's Python-exposed methods // (`call_method1`). One method-lookup per call vs. typed inherent - // calls — but the driver's `#[pymethods]` are private to its module, + // calls, but the driver's `#[pymethods]` are private to its module, // and Python dispatch keeps the adapter independent of changes in // the driver's Rust API surface. // ===================================================================== @@ -1664,7 +1681,7 @@ impl RedisRsAdapter { }) } - /// Redis-py-style alias for ``has_key`` — kept so cross-driver tests + /// Redis-py-style alias for ``has_key``, kept so cross-driver tests /// that use ``cache.get_client().exists(...)`` work against the /// adapter (which is what ``get_client()`` returns here). fn exists(slf: &Bound<'_, Self>, key: &str) -> PyResult { @@ -3502,8 +3519,8 @@ impl RedisRsAdapter { #[pyo3(signature = (**_kwargs))] fn close(&self, _kwargs: Option<&Bound<'_, PyDict>>) { // No-op. Django fires ``cache.close()`` on every ``request_finished`` - // signal as a request-cycle cleanup hook, not a shutdown call — - // dropping the multiplexed connection here would force a reconnect + // signal as a request-cycle cleanup hook, not a shutdown call. + // Dropping the multiplexed connection here would force a reconnect // per request and saturate the server with TCP handshakes. } @@ -3533,8 +3550,8 @@ impl RedisRsAdapter { // ``arename`` / ``arenamenx`` wrap the inner awaitable in a Python // coroutine (``rename_after``) that re-raises ``ERR no such key`` as - // ``ValueError``. The wrapped value is a coroutine — return type stays - // ``Py``. + // ``ValueError``. The wrapped value is a coroutine, so the return type + // stays ``Py``. fn arename( slf: &Bound<'_, Self>, src: &str, @@ -4017,8 +4034,8 @@ impl RedisRsAdapter { .unbind()) } - // ``aiter_keys`` returns a Python async generator (``ascan_iter_loop``) - // — not an awaitable — so the return type is genuinely ``Py``. + // ``aiter_keys`` returns a Python async generator (``ascan_iter_loop``), + // not an awaitable, so the return type is genuinely ``Py``. #[pyo3(signature = (pattern, itersize=None))] fn aiter_keys( slf: &Bound<'_, Self>, @@ -4110,8 +4127,8 @@ impl RedisRsAdapter { Ok(total) } - // ``adelete_pattern`` returns a Python coroutine — not an awaitable — - // so the return type is genuinely ``Py``. + // ``adelete_pattern`` returns a Python coroutine, not an awaitable, so + // the return type is genuinely ``Py``. #[pyo3(signature = (pattern, itersize=None))] fn adelete_pattern( slf: &Bound<'_, Self>, @@ -4254,8 +4271,8 @@ impl RedisRsAdapter { conn.hset(&key, &f, &v).await.into_raw_result() }); } - // Multi-field: HLEN before, HMSET, HLEN after — return diff. The - // cachex contract for multi-field hset is "count of *new* fields", + // Multi-field: HLEN before, HMSET, HLEN after, then return the diff. + // The cachex contract for multi-field hset is "count of *new* fields", // not the OK return that HMSET produces. adapter_async!(slf, conn, { let before: i64 = match conn.hlen(&key).await { diff --git a/crates/django-cachex-redis-rs/src/async_bridge.rs b/crates/django-cachex-redis-rs/src/async_bridge.rs index 83a6d51..e819284 100644 --- a/crates/django-cachex-redis-rs/src/async_bridge.rs +++ b/crates/django-cachex-redis-rs/src/async_bridge.rs @@ -5,7 +5,7 @@ // https://gitlab.com/glitchtip/django-vcache/-/blob/main/src/async_bridge.rs // // Keep this file in lockstep with upstream. If you want to diverge, -// open a discussion first — the design (5-poll busy-yield, OnceLock +// open a discussion first. The design (5-poll busy-yield, OnceLock // runtime with PID fork detection, oneshot channel + watcher task) is // the load-bearing part and must not drift accidentally. @@ -71,11 +71,11 @@ fn init_or_fork_runtime(pid: u32) -> &'static Runtime { } // ========================================================================= -// Result types — Rust-native, no GIL needed to construct +// Result types: Rust-native, no GIL needed to construct // ========================================================================= pub enum RawResult { - /// Successful operation with no return value — renders as Python None. + /// Successful operation with no return value. Renders as Python None. Nil, OptBytes(Option>), Bool(bool), @@ -91,7 +91,7 @@ pub enum RawResult { StringList(Vec), /// Field/value pairs for HGETALL/HMSET-style results (bytes value). BytesPairs(Vec<(Vec, Vec)>), - /// Key/value pairs for `aget_many` post-stampede-filter — surfaces as + /// Key/value pairs for `aget_many` post-stampede-filter. Surfaces as /// a Python ``dict[str, bytes]``. StringBytesPairs(Vec<(String, Vec)>), /// Member/score pairs for ZRANGE WITHSCORES, ZPOPMIN/MAX. @@ -107,15 +107,15 @@ pub enum RawResult { /// Used for EVAL/EVALSHA, INFO, CLIENT LIST, and other commands whose /// return shape varies enough that a typed variant doesn't help. Value(redis::Value), - /// XRANGE / XREVRANGE / XCLAIM (no JUSTID) — typed as + /// XRANGE / XREVRANGE / XCLAIM (no JUSTID), typed as /// ``list[tuple[str, dict[str, bytes]]]``; see :mod:`stream_decode`. StreamEntries(redis::Value), - /// XREAD / XREADGROUP — typed as + /// XREAD / XREADGROUP, typed as /// ``dict[str, list[tuple[str, dict[str, bytes]]]] | None``. StreamRead(redis::Value), - /// XCLAIM with ``JUSTID`` — typed as ``list[str]``. + /// XCLAIM with ``JUSTID``, typed as ``list[str]``. XClaimJustId(redis::Value), - /// XAUTOCLAIM — typed as + /// XAUTOCLAIM, typed as /// ``tuple[str, list[tuple[str, dict[str, bytes]]] | list[str], list[str]]``. /// Bool is the ``justid`` flag the call was issued with. Xautoclaim(redis::Value, bool), @@ -158,7 +158,7 @@ fn redis_value_to_py(py: Python<'_>, v: redis::Value) -> PyResult> { Ok(dict.into_any().unbind()) } redis::Value::Set(items) => { - // Redis sets via RESP3 — return as list to preserve order; Python can `set(...)` if needed. + // Redis sets via RESP3. Return as list to preserve order; Python can `set(...)` if needed. let py_items: Vec> = items .into_iter() .map(|item| redis_value_to_py(py, item)) @@ -180,7 +180,7 @@ fn redis_value_to_py(py: Python<'_>, v: redis::Value) -> PyResult> { redis::Value::ServerError(e) => Err(pyo3::exceptions::PyRuntimeError::new_err(format!( "{e:?}" ))), - // redis::Value is marked non_exhaustive — fall back to the Debug repr. + // redis::Value is marked non_exhaustive, so fall back to the Debug repr. other => Ok(PyString::new(py, &format!("{other:?}")).into_any().unbind()), } } @@ -305,9 +305,9 @@ impl RawResult { } // ========================================================================= -// RedisRsAwaitable — deferred-callback async bridge +// RedisRsAwaitable: deferred-callback async bridge // -// The tokio task sends its result via a oneshot channel — no GIL needed. +// The tokio task sends its result via a oneshot channel, so no GIL is needed. // // __next__ polls try_recv(). For fast local operations, the result is // usually ready on the first call → zero overhead, identical to the old @@ -333,7 +333,7 @@ struct DoneCallback { context: Option>, } -/// Callback-mode state — only allocated when an operation doesn't resolve +/// Callback-mode state, only allocated when an operation doesn't resolve /// within the busy-yield window (5 polls). Most fast ops never need this, /// so keeping it boxed avoids bloating every RedisRsAwaitable allocation. struct CallbackState { @@ -342,11 +342,11 @@ struct CallbackState { result_slot: Arc>>>, } -/// Post-await value transform — applied between `RawResult::into_py` and +/// Post-await value transform, applied between `RawResult::into_py` and /// delivery to the Python consumer. Replaces a stack of single-purpose /// Python `async def` wrappers from `django_cachex.adapters._async_helpers`. /// -/// The transforms here are pure value reshapes — they do NOT touch the +/// The transforms here are pure value reshapes. They do NOT touch the /// connection or perform I/O. Multi-step async (e.g. stampede TTL probes) /// stays in Python coroutines because it needs further awaits. #[derive(Clone, Copy, Debug)] @@ -384,7 +384,7 @@ pub enum AwaitTransform { /// nested `[[m, s], ...]` (RESP3) or flat `[m, s, m, s, ...]` (RESP2) /// and returns `[(m, float(s)), ...]`. DecodeZrangeWithScores, - /// ZREVRANGEBYSCORE WITHSCORES decoding — always flat + /// ZREVRANGEBYSCORE WITHSCORES decoding, always flat /// `[m, s, m, s, ...]` → `[(m, float(s)), ...]`. DecodeZrevrangebyscoreWithScores, /// XINFO STREAM flat-pair → dict (`adapter::parse_xinfo_pairs`). @@ -411,9 +411,9 @@ pub struct RedisRsAwaitable { rx: Option>, /// Post-await transform applied to the resolved value before delivery. transform: AwaitTransform, - /// Successful result value — stored for result() after StopIteration delivery. + /// Successful result value, stored for result() after StopIteration delivery. value: Option>, - /// Error exception object — raised by result() for the Task to propagate. + /// Error exception object, raised by result() for the Task to propagate. error: Option>, /// Whether we have a stored result (value or error). resolved: bool, @@ -423,7 +423,7 @@ pub struct RedisRsAwaitable { _asyncio_future_blocking: bool, /// Number of times __next__ has been called without a result. polls: u8, - /// Callback mode state — allocated lazily on 6th poll miss. + /// Callback mode state, allocated lazily on 6th poll miss. cb: Option>, } @@ -671,12 +671,12 @@ impl RedisRsAwaitable { fn __next__(slf: Py, py: Python<'_>) -> PyResult> { let mut this = slf.borrow_mut(py); - // Cancelled — raise CancelledError. + // Cancelled: raise CancelledError. if this.cancelled { return Err(cancelled_error(py)); } - // Already resolved — re-deliver stored result. + // Already resolved: re-deliver stored result. if this.resolved { if let Some(ref exc) = this.error { return Err(PyErr::from_value(exc.bind(py).clone())); @@ -750,7 +750,7 @@ impl RedisRsAwaitable { this.polls += 1; if this.polls <= 5 { - // Busy-yield for up to 5 iterations — covers nearly all fast ops + // Busy-yield for up to 5 iterations. This covers nearly all fast ops // (sub-ms driver operations resolve within 1-3 event loop ticks). // Callback mode has high fixed cost (get_running_loop + watcher // spawn + spawn_blocking + GIL acquisition), so busy-yield is diff --git a/crates/django-cachex-redis-rs/src/client.rs b/crates/django-cachex-redis-rs/src/client.rs index db6f9c3..8712d40 100644 --- a/crates/django-cachex-redis-rs/src/client.rs +++ b/crates/django-cachex-redis-rs/src/client.rs @@ -7,11 +7,11 @@ // Now reduced to the cross-cutting helpers ``adapter.rs`` uses: // * Connection-config types (``ClientCacheOpts``, ``TlsOpts``) and // the ``make_*`` factories that translate Python OPTIONS into them. -// * Error classification — ``classify``, ``to_py_err``, -// ``is_connection_error`` — splits Redis errors into +// * Error classification (``classify``, ``to_py_err``, +// ``is_connection_error``), which splits Redis errors into // ``ConnectionError`` (retryable / transport-level) vs // ``RuntimeError`` (server-side). -// * ``IntoRawResult`` + ``From`` impls — converts every typed +// * ``IntoRawResult`` + ``From`` impls, which convert every typed // ``RedisResult`` into a ``RawResult`` variant for the // awaitable bridge. // * Sync-side conversion helpers (``py_redis_value``, @@ -48,8 +48,8 @@ pub(crate) fn to_py_err(e: redis::RedisError) -> PyErr { } } -/// Server returned ``WRONGTYPE`` — applying a command to a key whose type -/// doesn't match (e.g. ``LPUSH`` on a key that holds a string). +/// Server returned ``WRONGTYPE``, meaning a command was applied to a key +/// whose type doesn't match (e.g. ``LPUSH`` on a key that holds a string). pub(crate) fn is_wrongtype(e: &redis::RedisError) -> bool { matches!(e.code(), Some("WRONGTYPE")) } diff --git a/crates/django-cachex-redis-rs/src/connection.rs b/crates/django-cachex-redis-rs/src/connection.rs index b1682cd..0a1d670 100644 --- a/crates/django-cachex-redis-rs/src/connection.rs +++ b/crates/django-cachex-redis-rs/src/connection.rs @@ -77,7 +77,7 @@ fn url_with_resp3(url: &str) -> String { if url.contains("protocol=") { return url.to_string(); } - // Handle fragment (#...) — query params must come before it. + // Handle fragment (#...). Query params must come before it. let (base, fragment) = match url.split_once('#') { Some((b, f)) => (b, Some(f)), None => (url, None), @@ -89,7 +89,7 @@ fn url_with_resp3(url: &str) -> String { } } -/// Config for blocking operations — no response timeout since BLMOVE/BLMPOP +/// Config for blocking operations. No response timeout, since BLMOVE/BLMPOP /// intentionally wait for data (possibly minutes). Never has caching. fn blocking_conn_manager_config() -> ConnectionManagerConfig { ConnectionManagerConfig::new() @@ -176,11 +176,11 @@ impl SentinelConn { } } -/// Inner connection enum — one per connection type. +/// Inner connection enum, one per connection type. /// All methods for individual Redis commands live here. /// /// Public so [`Conn`] can expose it as the target of its [`Deref`] -/// impls — non-blocking commands resolve straight to the inner without +/// impls. Non-blocking commands then resolve straight to the inner without /// per-method passthroughs. Crate is a `cdylib`, so the "public" boundary /// is just the PyO3 surface; this type is still effectively internal. #[derive(Clone)] @@ -794,7 +794,7 @@ impl ConnInner { } // ========================================================================= -// Connection config — stored for lazy blocking connection creation +// Connection config, stored for lazy blocking connection creation // ========================================================================= #[derive(Clone)] @@ -821,7 +821,7 @@ impl ConnInner { Self::Standard(c) => c.get_cache_statistics(), Self::Cluster(_) => None, // cluster connection doesn't expose this yet Self::Sentinel(s) => { - // Can't block here — return None if lock is contested. + // Can't block here, so return None if lock is contested. s.inner.try_read().ok().and_then(|c| c.get_cache_statistics()) } } @@ -829,13 +829,13 @@ impl ConnInner { } // ========================================================================= -// Conn — public wrapper with separate regular + blocking connections +// Conn: public wrapper with separate regular + blocking connections // ========================================================================= /// Public connection handle. Uses one connection for regular (fast) ops and /// a lazily-created second connection for blocking ops (BLMOVE, BLMPOP). /// -/// Redis processes commands sequentially per connection — a BLMOVE with a +/// Redis processes commands sequentially per connection, so a BLMOVE with a /// 1-second timeout blocks ALL other commands multiplexed on that connection. /// The separate blocking connection prevents this head-of-line blocking. #[derive(Clone)] @@ -849,8 +849,7 @@ pub struct Conn { // `conn.X(args).await` auto-resolve to `conn.regular.X(args).await` without // hand-written passthroughs for every command. Blocking commands (BLMOVE, // BL{POP,RPOP,MPOP}) bypass this by being defined as inherent methods on -// `Conn` — inherent methods take precedence over `Deref`-resolved -// ones. +// `Conn`, which take precedence over `Deref`-resolved ones. impl std::ops::Deref for Conn { type Target = ConnInner; fn deref(&self) -> &Self::Target { @@ -915,7 +914,7 @@ impl Conn { .cloned() } - // === Regular ops — delegate to self.regular === + // === Regular ops: delegate to self.regular === pub async fn get_bytes(&mut self, key: &str) -> RedisResult>> { self.regular.get_bytes(key).await @@ -1071,7 +1070,7 @@ impl Conn { self.regular.cache_statistics() } - // === Blocking ops — use separate connection === + // === Blocking ops: use separate connection === pub async fn blmove( &mut self, @@ -1329,8 +1328,8 @@ impl ConnInner { dispatch_cmd!(self, cmd) } - /// On Cluster, KEYS only hits one master and returns that node's keys — - /// callers needing a full-cluster scan should fan out themselves or use + /// On Cluster, KEYS only hits one master and returns that node's keys. + /// Callers needing a full-cluster scan should fan out themselves or use /// `scan_all` (which has the same single-node limitation here, mirroring /// vcache's behavior; documented for both). pub async fn keys(&mut self, pattern: &str) -> RedisResult> { @@ -2028,8 +2027,8 @@ impl ConnInner { /// /// `block_ms` adds the BLOCK option. The caller is responsible for routing /// blocking variants through the dedicated blocking connection (see - /// `Conn::xread`) — head-of-line blocking on the multiplexed - /// connection would otherwise stall every other command. + /// `Conn::xread`). Head-of-line blocking on the multiplexed connection + /// would otherwise stall every other command. pub async fn xread( &mut self, keys: &[String], diff --git a/crates/django-cachex-redis-rs/src/pipeline.rs b/crates/django-cachex-redis-rs/src/pipeline.rs index 0281a72..46d25ac 100644 --- a/crates/django-cachex-redis-rs/src/pipeline.rs +++ b/crates/django-cachex-redis-rs/src/pipeline.rs @@ -3,7 +3,7 @@ // // The shape of each command on the wire is ``(name, args)``. Per-command // parsers (Python callables in ``django_cachex.adapters._pipeline_parsers``) -// normalize the response shape — RESP2 flat lists → dicts, bytes → float, +// normalize the response shape: RESP2 flat lists → dicts, bytes → float, // stream entries → ``(id, fields)`` tuples, etc. // // Sync ``execute()`` resolves to ``list[Any]`` directly. Async @@ -1932,7 +1932,7 @@ impl RedisRsAsyncPipelineAdapter { } } -// Silence unused-import warnings — PyTuple is used via cast paths in extension. +// Silence unused-import warnings. PyTuple is used via cast paths in extension. const _: fn() = || { let _ = std::marker::PhantomData::; }; diff --git a/django_cachex/adapters/pipeline.py b/django_cachex/adapters/pipeline.py index 00b3f6a..77e1125 100644 --- a/django_cachex/adapters/pipeline.py +++ b/django_cachex/adapters/pipeline.py @@ -70,9 +70,13 @@ def __exit__(self, *args: object) -> None: def execute(self) -> list[Any]: """Execute all queued commands and decode the results.""" - results = self._pipeline_adapter.execute() decoders = self._decoders - self._decoders = [] + try: + results = self._pipeline_adapter.execute() + finally: + # The driver pipeline discards its queue on error; stale decoders + # would misalign against the next batch. + self._decoders = [] decoded = [] for result, decoder in zip(results, decoders, strict=True): decoded.append(decoder(result)) @@ -1633,21 +1637,27 @@ async def __aexit__(self, *args: object) -> None: await self._pipeline_adapter.reset() self._decoders.clear() - def __exit__(self, *args: object) -> None: - """Block sync ``with`` on an async pipeline. + def __enter__(self) -> Self: + """Block sync ``with`` on an async pipeline before any commands queue. - ``Pipeline.__exit__`` calls a sync ``reset()``. The async pipeline - adapter's ``reset()`` returns a coroutine that would be silently - discarded, leaving connection state dangling. + The inherited ``Pipeline.__exit__`` calls a sync ``reset()``; on an + async adapter that returns a coroutine which would be silently + discarded, leaving connection state dangling. Failing at ``__enter__`` + keeps the error next to the offending ``with`` statement rather than + after the block ran. """ msg = "AsyncPipeline requires 'async with', not 'with'" raise TypeError(msg) async def execute(self) -> list[Any]: # type: ignore[override] """Execute all queued commands asynchronously and decode the results.""" - results = await self._pipeline_adapter.execute() decoders = self._decoders - self._decoders = [] + try: + results = await self._pipeline_adapter.execute() + finally: + # The driver pipeline discards its queue on error; stale decoders + # would misalign against the next batch. + self._decoders = [] return [decoder(result) for result, decoder in zip(results, decoders, strict=True)] diff --git a/django_cachex/adapters/redis_rs.py b/django_cachex/adapters/redis_rs.py index 77f3fca..661ba7b 100644 --- a/django_cachex/adapters/redis_rs.py +++ b/django_cachex/adapters/redis_rs.py @@ -14,6 +14,7 @@ RespAsyncPipelineProtocol, RespPipelineProtocol, ) +from django_cachex.exceptions import NotSupportedError # The native ``_redis_rs`` extension ships in the optional # ``django-cachex-redis-rs`` binary package. When only the pure wheel is @@ -100,9 +101,19 @@ class RedisRsAdapter(_RustRedisRsAdapter, RespAdapterProtocol): The Rust ``RedisRsAdapter`` connects directly to redis-rs in ``__init__``; ``RespAdapterProtocol`` is a structural mixin, - runtime-checkable, contributes no methods. + runtime-checkable. It sits last in the MRO, so any command the Rust + class does not define resolves to the protocol's ``...`` body and + returns ``None`` rather than raising. Override such gaps explicitly. """ + def slowlog_get(self, count: int = 10) -> list[dict[str, Any]]: + """Reject SLOWLOG GET, which the Rust adapter does not implement.""" + raise NotSupportedError("slowlog_get", backend="redis-rs") + + def slowlog_len(self) -> int: + """Reject SLOWLOG LEN, which the Rust adapter does not implement.""" + raise NotSupportedError("slowlog_len", backend="redis-rs") + def lock( self, key: str, diff --git a/django_cachex/adapters/valkey_glide.py b/django_cachex/adapters/valkey_glide.py index 9651400..199b68e 100644 --- a/django_cachex/adapters/valkey_glide.py +++ b/django_cachex/adapters/valkey_glide.py @@ -16,15 +16,17 @@ import asyncio import datetime +import inspect import os import threading import time import weakref from typing import TYPE_CHECKING, Any, Self -from urllib.parse import urlparse +from urllib.parse import parse_qs, urlparse from django_cachex.adapters.protocols import RespAdapterProtocol, RespAsyncPipelineProtocol, RespPipelineProtocol from django_cachex.adapters.valkey_py import _options_key +from django_cachex.exceptions import maybe_wrap_wrongtype from django_cachex.stampede import ( StampedeConfig, get_timeout_with_buffer, @@ -44,6 +46,7 @@ # raises at backend instantiation time with an actionable message rather # than at module import time. try: + from glide import ClusterScanCursor as AsyncClusterScanCursor # ty: ignore[unresolved-import] from glide import GlideClient as AsyncGlideClient # ty: ignore[unresolved-import] from glide import GlideClientConfiguration as AsyncGlideClientConfiguration # ty: ignore[unresolved-import] from glide import GlideClusterClient as AsyncGlideClusterClient # ty: ignore[unresolved-import] @@ -51,8 +54,10 @@ GlideClusterClientConfiguration as AsyncGlideClusterClientConfiguration, ) from glide import NodeAddress as AsyncNodeAddress # ty: ignore[unresolved-import] + from glide import ServerCredentials as AsyncServerCredentials # ty: ignore[unresolved-import] from glide_sync import ( # ty: ignore[unresolved-import] Batch, + ClusterScanCursor, ConditionalChange, ExpirySet, ExpiryType, @@ -61,7 +66,9 @@ GlideClusterClient, GlideClusterClientConfiguration, NodeAddress, + ObjectType, RequestError, + ServerCredentials, ) from glide_sync.glide_client import GlideClient # ty: ignore[unresolved-import] except ImportError as _exc: @@ -85,6 +92,83 @@ def _check_installed() -> None: _set = set +# ============================================================================= +# WRONGTYPE translation +# ============================================================================= + + +async def _await_translated(awaitable: Any) -> Any: + try: + return await awaitable + except RequestError as exc: + wrapped = maybe_wrap_wrongtype(exc) + if wrapped is exc: + raise + raise wrapped from exc + + +def _translating(fn: Any) -> Any: + def call(*args: Any, **kwargs: Any) -> Any: + try: + result = fn(*args, **kwargs) + except RequestError as exc: + wrapped = maybe_wrap_wrongtype(exc) + if wrapped is exc: + raise + raise wrapped from exc + # The async client's methods are coroutine functions, so the error + # surfaces on await rather than on call. + return _await_translated(result) if inspect.isawaitable(result) else result + + return call + + +class _WrongTypeClient: + """Forward every attribute to a glide client, translating WRONGTYPE responses. + + Glide's Rust-backed clients have no ``execute_command`` seam for the patch + :mod:`~django_cachex.adapters.valkey_py` uses, so wrap the client instead. + Callables come back wrapped (cached per name); anything else passes through. + """ + + __slots__ = ("_glide_client", "_wrappers") + + def __init__(self, client: Any) -> None: + self._glide_client = client + self._wrappers: dict[str, Any] = {} + + def __getattr__(self, name: str) -> Any: + wrapper = self._wrappers.get(name) + if wrapper is not None: + return wrapper + attr = getattr(self._glide_client, name) + if not callable(attr): + return attr + wrapper = _translating(attr) + self._wrappers[name] = wrapper + return wrapper + + def __repr__(self) -> str: + return f"{type(self).__name__}({self._glide_client!r})" + + # Python looks dunders up on the type, so __getattr__ never sees them and + # ``with client:`` would fail even though the wrapped client supports it. + + def __enter__(self) -> Any: + self._glide_client.__enter__() + return self + + def __exit__(self, *args: object) -> Any: + return self._glide_client.__exit__(*args) + + async def __aenter__(self) -> Any: + await self._glide_client.__aenter__() + return self + + async def __aexit__(self, *args: object) -> Any: + return await self._glide_client.__aexit__(*args) + + # ============================================================================= # Process-wide client registries # ============================================================================= @@ -127,6 +211,60 @@ def _glide_config_key(servers: list[str], options: dict[str, Any]) -> tuple[Any, return (tuple(servers), _options_key(options)) +_TLS_SCHEMES = frozenset({"rediss", "valkeys"}) + + +def _parse_db(u: Any, options: dict[str, Any]) -> int | None: + """Database index: OPTIONS ``db`` beats the URL query, which beats the URL path.""" + if (db := options.get("db")) is not None: + return int(db) + query_db = parse_qs(u.query).get("db") + if query_db and query_db[-1].isdigit(): + return int(query_db[-1]) + path = u.path.lstrip("/") + if path.isdigit(): + return int(path) + return None + + +def _glide_config_kwargs( + servers: list[str], + options: dict[str, Any], + *, + credentials_cls: Any, + include_database: bool = True, +) -> dict[str, Any]: + """``Glide*ClientConfiguration`` kwargs from the first URL plus OPTIONS. + + ``credentials_cls`` is the sync or async ``ServerCredentials`` flavor; + cluster configs pass ``include_database=False`` (cluster only serves db 0). + """ + u = urlparse(servers[0]) + kwargs: dict[str, Any] = {} + + use_tls = u.scheme in _TLS_SCHEMES + for opt in ("use_tls", "ssl"): + if opt in options: + use_tls = bool(options[opt]) + break + if use_tls: + kwargs["use_tls"] = True + + username = options.get("username") or u.username or None + password = options.get("password") or u.password or None + if username or password: + kwargs["credentials"] = credentials_cls(password=password, username=username) + + if include_database and (db := _parse_db(u, options)) is not None: + kwargs["database_id"] = db + + if (request_timeout := options.get("request_timeout")) is not None: + kwargs["request_timeout"] = int(request_timeout) + if (client_name := options.get("client_name")) is not None: + kwargs["client_name"] = client_name + return kwargs + + # ============================================================================= # Encoding helpers # ============================================================================= @@ -143,6 +281,17 @@ def _enc(v: Any) -> bytes | str: return v +def _object_type(name: str | None) -> Any: + """Map a RESP type name to glide's ``ObjectType``, which SCAN requires.""" + # Glide reads ``type.value`` off the argument, so the plain string the + # protocol carries has to be looked up first. An unknown name yields no + # TYPE filter rather than raising. + if name is None: + return None + wanted = name.lower() + return next((t for t in ObjectType if t.value.lower() == wanted), None) + + def _enc_list(values: Iterable[Any]) -> list[bytes | str]: return [_enc(v) for v in values] @@ -151,12 +300,6 @@ def _enc_map(mapping: Mapping[Any, Any]) -> dict[Any, bytes | str]: return {k: _enc(v) for k, v in mapping.items()} -def _expiry(timeout: int | None) -> ExpirySet | None: - if timeout is None: - return None - return ExpirySet(ExpiryType.SEC, timeout) - - def _set_kw(*, ex: int | None = None, nx: bool = False, xx: bool = False, get: bool = False) -> dict[str, Any]: kw: dict[str, Any] = {} if ex is not None: @@ -503,6 +646,10 @@ def zadd(self, key: Any, mapping: Mapping[Any, float], **kwargs: Any) -> Self: args.append(b"NX") elif kwargs.get("xx"): args.append(b"XX") + if kwargs.get("gt"): + args.append(b"GT") + elif kwargs.get("lt"): + args.append(b"LT") if kwargs.get("ch"): args.append(b"CH") if kwargs.get("incr"): @@ -700,7 +847,22 @@ def lmove(self, src: Any, dst: Any, wherefrom: str, whereto: str) -> Self: # ---- streams (via custom_command for everything that's not single-response) ---- def xadd(self, key: Any, fields: Mapping[Any, Any], id: str = "*", **kwargs: Any) -> Self: - args = [b"XADD", key, _enc(id)] + args: list[Any] = [b"XADD", key] + if kwargs.get("nomkstream"): + args.append(b"NOMKSTREAM") + if (maxlen := kwargs.get("maxlen")) is not None: + args.append(b"MAXLEN") + if kwargs.get("approximate", True): + args.append(b"~") + args.append(str(maxlen).encode()) + elif (minid := kwargs.get("minid")) is not None: + args.append(b"MINID") + if kwargs.get("approximate", True): + args.append(b"~") + args.append(_enc(minid)) + if (limit := kwargs.get("limit")) is not None: + args.extend([b"LIMIT", str(limit).encode()]) + args.append(_enc(id)) for f, v in fields.items(): args.extend([_enc(f), _enc(v)]) self._batch.custom_command(args) @@ -813,6 +975,136 @@ def xtrim(self, key: Any, **kwargs: Any) -> Self: self._batch.custom_command(args) return self + def xack(self, key: Any, group: str, *ids: Any) -> Self: + self._batch.custom_command([b"XACK", key, _enc(group), *_enc_list(ids)]) + return self + + def xclaim( + self, + key: Any, + group: str, + consumer: str, + min_idle_time: int, + entry_ids: Sequence[str], + idle: int | None = None, + time: int | None = None, + retrycount: int | None = None, + force: bool = False, + justid: bool = False, + ) -> Self: + args: list[Any] = [ + b"XCLAIM", + key, + _enc(group), + _enc(consumer), + str(min_idle_time).encode(), + *_enc_list(entry_ids), + ] + if idle is not None: + args.extend([b"IDLE", str(idle).encode()]) + if time is not None: + args.extend([b"TIME", str(time).encode()]) + if retrycount is not None: + args.extend([b"RETRYCOUNT", str(retrycount).encode()]) + if force: + args.append(b"FORCE") + if justid: + args.append(b"JUSTID") + self._batch.custom_command(args) + if justid: + self._track(lambda r: _dec_keys(r or [])) + else: + self._track(_decode_stream_entries) + return self + + def xautoclaim( + self, + key: Any, + group: str, + consumer: str, + min_idle_time: int, + start_id: str = "0-0", + count: int | None = None, + justid: bool = False, + ) -> Self: + args: list[Any] = [ + b"XAUTOCLAIM", + key, + _enc(group), + _enc(consumer), + str(min_idle_time).encode(), + _enc(start_id), + ] + if count is not None: + args.extend([b"COUNT", str(count).encode()]) + if justid: + args.append(b"JUSTID") + self._batch.custom_command(args) + # Decode to redis-py shapes: a flat ID list for justid, else + # ``[next_id, [(id, fields), ...], deleted_ids]``. + if justid: + self._track(lambda r: _dec_keys(r[1] or [])) + else: + self._track( + lambda r: [ + _dec_str(r[0]), + _decode_stream_entries(r[1]), + _dec_keys(r[2]) if len(r) > 2 and r[2] else [], + ], + ) + return self + + def xgroup_create( + self, + key: Any, + group: str, + entry_id: str = "$", + mkstream: bool = False, + entries_read: int | None = None, + ) -> Self: + args: list[Any] = [b"XGROUP", b"CREATE", key, _enc(group), _enc(entry_id)] + if mkstream: + args.append(b"MKSTREAM") + if entries_read is not None: + args.extend([b"ENTRIESREAD", str(entries_read).encode()]) + self._batch.custom_command(args) + self._track(_ok_to_bool) + return self + + def xgroup_destroy(self, key: Any, group: str) -> Self: + self._batch.custom_command([b"XGROUP", b"DESTROY", key, _enc(group)]) + return self + + def xgroup_setid(self, key: Any, group: str, entry_id: str, *, entries_read: int | None = None) -> Self: + args: list[Any] = [b"XGROUP", b"SETID", key, _enc(group), _enc(entry_id)] + if entries_read is not None: + args.extend([b"ENTRIESREAD", str(entries_read).encode()]) + self._batch.custom_command(args) + self._track(_ok_to_bool) + return self + + def xgroup_delconsumer(self, key: Any, group: str, consumer: str) -> Self: + self._batch.custom_command([b"XGROUP", b"DELCONSUMER", key, _enc(group), _enc(consumer)]) + return self + + def xinfo_stream(self, key: Any, full: bool = False) -> Self: + args: list[Any] = [b"XINFO", b"STREAM", key] + if full: + args.append(b"FULL") + self._batch.custom_command(args) + self._track(_decode_xinfo) + return self + + def xinfo_groups(self, key: Any) -> Self: + self._batch.custom_command([b"XINFO", b"GROUPS", key]) + self._track(lambda r: [_decode_xinfo(g) for g in (r or [])]) + return self + + def xinfo_consumers(self, key: Any, group: str) -> Self: + self._batch.custom_command([b"XINFO", b"CONSUMERS", key, _enc(group)]) + self._track(lambda r: [_decode_xinfo(c) for c in (r or [])]) + return self + # ---- raw ---- def execute_command(self, *args: Any) -> Self: self._batch.custom_command(_enc_list(args)) @@ -945,8 +1237,9 @@ def _client(self) -> GlideClient: u = urlparse(self._servers[0]) cfg = GlideClientConfiguration( addresses=[NodeAddress(u.hostname or "localhost", u.port or 6379)], + **_glide_config_kwargs(self._servers, self._options, credentials_cls=ServerCredentials), ) - client = GlideClient.create(cfg) + client = _WrongTypeClient(GlideClient.create(cfg)) _GLIDE_SYNC_CLIENTS[self._config_key] = client return client @@ -980,8 +1273,9 @@ async def get_async_client(self, key: Any = None, *, write: bool = False) -> Asy u = urlparse(self._servers[0]) cfg = AsyncGlideClientConfiguration( addresses=[AsyncNodeAddress(u.hostname or "localhost", u.port or 6379)], + **_glide_config_kwargs(self._servers, self._options, credentials_cls=AsyncServerCredentials), ) - client = await AsyncGlideClient.create(cfg) + client = _WrongTypeClient(await AsyncGlideClient.create(cfg)) sub[self._config_key] = client return client @@ -1063,12 +1357,7 @@ def set_with_flags( nvalue = value actual_timeout = self.get_timeout_with_buffer(timeout, stampede_prevention) - if actual_timeout == 0: - return None if get else False - kw: dict[str, Any] = {} - if actual_timeout is not None: - kw["expiry"] = ExpirySet(ExpiryType.SEC, actual_timeout) if nx: kw["conditional_set"] = ConditionalChange.ONLY_IF_DOES_NOT_EXIST elif xx: @@ -1076,6 +1365,20 @@ def set_with_flags( if get: kw["return_old_value"] = True + if actual_timeout == 0: + # timeout=0 means expire immediately: run the SET unexpired so + # the nx/xx/get semantics still apply, then delete when it wrote. + result = client.set(key, _enc(nvalue), **kw) + if get: + executed = result is None if nx else (result is not None if xx else True) + else: + executed = result == "OK" + if executed: + client.delete([key]) + return result if get else result == "OK" + + if actual_timeout is not None: + kw["expiry"] = ExpirySet(ExpiryType.SEC, actual_timeout) result = client.set(key, _enc(nvalue), **kw) if get: return None if result is None else result @@ -1226,7 +1529,7 @@ def scan( count: int | None = None, _type: str | None = None, ) -> tuple[int, list[str]]: - result = self._client().scan(_enc(cursor), match=match, count=count, type=_type) + result = self._client().scan(_enc(cursor), match=match, count=count, type=_object_type(_type)) return int(_dec_str(result[0])), _dec_keys(result[1]) def iter_keys(self, pattern: str, itersize: int | None = None) -> Iterable[str]: @@ -1374,9 +1677,9 @@ def sscan( cursor: int = 0, match: str | None = None, count: int | None = None, - ) -> tuple[bytes, list]: + ) -> tuple[int, _set[Any]]: result = self._client().sscan(key, _enc(cursor), match=match, count=count) - return result[0], list(result[1]) + return int(_dec_str(result[0])), set(result[1]) def sscan_iter(self, key: str, match: str | None = None, count: int | None = None) -> Iterable[Any]: client = self._client() @@ -1400,6 +1703,10 @@ def zadd(self, key: str, mapping: Mapping[Any, float], **kwargs: Any) -> int: args.append(b"NX") elif kwargs.get("xx"): args.append(b"XX") + if kwargs.get("gt"): + args.append(b"GT") + elif kwargs.get("lt"): + args.append(b"LT") if kwargs.get("ch"): args.append(b"CH") if kwargs.get("incr"): @@ -1760,10 +2067,12 @@ def xpending( idle: int | None = None, ) -> Any: args: list[Any] = [b"XPENDING", key, _enc(group)] - if idle is not None: - args.extend([b"IDLE", str(idle).encode()]) is_range = start is not None and end is not None and count is not None if is_range: + # IDLE is only valid in the extended form, between the group + # name and the start/end/count range. + if idle is not None: + args.extend([b"IDLE", str(idle).encode()]) args.extend([_enc(start), _enc(end), str(count).encode()]) if consumer is not None: args.append(_enc(consumer)) @@ -1890,11 +2199,24 @@ def info(self, section: str | None = None) -> dict[str, Any]: def slowlog_len(self) -> int: return self._client().custom_command([b"SLOWLOG", b"LEN"]) - def slowlog_get(self, num: int | None = None) -> list[Any]: + def slowlog_get(self, num: int | None = None) -> list[dict[str, Any]]: args: list[bytes] = [b"SLOWLOG", b"GET"] if num is not None: args.append(str(num).encode()) - return self._client().custom_command(args) + raw = self._client().custom_command(args) + # Reshape each row into the dict form the other adapters return. + return [ + { + "id": entry[0], + "start_time": entry[1], + "duration": entry[2], + "command": [_dec_str(arg) for arg in (entry[3] or [])], + "client_address": _dec_str(entry[4]) if len(entry) > 4 else None, + "client_name": _dec_str(entry[5]) if len(entry) > 5 else None, + } + for entry in (raw or []) + if isinstance(entry, (list, tuple)) and len(entry) >= 4 + ] # ========================================================================= # Lock (sync). Inherits from base which uses redis-py-style lock @@ -2013,12 +2335,7 @@ async def aset_with_flags( nvalue = value actual_timeout = self.get_timeout_with_buffer(timeout, stampede_prevention) - if actual_timeout == 0: - return None if get else False - kw: dict[str, Any] = {} - if actual_timeout is not None: - kw["expiry"] = ExpirySet(ExpiryType.SEC, actual_timeout) if nx: kw["conditional_set"] = ConditionalChange.ONLY_IF_DOES_NOT_EXIST elif xx: @@ -2026,6 +2343,20 @@ async def aset_with_flags( if get: kw["return_old_value"] = True + if actual_timeout == 0: + # timeout=0 means expire immediately: run the SET unexpired so + # the nx/xx/get semantics still apply, then delete when it wrote. + result = await client.set(key, _enc(nvalue), **kw) + if get: + executed = result is None if nx else (result is not None if xx else True) + else: + executed = result == "OK" + if executed: + await client.delete([key]) + return result if get else result == "OK" + + if actual_timeout is not None: + kw["expiry"] = ExpirySet(ExpiryType.SEC, actual_timeout) result = await client.set(key, _enc(nvalue), **kw) if get: return None if result is None else result @@ -2176,7 +2507,8 @@ async def ascan( count: int | None = None, _type: str | None = None, ) -> tuple[int, list[str]]: - result = await (await self.get_async_client()).scan(_enc(cursor), match=match, count=count, type=_type) + client = await self.get_async_client() + result = await client.scan(_enc(cursor), match=match, count=count, type=_object_type(_type)) return int(_dec_str(result[0])), _dec_keys(result[1]) async def aiter_keys(self, pattern: str, itersize: int | None = None): @@ -2329,9 +2661,9 @@ async def asscan( cursor: int = 0, match: str | None = None, count: int | None = None, - ) -> tuple[bytes, list]: + ) -> tuple[int, _set[Any]]: result = await (await self.get_async_client()).sscan(key, _enc(cursor), match=match, count=count) - return result[0], list(result[1]) + return int(_dec_str(result[0])), set(result[1]) async def asscan_iter(self, key: str, match: str | None = None, count: int | None = None): client = await self.get_async_client() @@ -2356,6 +2688,10 @@ async def azadd(self, key: str, mapping: Mapping[Any, float], **kwargs: Any) -> args.append(b"NX") elif kwargs.get("xx"): args.append(b"XX") + if kwargs.get("gt"): + args.append(b"GT") + elif kwargs.get("lt"): + args.append(b"LT") if kwargs.get("ch"): args.append(b"CH") if kwargs.get("incr"): @@ -2734,10 +3070,12 @@ async def axpending( idle: int | None = None, ) -> Any: args: list[Any] = [b"XPENDING", key, _enc(group)] - if idle is not None: - args.extend([b"IDLE", str(idle).encode()]) is_range = start is not None and end is not None and count is not None if is_range: + # IDLE is only valid in the extended form, between the group + # name and the start/end/count range. + if idle is not None: + args.extend([b"IDLE", str(idle).encode()]) args.extend([_enc(start), _enc(end), str(count).encode()]) if consumer is not None: args.append(_enc(consumer)) @@ -2895,8 +3233,8 @@ class ValkeyGlideClusterAdapter(ValkeyGlideAdapter): """Cluster-mode adapter, wraps ``GlideClusterClient`` instead of standalone. Inherits the full command surface from :class:`ValkeyGlideAdapter`. - Only the client-construction hooks change. Multi-key operations must - hash to a single slot, use ``{tag}`` hash tags on related keys + Only the client-construction hooks and SCAN change. Multi-key operations + must hash to a single slot, use ``{tag}`` hash tags on related keys (matches :class:`~django_cachex.cache.resp.RespClusterCache` semantics for the other drivers). """ @@ -2912,6 +3250,53 @@ async def apipeline(self, *, transaction: bool = True) -> ValkeyGlideAsyncPipeli client = await self.get_async_client() return ValkeyGlideAsyncPipelineAdapter(client, transaction=False) + # A ``ClusterScanCursor`` can't round-trip through the protocol's int cursor, + # so drive the loop here and report one finished scan. + + def _scan_keys(self, match: str | None, count: int | None, _type: str | None) -> Iterable[str]: + client = self._client() + cursor = ClusterScanCursor() + object_type = _object_type(_type) + while not cursor.is_finished(): + cursor, keys = client.scan(cursor, match=match, count=count, type=object_type) + yield from _dec_keys(keys) + + def scan( + self, + cursor: int = 0, + match: str | None = None, + count: int | None = None, + _type: str | None = None, + ) -> tuple[int, list[str]]: + del cursor # Only ever 0: the previous call consumed the whole keyspace. + return 0, list(self._scan_keys(match, count, _type)) + + def iter_keys(self, pattern: str, itersize: int | None = None) -> Iterable[str]: + return self._scan_keys(pattern, itersize, None) + + async def _ascan_keys(self, match: str | None, count: int | None, _type: str | None): + client = await self.get_async_client() + cursor = AsyncClusterScanCursor() + object_type = _object_type(_type) + while not cursor.is_finished(): + cursor, keys = await client.scan(cursor, match=match, count=count, type=object_type) + for key in keys: + yield _dec_str(key) + + async def ascan( + self, + cursor: int = 0, + match: str | None = None, + count: int | None = None, + _type: str | None = None, + ) -> tuple[int, list[str]]: + del cursor + return 0, [key async for key in self._ascan_keys(match, count, _type)] + + async def aiter_keys(self, pattern: str, itersize: int | None = None): + async for key in self._ascan_keys(pattern, itersize, None): + yield key + def _client(self) -> Any: client = _GLIDE_SYNC_CLUSTER_CLIENTS.get(self._config_key) if client is not None: @@ -2919,8 +3304,16 @@ def _client(self) -> Any: with _GLIDE_SYNC_CLUSTER_LOCK: client = _GLIDE_SYNC_CLUSTER_CLIENTS.get(self._config_key) if client is None: - cfg = GlideClusterClientConfiguration(addresses=self._cluster_addresses()) - client = GlideClusterClient.create(cfg) + cfg = GlideClusterClientConfiguration( + addresses=self._cluster_addresses(), + **_glide_config_kwargs( + self._servers, + self._options, + credentials_cls=ServerCredentials, + include_database=False, + ), + ) + client = _WrongTypeClient(GlideClusterClient.create(cfg)) _GLIDE_SYNC_CLUSTER_CLIENTS[self._config_key] = client return client @@ -2941,8 +3334,16 @@ async def get_async_client(self, key: Any = None, *, write: bool = False) -> Any async with lock: client = sub.get(self._config_key) if client is None: - cfg = AsyncGlideClusterClientConfiguration(addresses=self._cluster_addresses_async()) - client = await AsyncGlideClusterClient.create(cfg) + cfg = AsyncGlideClusterClientConfiguration( + addresses=self._cluster_addresses_async(), + **_glide_config_kwargs( + self._servers, + self._options, + credentials_cls=AsyncServerCredentials, + include_database=False, + ), + ) + client = _WrongTypeClient(await AsyncGlideClusterClient.create(cfg)) sub[self._config_key] = client return client @@ -3088,7 +3489,6 @@ def __init__( self._sleep = sleep self._blocking = blocking self._timeout = timeout - self._initial_token = os.urandom(16).hex().encode() self._token_local: threading.local | None = threading.local() if thread_local else None self._token_shared: bytes | None = None @@ -3112,20 +3512,23 @@ def _token(self, value: bytes | None) -> None: def acquire(self, *, blocking: bool | None = None, timeout: float | None = None) -> bool: bl = self._blocking if blocking is None else blocking bt = self._timeout if timeout is None else timeout - deadline = time.monotonic() + bt if bt else None + deadline = time.monotonic() + bt if bt is not None else None kw: dict[str, Any] = {"conditional_set": ConditionalChange.ONLY_IF_DOES_NOT_EXIST} if self._lease is not None: kw["expiry"] = ExpirySet(ExpiryType.MILLSEC, int(self._lease * 1000)) while True: - result = self._client.set(self._key, self._initial_token, **kw) + # Fresh token per attempt: a reused token would let a stale + # holder release/extend a lock re-acquired by someone else. + token = os.urandom(16).hex().encode() + result = self._client.set(self._key, token, **kw) if result == "OK": - self._token = self._initial_token + self._token = token return True if not bl: return False - if deadline and time.monotonic() >= deadline: + if deadline is not None and time.monotonic() >= deadline: return False time.sleep(self._sleep) @@ -3196,7 +3599,6 @@ def __init__( self._sleep = sleep self._blocking = blocking self._timeout = timeout - self._initial_token = os.urandom(16).hex().encode() self._token_local: threading.local | None = threading.local() if thread_local else None self._token_shared: bytes | None = None @@ -3220,7 +3622,7 @@ def _token(self, value: bytes | None) -> None: async def acquire(self, *, blocking: bool | None = None, timeout: float | None = None) -> bool: bl = self._blocking if blocking is None else blocking bt = self._timeout if timeout is None else timeout - deadline = time.monotonic() + bt if bt else None + deadline = time.monotonic() + bt if bt is not None else None client = await self._adapter.get_async_client() kw: dict[str, Any] = {"conditional_set": ConditionalChange.ONLY_IF_DOES_NOT_EXIST} @@ -3228,13 +3630,16 @@ async def acquire(self, *, blocking: bool | None = None, timeout: float | None = kw["expiry"] = ExpirySet(ExpiryType.MILLSEC, int(self._lease * 1000)) while True: - result = await client.set(self._key, self._initial_token, **kw) + # Fresh token per attempt: a reused token would let a stale + # holder release/extend a lock re-acquired by someone else. + token = os.urandom(16).hex().encode() + result = await client.set(self._key, token, **kw) if result == "OK": - self._token = self._initial_token + self._token = token return True if not bl: return False - if deadline and time.monotonic() >= deadline: + if deadline is not None and time.monotonic() >= deadline: return False await asyncio.sleep(self._sleep) diff --git a/django_cachex/adapters/valkey_py.py b/django_cachex/adapters/valkey_py.py index c5de6eb..37e89af 100644 --- a/django_cachex/adapters/valkey_py.py +++ b/django_cachex/adapters/valkey_py.py @@ -100,6 +100,11 @@ def _options_key(options: dict[str, Any]) -> tuple[tuple[str, Any], ...]: return tuple(out) +def _raw_response(response: Any, **_options: Any) -> Any: + """Response callback that returns the driver's reply unparsed.""" + return response + + _VALKEY_ASYNC_POOLS: AsyncPoolsRegistry = weakref.WeakKeyDictionary() # Cluster-client caches, shared process-wide. Sync clusters are pooled by @@ -205,6 +210,10 @@ class ValkeyPyAdapter(RespAdapterProtocol): _async_pools = _VALKEY_ASYNC_POOLS + # Per-call client wrappers may be mutated (e.g. response callbacks) without + # leaking; the cluster adapter shares one client and overrides to False. + _per_call_clients: bool = True + if _VALKEY_AVAILABLE: _lib = valkey _client_class = valkey.Valkey @@ -509,9 +518,14 @@ def set_with_flags( actual_timeout = self.get_timeout_with_buffer(timeout, stampede_prevention) if actual_timeout == 0: + result = client.set(key, nvalue, nx=nx, xx=xx, get=get) if get: - return None - return False + executed = result is None if nx else (result is not None if xx else True) + else: + executed = bool(result) + if executed: + client.delete(key) + return result if get else bool(result) result = client.set(key, nvalue, ex=actual_timeout, nx=nx, xx=xx, get=get) if get: if result is None: @@ -540,9 +554,14 @@ async def aset_with_flags( actual_timeout = self.get_timeout_with_buffer(timeout, stampede_prevention) if actual_timeout == 0: + result = await client.set(key, nvalue, nx=nx, xx=xx, get=get) if get: - return None - return False + executed = result is None if nx else (result is not None if xx else True) + else: + executed = bool(result) + if executed: + await client.delete(key) + return result if get else bool(result) result = await client.set(key, nvalue, ex=actual_timeout, nx=nx, xx=xx, get=get) if get: if result is None: @@ -2612,6 +2631,11 @@ def xautoclaim( """Auto-claim pending messages that have been idle.""" client = self.get_client(key, write=True) + if justid and self._per_call_clients: + # The driver's JUSTID callback strips next_id and deleted; safe to + # override on a per-call client wrapper. + client.set_response_callback("XAUTOCLAIM", _raw_response) + result = client.xautoclaim( key, group, @@ -2622,13 +2646,16 @@ def xautoclaim( justid=justid, ) - if justid: - # redis-py returns flat list of claimed IDs (strips next_id/deleted) + if justid and not self._per_call_clients: + # Shared client (cluster): the driver callback already stripped + # next_id/deleted, so the cursor cannot be recovered. claimed: list[str] = [r.decode() if isinstance(r, bytes) else r for r in result] return ("", claimed, []) next_id = result[0].decode() if isinstance(result[0], bytes) else result[0] deleted = [d.decode() if isinstance(d, bytes) else d for d in result[2]] if len(result) > 2 else [] + if justid: + return (next_id, [r.decode() if isinstance(r, bytes) else r for r in result[1]], deleted) return (next_id, self._decode_stream_entries(result[1]), deleted) # ========================================================================= @@ -2891,6 +2918,11 @@ async def axautoclaim( """Auto-claim pending messages asynchronously.""" client = await self.get_async_client(key, write=True) + if justid and self._per_call_clients: + # The driver's JUSTID callback strips next_id and deleted; safe to + # override on a per-call client wrapper. + client.set_response_callback("XAUTOCLAIM", _raw_response) + result = await client.xautoclaim( key, group, @@ -2901,13 +2933,16 @@ async def axautoclaim( justid=justid, ) - if justid: - # redis-py returns flat list of claimed IDs (strips next_id/deleted) + if justid and not self._per_call_clients: + # Shared client (cluster): the driver callback already stripped + # next_id/deleted, so the cursor cannot be recovered. claimed: list[str] = [r.decode() if isinstance(r, bytes) else r for r in result] return ("", claimed, []) next_id = result[0].decode() if isinstance(result[0], bytes) else result[0] deleted = [d.decode() if isinstance(d, bytes) else d for d in result[2]] if len(result) > 2 else [] + if justid: + return (next_id, [r.decode() if isinstance(r, bytes) else r for r in result[1]], deleted) return (next_id, self._decode_stream_entries(result[1]), deleted) # ========================================================================= @@ -3121,7 +3156,6 @@ def _get_async_connection_pool(self, *, write: bool) -> Any: raise RuntimeError(msg) service_name, is_master, clean_url = self._parse_sentinel_url(index) - async_sentinel = self._get_async_sentinel() # Filter out parser_class - it's sync-specific and causes AttributeError on async connections pool_options: dict[str, Any] = ( @@ -3129,22 +3163,18 @@ def _get_async_connection_pool(self, *, write: bool) -> Any: if hasattr(self, "_pool_options") else {} ) - pool_options.update( - service_name=service_name, - sentinel_manager=async_sentinel, - is_master=is_master, - ) - # Key on the sentinel-aware fields plus the URL + options. The - # sentinel manager is per-loop (cached above) so we include its id in - # the key to avoid sharing a pool across two managers in the same loop. + # The key must be stable across adapter instances (asgiref hands each + # task a fresh one), so the fleet stands in for its sentinel manager. + sentinels = self._options.get("sentinels") or () key = ( self._async_sentinel_pool_class, clean_url, service_name, is_master, - id(async_sentinel), - _options_key({k: v for k, v in pool_options.items() if k != "sentinel_manager"}), + tuple(tuple(entry) for entry in sentinels), + _options_key(self._options.get("sentinel_kwargs") or {}), + _options_key(pool_options), index, ) @@ -3155,7 +3185,13 @@ def _get_async_connection_pool(self, *, write: bool) -> Any: pool = sub.get(key) if pool is None: - pool = self._async_sentinel_pool_class.from_url(clean_url, **pool_options) + pool = self._async_sentinel_pool_class.from_url( + clean_url, + service_name=service_name, + sentinel_manager=self._get_async_sentinel(), + is_master=is_master, + **pool_options, + ) sub[key] = pool return pool @@ -3172,6 +3208,10 @@ class ValkeyPyClusterAdapter(ValkeyPyAdapter): # correctly identify this adapter as cluster-shaped. _async_pool_class: builtins.type[Any] | None = None + # The cluster client is shared process-wide, so it must never be mutated + # per call (see ``ValkeyPyAdapter._per_call_clients``). + _per_call_clients: bool = False + # Subclasses must set these _cluster_class: builtins.type[Any] | None = None _async_cluster_class: builtins.type[Any] | None = None @@ -3206,16 +3246,15 @@ def _async_cluster(self) -> builtins.type[Any]: return self._async_cluster_class def _cluster_options(self) -> tuple[dict[str, Any], tuple[Any, ...]]: - """Build kwargs for the cluster constructor and a hashable cache key.""" + """Build extra kwargs for ``from_url`` and a hashable cache key. + + The server URL goes to ``from_url`` verbatim so TLS, auth, db and + query parameters survive; only OPTIONS-derived kwargs live here. + """ url = self._servers[0] - parsed_url = urlparse(url) cluster_options = { key_opt: value for key_opt, value in self._options.items() if key_opt not in self._CLIENT_ONLY_OPTIONS } - if parsed_url.hostname: - cluster_options["host"] = parsed_url.hostname - if parsed_url.port: - cluster_options["port"] = parsed_url.port return cluster_options, (self._cluster_class, url, _options_key(cluster_options)) @override @@ -3226,7 +3265,7 @@ def get_client(self, key: str | None = None, *, write: bool = False) -> Any: with self._clusters_lock: cluster = self._clusters.get(cache_key) if cluster is None: - cluster = self._cluster(**cluster_options) + cluster = self._cluster.from_url(self._servers[0], **cluster_options) self._clusters[cache_key] = cluster return _install_wrongtype_translation(cluster) @@ -3244,7 +3283,7 @@ async def get_async_client(self, key: str | None = None, *, write: bool = False) cluster = sub.get(cache_key) if cluster is None: - cluster = self._async_cluster(**cluster_options) + cluster = self._async_cluster.from_url(self._servers[0], **cluster_options) sub[cache_key] = cluster return _install_wrongtype_translation(cluster) @@ -4189,7 +4228,13 @@ async def execute(self) -> list[Any]: # type: ignore[override] @override async def reset(self) -> None: # type: ignore[override] """Discard buffered commands. ``redis.asyncio.Pipeline.reset()`` is awaitable.""" - await self._raw.reset() + reset = self._raw.reset + if inspect.iscoroutinefunction(reset): + await reset() + return + # Async ClusterPipeline has no reset(); the name resolves to the RESET + # server command. Clear the stack directly, like its own __aexit__. + self._raw._command_stack = [] __all__ = [ diff --git a/django_cachex/admin/admin.py b/django_cachex/admin/admin.py index 250576f..714a97c 100644 --- a/django_cachex/admin/admin.py +++ b/django_cachex/admin/admin.py @@ -7,7 +7,7 @@ from django.contrib.admin.utils import unquote from django.core.exceptions import PermissionDenied from django.http import HttpResponseRedirect -from django.urls import path, reverse +from django.urls import reverse from django.utils.safestring import mark_safe from .models import Cache, Key @@ -80,18 +80,6 @@ def has_delete_permission( ) -> bool: return False - def get_urls(self) -> list: - """Add custom URL patterns.""" - urls = super().get_urls() - custom_urls = [ - path( - "/change/", - self.admin_site.admin_view(self.change_view), - name="django_cachex_cache_change", - ), - ] - return custom_urls + urls - def _get_config(self) -> ViewConfig: return ViewConfig(help_messages=self._cachex_help_messages) @@ -105,7 +93,7 @@ def change_view( """Display cache details (info + slowlog combined).""" if not self.has_view_or_change_permission(request): raise PermissionDenied - return _cache_detail_view(request, object_id, self._get_config()) + return _cache_detail_view(request, unquote(object_id), self._get_config()) @admin.register(Key) @@ -228,18 +216,6 @@ class KeyAdmin(KeyAdminMixin, _KeyBase): # type: ignore[misc] ), } - def get_urls(self) -> list: - """Add custom URL patterns for key operations.""" - urls = super().get_urls() - custom_urls = [ - path( - "/change/", - self.admin_site.admin_view(self.change_view), - name="django_cachex_key_change", - ), - ] - return custom_urls + urls - def _get_config(self) -> ViewConfig: return ViewConfig(help_messages=self._cachex_help_messages) @@ -262,6 +238,13 @@ def change_view( reverse("admin:django_cachex_cache_changelist"), ) + # get_cache() raises on an unconfigured alias, so check first like add_view. + if Cache.get_by_name(cache_name) is None: + messages.error(request, f"Cache '{cache_name}' not found.") + return HttpResponseRedirect( + reverse("admin:django_cachex_cache_changelist"), + ) + return _key_detail_view(request, cache_name, key_name, self._get_config()) def add_view( diff --git a/django_cachex/admin/helpers.py b/django_cachex/admin/helpers.py index 94d8bf7..2d26173 100644 --- a/django_cachex/admin/helpers.py +++ b/django_cachex/admin/helpers.py @@ -196,7 +196,8 @@ def get_type_data( result = _fetch_type_data(cache, key, key_type, page=page) - # Add SHA1 fingerprints for CAS (compare-and-swap) protection. + # CAS fingerprints are hashed server-side, so backends without scripting + # (stock Django, LocMem, Database) get no conflict detection. if result and hasattr(cache, "eval_script"): _add_cas_fingerprints(cache, key, key_type, result) @@ -303,7 +304,11 @@ def _add_cas_fingerprints(cache: Any, key: str, key_type: str | None, result: di hash_sha1s = get_hash_field_sha1s(cache, key) result["field_entries"] = [(field, value, hash_sha1s.get(field, "")) for field, value in fields.items()] - except Exception: # noqa: BLE001 + except NotSupportedError: + # ``BaseCachex`` declares ``eval_script`` and raises, so LocMem and + # Database land here: render without CAS fingerprints. + return + except Exception: # CAS protection is best-effort. Mirror the warning emitted by # ``_key_detail_view`` (key_detail.py) so the operator knows the # next update will skip conflict detection. @@ -327,13 +332,14 @@ def get_size(cache: Any, key: str, key_type: str | None = None) -> int | None: return None def _string_size() -> int | None: - # Use STRLEN via raw client when available - try: - client = cache.get_client(write=False) - full_key = cache.make_key(key) - return client.strlen(full_key) - except NotSupportedError, AttributeError: - pass + # STRLEN over the raw client measures the stored bytes. Stock Django + # backends have no get_client; BaseCachex declares it and raises. + if hasattr(cache, "get_client"): + try: + client = cache.get_client(write=False) + return client.strlen(cache.make_key(key)) + except NotSupportedError: + pass # Fallback: compute Python object size (e.g. LocMemCache). Decode # failures for stale data must not break the size column, return None # so the row still renders and the user can delete the broken key. diff --git a/django_cachex/admin/models.py b/django_cachex/admin/models.py index 5c06a14..6516500 100644 --- a/django_cachex/admin/models.py +++ b/django_cachex/admin/models.py @@ -1,4 +1,5 @@ from typing import Any +from urllib.parse import quote, unquote from django.conf import settings from django.core.cache import InvalidCacheBackendError @@ -131,15 +132,20 @@ def __str__(self) -> str: @classmethod def make_pk(cls, cache_name: str, key_name: str) -> str: - """Create a primary key from cache name and key name.""" - return f"{cache_name}:{key_name}" + """Create a primary key from cache name and key name. + + The cache name is percent-encoded so a ``:`` in it can't be confused + with the separator; key names may contain ``:`` freely because + ``parse_pk`` splits on the first separator only. + """ + return f"{quote(cache_name, safe='')}:{key_name}" @classmethod def parse_pk(cls, pk: str) -> tuple[str, str]: """Parse a primary key into (cache_name, key_name).""" parts = pk.split(":", 1) if len(parts) == 2: - return parts[0], parts[1] + return unquote(parts[0]), parts[1] return "", pk @classmethod diff --git a/django_cachex/admin/queryset.py b/django_cachex/admin/queryset.py index feb5c87..7b288b0 100644 --- a/django_cachex/admin/queryset.py +++ b/django_cachex/admin/queryset.py @@ -12,11 +12,11 @@ from django.conf import settings from django.contrib import admin, messages from django.contrib.admin import ShowFacets -from django.contrib.admin.utils import unquote from django.http import HttpResponseRedirect from django.urls import reverse from django.utils import timezone from django.utils.html import format_html +from django.utils.http import urlencode from django.utils.safestring import mark_safe from django.utils.timesince import timeuntil from django.utils.translation import gettext_lazy as _ @@ -55,6 +55,7 @@ class CacheQuerySet: model = Cache ordered = True + totally_ordered = True # read by ChangeList since Django 6.1 db = "default" def __init__(self, data: list[Cache] | None = None): @@ -283,7 +284,7 @@ def support_display(self, obj: Cache) -> str: def keys_link(self, obj: Cache) -> str: if obj.support_level != "cachex": return mark_safe('-') - url = reverse("admin:django_cachex_key_changelist") + f"?cache={obj.name}" + url = reverse("admin:django_cachex_key_changelist") + "?" + urlencode({"cache": obj.name}) return format_html('{}', url, _("List Keys")) @@ -293,6 +294,10 @@ def keys_link(self, obj: Cache) -> str: KEY_TYPES = tuple(KeyType) +# Cap for the user-supplied ``?count=`` SCAN hint; uncapped, one changelist +# request could walk the whole keyspace in a single blocking SCAN. +MAX_SCAN_COUNT = 1000 + # Inline styles for type badges (theme-agnostic) _TYPE_STYLES: dict[str, str] = { "string": "background:#dbeafe;color:#1d4ed8;", @@ -314,6 +319,7 @@ class KeyQuerySet: model = Key ordered = True + totally_ordered = True # read by ChangeList since Django 6.1 db = "default" def __init__( @@ -365,12 +371,17 @@ def __getitem__(self, key: int | slice) -> KeyQuerySet | Key: return self._data[key] def filter(self, *args: Any, **kwargs: Any) -> KeyQuerySet: - """Filter keys. pk__in builds Key objects from composite PKs directly.""" + """Filter keys. pk__in builds Key objects from composite PKs directly. + + The pk values come from the action checkboxes, which carry the raw + ``str(obj.pk)``: no admin ``unquote()`` here, or keys containing + ``_XX`` sequences would be mangled. + """ clone = self._clone() if "pk__in" in kwargs: data = [] for pk in kwargs["pk__in"]: - cache_name, key_name = Key.parse_pk(unquote(str(pk))) + cache_name, key_name = Key.parse_pk(str(pk)) if cache_name: data.append(Key.from_cache_key(cache_name, key_name)) clone._data = data @@ -469,9 +480,16 @@ class KeyAdminMixin: show_facets = ShowFacets.NEVER show_full_result_count: ClassVar[bool] = False - def get_actions(self, request: HttpRequest) -> dict: - """Remove Django's built-in delete_selected (KeyQuerySet doesn't support it).""" - actions = super().get_actions(request) # type: ignore[misc] # ty: ignore[unresolved-attribute] + def get_actions(self, request: HttpRequest, action_location: Any = None) -> dict: + """Remove Django's built-in delete_selected (KeyQuerySet doesn't support it). + + ``action_location`` must appear literally in the signature: Django + 6.1's deprecation shim inspects it via ``get_func_args`` and emits + ``RemovedInDjango70Warning`` when absent. Django 6.0 calls without + it, so it is forwarded only when given. + """ + kwargs = {} if action_location is None else {"action_location": action_location} + actions = super().get_actions(request, **kwargs) # type: ignore[misc] # ty: ignore[unresolved-attribute] actions.pop("delete_selected", None) return actions @@ -613,7 +631,7 @@ def changelist_view( except Exception as exc: # noqa: BLE001 messages.error(request, f"Error clearing cache: {exc}") return HttpResponseRedirect( - reverse("admin:django_cachex_key_changelist") + f"?cache={cache_name}", + reverse("admin:django_cachex_key_changelist") + "?" + urlencode({"cache": cache_name}), ) # Help handling @@ -630,7 +648,7 @@ def changelist_view( except ValueError, TypeError: cursor = 0 try: - count = max(1, int(request.GET.get("count", 100))) + count = min(MAX_SCAN_COUNT, max(1, int(request.GET.get("count", 100)))) except ValueError, TypeError: count = 100 request._cachex_cursor = cursor # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] @@ -651,13 +669,20 @@ def changelist_view( @admin.action(description=_("Delete selected keys"), permissions=["delete"]) def delete_selected_keys(self, request: HttpRequest, queryset: KeyQuerySet) -> None: deleted = 0 + errors: list[str] = [] for key_obj in queryset: - with contextlib.suppress(Exception): + try: cache = get_cache(key_obj.cache_name) cache.delete(key_obj.key_name) deleted += 1 + except Exception as exc: # noqa: BLE001 + errors.append(f"'{key_obj.key_name}': {exc}") if deleted: messages.success(request, f"Successfully deleted {deleted} key(s).") + if errors: + shown = "; ".join(errors[:3]) + more = f" (+{len(errors) - 3} more)" if len(errors) > 3 else "" + messages.error(request, f"Failed to delete {len(errors)} key(s): {shown}{more}") # ------------------------------------------------------------------ # Display columns diff --git a/django_cachex/admin/templates/admin/django_cachex/cache/change_form.html b/django_cachex/admin/templates/admin/django_cachex/cache/change_form.html index df1099b..6ffe69e 100644 --- a/django_cachex/admin/templates/admin/django_cachex/cache/change_form.html +++ b/django_cachex/admin/templates/admin/django_cachex/cache/change_form.html @@ -26,7 +26,7 @@
    {% if cache_obj.support_level == "cachex" %}
  • - {% trans 'List Keys' %} + {% trans 'List Keys' %}
    • {% endif %}
  • diff --git a/django_cachex/admin/templates/admin/django_cachex/key/_key_detail_actions.html b/django_cachex/admin/templates/admin/django_cachex/key/_key_detail_actions.html index c9f0fc2..e2b04d7 100644 --- a/django_cachex/admin/templates/admin/django_cachex/key/_key_detail_actions.html +++ b/django_cachex/admin/templates/admin/django_cachex/key/_key_detail_actions.html @@ -1,5 +1,5 @@ {% load i18n %}
    - {% trans 'Back' %} + {% trans 'Back' %} {% trans 'Delete' %}
    diff --git a/django_cachex/admin/templates/admin/django_cachex/key/add_form.html b/django_cachex/admin/templates/admin/django_cachex/key/add_form.html index 74ef30a..4238ccf 100644 --- a/django_cachex/admin/templates/admin/django_cachex/key/add_form.html +++ b/django_cachex/admin/templates/admin/django_cachex/key/add_form.html @@ -15,7 +15,7 @@
    {% trans 'Home' %} › {% trans 'Caches' %} - › {{ cache_name }} + › {{ cache_name }} › {% trans 'Add Key' %}
    {% endblock %} @@ -25,7 +25,7 @@ @@ -67,7 +67,7 @@

    {% trans 'Key Details' %}

    - {% trans 'Back' %} + {% trans 'Back' %}
    diff --git a/django_cachex/admin/templates/admin/django_cachex/key/change_form.html b/django_cachex/admin/templates/admin/django_cachex/key/change_form.html index 88cbf1a..4ee8d43 100644 --- a/django_cachex/admin/templates/admin/django_cachex/key/change_form.html +++ b/django_cachex/admin/templates/admin/django_cachex/key/change_form.html @@ -16,7 +16,7 @@
    {% trans 'Home' %} › {% trans 'Caches' %} - › {{ cache_name }} + › {{ cache_name }} › {{ key }}
    {% endblock %} diff --git a/django_cachex/admin/templates/admin/django_cachex/key/change_list.html b/django_cachex/admin/templates/admin/django_cachex/key/change_list.html index 420a79b..7d90ef0 100644 --- a/django_cachex/admin/templates/admin/django_cachex/key/change_list.html +++ b/django_cachex/admin/templates/admin/django_cachex/key/change_list.html @@ -1,11 +1,11 @@ {% extends "admin/change_list.html" %} -{% load i18n %} +{% load i18n admin_urls %} {% block breadcrumbs %}
    {% trans 'Home' %} › {% trans 'Caches' %} - › {{ cache_name }} + › {{ cache_name }} › {% trans 'Keys' %}
    {% endblock %} @@ -14,10 +14,10 @@ @@ -38,10 +38,10 @@