refactor(graphql): separate subscription registry and publication state

Store item payload fingerprints separately from registrations and coordinate mutations with an injected Symfony Lock factory. Keep collection publications untracked and acknowledge prepared updates only after successful publication or dispatch.

Remove legacy subscription-manager adapters and document cache reset and shared-store configuration requirements.

BREAKING CHANGE: Custom subscription managers must implement operation-aware registration, getUpdates(), and acknowledge().

Author: psihius

composer.json

@@ -112,9 +112,11 @@
         "doctrine/inflector": "^2.0",
         "psr/cache": "^1.0 || ^2.0 || ^3.0",
         "psr/container": "^1.0 || ^2.0",
+        "psr/log": "^1.0 || ^2.0 || ^3.0",
         "symfony/deprecation-contracts": "^3.1",
         "symfony/http-foundation": "^6.4.14 || ^7.0 || ^8.0",
         "symfony/http-kernel": "^6.4 || ^7.0 || ^8.0",
+        "symfony/lock": "^6.4 || ^7.0 || ^8.0",
         "symfony/property-access": "^6.4 || ^7.0 || ^8.0",
         "symfony/property-info": "^6.4 || ^7.1 || ^8.0",
         "symfony/serializer": "^6.4 || ^7.0 || ^8.0",
@@ -157,7 +159,6 @@
         "phpstan/phpstan-phpunit": "^2.0",
         "phpstan/phpstan-symfony": "^2.0",
         "phpunit/phpunit": "^12.2",
-        "psr/log": "^1.0 || ^2.0 || ^3.0",
         "ramsey/uuid": "^4.7",
         "ramsey/uuid-doctrine": "^2.0",
         "soyuka/contexts": "^3.3.10",

src/GraphQl/README.md

@@ -10,3 +10,63 @@ The [GraphQL](https://graphql.org/) component of the [API Platform](https://api-
 > [report issues](https://github.com/api-platform/core/issues) and
 > [send Pull Requests](https://github.com/api-platform/core/pulls)
 > in the [core API Platform repository](https://github.com/api-platform/core).
+
+## Subscription state and publication
+
+The subscription registry stores IDs and selected fields. Individual-item payloads
+are represented by SHA-256 fingerprints in separate cache entries; collection
+subscriptions do not retain payloads or fingerprints. IDs and Mercure topics are
+unchanged.
+
+Symfony wires two cache pools:
+
+- `api_platform.graphql.cache.subscription`: the registration registry;
+- `api_platform.graphql.cache.subscription_fingerprint`: fingerprints, inheriting
+  the registry pool's adapter configuration by default.
+
+`SubscriptionStore` receives these pools and a Symfony `LockFactory` through
+injection. The `api_platform.graphql.subscription.lock_factory` alias points to
+Symfony's `lock.factory`, configured through `framework.lock`. It can instead
+reference a named lock factory when subscriptions need their own lock backend.
+API Platform does not construct or select a lock store. Query-only applications
+can leave locking disabled; registering subscriptions requires a configured
+factory and fails explicitly if it is missing. There is no unlocked fallback.
+
+When sharing subscription caches between servers, configure a shared lock backend
+(such as Redis) accessible to every registration and publication worker. Both cache
+pools must also use shared, coherent storage. Cache pools expose no portable way
+to derive their connection for the Lock component; configure the cache and lock
+backends together. Local locks are suitable only when all writers share the same
+local storage and lock scope.
+
+Registration and deletion lock their registry bucket. Successful item publications
+update their fingerprint under a separate per-subscription lock. Normalization,
+hashing, and Mercure/Messenger calls occur outside those locks. Versioned
+comparison detects fingerprints changed in the meantime. Conflicting completions
+invalidate the fingerprint so the next event is not incorrectly suppressed.
+Fingerprint write failures are logged and the stale fingerprint is discarded on
+a best-effort basis so delivery continues to other subscribers. Delete cleanup
+failures are also logged without dropping the delete recipients. Cache failures
+can leave stale entries if invalidation also fails; restore cache health and clear
+subscription state when necessary.
+
+Concurrent identical publications can still occur; this is change suppression,
+not an exactly-once delivery or ordering guarantee.
+
+The Doctrine publisher iterates `SubscriptionManagerInterface::getUpdates()`,
+publishes each prepared update, then explicitly calls `acknowledge()` for that update.
+The manager handles normalization; the store owns fingerprint comparison and
+acknowledgement. Preparing or iterating updates does not advance their fingerprints.
+A fingerprint advances only after successful hub acceptance for synchronous delivery
+or Messenger dispatch acceptance for asynchronous delivery. Worker retries remain
+Messenger's responsibility; acceptance does not imply client delivery.
+
+Custom subscription managers must implement the complete `SubscriptionManagerInterface`:
+registration receives the subscription operation, `getUpdates()` prepares updates
+for that operation, and `acknowledge()` records successful publication. The previous
+`getPushPayloads()` contract is removed.
+
+The cache layout is internal and has no migration path. When upgrading, stop old
+registration and publication workers, clear subscription state, and re-establish
+subscriptions. Old and new workers must not share this registry. Fingerprint cache
+eviction can cause an additional publication without losing registrations.

src/GraphQl/State/Processor/SubscriptionProcessor.php

@@ -14,7 +14,6 @@
 namespace ApiPlatform\GraphQl\State\Processor;
 
 use ApiPlatform\GraphQl\Subscription\MercureSubscriptionIriGeneratorInterface;
-use ApiPlatform\GraphQl\Subscription\OperationAwareSubscriptionManagerInterface;
 use ApiPlatform\GraphQl\Subscription\SubscriptionManagerInterface;
 use ApiPlatform\Metadata\GraphQl\Subscription;
 use ApiPlatform\Metadata\Operation;
@@ -41,11 +40,7 @@ public function process(mixed $data, Operation $operation, array $uriVariables =
             return $data;
         }
 
-        if ($this->subscriptionManager instanceof OperationAwareSubscriptionManagerInterface) {
-            $subscriptionId = $this->subscriptionManager->retrieveSubscriptionId($context, $data, $operation);
-        } else {
-            $subscriptionId = $this->subscriptionManager->retrieveSubscriptionId($context, $data);
-        }
+        $subscriptionId = $this->subscriptionManager->retrieveSubscriptionId($context, $data, $operation);
 
         if ($subscriptionId) {
             if (!$this->mercureSubscriptionIriGenerator) {

src/GraphQl/Subscription/OperationAwareSubscriptionManagerInterface.php

@@ -0,0 +1,27 @@
+<?php
+
+/*
+ * This file is part of the API Platform project.
+ *
+ * (c) Kévin Dunglas <dunglas@gmail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+declare(strict_types=1);
+
+namespace ApiPlatform\GraphQl\Subscription;
+
+/**
+ * A registration and its publication state at the time it was read.
+ *
+ * @internal
+ */
+final readonly class RegisteredSubscription
+{
+    /** @param array{hash: string, version: string}|null $fingerprint */
+    public function __construct(public string $key, public string $id, public array $fields, public bool $collection, public ?array $fingerprint)
+    {
+    }
+}