diff --git a/README.md b/README.md index 06bbd66..de56e89 100644 --- a/README.md +++ b/README.md @@ -22,6 +22,8 @@ visibility, maintenance focus, and faster feature enrichment for caching. - Unified `Cache` facade implementing PSR-6, PSR-16, `ArrayAccess`, and `Countable` - Adapter support for APCu, File, PHP Files, Memcached, Redis, Valkey, Redis Cluster, PDO (SQLite default), Shared Memory, MongoDB, and ScyllaDB - Tiered cache composition via `Cache::tiered()` (L1/L2/... descriptors or pool instances) +- Node-local APCu L1 + SQLite L2 cache via `NodeCache` +- Durable multi-node invalidation coordination via `ClusterCache` - Tagged invalidation with versioned tags: `setTagged()`, `invalidateTag()`, `invalidateTags()` - Stampede-safe `remember()` with pluggable lock providers - Per-adapter metrics counters and export hooks @@ -96,6 +98,43 @@ Request flow: - write L2 - optionally write L1 (controlled by `writeToL1`) +## Node-local cache (APCu -> SQLite) + +```php +use Infocyph\CacheLayer\Node\NodeCache; +use Infocyph\CacheLayer\Node\NodeCacheConfig; + +$cache = NodeCache::create(new NodeCacheConfig( + namespace: 'app', + sqliteFile: '/var/cache/my-app/cache.sqlite', +)); +``` + +This topology keeps an independent, disposable cache on each application +server: APCu provides hot in-memory reads when available, while SQLite keeps a +larger local cache across PHP-FPM restarts. It does not synchronize entries or +invalidation across servers. + +## Cluster invalidation + +```php +use Infocyph\CacheLayer\Cluster\ClusterCache; +use Infocyph\CacheLayer\Cluster\ClusterCacheConfig; + +$cluster = ClusterCache::create( + node: new NodeCacheConfig(sqliteFile: '/var/cache/my-app/cache.sqlite'), + cluster: new ClusterCacheConfig('production', gethostname()), + transport: $durableInvalidationTransport, +); + +$cluster->invalidateKey('product.42'); +$cluster->consume(); +``` + +Cluster Cache distributes only durable invalidation events. It never replicates +cached values, APCu state, or SQLite files; ordinary reads and writes remain +local to each application node. + ## Security Hardening CacheLayer includes optional payload/serialization hardening controls: diff --git a/docs/cluster/_content.inc b/docs/cluster/_content.inc new file mode 100644 index 0000000..4f6fada --- /dev/null +++ b/docs/cluster/_content.inc @@ -0,0 +1,660 @@ +======================== +Cluster Cache +======================== + +``ClusterCache`` coordinates *invalidation* between independent +:doc:`/node/index` instances. It does not copy cache values, APCu memory, or +SQLite files between machines. Normal reads and writes remain local and fast; +a durable shared event stream tells other nodes which local values to discard. + +This is a cache-coherence tool for data that may be briefly stale. It provides +durable eventual invalidation, not synchronous global consistency. + +.. cluster-overview-start + +Conceptual model +---------------- + +Each node owns a Node Cache and a durable cursor. All nodes in a logical +cluster share one replayable invalidation transport: + +.. code-block:: text + + shared durable event transport + +------------------------------------+ + | key / tag / namespace events | + +------------------------------------+ + ^ | + | publish | replay + | v + node A Node Cache + runtime node B Node Cache + runtime + APCu + local SQLite APCu + local SQLite + cursor A cursor B + +When node A invalidates a value, it removes the local value and publishes an +event. Node B's consumer replays that event and removes its own local value. +The next read on either node obtains a fresh value from the authoritative +source and caches it locally. + +What Cluster Cache is and is not +-------------------------------- + +Cluster Cache is: + +* a durable, replayable invalidation protocol on top of Node Cache; +* suitable for multiple application servers with independent local storage; +* able to distribute key, tag, and namespace invalidations; +* tolerant of an offline node when the transport retains events long enough. + +Cluster Cache is not: + +* a value-replication system or shared cache; +* a replacement for transactions, a database, sessions, or distributed locks; +* a guarantee that all nodes observe a change at the same instant; +* appropriate for payment idempotency, global counters, authorization state + that needs immediate revocation, or other authoritative/security-critical + state. + +Cluster identity, nodes, and namespaces +--------------------------------------- + +``ClusterCacheConfig`` has four settings: + +.. code-block:: php + + use Infocyph\CacheLayer\Cluster\ClusterCacheConfig; + + $clusterConfig = new ClusterCacheConfig( + cluster: 'production', + nodeId: 'catalog-web-01', + consumerBatchSize: 1_000, + invalidateLocallyFirst: true, + ); + +``cluster`` + A non-empty logical stream name. ``production``, ``staging``, and + ``tenant-a`` can use the same transport without consuming each other's + events. There is no fixed CacheLayer limit on cluster names; transport + capacity and retention determine the practical limit. + +``nodeId`` + A non-empty identity for this running node. It must be unique among nodes in + the same cluster. The producer's own events are not applied a second time, + but their cursor still advances. A stable hostname is suitable for long-lived + hosts; use a unique instance/pod identity for ephemeral infrastructure. + +``consumerBatchSize`` + Maximum events fetched by ``consume()`` when no explicit limit is supplied. + It must be greater than zero. Start with 1,000 and tune from observed event + volume and consumer runtime. + +``invalidateLocallyFirst`` + Defaults to ``true``. With the default, the initiating node clears its local + cache first and then publishes the event. With ``false``, it publishes first + and clears locally afterward. This selects failure order; it does not make a + two-system operation atomic. See :ref:`cluster-failure-order`. + +The Node Cache ``namespace`` is distinct from the cluster name. A namespace is +stored in every event and remote nodes only apply events for their matching +namespace. This allows independent cache domains to share a cluster transport, +although all events in that cluster are still read to advance each node's +cursor. + +Prerequisites and transport choice +---------------------------------- + +Every node needs: + +* a local writable SQLite file and optional APCu, as described in + :doc:`/node/index`; +* the same cluster name and a unique node ID; +* access to the same durable, replayable event transport; +* a consumer process or scheduled task that calls ``consume()``; +* retention longer than the largest expected node outage plus an operational + safety margin. + +``InvalidationTransportInterface`` is the boundary for the shared event +store. A production transport must assign sortable IDs, replay events after a +cursor, expose its oldest retained ID, and compare its cursor format for +recovery. Suitable technologies include PostgreSQL/MySQL tables, Redis Streams, +NATS JetStream, RabbitMQ durable queues designed for replay, and Kafka. + +Plain publish/subscribe is not sufficient: an offline node would lose the +invalidation and retain a stale local entry until its TTL expires. + +PDO transport +~~~~~~~~~~~~~ + +``PdoInvalidationTransport`` is included for a supplied PostgreSQL or MySQL +PDO connection. It creates a shared ``cachelayer_invalidation_events`` table +and index when constructed. The connection account therefore needs the required +DDL and read/write rights when the transport is initialized. + +.. code-block:: php + + use Infocyph\CacheLayer\Cluster\Transport\Pdo\PdoInvalidationTransport; + + $pdo = new PDO( + 'pgsql:host=db.internal;port=5432;dbname=application', + 'application', + $password, + [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION], + ); + + $transport = new PdoInvalidationTransport($pdo); + +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. + +.. cluster-topology-start + +Bootstrap one runtime per node +------------------------------ + +Build the Node Cache configuration, Cluster configuration, and transport once +in your application's service container or bootstrap. Keep the resulting +``ClusterRuntime`` and inject either the runtime or its cache facade where +needed. + +.. code-block:: php + + use Infocyph\CacheLayer\Cluster\ClusterCache; + use Infocyph\CacheLayer\Cluster\ClusterCacheConfig; + use Infocyph\CacheLayer\Node\NodeCacheConfig; + + $runtime = ClusterCache::create( + node: new NodeCacheConfig( + sqliteFile: '/var/cache/catalog/node-cache.sqlite', + namespace: 'catalog', + lockDirectory: '/var/cache/catalog/locks', + ), + cluster: new ClusterCacheConfig( + cluster: 'production', + nodeId: gethostname() ?: 'catalog-unknown-node', + ), + transport: $transport, + ); + + $cache = $runtime->cache(); + +The returned runtime exposes this small coordination API: + +=============================== ===================================================== +Method Meaning +=============================== ===================================================== +``cache(): Cache`` The normal local Node Cache facade. +``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. +``consume(?int $limit)`` Replay up to ``limit`` events and return the count read. +``recoverIfRequired(): bool`` Perform retention-gap recovery; normally called by consume. +=============================== ===================================================== + +The critical boundary is ``$runtime->cache()``: all ordinary ``get()``, +``set()``, ``remember()``, ``delete()``, and ``clear()`` calls are **local +only**. Use the runtime's three invalidation methods when an operation must be +distributed. + +Adding the third, fourth, or Nth node +------------------------------------- + +There is no central cluster object that receives a list of node instances and +no node-registration call. A cluster is formed when independently running +application nodes use the same ``cluster`` name and shared transport. Each +node creates its own runtime, local SQLite file, and consumer. + +For example, this bootstrap function is deployed on every web/worker node. The +deployment supplies a different ``$nodeId`` on each machine or instance: + +.. code-block:: php + + use Infocyph\CacheLayer\Cluster\ClusterCache; + use Infocyph\CacheLayer\Cluster\ClusterCacheConfig; + use Infocyph\CacheLayer\Cluster\ClusterRuntime; + use Infocyph\CacheLayer\Cluster\Transport\InvalidationTransportInterface; + use Infocyph\CacheLayer\Node\NodeCacheConfig; + + function bootCatalogCluster(string $nodeId, InvalidationTransportInterface $transport): ClusterRuntime + { + return ClusterCache::create( + node: new NodeCacheConfig( + sqliteFile: "/var/cache/catalog/{$nodeId}.sqlite", + namespace: 'catalog', + ), + cluster: new ClusterCacheConfig( + cluster: 'production-catalog', // identical on every node + nodeId: $nodeId, // unique on every node + ), + transport: $transport, // points to one shared event store + ); + } + + $nodeOne = bootCatalogCluster('catalog-web-01', $transport); + $nodeTwo = bootCatalogCluster('catalog-web-02', $transport); + $nodeThree = bootCatalogCluster('catalog-web-03', $transport); + // Continue with any number of nodes. + +In a real deployment these three calls run in separate processes/machines—not +in one PHP request. Each process normally creates its own PDO connection to +the same database or its own client connection to the selected transport. + +The node-addition flow is: + +1. Provision a private local cache directory/SQLite file for the new node. +2. Give it a unique ``nodeId`` while retaining the existing cluster name, + namespace, and transport configuration. +3. Start its consumer before or alongside application traffic. +4. The new node begins with no cursor, safely reads retained events, and then + serves local cache reads/writes like every other node. +5. Existing nodes require no configuration change; they already publish into + the common transport. + +When node one executes ``$nodeOne->invalidateKey('product.42')``, it invalidates +its own cache and publishes an event. Nodes two, three, and every later node +apply that event when their individual consumers next call ``consume()``. + +.. cluster-operations-start + +Local reads and writes +---------------------- + +Read paths do not contact the transport. This keeps the hot path independent +of the event-store availability. + +.. code-block:: php + + // Node-local read through APCu then SQLite; no transport call. + $product = $runtime->cache()->remember( + 'product.42', + function ($item) use ($products) { + $item->expiresAfter(300); + + return $products->find(42); + }, + tags: ['products', 'product.42'], + ); + +Direct writes are local too: + +.. code-block:: php + + $runtime->cache()->set('product.42', $product, 300); + $runtime->cache()->setTagged('catalog.home', $payload, ['products', 'home'], 60); + $runtime->cache()->setMultiple(['product.42' => $product, 'product.43' => $other], 300); + +This is intentional. A cache fill should not broadcast to every node because +each node has its own capacity, traffic pattern, and TTL. Only changes to the +underlying data need cross-node invalidation. + +Distributed invalidation operations +----------------------------------- + +After a source-data change commits, select the narrowest invalidation that +correctly covers affected entries. + +Key invalidation +~~~~~~~~~~~~~~~~ + +Use a key when one cached representation changed: + +.. code-block:: php + + $products->update(42, $input); // authoritative write has committed + $runtime->invalidateKey('product.42'); + +This deletes ``product.42`` locally and publishes a key event for other nodes. + +Tag invalidation +~~~~~~~~~~~~~~~~ + +Use tags when a change affects a group, list, or related derived entries: + +.. code-block:: php + + $products->update(42, $input); + $runtime->invalidateTag('products'); + +Any node that previously stored values with ``['products']`` treats those +entries as stale after consuming the event. You can still delete a direct key +locally if appropriate, then invalidate the shared tag: + +.. code-block:: php + + $runtime->cache()->delete('product.42'); // local direct representation + $runtime->invalidateTag('products'); // distributed aggregate invalidation + +Namespace invalidation +~~~~~~~~~~~~~~~~~~~~~~ + +Use ``clearNamespace()`` for a controlled broad reset, such as a schema or +serialization deployment that invalidates every entry in the namespace: + +.. code-block:: php + + $runtime->clearNamespace(); + +It is intentionally expensive because every participating node clears that +namespace after consuming the event. Prefer a key or tag for routine writes. + +End-to-end application workflows +-------------------------------- + +Read a product +~~~~~~~~~~~~~~ + +1. Request calls ``$runtime->cache()->remember('product.42', ...)``. +2. The node checks APCu, then local SQLite. +3. On a miss, the resolver reads the authoritative store and caches the value + locally. +4. No cluster event is created; cache fills are local. + +Update a product +~~~~~~~~~~~~~~~~ + +1. Commit the source-of-truth update. +2. Call ``$runtime->invalidateKey('product.42')`` and/or + ``$runtime->invalidateTag('products')``. +3. The initiating node invalidates its own local state and publishes events. +4. Other nodes' consumers replay those events and invalidate their local state. +5. Subsequent reads rebuild local entries from the authoritative source. + +Delete a product +~~~~~~~~~~~~~~~~ + +1. Commit the authoritative deletion. +2. Invalidate the direct representation and any affected list/search tags. +3. Allow consumers to propagate the same invalidation to other nodes. + +There is no ``ClusterRuntime::invalidateTags()`` method. Publish one event per +tag instead: + +.. code-block:: php + + $runtime->invalidateKey('product.42'); + $runtime->invalidateTag('products'); + $runtime->invalidateTag('search-results'); + +Do not call ``$runtime->cache()->invalidateTags()`` for distributed work; that +would only invalidate the initiating node. + +Consumer lifecycle +------------------ + +Each application node needs a consumer. ``consume()`` performs these steps: + +1. Check whether the node's stored cursor is older than retained events. +2. If it is, clear the local namespace and reset the cursor safely. +3. Fetch up to the requested number of events after the current cursor. +4. Ignore events emitted by this same node (they were already applied locally). +5. Apply matching-namespace events from other nodes. +6. Advance the local SQLite cursor after every event, including ignored events. + +``consume()`` returns the number of events *read*, not the number applied. It +returns zero when no events are available or when passed a limit below one. + +Scheduled polling +~~~~~~~~~~~~~~~~~ + +For moderate traffic, run a short CLI task on every node every few seconds or +minutes, according to the maximum acceptable stale window: + +.. code-block:: php + + // bin/cache-consume.php: bootstrap $runtime for this node first. + $processed = $runtime->consume(); + +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. + +Cluster consumers need scheduling +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +CacheLayer deliberately does not run a background daemon or install cron jobs: +the host application owns process supervision, credentials, logging, retries, +and graceful shutdown. However, **every node must run a consumer** for remote +invalidations to become effective. Choose one of these deployment models: + +``cron`` or framework scheduler + Good for modest event volume and a known stale-data budget. Schedule the + same small CLI command on every application node. The poll interval is part + of the remote-staleness budget. + +Supervised long-running worker + Good for low-latency or high-volume systems. Run a worker under systemd, + Supervisor, Kubernetes, or your framework queue/worker runtime. It calls + bounded ``consume()`` loops and polls/sleeps between iterations. + +Transport-native delivery + A custom transport can integrate with a durable broker's worker model, but + it still must preserve the replay/cursor contract. CacheLayer's consumer is + explicitly pull-based and works with scheduled polling for all transports. + +For a cron deployment, create a command that only bootstraps the runtime for +the current node and consumes one bounded batch: + +.. code-block:: php + + // bin/cache-cluster-consume.php + $processed = $runtime->consume(1_000); + +Then schedule it on **every** node. This example polls every minute; lower the +interval only when the resulting remote-staleness window is acceptable for the +application and the transport can handle the polling rate: + +.. code-block:: text + + * * * * * www-data /usr/bin/php /srv/application/bin/cache-cluster-consume.php + +Traditional cron cannot reliably poll more often than once a minute. For +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. + +Continuous worker with bounded drain +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +For higher event volume, use a supervised worker. Keep individual iterations +bounded so one constantly busy stream cannot starve health checks or graceful +shutdown handling in your worker framework: + +.. code-block:: php + + $limit = 1_000; + do { + $processed = $runtime->consume($limit); + } while ($processed === $limit); + +Then let the worker sleep/poll according to the transport and your supervision +model. CacheLayer does not start a daemon or provide a scheduler; deployment +owns that lifecycle. + +Startup and deployment behavior +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +It is safe to call ``consume()`` during worker startup before serving traffic. +That lets a node catch up before it begins filling local cache entries. Retain +events long enough for planned deployments, autoscaling delays, node outages, +and the time needed to drain a backlog. + +If a node has a new empty local SQLite file, it has no cursor and begins from +the retained stream. This may replay unnecessary invalidations but is safe. If +a node intentionally changes its ``nodeId``, it gets a distinct cursor record; +ensure the identity remains unique and monitor for unexpected backlog replay. + +.. cluster-reliability-start + +Retention, recovery, and offline nodes +-------------------------------------- + +Each node stores ``(cluster, nodeId, lastEventId)`` in its local Node Cache +SQLite database. The transport retains only a finite event history. If a node's +cursor predates the oldest available event, it cannot know every invalidation it +missed. Before normal consumption, Cluster Cache therefore: + +1. clears that node's entire local namespace; +2. resets its cursor to the oldest retained event ID; +3. resumes consumption on subsequent reads of the event stream. + +``recoverIfRequired()`` exposes this check directly and returns ``true`` when +a clear/recovery occurred. Calling ``consume()`` already calls it, so direct +calls are mainly useful for health checks or explicit startup handling. + +This safe reset is why event retention is an availability and cache-warmth +decision rather than a correctness shortcut. Short retention causes more full +local clears after outages; long retention increases transport storage and +catch-up work. + +For ``PdoInvalidationTransport``, prune events in small batches after a chosen +retention period: + +.. code-block:: php + + // Keep events for seven days. Run this repeatedly from maintenance. + $deleted = $transport->pruneBefore(time() - 7 * 24 * 60 * 60, 5_000); + +The PDO transport's ``pruneBefore()`` is intentionally not part of the generic +transport interface. Other transport implementations should use their own +retention mechanism. + +Retention-pruning schedule +~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Event pruning is separate from consumption. For the PDO transport, schedule a +maintenance command against the shared database once; do not run competing +pruners on every web node unless your job platform coordinates them. + +.. code-block:: php + + // bin/cache-cluster-prune.php + $transport->pruneBefore(time() - 7 * 24 * 60 * 60, 5_000); + +.. code-block:: text + + # One elected scheduler/worker, hourly; run repeatedly if the backlog is large. + 12 * * * * www-data /usr/bin/php /srv/application/bin/cache-cluster-prune.php + +Keep the retention window longer than the largest supported outage. Pruning +more often does not permit a shorter retention period safely. + +.. _cluster-failure-order: + +Failure ordering and transactional writes +----------------------------------------- + +An invalidation involves two independent actions: local invalidation and event +publication. ``invalidateLocallyFirst`` chooses which happens first. + +With the default ``true``: + +1. The initiating node becomes fresh immediately. +2. If publication fails, remote nodes are not notified and may be stale until + TTL or a later successful invalidation. + +With ``false``: + +1. Remote propagation is recorded before the initiating local clear. +2. If local invalidation fails, the originating node may remain stale while + remote nodes later become fresh. + +Neither sequence is atomic across the source database and the event transport. +For important database writes, use the PDO transactional-outbox helper with +the same database connection: + +.. code-block:: php + + use Infocyph\CacheLayer\Cluster\Event\InvalidationEvent; + + $pdo->beginTransaction(); + try { + updateProduct($pdo, 42, $input); + $transport->publishWithinTransaction( + $pdo, + InvalidationEvent::key( + 'production', + 'catalog', + 'product.42', + gethostname() ?: 'catalog-unknown-node', + ), + ); + $pdo->commit(); + } catch (Throwable $exception) { + if ($pdo->inTransaction()) { + $pdo->rollBack(); + } + + throw $exception; + } + + // The event is now durable. Clear this origin node explicitly after commit. + $runtime->cache()->delete('product.42'); + +``publishWithinTransaction()`` requires the exact PDO connection held by the +transport and an active transaction. It makes the source update and event +insertion atomic. It does not itself invalidate the origin node: origin events +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. + +Operations checklist +-------------------- + +At deployment: + +* create a local, private cache directory on every node; +* configure one shared durable transport per logical cluster; +* give every node a unique ID and use the same cluster name; +* start a consumer on every node before or alongside application traffic; +* set event retention longer than the expected maximum outage; +* choose bounded cache TTLs even with fast consumers. + +During normal operation: + +* use ``$runtime->cache()->remember()`` for local reads; +* use ``$runtime->cache()->set()`` only for local fills; +* call runtime ``invalidateKey()``, ``invalidateTag()``, or + ``clearNamespace()`` after committed source-data changes; +* run a consumer on every node and node SQLite pruning/checkpoint maintenance + separately; +* run shared-transport retention pruning from one coordinated scheduler; +* monitor consumer lag, transport errors, cursor age, cache hit rate, and + retention-prune activity. + +During an incident: + +* if a node is suspect, ``clearNamespace()`` gives a broad distributed reset; +* if a node was offline longer than retention, allow its consumer recovery to + clear its local namespace before it resumes; +* if the transport is unavailable, ordinary local reads continue, but remote + invalidations stop propagating—restore transport and process the backlog; +* do not rely on Cluster Cache alone for an immediate security revoke; enforce + that decision at the authoritative service or data store. + +Troubleshooting +--------------- + +``A remote node still has an old value`` + Confirm the source change called a *runtime* invalidation method, not + ``$runtime->cache()->delete()`` or ``invalidateTag()``. Then verify the + remote node's consumer is running, connected to the same cluster and shared + transport, and using the expected namespace. + +``A node repeatedly clears its cache on startup`` + Its cursor is likely behind event retention, its local SQLite file is being + discarded, or its node ID changes every deployment. Inspect retention, + volume persistence, and identity configuration. + +``Events are growing without bound`` + Configure transport retention. With the PDO transport, schedule bounded + ``pruneBefore()`` calls. Do not prune more aggressively than your longest + supported outage unless frequent full local-cache recovery is acceptable. + +``The origin node appears stale after a transactional outbox write`` + The consumer intentionally skips its own event. Ensure the application + explicitly deletes or invalidates the origin node after the transaction + commits, as shown above. + +``Can multiple clusters share one transport?`` + Yes. Use distinct ``cluster`` names. Events and cursors are partitioned by + cluster name; capacity, retention, and operational monitoring remain shared + concerns of the underlying transport. diff --git a/docs/cluster/index.rst b/docs/cluster/index.rst new file mode 100644 index 0000000..f315c32 --- /dev/null +++ b/docs/cluster/index.rst @@ -0,0 +1,21 @@ +========================= +Cluster Cache +========================= + +Cluster Cache distributes durable invalidation events between independent Node +Cache instances. It never replicates values or local cache files: reads and +writes stay on the current node, while key, tag, and namespace invalidations +are replayed by consumers on other nodes. + +Use the pages below to choose a transport, bootstrap every node, operate the +consumer, and plan recovery and retention. + +.. toctree:: + :maxdepth: 1 + + overview + topology + operations + reliability + +Cluster Cache builds on :doc:`/node/index`. diff --git a/docs/cluster/operations.rst b/docs/cluster/operations.rst new file mode 100644 index 0000000..0c71a4a --- /dev/null +++ b/docs/cluster/operations.rst @@ -0,0 +1,10 @@ +===================================== +Cluster Cache: Operations and Consumer +===================================== + +This page separates ordinary local cache operations from distributed +invalidation, then describes the consumer and scheduling lifecycle. + +.. include:: _content.inc + :start-after: .. cluster-operations-start + :end-before: Retention, recovery, and offline nodes diff --git a/docs/cluster/overview.rst b/docs/cluster/overview.rst new file mode 100644 index 0000000..1b7a00a --- /dev/null +++ b/docs/cluster/overview.rst @@ -0,0 +1,7 @@ +===================================== +Cluster Cache: Overview and Transport +===================================== + +.. include:: _content.inc + :start-after: .. cluster-overview-start + :end-before: Bootstrap one runtime per node diff --git a/docs/cluster/reliability.rst b/docs/cluster/reliability.rst new file mode 100644 index 0000000..7df2095 --- /dev/null +++ b/docs/cluster/reliability.rst @@ -0,0 +1,9 @@ +===================================== +Cluster Cache: Recovery and Reliability +===================================== + +This page covers retention gaps, recovery, transaction ordering, deployment +checklists, and production troubleshooting. + +.. include:: _content.inc + :start-after: .. cluster-reliability-start diff --git a/docs/cluster/topology.rst b/docs/cluster/topology.rst new file mode 100644 index 0000000..233fd2d --- /dev/null +++ b/docs/cluster/topology.rst @@ -0,0 +1,10 @@ +===================================== +Cluster Cache: Topology and Node Setup +===================================== + +This page explains the runtime API and the exact pattern for adding the third, +fourth, or Nth independently running application node. + +.. include:: _content.inc + :start-after: .. cluster-topology-start + :end-before: Local reads and writes diff --git a/docs/index.rst b/docs/index.rst index a03d14b..06a8c8b 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -51,6 +51,8 @@ Quick Start adapters/index cookbook metrics-and-locking + node/index + cluster/index security serializer memoize diff --git a/docs/node/_content.inc b/docs/node/_content.inc new file mode 100644 index 0000000..cd537e6 --- /dev/null +++ b/docs/node/_content.inc @@ -0,0 +1,427 @@ +======================= +Node-local Cache +======================= + +``NodeCache`` is an opinionated cache for one application server. It combines +an optional APCu L1 (hot, process-local memory) with a SQLite L2 (durable local +disk). The returned object is the usual :class:`Cache` facade, so application +code uses the same PSR-6, PSR-16, tagged-cache, and ``remember()`` API as every +other CacheLayer backend. + +Use it when every server may safely hold its own, disposable copy of cached +data. It is the storage foundation for :doc:`/cluster/index`, but it is also +useful on its own. + +.. node-overview-start + +What Node Cache is and is not +----------------------------- + +Node Cache is: + +* a local APCu -> SQLite cache topology; +* persistent across PHP-FPM or application-worker restarts when the local + SQLite file survives; +* compatible with normal ``Cache`` reads, writes, tags, batches, deferred + writes, metrics, and ``remember()``; +* fail-open by default for APCu and SQLite runtime failures. + +Node Cache is not: + +* a shared cache between servers or containers; +* a distributed lock, queue, counter, rate limiter, or session store; +* a substitute for an authoritative database; +* an immediate global-revocation mechanism for security-sensitive data. + +For cross-node invalidation while retaining local cached values, use +:doc:`/cluster/index`. + +Topology and request lifecycle +------------------------------ + +The topology is deliberately simple: + +.. code-block:: text + + PHP request / worker + | + v + Cache facade (PSR-6 / PSR-16 / tags / remember) + | + v + APCu L1, when enabled and available + | + | miss + v + SQLite L2 on this node's local disk + | + | miss + v + application resolver / authoritative source + +Read lifecycle +~~~~~~~~~~~~~~ + +1. APCu is checked when enabled. +2. On an APCu miss, SQLite is checked. +3. A SQLite hit is promoted to APCu with its remaining TTL. +4. A miss returns the caller's default, or ``remember()`` resolves the value. +5. Expired rows are misses. They are left for bounded maintenance rather than + being deleted on every read. + +Write lifecycle +~~~~~~~~~~~~~~~ + +1. A normal write is persisted to SQLite first. +2. A successful SQLite write is copied to APCu when APCu is available. +3. A delete or namespace clear is attempted across both layers. +4. Tagged entries and tag-version metadata are local to this node. + +With the default ``failOpen: true``, an L2 failure can fall back to an L1 write +so the request can continue. This is a resilience choice: the L1-only value is +lost when that PHP/APCu process is restarted. Set ``failOpen: false`` when a +storage failure must be surfaced to the application instead. + +Requirements and filesystem placement +-------------------------------------- + +Node Cache requires ``ext-pdo`` with the SQLite driver. APCu is optional. +APCu is used only if all of the following are true: + +* ``apcuEnabled`` is ``true`` (the default); +* the APCu extension is loaded; +* APCu is enabled for the executing PHP SAPI. + +Put the SQLite database in a writable directory that is local to the node, +outside the repository and web root. SQLite creates companion ``-wal`` and +``-shm`` files, so the directory—not merely the database file—must be writable +by the application account. + +Do not use NFS, SMB, a synchronized folder, or another network/shared +filesystem for the Node Cache SQLite file. The factory rejects symlinked and +world-writable cache directories as a safety measure. + +Recommended layout: + +.. code-block:: text + + /var/cache/my-application/ + node-cache.sqlite + node-cache.sqlite-wal + node-cache.sqlite-shm + locks/ # optional, used by remember() file locks + +Configuration and bootstrap +--------------------------- + +Create one ``NodeCacheConfig`` for each application node. The SQLite path and +lock directory should be local; the namespace names the logical cache within +that database. + +.. code-block:: php + + use Infocyph\CacheLayer\Node\NodeCache; + use Infocyph\CacheLayer\Node\NodeCacheConfig; + + $config = new NodeCacheConfig( + sqliteFile: '/var/cache/my-application/node-cache.sqlite', + namespace: 'catalog', + lockDirectory: '/var/cache/my-application/locks', + busyTimeoutMs: 1_000, + apcuEnabled: true, + failOpen: true, + ); + + $cache = NodeCache::create($config); + +Configuration reference +~~~~~~~~~~~~~~~~~~~~~~~ + +``sqliteFile`` + Required path to the SQLite database. It must not contain a NUL byte. + +``namespace`` + Logical cache partition inside the SQLite database. It defaults to + ``default`` and is sanitized before use. Use a stable, application-specific + name such as ``catalog`` or ``billing``. + +``lockDirectory`` + Optional directory for the file lock provider used by ``remember()``. Leave + it ``null`` to use the library default, or point it at a local writable + directory for an explicit deployment layout. + +``busyTimeoutMs`` + SQLite busy timeout in milliseconds. The default is 1,000. Increase it + cautiously only when legitimate local write contention is observed; it does + not make a shared/network filesystem safe. + +``apcuEnabled`` + Enables APCu L1 when the extension and SAPI support it. Set it to ``false`` + for deterministic SQLite-only behavior, CLI jobs, or troubleshooting. + +``failOpen`` + Defaults to ``true``. When true, recoverable layer failures are treated as + cache degradation where possible. When false, APCu/SQLite failures propagate + to the caller. + +.. node-operations-start + +Normal cache operations +----------------------- + +``NodeCache::create()`` returns ``Infocyph\CacheLayer\Cache\Cache``. Do not +reach into its SQLite adapter for ordinary application work. Use the facade. + +Simple read and write +~~~~~~~~~~~~~~~~~~~~~ + +.. code-block:: php + + $cache->set('product.42', ['id' => 42, 'name' => 'Keyboard'], 300); + + $product = $cache->get('product.42'); + $missing = $cache->get('does-not-exist', 'fallback'); + + $cache->delete('product.42'); + $cache->clear(); // clears this namespace on this node only + +``set()`` accepts an integer TTL, a ``DateInterval``, or ``null``. ``null`` +uses no explicit expiry; prefer bounded TTLs for data that can change. A TTL of +zero expires immediately. Valid keys contain only letters, digits, ``_``, +``.``, and ``-``. + +Read-through caching with ``remember()`` +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Use ``remember()`` for the normal cache-aside read path. It reads first, then +uses the configured local lock to prevent many concurrent requests on the same +node from doing the same expensive work. + +.. code-block:: php + + $product = $cache->remember( + 'product.42', + function ($item) use ($repository) { + $item->expiresAfter(300); + + return $repository->find(42); + }, + tags: ['products', 'product.42'], + ); + +The lock is node-local. Two different servers may both recompute a missing +value; that is expected for Node Cache. Use short, suitable TTLs and make the +resolver safe to run more than once. + +Tags and invalidation +~~~~~~~~~~~~~~~~~~~~~ + +Tags group related cached values. CacheLayer invalidates them by changing local +tag versions rather than scanning every cached key. + +.. code-block:: php + + $cache->setTagged( + 'category.10.products', + $products, + ['products', 'category.10'], + 300, + ); + + $cache->invalidateTag('category.10'); + $cache->invalidateTags(['products', 'search-results']); + +Tag invalidation affects this node only. In a multi-node deployment, +``invalidateTag()`` on a Node Cache does not notify other machines; call +``ClusterRuntime::invalidateTag()`` instead when Cluster Cache is configured. + +PSR-6 items and deferred writes +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +The facade also acts as a PSR-6 pool. Deferred items are held in memory until +``commit()`` and are then written to SQLite in a batch before APCu is updated. + +.. code-block:: php + + $first = $cache->getItem('product.42') + ->set($product) + ->expiresAfter(300); + $second = $cache->getItem('product.43') + ->set($otherProduct) + ->expiresAfter(300); + + $cache->saveDeferred($first); + $cache->saveDeferred($second); + $cache->commit(); + +For simple batches, use the PSR-16 methods instead: + +.. code-block:: php + + $cache->setMultiple([ + 'product.42' => $product, + 'product.43' => $otherProduct, + ], 300); + + $products = $cache->getMultiple(['product.42', 'product.43']); + $cache->deleteMultiple(['product.42', 'product.43']); + +``saveDeferred()`` is for batching in one process. It does not survive a +process crash before ``commit()`` and is not an asynchronous queue. + +Common source-data workflows +---------------------------- + +Create or update a record +~~~~~~~~~~~~~~~~~~~~~~~~~ + +After the authoritative database transaction succeeds, either delete the exact +key or invalidate the business tag. Avoid writing a speculative database value +to cache before its transaction commits. + +.. code-block:: php + + $database->transaction(function () use ($repository, $input): void { + $repository->updateProduct(42, $input); + }); + + $cache->delete('product.42'); + $cache->invalidateTags(['products', 'category.10']); + +The next reader rebuilds entries through ``remember()``. This approach avoids +having cache state become the source of truth. + +Delete a record +~~~~~~~~~~~~~~~ + +Delete the exact record key and any aggregate tags after the source deletion is +committed: + +.. code-block:: php + + $repository->deleteProduct(42); + $cache->delete('product.42'); + $cache->invalidateTag('products'); + +When a direct key and its aggregate/list views may be cached, invalidating both +is normal and safe. + +.. node-production-start + +Maintenance lifecycle +--------------------- + +Run maintenance independently on every node. Reads correctly ignore expired +rows, but pruning prevents the local SQLite file from accumulating them. + +.. code-block:: php + + use Infocyph\CacheLayer\Node\NodeCache; + + $maintenance = NodeCache::maintenance($config); + $deleted = $maintenance->pruneExpired(5_000); + $maintenance->checkpoint(); + $maintenance->optimize(); + +``pruneExpired($limit)`` + Deletes at most ``$limit`` expired rows in the configured namespace. It + returns the number deleted. Run it periodically and keep the limit bounded. + +``checkpoint()`` + Requests a passive SQLite WAL checkpoint. It is suitable for regular + maintenance because it does not force active writers to stop. + +``optimize()`` + Runs SQLite ``PRAGMA optimize``. Schedule it less frequently than pruning, + for example as part of a daily maintenance job. + +A simple scheduled task can perform a small bounded prune every few minutes +and checkpoint periodically. Do not run ``VACUUM`` on the request path; plan it +separately if disk-space reclamation is required. + +Example cron schedule +~~~~~~~~~~~~~~~~~~~~~ + +CacheLayer does not install a cron entry or start a scheduler. Wire the +maintenance API into your application's CLI bootstrap, then schedule that +command with the mechanism your deployment already uses. + +.. code-block:: php + + // bin/cache-node-maintenance.php + $maintenance = NodeCache::maintenance($config); + $maintenance->pruneExpired(5_000); + $maintenance->checkpoint(); + + // Run optimize less frequently, for example from a separate daily command. + +For a traditional cron-based deployment, a conservative starting point is: + +.. code-block:: text + + # Every five minutes: bounded expired-row cleanup and passive WAL checkpoint. + */5 * * * * www-data /usr/bin/php /srv/application/bin/cache-node-maintenance.php + + # Daily: bootstrap a command that calls $maintenance->optimize(). + 17 3 * * * www-data /usr/bin/php /srv/application/bin/cache-node-optimize.php + +Use your framework scheduler, Kubernetes ``CronJob``, systemd timer, or another +platform scheduler instead when that is how application jobs are operated. +Only one maintenance task is needed per local SQLite file; it is safe to use a +bounded prune limit if overlap cannot be ruled out. + +Deployment and process guidance +------------------------------- + +* Create the cache directory with ownership restricted to the application + account before starting PHP-FPM/workers. +* Persist the SQLite file on the node if warm cache across process restarts is + useful. Treat it as disposable: deleting it only causes cold misses. +* In containers, mount a node-local writable volume. Do not mount the same + SQLite cache volume into multiple nodes. +* Configure APCu for the PHP SAPI that actually serves traffic. A CLI command + may not have the same APCu settings as PHP-FPM. +* Use the same namespace for processes that should share a node-local cache; + use distinct namespaces to isolate applications or domains. +* Keep cache TTLs shorter than the maximum period during which stale data would + be tolerable. + +Observability and failures +-------------------------- + +``exportMetrics()`` exposes adapter counters, including cache hits/misses and +Node Cache layer events. Export it through your application's telemetry path: + +.. code-block:: php + + $metrics = $cache->exportMetrics(); + +Layer failures are represented by metrics such as APCu/SQLite failures when +``failOpen`` is enabled. Alert on sustained failures, a sharp fall in hit rate, +rapid SQLite file growth, and repeated lock contention. With ``failOpen`` +disabled, also handle the propagated cache exception according to the request's +availability requirements. + +Troubleshooting +--------------- + +``APCu is not being used`` + Check that ``apcuEnabled`` is true, the APCu extension is loaded, and APCu is + enabled for the active SAPI. SQLite-only operation is an intended fallback. + +``SQLite database is locked`` + Confirm that the database is on a local filesystem, inspect concurrent local + writers, then consider a modest ``busyTimeoutMs`` increase. Do not solve + this by placing the file on a shared filesystem. + +``Data looks stale on another server`` + This is expected with Node Cache alone. Add Cluster Cache and consume its + invalidation stream on every node, or rely on an appropriately short TTL. + +``The SQLite file keeps growing`` + Schedule ``pruneExpired()`` and periodic checkpoints. Expired rows are not + removed by reads. Review TTL choices and cache cardinality as well. + +``A cache failure should fail the request`` + Create the configuration with ``failOpen: false`` and ensure the application + handles the resulting exception at the appropriate boundary. diff --git a/docs/node/index.rst b/docs/node/index.rst new file mode 100644 index 0000000..09a0113 --- /dev/null +++ b/docs/node/index.rst @@ -0,0 +1,19 @@ +===================== +Node Cache +===================== + +Node Cache is CacheLayer's opinionated cache for a single application server: +optional APCu provides a hot L1 and a local SQLite database provides durable L2 +storage. It returns the normal ``Cache`` facade, so application code keeps the +same PSR-6, PSR-16, tagging, and ``remember()`` APIs. + +Use these focused pages in order: + +.. toctree:: + :maxdepth: 1 + + overview + operations + production + +For cross-server invalidation, continue with :doc:`/cluster/index`. diff --git a/docs/node/operations.rst b/docs/node/operations.rst new file mode 100644 index 0000000..aebcc13 --- /dev/null +++ b/docs/node/operations.rst @@ -0,0 +1,10 @@ +=========================== +Node Cache: Cache Operations +=========================== + +This page covers the normal application-facing ``Cache`` facade: reads, +writes, tags, deferred items, and source-data invalidation patterns. + +.. include:: _content.inc + :start-after: .. node-operations-start + :end-before: Maintenance lifecycle diff --git a/docs/node/overview.rst b/docs/node/overview.rst new file mode 100644 index 0000000..f02a3df --- /dev/null +++ b/docs/node/overview.rst @@ -0,0 +1,7 @@ +=============================== +Node Cache: Overview and Setup +=============================== + +.. include:: _content.inc + :start-after: .. node-overview-start + :end-before: Normal cache operations diff --git a/docs/node/production.rst b/docs/node/production.rst new file mode 100644 index 0000000..7c569fb --- /dev/null +++ b/docs/node/production.rst @@ -0,0 +1,9 @@ +======================================= +Node Cache: Maintenance and Production +======================================= + +This page covers scheduled SQLite maintenance, deployment, observability, and +failure diagnosis. + +.. include:: _content.inc + :start-after: .. node-production-start diff --git a/src/Cluster/ClusterCache.php b/src/Cluster/ClusterCache.php new file mode 100644 index 0000000..e882a18 --- /dev/null +++ b/src/Cluster/ClusterCache.php @@ -0,0 +1,45 @@ +sqliteFile, $cluster->cluster, $cluster->nodeId); + $recovery = new ClusterRecoveryManager($cache, $cursorStore, $transport, $cluster->cluster); + $coordinator = new ClusterCoordinator( + $cache, + $cluster->cluster, + $node->namespace, + $cluster->nodeId, + $transport, + $cluster->invalidateLocallyFirst, + ); + $consumer = new InvalidationConsumer( + $transport, + $cursorStore, + new InvalidationHandler($cache, $node->namespace), + $recovery, + $cluster->cluster, + $cluster->nodeId, + ); + + return new ClusterRuntime($cache, $coordinator, $consumer, $recovery, $cluster->consumerBatchSize); + } +} diff --git a/src/Cluster/ClusterCacheConfig.php b/src/Cluster/ClusterCacheConfig.php new file mode 100644 index 0000000..04968a6 --- /dev/null +++ b/src/Cluster/ClusterCacheConfig.php @@ -0,0 +1,29 @@ +cache; + } + + public function clearNamespace(): void + { + $this->coordinator->clearNamespace(); + } + + public function consume(?int $limit = null): int + { + return $this->consumer->consume($limit ?? $this->consumerBatchSize); + } + + public function invalidateKey(string $key): void + { + $this->coordinator->invalidateKey($key); + } + + public function invalidateTag(string $tag): void + { + $this->coordinator->invalidateTag($tag); + } + + public function recoverIfRequired(): bool + { + return $this->recovery->recoverIfRequired(); + } +} diff --git a/src/Cluster/Consumer/InvalidationConsumer.php b/src/Cluster/Consumer/InvalidationConsumer.php new file mode 100644 index 0000000..6318652 --- /dev/null +++ b/src/Cluster/Consumer/InvalidationConsumer.php @@ -0,0 +1,41 @@ +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); + } + + $this->cursorStore->advance((string) $event->id); + } + + return count($batch->events); + } +} diff --git a/src/Cluster/Consumer/InvalidationHandler.php b/src/Cluster/Consumer/InvalidationHandler.php new file mode 100644 index 0000000..3d95266 --- /dev/null +++ b/src/Cluster/Consumer/InvalidationHandler.php @@ -0,0 +1,44 @@ +namespace !== $this->namespace) { + return; + } + + match ($event->type) { + InvalidationEventType::Key => $this->invalidateKey($event), + InvalidationEventType::Namespace => $this->cache->clear(), + InvalidationEventType::Tag => $this->invalidateTag($event), + }; + } + + private function invalidateKey(InvalidationEvent $event): void + { + if ($event->identifier !== null) { + $this->cache->delete($event->identifier); + } + } + + private function invalidateTag(InvalidationEvent $event): void + { + if ($event->identifier !== null) { + $this->cache->invalidateTag($event->identifier); + } + } +} diff --git a/src/Cluster/Coordinator/ClusterCoordinator.php b/src/Cluster/Coordinator/ClusterCoordinator.php new file mode 100644 index 0000000..9edf836 --- /dev/null +++ b/src/Cluster/Coordinator/ClusterCoordinator.php @@ -0,0 +1,60 @@ +cluster, $this->namespace, $this->nodeId); + $this->coordinate($event, fn(): bool => $this->cache->clear()); + } + + public function invalidateKey(string $key): void + { + $event = InvalidationEvent::key($this->cluster, $this->namespace, $key, $this->nodeId); + $this->coordinate($event, fn(): bool => $this->cache->delete($key)); + } + + public function invalidateTag(string $tag): void + { + $event = InvalidationEvent::tag($this->cluster, $this->namespace, $tag, $this->nodeId); + $this->coordinate($event, fn(): bool => $this->cache->invalidateTag($tag)); + } + + private function coordinate(InvalidationEvent $event, callable $invalidate): void + { + if ($this->invalidateLocallyFirst) { + $this->invalidate($invalidate); + $this->transport->publish($event); + + return; + } + + $this->transport->publish($event); + $this->invalidate($invalidate); + } + + private function invalidate(callable $invalidate): void + { + if (!$invalidate()) { + throw new ClusterCacheException('Unable to invalidate the local node cache.'); + } + } +} diff --git a/src/Cluster/Cursor/CursorStoreInterface.php b/src/Cluster/Cursor/CursorStoreInterface.php new file mode 100644 index 0000000..f60c2aa --- /dev/null +++ b/src/Cluster/Cursor/CursorStoreInterface.php @@ -0,0 +1,14 @@ +connection = $connection ?? NodeSqliteConnection::create( + new NodeCacheConfig($sqliteFile, 'cluster-cursor'), + ); + $this->createSchemaIfMissing(); + } + + public function advance(string $eventId): void + { + if ($eventId === '') { + throw new ClusterCacheException('Cluster cursor event IDs cannot be empty.'); + } + + $this->write($eventId); + } + + public function current(): ?string + { + try { + $statement = $this->connection->prepare( + 'SELECT last_event_id FROM cachelayer_cluster_cursors ' + . 'WHERE cluster_name = :cluster AND node_id = :node_id LIMIT 1', + ); + $statement->execute([':cluster' => $this->cluster, ':node_id' => $this->nodeId]); + $cursor = $statement->fetchColumn(); + } catch (PDOException $exception) { + throw new ClusterCacheException('Unable to read the cluster cursor.', 0, $exception); + } + + return is_string($cursor) && $cursor !== '' ? $cursor : null; + } + + public function reset(?string $eventId): void + { + $this->write($eventId); + } + + private function createSchemaIfMissing(): void + { + try { + $this->connection->exec( + 'CREATE TABLE IF NOT EXISTS cachelayer_cluster_cursors (' + . 'cluster_name TEXT NOT NULL, node_id TEXT NOT NULL, last_event_id TEXT, updated_at INTEGER NOT NULL, ' + . 'PRIMARY KEY (cluster_name, node_id)) WITHOUT ROWID', + ); + } catch (PDOException $exception) { + throw new ClusterCacheException('Unable to initialize the cluster cursor store.', 0, $exception); + } + } + + private function write(?string $eventId): void + { + try { + $statement = $this->connection->prepare( + 'INSERT INTO cachelayer_cluster_cursors (cluster_name, node_id, last_event_id, updated_at) ' + . 'VALUES (:cluster, :node_id, :event_id, :updated_at) ' + . 'ON CONFLICT(cluster_name, node_id) DO UPDATE SET ' + . 'last_event_id = excluded.last_event_id, updated_at = excluded.updated_at', + ); + $statement->execute([ + ':cluster' => $this->cluster, + ':node_id' => $this->nodeId, + ':event_id' => $eventId, + ':updated_at' => time(), + ]); + } catch (PDOException $exception) { + throw new ClusterCacheException('Unable to update the cluster cursor.', 0, $exception); + } + } +} diff --git a/src/Cluster/Event/InvalidationBatch.php b/src/Cluster/Event/InvalidationBatch.php new file mode 100644 index 0000000..42c9cbf --- /dev/null +++ b/src/Cluster/Event/InvalidationBatch.php @@ -0,0 +1,23 @@ + $events + */ + public function __construct(public array $events) + { + foreach ($events as $event) { + if ($event->id === null || $event->id === '') { + throw new ClusterCacheException('Consumed invalidation events require a transport-assigned ID.'); + } + } + } +} diff --git a/src/Cluster/Event/InvalidationEvent.php b/src/Cluster/Event/InvalidationEvent.php new file mode 100644 index 0000000..a9bed1e --- /dev/null +++ b/src/Cluster/Event/InvalidationEvent.php @@ -0,0 +1,52 @@ +cluster, $this->namespace, $this->type, $this->identifier, $this->originNodeId, $this->createdAt); + } +} diff --git a/src/Cluster/Event/InvalidationEventType.php b/src/Cluster/Event/InvalidationEventType.php new file mode 100644 index 0000000..f3e18fe --- /dev/null +++ b/src/Cluster/Event/InvalidationEventType.php @@ -0,0 +1,14 @@ +cursorStore->current(); + $oldest = $this->transport->oldestAvailableId($this->cluster); + if ($cursor === null || $oldest === null || !$this->transport->isCursorBefore($cursor, $oldest)) { + return false; + } + + $this->cache->clear(); + $this->cursorStore->reset($oldest); + + return true; + } +} diff --git a/src/Cluster/Transport/InvalidationTransportInterface.php b/src/Cluster/Transport/InvalidationTransportInterface.php new file mode 100644 index 0000000..afb6781 --- /dev/null +++ b/src/Cluster/Transport/InvalidationTransportInterface.php @@ -0,0 +1,19 @@ +connection->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION); + $driver = $connection->getAttribute(PDO::ATTR_DRIVER_NAME); + $this->driver = is_string($driver) ? $driver : ''; + $this->createSchemaIfMissing(); + } + + public function consumeAfter(string $cluster, ?string $cursor, int $limit): InvalidationBatch + { + if ($limit < 1) { + return new InvalidationBatch([]); + } + + try { + $statement = $this->connection->prepare($this->consumeSql($cursor)); + $statement->bindValue(':cluster', $cluster, PDO::PARAM_STR); + if ($cursor !== null) { + $statement->bindValue(':cursor', $this->eventId($cursor), PDO::PARAM_STR); + } + $statement->bindValue(':limit', $limit, PDO::PARAM_INT); + $statement->execute(); + } catch (PDOException $exception) { + throw new ClusterTransportException('Unable to consume invalidation events.', 0, $exception); + } + + $events = []; + foreach ($statement->fetchAll(PDO::FETCH_ASSOC) as $row) { + if (!is_array($row)) { + continue; + } + $events[] = $this->eventFromRow($this->normalizeRow($row)); + } + + return new InvalidationBatch($events); + } + + public function isCursorBefore(string $cursor, string $oldestAvailableId): bool + { + $cursor = $this->eventId($cursor); + $oldestAvailableId = $this->eventId($oldestAvailableId); + + return strlen($cursor) === strlen($oldestAvailableId) + ? strcmp($cursor, $oldestAvailableId) < 0 + : strlen($cursor) < strlen($oldestAvailableId); + } + + public function oldestAvailableId(string $cluster): ?string + { + try { + $statement = $this->connection->prepare( + 'SELECT MIN(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 the oldest invalidation event ID.', 0, $exception); + } + + return is_int($id) || (is_string($id) && ctype_digit($id)) ? (string) $id : null; + } + + public function pruneBefore(int $retentionBoundary, int $limit = 5_000): int + { + if ($limit < 1) { + return 0; + } + + try { + $statement = $this->connection->prepare($this->pruneSql()); + $statement->bindValue(':boundary', $retentionBoundary, PDO::PARAM_INT); + $statement->bindValue(':limit', $limit, PDO::PARAM_INT); + $statement->execute(); + } catch (PDOException $exception) { + throw new ClusterTransportException('Unable to prune invalidation events.', 0, $exception); + } + + return $statement->rowCount(); + } + + public function publish(InvalidationEvent $event): string + { + return $this->insert($this->connection, $event); + } + + public function publishWithinTransaction(PDO $connection, InvalidationEvent $event): string + { + if ($connection !== $this->connection || !$connection->inTransaction()) { + throw new ClusterTransportException( + 'Transactional outbox publishing requires this transport connection and an active transaction.', + ); + } + + return $this->insert($connection, $event); + } + + private function consumeSql(?string $cursor): string + { + $where = 'cluster_name = :cluster'; + if ($cursor !== null) { + $where .= ' AND event_id > :cursor'; + } + + return 'SELECT event_id, cluster_name, namespace_name, event_type, identifier, origin_node_id, created_at ' + . 'FROM ' . self::TABLE . ' WHERE ' . $where . ' ORDER BY event_id ASC LIMIT :limit'; + } + + private function createClusterIndexIfMissing(): void + { + $ifNotExists = $this->driver === 'mysql' ? '' : ' IF NOT EXISTS'; + + try { + $this->connection->exec( + 'CREATE INDEX' . $ifNotExists . ' cachelayer_invalidation_events_cluster_idx ' + . 'ON ' . self::TABLE . ' (cluster_name, event_id)', + ); + } catch (PDOException $exception) { + if ($this->driver !== 'mysql' || !$this->isDuplicateIndex($exception)) { + throw new ClusterTransportException('Unable to initialize the PDO invalidation transport index.', 0, $exception); + } + } + } + + private function createSchemaIfMissing(): void + { + try { + $this->connection->exec($this->createTableSql()); + } catch (PDOException $exception) { + throw new ClusterTransportException('Unable to initialize the PDO invalidation transport schema.', 0, $exception); + } + + $this->createClusterIndexIfMissing(); + } + + private function createTableSql(): string + { + $id = match ($this->driver) { + 'mysql' => 'BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY', + 'pgsql' => 'BIGSERIAL PRIMARY KEY', + default => 'INTEGER PRIMARY KEY AUTOINCREMENT', + }; + + return 'CREATE TABLE IF NOT EXISTS ' . self::TABLE . ' (' + . 'event_id ' . $id . ', cluster_name VARCHAR(128) NOT NULL, namespace_name VARCHAR(128) NOT NULL, ' + . 'event_type VARCHAR(32) NOT NULL, identifier VARCHAR(512) NULL, origin_node_id VARCHAR(255) NOT NULL, ' + . 'created_at BIGINT NOT NULL)'; + } + + /** + * @param array $row The row argument. + * @phpstan-param array $row + */ + private function eventFromRow(array $row): InvalidationEvent + { + $type = InvalidationEventType::tryFrom($this->requiredString($row, 'event_type')); + if ($type === null) { + throw new ClusterTransportException('Invalid invalidation event record returned by PDO transport.'); + } + + $identifier = $row['identifier'] ?? null; + if ($identifier !== null && !is_string($identifier)) { + throw new ClusterTransportException('Invalid invalidation event identifier returned by PDO transport.'); + } + + return new InvalidationEvent( + $this->eventIdFromValue($row['event_id'] ?? null), + $this->requiredString($row, 'cluster_name'), + $this->requiredString($row, 'namespace_name'), + $type, + $identifier, + $this->requiredString($row, 'origin_node_id'), + $this->timestampFromValue($row['created_at'] ?? null), + ); + } + + private function eventId(string $id): string + { + if (!ctype_digit($id)) { + throw new ClusterTransportException('PDO invalidation transport requires unsigned numeric event IDs.'); + } + + $normalized = ltrim($id, '0'); + + return $normalized === '' ? '0' : $normalized; + } + + private function eventIdFromValue(mixed $value): string + { + if (is_int($value)) { + return (string) $value; + } + + if (is_string($value) && ctype_digit($value)) { + return $value; + } + + throw new ClusterTransportException('Invalid invalidation event ID returned by PDO transport.'); + } + + private function insert(PDO $connection, InvalidationEvent $event): string + { + try { + $statement = $connection->prepare( + 'INSERT INTO ' . self::TABLE . ' ' + . '(cluster_name, namespace_name, event_type, identifier, origin_node_id, created_at) ' + . 'VALUES (:cluster, :namespace, :type, :identifier, :origin, :created_at)', + ); + $statement->execute([ + ':cluster' => $event->cluster, + ':namespace' => $event->namespace, + ':type' => $event->type->value, + ':identifier' => $event->identifier, + ':origin' => $event->originNodeId, + ':created_at' => $event->createdAt, + ]); + $id = $connection->lastInsertId(); + } catch (PDOException $exception) { + throw new ClusterTransportException('Unable to publish an invalidation event.', 0, $exception); + } + + if (!is_string($id) || $id === '') { + throw new ClusterTransportException('PDO invalidation transport did not return an event ID.'); + } + + return $id; + } + + private function isDuplicateIndex(PDOException $exception): bool + { + $errorInfo = $exception->errorInfo; + + return is_array($errorInfo) && ($errorInfo[1] ?? null) === 1061; + } + + /** + * @param array $row The row argument. + * @phpstan-param array $row + * @phpstan-return array + */ + private function normalizeRow(array $row): array + { + $normalized = []; + foreach ($row as $key => $value) { + if (is_string($key)) { + $normalized[$key] = $value; + } + } + + return $normalized; + } + + private function pruneSql(): string + { + $selection = 'SELECT event_id FROM ' . self::TABLE . ' WHERE created_at < :boundary ORDER BY event_id LIMIT :limit'; + if ($this->driver === 'mysql') { + $selection = 'SELECT event_id FROM (' . $selection . ') AS cachelayer_prunable_events'; + } + + return 'DELETE FROM ' . self::TABLE . ' WHERE event_id IN (' . $selection . ')'; + } + + private function requiredString(mixed $row, string $key): string + { + if (!is_array($row)) { + throw new ClusterTransportException('Invalid PDO transport event row.'); + } + + $value = $row[$key] ?? null; + if (!is_string($value) || $value === '') { + throw new ClusterTransportException("Invalid {$key} returned by PDO transport."); + } + + return $value; + } + + private function timestampFromValue(mixed $value): int + { + if (is_int($value)) { + return $value; + } + + if (is_string($value) && ctype_digit($value)) { + return (int) $value; + } + + throw new ClusterTransportException('Invalid invalidation event timestamp returned by PDO transport.'); + } +} diff --git a/src/Node/Adapter/NodeCacheAdapter.php b/src/Node/Adapter/NodeCacheAdapter.php new file mode 100644 index 0000000..2c83516 --- /dev/null +++ b/src/Node/Adapter/NodeCacheAdapter.php @@ -0,0 +1,287 @@ +runAcrossLayers(static fn(CacheItemPoolInterface $pool): bool => $pool->clear()); + } + + #[\Override] + public function commit(): bool + { + $items = array_values($this->deferred); + if ($items === []) { + return true; + } + + try { + $stored = $this->l2->saveMany($items); + } catch (Throwable $exception) { + if (!$this->failOpen) { + throw $exception; + } + + if (!$this->saveAllToL1($items)) { + return false; + } + + $this->deferred = []; + + return true; + } + + if (!$stored) { + return false; + } + + $this->deferred = []; + if ($this->l1 === null) { + return true; + } + + $this->saveAllToL1($items); + + return true; + } + + public function count(): int + { + try { + return count($this->l2); + } catch (Throwable $exception) { + if (!$this->failOpen) { + throw $exception; + } + + $this->metric('sqlite_failure'); + + return 0; + } + } + + public function deleteItem(string $key): bool + { + return $this->runAcrossLayers(static fn(CacheItemPoolInterface $pool): bool => $pool->deleteItem($key)); + } + + /** + * @param array $keys The keys argument. + * @phpstan-param list $keys + */ + public function deleteItems(array $keys): bool + { + return $this->runAcrossLayers(static fn(CacheItemPoolInterface $pool): bool => $pool->deleteItems($keys)); + } + + public function getItem(string $key): GenericCacheItem + { + $l1Item = $this->itemFromL1($key); + if ($l1Item !== null) { + return $l1Item; + } + + try { + $l2Item = $this->l2->getItem($key); + } catch (Throwable $exception) { + if (!$this->failOpen) { + throw $exception; + } + + $this->metric('sqlite_failure'); + + return new GenericCacheItem($this, $key); + } + + if (!$l2Item->isHit()) { + $this->metric('sqlite_miss'); + + return new GenericCacheItem($this, $key); + } + + $this->metric('sqlite_hit'); + $this->saveToL1($l2Item); + + return $this->nodeItem($l2Item); + } + + public function hasItem(string $key): bool + { + return $this->getItem($key)->isHit(); + } + + /** + * @param array $keys The keys argument. + * @phpstan-param list $keys + * @phpstan-return array + */ + public function multiFetch(array $keys): array + { + return $this->multiFetchItems($keys, $this->getItem(...)); + } + + public function save(CacheItemInterface $item): bool + { + if (!$this->supportsItem($item)) { + return false; + } + + try { + $stored = $this->l2->save($item); + } catch (Throwable $exception) { + if (!$this->failOpen) { + throw $exception; + } + + $this->metric('sqlite_failure'); + + return $this->saveToL1($item); + } + + if (!$stored) { + $this->metric('write_failure'); + + return false; + } + + $this->metric('write_success'); + $this->saveToL1($item); + + return true; + } + + protected function supportsItem(CacheItemInterface $item): bool + { + return $item instanceof GenericCacheItem; + } + + private function itemFromL1(string $key): ?GenericCacheItem + { + if ($this->l1 === null) { + return null; + } + + try { + $item = $this->l1->getItem($key); + } catch (Throwable $exception) { + if (!$this->failOpen) { + throw $exception; + } + + $this->metric('apcu_failure'); + + return null; + } + + if (!$item->isHit()) { + $this->metric('apcu_miss'); + + return null; + } + + $this->metric('apcu_hit'); + + return $this->nodeItem($item); + } + + private function metric(string $name): void + { + $this->metrics->increment(self::class, $name); + } + + private function nodeItem(CacheItemInterface $item): GenericCacheItem + { + $nodeItem = new GenericCacheItem($this, $item->getKey(), $item->get(), true); + $expires = CachePayloadCodec::expirationFromItem($item); + if ($expires['ttl'] !== null) { + $nodeItem->expiresAfter($expires['ttl']); + } + + return $nodeItem; + } + + private function runAcrossLayers(callable $operation): bool + { + $success = false; + $failure = null; + + foreach ([$this->l2, $this->l1] as $pool) { + if (!$pool instanceof CacheItemPoolInterface) { + continue; + } + + try { + $success = $operation($pool) || $success; + } catch (Throwable $exception) { + $failure ??= $exception; + } + } + + if ($failure !== null && !$this->failOpen) { + throw $failure; + } + + return $success; + } + + /** + * @param array $items The items argument. + * @phpstan-param list $items + */ + private function saveAllToL1(array $items): bool + { + $success = true; + foreach ($items as $item) { + $success = $this->saveToL1($item) && $success; + } + + return $success; + } + + private function saveToL1(CacheItemInterface $item): bool + { + if ($this->l1 === null) { + return true; + } + + try { + $target = $this->l1->getItem($item->getKey()); + $target->set($item->get()); + $expires = CachePayloadCodec::expirationFromItem($item); + $target->expiresAfter($expires['ttl']); + $stored = $this->l1->save($target); + } catch (Throwable $exception) { + if (!$this->failOpen) { + throw $exception; + } + + $this->metric('apcu_failure'); + + return false; + } + + $this->metric($stored ? 'promotion_success' : 'promotion_failure'); + + return $stored; + } +} diff --git a/src/Node/Adapter/NodeSqliteCacheAdapter.php b/src/Node/Adapter/NodeSqliteCacheAdapter.php new file mode 100644 index 0000000..f7bc852 --- /dev/null +++ b/src/Node/Adapter/NodeSqliteCacheAdapter.php @@ -0,0 +1,257 @@ +createSchemaIfMissing(); + $this->deleteStatement = $connection->prepare( + 'DELETE FROM ' . self::TABLE . ' WHERE namespace = :namespace AND cache_key = :cache_key', + ); + $this->lookupStatement = $connection->prepare( + 'SELECT payload, expires_at FROM ' . self::TABLE . ' WHERE namespace = :namespace ' + . 'AND cache_key = :cache_key AND (expires_at IS NULL OR expires_at > :current_time) LIMIT 1', + ); + $this->upsertStatement = $connection->prepare( + 'INSERT INTO ' . self::TABLE . ' (namespace, cache_key, payload, expires_at) ' + . 'VALUES (:namespace, :cache_key, :payload, :expires_at) ' + . 'ON CONFLICT(namespace, cache_key) DO UPDATE SET ' + . 'payload = excluded.payload, expires_at = excluded.expires_at', + ); + } + + public function clear(): bool + { + try { + $statement = $this->connection->prepare('DELETE FROM ' . self::TABLE . ' WHERE namespace = :namespace'); + $ok = $statement->execute([':namespace' => $this->namespace]); + $this->deferred = []; + + return $ok; + } catch (PDOException $exception) { + throw $this->storageException('Unable to clear the node SQLite cache.', $exception); + } + } + + #[\Override] + public function commit(): bool + { + $items = array_values($this->deferred); + if (!$this->saveMany($items)) { + return false; + } + + $this->deferred = []; + + return true; + } + + public function connection(): PDO + { + return $this->connection; + } + + public function count(): int + { + try { + $statement = $this->connection->prepare( + 'SELECT COUNT(*) FROM ' . self::TABLE . ' WHERE namespace = :namespace ' + . 'AND (expires_at IS NULL OR expires_at > :current_time)', + ); + $statement->execute([':namespace' => $this->namespace, ':current_time' => time()]); + $count = $statement->fetchColumn(); + + return is_numeric($count) ? max(0, (int) $count) : 0; + } catch (PDOException $exception) { + throw $this->storageException('Unable to count node SQLite cache entries.', $exception); + } + } + + public function deleteItem(string $key): bool + { + try { + return $this->deleteStatement->execute([':namespace' => $this->namespace, ':cache_key' => $key]); + } catch (PDOException $exception) { + throw $this->storageException("Unable to delete node SQLite cache key '{$key}'.", $exception); + } + } + + /** + * @param array $keys The keys argument. + * @phpstan-param list $keys + */ + public function deleteItems(array $keys): bool + { + if ($keys === []) { + return true; + } + + try { + $this->connection->beginTransaction(); + foreach ($keys as $key) { + $this->deleteStatement->execute([':namespace' => $this->namespace, ':cache_key' => $key]); + } + $this->connection->commit(); + + return true; + } catch (PDOException $exception) { + $this->rollBack(); + + throw $this->storageException('Unable to delete node SQLite cache keys.', $exception); + } + } + + public function getItem(string $key): GenericCacheItem + { + try { + $this->lookupStatement->execute([ + ':namespace' => $this->namespace, + ':cache_key' => $key, + ':current_time' => time(), + ]); + $row = $this->lookupStatement->fetch(); + } catch (PDOException $exception) { + throw $this->storageException("Unable to read node SQLite cache key '{$key}'.", $exception); + } + + if (!is_array($row) || !is_string($row['payload'] ?? null)) { + return new GenericCacheItem($this, $key); + } + + $record = CachePayloadCodec::decode($row['payload']); + if ($record === null || CachePayloadCodec::isExpired($record['expires'])) { + return new GenericCacheItem($this, $key); + } + + return new GenericCacheItem( + $this, + $key, + $record['value'], + true, + CachePayloadCodec::toDateTime($record['expires']), + ); + } + + public function hasItem(string $key): bool + { + return $this->getItem($key)->isHit(); + } + + /** + * @param array $keys The keys argument. + * @phpstan-param list $keys + * @phpstan-return array + */ + public function multiFetch(array $keys): array + { + return $this->multiFetchItems($keys, $this->getItem(...)); + } + + public function save(CacheItemInterface $item): bool + { + return $this->saveMany([$item]); + } + + /** + * @param array $items The items argument. + * @phpstan-param list $items + */ + public function saveMany(array $items): bool + { + foreach ($items as $item) { + if (!$this->supportsItem($item)) { + return false; + } + } + + if ($items === []) { + return true; + } + + try { + $this->connection->beginTransaction(); + foreach ($items as $item) { + $this->persistItem($item); + } + $this->connection->commit(); + + return true; + } catch (PDOException $exception) { + $this->rollBack(); + + throw $this->storageException('Unable to store node SQLite cache entries.', $exception); + } + } + + protected function supportsItem(CacheItemInterface $item): bool + { + return $item instanceof GenericCacheItem; + } + + private function createSchemaIfMissing(): void + { + try { + $this->connection->exec( + 'CREATE TABLE IF NOT EXISTS ' . self::TABLE . ' (' + . 'namespace TEXT NOT NULL, cache_key TEXT NOT NULL, payload BLOB NOT NULL, ' + . 'expires_at INTEGER, PRIMARY KEY (namespace, cache_key)) WITHOUT ROWID', + ); + $this->connection->exec( + 'CREATE INDEX IF NOT EXISTS cachelayer_node_entries_expiry_idx ' + . 'ON ' . self::TABLE . ' (namespace, expires_at)', + ); + } catch (PDOException $exception) { + throw $this->storageException('Unable to initialize the node SQLite cache schema.', $exception); + } + } + + private function persistItem(CacheItemInterface $item): void + { + $expires = CachePayloadCodec::expirationFromItem($item); + if ($expires['ttl'] === 0) { + $this->deleteStatement->execute([':namespace' => $this->namespace, ':cache_key' => $item->getKey()]); + + return; + } + + $this->upsertStatement->bindValue(':namespace', $this->namespace, PDO::PARAM_STR); + $this->upsertStatement->bindValue(':cache_key', $item->getKey(), PDO::PARAM_STR); + $this->upsertStatement->bindValue(':payload', CachePayloadCodec::encode($item->get(), $expires['expiresAt']), PDO::PARAM_LOB); + $this->upsertStatement->bindValue(':expires_at', $expires['expiresAt'], $expires['expiresAt'] === null ? PDO::PARAM_NULL : PDO::PARAM_INT); + $this->upsertStatement->execute(); + } + + private function rollBack(): void + { + if ($this->connection->inTransaction()) { + $this->connection->rollBack(); + } + } + + private function storageException(string $message, PDOException $exception): NodeCacheStorageException + { + return new NodeCacheStorageException($message, 0, $exception); + } +} diff --git a/src/Node/Connection/NodeSqliteConnection.php b/src/Node/Connection/NodeSqliteConnection.php new file mode 100644 index 0000000..243568c --- /dev/null +++ b/src/Node/Connection/NodeSqliteConnection.php @@ -0,0 +1,65 @@ +sqliteFile); + + try { + $connection = new PDO( + 'sqlite:' . $config->sqliteFile, + null, + null, + [ + PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC, + PDO::ATTR_EMULATE_PREPARES => false, + PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION, + PDO::ATTR_STRINGIFY_FETCHES => false, + ], + ); + $connection->exec('PRAGMA journal_mode = WAL'); + $connection->exec('PRAGMA synchronous = NORMAL'); + $connection->exec('PRAGMA busy_timeout = ' . $config->busyTimeoutMs); + $connection->exec('PRAGMA temp_store = MEMORY'); + } catch (PDOException $exception) { + throw new NodeCacheConfigurationException('Unable to initialize the node SQLite cache.', 0, $exception); + } + + return $connection; + } + + private static function prepareDirectory(string $file): void + { + if (is_link($file)) { + throw new NodeCacheConfigurationException("Refusing symlinked SQLite cache file: {$file}"); + } + + $directory = dirname($file); + if (is_link($directory)) { + throw new NodeCacheConfigurationException("Refusing symlinked SQLite cache directory: {$directory}"); + } + + if (!is_dir($directory) && !mkdir($directory, 0750, true) && !is_dir($directory)) { + throw new NodeCacheConfigurationException("Unable to create SQLite cache directory: {$directory}"); + } + + if (!is_writable($directory)) { + throw new NodeCacheConfigurationException("SQLite cache directory is not writable: {$directory}"); + } + + $permissions = fileperms($directory); + if ($permissions !== false && (($permissions & 0x0002) === 0x0002)) { + throw new NodeCacheConfigurationException("SQLite cache directory must not be world-writable: {$directory}"); + } + } +} diff --git a/src/Node/Exception/NodeCacheConfigurationException.php b/src/Node/Exception/NodeCacheConfigurationException.php new file mode 100644 index 0000000..3d423f0 --- /dev/null +++ b/src/Node/Exception/NodeCacheConfigurationException.php @@ -0,0 +1,7 @@ +connection->query('PRAGMA wal_checkpoint(PASSIVE)'); + } + + public function optimize(): void + { + $this->connection->exec('PRAGMA optimize'); + } + + public function pruneExpired(int $limit = 5_000): int + { + return $this->pruner->pruneExpired($limit); + } +} diff --git a/src/Node/Maintenance/NodeCachePruner.php b/src/Node/Maintenance/NodeCachePruner.php new file mode 100644 index 0000000..878538f --- /dev/null +++ b/src/Node/Maintenance/NodeCachePruner.php @@ -0,0 +1,42 @@ +connection->prepare( + <<<'SQL' + DELETE FROM cachelayer_node_entries + WHERE (namespace, cache_key) IN ( + SELECT namespace, cache_key + FROM cachelayer_node_entries + WHERE namespace = :namespace + AND expires_at IS NOT NULL + AND expires_at <= :current_time + LIMIT :limit + ) + SQL, + ); + $statement->bindValue(':namespace', $this->namespace, PDO::PARAM_STR); + $statement->bindValue(':current_time', time(), PDO::PARAM_INT); + $statement->bindValue(':limit', $limit, PDO::PARAM_INT); + $statement->execute(); + + return $statement->rowCount(); + } +} diff --git a/src/Node/NodeCache.php b/src/Node/NodeCache.php new file mode 100644 index 0000000..815eff5 --- /dev/null +++ b/src/Node/NodeCache.php @@ -0,0 +1,56 @@ +namespace), + $config->failOpen, + $metrics, + ); + + return new Cache( + $adapter, + new FileLockProvider($config->lockDirectory), + $metrics, + ); + } + + public static function maintenance(NodeCacheConfig $config): NodeCacheMaintenance + { + $connection = NodeSqliteConnection::create($config); + $adapter = new NodeSqliteCacheAdapter($connection, $config->namespace); + + return new NodeCacheMaintenance( + $adapter->connection(), + new NodeCachePruner($adapter->connection(), $config->namespace), + ); + } + + private static function createApcuAdapter(NodeCacheConfig $config): ?ApcuCacheAdapter + { + if (!$config->apcuEnabled || !extension_loaded('apcu') || !apcu_enabled()) { + return null; + } + + return new ApcuCacheAdapter($config->namespace); + } +} diff --git a/src/Node/NodeCacheConfig.php b/src/Node/NodeCacheConfig.php new file mode 100644 index 0000000..bf68e02 --- /dev/null +++ b/src/Node/NodeCacheConfig.php @@ -0,0 +1,39 @@ +namespace = sanitize_cache_ns($namespace); + } +} diff --git a/tests/Cluster/ClusterCacheTest.php b/tests/Cluster/ClusterCacheTest.php new file mode 100644 index 0000000..803e3e1 --- /dev/null +++ b/tests/Cluster/ClusterCacheTest.php @@ -0,0 +1,112 @@ +clusterDirectory = sys_get_temp_dir() . '/cachelayer-cluster-' . uniqid(); + $this->transport = new InMemoryInvalidationTransport(); + $this->clusterConfigA = new ClusterCacheConfig('test-cluster', 'node-a'); + $this->clusterConfigB = new ClusterCacheConfig('test-cluster', 'node-b'); + $this->nodeConfigA = new NodeCacheConfig( + $this->clusterDirectory . '/node-a.sqlite', + 'application', + apcuEnabled: false, + ); + $this->nodeConfigB = new NodeCacheConfig( + $this->clusterDirectory . '/node-b.sqlite', + 'application', + apcuEnabled: false, + ); + $this->nodeA = ClusterCache::create($this->nodeConfigA, $this->clusterConfigA, $this->transport); + $this->nodeB = ClusterCache::create($this->nodeConfigB, $this->clusterConfigB, $this->transport); +}); + +afterEach(function () { + if (!is_dir($this->clusterDirectory)) { + return; + } + + $files = new RecursiveIteratorIterator( + new RecursiveDirectoryIterator($this->clusterDirectory, FilesystemIterator::SKIP_DOTS), + RecursiveIteratorIterator::CHILD_FIRST, + ); + foreach ($files as $file) { + $file->isDir() ? rmdir($file->getPathname()) : unlink($file->getPathname()); + } + rmdir($this->clusterDirectory); +}); + +test('cluster key, tag, and namespace invalidations are replayed to another node', function () { + $this->nodeA->cache()->set('product.42', 'A', 300); + $this->nodeB->cache()->set('product.42', 'B', 300); + + $this->nodeA->invalidateKey('product.42'); + + expect($this->nodeA->cache()->get('product.42'))->toBeNull() + ->and($this->nodeB->cache()->get('product.42'))->toBe('B') + ->and($this->nodeB->consume())->toBe(1) + ->and($this->nodeB->cache()->get('product.42'))->toBeNull(); + + $this->nodeA->cache()->setTagged('product.43', 'A', ['products'], 300); + $this->nodeB->cache()->setTagged('product.43', 'B', ['products'], 300); + $this->nodeA->invalidateTag('products'); + + expect($this->nodeB->consume())->toBe(1) + ->and($this->nodeB->cache()->get('product.43'))->toBeNull(); + + $this->nodeA->cache()->set('settings', 'A', 300); + $this->nodeB->cache()->set('settings', 'B', 300); + $this->nodeA->clearNamespace(); + + expect($this->nodeB->consume())->toBe(1) + ->and($this->nodeB->cache()->get('settings'))->toBeNull(); +}); + +test('recovery clears a local node when its cursor predates retained events', 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')); + $this->nodeB->consume(1); + $this->nodeB->cache()->set('stale', 'value', 300); + $this->transport->discardBefore('test-cluster', 3); + + expect($this->nodeB->recoverIfRequired())->toBeTrue() + ->and($this->nodeB->cache()->get('stale'))->toBeNull() + ->and($this->nodeB->recoverIfRequired())->toBeFalse(); +}); + +test('the origin node advances its cursor without replaying its own invalidation', function () { + $this->nodeA->cache()->set('origin.only', 'value', 300); + $this->nodeA->invalidateKey('origin.only'); + + expect($this->nodeA->consume())->toBe(1) + ->and($this->nodeA->cache()->get('origin.only'))->toBeNull() + ->and($this->nodeA->consume())->toBe(0); +}); + +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); + $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->isCursorBefore($first, $second))->toBeTrue() + ->and($transport->pruneBefore(time() + 1, 1))->toBe(1) + ->and($transport->oldestAvailableId('pdo-cluster'))->toBe($second); + + $connection->beginTransaction(); + $transport->publishWithinTransaction( + $connection, + InvalidationEvent::namespace('pdo-cluster', 'application', 'writer'), + ); + $connection->rollBack(); + + expect($transport->consumeAfter('pdo-cluster', $second, 10)->events)->toBe([]); +}); diff --git a/tests/Cluster/Support/InMemoryInvalidationTransport.php b/tests/Cluster/Support/InMemoryInvalidationTransport.php new file mode 100644 index 0000000..5f2a3b2 --- /dev/null +++ b/tests/Cluster/Support/InMemoryInvalidationTransport.php @@ -0,0 +1,54 @@ +> */ + private array $events = []; + + private int $nextId = 1; + + public function consumeAfter(string $cluster, ?string $cursor, int $limit): InvalidationBatch + { + $after = $cursor === null ? 0 : (int) $cursor; + $events = array_filter( + $this->events[$cluster] ?? [], + static fn(InvalidationEvent $event): bool => (int) $event->id > $after, + ); + + return new InvalidationBatch(array_values(array_slice($events, 0, $limit))); + } + + public function discardBefore(string $cluster, int $eventId): void + { + $this->events[$cluster] = array_values(array_filter( + $this->events[$cluster] ?? [], + static fn(InvalidationEvent $event): bool => (int) $event->id >= $eventId, + )); + } + + public function isCursorBefore(string $cursor, string $oldestAvailableId): bool + { + return (int) $cursor < (int) $oldestAvailableId; + } + + public function oldestAvailableId(string $cluster): ?string + { + return $this->events[$cluster][0]->id ?? null; + } + + public function publish(InvalidationEvent $event): string + { + $id = (string) $this->nextId++; + $this->events[$event->cluster][] = $event->withId($id); + + return $id; + } +} diff --git a/tests/Node/NodeCacheTest.php b/tests/Node/NodeCacheTest.php new file mode 100644 index 0000000..49e2a06 --- /dev/null +++ b/tests/Node/NodeCacheTest.php @@ -0,0 +1,98 @@ +nodeCacheDirectory = sys_get_temp_dir() . '/cachelayer-node-' . uniqid(); + $this->nodeCacheFile = $this->nodeCacheDirectory . '/cache.sqlite'; + $this->nodeConfig = new NodeCacheConfig( + sqliteFile: $this->nodeCacheFile, + namespace: 'node.tests', + apcuEnabled: false, + lockDirectory: $this->nodeCacheDirectory . '/locks', + ); +}); + +afterEach(function () { + if (!is_dir($this->nodeCacheDirectory)) { + return; + } + + $files = new RecursiveIteratorIterator( + new RecursiveDirectoryIterator($this->nodeCacheDirectory, FilesystemIterator::SKIP_DOTS), + RecursiveIteratorIterator::CHILD_FIRST, + ); + foreach ($files as $file) { + $file->isDir() ? rmdir($file->getPathname()) : unlink($file->getPathname()); + } + rmdir($this->nodeCacheDirectory); +}); + +test('node cache factory provides SQLite-backed facade behavior and local tags', function () { + $cache = NodeCache::create($this->nodeConfig); + + expect($cache->setTagged('user.42', ['name' => 'Ada'], ['users'], 300))->toBeTrue() + ->and($cache->get('user.42'))->toBe(['name' => 'Ada']) + ->and($cache->invalidateTag('users'))->toBeTrue() + ->and($cache->get('user.42'))->toBeNull(); +}); + +test('node cache supports deferred writes through its composed PSR-6 item', function () { + $cache = NodeCache::create($this->nodeConfig); + $item = $cache->getItem('deferred')->set('value')->expiresAfter(300); + + expect($cache->saveDeferred($item))->toBeTrue() + ->and($cache->commit())->toBeTrue() + ->and($cache->get('deferred'))->toBe('value'); +}); + +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); + $l2 = new NodeSqliteCacheAdapter($connection, $this->nodeConfig->namespace); + $cache = new Cache(new NodeCacheAdapter($l1, $l2, false)); + + expect($cache->set('promoted', 'value', 300))->toBeTrue(); + $l1->clear(); + + $item = $cache->getItem('promoted'); + + expect($item->isHit())->toBeTrue() + ->and($item->get())->toBe('value') + ->and($l1->getItem('promoted')->isHit())->toBeTrue(); + + $item->set('changed'); + expect($cache->save($item))->toBeTrue() + ->and($l2->getItem('promoted')->get())->toBe('changed'); +}); + +test('expired rows remain outside the read path until bounded pruning', function () { + $connection = NodeSqliteConnection::create($this->nodeConfig); + $adapter = new NodeSqliteCacheAdapter($connection, $this->nodeConfig->namespace); + $statement = $connection->prepare( + 'INSERT INTO cachelayer_node_entries (namespace, cache_key, payload, expires_at) VALUES (?, ?, ?, ?)', + ); + $statement->execute([$this->nodeConfig->namespace, 'stale', 'invalid-payload', time() - 1]); + + expect($adapter->getItem('stale')->isHit())->toBeFalse() + ->and((int) $connection->query("SELECT COUNT(*) FROM cachelayer_node_entries WHERE cache_key = 'stale'")->fetchColumn()) + ->toBe(1) + ->and((new NodeCachePruner($connection, $this->nodeConfig->namespace))->pruneExpired(1)) + ->toBe(1) + ->and((int) $connection->query("SELECT COUNT(*) FROM cachelayer_node_entries WHERE cache_key = 'stale'")->fetchColumn()) + ->toBe(0); +}); + +test('node cache configuration rejects invalid paths and timeouts', function () { + expect(fn () => new NodeCacheConfig('', 'app'))->toThrow(NodeCacheConfigurationException::class) + ->and(fn () => new NodeCacheConfig('/tmp/cache.sqlite', 'app', busyTimeoutMs: -1)) + ->toThrow(NodeCacheConfigurationException::class); +});