From 14a5f2a169672232068c64c90099a4a00ec8fa2b Mon Sep 17 00:00:00 2001 From: "A. B. M. Mahmudul Hasan" Date: Wed, 15 Jul 2026 10:38:52 +0600 Subject: [PATCH] docs_atom --- docs/cluster/_content.inc | 96 +++++++++ docs/counters.rst | 51 +++++ docs/index.rst | 1 + docs/node/_content.inc | 16 ++ src/Cluster/ClusterCache.php | 17 +- src/Cluster/ClusterRuntime.php | 89 +++++++- src/Cluster/Consumer/InvalidationConsumer.php | 28 ++- .../Coordinator/ClusterCoordinator.php | 17 ++ src/Cluster/Cursor/CursorStoreInterface.php | 2 + src/Cluster/Cursor/SqliteCursorStore.php | 18 ++ src/Cluster/Health/ClusterStatus.php | 22 ++ src/Cluster/Health/ClusterStatusTracker.php | 61 ++++++ src/Cluster/Outbox/ClusterOutbox.php | 91 ++++++++ ...nvalidationTransportInspectorInterface.php | 12 ++ .../Pdo/PdoInvalidationTransport.php | 54 ++++- .../RedisStreamInvalidationTransport.php | 196 ++++++++++++++++++ ...actionalInvalidationTransportInterface.php | 13 ++ src/Counter/AtomicCounterStoreInterface.php | 16 ++ src/Counter/AtomicCounterValue.php | 13 ++ src/Counter/AtomicCounters.php | 51 +++++ .../Exception/AtomicCounterException.php | 9 + src/Counter/RedisAtomicCounterStore.php | 102 +++++++++ src/Node/NodeCache.php | 2 +- src/Node/NodeCacheConfig.php | 2 + tests/Cluster/ClusterCacheTest.php | 82 +++++++- .../Support/InMemoryInvalidationTransport.php | 16 +- tests/Node/NodeCacheTest.php | 40 ++++ 27 files changed, 1100 insertions(+), 17 deletions(-) create mode 100644 docs/counters.rst create mode 100644 src/Cluster/Health/ClusterStatus.php create mode 100644 src/Cluster/Health/ClusterStatusTracker.php create mode 100644 src/Cluster/Outbox/ClusterOutbox.php create mode 100644 src/Cluster/Transport/InvalidationTransportInspectorInterface.php create mode 100644 src/Cluster/Transport/RedisStreamInvalidationTransport.php create mode 100644 src/Cluster/Transport/TransactionalInvalidationTransportInterface.php create mode 100644 src/Counter/AtomicCounterStoreInterface.php create mode 100644 src/Counter/AtomicCounterValue.php create mode 100644 src/Counter/AtomicCounters.php create mode 100644 src/Counter/Exception/AtomicCounterException.php create mode 100644 src/Counter/RedisAtomicCounterStore.php diff --git a/docs/cluster/_content.inc b/docs/cluster/_content.inc index 4f6fada..ff6ec54 100644 --- a/docs/cluster/_content.inc +++ b/docs/cluster/_content.inc @@ -145,6 +145,36 @@ DDL and read/write rights when the transport is initialized. SQLite support in this class is useful for a local test only. Do **not** use a SQLite database as a shared multi-node event transport in production. +It is rejected by default and requires an explicit test-only opt-in: + +.. code-block:: php + + $transport = new PdoInvalidationTransport($sqliteConnection, allowSqliteForTesting: true); + +Redis and Valkey Streams transport +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +``RedisStreamInvalidationTransport`` is included for lower-latency, +higher-volume invalidation workloads. It appends events to one durable stream +per cluster and each node replays from its own persisted stream cursor; it does +not use a consumer group because every node must receive every invalidation. + +.. code-block:: php + + use Infocyph\CacheLayer\Cluster\Transport\RedisStreamInvalidationTransport; + + $transport = new RedisStreamInvalidationTransport( + $redis, + prefix: 'application:cache-invalidation:', + maxLength: 100_000, + ); + +``maxLength`` bounds retained stream entries using Redis's approximate trimming. +Set it large enough to cover the greatest expected outage and replay backlog; +a node that falls behind the retained stream is safely recovered by clearing +its local namespace. Redis/Valkey persistence and replication settings are an +application operational requirement—the stream must be durable enough for the +invalidation guarantee you need. .. cluster-topology-start @@ -186,8 +216,10 @@ Method Meaning ``invalidateKey(string $key)`` Invalidate one key locally and publish an event. ``invalidateTag(string $tag)`` Invalidate one tag locally and publish an event. ``clearNamespace()`` Clear the local namespace and publish an event. +``invalidateTags(array $tags)`` Publish one invalidation for each unique tag. ``consume(?int $limit)`` Replay up to ``limit`` events and return the count read. ``recoverIfRequired(): bool`` Perform retention-gap recovery; normally called by consume. +``status(): ClusterStatus`` Report cursor, transport, consume, and recovery health. =============================== ===================================================== The critical boundary is ``$runtime->cache()``: all ordinary ``get()``, @@ -403,6 +435,13 @@ minutes, according to the maximum acceptable stale window: // bin/cache-consume.php: bootstrap $runtime for this node first. $processed = $runtime->consume(); +For a worker that should catch up without running forever, use ``drain()``. +It reads until it receives a partial batch or reaches ``maxBatches``: + +.. code-block:: php + + $processed = $runtime->drain(limit: 1_000, maxBatches: 50); + The schedule determines remote invalidation latency. For example, a task every five seconds means a healthy remote node may serve stale data for nearly five seconds plus normal request timing. TTL remains the final safety bound. @@ -451,6 +490,30 @@ sub-minute propagation, use a framework scheduler that supports it or a supervised worker. Ensure the scheduler prevents overlapping runs on the same node, or make each invocation deliberately short and bounded. +Cluster health status +~~~~~~~~~~~~~~~~~~~~~ + +Use ``status()`` in a health endpoint or scheduled diagnostic. It returns a +``ClusterStatus`` value with the local cursor and update time, oldest/newest +transport event IDs, pending-event count when the transport can calculate it, +and the most recent consume count, error, and recovery time. + +.. code-block:: php + + $status = $runtime->status(); + + if ($status->lastConsumeError !== null) { + reportCacheConsumerFailure($status->lastConsumeError); + } + + if ($status->pendingEventCount !== null && $status->pendingEventCount > 10_000) { + alertOnConsumerLag($status); + } + +The PDO transport implements detailed inspection. A custom transport that does +not implement the optional inspection contract reports ``null`` for newest ID +and pending count rather than inventing a lag estimate. + Continuous worker with bounded drain ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ @@ -596,6 +659,39 @@ are skipped by that node's consumer, so the explicit local delete after a successful commit is required. Treat a failure of that delete as operationally important and retry/alert; a bounded TTL still limits the stale window. +``PdoInvalidationTransport`` implements +``TransactionalInvalidationTransportInterface``. Framework/database +integrations should depend on that interface rather than the concrete PDO class +when they need to attach an invalidation event to a database transaction. + +CacheLayer also provides a framework-neutral ``ClusterOutbox`` helper through +``$runtime->outbox($pdo)``. Create it within the source transaction, publish +its invalidations alongside the source write, commit, then apply its local +invalidation after the commit succeeds: + +.. code-block:: php + + $pdo->beginTransaction(); + try { + updateProduct($pdo, 42, $input); + $outbox = $runtime->outbox($pdo); + $outbox->invalidateKey('product.42'); + $pdo->commit(); + $outbox->applyLocally(); + } catch (Throwable $exception) { + if ($pdo->inTransaction()) { + $pdo->rollBack(); + } + + throw $exception; + } + +The outbox event is deliberately replayed by its origin node as well. Therefore +if the process fails after commit but before ``applyLocally()``, the next local +consumer run still invalidates the origin cache safely. A framework can wrap +the final local application in its after-commit callback; transaction ownership +and callback registration remain application concerns. + Operations checklist -------------------- diff --git a/docs/counters.rst b/docs/counters.rst new file mode 100644 index 0000000..a3b6e8f --- /dev/null +++ b/docs/counters.rst @@ -0,0 +1,51 @@ +=============== +Atomic Counters +=============== + +Atomic counters are separate from the normal ``Cache`` facade. They are for +shared integer state where ``get()`` followed by ``set()`` would race across +PHP workers or application nodes: rate-limit windows, authentication lockout +attempts, quotas, and replay-attempt counts. + +Use Redis or Valkey as the shared backend. Node Cache, APCu, and SQLite are not +distributed atomic-counter stores. + +.. code-block:: php + + use Infocyph\CacheLayer\Counter\AtomicCounters; + + $counters = AtomicCounters::redis('authentication'); + $attempt = $counters->increment('login.203.0.113.10', ttlSeconds: 900); + + if ($attempt->value > 5) { + denyLogin(); + } + +``increment()`` and ``decrement()`` execute as one Redis/Valkey operation. +When a positive TTL is supplied, it is assigned only when that key is first +created; later increments do not extend the fixed window. Each operation +returns ``AtomicCounterValue`` with the resulting ``value`` and an +``initialized`` flag. + +.. code-block:: php + + $first = $counters->increment('quota.42', 1, 3600); + // $first->value === 1; $first->initialized === true + + $later = $counters->increment('quota.42', 3, 3600); + // $later->value === 4; $later->initialized === false; TTL is unchanged. + +Use ``get()`` for a read, ``delete()`` to remove a counter, and +``decrement()`` for an atomic decrement. Counter keys allow letters, digits, +``_``, ``.``, and ``-``. A provided TTL must be a positive integer. + +Valkey uses the same Redis-compatible client contract: + +.. code-block:: php + + $counters = AtomicCounters::valkey('rate-limits', 'valkey://127.0.0.1:6379'); + +Counters are not a general replacement for authoritative accounting. For +payment, entitlement, or security decisions requiring durable auditability, +keep the authoritative record in a transactional data store and use counters +only for the bounded coordination/window concern. diff --git a/docs/index.rst b/docs/index.rst index 06a8c8b..89356f5 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -48,6 +48,7 @@ Quick Start :caption: Guide cache + counters adapters/index cookbook metrics-and-locking diff --git a/docs/node/_content.inc b/docs/node/_content.inc index cd537e6..65e0f1d 100644 --- a/docs/node/_content.inc +++ b/docs/node/_content.inc @@ -164,6 +164,22 @@ Configuration reference cache degradation where possible. When false, APCu/SQLite failures propagate to the caller. +``lockProvider`` + Optional ``LockProviderInterface`` used by ``remember()``. When omitted, + Node Cache uses its local file-lock default. Supply a shared + ``RedisLockProvider`` or ``ValkeyLockProvider`` when multiple application + nodes must coordinate one ``remember()`` resolver. + +.. code-block:: php + + use Infocyph\CacheLayer\Cache\Lock\RedisLockProvider; + + $config = new NodeCacheConfig( + sqliteFile: '/var/cache/my-application/node-cache.sqlite', + namespace: 'catalog', + lockProvider: new RedisLockProvider($redis, 'catalog:remember:'), + ); + .. node-operations-start Normal cache operations diff --git a/src/Cluster/ClusterCache.php b/src/Cluster/ClusterCache.php index e882a18..25ac6ab 100644 --- a/src/Cluster/ClusterCache.php +++ b/src/Cluster/ClusterCache.php @@ -8,6 +8,7 @@ use Infocyph\CacheLayer\Cluster\Consumer\InvalidationHandler; use Infocyph\CacheLayer\Cluster\Coordinator\ClusterCoordinator; use Infocyph\CacheLayer\Cluster\Cursor\SqliteCursorStore; +use Infocyph\CacheLayer\Cluster\Health\ClusterStatusTracker; use Infocyph\CacheLayer\Cluster\Recovery\ClusterRecoveryManager; use Infocyph\CacheLayer\Cluster\Transport\InvalidationTransportInterface; use Infocyph\CacheLayer\Node\NodeCache; @@ -22,6 +23,7 @@ public static function create( ): ClusterRuntime { $cache = NodeCache::create($node); $cursorStore = new SqliteCursorStore($node->sqliteFile, $cluster->cluster, $cluster->nodeId); + $status = new ClusterStatusTracker(); $recovery = new ClusterRecoveryManager($cache, $cursorStore, $transport, $cluster->cluster); $coordinator = new ClusterCoordinator( $cache, @@ -38,8 +40,21 @@ public static function create( $recovery, $cluster->cluster, $cluster->nodeId, + $status, ); - return new ClusterRuntime($cache, $coordinator, $consumer, $recovery, $cluster->consumerBatchSize); + return new ClusterRuntime( + $cache, + $coordinator, + $consumer, + $recovery, + $cluster->consumerBatchSize, + $cursorStore, + $transport, + $status, + $cluster->cluster, + $cluster->nodeId, + $node->namespace, + ); } } diff --git a/src/Cluster/ClusterRuntime.php b/src/Cluster/ClusterRuntime.php index 00f361c..847546e 100644 --- a/src/Cluster/ClusterRuntime.php +++ b/src/Cluster/ClusterRuntime.php @@ -6,8 +6,18 @@ use Infocyph\CacheLayer\Cache\Cache; use Infocyph\CacheLayer\Cluster\Consumer\InvalidationConsumer; +use Infocyph\CacheLayer\Cluster\Consumer\InvalidationHandler; use Infocyph\CacheLayer\Cluster\Coordinator\ClusterCoordinator; +use Infocyph\CacheLayer\Cluster\Cursor\CursorStoreInterface; +use Infocyph\CacheLayer\Cluster\Exception\ClusterCacheException; +use Infocyph\CacheLayer\Cluster\Health\ClusterStatus; +use Infocyph\CacheLayer\Cluster\Health\ClusterStatusTracker; +use Infocyph\CacheLayer\Cluster\Outbox\ClusterOutbox; use Infocyph\CacheLayer\Cluster\Recovery\ClusterRecoveryManager; +use Infocyph\CacheLayer\Cluster\Transport\InvalidationTransportInspectorInterface; +use Infocyph\CacheLayer\Cluster\Transport\InvalidationTransportInterface; +use Infocyph\CacheLayer\Cluster\Transport\TransactionalInvalidationTransportInterface; +use PDO; final readonly class ClusterRuntime { @@ -17,6 +27,12 @@ public function __construct( private InvalidationConsumer $consumer, private ClusterRecoveryManager $recovery, private int $consumerBatchSize, + private CursorStoreInterface $cursorStore, + private InvalidationTransportInterface $transport, + private ClusterStatusTracker $status, + private string $cluster, + private string $nodeId, + private string $namespace, ) {} public function cache(): Cache @@ -34,6 +50,25 @@ public function consume(?int $limit = null): int return $this->consumer->consume($limit ?? $this->consumerBatchSize); } + public function drain(?int $limit = null, int $maxBatches = 100): int + { + $limit ??= $this->consumerBatchSize; + if ($limit < 1 || $maxBatches < 1) { + throw new ClusterCacheException('Cluster drain limit and maximum batches must be greater than zero.'); + } + + $total = 0; + for ($batch = 0; $batch < $maxBatches; ++$batch) { + $processed = $this->consumer->consume($limit); + $total += $processed; + if ($processed < $limit) { + break; + } + } + + return $total; + } + public function invalidateKey(string $key): void { $this->coordinator->invalidateKey($key); @@ -44,8 +79,60 @@ public function invalidateTag(string $tag): void $this->coordinator->invalidateTag($tag); } + /** + * @param array $tags The tags argument. + * @phpstan-param list $tags + */ + public function invalidateTags(array $tags): void + { + $this->coordinator->invalidateTags($tags); + } + + public function outbox(PDO $connection): ClusterOutbox + { + if (!$this->transport instanceof TransactionalInvalidationTransportInterface) { + throw new ClusterCacheException('The configured Cluster Cache transport does not support transactional invalidation.'); + } + + return new ClusterOutbox( + $connection, + $this->transport, + new InvalidationHandler($this->cache, $this->namespace), + $this->cluster, + $this->namespace, + $this->nodeId, + ); + } + public function recoverIfRequired(): bool { - return $this->recovery->recoverIfRequired(); + $recovered = $this->recovery->recoverIfRequired(); + if ($recovered) { + $this->status->recordRecovery(); + } + + return $recovered; + } + + public function status(): ClusterStatus + { + $cursor = $this->cursorStore->current(); + $oldest = $this->transport->oldestAvailableId($this->cluster); + $newest = null; + $pending = null; + if ($this->transport instanceof InvalidationTransportInspectorInterface) { + $newest = $this->transport->newestAvailableId($this->cluster); + $pending = $this->transport->countAfter($this->cluster, $cursor); + } + + return $this->status->snapshot( + $this->cluster, + $this->nodeId, + $cursor, + $this->cursorStore->updatedAt(), + $oldest, + $newest, + $pending, + ); } } diff --git a/src/Cluster/Consumer/InvalidationConsumer.php b/src/Cluster/Consumer/InvalidationConsumer.php index 6318652..9214831 100644 --- a/src/Cluster/Consumer/InvalidationConsumer.php +++ b/src/Cluster/Consumer/InvalidationConsumer.php @@ -5,8 +5,10 @@ namespace Infocyph\CacheLayer\Cluster\Consumer; use Infocyph\CacheLayer\Cluster\Cursor\CursorStoreInterface; +use Infocyph\CacheLayer\Cluster\Health\ClusterStatusTracker; use Infocyph\CacheLayer\Cluster\Recovery\ClusterRecoveryManager; use Infocyph\CacheLayer\Cluster\Transport\InvalidationTransportInterface; +use Throwable; final readonly class InvalidationConsumer { @@ -17,6 +19,7 @@ public function __construct( private ClusterRecoveryManager $recovery, private string $cluster, private string $nodeId, + private ClusterStatusTracker $status, ) {} public function consume(int $limit = 1_000): int @@ -25,17 +28,26 @@ public function consume(int $limit = 1_000): int return 0; } - $this->recovery->recoverIfRequired(); - $batch = $this->transport->consumeAfter($this->cluster, $this->cursorStore->current(), $limit); + try { + $recovered = $this->recovery->recoverIfRequired(); + $batch = $this->transport->consumeAfter($this->cluster, $this->cursorStore->current(), $limit); - foreach ($batch->events as $event) { - if ($event->originNodeId !== $this->nodeId) { - $this->handler->handle($event); + foreach ($batch->events as $event) { + if ($event->originNodeId !== $this->nodeId) { + $this->handler->handle($event); + } + + $this->cursorStore->advance((string) $event->id); } - $this->cursorStore->advance((string) $event->id); - } + $count = count($batch->events); + $this->status->recordConsume($count, $recovered); - return count($batch->events); + return $count; + } catch (Throwable $exception) { + $this->status->recordFailure($exception); + + throw $exception; + } } } diff --git a/src/Cluster/Coordinator/ClusterCoordinator.php b/src/Cluster/Coordinator/ClusterCoordinator.php index 9edf836..9198bb4 100644 --- a/src/Cluster/Coordinator/ClusterCoordinator.php +++ b/src/Cluster/Coordinator/ClusterCoordinator.php @@ -38,6 +38,23 @@ public function invalidateTag(string $tag): void $this->coordinate($event, fn(): bool => $this->cache->invalidateTag($tag)); } + /** + * @param array $tags The tags argument. + * @phpstan-param list $tags + */ + public function invalidateTags(array $tags): void + { + $seen = []; + foreach ($tags as $tag) { + if (isset($seen[$tag])) { + continue; + } + + $seen[$tag] = true; + $this->invalidateTag($tag); + } + } + private function coordinate(InvalidationEvent $event, callable $invalidate): void { if ($this->invalidateLocallyFirst) { diff --git a/src/Cluster/Cursor/CursorStoreInterface.php b/src/Cluster/Cursor/CursorStoreInterface.php index f60c2aa..34056ca 100644 --- a/src/Cluster/Cursor/CursorStoreInterface.php +++ b/src/Cluster/Cursor/CursorStoreInterface.php @@ -11,4 +11,6 @@ public function advance(string $eventId): void; public function current(): ?string; public function reset(?string $eventId): void; + + public function updatedAt(): ?int; } diff --git a/src/Cluster/Cursor/SqliteCursorStore.php b/src/Cluster/Cursor/SqliteCursorStore.php index 322b995..ff596ea 100644 --- a/src/Cluster/Cursor/SqliteCursorStore.php +++ b/src/Cluster/Cursor/SqliteCursorStore.php @@ -56,6 +56,24 @@ public function reset(?string $eventId): void $this->write($eventId); } + public function updatedAt(): ?int + { + try { + $statement = $this->connection->prepare( + 'SELECT updated_at FROM cachelayer_cluster_cursors ' + . 'WHERE cluster_name = :cluster AND node_id = :node_id LIMIT 1', + ); + $statement->execute([':cluster' => $this->cluster, ':node_id' => $this->nodeId]); + $updatedAt = $statement->fetchColumn(); + } catch (PDOException $exception) { + throw new ClusterCacheException('Unable to read the cluster cursor update time.', 0, $exception); + } + + return is_int($updatedAt) || (is_string($updatedAt) && ctype_digit($updatedAt)) + ? (int) $updatedAt + : null; + } + private function createSchemaIfMissing(): void { try { diff --git a/src/Cluster/Health/ClusterStatus.php b/src/Cluster/Health/ClusterStatus.php new file mode 100644 index 0000000..c41792f --- /dev/null +++ b/src/Cluster/Health/ClusterStatus.php @@ -0,0 +1,22 @@ +lastConsumedAt = time(); + $this->lastConsumeCount = $count; + $this->lastConsumeError = null; + if ($recovered) { + $this->recordRecovery(); + } + } + + public function recordFailure(\Throwable $exception): void + { + $this->lastConsumedAt = time(); + $this->lastConsumeError = $exception->getMessage(); + } + + public function recordRecovery(): void + { + $this->lastRecoveryAt = time(); + } + + public function snapshot( + string $cluster, + string $nodeId, + ?string $cursor, + ?int $cursorUpdatedAt, + ?string $oldestAvailableEventId, + ?string $newestAvailableEventId, + ?int $pendingEventCount, + ): ClusterStatus { + return new ClusterStatus( + $cluster, + $nodeId, + $cursor, + $cursorUpdatedAt, + $oldestAvailableEventId, + $newestAvailableEventId, + $pendingEventCount, + $this->lastConsumedAt, + $this->lastConsumeCount, + $this->lastConsumeError, + $this->lastRecoveryAt, + ); + } +} diff --git a/src/Cluster/Outbox/ClusterOutbox.php b/src/Cluster/Outbox/ClusterOutbox.php new file mode 100644 index 0000000..7494722 --- /dev/null +++ b/src/Cluster/Outbox/ClusterOutbox.php @@ -0,0 +1,91 @@ + */ + private array $events = []; + + private bool $localInvalidationApplied = false; + + public function __construct( + private readonly PDO $connection, + private readonly TransactionalInvalidationTransportInterface $transport, + private readonly InvalidationHandler $handler, + private readonly string $cluster, + private readonly string $namespace, + private readonly string $nodeId, + ) { + if (!$this->connection->inTransaction()) { + throw new ClusterCacheException('Cluster outbox creation requires an active database transaction.'); + } + } + + public function applyLocally(): void + { + if ($this->connection->inTransaction()) { + throw new ClusterCacheException('Apply Cluster outbox local invalidation only after the transaction commits.'); + } + if ($this->localInvalidationApplied) { + return; + } + + foreach ($this->events as $event) { + $this->handler->handle($event); + } + + $this->localInvalidationApplied = true; + } + + public function clearNamespace(): void + { + $this->publish(InvalidationEvent::namespace($this->cluster, $this->namespace, $this->outboxOrigin())); + } + + public function invalidateKey(string $key): void + { + $this->publish(InvalidationEvent::key($this->cluster, $this->namespace, $key, $this->outboxOrigin())); + } + + public function invalidateTag(string $tag): void + { + $this->publish(InvalidationEvent::tag($this->cluster, $this->namespace, $tag, $this->outboxOrigin())); + } + + /** + * @param array $tags The tags argument. + * @phpstan-param list $tags + */ + public function invalidateTags(array $tags): void + { + $seen = []; + foreach ($tags as $tag) { + if (isset($seen[$tag])) { + continue; + } + + $seen[$tag] = true; + $this->invalidateTag($tag); + } + } + + private function outboxOrigin(): string + { + return $this->nodeId . ':outbox'; + } + + private function publish(InvalidationEvent $event): void + { + $this->transport->publishWithinTransaction($this->connection, $event); + $this->events[] = $event; + } +} diff --git a/src/Cluster/Transport/InvalidationTransportInspectorInterface.php b/src/Cluster/Transport/InvalidationTransportInspectorInterface.php new file mode 100644 index 0000000..f46285a --- /dev/null +++ b/src/Cluster/Transport/InvalidationTransportInspectorInterface.php @@ -0,0 +1,12 @@ +connection->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION); $driver = $connection->getAttribute(PDO::ATTR_DRIVER_NAME); $this->driver = is_string($driver) ? $driver : ''; + if (!in_array($this->driver, ['mysql', 'pgsql', 'sqlite'], true)) { + throw new ClusterTransportException('PDO invalidation transport supports MySQL and PostgreSQL only.'); + } + if ($this->driver === 'sqlite' && !$allowSqliteForTesting) { + throw new ClusterTransportException('SQLite is not a supported shared Cluster Cache transport.'); + } $this->createSchemaIfMissing(); } @@ -55,6 +62,27 @@ public function consumeAfter(string $cluster, ?string $cursor, int $limit): Inva return new InvalidationBatch($events); } + public function countAfter(string $cluster, ?string $cursor): int + { + try { + $sql = 'SELECT COUNT(*) FROM ' . self::TABLE . ' WHERE cluster_name = :cluster'; + if ($cursor !== null) { + $sql .= ' AND event_id > :cursor'; + } + $statement = $this->connection->prepare($sql); + $statement->bindValue(':cluster', $cluster, PDO::PARAM_STR); + if ($cursor !== null) { + $statement->bindValue(':cursor', $this->eventId($cursor), PDO::PARAM_STR); + } + $statement->execute(); + $count = $statement->fetchColumn(); + } catch (PDOException $exception) { + throw new ClusterTransportException('Unable to count pending invalidation events.', 0, $exception); + } + + return is_numeric($count) ? (int) $count : 0; + } + public function isCursorBefore(string $cursor, string $oldestAvailableId): bool { $cursor = $this->eventId($cursor); @@ -65,6 +93,11 @@ public function isCursorBefore(string $cursor, string $oldestAvailableId): bool : strlen($cursor) < strlen($oldestAvailableId); } + public function newestAvailableId(string $cluster): ?string + { + return $this->eventBoundary($cluster, 'MAX'); + } + public function oldestAvailableId(string $cluster): ?string { try { @@ -166,6 +199,21 @@ private function createTableSql(): string . 'created_at BIGINT NOT NULL)'; } + private function eventBoundary(string $cluster, string $aggregate): ?string + { + try { + $statement = $this->connection->prepare( + 'SELECT ' . $aggregate . '(event_id) FROM ' . self::TABLE . ' WHERE cluster_name = :cluster', + ); + $statement->execute([':cluster' => $cluster]); + $id = $statement->fetchColumn(); + } catch (PDOException $exception) { + throw new ClusterTransportException('Unable to read invalidation event boundary.', 0, $exception); + } + + return is_int($id) || (is_string($id) && ctype_digit($id)) ? (string) $id : null; + } + /** * @param array $row The row argument. * @phpstan-param array $row diff --git a/src/Cluster/Transport/RedisStreamInvalidationTransport.php b/src/Cluster/Transport/RedisStreamInvalidationTransport.php new file mode 100644 index 0000000..ac9272e --- /dev/null +++ b/src/Cluster/Transport/RedisStreamInvalidationTransport.php @@ -0,0 +1,196 @@ +client->xRead([$this->stream($cluster) => $cursor ?? '0-0'], $limit); + } catch (\RedisException $exception) { + throw new ClusterTransportException('Unable to consume Redis Stream invalidation events.', 0, $exception); + } + + if (!is_array($streams)) { + return new InvalidationBatch([]); + } + + $entries = $streams[$this->stream($cluster)] ?? null; + if (!is_array($entries)) { + return new InvalidationBatch([]); + } + + $events = []; + foreach ($entries as $id => $fields) { + if (!is_string($id) || !is_array($fields)) { + continue; + } + $events[] = $this->eventFromFields($id, $fields); + } + + return new InvalidationBatch($events); + } + + public function isCursorBefore(string $cursor, string $oldestAvailableId): bool + { + return $this->compareIds($cursor, $oldestAvailableId) < 0; + } + + public function oldestAvailableId(string $cluster): ?string + { + return $this->boundary($cluster, false); + } + + public function publish(InvalidationEvent $event): string + { + try { + $id = $this->client->xAdd( + $this->stream($event->cluster), + '*', + [ + 'cluster' => $event->cluster, + 'namespace' => $event->namespace, + 'type' => $event->type->value, + 'identifier' => $event->identifier ?? '', + 'has_identifier' => $event->identifier === null ? '0' : '1', + 'origin' => $event->originNodeId, + 'created_at' => (string) $event->createdAt, + ], + $this->maxLength, + true, + ); + } catch (\RedisException $exception) { + throw new ClusterTransportException('Unable to publish Redis Stream invalidation event.', 0, $exception); + } + + if (!is_string($id) || $id === '') { + throw new ClusterTransportException('Redis Stream transport did not return an event ID.'); + } + + return $id; + } + + private function boundary(string $cluster, bool $newest): ?string + { + try { + $entries = $newest + ? $this->client->xRevRange($this->stream($cluster), '+', '-', 1) + : $this->client->xRange($this->stream($cluster), '-', '+', 1); + } catch (\RedisException $exception) { + throw new ClusterTransportException('Unable to read Redis Stream invalidation boundary.', 0, $exception); + } + + if (!is_array($entries) || $entries === []) { + return null; + } + + $id = array_key_first($entries); + + return is_string($id) ? $id : null; + } + + private function clusterFromStreamEvent(mixed $fields): string + { + return $this->field($fields, 'cluster'); + } + + private function compareIds(string $left, string $right): int + { + [$leftMilliseconds, $leftSequence] = $this->idParts($left); + [$rightMilliseconds, $rightSequence] = $this->idParts($right); + + return $leftMilliseconds === $rightMilliseconds + ? $leftSequence <=> $rightSequence + : $leftMilliseconds <=> $rightMilliseconds; + } + + /** + * @param string $id The ID argument. + * @param array $fields The fields argument. + * @phpstan-param array $fields + */ + private function eventFromFields(string $id, array $fields): InvalidationEvent + { + $type = InvalidationEventType::tryFrom($this->field($fields, 'type')); + if ($type === null) { + throw new ClusterTransportException('Redis Stream event has an invalid event type.'); + } + + $identifier = $this->field($fields, 'has_identifier') === '1' + ? $this->field($fields, 'identifier') + : null; + + return new InvalidationEvent( + $id, + $this->clusterFromStreamEvent($fields), + $this->field($fields, 'namespace'), + $type, + $identifier, + $this->field($fields, 'origin'), + (int) $this->field($fields, 'created_at'), + ); + } + + private function field(mixed $fields, string $name): string + { + if (!is_array($fields)) { + throw new ClusterTransportException('Redis Stream event fields must be an array.'); + } + + $value = $fields[$name] ?? null; + if (!is_string($value) || $value === '') { + throw new ClusterTransportException("Redis Stream event has an invalid {$name} field."); + } + + return $value; + } + + /** + * @param string $id The ID argument. + * @return array The ID parts. + * @phpstan-return array{int, int} + */ + private function idParts(string $id): array + { + $parts = explode('-', $id, 2); + if (count($parts) !== 2 || !ctype_digit($parts[0]) || !ctype_digit($parts[1])) { + throw new ClusterTransportException('Redis Stream event IDs must have the form milliseconds-sequence.'); + } + + return [(int) $parts[0], (int) $parts[1]]; + } + + private function stream(string $cluster): string + { + if ($cluster === '') { + throw new ClusterTransportException('Redis Stream cluster name cannot be empty.'); + } + + return $this->prefix . $cluster; + } +} diff --git a/src/Cluster/Transport/TransactionalInvalidationTransportInterface.php b/src/Cluster/Transport/TransactionalInvalidationTransportInterface.php new file mode 100644 index 0000000..72f5000 --- /dev/null +++ b/src/Cluster/Transport/TransactionalInvalidationTransportInterface.php @@ -0,0 +1,13 @@ +connect($host, $port); + if (is_string($parts['pass'] ?? null) && $parts['pass'] !== '') { + $client->auth($parts['pass']); + } + if (is_string($parts['path'] ?? null) && $parts['path'] !== '/') { + $client->select((int) ltrim($parts['path'], '/')); + } + + return $client; + } +} diff --git a/src/Counter/Exception/AtomicCounterException.php b/src/Counter/Exception/AtomicCounterException.php new file mode 100644 index 0000000..661f217 --- /dev/null +++ b/src/Counter/Exception/AtomicCounterException.php @@ -0,0 +1,9 @@ + 0 then + redis.call('EXPIRE', KEYS[1], ARGV[2]) +end +return { value, exists == 0 and 1 or 0 } +LUA; + + private string $namespace; + + public function __construct( + private \Redis $client, + string $namespace = 'default', + ) { + if (!class_exists(\Redis::class)) { + throw new AtomicCounterException('phpredis extension not loaded'); + } + + $this->namespace = sanitize_cache_ns($namespace); + } + + public function decrement(string $key, int $by = 1, ?int $ttlSeconds = null): AtomicCounterValue + { + if ($by < 1) { + throw new AtomicCounterException('Atomic counter decrement value must be greater than zero.'); + } + + return $this->change($key, -$by, $ttlSeconds); + } + + public function delete(string $key): bool + { + return $this->client->del($this->map($key)) !== false; + } + + public function get(string $key): ?int + { + $value = $this->client->get($this->map($key)); + if ($value === false || $value === null) { + return null; + } + + if (!is_string($value) || !preg_match('/^-?\d+$/D', $value)) { + throw new AtomicCounterException('Atomic counter contains a non-integer value.'); + } + + return (int) $value; + } + + public function increment(string $key, int $by = 1, ?int $ttlSeconds = null): AtomicCounterValue + { + if ($by < 1) { + throw new AtomicCounterException('Atomic counter increment value must be greater than zero.'); + } + + return $this->change($key, $by, $ttlSeconds); + } + + private function change(string $key, int $by, ?int $ttlSeconds): AtomicCounterValue + { + $ttl = $this->normalizeTtl($ttlSeconds); + $result = $this->client->eval(self::INCREMENT_SCRIPT, [$this->map($key), (string) $by, (string) $ttl], 1); + if (!is_array($result) || !isset($result[0], $result[1]) || !is_numeric($result[0]) || !is_numeric($result[1])) { + throw new AtomicCounterException('Unable to update atomic counter.'); + } + + return new AtomicCounterValue((int) $result[0], (int) $result[1] === 1); + } + + private function map(string $key): string + { + if (!preg_match('/^[A-Za-z0-9_.-]+$/D', $key)) { + throw new AtomicCounterException('Atomic counter key is invalid.'); + } + + return $this->namespace . ':counter:' . $key; + } + + private function normalizeTtl(?int $ttlSeconds): int + { + if ($ttlSeconds === null) { + return -1; + } + + if ($ttlSeconds < 1) { + throw new AtomicCounterException('Atomic counter TTL must be greater than zero when provided.'); + } + + return $ttlSeconds; + } +} diff --git a/src/Node/NodeCache.php b/src/Node/NodeCache.php index 815eff5..ee167ef 100644 --- a/src/Node/NodeCache.php +++ b/src/Node/NodeCache.php @@ -29,7 +29,7 @@ public static function create(NodeCacheConfig $config): Cache return new Cache( $adapter, - new FileLockProvider($config->lockDirectory), + $config->lockProvider ?? new FileLockProvider($config->lockDirectory), $metrics, ); } diff --git a/src/Node/NodeCacheConfig.php b/src/Node/NodeCacheConfig.php index bf68e02..bceb82a 100644 --- a/src/Node/NodeCacheConfig.php +++ b/src/Node/NodeCacheConfig.php @@ -4,6 +4,7 @@ namespace Infocyph\CacheLayer\Node; +use Infocyph\CacheLayer\Cache\Lock\LockProviderInterface; use Infocyph\CacheLayer\Node\Exception\NodeCacheConfigurationException; final readonly class NodeCacheConfig @@ -17,6 +18,7 @@ public function __construct( public int $busyTimeoutMs = 1_000, public bool $apcuEnabled = true, public bool $failOpen = true, + public ?LockProviderInterface $lockProvider = null, ) { if ($sqliteFile === '' || str_contains($sqliteFile, "\0")) { throw new NodeCacheConfigurationException('The SQLite cache file path is invalid.'); diff --git a/tests/Cluster/ClusterCacheTest.php b/tests/Cluster/ClusterCacheTest.php index 803e3e1..14e08ad 100644 --- a/tests/Cluster/ClusterCacheTest.php +++ b/tests/Cluster/ClusterCacheTest.php @@ -89,14 +89,55 @@ ->and($this->nodeA->consume())->toBe(0); }); +test('cluster bulk tag invalidation publishes each unique tag once', function () { + $this->nodeB->cache()->setTagged('products.list', 'products', ['products'], 300); + $this->nodeB->cache()->setTagged('search.list', 'search', ['search'], 300); + + $this->nodeA->invalidateTags(['products', 'search', 'products']); + + expect($this->nodeB->consume())->toBe(2) + ->and($this->nodeB->cache()->get('products.list'))->toBeNull() + ->and($this->nodeB->cache()->get('search.list'))->toBeNull(); +}); + +test('cluster status reports cursor position, pending events, and consume results', function () { + $this->transport->publish(InvalidationEvent::key('test-cluster', 'application', 'first', 'writer')); + $this->transport->publish(InvalidationEvent::key('test-cluster', 'application', 'second', 'writer')); + + $before = $this->nodeB->status(); + $this->nodeB->consume(1); + $after = $this->nodeB->status(); + + expect($before->cursor)->toBeNull() + ->and($before->oldestAvailableEventId)->toBe('1') + ->and($before->newestAvailableEventId)->toBe('2') + ->and($before->pendingEventCount)->toBe(2) + ->and($after->cursor)->toBe('1') + ->and($after->cursorUpdatedAt)->toBeInt() + ->and($after->pendingEventCount)->toBe(1) + ->and($after->lastConsumeCount)->toBe(1) + ->and($after->lastConsumeError)->toBeNull(); +}); + +test('cluster drain consumes a bounded sequence of event batches', function () { + $this->transport->publish(InvalidationEvent::key('test-cluster', 'application', 'first', 'writer')); + $this->transport->publish(InvalidationEvent::key('test-cluster', 'application', 'second', 'writer')); + $this->transport->publish(InvalidationEvent::key('test-cluster', 'application', 'third', 'writer')); + + expect($this->nodeB->drain(limit: 1, maxBatches: 2))->toBe(2) + ->and($this->nodeB->consume())->toBe(1); +}); + test('PDO transport replays events, prunes them in batches, and participates in an outbox transaction', function () { $connection = new \PDO('sqlite:' . $this->clusterDirectory . '/transport.sqlite'); - $transport = new PdoInvalidationTransport($connection); + $transport = new PdoInvalidationTransport($connection, allowSqliteForTesting: true); $first = $transport->publish(InvalidationEvent::key('pdo-cluster', 'application', 'first', 'writer')); $second = $transport->publish(InvalidationEvent::tag('pdo-cluster', 'application', 'products', 'writer')); expect($transport->consumeAfter('pdo-cluster', $first, 10)->events)->toHaveCount(1) ->and($transport->oldestAvailableId('pdo-cluster'))->toBe($first) + ->and($transport->newestAvailableId('pdo-cluster'))->toBe($second) + ->and($transport->countAfter('pdo-cluster', $first))->toBe(1) ->and($transport->isCursorBefore($first, $second))->toBeTrue() ->and($transport->pruneBefore(time() + 1, 1))->toBe(1) ->and($transport->oldestAvailableId('pdo-cluster'))->toBe($second); @@ -110,3 +151,42 @@ expect($transport->consumeAfter('pdo-cluster', $second, 10)->events)->toBe([]); }); + +test('PDO transport refuses SQLite unless it is explicitly test-only', function () { + $connection = new \PDO('sqlite:' . $this->clusterDirectory . '/unsafe-transport.sqlite'); + + expect(fn () => new PdoInvalidationTransport($connection)) + ->toThrow(\Infocyph\CacheLayer\Cluster\Exception\ClusterTransportException::class); +}); + +test('transactional outbox publishes with the source transaction and applies locally after commit', function () { + $connection = new \PDO('sqlite:' . $this->clusterDirectory . '/outbox.sqlite'); + $transport = new PdoInvalidationTransport($connection, allowSqliteForTesting: true); + $runtime = ClusterCache::create($this->nodeConfigA, $this->clusterConfigA, $transport); + $runtime->cache()->set('product.42', 'stale', 300); + + $connection->beginTransaction(); + $outbox = $runtime->outbox($connection); + $outbox->invalidateKey('product.42'); + $connection->commit(); + + expect($runtime->cache()->get('product.42'))->toBe('stale'); + + $outbox->applyLocally(); + + expect($runtime->cache()->get('product.42'))->toBeNull(); +}); + +test('transactional outbox events replay on their origin node after a post-commit crash window', function () { + $connection = new \PDO('sqlite:' . $this->clusterDirectory . '/outbox-replay.sqlite'); + $transport = new PdoInvalidationTransport($connection, allowSqliteForTesting: true); + $runtime = ClusterCache::create($this->nodeConfigA, $this->clusterConfigA, $transport); + $runtime->cache()->set('product.42', 'stale', 300); + + $connection->beginTransaction(); + $runtime->outbox($connection)->invalidateKey('product.42'); + $connection->commit(); + + expect($runtime->consume())->toBe(1) + ->and($runtime->cache()->get('product.42'))->toBeNull(); +}); diff --git a/tests/Cluster/Support/InMemoryInvalidationTransport.php b/tests/Cluster/Support/InMemoryInvalidationTransport.php index 5f2a3b2..6973478 100644 --- a/tests/Cluster/Support/InMemoryInvalidationTransport.php +++ b/tests/Cluster/Support/InMemoryInvalidationTransport.php @@ -6,9 +6,9 @@ use Infocyph\CacheLayer\Cluster\Event\InvalidationBatch; use Infocyph\CacheLayer\Cluster\Event\InvalidationEvent; -use Infocyph\CacheLayer\Cluster\Transport\InvalidationTransportInterface; +use Infocyph\CacheLayer\Cluster\Transport\InvalidationTransportInspectorInterface; -final class InMemoryInvalidationTransport implements InvalidationTransportInterface +final class InMemoryInvalidationTransport implements InvalidationTransportInspectorInterface { /** @var array> */ private array $events = []; @@ -44,6 +44,18 @@ public function oldestAvailableId(string $cluster): ?string return $this->events[$cluster][0]->id ?? null; } + public function newestAvailableId(string $cluster): ?string + { + $events = $this->events[$cluster] ?? []; + + return $events === [] ? null : $events[array_key_last($events)]->id; + } + + public function countAfter(string $cluster, ?string $cursor): int + { + return count($this->consumeAfter($cluster, $cursor, PHP_INT_MAX)->events); + } + public function publish(InvalidationEvent $event): string { $id = (string) $this->nextId++; diff --git a/tests/Node/NodeCacheTest.php b/tests/Node/NodeCacheTest.php index 49e2a06..46ff831 100644 --- a/tests/Node/NodeCacheTest.php +++ b/tests/Node/NodeCacheTest.php @@ -2,6 +2,8 @@ use Infocyph\CacheLayer\Cache\Adapter\ArrayCacheAdapter; use Infocyph\CacheLayer\Cache\Cache; +use Infocyph\CacheLayer\Cache\Lock\LockHandle; +use Infocyph\CacheLayer\Cache\Lock\LockProviderInterface; use Infocyph\CacheLayer\Node\Adapter\NodeCacheAdapter; use Infocyph\CacheLayer\Node\Adapter\NodeSqliteCacheAdapter; use Infocyph\CacheLayer\Node\Connection\NodeSqliteConnection; @@ -54,6 +56,44 @@ ->and($cache->get('deferred'))->toBe('value'); }); +test('node cache uses the configured shared lock provider for remember operations', function () { + $lock = new class implements LockProviderInterface { + public int $acquired = 0; + + public int $released = 0; + + public function acquire(string $key, float $waitSeconds): ?LockHandle + { + if ($waitSeconds < 0) { + return null; + } + + ++$this->acquired; + + return new LockHandle($key, 'test-lock'); + } + + public function release(?LockHandle $handle): void + { + if (!$handle instanceof LockHandle) { + return; + } + + ++$this->released; + } + }; + $config = new NodeCacheConfig( + sqliteFile: $this->nodeCacheFile, + namespace: 'node.tests', + apcuEnabled: false, + lockProvider: $lock, + ); + + expect(NodeCache::create($config)->remember('locked', fn(): string => 'value', 300))->toBe('value') + ->and($lock->acquired)->toBe(1) + ->and($lock->released)->toBe(1); +}); + test('SQLite hits promote into an L1 cache without returning the child item', function () { $connection = NodeSqliteConnection::create($this->nodeConfig); $l1 = new ArrayCacheAdapter($this->nodeConfig->namespace);