diff --git a/README.md b/README.md index ab0aa85..0b12b08 100644 --- a/README.md +++ b/README.md @@ -101,15 +101,22 @@ $state = $atomic->getAndDelete('oauth.state.42'); | Backend | Atomic cache coordination | Scope | |---|---|---| | Array memory | Yes | one PHP process | +| WeakMap | Yes | one PHP process / adapter instance | | Shared memory | Yes | one host / shared SysV segment | | Redis / Valkey | Yes | supplied authoritative Redis-compatible store | | Redis Cluster | Yes | stable CacheLayer hash-slot bucket | | MongoDB | Yes | supplied authoritative collection | -| APCu / Memcached | No | no full atomic consume primitive | -| PDO / SQLite | No | no cross-driver atomic contract in 3.3 | -| File / PHP files | No | ordinary writers do not share one atomic key lock | -| ScyllaDB | No | no full three-operation contract exposed | -| WeakMap / Null / Tiered | No | not one authoritative coordination domain | +| Memcached | Yes | supplied Memcached key/CAS domain | +| SQLite PDO | Yes | supplied SQLite database | +| PostgreSQL PDO | Yes | supplied PostgreSQL database | +| MySQL / MariaDB PDO | Yes | supplied transactional database | +| File / PHP files | Yes | one reliable filesystem lock domain | +| APCu | No | no arbitrary-value CAS / consume primitive | +| Generic / unknown PDO | No | no portable cross-driver atomic contract | +| ScyllaDB | No | ordinary cache writes do not use the LWT architecture required for the full contract | +| Null / Tiered | No | no single retained authoritative coordination domain | + +Memcached uses native `add()`/CAS primitives and a reserved tombstone for linearizable logical consumption. File and PHP-file caches route ordinary key writers and atomic mutations through the same deterministic per-key `flock()`; their guarantee is therefore limited to a filesystem where that lock domain is reliable. PDO capability is runtime-qualified: SQLite, PostgreSQL, and MySQL/MariaDB expose it, while unknown PDO drivers do not. SQLite coordinates writers with `BEGIN IMMEDIATE`; PostgreSQL and MySQL/MariaDB use transactional row locking. Atomic PDO calls require CacheLayer to own the transaction and reject execution inside a caller-owned active transaction. Use dedicated, untagged keys for portable replay claims, nonces, challenges, state transitions, and one-time state. Tag rotation is a separate invalidation mechanism and is not part of the portable atomic linearization boundary. Redis/Valkey, Redis Cluster, and MongoDB therefore reject `compareAndSet()` on tagged records. Array memory and SharedMemory can validate tag generations inside their own local atomic domain, but callers should not depend on tagged CAS when code must be portable across backends. Tiered caches remain non-atomic even when an individual tier supports the capability. diff --git a/docs/cache.rst b/docs/cache.rst index d97a129..1537c55 100644 --- a/docs/cache.rst +++ b/docs/cache.rst @@ -107,32 +107,54 @@ never implements these methods as public ``has()/get()`` followed by The supported facade backends are: -================= ====== ================================================ -Backend Atomic Consistency domain -================= ====== ================================================ -Array memory yes one PHP process -Shared memory yes one host / shared SysV segment -Redis / Valkey yes the supplied authoritative Redis-compatible store -Redis Cluster yes the key's stable CacheLayer hash-slot bucket -MongoDB yes the supplied authoritative MongoDB collection -APCu no no full atomic consume primitive -Memcached no no full atomic consume primitive -PDO / SQLite no no cross-driver contract is claimed in 3.3 -File / PHP files no ordinary writers do not share an atomic key lock -ScyllaDB no no full three-operation contract is exposed -WeakMap no process-local object cache, not coordination -Null store no non-authoritative sink -Tiered cache no tiers cannot form one linearizable authority -================= ====== ================================================ +========================== ====== ==================================================== +Backend Atomic Consistency domain +========================== ====== ==================================================== +Array memory yes one PHP process +WeakMap yes one PHP process / adapter instance +Shared memory yes one host / shared SysV segment +Redis / Valkey yes the supplied authoritative Redis-compatible store +Redis Cluster yes the key's stable CacheLayer hash-slot bucket +MongoDB yes the supplied authoritative MongoDB collection +Memcached yes the supplied authoritative Memcached CAS domain +PDO: SQLite yes the supplied SQLite database +PDO: PostgreSQL yes the supplied PostgreSQL database +PDO: MySQL / MariaDB yes the supplied transactional database +File yes one reliable filesystem ``flock()`` domain +PHP files yes one reliable filesystem ``flock()`` domain +APCu no no arbitrary-value CAS/consume primitive +PDO: other drivers no no portable transaction/locking contract is claimed +ScyllaDB no LWT is not mixed with ordinary cache writes +Null store no non-authoritative sink +Tiered cache no tiers cannot form one linearizable authority +========================== ====== ==================================================== + +Memcached uses native CAS tokens for replacement. Atomic consume linearizes by +CAS-replacing the value with an internal short-lived tombstone; ordinary reads +treat the tombstone as a miss, while ``setIfAbsent()`` can immediately reclaim +it through CAS. The tombstone is never exposed as a logical cache value. + +File and PHP-file stores use deterministic per-key ``flock()`` files. Ordinary +``save()`` and ``deleteItem()`` mutations take the same key lock as atomic +operations, while reads remain lock-free over atomic rename replacement. Their +atomic guarantee therefore applies only inside a filesystem domain where +``flock()`` semantics are reliable. + +PDO atomic capability is driver-qualified. SQLite uses an immediate writer +transaction; PostgreSQL and MySQL/MariaDB use transactional row locking plus +conflict-safe inserts. Other PDO drivers return no atomic capability rather than +claiming a cross-driver guarantee. Atomic PDO operations require ownership of +the PDO transaction and reject execution inside an already-active caller +transaction. Use dedicated, untagged keys for portable replay claims, nonces, challenges, state transitions, one-time state, and similar coordination. Tag invalidation is a separate cache coordination mechanism and is not part of the portable -atomic linearization boundary. Redis/Valkey, Redis Cluster, and MongoDB reject -``compareAndSet()`` on tagged records because their tag metadata cannot join the -same atomic replacement condition. Array memory and SharedMemory can validate -tag generations inside their local atomic domain, but portable protocols should -not depend on tagged CAS behavior. +atomic linearization boundary. Redis/Valkey, Redis Cluster, MongoDB, and +Memcached reject ``compareAndSet()`` on tagged records because their tag +metadata cannot join the same atomic replacement condition. Process-local and +single-lock-domain backends can validate tag generations inside their atomic +domain, but portable protocols should not depend on tagged CAS behavior. ``atomic()`` returns ``null`` rather than emulating missing backend primitives. This remains true for a tiered facade even when one or more individual tiers diff --git a/src/Cache/Adapter/ConditionalAtomicCachePoolInterface.php b/src/Cache/Adapter/ConditionalAtomicCachePoolInterface.php new file mode 100644 index 0000000..6a9c1b7 --- /dev/null +++ b/src/Cache/Adapter/ConditionalAtomicCachePoolInterface.php @@ -0,0 +1,11 @@ +createDirectory($namespace, $baseDir); } + public function atomicCompareAndSet(string $key, mixed $expected, CacheItemInterface $replacement): bool + { + if (!$this->supportsItem($replacement)) { + return false; + } + $expiration = CachePayloadCodec::expirationFromItem($replacement); + if ($expiration['ttl'] !== null && $expiration['ttl'] <= 0) { + return false; + } + + return $this->withKeyLock($key, function () use ($key, $expected, $replacement): bool { + $record = $this->readLiveRecordUnlocked($key); + if (!$record instanceof CacheRecord || $record->value !== $expected) { + return false; + } + + return $this->persistItemUnlocked($replacement); + }); + } + + public function atomicGetAndDelete(string $key): CacheItemInterface + { + return $this->withKeyLock($key, function () use ($key): CacheItemInterface { + $record = $this->readLiveRecordUnlocked($key); + if (!$record instanceof CacheRecord) { + return $this->genericMiss($key); + } + $this->deleteItemUnlocked($key); + + return $this->genericItemFromRecord($key, $record); + }); + } + + public function atomicSetIfAbsent(CacheItemInterface $item): bool + { + if (!$this->supportsItem($item)) { + return false; + } + $expiration = CachePayloadCodec::expirationFromItem($item); + if ($expiration['ttl'] !== null && $expiration['ttl'] <= 0) { + return false; + } + + return $this->withKeyLock($item->getKey(), function () use ($item): bool { + if ($this->readLiveRecordUnlocked($item->getKey()) instanceof CacheRecord) { + return false; + } + + return $this->persistItemUnlocked($item); + }); + } + public function clear(): bool { $ok = true; @@ -58,15 +95,10 @@ public function clear(): bool public function deleteItem(string $key): bool { - $file = $this->fileFor($key); - - return !is_file($file) || unlink($file); + return $this->withKeyLock($key, fn(): bool => $this->deleteItemUnlocked($key)); } - /** - * @param array $keys The keys argument. - * @phpstan-param list $keys - */ + /** @param list $keys */ public function deleteItems(array $keys): bool { $ok = true; @@ -79,23 +111,18 @@ public function deleteItems(array $keys): bool public function getItem(string $key): CacheItem { - $file = $this->fileFor($key); - - if (is_file($file)) { - $raw = file_get_contents($file); - if (is_string($raw)) { - $record = $this->decodeRecordFromBlob($raw); - if ($record !== null) { - return $this->genericItemFromRecord($key, $record); - } - } - unlink($file); + $record = $this->readLiveRecordUnlocked($key); + if ($record instanceof CacheRecord) { + return $this->genericItemFromRecord($key, $record); } return new CacheItem($this, $key); } - /** @param list $tags */ + /** + * @param list $tags + * @return array + */ #[\Override] public function getTagGenerations(array $tags): array { @@ -127,28 +154,8 @@ public function hasItem(string $key): bool public function multiFetch(array $keys): array { $items = []; - $stale = []; foreach ($keys as $key) { - $file = $this->fileFor($key); - $raw = is_file($file) ? file_get_contents($file) : false; - if (!is_string($raw)) { - $items[$key] = $this->genericMiss($key); - - continue; - } - $record = $this->decodeRecordFromBlob($raw); - if ($record === null) { - $stale[] = $file; - $items[$key] = $this->genericMiss($key); - - continue; - } - $items[$key] = $this->genericItemFromRecord($key, $record); - } - foreach ($stale as $file) { - if (is_file($file)) { - unlink($file); - } + $items[$key] = $this->getItem($key); } return $items; @@ -199,18 +206,12 @@ private function createDirectory(string $ns, ?string $baseDir): void $root = $baseDir . DIRECTORY_SEPARATOR . 'cache_' . $ns . DIRECTORY_SEPARATOR; $this->dataDirectory = $root . 'data' . DIRECTORY_SEPARATOR; $this->metadataDirectory = $root . 'meta' . DIRECTORY_SEPARATOR; - - if (is_dir($this->dataDirectory) && is_dir($this->metadataDirectory)) { - $this->assertSecureDirectory($baseDir, 'Cache base directory'); - $this->assertSecureDirectory($this->dataDirectory, 'Cache data directory'); - $this->assertSecureDirectory($this->metadataDirectory, 'Cache metadata directory'); - - return; - } + $this->lockDirectory = $root . 'locks' . DIRECTORY_SEPARATOR; $this->ensureBaseDirectoryExists($baseDir); - $this->ensureCacheDirectoryExists($this->dataDirectory); - $this->ensureCacheDirectoryExists($this->metadataDirectory); + foreach ([$this->dataDirectory, $this->metadataDirectory, $this->lockDirectory] as $directory) { + $this->ensureCacheDirectoryExists($directory); + } } private function defaultBaseDirectory(): string @@ -220,37 +221,34 @@ private function defaultBaseDirectory(): string . str_replace('/', DIRECTORY_SEPARATOR, self::DEFAULT_BASE_DIR); } + private function deleteItemUnlocked(string $key): bool + { + $file = $this->fileFor($key); + + return !is_file($file) || unlink($file); + } + private function ensureBaseDirectoryExists(string $baseDir): void { $this->assertPathNotSymlink($baseDir, 'Cache base directory'); - if (file_exists($baseDir) && !is_dir($baseDir)) { - throw new RuntimeException( - 'Cache base path ' . realpath($baseDir) . ' exists and is *not* a directory', - ); + throw new RuntimeException('Cache base path ' . realpath($baseDir) . ' exists and is *not* a directory'); } - if (!is_dir($baseDir) && !mkdir($baseDir, 0700, true) && !is_dir($baseDir)) { $this->throwCreationError('Failed to create base directory ' . $baseDir); } - $this->assertSecureDirectory($baseDir, 'Cache base directory'); } private function ensureCacheDirectoryExists(string $cacheDir): void { $this->assertPathNotSymlink($cacheDir, 'Cache directory'); - if (file_exists($cacheDir) && !is_dir($cacheDir)) { - throw new RuntimeException( - realpath($cacheDir) . ' exists and is not a directory', - ); + throw new RuntimeException(realpath($cacheDir) . ' exists and is not a directory'); } - - if (!mkdir($cacheDir, 0700, true) && !is_dir($cacheDir)) { + if (!is_dir($cacheDir) && !mkdir($cacheDir, 0700, true) && !is_dir($cacheDir)) { $this->throwCreationError('Failed to create cache directory ' . $cacheDir); } - $this->assertSecureDirectory($cacheDir, 'Cache directory'); } @@ -266,11 +264,15 @@ private function metadataFileFor(string $tag): string private function persistItem(CacheItemInterface $item): bool { + return $this->withKeyLock($item->getKey(), fn(): bool => $this->persistItemUnlocked($item)); + } + private function persistItemUnlocked(CacheItemInterface $item): bool + { $expires = CachePayloadCodec::expirationFromItem($item); $ttl = $expires['ttl']; if ($ttl !== null && $ttl <= 0) { - return $this->deleteItem($item->getKey()); + return $this->deleteItemUnlocked($item->getKey()); } $blob = $this->encodeItem($item, $expires['expiresAt']); @@ -278,7 +280,6 @@ private function persistItem(CacheItemInterface $item): bool if ($tmp === false) { return false; } - if (file_put_contents($tmp, $blob, LOCK_EX) === false) { if (is_file($tmp)) { unlink($tmp); @@ -286,7 +287,6 @@ private function persistItem(CacheItemInterface $item): bool return false; } - if (!rename($tmp, $this->fileFor($item->getKey()))) { if (is_file($tmp)) { unlink($tmp); @@ -298,10 +298,61 @@ private function persistItem(CacheItemInterface $item): bool return true; } + private function readLiveRecordUnlocked(string $key): ?CacheRecord + { + $file = $this->fileFor($key); + $raw = is_file($file) ? file_get_contents($file) : false; + $record = is_string($raw) ? $this->decodeRecordFromBlob($raw) : null; + if (!$record instanceof CacheRecord || !$this->recordTagsAreCurrent($record)) { + return null; + } + + return $record; + } + + private function recordTagsAreCurrent(CacheRecord $record): bool + { + foreach ($record->tags as $tag => $generation) { + $current = is_file($this->metadataFileFor($tag)) + ? file_get_contents($this->metadataFileFor($tag)) + : false; + if (!is_string($current) || strtolower($current) !== $generation) { + return false; + } + } + + return true; + } + private function throwCreationError(string $prefix): void { $err = error_get_last()['message'] ?? 'unknown error'; throw new RuntimeException($prefix . ": $err"); } + + /** + * @template T + * @param callable(): T $callback + * @return T + */ + private function withKeyLock(string $key, callable $callback): mixed + { + $path = $this->lockDirectory . hash('xxh128', $key) . '.lock'; + $handle = fopen($path, 'c'); + if (!is_resource($handle) || !flock($handle, LOCK_EX)) { + if (is_resource($handle)) { + fclose($handle); + } + + throw new RuntimeException('Unable to acquire file cache key lock.'); + } + + try { + return $callback(); + } finally { + flock($handle, LOCK_UN); + fclose($handle); + } + } } diff --git a/src/Cache/Adapter/MemcachedCacheAdapter.php b/src/Cache/Adapter/MemcachedCacheAdapter.php index 4f97257..2578b50 100644 --- a/src/Cache/Adapter/MemcachedCacheAdapter.php +++ b/src/Cache/Adapter/MemcachedCacheAdapter.php @@ -5,12 +5,15 @@ namespace Infocyph\CacheLayer\Cache\Adapter; use Infocyph\CacheLayer\Cache\CacheInput; +use Infocyph\CacheLayer\Cache\CacheRecord; use Infocyph\CacheLayer\Cache\Item\CacheItem; use Psr\Cache\CacheItemInterface; use RuntimeException; -final class MemcachedCacheAdapter extends AbstractCacheAdapter +final class MemcachedCacheAdapter extends AbstractCacheAdapter implements AtomicCachePoolInterface { + private const string ATOMIC_TOMBSTONE = "\0cachelayer:atomic:tombstone:v1\0"; + private readonly \Memcached $client; private readonly string $namespace; @@ -33,6 +36,117 @@ public function __construct( } } + public function atomicCompareAndSet( + string $key, + mixed $expected, + CacheItemInterface $replacement, + ): bool { + if (!$this->supportsItem($replacement)) { + return false; + } + + $expiration = CachePayloadCodec::expirationFromItem($replacement); + $ttl = $expiration['ttl']; + if ($ttl !== null && $ttl <= 0) { + return false; + } + + $mapped = $this->mapData($key); + $extended = $this->extendedGet($mapped); + if ($extended === null) { + return false; + } + + $blob = $extended['value']; + if ($blob === self::ATOMIC_TOMBSTONE) { + return false; + } + + $record = $this->decodeRecordFromBlob($blob); + if (!$record instanceof CacheRecord + || $record->namespaceGeneration !== $this->namespaceGeneration() + || $record->tags !== [] + || $record->value !== $expected) { + return false; + } + + $replacementBlob = $this->encodeItem( + $replacement, + $expiration['expiresAt'], + $this->namespaceGeneration(), + ); + + return $this->client->cas( + $extended['cas'], + $mapped, + $replacementBlob, + $ttl ?? 0, + ); + } + + public function atomicGetAndDelete(string $key): CacheItemInterface + { + $mapped = $this->mapData($key); + $extended = $this->extendedGet($mapped); + if ($extended === null || $extended['value'] === self::ATOMIC_TOMBSTONE) { + return $this->genericMiss($key); + } + + $record = $this->decodeRecordFromBlob($extended['value']); + if (!$record instanceof CacheRecord + || $record->namespaceGeneration !== $this->namespaceGeneration() + || !$this->recordTagsAreCurrent($record)) { + return $this->genericMiss($key); + } + + if (!$this->client->cas($extended['cas'], $mapped, self::ATOMIC_TOMBSTONE, 1)) { + return $this->genericMiss($key); + } + + return $this->genericItemFromRecord($key, $record); + } + + public function atomicSetIfAbsent(CacheItemInterface $item): bool + { + if (!$this->supportsItem($item)) { + return false; + } + + $expiration = CachePayloadCodec::expirationFromItem($item); + $ttl = $expiration['ttl']; + if ($ttl !== null && $ttl <= 0) { + return false; + } + + $mapped = $this->mapData($item->getKey()); + $blob = $this->encodeItem( + $item, + $expiration['expiresAt'], + $this->namespaceGeneration(), + ); + + if ($this->client->add($mapped, $blob, $ttl ?? 0)) { + return true; + } + + $extended = $this->extendedGet($mapped); + if ($extended === null) { + return $this->client->add($mapped, $blob, $ttl ?? 0); + } + + $current = $extended['value']; + $record = $current === self::ATOMIC_TOMBSTONE + ? null + : $this->decodeRecordFromBlob($current); + if ($record instanceof CacheRecord + && $record->namespaceGeneration === $this->namespaceGeneration() + && $this->recordTagsAreCurrent($record)) { + return false; + } + + return $this->client->cas($extended['cas'], $mapped, $blob, $ttl ?? 0); + } + public function clear(): bool { $cleared = $this->client->set($this->generationKey(), self::newGeneration()); @@ -80,6 +194,9 @@ public function getItem(string $key): CacheItem $stored = is_array($stored) ? $stored : []; $generation = $this->namespaceGeneration($stored[$this->generationKey()] ?? null); $blob = $stored[$mapped] ?? null; + if ($blob === self::ATOMIC_TOMBSTONE) { + return $this->genericMiss($key); + } $record = is_string($blob) ? $this->decodeRecordFromBlob($blob) : null; if ($record !== null && $record->namespaceGeneration === $generation) { return $this->genericItemFromRecord($key, $record); @@ -139,6 +256,11 @@ public function multiFetch(array $keys): array foreach ($keys as $key) { $mapped = $this->mapData($key); $blob = $stored[$mapped] ?? null; + if ($blob === self::ATOMIC_TOMBSTONE) { + $items[$key] = $this->genericMiss($key); + + continue; + } $record = is_string($blob) ? $this->decodeRecordFromBlob($blob) : null; if ($record === null || $record->namespaceGeneration !== $generation) { $items[$key] = $this->genericMiss($key); @@ -222,6 +344,21 @@ public function saveItems(array $items): bool return true; } + /** @return array{value:string, cas:int|float}|null */ + private function extendedGet(string $key): ?array + { + $value = $this->client->get($key, null, \Memcached::GET_EXTENDED); + if (!is_array($value) || !is_string($value['value'] ?? null)) { + return null; + } + $cas = $value['cas'] ?? null; + if (!is_int($cas) && !is_float($cas)) { + return null; + } + + return ['value' => $value['value'], 'cas' => $cas]; + } + private function generationKey(): string { return $this->namespace . ':m:generation'; @@ -260,6 +397,22 @@ private function namespaceGeneration(mixed $value = null): string return $generation; } + private function recordTagsAreCurrent(CacheRecord $record): bool + { + if ($record->tags === []) { + return true; + } + + $current = $this->getTagGenerations(array_keys($record->tags)); + foreach ($record->tags as $tag => $generation) { + if (($current[$tag] ?? null) !== $generation) { + return false; + } + } + + return true; + } + private function tagGeneration(string $key, mixed $value): string { $generation = self::normalizeGeneration($value); diff --git a/src/Cache/Adapter/PdoAtomicOperations.php b/src/Cache/Adapter/PdoAtomicOperations.php new file mode 100644 index 0000000..6b8b379 --- /dev/null +++ b/src/Cache/Adapter/PdoAtomicOperations.php @@ -0,0 +1,218 @@ +supportsAtomicCache() || !$this->supportsItem($replacement)) { + return false; + } + $expiration = CachePayloadCodec::expirationFromItem($replacement); + if ($expiration['ttl'] !== null && $expiration['ttl'] <= 0) { + return false; + } + + return $this->atomicTransaction(function () use ($key, $expected, $replacement, $expiration): bool { + $row = $this->atomicFetchRow($key); + $record = $row === null ? null : $this->atomicRecordFromRow($row); + if (!$record instanceof CacheRecord + || !$this->atomicRecordTagsAreCurrent($record) + || $record->value !== $expected) { + return false; + } + + return $this->atomicUpdateExisting( + $key, + $this->encodeItem($replacement, $expiration['expiresAt']), + $expiration['expiresAt'], + ); + }); + } + + public function atomicGetAndDelete(string $key): CacheItemInterface + { + if (!$this->supportsAtomicCache()) { + return $this->genericMiss($key); + } + + return $this->atomicTransaction(function () use ($key): CacheItemInterface { + $row = $this->atomicFetchRow($key); + if ($row === null) { + return $this->genericMiss($key); + } + + $record = $this->atomicRecordFromRow($row); + $this->deleteItem($key); + if (!$record instanceof CacheRecord || !$this->atomicRecordTagsAreCurrent($record)) { + return $this->genericMiss($key); + } + + return $this->genericItemFromRecord($key, $record); + }); + } + + public function atomicSetIfAbsent(CacheItemInterface $item): bool + { + if (!$this->supportsAtomicCache() || !$this->supportsItem($item)) { + return false; + } + $expiration = CachePayloadCodec::expirationFromItem($item); + if ($expiration['ttl'] !== null && $expiration['ttl'] <= 0) { + return false; + } + + return $this->atomicTransaction(function () use ($item, $expiration): bool { + $key = $item->getKey(); + $row = $this->atomicFetchRow($key); + $record = $row === null ? null : $this->atomicRecordFromRow($row); + if ($record instanceof CacheRecord && $this->atomicRecordTagsAreCurrent($record)) { + return false; + } + + $payload = $this->encodeItem($item, $expiration['expiresAt']); + if ($row !== null) { + return $this->atomicUpdateExisting($key, $payload, $expiration['expiresAt']); + } + + return $this->atomicInsertIfMissing($key, $payload, $expiration['expiresAt']); + }); + } + + public function supportsAtomicCache(): bool + { + return in_array($this->driver, ['sqlite', 'pgsql', 'mysql', 'mariadb'], true); + } + + /** @return array{payload:string, expires:int|null}|null */ + private function atomicFetchRow(string $key): ?array + { + $suffix = in_array($this->driver, ['pgsql', 'mysql', 'mariadb'], true) ? ' FOR UPDATE' : ''; + $statement = $this->pdo->prepare( + "SELECT payload, expires FROM {$this->table} " + . "WHERE namespace = ? AND kind = ? AND cache_key = ?{$suffix}", + ); + $statement->execute([$this->namespace, self::KIND_DATA, $key]); + $row = $statement->fetch(PDO::FETCH_ASSOC); + if (!is_array($row)) { + return null; + } + + $payload = $row['payload'] ?? null; + if (is_resource($payload)) { + $payload = stream_get_contents($payload); + } + if (!is_string($payload)) { + return null; + } + + return [ + 'payload' => $payload, + 'expires' => is_numeric($row['expires'] ?? null) ? (int) $row['expires'] : null, + ]; + } + + private function atomicInsertIfMissing(string $key, string $payload, ?int $expires): bool + { + if (in_array($this->driver, ['pgsql', 'sqlite'], true)) { + $sql = "INSERT INTO {$this->table} (namespace, kind, cache_key, payload, expires) " + . 'VALUES (?, ?, ?, ?, ?) ON CONFLICT (namespace, kind, cache_key) DO NOTHING'; + } else { + $sql = "INSERT IGNORE INTO {$this->table} " + . '(namespace, kind, cache_key, payload, expires) VALUES (?, ?, ?, ?, ?)'; + } + $statement = $this->pdo->prepare($sql); + $statement->execute([$this->namespace, self::KIND_DATA, $key, $payload, $expires]); + + return $statement->rowCount() === 1; + } + + /** @param array{payload:string, expires:int|null} $row */ + private function atomicRecordFromRow(array $row): ?CacheRecord + { + if (CachePayloadCodec::isExpired($row['expires'])) { + return null; + } + + $record = $this->decodeRecordFromBlob($row['payload']); + + return $record instanceof CacheRecord ? $record : null; + } + + private function atomicRecordTagsAreCurrent(CacheRecord $record): bool + { + if ($record->tags === []) { + return true; + } + + $rows = $this->fetchRows(self::KIND_TAG, array_keys($record->tags)); + foreach ($record->tags as $tag => $generation) { + if (($rows[$tag]['payload'] ?? null) !== $generation) { + return false; + } + } + + return true; + } + + /** + * @template T + * @param callable(): T $callback + * @return T + */ + private function atomicTransaction(callable $callback): mixed + { + if ($this->pdo->inTransaction()) { + throw new RuntimeException('Atomic PDO cache operations require ownership of the PDO transaction.'); + } + + $sqlite = $this->driver === 'sqlite'; + if ($sqlite) { + $this->pdo->exec('BEGIN IMMEDIATE'); + } else { + $this->pdo->beginTransaction(); + } + + try { + $result = $callback(); + if ($sqlite) { + $this->pdo->exec('COMMIT'); + } else { + $this->pdo->commit(); + } + + return $result; + } catch (Throwable $failure) { + if ($sqlite) { + $this->pdo->exec('ROLLBACK'); + } else { + $this->pdo->rollBack(); + } + + throw $failure; + } + } + + private function atomicUpdateExisting(string $key, string $payload, ?int $expires): bool + { + $statement = $this->pdo->prepare( + "UPDATE {$this->table} SET payload = ?, expires = ? " + . 'WHERE namespace = ? AND kind = ? AND cache_key = ?', + ); + + return $statement->execute([$payload, $expires, $this->namespace, self::KIND_DATA, $key]); + } +} diff --git a/src/Cache/Adapter/PdoCacheAdapter.php b/src/Cache/Adapter/PdoCacheAdapter.php index 0fb4ba9..0433f84 100644 --- a/src/Cache/Adapter/PdoCacheAdapter.php +++ b/src/Cache/Adapter/PdoCacheAdapter.php @@ -11,8 +11,10 @@ use Psr\Cache\CacheItemInterface; use RuntimeException; -final class PdoCacheAdapter extends AbstractCacheAdapter +final class PdoCacheAdapter extends AbstractCacheAdapter implements ConditionalAtomicCachePoolInterface { + use PdoAtomicOperations; + private const int BATCH_SIZE = 250; private const string DEFAULT_SQLITE_DIR = 'cachelayer/pdo'; diff --git a/src/Cache/Adapter/PhpFilesCacheAdapter.php b/src/Cache/Adapter/PhpFilesCacheAdapter.php index 3fac0cb..f72d6ef 100644 --- a/src/Cache/Adapter/PhpFilesCacheAdapter.php +++ b/src/Cache/Adapter/PhpFilesCacheAdapter.php @@ -5,11 +5,12 @@ namespace Infocyph\CacheLayer\Cache\Adapter; use Infocyph\CacheLayer\Cache\CacheInput; +use Infocyph\CacheLayer\Cache\CacheRecord; use Infocyph\CacheLayer\Cache\Item\CacheItem; use Psr\Cache\CacheItemInterface; use RuntimeException; -final class PhpFilesCacheAdapter extends AbstractCacheAdapter +final class PhpFilesCacheAdapter extends AbstractCacheAdapter implements AtomicCachePoolInterface { use SecuresFilesystemDirectories; @@ -17,6 +18,8 @@ final class PhpFilesCacheAdapter extends AbstractCacheAdapter private string $dataDirectory; + private string $lockDirectory; + private string $metadataDirectory; public function __construct(string $namespace = 'default', ?string $baseDir = null) @@ -24,6 +27,58 @@ public function __construct(string $namespace = 'default', ?string $baseDir = nu $this->createDirectory($namespace, $baseDir); } + public function atomicCompareAndSet(string $key, mixed $expected, CacheItemInterface $replacement): bool + { + if (!$this->supportsItem($replacement)) { + return false; + } + $expiration = CachePayloadCodec::expirationFromItem($replacement); + if ($expiration['ttl'] !== null && $expiration['ttl'] <= 0) { + return false; + } + + return $this->withKeyLock($key, function () use ($key, $expected, $replacement): bool { + $record = $this->readLiveRecordUnlocked($key); + if (!$record instanceof CacheRecord || $record->value !== $expected) { + return false; + } + + return $this->persistItemUnlocked($replacement); + }); + } + + public function atomicGetAndDelete(string $key): CacheItemInterface + { + return $this->withKeyLock($key, function () use ($key): CacheItemInterface { + $record = $this->readLiveRecordUnlocked($key); + if (!$record instanceof CacheRecord) { + return $this->genericMiss($key); + } + $this->deleteItemUnlocked($key); + + return $this->genericItemFromRecord($key, $record); + }); + } + + public function atomicSetIfAbsent(CacheItemInterface $item): bool + { + if (!$this->supportsItem($item)) { + return false; + } + $expiration = CachePayloadCodec::expirationFromItem($item); + if ($expiration['ttl'] !== null && $expiration['ttl'] <= 0) { + return false; + } + + return $this->withKeyLock($item->getKey(), function () use ($item): bool { + if ($this->readLiveRecordUnlocked($item->getKey()) instanceof CacheRecord) { + return false; + } + + return $this->persistItemUnlocked($item); + }); + } + public function clear(): bool { $ok = true; @@ -34,7 +89,6 @@ public function clear(): bool foreach (glob($this->metadataDirectory . '*') ?: [] as $file) { $ok = (!is_file($file) || unlink($file)) && $ok; } - $this->deferred = []; return $ok; @@ -42,21 +96,15 @@ public function clear(): bool public function deleteItem(string $key): bool { - $file = $this->fileFor($key); - $this->invalidateOpcache($file); - - return !is_file($file) || unlink($file); + return $this->withKeyLock($key, fn(): bool => $this->deleteItemUnlocked($key)); } - /** - * @param array $keys The keys argument. - * @phpstan-param list $keys - */ + /** @param list $keys */ public function deleteItems(array $keys): bool { $ok = true; foreach ($keys as $key) { - $ok = $this->deleteItem((string) $key) && $ok; + $ok = $this->deleteItem($key) && $ok; } return $ok; @@ -64,27 +112,18 @@ public function deleteItems(array $keys): bool public function getItem(string $key): CacheItem { - $file = $this->fileFor($key); - if (!is_file($file)) { - return $this->genericMiss($key); - } - - $row = require $file; - $payload = is_array($row) && is_string($row['p'] ?? null) - ? $row['p'] - : null; - if (!is_string($payload)) { - return $this->genericDeleteAndMiss($key); + $record = $this->readLiveRecordUnlocked($key); + if ($record instanceof CacheRecord) { + return $this->genericItemFromRecord($key, $record); } - return $this->genericFromBase64WithInvalidator( - $key, - $payload, - fn(): bool => $this->deleteItem($key), - ); + return $this->genericMiss($key); } - /** @param list $tags */ + /** + * @param list $tags + * @return array + */ #[\Override] public function getTagGenerations(array $tags): array { @@ -110,31 +149,14 @@ public function hasItem(string $key): bool } /** - * @param array $keys The keys argument. - * @phpstan-param list $keys - * @phpstan-return array + * @param list $keys + * @return array */ public function multiFetch(array $keys): array { $items = []; - $stale = []; foreach ($keys as $key) { - $file = $this->fileFor($key); - if (!is_file($file)) { - $items[$key] = $this->genericMiss($key); - - continue; - } - $row = require $file; - $payload = is_array($row) && is_string($row['p'] ?? null) ? $row['p'] : null; - $item = $this->genericFromBase64WithInvalidator($key, $payload, static fn(): bool => true); - if (!$item->isHit()) { - $stale[] = $key; - } - $items[$key] = $item; - } - if ($stale !== []) { - $this->deleteItems($stale); + $items[$key] = $this->getItem($key); } return $items; @@ -184,33 +206,18 @@ private function createDirectory(string $ns, ?string $baseDir): void $root = $baseDir . DIRECTORY_SEPARATOR . 'cache_' . $ns . DIRECTORY_SEPARATOR; $this->dataDirectory = $root . 'data' . DIRECTORY_SEPARATOR; $this->metadataDirectory = $root . 'meta' . DIRECTORY_SEPARATOR; + $this->lockDirectory = $root . 'locks' . DIRECTORY_SEPARATOR; - $this->assertPathNotSymlink($baseDir, 'PHP cache base directory'); - $this->assertPathNotSymlink($this->dataDirectory, 'PHP cache data directory'); - $this->assertPathNotSymlink($this->metadataDirectory, 'PHP cache metadata directory'); - - if (!is_dir($baseDir) && !mkdir($baseDir, 0700, true) && !is_dir($baseDir)) { - throw new RuntimeException("Unable to create PHP cache base directory: {$baseDir}"); - } - - if (!is_dir($this->dataDirectory) - && !mkdir($this->dataDirectory, 0700, true) - && !is_dir($this->dataDirectory)) { - throw new RuntimeException("Unable to create PHP cache data directory: {$this->dataDirectory}"); - } - if (!is_dir($this->metadataDirectory) - && !mkdir($this->metadataDirectory, 0700, true) - && !is_dir($this->metadataDirectory)) { - throw new RuntimeException("Unable to create PHP cache metadata directory: {$this->metadataDirectory}"); + foreach ([$baseDir, $this->dataDirectory, $this->metadataDirectory, $this->lockDirectory] as $directory) { + $this->assertPathNotSymlink($directory, 'PHP cache directory'); + if (!is_dir($directory) && !mkdir($directory, 0700, true) && !is_dir($directory)) { + throw new RuntimeException("Unable to create PHP cache directory: {$directory}"); + } + $this->assertSecureDirectory($directory, 'PHP cache directory'); } - - $this->assertSecureDirectory($baseDir, 'PHP cache base directory'); - if (!is_writable($this->dataDirectory) || !is_writable($this->metadataDirectory)) { + if (!is_writable($this->dataDirectory) || !is_writable($this->metadataDirectory) || !is_writable($this->lockDirectory)) { throw new RuntimeException('PHP cache directories are not writable.'); } - - $this->assertSecureDirectory($this->dataDirectory, 'PHP cache data directory'); - $this->assertSecureDirectory($this->metadataDirectory, 'PHP cache metadata directory'); } private function defaultBaseDirectory(): string @@ -220,6 +227,14 @@ private function defaultBaseDirectory(): string . str_replace('/', DIRECTORY_SEPARATOR, self::DEFAULT_BASE_DIR); } + private function deleteItemUnlocked(string $key): bool + { + $file = $this->fileFor($key); + $this->invalidateOpcache($file); + + return !is_file($file) || unlink($file); + } + private function fileFor(string $key): string { return $this->dataDirectory . hash('xxh128', $key) . '.php'; @@ -239,22 +254,24 @@ private function metadataFileFor(string $tag): string private function persistItem(CacheItemInterface $item): bool { + return $this->withKeyLock($item->getKey(), fn(): bool => $this->persistItemUnlocked($item)); + } + private function persistItemUnlocked(CacheItemInterface $item): bool + { $expires = CachePayloadCodec::expirationFromItem($item); if ($expires['ttl'] !== null && $expires['ttl'] <= 0) { - return $this->deleteItem($item->getKey()); + return $this->deleteItemUnlocked($item->getKey()); } $blob = $this->encodeItem($item, $expires['expiresAt']); $payload = var_export(base64_encode($blob), true); $code = " {$payload}];\n"; - $file = $this->fileFor($item->getKey()); $tmp = tempnam($this->dataDirectory, 'pc_'); if ($tmp === false) { return false; } - if (file_put_contents($tmp, $code, LOCK_EX) === false) { if (is_file($tmp)) { unlink($tmp); @@ -262,7 +279,6 @@ private function persistItem(CacheItemInterface $item): bool return false; } - $this->invalidateOpcache($file); if (!rename($tmp, $file)) { if (is_file($tmp)) { @@ -274,4 +290,60 @@ private function persistItem(CacheItemInterface $item): bool return true; } + + private function readLiveRecordUnlocked(string $key): ?CacheRecord + { + $file = $this->fileFor($key); + if (!is_file($file)) { + return null; + } + $row = require $file; + $payload = is_array($row) && is_string($row['p'] ?? null) ? $row['p'] : null; + $blob = is_string($payload) ? base64_decode($payload, true) : false; + $record = is_string($blob) ? $this->decodeRecordFromBlob($blob) : null; + if (!$record instanceof CacheRecord || !$this->recordTagsAreCurrent($record)) { + return null; + } + + return $record; + } + + private function recordTagsAreCurrent(CacheRecord $record): bool + { + foreach ($record->tags as $tag => $generation) { + $current = is_file($this->metadataFileFor($tag)) + ? file_get_contents($this->metadataFileFor($tag)) + : false; + if (!is_string($current) || strtolower($current) !== $generation) { + return false; + } + } + + return true; + } + + /** + * @template T + * @param callable(): T $callback + * @return T + */ + private function withKeyLock(string $key, callable $callback): mixed + { + $path = $this->lockDirectory . hash('xxh128', $key) . '.lock'; + $handle = fopen($path, 'c'); + if (!is_resource($handle) || !flock($handle, LOCK_EX)) { + if (is_resource($handle)) { + fclose($handle); + } + + throw new RuntimeException('Unable to acquire PHP-file cache key lock.'); + } + + try { + return $callback(); + } finally { + flock($handle, LOCK_UN); + fclose($handle); + } + } } diff --git a/src/Cache/Adapter/WeakMapCacheAdapter.php b/src/Cache/Adapter/WeakMapCacheAdapter.php index a805f1a..def3b3e 100644 --- a/src/Cache/Adapter/WeakMapCacheAdapter.php +++ b/src/Cache/Adapter/WeakMapCacheAdapter.php @@ -10,7 +10,7 @@ use Psr\Cache\CacheItemInterface; use WeakReference; -final class WeakMapCacheAdapter extends AbstractCacheAdapter +final class WeakMapCacheAdapter extends AbstractCacheAdapter implements AtomicCachePoolInterface { private readonly string $ns; @@ -31,6 +31,58 @@ public function __construct(string $namespace = 'default') $this->ns = CacheInput::namespace($namespace); } + public function atomicCompareAndSet( + string $key, + mixed $expected, + CacheItemInterface $replacement, + ): bool { + if (!$this->supportsItem($replacement)) { + return false; + } + + $expiration = CachePayloadCodec::expirationFromItem($replacement); + if ($expiration['ttl'] !== null && $expiration['ttl'] <= 0) { + return false; + } + + $current = $this->getItem($key); + if (!$current->isHit() || $current->get() !== $expected) { + return false; + } + + return $this->persistItem($replacement, $expiration); + } + + public function atomicGetAndDelete(string $key): CacheItemInterface + { + $current = $this->getItem($key); + if (!$current->isHit()) { + return $current; + } + + $this->deleteItem($key); + + return $current; + } + + public function atomicSetIfAbsent(CacheItemInterface $item): bool + { + if (!$this->supportsItem($item)) { + return false; + } + + $expiration = CachePayloadCodec::expirationFromItem($item); + if ($expiration['ttl'] !== null && $expiration['ttl'] <= 0) { + return false; + } + + if ($this->getItem($item->getKey())->isHit()) { + return false; + } + + return $this->persistItem($item, $expiration); + } + public function clear(): bool { $this->scalarStore = []; diff --git a/src/Cache/AtomicCache.php b/src/Cache/AtomicCache.php index 762ac03..807d3aa 100644 --- a/src/Cache/AtomicCache.php +++ b/src/Cache/AtomicCache.php @@ -7,6 +7,7 @@ use DateInterval; use DateTimeInterface; use Infocyph\CacheLayer\Cache\Adapter\AtomicCachePoolInterface; +use Infocyph\CacheLayer\Cache\Adapter\ConditionalAtomicCachePoolInterface; use Infocyph\CacheLayer\Cache\Adapter\InternalCachePoolInterface; use Infocyph\CacheLayer\Cache\Metrics\CacheMetricsCollectorInterface; use Infocyph\CacheLayer\Exceptions\CacheBackendException; @@ -30,6 +31,9 @@ public static function fromAdapter( if (!$adapter instanceof AtomicCachePoolInterface) { return null; } + if ($adapter instanceof ConditionalAtomicCachePoolInterface && !$adapter->supportsAtomicCache()) { + return null; + } return new self($adapter, $options, $metrics); } diff --git a/tests/Cache/AtomicBackendExpansionTest.php b/tests/Cache/AtomicBackendExpansionTest.php new file mode 100644 index 0000000..bbd41a3 --- /dev/null +++ b/tests/Cache/AtomicBackendExpansionTest.php @@ -0,0 +1,99 @@ +isDir() ? rmdir($entry->getPathname()) : unlink($entry->getPathname()); + } + rmdir($directory); +}; + +test('WeakMap supports the full atomic cache contract', function () { + $cache = Cache::weakMap('atomic-weak-map-contract'); + $atomic = $cache->atomic(); + + expect($atomic)->toBeInstanceOf(AtomicCacheInterface::class) + ->and($atomic->setIfAbsent('claim', 1, 30))->toBeTrue() + ->and($atomic->setIfAbsent('claim', 2, 30))->toBeFalse() + ->and($atomic->compareAndSet('claim', '1', 2, 30))->toBeFalse() + ->and($atomic->compareAndSet('claim', 1, 2, 30))->toBeTrue() + ->and($atomic->getAndDelete('claim', 'missing'))->toBe(2) + ->and($atomic->getAndDelete('claim', 'missing'))->toBe('missing'); +}); + +test('file and PHP-file stores support the full atomic cache contract', function () use ($cleanupTree) { + foreach (['file', 'phpFiles'] as $factory) { + $directory = sys_get_temp_dir() . '/cachelayer-atomic-' . strtolower($factory) . '-' . uniqid('', true); + try { + $cache = $factory === 'file' + ? Cache::file('atomic-files', $directory) + : Cache::phpFiles('atomic-files', $directory); + $atomic = $cache->atomic(); + + expect($atomic)->toBeInstanceOf(AtomicCacheInterface::class) + ->and($atomic->setIfAbsent('claim', 'first', 30))->toBeTrue() + ->and($atomic->setIfAbsent('claim', 'second', 30))->toBeFalse() + ->and($atomic->compareAndSet('claim', 'first', 'updated', 30))->toBeTrue() + ->and($cache->get('claim'))->toBe('updated') + ->and($atomic->getAndDelete('claim', 'missing'))->toBe('updated') + ->and($cache->has('claim'))->toBeFalse(); + } finally { + $cleanupTree($directory); + } + } +}); + +test('SQLite PDO supports the full atomic cache contract', function () { + $file = sys_get_temp_dir() . '/cachelayer-atomic-pdo-' . uniqid('', true) . '.sqlite'; + try { + $cache = Cache::sqlite('atomic-pdo', $file); + $atomic = $cache->atomic(); + + expect($atomic)->toBeInstanceOf(AtomicCacheInterface::class) + ->and($atomic->setIfAbsent('claim', null, 30))->toBeTrue() + ->and($atomic->compareAndSet('claim', null, 'updated', 30))->toBeTrue() + ->and($atomic->getAndDelete('claim', 'missing'))->toBe('updated') + ->and($atomic->getAndDelete('claim', 'missing'))->toBe('missing'); + } finally { + if (is_file($file)) { + unlink($file); + } + } +}); + +if (class_exists(Memcached::class)) { + $host = getenv('IC_MEMCACHED_HOST') ?: getenv('CACHELAYER_MEMCACHED_HOST') ?: '127.0.0.1'; + $port = (int) (getenv('IC_MEMCACHED_PORT') ?: getenv('CACHELAYER_MEMCACHED_PORT') ?: '11211'); + $probe = new Memcached(); + $probe->addServer($host, $port); + $available = $probe->set('cachelayer-atomic-probe', 'ok') + && $probe->getResultCode() === Memcached::RES_SUCCESS; + + test('Memcached supports CAS-backed atomic cache operations', function () use ($host, $port) { + $client = new Memcached(); + $client->addServer($host, $port); + $client->flush(); + $cache = Cache::memcached('atomic-memcached', [[$host, $port, 0]], $client); + $atomic = $cache->atomic(); + + expect($atomic)->toBeInstanceOf(AtomicCacheInterface::class) + ->and($atomic->setIfAbsent('claim', 'first', 30))->toBeTrue() + ->and($atomic->setIfAbsent('claim', 'second', 30))->toBeFalse() + ->and($atomic->compareAndSet('claim', 'first', 'updated', 30))->toBeTrue() + ->and($atomic->getAndDelete('claim', 'missing'))->toBe('updated') + ->and($cache->has('claim'))->toBeFalse() + ->and($atomic->setIfAbsent('claim', 'reclaimed', 30))->toBeTrue() + ->and($cache->get('claim'))->toBe('reclaimed'); + })->skip(!$available, 'No Memcached server available.'); +} diff --git a/tests/Cache/AtomicCacheCapabilityTest.php b/tests/Cache/AtomicCacheCapabilityTest.php index 9b0ef77..3d35e00 100644 --- a/tests/Cache/AtomicCacheCapabilityTest.php +++ b/tests/Cache/AtomicCacheCapabilityTest.php @@ -12,24 +12,21 @@ test('cache exposes atomic capability only when the adapter guarantees it', function () { $memory = Cache::memory('atomic-memory'); + $weakMap = Cache::weakMap('atomic-weak-map'); $null = Cache::nullStore(); expect($memory)->toBeInstanceOf(AtomicCacheProviderInterface::class) ->and($memory->atomic())->toBeInstanceOf(AtomicCacheInterface::class) + ->and($weakMap->atomic())->toBeInstanceOf(AtomicCacheInterface::class) ->and($null->atomic())->toBeNull(); }); -test('unsupported and composite stores do not advertise atomic capability', function () { +test('supported SQLite PDO store advertises atomic capability', function () { $sqliteFile = sys_get_temp_dir() . '/cachelayer-atomic-' . uniqid('', true) . '.sqlite'; $sqlite = Cache::sqlite('atomic-sqlite', $sqliteFile); - $tiered = Cache::tiered([ - new ArrayCacheAdapter('atomic-tier-l1'), - new ArrayCacheAdapter('atomic-tier-l2'), - ], namespace: 'atomic-tiered'); try { - expect($sqlite->atomic())->toBeNull() - ->and($tiered->atomic())->toBeNull(); + expect($sqlite->atomic())->toBeInstanceOf(AtomicCacheInterface::class); } finally { unset($sqlite); if (is_file($sqliteFile)) { @@ -38,6 +35,15 @@ } }); +test('composite stores do not advertise atomic capability', function () { + $tiered = Cache::tiered([ + new ArrayCacheAdapter('atomic-tier-l1'), + new ArrayCacheAdapter('atomic-tier-l2'), + ], namespace: 'atomic-tiered'); + + expect($tiered->atomic())->toBeNull(); +}); + test('set if absent has one-winner semantics and preserves existing values', function () { $cache = Cache::memory('atomic-set-if-absent'); $atomic = $cache->atomic();