diff --git a/.secrets.baseline b/.secrets.baseline index 44d94eac..7d0674c7 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -215,7 +215,7 @@ "filename": "src/cachekit/cache_handler.py", "hashed_secret": "5baa61e4c9b93f3f0682250b6cf8331b7ee68fd8", "is_verified": false, - "line_number": 719 + "line_number": 982 } ], "src/cachekit/serializers/interop_serializer.py": [ @@ -2020,5 +2020,5 @@ } ] }, - "generated_at": "2026-11-01T05:43:36Z" + "generated_at": "2026-11-02T01:40:34Z" } diff --git a/README.md b/README.md index 51526fe2..50ccc403 100644 --- a/README.md +++ b/README.md @@ -204,7 +204,7 @@ Setting `CACHEKIT_MASTER_KEY` instead of passing `master_key=` means every prese > > 🔒 `@cache.secure` forces `integrity_checking=True` — passing `integrity_checking=False` raises `ConfigurationError` at decoration, including as an override next to `config=DecoratorConfig.secure(...)`. Its encryption is fixed the same way: `encryption=` raises `ConfigurationError` on `@cache.secure` and next to `config=DecoratorConfig.secure(...)`, so pass `fail_closed=` and the other encryption options to the preset itself. `@cache.secure` also rejects `config=`; the RORO form is `@cache(config=DecoratorConfig.secure(...))`. > -> **Default TTL** follows the cross-SDK [intent-preset spec](https://github.com/cachekit-io/protocol/blob/main/spec/intent-presets.md#default-ttl) (`@cache.io` 3600 s) — the same numbers as cachekit-rs and cachekit-ts. `ttl=` overrides it; `ttl=None` is the explicit never-expire opt-in ([details](docs/configuration.md#intent-presets)). +> **Default TTL** follows the cross-SDK [intent-preset spec](https://github.com/cachekit-io/protocol/blob/main/spec/intent-presets.md#default-ttl) (`@cache.io` 3600 s) — the same numbers as cachekit-rs and cachekit-ts. `ttl=` overrides it; `ttl=None` is the explicit no-TTL opt-in: never-expire on Redis, Memcached and File, while CachekitIO stores the entry for your plan's maximum TTL and keeps it forever only on a plan without a maximum, or where none applied when the entry was written ([details](docs/configuration.md#intent-presets), [plan maximums](docs/backends/cachekitio.md#ttl-range-and-plan-maximum)). > > **Cross-process L1 eviction** is not a preset feature: every preset's `invalidate_cache()` deletes from L1 and L2 alike, and evicting other processes' L1 copies is one process-wide opt-in, `CACHEKIT_INVALIDATION_LISTENER_ENABLED` ([L1 invalidation](docs/features/l1-invalidation.md#cross-process-l1-eviction)). > diff --git a/docs/api-reference.md b/docs/api-reference.md index 68b9c5a4..f039422e 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -82,7 +82,7 @@ def your_function(args): #### Core Parameters -- **`ttl`** (`int | None`, default: `None` on bare `@cache`; each preset sets its own — see the [preset matrix](configuration.md#intent-presets)) - Cache time-to-live in seconds. `None` = no expiration; on a preset that is an explicit opt-in. +- **`ttl`** (`int | None`, default: `None` on bare `@cache`; each preset sets its own — see the [preset matrix](configuration.md#intent-presets)) - Cache time-to-live in seconds. `None` = no TTL: no expiration on Redis, Memcached and File; one year in L1-only mode (`backend=None`); on CachekitIO the server stores the entry for your plan's maximum TTL ([plan maximums](backends/cachekitio.md#ttl-range-and-plan-maximum)). On a preset it is an explicit opt-in. - **`namespace`** (`str | None`, default: `None`) - Cache key prefix for organization - **`serializer`** (`str | SerializerProtocol`, default: `"default"`) - Serializer name (`"default"`, `"std"`, `"auto"`, `"arrow"`, `"orjson"`) or `SerializerProtocol` instance - **`integrity_checking`** (`bool`, default: `True`; any non-bool raises `ConfigurationError`) - Enable xxHash3-64 checksums for corruption detection (non-cryptographic — detects bit rot and storage bugs, NOT tampering; tamper resistance requires encryption). Behaviour when a writer and reader disagree on this flag: [cross-config reads](serializers/auto.md#cross-config-reads-integrity_checking-mismatch) @@ -222,7 +222,7 @@ def get_exchange_rates(): - Inherits all production-grade reliability features: circuit breaker, backpressure, full monitoring - L1 in-memory cache is enabled — hot data is served from process memory without an HTTP round-trip (each L1 hit deserializes the stored bytes) -- Default `ttl=3600`; with that default the stale-while-revalidate window is also 3600 s. `ttl=None` disables both expiry and SWR. +- Default `ttl=3600`; with that default the stale-while-revalidate window is also 3600 s. `ttl=None` sends no TTL and disables SWR: the server stores the entry for your plan's maximum TTL, and never expires it only on a plan without a maximum, or where none applied when the entry was written ([plan maximums](backends/cachekitio.md#ttl-range-and-plan-maximum)). - Standard `ttl`, `namespace`, `serializer`, and other `@cache(...)` kwargs are all supported as overrides ### Health Check Methods diff --git a/docs/backends/cachekitio.md b/docs/backends/cachekitio.md index fb8d360f..254ed9ba 100644 --- a/docs/backends/cachekitio.md +++ b/docs/backends/cachekitio.md @@ -135,6 +135,44 @@ async with backend.acquire_lock("my-key", timeout=30.0, blocking_timeout=5.0) as The `@cache` decorators use this automatically on cache miss for **async** functions — see [Distributed Locking](../features/distributed-locking.md). +## TTL Range and Plan Maximum + +A TTL is 1 to 7,776,000 seconds (90 days), and `ttl + stale_ttl` stays within the same 7,776,000. +The decorator rejects a larger `ttl + stale_ttl` at decoration time. Your plan then sets the +longest an entry is kept, its stale window included: + +| Plan | Maximum TTL | +| :--- | :--- | +| Free | 86,400 s (24 hours) | +| Startup | 2,592,000 s (30 days) | +| Professional | 7,776,000 s (90 days) | +| Enterprise | None | + +A longer TTL is clamped to the maximum, not rejected, and the stale window gets only what the +maximum leaves after the fresh TTL. A write without a TTL (`ttl=None`) is stored for the plan +maximum. An entry never expires only on a plan without a maximum, or where none applied when +it was written: a write whose plan the service could not determine is stored as stated, and an +entry written before the maximum applied keeps its timing until its next write. `refresh_ttl` is +clamped the same way. + +The server reports the result in the `PUT` response's `X-CacheKit-Fresh-For`, and the decorator +bounds its in-process (L1) copy of the value it wrote by it: the copy lives at most +`min(ttl, fresh_for)` from the response, so a clamped write is never served from L1 past the +plan maximum. A malformed or `0` value keeps no L1 copy; a server that sends no header leaves the +copy at `ttl`. The `PATCH` response of `refresh_ttl` does not report the applied TTL, so it never +lengthens an L1 copy. A `CachekitIOBackend` subclass that overrides `set` keeps this bound as long +as its override calls `super().set(...)` on the same thread. The same holds for a wrapper or proxy +backend whose `set` calls a `CachekitIOBackend`'s `set` on the same thread, provided the decorator +can find that backend at its first L2 write: in an instance or class attribute (`__slots__` included), or in a dict, +list, tuple or set there (a router or a shard ring), at most three wrappers down. A wrapper too +large to search is assumed to hold one. A wrapper that holds its backend only in a closure, +or creates it on first use (even into an attribute: the search runs before the first write and its +answer is kept), is not found: its L1 copies keep `ttl`, and the decorator logs this once per +wrapper class. Hold the backend in an attribute from construction to get the bound. A wrapper that delegates on another thread (a timeout wrapper with its own executor) must +run the call under `contextvars.copy_context().run`; otherwise its L1 copies keep `ttl` and ignore +Fresh-For. A subclass that writes any other way gives the decorator no response to read, so it +keeps no L1 copy of that write. + ## TTL Inspection (async only) ```python notest diff --git a/docs/configuration.md b/docs/configuration.md index fe19f636..bad7bae7 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -197,14 +197,14 @@ def report(): Rules and behavior: -- Requires a positive `ttl`; `ttl + stale_ttl` is capped at 2,592,000 s (30 days). Violations raise `ConfigurationError` at decoration time. +- Requires a positive `ttl`; `ttl + stale_ttl` is capped at 7,776,000 s (90 days, the protocol's TTL range). Violations raise `ConfigurationError` at decoration time. A plan whose [maximum TTL](backends/cachekitio.md#ttl-range-and-plan-maximum) is lower clamps the write server-side instead of rejecting it: the fresh TTL is kept first and the stale window gets what is left. - **CachekitIO only, known at decoration time** — `@cache.io` or an explicit `backend=CachekitIOBackend()`. Other backends have no read-side freshness signal and raise `ConfigurationError` if `stale_ttl` is set; so does a CachekitIO backend resolved lazily from `CACHEKIT_API_KEY` under another preset (the remaining-freshness bound below still applies to its reads). - Concurrent stale hits trigger at most one revalidation: per-process dedup plus (async functions) a non-blocking distributed lease on the backend's lock. Contested = serve stale, don't wait. - At most 32 revalidations per decorated function run at once. A stale hit that finds the limit reached serves stale without revalidating, and a later hit retries. An async revalidation stranded on a stopped event loop (sync code that calls `loop.run_until_complete` and leaves the loop stopped, or closes it without cancelling its tasks) gives up its turn 30 seconds after it was scheduled and is cancelled, so a stale hit on another loop can revalidate that key again. If its loop resumes, it stores nothing, even when your function swallows the cancellation and returns. A write it had already sent before its loop stopped is not recalled, and can land after a newer revalidation's. - A failed background recompute never reaches the caller: the entry keeps serving stale until its hard eviction bound, after which the next call takes the ordinary synchronous miss path. It logs a WARNING `SWR revalidation failed`, as does a revalidation skipped because the call's arguments cannot be deep-copied (`SWR revalidation skipped`) or one that could not be scheduled. It names the function by the same digest as the L1-only refresh WARNING. Each fires at most once a minute per function and carries the count since the last one; the occurrences in between log at DEBUG. - The background recompute runs with a **snapshot of the caller's `contextvars`** (contextvar-based tenant extraction works), but outside the request otherwise — don't rely on other request-scoped resources (open sessions, connections) inside functions that enable SWR. - Stale values are never written to the L1 in-memory cache, and stale reads still count as cache **hits** for metered-misses billing. -- On the CachekitIO backend, every read (SWR-configured or not) also carries the server's remaining freshness (`X-CacheKit-Fresh-For`, [protocol spec](https://github.com/cachekit-io/protocol/blob/main/spec/saas-api.md#remaining-freshness)): an L2 hit backfilled into L1 lives at most `min(ttl, remaining)` locally (with `ttl=None`, L1's own 300-second default lifetime, capped by `remaining`), so a value read near the end of its server-side freshness window is never served fresh from L1 past the server's bound. Pre-signal servers omit the header and behavior is unchanged. +- On the CachekitIO backend, every read (SWR-configured or not) also carries the server's remaining freshness (`X-CacheKit-Fresh-For`, [protocol spec](https://github.com/cachekit-io/protocol/blob/main/spec/saas-api.md#remaining-freshness)): an L2 hit backfilled into L1 lives at most `min(ttl, remaining)` locally (with `ttl=None`, L1's own 300-second default lifetime, capped by `remaining`), so a value read near the end of its server-side freshness window is never served fresh from L1 past the server's bound. Pre-signal servers omit the header and behavior is unchanged. The same bound applies to the copy a call keeps of a value it just wrote: the `PUT` response carries `X-CacheKit-Fresh-For` too, so that copy lives at most `min(ttl, fresh_for)` from the response (shorter when the server clamped the TTL to the plan maximum), a malformed or `0` value keeps no L1 copy, and an absent header leaves it at `ttl`. ### File Backend Environment Variables @@ -392,7 +392,7 @@ def secure_function(): | `secure()` | 600 s | ❌ | 100 MB | Function runs, no error | AES-256-GCM encryption required; refuses `backend=None`, so no L1-only SWR | | `io()` | 3600 s | ✓ | 100 MB | Function runs, no error | CachekitIO managed SaaS backend (closed beta — [request access](https://cachekit.io)); past-TTL [SWR](#stale-while-revalidate-stale_ttl) default-on (`stale_ttl = ttl`) | -**Default TTL** is fixed by the cross-SDK [intent-preset spec](https://github.com/cachekit-io/protocol/blob/main/spec/intent-presets.md#default-ttl) so a `production` entry expires at the same moment in Python, Rust and TypeScript; `dev()` / `test()` are SDK-local, Python-only presets and take `minimal`'s 300 s. Pass `ttl=` to override, or `ttl=None` to opt in to never-expire explicitly. The spec forbids a process-wide TTL override, so there is no `CACHEKIT_DEFAULT_TTL` (the name is reserved and ignored). Two consequences of a finite default: an entry this process writes also lives in its L1 for the same TTL (previously L1's own 300 s when no `ttl` was set), and the presets' `swr_enabled` / `stale_ttl` features — which need a positive `ttl` — are now active without an explicit `ttl=`. +**Default TTL** is fixed by the cross-SDK [intent-preset spec](https://github.com/cachekit-io/protocol/blob/main/spec/intent-presets.md#default-ttl) so a `production` entry expires at the same moment in Python, Rust and TypeScript; `dev()` / `test()` are SDK-local, Python-only presets and take `minimal`'s 300 s. Pass `ttl=` to override, or `ttl=None` to opt in to no TTL explicitly: never-expire, except on CachekitIO, where the server applies your plan's [maximum TTL](backends/cachekitio.md#ttl-range-and-plan-maximum). The spec forbids a process-wide TTL override, so there is no `CACHEKIT_DEFAULT_TTL` (the name is reserved and ignored). Two consequences of a finite default: an entry this process writes also lives in its L1 for the same TTL (previously L1's own 300 s when no `ttl` was set), and the presets' `swr_enabled` / `stale_ttl` features — which need a positive `ttl` — are now active without an explicit `ttl=`. **Invalidation is not a preset feature.** Every preset's `invalidate_cache()` deletes from L1 and L2 alike. Evicting other processes' L1 copies is the process-wide `CACHEKIT_INVALIDATION_LISTENER_ENABLED` ([Environment Variables](#environment-variables)), off by default and set by no preset ([Cross-Process L1 Eviction](features/l1-invalidation.md#cross-process-l1-eviction)). On the tenant-scoped Redis backend, an entry with `ttl=None` or a TTL above 7 days outlives the key registry's tracking set, so call `invalidate_cache()` before you retire such a function (**Set lifetime** under [Whole-Function Invalidation](features/l1-invalidation.md#whole-function-invalidation)). diff --git a/docs/features/l1-invalidation.md b/docs/features/l1-invalidation.md index 0b4ac9fa..db397504 100644 --- a/docs/features/l1-invalidation.md +++ b/docs/features/l1-invalidation.md @@ -268,7 +268,7 @@ Things to know: **Tenant scope:** with the tenant-scoped Redis backend (env auto-detection, or `RedisBackendProvider(...).get_shared_backend()`), each tenant's entries live under its own `t:{tenant}:` prefix. `invalidate_cache()` — with or without arguments — deletes only the L2 entries of the tenant set in `tenant_context` for the calling context (`default` when none is set); other tenants' entries stay cached and tracked. L1 is not tenant-scoped: within a process, all tenants share one L1 entry per cache key, so a tenant can be served the value another tenant cached, and `invalidate_cache()` evicts that entry for every tenant. The exception is an encrypted cache: its AAD [binds the `t:{tenant}:` prefix](zero-knowledge-encryption.md#cache-key-binding), so an L1 entry another tenant cached fails authentication for this one, is skipped as a miss, and the read goes on to L2. Disable L1 on functions whose results differ by tenant: `@cache(..., l1_enabled=False)`, or with a preset `@cache.production(..., l1_enabled=False)`, which keeps the preset's other L1 settings. -**Writes in flight.** An invalidation does not stop a write that began before it. A call that missed the cache before an invalidation still stores its result after it, and so does a `@cache.io` [`stale_ttl`](../configuration.md#stale-while-revalidate-stale_ttl) refresh. The value it stores was computed no earlier than the call started, and it expires by its TTL. With `ttl=None` that is never for an L2 entry, and one year for an entry in L1-only mode. With a backend, the store is two steps, L2 then L1, so an invalidation that arrives between them, from this process or [announced](#cross-process-l1-eviction) by another, can leave the value in this process's L1 alone. Likewise, a read that fetched the old value from L2 just before an invalidation can copy it into this process's L1 just after. Either copy expires by its L1 TTL. In L1-only mode a background SWR refresh never lands after an invalidation: it is dropped when the entry it refreshes has been removed. A call that starts after an invalidation in this process never joins a miss that started before it ([in-process single-flight](distributed-locking.md#in-process-single-flight-before-the-lock)): it runs its own. +**Writes in flight.** An invalidation does not stop a write that began before it. A call that missed the cache before an invalidation still stores its result after it, and so does a `@cache.io` [`stale_ttl`](../configuration.md#stale-while-revalidate-stale_ttl) refresh. The value it stores was computed no earlier than the call started, and it expires by its TTL. With `ttl=None` that is never for an L2 entry (on CachekitIO, your plan's [maximum TTL](../backends/cachekitio.md#ttl-range-and-plan-maximum)), and one year for an entry in L1-only mode. With a backend, the store is two steps, L2 then L1, so an invalidation that arrives between them, from this process or [announced](#cross-process-l1-eviction) by another, can leave the value in this process's L1 alone. Likewise, a read that fetched the old value from L2 just before an invalidation can copy it into this process's L1 just after. Either copy expires by its L1 TTL. In L1-only mode a background SWR refresh never lands after an invalidation: it is dropped when the entry it refreshes has been removed. A call that starts after an invalidation in this process never joins a miss that started before it ([in-process single-flight](distributed-locking.md#in-process-single-flight-before-the-lock)): it runs its own. ### Upgrading: Custom `key=` Keys Carry the Function's Identity diff --git a/src/cachekit/backends/base.py b/src/cachekit/backends/base.py index 7bdc2ebe..a98324ca 100644 --- a/src/cachekit/backends/base.py +++ b/src/cachekit/backends/base.py @@ -12,11 +12,21 @@ from collections.abc import Callable, Iterable from contextlib import AbstractAsyncContextManager +from contextvars import ContextVar from typing import Any, BinaryIO, Optional, Protocol, runtime_checkable # Re-export BackendError for convenience (public API) from cachekit.backends.errors import BackendError # noqa: F401 +# Write freshness (API-103). A backend with ``_reports_write_deadline = True`` appends to this list, on each +# successful ``set`` while the decorator holds one open, ``(received, fresh_until)``: the ``time.time()`` its +# 2xx arrived, and the instant the server's freshness for the written value runs out (``received`` plus the +# remaining freshness the response reports, ``math.inf`` when it reports none). The decorator bounds its L1 +# copy of the value from them (``cache_handler._written_copy_expiry``). A list, not a value: a write run by +# ``asyncio.to_thread`` sees a copy of the context, which shares the list but not a rebinding. Private, like +# ``_set_tracked``: not part of the backend contract, and free to change. +_WRITE_DEADLINES: ContextVar[Optional[list[tuple[float, float]]]] = ContextVar("cachekit_write_deadlines", default=None) + @runtime_checkable class BaseBackend(Protocol): diff --git a/src/cachekit/backends/cachekitio/backend.py b/src/cachekit/backends/cachekitio/backend.py index 981d110b..98d42a2b 100644 --- a/src/cachekit/backends/cachekitio/backend.py +++ b/src/cachekit/backends/cachekitio/backend.py @@ -23,6 +23,7 @@ from urllib3.exceptions import ClosedPoolError from cachekit.backends._uninterrupted import _await_uninterrupted, _clear_interrupted, _raised_during_request +from cachekit.backends.base import _WRITE_DEADLINES from cachekit.backends.cachekitio.client import ClientLease, lease_http_client from cachekit.backends.cachekitio.config import CachekitIOBackendConfig from cachekit.backends.cachekitio.error_handler import HTTPStatusError, HTTPTransportError, classify_http_error @@ -81,7 +82,8 @@ def _log_release_failure(lock_key: str, exc: BaseException) -> None: STALE_TTL_HEADER = "X-CacheKit-Stale-TTL" FRESHNESS_HEADER = "X-CacheKit-Freshness" FRESH_FOR_HEADER = "X-CacheKit-Fresh-For" -# A Fresh-For value never exceeds the 30-day TTL cap, which is seven digits. +# The Fresh-For header's own 30-day cap (seven digits), set by the spec independently of the +# 90-day TTL range: it only ever shortens a local copy's bound. Never derive one from the other. _FRESH_FOR_MAX_S = 2_592_000 _FRESH_FOR_MAX_DIGITS = 7 @@ -160,7 +162,7 @@ def _parse_fresh_for(value: str | None) -> int | None: """Map an ``X-CacheKit-Fresh-For`` value to seconds (spec/saas-api.md#remaining-freshness). Absent = pre-signal server → None (legacy behavior: no bound). Anything but 1-7 ASCII - digits at most the 30-day TTL cap = drift → 0 (do not extend local service — the + digits at most the header's 30-day cap = drift → 0 (do not extend local service — the conservative action, mirroring the unrecognized-freshness → stale rule). The shape check runs before ``int()``, which accepts ``+5``, ``1_0``, whitespace and non-ASCII digits. Takes the header string, not a response, so it outlives the HTTP client. @@ -627,7 +629,9 @@ def set(self, key: str, value: bytes, ttl: int | None = None, stale_ttl: int | N Args: key: Cache key value: Bytes to cache - ttl: Time-to-live in seconds (optional) + ttl: Time-to-live in seconds (optional). Omitted, the entry gets the plan's + maximum TTL, and never expires only on a plan without a maximum, or where + none applied when the entry was written. stale_ttl: Stale-grace window in seconds past the fresh TTL (SWR). Only honoured alongside an explicit ``ttl``; pre-SWR servers ignore it. @@ -635,7 +639,27 @@ def set(self, key: str, value: bytes, ttl: int | None = None, stale_ttl: int | N Raises: BackendError: If operation fails """ - self._request_sync("PUT", self._encode_key(key), body=value, headers=self._set_headers(ttl, stale_ttl)) + response = self._request_sync("PUT", self._encode_key(key), body=value, headers=self._set_headers(ttl, stale_ttl)) + seen = _WRITE_DEADLINES.get() + if seen is not None: + seen.append(self._write_freshness(response)) + + # set reports its write's freshness through _WRITE_DEADLINES; see cache_handler._written_copy_expiry. + _reports_write_deadline = True + + @staticmethod + def _write_freshness(response: BaseHTTPResponse) -> tuple[float, float]: + """``(received, fresh_until)`` for a value this PUT stored (spec/saas-api.md#remaining-freshness). + + The decorator bounds its L1 copy of the value by them (API-103): the server may have clamped + ``ttl`` to the plan's maximum. Both are timed from the 2xx's receipt, on the thread that received + it, so a later hop back to the event loop cannot restart the bound. ``X-CacheKit-Fresh-For`` is + read with the GET grammar (``_parse_fresh_for``): absent (the stated TTL was applied) gives + ``math.inf``, malformed gives the receipt itself. + """ + received = time.time() + fresh_for = CachekitIOBackend._fresh_for(response) + return received, math.inf if fresh_for is None else received + fresh_for @staticmethod def _set_headers(ttl: int | None, stale_ttl: int | None) -> dict[str, str]: @@ -819,7 +843,9 @@ async def set_async(self, key: str, value: bytes, ttl: int | None = None, stale_ Args: key: Cache key value: Bytes to cache - ttl: Time-to-live in seconds (optional) + ttl: Time-to-live in seconds (optional). Omitted, the entry gets the plan's + maximum TTL, and never expires only on a plan without a maximum, or where + none applied when the entry was written. stale_ttl: Stale-grace window in seconds past the fresh TTL (SWR). Only honoured alongside an explicit ``ttl``; pre-SWR servers ignore it. @@ -1136,7 +1162,10 @@ async def get_ttl(self, key: str) -> int | None: key: Cache key Returns: - TTL in seconds, None if key doesn't exist or has no expiry + Seconds until the entry is evicted, its stale window included; None if the key + doesn't exist, the request failed, or the entry has no expiry (stored without a TTL + on a plan without a maximum, or where none applied when it was written, which can + happen on any plan) Raises: BackendError: If the key is reserved (see ``_encode_key``); encoded outside the @@ -1153,6 +1182,9 @@ async def get_ttl(self, key: str) -> int | None: async def refresh_ttl(self, key: str, ttl: int) -> bool: """Refresh/update TTL for existing key. + The ``PATCH`` response says nothing about the applied TTL (the server may clamp it to the + plan maximum), so the decorator never derives a local (L1) bound from it (API-104). + Args: key: Cache key ttl: New TTL in seconds diff --git a/src/cachekit/cache_handler.py b/src/cachekit/cache_handler.py index caae1f47..d23dd58a 100644 --- a/src/cachekit/cache_handler.py +++ b/src/cachekit/cache_handler.py @@ -7,18 +7,20 @@ from __future__ import annotations import asyncio +import collections import contextlib import contextvars import sys import threading import types import warnings -from collections.abc import Callable, Iterator +from collections.abc import Callable, Iterable, Iterator from typing import TYPE_CHECKING, Any, BinaryIO, NamedTuple, Optional, Protocol, TypeGuard, Union, runtime_checkable from pydantic import SecretBytes, SecretStr from cachekit.backends.base import ( + _WRITE_DEADLINES, BackendError, BaseBackend, BufferHandle, @@ -50,6 +52,7 @@ from cachekit.hash_utils import redact_cache_key, redact_error_for_log from cachekit.interop import InteropError from cachekit.key_generator import CacheKeyGenerator +from cachekit.l1_cache import DEFAULT_L1_TTL_SECONDS from cachekit.serializers.auto_serializer import AutoSerializer from cachekit.serializers.base import ( EnvelopeShapeError, @@ -399,6 +402,228 @@ def _fuses_tracked_write(backend: object) -> bool: return True +_unreported_write_classes: set[type] = set() +# Three wrapper levels covers a metrics wrapper around a router around a backend; deeper nesting is rare. +_REACH_DEPTH = 3 +# Values read at most (attribute values, unset slots, container items and the type entries scanned to find them) before +# the search gives up and answers True. The answer is cached per handler, so a large object graph costs the +# deadline list, never a missed bound. +_REACH_READS = 4096 +_REACH_LEAVES = (str, bytes, bytearray, memoryview, int, float, complex, bool, type(None), type) +# Containers are walked with the base type's own iterator, so a subclass's __iter__ never runs. +_REACH_CONTAINERS: tuple[tuple[type, Callable[[Any], Iterator[object]]], ...] = ( + (dict, lambda d: iter(dict.values(d))), # a C view; iter() on it runs no user code + (list, list.__iter__), + (tuple, tuple.__iter__), + (set, set.__iter__), + (frozenset, frozenset.__iter__), +) +# Type internals read through type's own descriptors, so a metaclass property or __getattribute__ never runs. +_TYPE_MRO: Any = vars(type)["__mro__"] +_TYPE_DICT: Any = vars(type)["__dict__"] +_TYPE_MODULE: Any = vars(type)["__module__"] +_TYPE_QUALNAME: Any = vars(type)["__qualname__"] +_unreachable_wrapper_classes: set[tuple[str, str]] = set() # names, not classes: a class per mock is not kept +_MISSING = object() # not found yet: None is a value a namespace can hold +# A type's scan: (the type, kept alive so its id is not reused; member descriptors; getset __dict__; +# class-level flag; entries scanned). +_TypeScan = tuple[type, list[Any], object, object, int, list[object]] +# Class-namespace values that are code or descriptors, not held objects: skipped by type identity, so no +# metaclass __hash__ runs, and never called. +_CLASS_CODE = ( + types.FunctionType, + types.BuiltinFunctionType, + types.MemberDescriptorType, + types.GetSetDescriptorType, + types.WrapperDescriptorType, + types.MethodDescriptorType, + types.ClassMethodDescriptorType, + types.MethodWrapperType, + classmethod, + staticmethod, + property, +) + + +def _scan_type(kind: type) -> _TypeScan: + """What reading an instance of ``kind`` needs, from one scan of its MRO: its member descriptors, its getset + ``__dict__`` descriptor, its class-level ``_reports_write_deadline`` value, the entries scanned, and the + objects its classes hold (``inner = backend`` on a wrapper class), read from the namespaces as stored, so no + descriptor runs; functions and descriptors are skipped.""" + members: list[Any] = [] + class_values: list[object] = [] + scanned = 0 + instance_dict: object = _MISSING + flag: object = _MISSING + for klass in _TYPE_MRO.__get__(kind): + namespace = _TYPE_DICT.__get__(klass) + scanned += 1 + len(namespace) + members += [d for d in namespace.values() if type(d) is types.MemberDescriptorType] + # Every name, dunders included (``__wrapped__ = backend``): interpreter metadata is strings, ints, types + # and tuples of them, which the walk treats as leaves, or code and descriptors, skipped here. + class_values += [ + value + for value in namespace.values() + if not any(type(value) is code for code in _CLASS_CODE) and not issubclass(type(value), _REACH_LEAVES) + ] + if instance_dict is _MISSING and "__dict__" in namespace: + instance_dict = namespace["__dict__"] + if flag is _MISSING and "_reports_write_deadline" in namespace: + flag = namespace["_reports_write_deadline"] + return kind, members, instance_dict, flag, scanned, class_values + + +def _read_object( + obj: object, types_read: dict[int, _TypeScan], class_depths: dict[int, int], depth: int +) -> tuple[bool, list[object], int]: + """One read of ``obj`` at ``depth``: (it says it reports a write deadline, its attribute values, reads spent). + A type is scanned once per search (``types_read``), and charged then. Values are read only below + ``_REACH_DEPTH``. A type's class values are yielded whenever one of its instances is read shallower than any + earlier yield of them (``class_depths``), so a type first met at the depth limit, or first met by a longer + path, still has its class-held backend searched from the shallowest route: at most ``_REACH_DEPTH`` yields + per type. + + Storage is read only through the C descriptors Python makes for it: the getset ``__dict__`` and every + member descriptor along the MRO, so a property named ``__dict__`` or shadowing a slot is skipped, never + called. Member descriptors are deliberately every C member field, not only ``__slots__``: that is how a + bound method's ``__self__`` is found (``self.write = backend.set``), and a function's ``__globals__`` + only spends the budget, which fails open. Do not make functions leaves: that turns a True into a False. + """ + kind = type(obj) + spent = 0 + scan = types_read.get(id(kind)) # by id: hashing the type could run a metaclass __hash__ + if scan is None: + scan = types_read[id(kind)] = _scan_type(kind) + spent = scan[4] + _, members, instance_dict, flag, _, class_values = scan + want_values = depth < _REACH_DEPTH # at the limit only the flag is read + values: list[object] = [] + unset = 0 + for descriptor in members if want_values else (): + try: + values.append(descriptor.__get__(obj, kind)) + except AttributeError: # an unset slot: a read the caller never walks, so it is charged here + unset += 1 + if type(instance_dict) is types.GetSetDescriptorType: + held = instance_dict.__get__(obj, kind) # type: ignore[attr-defined] + if type(held) is dict: + if "_reports_write_deadline" in held: # an instance value shadows the class's + flag = held["_reports_write_deadline"] + if want_values: + values += dict.values(held) + if want_values and depth < class_depths.get(id(kind), _REACH_DEPTH): + class_depths[id(kind)] = depth + values += class_values + return flag is True, values, spent + unset # the caller charges each value it walks + + +def _reaches_write_deadlines(backend: object) -> bool: + """Whether a write through ``backend`` can report a write deadline (API-103): the backend says it reports + (read with ``getattr``, so an attribute-forwarding proxy forwards it), or it holds a backend that says so, + up to ``_REACH_DEPTH`` wrappers down, in an attribute (``__dict__`` or ``__slots__``) or in a dict, list, + tuple or set there (a metrics, timeout, router or shard-ring wrapper over a CachekitIOBackend). + + One breadth-first walk with one counter: every value read counts, container items and the type entries + scanned included. An object is expanded again only when reached by a shorter path (``seen`` keeps the + shallowest depth), so a cycle ends and a wrapper shared by a long and a short path is searched to its + full reach. Below the root everything is read through Python's own descriptors (``_read_object``), so a + lazy object is not evaluated and no property, ``__getattr__``, ``__iter__`` override or metaclass hook + runs. If the walk raises, or spends ``_REACH_READS`` reads without an answer, the answer is True: opening + the deadline list is always safe, skipping it is not. A wrapper holding its backend only in a closure, or + creating it on first use (even into an attribute: the search runs before the first write), is out of + reach: logged once per class, and its L1 copies keep the stated ttl. + """ + if backend is None: + return False + try: + if getattr(backend, "_reports_write_deadline", False) is True: + return True + queue: collections.deque[tuple[object, int]] = collections.deque([(backend, 0)]) + seen = {id(backend): 0} # id -> the shallowest depth it was reached at + types_read: dict[int, _TypeScan] = {} # each type's MRO is scanned once per search + class_depths: dict[int, int] = {} # type id -> the shallowest depth its class values were yielded at + reads = 0 + while queue: + obj, depth = queue.popleft() + kind = type(obj) + walk = next((it for base, it in _REACH_CONTAINERS if issubclass(kind, base)), None) + batches: list[tuple[Iterable[object], int]] = [] + if walk is not None: + batches.append((walk(obj), depth)) # a container: its items stay at the holder's depth + # An object, or a container subclass (a dict router with self.inner): read its own storage too. + # Identity, not `in`: hashing the type could run a metaclass __hash__. + if walk is None or not any(kind is base for base, _ in _REACH_CONTAINERS): + reports, values, scanned = _read_object(obj, types_read, class_depths, depth) + if reports and obj is not backend: # the root's own flag was read with getattr above + return True + reads += scanned + if reads > _REACH_READS: + return True + if depth < _REACH_DEPTH: # at the limit its flag is read; its own attributes are past the reach + batches.append((values, depth + 1)) + for children, child_depth in batches: + for child in children: + reads += 1 + if reads > _REACH_READS: + return True + if issubclass(type(child), _REACH_LEAVES) or seen.get(id(child), _REACH_DEPTH + 1) <= child_depth: + continue + seen[id(child)] = child_depth + queue.append((child, child_depth)) + kind = type(backend) + module = _TYPE_MODULE.__get__(kind) + name = (module if isinstance(module, str) else "", _TYPE_QUALNAME.__get__(kind)) + if name not in _unreachable_wrapper_classes and not name[0].startswith("cachekit."): + _unreachable_wrapper_classes.add(name) + get_logger().info( + f"{name[1]} holds no CachekitIOBackend in its attributes: its writes keep L1 copies for their " + "ttl. If it wraps one, hold it in an attribute before the first write." + ) + except Exception as exc: # noqa: BLE001 - the safe answer: open the list + get_logger().debug(f"write-deadline search failed: {type(exc).__name__}") + return True + return False + + +def write_deadlines_reachable(handler: Any) -> bool: + """``_reaches_write_deadlines`` for the handler's backend, cached on the handler. The decorator always + installs a ``StandardCacheHandler``; a custom ``CacheHandlerStrategy`` (a public extension point) that + cannot take the attribute (``__slots__``) is searched again on each write, bounded by ``_REACH_READS``.""" + reachable = _reaches_write_deadlines(getattr(handler, "backend", None)) + with contextlib.suppress(AttributeError): + handler._write_deadlines_reachable = reachable + return reachable + + +def _written_copy_expiry( + backend: object, seen: list[tuple[float, float]], stored: bool, ttl: Optional[float] +) -> Optional[float]: + """When the writer's L1 copy of a value it wrote expires (API-103), from what the write reported into + ``seen`` (``_WRITE_DEADLINES``); None keeps the stated ttl from the put. + + ``min(local_ttl, effective_ttl, fresh_for)``, every term timed from the 2xx's receipt: ttl stands for + local_ttl and effective_ttl (the L1 default for ttl=None), Fresh-For for the last, unbounded when + absent. A report wins even when the write then failed (an override that raised after its 2xx). With + none: None for a failed write (API-36/37 bound the copy) or a backend that never reports, and 0.0, no + copy, when the backend says it reports but the write never reached its own ``set`` (an override that + does not delegate, or delegates on another thread without the context), so the bound is unknown: that + turns L1 off for the backend's writes, logged once per class. A nested write that reports into the + same list can only shorten the bound. + """ + if seen: + local_ttl = DEFAULT_L1_TTL_SECONDS if ttl is None else ttl + return min(min(received + local_ttl, fresh_until) for received, fresh_until in seen) + if not stored or getattr(backend, "_reports_write_deadline", False) is not True: + return None + if type(backend) not in _unreported_write_classes: + _unreported_write_classes.add(type(backend)) + get_logger().warning( + f"{type(backend).__name__}.set reported no write deadline: its writes keep no L1 copy. " + "Delegate to the base set on the same thread, or under contextvars.copy_context().run." + ) + return 0.0 + + def _set_l2( handler: Any, key: str, @@ -407,24 +632,62 @@ def _set_l2( stale_ttl: int | None, registry_id: str | None, on_track_error: Callable[[str, Exception], None], -) -> tuple[bool, bool]: - """A buffered L2 write through ``handler``: (stored, tracked). + on_report: Optional[Callable[[tuple[float, float]], None]] = None, +) -> tuple[bool, bool, Optional[float]]: + """A buffered L2 write through ``handler``: (stored, tracked, l1_until). ``on_report`` is called with each + ``(received, fresh_until)`` the backend reports, as it reports it (``_ReportedDeadlines``). With a ``registry_id``, the key-registry write rides in the same round trip where the handler can (``set_tracked``), and tracked is True: its failure has gone to ``on_track_error``. Otherwise tracked is - False and the caller tracks a stored key itself. getattr: custom CacheHandlerStrategy impls predate - set_tracked, as they predate set_streaming. + False and the caller tracks a stored key itself. l1_until bounds the writer's L1 copy + (``_written_copy_expiry``), None where the backend reports no freshness. getattr: custom CacheHandlerStrategy + impls predate set_tracked, as they predate set_streaming. """ set_tracked = None if registry_id is None else getattr(handler, "set_tracked", None) if set_tracked is not None: stored = set_tracked(key, value, ttl, registry_id, on_track_error) if stored is not None: - return stored, True - # Only thread the SWR kwarg when set: strategy implementations without - # **metadata (tests, custom handlers) must keep working unchanged. - if stale_ttl is not None: - return bool(handler.set(key, value, ttl, stale_ttl=stale_ttl)), False - return bool(handler.set(key, value, ttl)), False + return stored, True, None + # The instruction-gated miss path: the cached answer is a plain attribute load (getattr's call costs the + # gate measurably; try is free until it raises), and a backend that cannot report takes its own early return + # (one set call under a conditional token measured +1.02% on `miss`, this +0.79%). + try: + reachable = handler._write_deadlines_reachable + except AttributeError: # a custom strategy that does not carry the cache + reachable = None + if not (write_deadlines_reachable(handler) if reachable is None else reachable): + # Only thread the SWR kwarg when set: strategy implementations without + # **metadata (tests, custom handlers) must keep working unchanged. + if stale_ttl is not None: + return bool(handler.set(key, value, ttl, stale_ttl=stale_ttl)), False, None + return bool(handler.set(key, value, ttl)), False, None + # Around the handler's own set, so an override of it still runs. + seen: list[tuple[float, float]] = [] if on_report is None else _ReportedDeadlines(on_report) + token = _WRITE_DEADLINES.set(seen) + try: + if stale_ttl is not None: + stored = bool(handler.set(key, value, ttl, stale_ttl=stale_ttl)) + else: + stored = bool(handler.set(key, value, ttl)) + finally: + _WRITE_DEADLINES.reset(token) + return stored, False, _written_copy_expiry(getattr(handler, "backend", None), seen, stored, ttl) + + +class _ReportedDeadlines(list[tuple[float, float]]): + """A ``_WRITE_DEADLINES`` list that applies each report the moment the backend appends it, from inside the + backend's ``set``: a write-behind copy is bounded at the 2xx's receipt, not when a ``set`` override's + post-response work (an audit hook) returns.""" + + __slots__ = ("_on_report",) + + def __init__(self, on_report: Callable[[tuple[float, float]], None]) -> None: + super().__init__() + self._on_report = on_report + + def append(self, entry: tuple[float, float]) -> None: + super().append(entry) + self._on_report(entry) class _MultiDeleteBackend(Protocol): @@ -1707,13 +1970,15 @@ class StoreOutcome(NamedTuple): a successful stream has none. Key tracking follows stored, L1 follows envelope. deferred says the backend write was handed to a write-behind writer (stored is then False: the writer tracks it). tracked says the key-registry write rode with the backend write (``set_tracked``), its failure - already reported: the caller must not track the key again. + already reported: the caller must not track the key again. l1_until is the instant the + caller's L1 copy must stop serving by (``_written_copy_expiry``; None keeps the stated ttl). """ envelope: Optional[bytes] stored: bool deferred: bool = False tracked: bool = False + l1_until: Optional[float] = None class CacheOperationHandler: @@ -2198,14 +2463,14 @@ def store_result( serialized_data = self.serialization_handler.serialize_data(result, args, kwargs, cache_key) if defer is not None and defer(serialized_data): return StoreOutcome(envelope=serialized_data, stored=False, deferred=True) - stored, tracked = _set_l2( + stored, tracked, l1_until = _set_l2( self._cache_handler, cache_key, serialized_data, ttl, stale_ttl, registry_id, on_track_error ) if stored: get_logger().cache_stored(cache_key, ttl) # The envelope goes to L1 even when the backend write failed (L1 serves it only within the per-entry share) - return StoreOutcome(envelope=serialized_data, stored=bool(stored), tracked=tracked) + return StoreOutcome(envelope=serialized_data, stored=bool(stored), tracked=tracked, l1_until=l1_until) except (InteropError, KeyringConfigurationError): # Interop/v1 data-model rejection (spec-mandated; matches cachekit-ts), or # a LOCAL keyring config fault, which a cold key first hits here, on the @@ -2408,6 +2673,7 @@ def __init__( ttl_refresh_threshold: Threshold for TTL refresh (0.0-1.0, default 0.5 = 50%) """ self.backend = backend + self._write_deadlines_reachable: Optional[bool] = None # see write_deadlines_reachable self.backpressure_controller = backpressure_controller self.ttl_refresh_threshold = ttl_refresh_threshold diff --git a/src/cachekit/config/decorator.py b/src/cachekit/config/decorator.py index a6d9e046..0e5d45d8 100644 --- a/src/cachekit/config/decorator.py +++ b/src/cachekit/config/decorator.py @@ -182,7 +182,7 @@ def local_function(): # positive ttl and an SWR-capable backend (CachekitIO). None = preset # decides (io() defaults it to ttl via swr_by_default); 0 = explicitly off. stale_ttl: int | None = None - # Preset flag: default stale_ttl to ttl (capped to the shared 30-day bound) + # Preset flag: default stale_ttl to ttl (capped to the shared 90-day TTL range) # when the backend supports SWR and the user didn't say otherwise. swr_by_default: bool = False namespace: str | None = None @@ -364,7 +364,8 @@ def minimal(cls, **kwargs: Any) -> DecoratorConfig: Args: **kwargs: Overrides; each wins over the preset's value (ttl, namespace, backend, integrity_checking=True to opt-in, l1, circuit_breaker, backpressure, monitoring, etc.) - Default ttl=300 (protocol/spec/intent-presets.md); ttl=None = never expire. + Default ttl=300 (protocol/spec/intent-presets.md); ttl=None sends no TTL: + never expiring on Redis, Memcached and File; CachekitIO stores it for the plan maximum. Returns: DecoratorConfig with minimal protections preset @@ -411,7 +412,8 @@ def production(cls, **kwargs: Any) -> DecoratorConfig: Args: **kwargs: Overrides; each wins over the preset's value (ttl, namespace, backend, integrity_checking, l1, circuit_breaker, backpressure, monitoring, etc.) - Default ttl=600 (protocol/spec/intent-presets.md); ttl=None = never expire. + Default ttl=600 (protocol/spec/intent-presets.md); ttl=None sends no TTL: + never expiring on Redis, Memcached and File; CachekitIO stores it for the plan maximum. Returns: DecoratorConfig with production-grade protections @@ -469,7 +471,8 @@ def secure( per-tenant key derivation. Not a tenancy boundary: see docs/features/zero-knowledge-encryption.md **kwargs: Overrides (ttl, namespace, backend, l1, circuit_breaker, backpressure, monitoring, etc.) - integrity_checking other than True is rejected, and encryption= is not an override. - Default ttl=600 (protocol/spec/intent-presets.md); ttl=None = never expire. + Default ttl=600 (protocol/spec/intent-presets.md); ttl=None sends no TTL: + never expiring on Redis, Memcached and File; CachekitIO stores it for the plan maximum. fail_closed=True raises DecryptionAuthenticationError to the caller on AES-GCM auth failure / key-fingerprint mismatch instead of silently recomputing (default None defers to CACHEKIT_ENCRYPTION_FAIL_CLOSED, which defaults to False) @@ -562,7 +565,8 @@ def dev(cls, **kwargs: Any) -> DecoratorConfig: Args: **kwargs: Overrides; each wins over the preset's value (ttl, namespace, backend, integrity_checking, l1, circuit_breaker, backpressure, monitoring, etc.) - Default ttl=300 (SDK-local preset; spec rule 4 forbids never-expire as a default); ttl=None = never expire. + Default ttl=300 (SDK-local preset; spec rule 4 forbids never-expire as a default); ttl=None sends no TTL: + never expiring on Redis, Memcached and File; CachekitIO stores it for the plan maximum. Returns: DecoratorConfig optimized for development @@ -607,7 +611,8 @@ def test(cls, **kwargs: Any) -> DecoratorConfig: Args: **kwargs: Overrides; each wins over the preset's value (ttl, namespace, backend, integrity_checking, l1, circuit_breaker, backpressure, monitoring, etc.) - Default ttl=300 (SDK-local preset; spec rule 4 forbids never-expire as a default); ttl=None = never expire. + Default ttl=300 (SDK-local preset; spec rule 4 forbids never-expire as a default); ttl=None sends no TTL: + never expiring on Redis, Memcached and File; CachekitIO stores it for the plan maximum. Returns: DecoratorConfig optimized for testing @@ -662,8 +667,9 @@ def io(cls, api_key: str | SecretStr | None = None, **kwargs: Any) -> DecoratorC **kwargs: Overrides (ttl, namespace, integrity_checking, swr_by_default, l1, circuit_breaker, backpressure, monitoring, etc.). ``backend`` is not one — io always caches through its own CachekitIOBackend and rejects ``backend=``. - Default ttl=3600 (protocol/spec/intent-presets.md); ttl=None = never expire - (and disables the stale_ttl SWR window, which needs a positive ttl). + Default ttl=3600 (protocol/spec/intent-presets.md); ttl=None sends no TTL, so the + server stores the entry for the plan's maximum TTL (never expiring only on a plan + without a maximum, or where none applied when the entry was written), and disables the stale_ttl SWR window, which needs a positive ttl. Returns: DecoratorConfig with CachekitIOBackend diff --git a/src/cachekit/decorators/intent.py b/src/cachekit/decorators/intent.py index 5b324d97..6107df52 100644 --- a/src/cachekit/decorators/intent.py +++ b/src/cachekit/decorators/intent.py @@ -107,7 +107,9 @@ def cache( entry is served immediately and the function re-runs in the background, so no request pays the recompute at a TTL boundary. Requires a positive ``ttl`` and an SWR-capable backend (CachekitIO); ``ttl + stale_ttl`` - is capped at 2,592,000 s (30 days). ``@cache.io`` defaults it to + is capped at 7,776,000 s (90 days, the protocol's TTL range; a plan + with a lower maximum TTL shortens the stale window first, server-side). + ``@cache.io`` defaults it to ``ttl`` — pass ``stale_ttl=0`` to opt out. The background recompute sees a snapshot of the caller's ``contextvars`` (so contextvar-based tenant extraction works), but no other request-scoped resources — diff --git a/src/cachekit/decorators/wrapper.py b/src/cachekit/decorators/wrapper.py index a65d51dd..aecb0fdc 100644 --- a/src/cachekit/decorators/wrapper.py +++ b/src/cachekit/decorators/wrapper.py @@ -20,6 +20,7 @@ from .. import invalidation, l1_cache from .._rust_serializer import HitCounter, HitMemo from .._rust_serializer import memo_key as _memo_key +from ..backends.base import _WRITE_DEADLINES from ..backends.errors import BackendError, UnsupportedTenantError from ..cache_handler import ( CacheHandlerStrategy, @@ -31,6 +32,7 @@ TenantResolutionError, _set_l2, _supports_multi_delete, + _written_copy_expiry, get_backend_provider, get_logger, handle_decrypt_failure, @@ -42,6 +44,7 @@ supports_swr, supports_ttl_inspection, warn_ttl_refresh_unsupported, + write_deadlines_reachable, ) from ..config.validation import ConfigurationError, hide_secret from ..interop import ( @@ -221,6 +224,22 @@ def _ttl_refresh_done_callback(task: asyncio.Task, cache_key: str) -> None: pass +class _DeferredCopy: + """Who drops a write-behind miss's L1 copy when it would outlive its write's bound (API-103). + + The writer can land before or after the miss's L1 put, and both run under ``lock``. A put that comes + second puts the copy already bounded by ``l1_until``; a write that comes second compares ``l1_until`` + with the ``put_until`` the put recorded, and drops a copy that would outlive it. + """ + + __slots__ = ("l1_until", "lock", "put_until") + + def __init__(self) -> None: + self.lock = threading.Lock() + self.put_until: float | None = None # the expires_at the miss's L1 copy was put with + self.l1_until: float | None = None # the write's bound, once it landed with one + + class _LockPhase: """How far a lock clause got, so its handler can tell where a lock error was raised. @@ -963,7 +982,7 @@ def _install_cache_handler(resolved: BaseBackend) -> None: # its fresh TTL and labels the read stale; we return the stale value immediately # and re-run the wrapped function in the background. Requires an SWR-capable # backend (CachekitIO — the server signals freshness on read). - _max_total_ttl = 2_592_000 # 30-day storage cap, shared with the stale window (spec) + _max_total_ttl = 7_776_000 # 90-day TTL range, shared with the stale window (spec); the plan maximum may clamp it def _l2_freshness_capable() -> bool: """Freshness capability of the backend as RESOLVED so far. The read paths @@ -988,7 +1007,7 @@ def _l2_freshness_capable() -> bool: if ttl is None or ttl <= 0: raise ConfigurationError("stale_ttl requires a positive ttl (the stale window starts where freshness ends)") if ttl + stale_ttl > _max_total_ttl: - raise ConfigurationError(f"ttl + stale_ttl must not exceed {_max_total_ttl} seconds (30-day storage cap)") + raise ConfigurationError(f"ttl + stale_ttl must not exceed {_max_total_ttl} seconds (90-day TTL range)") if not _l2_swr_capable_at_decoration: raise ConfigurationError( "stale_ttl requires an SWR-capable backend (CachekitIO) known at decoration time. " @@ -1004,7 +1023,7 @@ def _l2_freshness_capable() -> bool: and _l2_swr_capable_at_decoration ): # Preset default (io()): stale window = ttl, capped so the total stays - # within the 30-day bound. stale_ttl=0 opts out explicitly. A ttl at or + # within the 90-day range. stale_ttl=0 opts out explicitly. A ttl at or # above the cap leaves no window headroom -> no default (never negative). # Derived from the TTL as the wire sends it (whole seconds, ceiled), so a # fractional ttl near the cap cannot round the sum past it into a 400. @@ -1195,7 +1214,9 @@ def _done(t: asyncio.Task[None]) -> None: task.add_done_callback(_done) - def _put_l1(cache_key: str, serialized_data: Any, l1_ttl: int | None, *, twin: str | None) -> None: + def _put_l1( + cache_key: str, serialized_data: Any, l1_ttl: int | None, *, twin: str | None, expires_at: float | None = None + ) -> None: """The only L1 write in this wrapper: put the serialized bytes (str payloads encoded), then record the key, and its pre-0.20.0 twin if it has one (_twin_key), in _cached_keys. @@ -1209,9 +1230,22 @@ def _put_l1(cache_key: str, serialized_data: Any, l1_ttl: int | None, *, twin: s """ if _l1_cache and serialized_data: _b = serialized_data.encode("utf-8") if isinstance(serialized_data, str) else serialized_data - _l1_cache.put(cache_key, _b, redis_ttl=l1_ttl) + _l1_cache.put(cache_key, _b, redis_ttl=l1_ttl, expires_at=expires_at) _record(cache_key, twin) + def _put_l1_written(cache_key: str, serialized_data: Any, l1_until: float | None, *, twin: str | None) -> None: + """The writer's own L1 copy of a value it stored (API-103): served until l1_until, the instant its + write's 2xx bounds it to (``cache_handler._written_copy_expiry``). + + l1_until is timed from the 2xx's receipt, not from this put, so a stall between the two (the hop + back to the event loop) shortens the copy instead of restarting its bound. None (the backend + reports no freshness, or the write failed) keeps the stated ttl from the put. + """ + if l1_until is None: + _put_l1(cache_key, serialized_data, ttl, twin=twin) + else: + _put_l1(cache_key, serialized_data, None, twin=twin, expires_at=l1_until) + def _record(cache_key: str, twin: str | None) -> None: """Record ``cache_key`` and its ``twin`` (if any) in _cached_keys under the current L2 scope, in one step. @@ -1271,18 +1305,32 @@ async def _track_and_record_async(cache_key: str) -> None: if _is_trackable(): await asyncio.to_thread(_track_and_record, cache_key) - async def _set_l2_async(cache_key: str, serialized_data: bytes) -> tuple[bool, bool]: - """An async miss's L2 write: (stored, tracked). Where the handler can (set_tracked_async), the + async def _set_l2_async(cache_key: str, serialized_data: bytes) -> tuple[bool, bool, float | None]: + """An async L2 write: (stored, tracked, l1_until). Where the handler can (set_tracked_async), the key-registry write rides in the same round trip and thread hop, and tracked is True; otherwise the - caller tracks a stored key itself (_track_and_record_async).""" + caller tracks a stored key itself (_track_and_record_async). l1_until bounds the writer's L1 copy + (cache_handler._written_copy_expiry), None where the backend reports no freshness.""" handler = operation_handler.cache_handler # Asked first: a backend without a registry never pays for the probe. Custom strategies predate set_tracked_async. set_tracked_async = getattr(handler, "set_tracked_async", None) if _is_trackable() else None if set_tracked_async is not None: stored = await set_tracked_async(cache_key, serialized_data, ttl, _registry_id, _warn_track_failure) if stored is not None: - return stored, True - return await handler.set_async(cache_key, serialized_data, ttl=ttl, stale_ttl=_stale_ttl), False # type: ignore[union-attr] + return stored, True, None + reachable = getattr(handler, "_write_deadlines_reachable", None) # the async path is not gated: one branch + seen: list[tuple[float, float]] | None = ( + [] if (write_deadlines_reachable(handler) if reachable is None else reachable) else None + ) + # Around the handler's own set_async. to_thread copies the context. + token = None if seen is None else _WRITE_DEADLINES.set(seen) + try: + stored = await handler.set_async(cache_key, serialized_data, ttl=ttl, stale_ttl=_stale_ttl) # type: ignore[union-attr] + finally: + if token is not None: + _WRITE_DEADLINES.reset(token) + if seen is None: + return stored, False, None + return stored, False, _written_copy_expiry(getattr(handler, "backend", None), seen, stored, ttl) def _l1_backfill_ttl(fresh_for: int | None) -> Any: """L1 TTL for a backfill from an L2 read, bounded by the server's remaining @@ -1297,7 +1345,7 @@ def _l1_backfill_ttl(fresh_for: int | None) -> Any: The signal may only ever SHORTEN the L1 lifetime: with ttl=None the legacy baseline is L1Cache's own default (DEFAULT_L1_TTL_SECONDS), so a long server remainder is clamped to it — returning raw fresh_for would - EXTEND local service up to the 30-day cap and turn the bound into the + EXTEND local service up to Fresh-For's 30-day cap and turn the bound into the very freshness-extension it exists to prevent (CWE-613: server-side DELETE-as-revocation relies on the ≤300s ageout). """ @@ -1445,12 +1493,10 @@ async def _l2_swr_recompute_store_async( serialized_data = operation_handler.serialization_handler.serialize_data( result, call_args, call_kwargs, cache_key=cache_key ) - stored = await operation_handler.cache_handler.set_async( # type: ignore[attr-defined] - cache_key, serialized_data, ttl=ttl, stale_ttl=_stale_ttl - ) - # Refresh L1 with the new fresh bytes (mirrors the miss-path store). - _put_l1(cache_key, serialized_data, ttl, twin=_twin_key(call_args, call_kwargs)) - if stored: + stored, tracked, l1_until = await _set_l2_async(cache_key, serialized_data) + # Refresh L1 with the new fresh bytes (mirrors the miss-path store), bounded by the write (API-103). + _put_l1_written(cache_key, serialized_data, l1_until, twin=_twin_key(call_args, call_kwargs)) + if stored and not tracked: await _track_and_record_async(cache_key) async def _l2_swr_revalidate_async( @@ -1486,10 +1532,10 @@ def _l2_swr_revalidate_sync(slot: object, cache_key: str, call_args: tuple[Any, serialized_data = operation_handler.serialization_handler.serialize_data( result, call_args, call_kwargs, cache_key=cache_key ) - stored = operation_handler.cache_handler.set( # type: ignore[attr-defined] - cache_key, serialized_data, ttl=ttl, stale_ttl=_stale_ttl - ) - _put_l1(cache_key, serialized_data, ttl, twin=_twin_key(call_args, call_kwargs)) + # No registry_id: the key is tracked below, as before. + handler = operation_handler.cache_handler + stored, _, l1_until = _set_l2(handler, cache_key, serialized_data, ttl, _stale_ttl, None, _warn_track_failure) + _put_l1_written(cache_key, serialized_data, l1_until, twin=_twin_key(call_args, call_kwargs)) if stored: _track_and_record(cache_key) except Exception as exc: # noqa: BLE001 — spec: revalidation failure must never surface to callers @@ -1761,7 +1807,9 @@ def _warn_deferred_write(cache_key: str, reason: str) -> None: else None ) - def _send_deferred(cache_key: str, envelope: bytes, twin: str | None, handler: CacheHandlerStrategy) -> None: + def _send_deferred( + cache_key: str, envelope: bytes, twin: str | None, handler: CacheHandlerStrategy, copy: _DeferredCopy + ) -> None: """A write-behind miss write, on the writer's thread in the caller's context: the backend write store_result would have made, then what the miss path does after a stored one. @@ -1773,8 +1821,25 @@ def _send_deferred(cache_key: str, envelope: bytes, twin: str | None, handler: C start = time.perf_counter() # Trackable first: a backend without a registry never pays for set_tracked's probe. registry_id = _registry_id if _is_trackable() else None - stored, tracked = _set_l2(handler, cache_key, envelope, ttl, _stale_ttl, registry_id, _warn_track_failure) + local_ttl = DEFAULT_L1_TTL_SECONDS if ttl is None else ttl + + def bound_at_receipt(report: tuple[float, float]) -> None: + # Called from inside the backend's set as the 2xx is read: the copy is bounded before any post-response + # work in a set override runs, not when the handler call returns. + received, fresh_until = report + _drop_copy_outliving_write(cache_key, min(received + local_ttl, fresh_until), copy) + + try: + stored, tracked, l1_until = _set_l2( + handler, cache_key, envelope, ttl, _stale_ttl, registry_id, _warn_track_failure, on_report=bound_at_receipt + ) + except BaseException: + # The write may have had its 2xx before it raised, and its bound is lost with the return: no copy. + _drop_copy_outliving_write(cache_key, 0.0, copy) + raise duration_ms = (time.perf_counter() - start) * 1000 + # Stored or not: a write that failed after its 2xx still bounds the copy (_written_copy_expiry). + _drop_copy_outliving_write(cache_key, l1_until, copy) if stored: get_logger().cache_stored(cache_key, ttl) _record(cache_key, twin) @@ -1791,11 +1856,41 @@ def _send_deferred(cache_key: str, envelope: bytes, twin: str | None, handler: C serializer="rust", ) - def _defer_write(cache_key: str, twin: str | None, envelope: bytes) -> bool: + def _drop_copy_outliving_write(cache_key: str, l1_until: float | None, copy: _DeferredCopy) -> None: + """Hold a write-behind copy to the bound of its write's 2xx (API-103), once that write lands. + + A copy the miss already put, for ttl before the bound was known, that would serve past l1_until (the + server clamped the TTL to the plan maximum, sent 0 or a malformed value, or the put came well after + the 2xx) is dropped, not re-put: a re-put could replace a newer value, and the next read backfills + under its own bound. A put still to come reads l1_until and is bounded by it (_put_deferred_copy). + """ + if l1_until is None or _l1_cache is None: + return + with copy.lock: + # The smallest bound reported so far: a later report (or the final one) can only shorten it. + l1_until = l1_until if copy.l1_until is None else min(copy.l1_until, l1_until) + copy.l1_until = l1_until + put_until = copy.put_until + # L1 stops serving a copy ttl_buffer_seconds before the expiry it was put with. + if put_until is not None and put_until - _l1_cache.ttl_buffer_seconds > l1_until: + _l1_cache.invalidate(cache_key) + + def _put_deferred_copy(cache_key: str, serialized_data: Any, twin: str | None, copy: _DeferredCopy) -> None: + """A deferred miss's L1 put. Under the latch, so it never publishes a copy past a bound its write has + already landed with: bounded by it when the write came first, else for ttl, and then dropped by + _drop_copy_outliving_write if the bound turns out shorter.""" + with copy.lock: + # The exact expiry the copy is put with, recorded in both branches, so any later and shorter report + # (a second PUT in the same write) still finds the copy and drops it. + local_until = time.time() + (DEFAULT_L1_TTL_SECONDS if ttl is None else ttl) + copy.put_until = local_until if copy.l1_until is None else min(local_until, copy.l1_until) + _put_l1(cache_key, serialized_data, None, twin=twin, expires_at=copy.put_until) + + def _defer_write(cache_key: str, twin: str | None, envelope: bytes, *, copy: _DeferredCopy) -> bool: """Hand a miss's serialized value to the writer. False if it refused: the caller writes it now.""" handler = operation_handler.cache_handler assert _write_behind is not None and handler is not None # noqa: S101 - only on the miss path, write_behind on - send = functools.partial(_send_deferred, cache_key, envelope, twin, handler) + send = functools.partial(_send_deferred, cache_key, envelope, twin, handler, copy) return _write_behind.submit((_l2_scope(), cache_key), len(envelope), send) def _cancel_deferred(match: Callable[[Entry], bool], *, whole_function: bool, wait: bool = True) -> None: @@ -2513,6 +2608,7 @@ def _backed_miss_sync( # landing between the two leaves the same state either way. # Asked once, before the write: a backend without a registry never pays for set_tracked's probe. trackable = _is_trackable() + deferred_copy = None if _write_behind is None else _DeferredCopy() outcome = operation_handler.store_result( cache_key, result, @@ -2520,15 +2616,26 @@ def _backed_miss_sync( args, kwargs, stale_ttl=_stale_ttl, - defer=None if _write_behind is None else functools.partial(_defer_write, cache_key, twin_key), + defer=None + if deferred_copy is None + else functools.partial(_defer_write, cache_key, twin_key, copy=deferred_copy), registry_id=_registry_id if trackable else None, on_track_error=_warn_track_failure, ) envelope = outcome.envelope set_duration_ms = (time.perf_counter() - set_start) * 1000 - # Also store in L1 cache for fast subsequent access (using serialized bytes) - _put_l1(cache_key, outcome.envelope, ttl, twin=twin_key) + # Also store in L1 cache for fast subsequent access (using serialized bytes), bounded by the + # write's Fresh-For (API-103). A deferred write has none yet: _send_deferred bounds that copy. + if outcome.deferred and deferred_copy is not None: + _put_deferred_copy(cache_key, outcome.envelope, twin_key, deferred_copy) + else: + # None on every backend but CachekitIO: the stated ttl, without the helper's frame (gated miss path). + l1_until = outcome.l1_until + if l1_until is None: + _put_l1(cache_key, outcome.envelope, ttl, twin=twin_key) + else: + _put_l1_written(cache_key, outcome.envelope, l1_until, twin=twin_key) if outcome.stored and trackable and not outcome.tracked: _track_and_record(cache_key) @@ -3055,11 +3162,12 @@ async def _backed_miss_async( ) # Store in Redis with TTL - stored, tracked = await _set_l2_async(cache_key, serialized_data) + stored, tracked, l1_until = await _set_l2_async(cache_key, serialized_data) set_duration_ms = (time.perf_counter() - set_start) * 1000 - # Also store in L1 cache for fast subsequent access (using serialized bytes) - _put_l1(cache_key, serialized_data, ttl, twin=twin_key) + # Also store in L1 cache for fast subsequent access (using serialized bytes), + # bounded by the write's Fresh-For (API-103) + _put_l1_written(cache_key, serialized_data, l1_until, twin=twin_key) if stored and not tracked: await _track_and_record_async(cache_key) @@ -3173,14 +3281,20 @@ async def _backed_miss_async( # Write-behind (unlocked path only: the locked path's write must land inside its lock, or # waiters re-read before L2 holds the value and recompute). A refusal writes here, as without it. - deferred = _write_behind is not None and _defer_write(cache_key, twin_key, serialized_data) + deferred_copy = None if _write_behind is None else _DeferredCopy() + deferred = deferred_copy is not None and _defer_write(cache_key, twin_key, serialized_data, copy=deferred_copy) stored = tracked = False # a deferred write is not stored yet: the writer tracks it once it lands + l1_until = None if not deferred: - stored, tracked = await _set_l2_async(cache_key, serialized_data) + stored, tracked, l1_until = await _set_l2_async(cache_key, serialized_data) set_duration_ms = (time.perf_counter() - set_start) * 1000 - # Also store in L1 cache for fast subsequent access (using serialized bytes) - _put_l1(cache_key, serialized_data, ttl, twin=twin_key) + # Also store in L1 cache for fast subsequent access (using serialized bytes), bounded by the + # write's Fresh-For (API-103). A deferred write has none yet: _send_deferred bounds that copy. + if deferred and deferred_copy is not None: + _put_deferred_copy(cache_key, serialized_data, twin_key, deferred_copy) + else: + _put_l1_written(cache_key, serialized_data, l1_until, twin=twin_key) if stored and not tracked: await _track_and_record_async(cache_key) diff --git a/src/cachekit/l1_cache.py b/src/cachekit/l1_cache.py index 885455cc..1894702b 100644 --- a/src/cachekit/l1_cache.py +++ b/src/cachekit/l1_cache.py @@ -304,7 +304,9 @@ def put( """Store value in L1 cache with TTL. A value larger than an eighth of max_memory_bytes is not stored, and any older entry under - the key is dropped; callers serve it from L2, or recompute it if L2 did not store it. + the key is dropped; callers serve it from L2, or recompute it if L2 did not store it. The + same holds, for every caller, for a value whose expiry is non-finite or already past once + ttl_buffer_seconds is taken off: a declined put never leaves the previous value serving. Args: key: Cache key @@ -328,6 +330,11 @@ def put( "the mapped file for the entry's TTL. Materialize to bytes before caching." ) + # Before the first _state use, the declined put's removal below included: the take-over + # replaces a state whose lock fork orphaned. + if self._before_store is not None: + self._before_store() + # Calculate expiry time current_time = time.time() if expires_at is not None: @@ -339,7 +346,10 @@ def put( # Skip caching if the effective TTL is non-finite (NaN/inf would create an # immortal entry that never expires) or too short (would expire immediately). + # Any older entry under the key is dropped, as for an oversized value below: a put that + # declines to store a newer value must not leave the previous one serving. if not math.isfinite(expiry) or expiry <= current_time: + self._on_current_state(_L1State.remove, key) logger.debug( "Skipping L1 cache for key %s - non-finite or too-short TTL (effective expiry: %r)", redact_key_for_log(key), @@ -347,10 +357,6 @@ def put( ) return - # Before the first _state use: the take-over replaces a state whose lock fork orphaned. - if self._before_store is not None: - self._before_store() - # Estimate size size = self._estimate_size(value) diff --git a/tests/unit/test_cachekitio_write_bound.py b/tests/unit/test_cachekitio_write_bound.py new file mode 100644 index 00000000..ac25ff3f --- /dev/null +++ b/tests/unit/test_cachekitio_write_bound.py @@ -0,0 +1,1409 @@ +"""The writer's own L1 copy over CachekitIO (protocol spec/saas-api.md#remaining-freshness). + +API-103: once the store accepts a PUT, the decorator must not serve its L1 copy of the value after +min(local_ttl, effective_ttl, fresh_for), where fresh_for is the PUT 2xx's X-CacheKit-Fresh-For read with +the GET grammar: absent = no bound (today's behaviour), malformed or 0 = no L1 copy at all. The server +sends a Fresh-For below the stated TTL when it clamped the TTL to the plan's maximum. + +API-104: a PATCH /ttl response never creates, refreshes or extends an L1 bound. + +Everything runs through the backend's real client on a fake pool (tests/utils/cachekitio_fakes.py), so the +header is parsed exactly as in production. Wall-clock time is a fake clock: the ``time`` module of L1, the +decorator and the backend is swapped for a shim whose ``time()`` the test advances, so the PUT's receipt, the +L1 put and L1 expiry all read one clock. +""" + +from __future__ import annotations + +import asyncio +import dataclasses +import math +import threading +import time +import types +from collections.abc import Callable, Iterator +from dataclasses import dataclass, field +from typing import Any, NamedTuple +from urllib.parse import unquote + +import pytest +from urllib3 import HTTPResponse + +from cachekit import cache +from cachekit import cache_handler as cache_handler_module +from cachekit import l1_cache as l1_module +from cachekit.backends.base import _WRITE_DEADLINES +from cachekit.backends.cachekitio import backend as backend_module +from cachekit.backends.cachekitio.backend import FRESH_FOR_HEADER, FRESHNESS_HEADER, CachekitIOBackend +from cachekit.cache_handler import _REACH_READS, StandardCacheHandler, _reaches_write_deadlines, _set_l2 +from cachekit.decorators import wrapper as wrapper_module +from cachekit.decorators import write_behind as wb +from cachekit.l1_cache import DEFAULT_L1_TTL_SECONDS +from tests.utils.cachekitio_fakes import FakeRequest, fake_backend, response + +pytestmark = pytest.mark.unit + +_PREFIX = "/v1/cache/" +_GUARD_SECONDS = 10.0 # hang guard only: hitting it fails the test, it never decides an outcome + + +@dataclass +class _FakeSaaS: + """Stores PUT bodies; answers each PUT 2xx with ``put_fresh_for`` (None omits the header).""" + + put_fresh_for: str | None = None + fail_writes: bool = False + get_fresh_for: str | None = None + stale: bool = False # label every hit stale, as in the entry's stale window + store: dict[str, bytes] = field(default_factory=dict) + ops: list[str] = field(default_factory=list) + patched: threading.Event = field(default_factory=threading.Event) + + def respond(self, request: FakeRequest) -> HTTPResponse: + encoded, _, suffix = request.path[len(_PREFIX) :].partition("/") + key = unquote(encoded) + op = f"{request.method} /{suffix}" if suffix else request.method + self.ops.append(op) + if op == "PUT" and self.fail_writes: + return response(500) + if op == "PUT": + assert request.body is not None + self.store[key] = request.body + return response(200, headers=None if self.put_fresh_for is None else {FRESH_FOR_HEADER: self.put_fresh_for}) + if op == "GET": + if key not in self.store: + return response(404) + if self.stale: + return response(200, self.store[key], headers={FRESHNESS_HEADER: "stale", FRESH_FOR_HEADER: "0"}) + headers = None if self.get_fresh_for is None else {FRESH_FOR_HEADER: self.get_fresh_for} + return response(200, self.store[key], headers=headers) + if op == "GET /ttl": + return response(200, json={"ttl": 1}) # under any refresh threshold: a refresh is due + if op == "POST /lock": + return response(200, json={"lock_id": "lock-1"}) + if op == "PATCH /ttl": + self.patched.set() + return response(200, json={"success": True}) + return response(200) + + def gets(self) -> int: + return self.ops.count("GET") + + +class _Clock: + """The wall clock: ``now`` is what L1, the decorator and the backend read as ``time.time()``.""" + + def __init__(self) -> None: + self.now = 1_000_000.0 + + def advance(self, seconds: float) -> None: + self.now += seconds + + +@pytest.fixture +def clock(monkeypatch: pytest.MonkeyPatch) -> _Clock: + clock = _Clock() + shim = types.SimpleNamespace(**{name: getattr(time, name) for name in dir(time) if not name.startswith("_")}) + shim.time = lambda: clock.now + for module in (l1_module, wrapper_module, backend_module): + monkeypatch.setattr(module, "time", shim) + return clock + + +@pytest.fixture +def saas() -> _FakeSaaS: + return _FakeSaaS() + + +@pytest.fixture +def backend(saas: _FakeSaaS) -> CachekitIOBackend: + backend, _ = fake_backend(saas.respond) + return backend + + +_namespaces = iter(range(1_000_000)) + + +def _namespace() -> str: + """A namespace no other test shares: L1 caches are per-namespace and process-wide.""" + return f"write-bound-{next(_namespaces)}-{time.monotonic_ns()}" + + +def _reported(backend: CachekitIOBackend) -> list[tuple[float, float]]: + """What one ``backend.set`` reports through _WRITE_DEADLINES.""" + seen: list[tuple[float, float]] = [] + token = _WRITE_DEADLINES.set(seen) + try: + backend.set("k", b"v", ttl=60) + finally: + _WRITE_DEADLINES.reset(token) + return seen + + +def _write(handler: Any, ttl: int | None = 60) -> tuple[bool, bool, float | None]: + """One decorator L2 write through ``handler``: (stored, tracked, l1_until).""" + return _set_l2(handler, "k", b"v", ttl, None, None, lambda *_: None) + + +class TestBackendReportsPutFreshFor: + """``set`` reports the PUT 2xx's receipt and its header, read with the GET grammar (API-32/33).""" + + @pytest.mark.parametrize( + ("header", "expected"), + [ + (None, None), # absent: the stated TTL was applied + ("5", 5), + ("0", 0), + ("2592000", 2592000), # the header's own 30-day cap, below the 90-day TTL range + ("2592001", 0), # over the cap: malformed + ("7776000", 0), # the TTL range maximum is not a valid Fresh-For + ("+5", 0), + ("", 0), + ("00000005", 0), # eight digits: the length check runs first + ], + ) + def test_header_grammar( + self, saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock, header: str | None, expected: int | None + ) -> None: + saas.put_fresh_for = header + assert _reported(backend) == [(clock.now, math.inf if expected is None else clock.now + expected)] + assert saas.ops == ["PUT"] + + def test_timed_from_receipt( + self, saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock, monkeypatch: pytest.MonkeyPatch + ) -> None: + """The deadline is fixed when the 2xx arrives: time spent after it, parsing included, does not move it.""" + parse = CachekitIOBackend._fresh_for + + def slow_parse(response: Any) -> int | None: + clock.advance(3) + return parse(response) + + monkeypatch.setattr(CachekitIOBackend, "_fresh_for", staticmethod(slow_parse)) + saas.put_fresh_for = "5" + received = clock.now + assert _reported(backend) == [(received, received + 5)] + assert clock.now == received + 3 # the slow parse ran: time did pass after receipt + + def test_set_keeps_its_none_return(self, saas: _FakeSaaS, backend: CachekitIOBackend) -> None: + saas.put_fresh_for = "5" + assert backend.set("k", b"v", ttl=60) is None # type: ignore[func-returns-value] + + +class TestWriteBound: + """``_set_l2`` turns what the backend reported into the instant the writer's L1 copy stops serving.""" + + def test_header_below_ttl(self, saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock) -> None: + saas.put_fresh_for = "5" + assert _write(StandardCacheHandler(backend)) == (True, False, clock.now + 5) + assert _WRITE_DEADLINES.get() is None # closed what it opened + + def test_absent_header_bounds_the_stated_ttl_from_receipt( + self, saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock + ) -> None: + assert _write(StandardCacheHandler(backend)) == (True, False, clock.now + 60) + + def test_no_ttl_uses_the_l1_default(self, saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock) -> None: + assert _write(StandardCacheHandler(backend), ttl=None) == (True, False, clock.now + DEFAULT_L1_TTL_SECONDS) + + async def test_async_write_shares_the_list(self, saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock) -> None: + """``set_async`` runs the write on a worker thread, in a copy of the context: the list still reaches it.""" + saas.put_fresh_for = "7" + seen: list[tuple[float, float]] = [] + token = _WRITE_DEADLINES.set(seen) + try: + assert await StandardCacheHandler(backend).set_async("k", b"v", ttl=60) is True + finally: + _WRITE_DEADLINES.reset(token) + assert seen == [(clock.now, clock.now + 7)] + + def test_failed_write_reports_no_bound(self, saas: _FakeSaaS, backend: CachekitIOBackend) -> None: + saas.fail_writes = True + assert _write(StandardCacheHandler(backend)) == (False, False, None) + + def test_other_backends_report_none(self) -> None: + written: list[tuple[Any, ...]] = [] + + class _Plain: + def set(self, key: str, value: bytes, ttl: int | None = None) -> None: + written.append((key, value, ttl)) + + assert _write(StandardCacheHandler(_Plain())) == (True, False, None) # type: ignore[arg-type] + assert written == [("k", b"v", 60)] + + def test_set_override_still_runs(self, saas: _FakeSaaS, clock: _Clock) -> None: + """A subclass of ``set`` keeps running on every write, and one that delegates still reports its + write's bound: the base ``set`` reports it, not a private variant the override would bypass.""" + seen: list[str] = [] + + class _Recording(CachekitIOBackend): + def set(self, key: str, value: bytes, ttl: int | None = None, stale_ttl: int | None = None) -> None: + seen.append(key) + super().set(key, value, ttl, stale_ttl) + + backend, _ = fake_backend(saas.respond) + backend.__class__ = _Recording + saas.put_fresh_for = "5" + assert _write(StandardCacheHandler(backend)) == (True, False, clock.now + 5) + assert seen == ["k"] + + def test_override_that_never_reports_keeps_no_copy(self, saas: _FakeSaaS) -> None: + """A CachekitIO ``set`` override that writes without reaching the base ``set`` reports nothing: its + 2xx's bound is unknown, so the write keeps no L1 copy (0.0) rather than assume the stated TTL.""" + + class _Bypassing(CachekitIOBackend): + def set(self, key: str, value: bytes, ttl: int | None = None, stale_ttl: int | None = None) -> None: + self._request_sync("PUT", self._encode_key(key), body=value, headers=self._set_headers(ttl, stale_ttl)) + + backend, _ = fake_backend(saas.respond) + backend.__class__ = _Bypassing + saas.put_fresh_for = "5" + assert _write(StandardCacheHandler(backend)) == (True, False, 0.0) + assert saas.ops == ["PUT"] + + def test_nested_write_restores_the_outer_list(self, saas: _FakeSaaS, clock: _Clock) -> None: + """A write made inside another backend's ``set`` reports into its own list, and the outer write's + list is back in place for the outer 2xx.""" + inner_saas = _FakeSaaS(put_fresh_for="2592000") + inner, _ = fake_backend(inner_saas.respond) + nested: list[tuple[bool, bool, float | None]] = [] + + class _Nested(CachekitIOBackend): + def set(self, key: str, value: bytes, ttl: int | None = None, stale_ttl: int | None = None) -> None: + nested.append(_write(StandardCacheHandler(inner))) + super().set(key, value, ttl, stale_ttl) + + outer, _ = fake_backend(saas.respond) + outer.__class__ = _Nested + saas.put_fresh_for = "0" + assert _write(StandardCacheHandler(outer)) == (True, False, clock.now) + assert nested == [(True, False, clock.now + 60)] + assert _WRITE_DEADLINES.get() is None + + def test_reachability_is_decided_once_per_handler(self, backend: CachekitIOBackend) -> None: + """A backend that cannot report never opens the deadline list (the miss path's instruction gate); the + answer is cached on the handler, and found through wrappers, lists and forwarded attributes.""" + + class _Holder: + def __init__(self, *inner: object) -> None: + self.inner = list(inner) + + def set(self, key: str, value: bytes, ttl: int | None = None) -> None: + pass + + class _Plain: + def set(self, key: str, value: bytes, ttl: int | None = None) -> None: + pass + + assert _reaches_write_deadlines(backend) + assert _reaches_write_deadlines(_Holder(_Plain(), backend)) # a tiered wrapper's list + assert _reaches_write_deadlines(_Holder(_Holder(_Holder(backend)))) # three wrappers down + assert not _reaches_write_deadlines(_Holder(_Holder(_Holder(_Holder(backend))))) # past the depth limit + assert not _reaches_write_deadlines(_Plain()) + handler = StandardCacheHandler(_Plain()) # type: ignore[arg-type] + assert _write(handler) == (True, False, None) + assert handler._write_deadlines_reachable is False + + def test_reachability_reads_slots(self, backend: CachekitIOBackend) -> None: + """A slotted wrapper (``__slots__``, ``@dataclass(slots=True)``) has no ``__dict__``: its slots are read.""" + + @dataclasses.dataclass(slots=True) + class _SlottedDataclass: + inner: CachekitIOBackend + + class _Private: + __slots__ = ("__inner",) + + def __init__(self, inner: CachekitIOBackend) -> None: + self.__inner = inner + + class _Unset: + __slots__ = ("inner",) + + assert _reaches_write_deadlines(_SlottedDataclass(backend)) + assert _reaches_write_deadlines(_Private(backend)) # name-mangled slot + assert not _reaches_write_deadlines(_Unset()) + + def test_documented_blind_spots_keep_the_stated_ttl(self, saas: _FakeSaaS, backend: CachekitIOBackend) -> None: + """The two shapes the docs name as out of reach: a backend held only in a closure, and one assigned + to an attribute on first use (the search runs before the first write and is cached).""" + + class _Closure: + def set(self, key: str, value: bytes, ttl: int | None = None) -> None: + backend.set(key, value, ttl) + + class _BuiltOnFirstUse: + def __init__(self) -> None: + self.inner: CachekitIOBackend | None = None + + def set(self, key: str, value: bytes, ttl: int | None = None) -> None: + if self.inner is None: + self.inner = backend + self.inner.set(key, value, ttl) + + saas.put_fresh_for = "0" + assert _write(StandardCacheHandler(_Closure())) == (True, False, None) # type: ignore[arg-type] + handler = StandardCacheHandler(_BuiltOnFirstUse()) # type: ignore[arg-type] + assert _write(handler) == (True, False, None) + assert _write(handler) == (True, False, None) # still: the answer was cached at the first write + + def test_reachability_through_containers(self, backend: CachekitIOBackend) -> None: + """Routers, shard rings and pools: dict values, sets and nested containers, without spending depth.""" + + class _Holder: + def __init__(self, held: object) -> None: + self.held = held + + assert _reaches_write_deadlines(_Holder({"eu": object(), "us": backend})) # router + assert _reaches_write_deadlines(_Holder([(17, object()), (42, backend)])) # (hash, backend) ring + assert _reaches_write_deadlines(_Holder({backend})) + assert _reaches_write_deadlines(_Holder({"tier": [[{"x": backend}]]})) # nested, one wrapper level + + def test_an_ordinary_backend_answers_false_without_failing_open(self, monkeypatch: pytest.MonkeyPatch) -> None: + """A plain backend with dicts, lists, sets and slots in its attributes is searched to a clean False: + a search that raised would answer True, opening the list on every write of a non-reporting backend.""" + failed: list[str] = [] + monkeypatch.setattr(cache_handler_module.get_logger(), "debug", lambda message, *a, **k: failed.append(message)) + + class _Plain: + __slots__ = ("store", "pool", "tags", "__dict__") + + def __init__(self) -> None: + self.store = {"k": b"v", "n": {"nested": [object()]}} + self.pool = [object(), (object(), {"x": object()})] + self.tags = {"a", "b"} + self.extra = frozenset({1}) + + assert _reaches_write_deadlines(_Plain()) is False + assert failed == [] + + def test_a_cyclic_container_ends(self) -> None: + ring: list[object] = [] + ring.append(ring) + + class _Holder: + def __init__(self) -> None: + self.ring = ring + + assert _reaches_write_deadlines(_Holder()) is False # returns, and finds nothing + + def test_one_counter_bounds_any_container_shape(self) -> None: + """Every value read counts, leaves included, and a shared container is read once: shapes that grow + super-linearly when re-read finish at once and fail open.""" + + class _Wide: + def __init__(self) -> None: + row = [0] * 10**6 + self.grid = [row] * 10**6 + self.nested = [[[[0] * 100] * 100] * 100] * 100 + + assert _reaches_write_deadlines(_Wide()) # finishes only if every read counts and shared lists are read once + + def test_a_shared_wrapper_reached_by_a_longer_path_first_is_searched_to_full_reach(self, backend: CachekitIOBackend) -> None: + """``seen`` keeps the shallowest depth: a router first reached through a metrics wrapper (one level + deeper) is expanded again when the shorter route through a dict reaches it.""" + + class _Shard: + def __init__(self) -> None: + self.backend = backend + + class _Router: + def __init__(self) -> None: + self.shard = _Shard() + + class _Metrics: + def __init__(self, inner: object) -> None: + self.inner = inner + + class _Root: + def __init__(self) -> None: + router = _Router() + self.metrics = _Metrics(router) + self.routes = {"eu": router} + + assert _reaches_write_deadlines(_Root()) + + def test_a_bucket_within_budget_does_not_hide_a_route(self, backend: CachekitIOBackend) -> None: + """A bucket that leaves room in the budget is read in full, then the router's backend is found; + the same shape with no backend answers False, so the True is a find, not exhaustion.""" + + class _Wrapper: + def __init__(self, inner: object) -> None: + self.buckets = [[object() for _ in range(_REACH_READS - 100)]] + self.routes = {"us": [inner]} + + assert _reaches_write_deadlines(_Wrapper(backend)) + assert _reaches_write_deadlines(_Wrapper(object())) is False + + def test_properties_named_like_storage_never_run(self, backend: CachekitIOBackend) -> None: + """A property named ``__dict__`` (a lazy proxy) or overriding a slot name is skipped, not called.""" + ran: list[str] = [] + + class _LazyProxy: + @property + def __dict__(self) -> dict[str, object]: # type: ignore[override] + ran.append("__dict__") + return {"inner": backend} + + class _Base: + __slots__ = ("computed",) + + class _Overriding(_Base): + __slots__ = () + + @property + def computed(self) -> object: # type: ignore[override] + ran.append("computed") + return backend + + class _Holder: + def __init__(self) -> None: + self.lazy = _LazyProxy() + self.overriding = _Overriding() + self.hooked = _Hooked() + + class _Meta(type): + def __getattribute__(cls, name: str) -> Any: + ran.append(f"meta:{name}") + return super().__getattribute__(name) + + class _Hooked(metaclass=_Meta): + def __init__(self) -> None: + self.value = 1 + + ran.clear() + holder = _Holder() + ran.clear() # building the objects above may touch the metaclass; the search must not + assert _reaches_write_deadlines(holder) is False + assert ran == [] + + def test_a_container_root_holding_the_backend_is_reached(self, backend: CachekitIOBackend) -> None: + """A root that is itself a dict, list or tuple subclass: its items sit at depth 0, and their flags count.""" + + class _Router(dict): # type: ignore[type-arg] + pass + + class _Ring(list): # type: ignore[type-arg] + pass + + class _Pair(NamedTuple): + name: str + inner: object + + assert _reaches_write_deadlines(_Router(us=backend)) + assert _reaches_write_deadlines(_Ring([object(), backend])) + assert _reaches_write_deadlines(_Pair("us", backend)) + assert _reaches_write_deadlines(_Router(us=object())) is False + + def test_a_container_subclass_attribute_is_reached(self, backend: CachekitIOBackend) -> None: + """A dict-subclass router holding its backend in an attribute, not as an item: its storage is read + as well as its items, at the root and below it.""" + + class _Router(dict): # type: ignore[type-arg] + def __init__(self, inner: object) -> None: + super().__init__(eu=object()) + self.inner = inner + + class _Holder: + def __init__(self, inner: object) -> None: + self.router = _Router(inner) + + assert _reaches_write_deadlines(_Router(backend)) + assert _reaches_write_deadlines(_Holder(backend)) + assert _reaches_write_deadlines(_Router(object())) is False + + def test_class_held_backend_reached_by_a_longer_path_first(self, backend: CachekitIOBackend) -> None: + """The panel's probe: a router met first through a metrics wrapper (one level deeper), then through a + dict, with its shard's backend held on the shard's class. Its class values are yielded again from the + shorter route.""" + + class _ShardC: + pass + + _ShardC.backend = backend # type: ignore[attr-defined] + + class _Router: + def __init__(self) -> None: + self.shard = _ShardC() + + class _Metrics: + def __init__(self, inner: object) -> None: + self.inner = inner + + class _Root: + def __init__(self) -> None: + router = _Router() + self.metrics = _Metrics(router) + self.routes = {"eu": router} + + assert _reaches_write_deadlines(_Root()) + + def test_class_held_backend_after_its_type_was_first_met_at_the_limit(self, backend: CachekitIOBackend) -> None: + """Cobel's probe: a Holder first read at the depth limit (flag only), then another Holder at a shallow + depth through nested lists: its class-held backend is still searched.""" + + class _Holder: + inner = backend + + class _Box: + def __init__(self, inner: object) -> None: + self.inner = inner + + class _Root: + def __init__(self) -> None: + self.deep = _Box(_Box(_Holder())) + self.shallow = [[[_Holder()]]] + + assert _reaches_write_deadlines(_Root()) + + def test_a_none_flag_on_a_subclass_stops_the_lookup(self) -> None: + """As Python's own lookup: the first namespace holding the name wins, even with None.""" + + class _Base: + _reports_write_deadline = True + + class _Off(_Base): + _reports_write_deadline = None + + class _Holder: + def __init__(self, inner: object) -> None: + self.inner = inner + + assert _reaches_write_deadlines(_Holder(_Base())) + assert _reaches_write_deadlines(_Holder(_Off())) is False + + def test_forwarding_proxy_is_reached(self, backend: CachekitIOBackend) -> None: + """The backend lives only in a closure: the proxy is found through the root's forwarded attribute.""" + + class _Proxy: + def __getattr__(self, name: str) -> Any: + return getattr(backend, name) + + assert _reaches_write_deadlines(_Proxy()) + + def test_search_runs_no_user_code_below_the_root(self, backend: CachekitIOBackend) -> None: + """A lazy object is never evaluated, a list subclass's ``__iter__`` never runs, and a property never + fires: nothing below the root raises into the write.""" + ran: list[str] = [] + + class _Lazy: + def __getattr__(self, name: str) -> Any: + ran.append(name) + raise RuntimeError("factory failed") + + @property + def __class__(self) -> type: # type: ignore[override] + ran.append("__class__") + raise RuntimeError("factory failed") + + class _List(list): # type: ignore[type-arg] + def __iter__(self) -> Any: + ran.append("__iter__") + raise RuntimeError("iteration failed") + + class _Holder: + def __init__(self) -> None: + self.lazy = _Lazy() + self.items = _List([object(), backend]) + + assert _reaches_write_deadlines(_Holder()) + assert ran == [] + + def test_a_search_that_raises_opens_the_list(self, backend: CachekitIOBackend, monkeypatch: pytest.MonkeyPatch) -> None: + """Opening the list is always safe, so a search that fails answers True.""" + + def broken(obj: object, types_read: object, class_depths: object, depth: int) -> tuple[bool, list[object], int]: + raise RuntimeError("unexpected") + + monkeypatch.setattr(cache_handler_module, "_read_object", broken) + assert _reaches_write_deadlines(object()) + + def test_a_none_backend_or_odd_module_neither_raises_nor_logs(self, monkeypatch: pytest.MonkeyPatch) -> None: + logged: list[str] = [] + monkeypatch.setattr(cache_handler_module.get_logger(), "info", lambda message, *a, **k: logged.append(message)) + assert _reaches_write_deadlines(None) is False + assert logged == [] + odd = type("Odd", (), {}) + odd.__module__ = None # type: ignore[assignment] + assert _reaches_write_deadlines(odd()) is False + assert len(logged) == 1 + + def test_a_raising_metaclass_name_does_not_escape(self) -> None: + class _Meta(type): + @property + def __name__(cls) -> str: # type: ignore[override] + raise RuntimeError("no name") + + class _Odd(metaclass=_Meta): + pass + + assert _reaches_write_deadlines(_Odd()) is False # a clean answer: logged by qualname, nothing raised + + def test_unreachable_log_keeps_no_class_alive(self) -> None: + """The once-per-class log is keyed by names, so throwaway classes (one per MagicMock) are not kept.""" + import gc + import weakref + + kind = type("Throwaway", (), {}) + ref = weakref.ref(kind) + assert _reaches_write_deadlines(kind()) is False + del kind + gc.collect() + assert ref() is None + + def test_budget_exhaustion_opens_the_list(self, backend: CachekitIOBackend) -> None: + """A wide first level used to answer False and drop the bound; now the walk fails open.""" + + class _Striped: + def __init__(self, inner: object) -> None: + self.locks = {i: threading.Lock() for i in range(5000)} + self.inner = inner + + assert _reaches_write_deadlines(_Striped(object())) # nothing to find, but too wide to be sure + + def test_each_attribute_value_is_charged_once(self) -> None: + """A backend whose attributes or set slots fit the budget when each value is charged once answers a + clean False: charging a value both when read and when walked spends the budget and opens the list on + every write.""" + + class _Wide: + def __init__(self) -> None: + for i in range(_REACH_READS * 3 // 4): + setattr(self, f"a{i}", i) + + class _Slotted: + __slots__ = tuple(f"s{i}" for i in range(50)) + + def __init__(self) -> None: + for i, name in enumerate(self.__slots__): + setattr(self, name, i) + + class _Pool: + def __init__(self) -> None: + self.entries = [_Slotted() for _ in range(_REACH_READS * 5 // 8 // 50)] + + assert _reaches_write_deadlines(_Wide()) is False + assert _reaches_write_deadlines(_Pool()) is False + + def test_unset_slots_spend_the_budget(self) -> None: + """Reading an unset slot is still a read: many instances with many unset slots spend the budget and + fail open, instead of a descriptor call per slot per instance going uncharged.""" + + class _Sparse: + __slots__ = tuple(f"s{i}" for i in range(100)) + + class _Holder: + def __init__(self) -> None: + self.items = [_Sparse() for _ in range(1000)] + + assert _reaches_write_deadlines(_Holder()) + + def test_composing_wrapper_keeps_the_bound(self, saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock) -> None: + """A wrapper that composes a CachekitIO backend without forwarding its attributes (a metrics class) + still reaches the base ``set`` on the same thread: the list is open around every write.""" + + class _Metrics: + def __init__(self, inner: CachekitIOBackend) -> None: + self.inner = inner + self.writes = 0 + + def set(self, key: str, value: bytes, ttl: int | None = None) -> None: + self.writes += 1 + self.inner.set(key, value, ttl) + + saas.put_fresh_for = "0" + assert _write(StandardCacheHandler(_Metrics(backend))) == (True, False, clock.now) # type: ignore[arg-type] + + def test_override_raising_after_its_2xx_keeps_the_bound(self, saas: _FakeSaaS, clock: _Clock) -> None: + """The write failed as far as the handler knows, but a 2xx was received: its bound still applies.""" + + class _Raising(CachekitIOBackend): + def set(self, key: str, value: bytes, ttl: int | None = None, stale_ttl: int | None = None) -> None: + super().set(key, value, ttl, stale_ttl) + raise RuntimeError("audit hook failed") + + backend, _ = fake_backend(saas.respond) + backend.__class__ = _Raising + saas.put_fresh_for = "5" + assert _write(StandardCacheHandler(backend)) == (False, False, clock.now + 5) + + def test_forwarding_proxy_that_bypasses_set_keeps_no_copy(self, saas: _FakeSaaS, backend: CachekitIOBackend) -> None: + """The capability is read off the instance, so a proxy that forwards attributes forwards it: a write + that then never reaches the base ``set`` has an unknown bound, and keeps no copy (0.0).""" + + class _Proxy: + def __init__(self, inner: CachekitIOBackend) -> None: + self.inner = inner + + def __getattr__(self, name: str) -> Any: + return getattr(self.inner, name) + + def set(self, key: str, value: bytes, ttl: int | None = None, stale_ttl: int | None = None) -> None: + inner = self.inner + inner._request_sync("PUT", inner._encode_key(key), body=value, headers=inner._set_headers(ttl, stale_ttl)) + + saas.put_fresh_for = "5" + assert _write(StandardCacheHandler(_Proxy(backend))) == (True, False, 0.0) # type: ignore[arg-type] + assert saas.ops == ["PUT"] + + +def _decorated( + backend: CachekitIOBackend, *, is_async: bool, ttl: int | None = 60, **kwargs: Any +) -> tuple[Callable[[], Any], list[int]]: + """A decorated function counting its executions, run synchronously either way.""" + calls: list[int] = [] + namespace = _namespace() + if is_async: + + @cache(backend=backend, ttl=ttl, namespace=namespace, **kwargs) + async def compute_async() -> int: + calls.append(1) + return len(calls) + + return (lambda: asyncio.run(compute_async())), calls + + @cache(backend=backend, ttl=ttl, namespace=namespace, **kwargs) + def compute() -> int: + calls.append(1) + return len(calls) + + return compute, calls + + +@pytest.mark.parametrize("is_async", [False, True], ids=["sync", "async"]) +class TestWriterCopyBound: + """API-103 on the miss path: the L1 copy of a stored value lives min(local_ttl, effective_ttl, fresh_for). + + Each test counts the entry GETs after the miss: the async miss also re-reads under its fill lock. + """ + + def test_header_below_ttl_bounds_the_copy( + self, saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock, is_async: bool + ) -> None: + saas.put_fresh_for = "5" # the plan maximum clamped ttl=60 down to 5 s + compute, calls = _decorated(backend, is_async=is_async) + assert compute() == 1 # miss: PUT, then the bounded L1 copy + gets = saas.gets() + clock.advance(3.9) # L1 drops an entry ttl_buffer_seconds (1 s) early: it lives until +4 + assert compute() == 1 + assert saas.gets() == gets # served from L1 + clock.advance(1.2) # +5.1: past fresh_for, the copy is gone + assert compute() == 1 # from L2, not recomputed + assert saas.gets() == gets + 1 + assert len(calls) == 1 + + def test_absent_header_keeps_the_stated_ttl( + self, saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock, is_async: bool + ) -> None: + compute, calls = _decorated(backend, is_async=is_async) + assert compute() == 1 + gets = saas.gets() + clock.advance(58.5) # still inside min(local_ttl, effective_ttl) = 60, less the 1 s buffer + assert compute() == 1 + assert saas.gets() == gets + clock.advance(1.0) # +59.5: past it + assert compute() == 1 + assert saas.gets() == gets + 1 + assert len(calls) == 1 + + def test_header_above_ttl_never_extends( + self, saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock, is_async: bool + ) -> None: + saas.put_fresh_for = "2592000" + compute, _ = _decorated(backend, is_async=is_async) + assert compute() == 1 + gets = saas.gets() + clock.advance(59.5) + assert compute() == 1 + assert saas.gets() == gets + 1 # the 60 s copy expired: Fresh-For never lengthens the bound + + def test_no_ttl_is_bounded_by_the_header( + self, saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock, is_async: bool + ) -> None: + """No TTL: the server stores the plan maximum and says so; L1 keeps min(default 300 s, fresh_for).""" + saas.put_fresh_for = "10" + compute, _ = _decorated(backend, is_async=is_async, ttl=None) + assert compute() == 1 + gets = saas.gets() + clock.advance(8.5) + assert compute() == 1 + assert saas.gets() == gets + clock.advance(1.0) + assert compute() == 1 + assert saas.gets() == gets + 1 + + def test_time_after_receipt_shortens_the_copy( + self, saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock, monkeypatch: pytest.MonkeyPatch, is_async: bool + ) -> None: + """The bound runs from the 2xx's receipt, not from the L1 put: a stall between the two (here 2.2 s, + on the async path the hop back to a busy event loop) is spent from the copy, not added to it.""" + parse = CachekitIOBackend._fresh_for + + def stalled_parse(response: Any) -> int | None: + clock.advance(2.2) + return parse(response) + + monkeypatch.setattr(CachekitIOBackend, "_fresh_for", staticmethod(stalled_parse)) + saas.put_fresh_for = "2" + compute, calls = _decorated(backend, is_async=is_async, distributed_lock=False) + assert compute() == 1 + gets = saas.gets() + assert compute() == 1 # the receipt is already 2.2 s ago: no copy may serve + assert saas.gets() == gets + 1 + assert len(calls) == 1 + + def test_absent_header_times_the_stated_ttl_from_receipt( + self, saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock, monkeypatch: pytest.MonkeyPatch, is_async: bool + ) -> None: + """No Fresh-For (an older server): the stated TTL bounds the copy, still timed from the 2xx's receipt.""" + parse = CachekitIOBackend._fresh_for + + def stalled_parse(response: Any) -> int | None: + clock.advance(2.2) + return parse(response) + + monkeypatch.setattr(CachekitIOBackend, "_fresh_for", staticmethod(stalled_parse)) + compute, calls = _decorated(backend, is_async=is_async, ttl=2, distributed_lock=False) + assert compute() == 1 + gets = saas.gets() + assert compute() == 1 # the receipt is already 2.2 s ago, past ttl=2: no copy may serve + assert saas.gets() == gets + 1 + assert len(calls) == 1 + + def test_handler_set_override_is_not_bypassed( + self, saas: _FakeSaaS, backend: CachekitIOBackend, monkeypatch: pytest.MonkeyPatch, is_async: bool + ) -> None: + """The miss writes through the handler's own set / set_async, so a patch of either still decides.""" + calls: list[str] = [] + + def refusing(self: StandardCacheHandler, *args: Any, **kwargs: Any) -> bool: + calls.append("set") + return False + + async def refusing_async(self: StandardCacheHandler, *args: Any, **kwargs: Any) -> bool: + calls.append("set_async") + return False + + if is_async: + monkeypatch.setattr(StandardCacheHandler, "set_async", refusing_async) + else: + monkeypatch.setattr(StandardCacheHandler, "set", refusing) + compute, _ = _decorated(backend, is_async=is_async, distributed_lock=False) + assert compute() == 1 + assert calls + assert "PUT" not in saas.ops + + def test_container_subclass_router_keeps_a_zero_bound( + self, saas: _FakeSaaS, backend: CachekitIOBackend, is_async: bool + ) -> None: + """Helena Eagen's probe: a dict-subclass router with ``self.inner``, Fresh-For 0, keeps no copy.""" + + class _Router(dict): # type: ignore[type-arg] + def __init__(self, inner: CachekitIOBackend) -> None: + super().__init__() + self.inner = inner + + def get(self, key: str) -> bytes | None: # type: ignore[override] + return self.inner.get(key) + + def set(self, key: str, value: bytes, ttl: int | None = None) -> None: + self.inner.set(key, value, ttl) + + def delete(self, key: str) -> bool: + return self.inner.delete(key) + + def exists(self, key: str) -> bool: + return self.inner.exists(key) + + saas.put_fresh_for = "0" + compute, calls = _decorated(_Router(backend), is_async=is_async, distributed_lock=False) # type: ignore[arg-type] + assert compute() == 1 + gets = saas.gets() + assert compute() == 1 + assert saas.gets() == gets + 1 + assert len(calls) == 1 + + def test_composing_wrapper_keeps_a_zero_bound(self, saas: _FakeSaaS, backend: CachekitIOBackend, is_async: bool) -> None: + """A composing wrapper (no attribute forwarding) over CachekitIO: a 0 Fresh-For keeps no copy.""" + + class _Metrics: + def __init__(self, inner: CachekitIOBackend) -> None: + self.inner = inner + + def get(self, key: str) -> bytes | None: + return self.inner.get(key) + + def set(self, key: str, value: bytes, ttl: int | None = None) -> None: + self.inner.set(key, value, ttl) + + def delete(self, key: str) -> bool: + return self.inner.delete(key) + + def exists(self, key: str) -> bool: + return self.inner.exists(key) + + saas.put_fresh_for = "0" + compute, calls = _decorated(_Metrics(backend), is_async=is_async, distributed_lock=False) # type: ignore[arg-type] + assert compute() == 1 + gets = saas.gets() + assert compute() == 1 + assert saas.gets() == gets + 1 + assert len(calls) == 1 + + @pytest.mark.parametrize("header", ["0", "garbage", "+5", "3000000"]) + def test_malformed_or_zero_keeps_no_copy( + self, saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock, is_async: bool, header: str + ) -> None: + saas.put_fresh_for = header + compute, calls = _decorated(backend, is_async=is_async) + assert compute() == 1 + gets = saas.gets() + assert compute() == 1 # no clock movement: an L1 copy would have served this + assert saas.gets() == gets + 1 + assert len(calls) == 1 + + +class TestWriteBehindCopyBound: + """A write-behind miss puts its L1 copy before the PUT (API-36/37). Once the 2xx lands, a Fresh-For that + would end the copy sooner drops it; a value the copy already honours keeps it. A put that comes after the + 2xx is bounded by it from the start.""" + + @pytest.fixture(autouse=True) + def _writers(self, monkeypatch: pytest.MonkeyPatch) -> Iterator[list[wb.WriteBehind]]: + created: list[wb.WriteBehind] = [] + init = wb.WriteBehind.__init__ + + def recording_init(self: wb.WriteBehind, *args: Any, **kwargs: Any) -> None: + init(self, *args, **kwargs) + created.append(self) + + monkeypatch.setattr(wb.WriteBehind, "__init__", recording_init) + self.created = created + yield created + + def _drain(self) -> None: + for writer in self.created: + assert writer.close(time.monotonic() + _GUARD_SECONDS) == [], "a deferred write never finished" + + def _land_before_put(self, monkeypatch: pytest.MonkeyPatch, then: Callable[[], None] = lambda: None) -> None: + """Hold the miss after it hands its write to the writer until that write has landed, then run + ``then``: the miss's L1 put comes after the 2xx, and must see its bound.""" + submit = wb.WriteBehind.submit + + def submit_and_land(writer: wb.WriteBehind, *args: Any, **kwargs: Any) -> bool: + queued = submit(writer, *args, **kwargs) + self._drain() + then() + return queued + + monkeypatch.setattr(wb.WriteBehind, "submit", submit_and_land) + + @pytest.mark.parametrize("is_async", [False, True], ids=["sync", "async"]) + @pytest.mark.parametrize( + ("header", "copy_kept"), + [ + (None, True), + ("59", True), # floor(fresh_until - now) of an unclamped ttl=60: the 1 s buffer already covers it + ("5", False), # clamped by the plan maximum + ("0", False), + ("garbage", False), + ], + ) + def test_landed_write_drops_a_copy_it_would_outlive( + self, saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock, header: str | None, copy_kept: bool, is_async: bool + ) -> None: + """The usual order: the miss puts its copy for ttl, then the write lands and drops it if its bound is shorter.""" + saas.put_fresh_for = header + # The async locked miss path never defers (its write must land inside the lock). + compute, _ = _decorated(backend, is_async=is_async, write_behind=True, distributed_lock=False) + assert compute() == 1 + self._drain() + assert saas.ops == ["GET", "PUT"] + assert compute() == 1 + assert saas.gets() == (1 if copy_kept else 2) + + @pytest.mark.parametrize("is_async", [False, True], ids=["sync", "async"]) + @pytest.mark.parametrize( + ("header", "bound"), + [(None, 60), ("59", 59), ("5", 5), ("0", 0), ("garbage", 0)], + ) + def test_put_after_landing_is_bounded( + self, + saas: _FakeSaaS, + backend: CachekitIOBackend, + clock: _Clock, + monkeypatch: pytest.MonkeyPatch, + header: str | None, + bound: int, + is_async: bool, + ) -> None: + """A write that lands before the miss's put bounds the put itself: a copy with no time left is never + published, not even until a later drop, and a clamped one serves only inside its bound.""" + self._land_before_put(monkeypatch) + original_put = l1_module.L1Cache.put + visible: list[bool] = [] + + def observed_put(self_: l1_module.L1Cache, key: str, value: bytes, **kwargs: Any) -> None: + original_put(self_, key, value, **kwargs) + visible.append(self_.get(key)[0]) # what a concurrent reader sees right after the put + + monkeypatch.setattr(l1_module.L1Cache, "put", observed_put) + saas.put_fresh_for = header + compute, _ = _decorated(backend, is_async=is_async, write_behind=True, distributed_lock=False) + assert compute() == 1 + assert visible == [bound > 1] + gets = saas.gets() + clock.advance(max(bound - 1.5, 0)) # inside the bound, less the 1 s buffer + assert compute() == 1 + assert saas.gets() == (gets + 1 if bound <= 1 else gets) + clock.advance(1.0) # past it + assert compute() == 1 + assert saas.gets() > gets + + @pytest.mark.parametrize("is_async", [False, True], ids=["sync", "async"]) + def test_put_long_after_landing_keeps_the_receipt_bound( + self, + saas: _FakeSaaS, + backend: CachekitIOBackend, + clock: _Clock, + monkeypatch: pytest.MonkeyPatch, + is_async: bool, + ) -> None: + """A copy put 10 s after its write's 2xx serves to receipt + 59 (less the buffer), not put + 59 or + put + 60: the bound is an instant taken at the receipt.""" + self._land_before_put(monkeypatch, then=lambda: clock.advance(10)) + saas.put_fresh_for = "59" + received = clock.now + compute, _ = _decorated(backend, is_async=is_async, write_behind=True, distributed_lock=False) + assert compute() == 1 + gets = saas.gets() + clock.now = received + 57.5 + assert compute() == 1 + assert saas.gets() == gets + clock.now = received + 58.5 # past receipt + 59 less the buffer + assert compute() == 1 + assert saas.gets() == gets + 1 + + @pytest.mark.parametrize("is_async", [False, True], ids=["sync", "async"]) + @pytest.mark.parametrize("lands_first", [False, True], ids=["put-first", "write-lands-first"]) + @pytest.mark.parametrize("raiser", ["backend", "handler"]) + def test_write_failing_after_its_2xx_keeps_no_copy( + self, + saas: _FakeSaaS, + backend: CachekitIOBackend, + clock: _Clock, + monkeypatch: pytest.MonkeyPatch, + raiser: str, + lands_first: bool, + is_async: bool, + ) -> None: + """A deferred write whose PUT got Fresh-For 0 and which then raised: in the backend's ``set`` (the + handler catches it, so the write is not stored but its bound was reported), or in a patched handler + ``set`` (so ``_set_l2`` raises and the bound is lost). Either way no copy serves.""" + raised: list[str] = [] + if raiser == "backend": + + class _Raising(CachekitIOBackend): + def set(self, key: str, value: bytes, ttl: int | None = None, stale_ttl: int | None = None) -> None: + super().set(key, value, ttl, stale_ttl) + raised.append("backend") + raise RuntimeError("audit hook failed") + + backend.__class__ = _Raising + else: + original = StandardCacheHandler.set + + def raising(self_: StandardCacheHandler, *args: Any, **kwargs: Any) -> bool: + original(self_, *args, **kwargs) + raised.append("handler") + raise RuntimeError("audit hook failed") + + monkeypatch.setattr(StandardCacheHandler, "set", raising) + if lands_first: + self._land_before_put(monkeypatch) + saas.put_fresh_for = "0" + compute, _ = _decorated(backend, is_async=is_async, write_behind=True, distributed_lock=False) + assert compute() == 1 + self._drain() + assert saas.ops.count("PUT") == 1 + assert raised == [raiser] # the write really failed after its 2xx + gets = saas.gets() + clock.advance(30) + assert compute() == 1 + assert saas.gets() == gets + 1 # no L1 copy served + + +class TestPatchTtlNeverExtendsL1: + """API-104: refresh_ttl_on_get's background PATCH /ttl leaves every L1 bound as it was.""" + + async def test_refresh_leaves_the_backfill_bound(self, saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock) -> None: + namespace = _namespace() + + @cache(backend=backend, ttl=60, namespace=namespace, refresh_ttl_on_get=True) + async def compute() -> int: + return 1 + + assert await compute() == 1 # miss + PUT + l1 = l1_module.get_l1_cache(namespace) + l1.invalidate_many(list(l1._state.cache)) # drop L1 only: the next read is an L2 hit + saas.get_fresh_for = "3" # 3 s of freshness left: the backfill copy lives until +2 (1 s buffer) + assert await compute() == 1 # L2 hit -> bounded backfill, and a refresh is due + assert await asyncio.to_thread(saas.patched.wait, _GUARD_SECONDS), "no PATCH /ttl was sent" + await asyncio.sleep(0) # let the refresh task finish after its response + clock.advance(2.5) # past the backfill bound; a PATCH-derived 60 s bound would still serve L1 + gets = saas.gets() + assert await compute() == 1 + assert saas.gets() == gets + 1 + + +def _wait_for(predicate: Callable[[], bool]) -> bool: + deadline = time.monotonic() + _GUARD_SECONDS + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(0.01) + return predicate() + + +@pytest.mark.parametrize("is_async", [False, True], ids=["sync", "async"]) +class TestRevalidationCopyBound: + """API-103 on the SWR revalidation write: its L1 refresh is bounded by the revalidation PUT's 2xx.""" + + @pytest.mark.parametrize(("header", "l1_ttl"), [(None, 60), ("5", 5), ("0", 0)]) + def test_revalidated_copy_is_bounded( + self, + saas: _FakeSaaS, + backend: CachekitIOBackend, + clock: _Clock, + monkeypatch: pytest.MonkeyPatch, + is_async: bool, + header: str | None, + l1_ttl: int, + ) -> None: + namespace = _namespace() + calls: list[int] = [] + puts: list[Any] = [] # lifetime of each L1 put, from the put + put_seen = threading.Event() + original_put = l1_module.L1Cache.put + + def spy(self: l1_module.L1Cache, key: str, value: bytes, redis_ttl: Any = None, expires_at: Any = None) -> None: + original_put(self, key, value, redis_ttl=redis_ttl, expires_at=expires_at) + puts.append(redis_ttl if expires_at is None else expires_at - clock.now) + put_seen.set() + + if is_async: + + @cache(backend=backend, ttl=60, stale_ttl=60, namespace=namespace) + async def compute_async() -> int: + calls.append(1) + return len(calls) + + def compute() -> Any: + async def call_and_settle() -> int: + result = await compute_async() + if saas.stale: # a revalidation runs on this loop: hold it open until its L1 put + await asyncio.to_thread(put_seen.wait, _GUARD_SECONDS) + return result + + return asyncio.run(call_and_settle()) + else: + + @cache(backend=backend, ttl=60, stale_ttl=60, namespace=namespace) + def compute() -> int: + calls.append(1) + return len(calls) + + assert compute() == 1 # miss + PUT (no header: a 60 s copy) + l1 = l1_module.get_l1_cache(namespace) + l1.invalidate_many(list(l1._state.cache)) # the next read reaches L2, which labels it stale + saas.stale = True + saas.put_fresh_for = header + monkeypatch.setattr(l1_module.L1Cache, "put", spy) + assert compute() == 1 # stale served; the revalidation recomputes and writes in the background + assert put_seen.wait(_GUARD_SECONDS), "the revalidation never refreshed L1" + assert len(calls) == 2 + assert saas.ops.count("PUT") == 2 + assert puts == [l1_ttl] # min(local_ttl, effective_ttl, fresh_for) + assert len(l1._state.cache) == (1 if l1_ttl else 0) # 0: no copy kept + + +class TestL1DeclinedPutDropsOlderEntry: + """A put L1 declines because the value would already be expired (a 0 Fresh-For) drops the older value + under the key, as an oversized put does: the newer write must not leave the previous one serving.""" + + def test_zero_ttl_put_drops_the_older_entry(self, clock: _Clock) -> None: + l1 = l1_module.L1Cache(namespace=_namespace()) + l1.put("k", b"old", redis_ttl=60) + assert l1.get("k") == (True, b"old") + l1.put("k", b"new", redis_ttl=0) + assert l1.get("k") == (False, None) + + def test_declined_put_runs_the_fork_take_over_first(self, clock: _Clock) -> None: + """The removal touches L1's state, so a child forked without hooks must take it over first: the + lock it would otherwise wait on can be one the fork orphaned.""" + order: list[str] = [] + l1 = l1_module.L1Cache(namespace=_namespace(), before_store=lambda: order.append("take-over")) + original = l1._on_current_state + + def recording(*args: Any, **kwargs: Any) -> None: + order.append("state") + original(*args, **kwargs) + + l1._on_current_state = recording # type: ignore[method-assign] + l1.put("k", b"new", redis_ttl=0) + assert order == ["take-over", "state"] + + +@pytest.mark.parametrize("is_async", [False, True], ids=["sync", "async"]) +def test_class_attribute_backend_is_bounded(saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock, is_async: bool) -> None: + """Cobel's probe: a wrapper holding its backend as a class attribute (``inner = backend``) is found, so a + 0 Fresh-For keeps no copy.""" + + class _Wrapper: + inner = backend + + def get(self, key: str) -> bytes | None: + return self.inner.get(key) + + def set(self, key: str, value: bytes, ttl: int | None = None) -> None: + self.inner.set(key, value, ttl) + + def delete(self, key: str) -> bool: + return self.inner.delete(key) + + def exists(self, key: str) -> bool: + return self.inner.exists(key) + + saas.put_fresh_for = "0" + compute, _ = _decorated(_Wrapper(), is_async=is_async, distributed_lock=False) # type: ignore[arg-type] + assert compute() == 1 + gets = saas.gets() + assert compute() == 1 + assert saas.gets() == gets + 1 + + +@pytest.mark.parametrize("is_async", [False, True], ids=["sync", "async"]) +def test_deferred_copy_is_bounded_at_receipt_not_after_post_response_work( + saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock, monkeypatch: pytest.MonkeyPatch, is_async: bool +) -> None: + """Cobel's probe: a ``set`` override that blocks in post-response work after ``super().set`` returns. The + write-behind copy is bounded when the 2xx is read, so at receipt + 6 with Fresh-For 5 the call reaches L2. + The server also sends Fresh-For on reads, so a read's backfill cannot stand in for the writer's copy.""" + received = threading.Event() + release = threading.Event() + writers: list[wb.WriteBehind] = [] + init = wb.WriteBehind.__init__ + + def record(self: wb.WriteBehind, *args: Any, **kwargs: Any) -> None: + init(self, *args, **kwargs) + writers.append(self) + + monkeypatch.setattr(wb.WriteBehind, "__init__", record) + + class _Delayed(CachekitIOBackend): + def set(self, key: str, value: bytes, ttl: int | None = None, stale_ttl: int | None = None) -> None: + super().set(key, value, ttl, stale_ttl) + received.set() + assert release.wait(_GUARD_SECONDS) + + backend.__class__ = _Delayed + saas.put_fresh_for = "5" + saas.get_fresh_for = "5" + compute, _ = _decorated(backend, is_async=is_async, write_behind=True, distributed_lock=False) + try: + assert compute() == 1 + assert received.wait(_GUARD_SECONDS) + assert compute() == 1 # the copy was dropped at receipt: this reaches L2 and backfills under its own bound + clock.advance(6) + gets = saas.gets() + assert compute() == 1 + assert saas.gets() == gets + 1 + finally: + release.set() + for writer in writers: + assert writer.close(time.monotonic() + _GUARD_SECONDS) == [] + + +@pytest.mark.parametrize("is_async", [False, True], ids=["sync", "async"]) +def test_wrapped_class_attribute_backend_is_bounded( + saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock, is_async: bool +) -> None: + """Cobel's probe: a class-level ``__wrapped__ = backend`` is held storage, not interpreter metadata.""" + + class _Wrapper: + __wrapped__ = backend + + def get(self, key: str) -> bytes | None: + return self.__wrapped__.get(key) + + def set(self, key: str, value: bytes, ttl: int | None = None) -> None: + self.__wrapped__.set(key, value, ttl) + + def delete(self, key: str) -> bool: + return self.__wrapped__.delete(key) + + def exists(self, key: str) -> bool: + return self.__wrapped__.exists(key) + + saas.put_fresh_for = "0" + compute, _ = _decorated(_Wrapper(), is_async=is_async, distributed_lock=False) # type: ignore[arg-type] + assert compute() == 1 + gets = saas.gets() + assert compute() == 1 + assert saas.gets() == gets + 1 + + +@pytest.mark.parametrize("is_async", [False, True], ids=["sync", "async"]) +def test_a_later_shorter_report_revokes_a_copy_published_after_the_first( + saas: _FakeSaaS, backend: CachekitIOBackend, clock: _Clock, monkeypatch: pytest.MonkeyPatch, is_async: bool +) -> None: + """Cobel's probe: a set override making two PUTs. The first reports Fresh-For 59 before the miss publishes + its copy (bounded by it); the second reports 0 after, and must still revoke the published copy.""" + first = threading.Event() + proceed = threading.Event() + second = threading.Event() + release = threading.Event() + writers: list[wb.WriteBehind] = [] + init = wb.WriteBehind.__init__ + submit = wb.WriteBehind.submit + + def record(self: wb.WriteBehind, *args: Any, **kwargs: Any) -> None: + init(self, *args, **kwargs) + writers.append(self) + + def submit_after_first(self: wb.WriteBehind, *args: Any, **kwargs: Any) -> bool: + queued = submit(self, *args, **kwargs) + assert first.wait(_GUARD_SECONDS) + return queued + + monkeypatch.setattr(wb.WriteBehind, "__init__", record) + monkeypatch.setattr(wb.WriteBehind, "submit", submit_after_first) + + class _Twice(CachekitIOBackend): + def set(self, key: str, value: bytes, ttl: int | None = None, stale_ttl: int | None = None) -> None: + super().set(key, value, ttl, stale_ttl) + first.set() + assert proceed.wait(_GUARD_SECONDS) + super().set(key, value, ttl, stale_ttl) + second.set() + assert release.wait(_GUARD_SECONDS) + + backend.__class__ = _Twice + saas.put_fresh_for = "59" + compute, _ = _decorated(backend, is_async=is_async, write_behind=True, distributed_lock=False) + try: + assert compute() == 1 # the first report lands before the miss publishes its copy + saas.put_fresh_for = "0" + proceed.set() + assert second.wait(_GUARD_SECONDS) + release.set() + for writer in writers: + assert writer.close(time.monotonic() + _GUARD_SECONDS) == [] + gets = saas.gets() + assert compute() == 1 + assert saas.gets() == gets + 1 + finally: + proceed.set() + release.set() + for writer in writers: + writer.close(time.monotonic() + _GUARD_SECONDS) diff --git a/tests/unit/test_swr_decorator.py b/tests/unit/test_swr_decorator.py index 9eca5951..acf733c1 100644 --- a/tests/unit/test_swr_decorator.py +++ b/tests/unit/test_swr_decorator.py @@ -30,7 +30,7 @@ from cachekit.config.validation import ConfigurationError from tests.utils.fork_helpers import child_outcome, report, run_fresh -_CAP = 2_592_000 # 30-day storage cap shared by ttl + stale_ttl +_CAP = 7_776_000 # 90-day TTL range shared by ttl + stale_ttl class FakeSWRBackend: @@ -240,7 +240,7 @@ def compute() -> str: assert compute() == "v" assert backend.set_calls == [(60, None)] # no window: explicit opt-out - def test_preset_default_caps_window_to_30_day_total(self) -> None: + def test_preset_default_caps_window_to_90_day_total(self) -> None: backend = FakeSWRBackend() ttl = _CAP - 100 # leaves only 100s of window headroom config = DecoratorConfig(backend=backend, ttl=ttl, swr_by_default=True) @@ -255,7 +255,7 @@ def compute() -> str: def test_preset_default_window_fits_the_cap_after_rounding(self) -> None: """A fractional ttl is ceiled on the wire, so the derived window is bounded by the ceiled TTL.""" backend = FakeSWRBackend() - ttl = _CAP / 2 + 0.5 # goes out as 1_296_001 + ttl = _CAP / 2 + 0.5 # goes out as 3_888_001 config = DecoratorConfig(backend=backend, ttl=ttl, swr_by_default=True) # type: ignore[arg-type] @cache(config=config) @@ -273,11 +273,32 @@ def test_stale_ttl_without_ttl_raises(self) -> None: def f() -> None: ... def test_total_over_cap_raises(self) -> None: - with pytest.raises(ConfigurationError, match="30-day"): + with pytest.raises(ConfigurationError, match="90-day"): - @cache(backend=FakeSWRBackend(), ttl=2_000_000, stale_ttl=1_000_000, l1_enabled=False) + @cache(backend=FakeSWRBackend(), ttl=_CAP - 1_000_000, stale_ttl=1_000_001, l1_enabled=False) def f() -> None: ... + def test_total_at_cap_accepted(self) -> None: + """ttl + stale_ttl == 7,776,000 is the largest valid write; the old 30-day cap no longer applies.""" + backend = FakeSWRBackend() + + @cache(backend=backend, ttl=_CAP - 1_000_000, stale_ttl=1_000_000, l1_enabled=False) + def f() -> str: + return "v" + + assert f() == "v" + assert backend.set_calls == [(_CAP - 1_000_000, 1_000_000)] + + def test_total_past_old_30_day_cap_accepted(self) -> None: + backend = FakeSWRBackend() + + @cache(backend=backend, ttl=2_000_000, stale_ttl=1_000_000, l1_enabled=False) + def f() -> str: + return "v" + + assert f() == "v" + assert backend.set_calls == [(2_000_000, 1_000_000)] + def test_non_swr_backend_raises(self) -> None: class PlainBackend: def get(self, k: str) -> None: @@ -743,7 +764,7 @@ class TestSWRSchedulingHardening: slot release when scheduling fails.""" def test_default_window_off_when_ttl_at_or_above_cap(self) -> None: - """ttl ≥ the 30-day cap leaves no window headroom: SWR silently off, + """ttl ≥ the 90-day cap leaves no window headroom: SWR silently off, never a negative stale_ttl.""" backend = FakeSWRBackend() config = DecoratorConfig(backend=backend, ttl=_CAP + 100, swr_by_default=True)