docs_atom

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
 --------------------
 

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.

docs/index.rst

@@ -48,6 +48,7 @@ Quick Start
    :caption: Guide
 
    cache
+   counters
    adapters/index
    cookbook
    metrics-and-locking

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

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,
+        );
     }
 }

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<string> $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,
+        );
     }
 }