Repository navigation
Conversation
soyuka
left a comment
There was a problem hiding this comment.
- Private fields config option probably should also use expression language same way topics do
Not sure to understand this functionality, did I miss something?
- Caching item of the subscriptionCollection is going to become quite sizeable when you have a lot of segmentation based on owners of the data and I don't have a good idea how to segment that
I don't think we should send collection updates, only items inside a collection should be sufficient.
When a data is deleted you can still send an update on its IRI (as mercure topic) with a null value? I'm not sure why data should be part of a cache key, instead we should use IRIs just as in REST? Or is it that you want to send a mercure update with only the fields selected in the GraphQl Query?
- Mercure payload format probably needs updating - with it now containing all 3 types of operations, I think it needs a field that contains what type of operation it was - create, update or deleted so mercure consumers can decide what to do
IMO, we should always consider a mercure update as a modified element, it's a new one if you don't have it in your referential, or else it's updated. You can detect it's deleted when the value is null.
- Deleted stdObject contains too little information, kind'a want ability to retain some of the data in there, like for example the parent relation field that points to the parent ID the record was deleted for.
Not sure why we would need this. It makes things really complicated when we look for data associations, when do you stop looking for relations?
https://api-platform.com/docs/core/mercure/#dispatching-restrictive-updates-security-mode - this, but for the graphql subscription. It's a way to restrict data pushing to specific user/resource/whatever way you want to restrict access based on resource field values. I use it to restrict access between accounts in SaaS service.
As you can see here, it first tries to get subscriptions from full IRI. If it's an update operation - the item is already in cache, if not - it goes to collection cache (which would happen only for create operation).
See that second foreach there? It filters out based on field combo subscriptions that are not addressed to the specific combo of fields. The _private_field$field item makes sure with it's value that you send updates to a subscription id that is restricted to the same field value. IF you use There's the corresponding change in SubscriptionManager::retrieveSubscriptionId which gets called in SubscriptionProcessor which is where the subscription id is created for a given IRI, that subscription id is created from fields that were selected as return data for the subscription. In there I also inject those private field and their data from the resource in question, which affects the value of generated subscription id. The subscription id always is generated as a resource wide subscription, it's the same id regardless what IRI you pass as long as it's the same resource type. The cache just was stored per resource IRI (aka per item), so I just added also the collection level cache item so you can push updates for "create" operation, since that adds a new item and you need to know what subscriptions you need to push that update into.
I did modify the output for the "delete" operation to contain
|
|
This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions. |
|
I definitely need time to review this I think it's still something that we should integrate. |
Feel free to reach out on Symfony Slack into my DM's or tag me on #api-platform channel about this whole thing. I will be able to do changes/fixes to this whole thing next week cause I think we found a small bug to fix, so I will be touching the code :) |
ead72e1 to
c9cefd8
Compare
|
I found some flaws in my previous implementation, split out completely the handling of collection updates and so that collection subscriptions generate their own subscription ID instead of piggybacking on individual item subscriptions. I've also added relationship handling for mercure updates, because those have never been handled so you could not any -to-many relationship data on an update, even if you asked for it in the GraphQL query. Nothing complicated in there, but it works and will allow people to have simple related payloads to be included with the mercure update payload. There is more I would do, but it would require some major amount of time and effort that I simply just don't have: I would split out mercure update pipeline completely from graphql processing pipeline because part of GraphQL library work is happening at the response time once it;s out of ApiPlatform's graphql internals and as you can see from my latest commit, there are some things that have been missing because mercure updates happen as a side channel as a doctrine event listener. |
decfc5b to
9e61e92
Compare
|
don't hesitate to add functional tests to help me understand the needs behind that, there are tests inside |
Yeah, I've been slowly working on it. In other news, my last commit is more of a temporary fix for our production system, but the cache pool that's used there right now needs to be replaced. The system cache pool does not have locking and protection from parallel writing, which under 20+ users, started to overwrite each other's changes. I've also added private field data to deleted object data in the doctrine listener, probably same needs to happen in ODM listener. I needed to read deleted item fields and pass them on to the subscription manager. |
ef602bf to
cea37ca
Compare
|
Thanks for the review, I will adjust to feedback. I do have a question about collections, if I implement CollectionOperationInterface for the collection subscription, I assume the system will allow operating on collections via the reference to the resource aka I would like some references what to read and what files to dig into to explore proper collection subscription implementation and what helpers and components do I need to look at, since I'm not very familiar with broader internals. Like I did not know and would not find IdentifiersExtractor without the suggestion for quite a while if I continued to dig on my own :) |
0e1eb28 to
6d97b96
Compare
|
Performed a rework of the feature based on feedback and did some internal cache related changes that changed how this works quite a bit. Documentation PR: api-platform/docs#2267 The biggest change is that collection subscriptions are now modeled as their own GraphQL operation type via Current behavior
{
"type": "delete",
"payload": {
"id": "/books/1",
"iri": "https://example.com/books/1",
"type": "Book"
}
}I kept the delete payload explicit on purpose. For item subscriptions you could argue Internal changesThe old collection IRI derivation hack is gone. Collection subscription lookup is now based on resource metadata +
This is meant for cases where a client is authenticated and allowed to see multiple scopes, but subscriptions still need to be isolated by something like organization / workspace / inbox / whatever the app defines. This is now validated explicitly, the extraction logic was centralized, and it now uses Symfony property access instead of raw getter assumptions. Cache behavior changed quite a bit too.
This makes the internal cache model much more predictable and fixes a few edge cases from the earlier implementation, especially around private subscriptions and collection notifications. I also addressed the main review points around:
TestingTesting is much broader now too. The feature is covered in:
|
f18a3ae to
ffeb112
Compare
Use the registering collection operation when looking up subscriptions for create, update and delete events. Avoid selecting a collection query or only the first collection subscription, and keep item lookups limited to item subscription operations. Cover multiple collection subscriptions and HTTP registration followed by publication for all three event types.
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(). Signed-off-by: psihius <arvids.godjuks@gmail.com>
2db0462 to
cb1f7cf
Compare
Implementation walkthroughThis PR extends GraphQL Mercure subscriptions to cover item deletion and collection events, and reworks registration and publication to support multiple operations, private partitions, and concurrent workers.
A collection subscription sends the changed item, not the entire collection. It does not store a growing collection snapshot. The important distinction throughout the implementation is that metadata determines which operations may publish; cached registrations determine which subscriptions receive those publications. The registry does not replace operation metadata. 1. A client registers its selected fieldsFor example, a resource can expose both operation kinds: #[ApiResource(
mercure: true,
graphQlOperations: [
new Subscription(name: 'watch'),
new SubscriptionCollection(name: 'watchCollection'),
],
)]The client submits a GraphQL subscription with its field selection and receives the initial result and a SubscriptionProcessor resolves the operation's Mercure options and calls SubscriptionManager::retrieveSubscriptionId(). The manager builds three distinct pieces of identity:
SubscriptionStore::register() stores only IDs and selections in the registry. Conceptually: Operation metadata, delivery settings, and complete payloads are not stored in these registry entries. 2. Doctrine changes are matched against resource metadataDuring collectGraphQlPublications() still loops over the resource's declared GraphQL operations. It selects The listener collects the applicable operations into one batch per changed object. For deletion, it captures the IRI, resource type, and private partition values before the entity disappears. REST publication continues without GraphQL subscriptions. GraphQL publication can likewise remain enabled when REST publication is disabled. Each path uses its own operation's configuration, avoiding HTTP metadata overriding a subscription's privacy or delivery settings. 3. Operations share cache reads, while retaining their own behaviorAfter the flush, SubscriptionManager::getUpdates() groups the supplied operations by bucket key, reads each bucket once, and matches registrations to the eligible operations: $subscriptions = $this->store->getSubscriptions($key, $collection);
foreach ($operations as $name => ['object' => $object, 'operation' => $operation]) {
foreach ($subscriptions[$name] ?? [] as $subscription) {
// Normalize the selected fields using this operation.
// Prepare an update if publication is needed.
}
}For example, three item operations sharing an item and private partition need one registry read, rather than three. If only two have registrations, only those two perform normalization and prepare updates. A collection bucket requires a separate read; different private partitions also remain separate. An eligible operation without registrations still undergoes metadata evaluation and grouping, but produces no payload. A cached operation removed from metadata, or disabled for this event, produces no publication. The actual subscription operation is passed through normalization, preserving its normalization settings. Subscription behavior is identified by operation type rather than magic operation names. ItemNormalizer also retains subscription context when normalizing selected nested relationships. 4. Prepared updates are delivered, then acknowledgedThe manager yields each prepared update together with the operation supplied by the listener. postFlush() processes deletes first, then creates, then updates. publishUpdates() uses each operation's resolved options to publish synchronously or dispatch through Messenger. Create/update notifications are acknowledged only after successful publication or dispatch; delete notifications are acknowledged after every delivery attempt, including a failed one. GraphQL updates use the topic generated from the subscription ID. There is no need to configure Create/update payloads retain the selected GraphQL response structure. Deletes use a small envelope: {
"type": "delete",
"payload": {
"id": "/messages/42",
"iri": "https://example.com/messages/42",
"type": "Message"
}
}Deletion cleanup and delivery failuresOnce Doctrine has successfully flushed a deletion, its item registrations must be retired even if Mercure delivery fails. The flow is:
For example, if two deleted items have active subscriptions and the first item's hub publication fails, the second item's buckets are still cleaned up and its delete notifications are attempted before the original error is rethrown. Doctrine does not replay a flushed deletion after delivery failure, so retaining item registrations would not provide a retry mechanism. A delete Private partitions and Mercure permissionsThis provides a GraphQL counterpart to the documented Dispatching Restrictive Updates (Security Mode). In the REST example, the publisher adds an owner-specific alternate topic, and the subscriber's JWT authorizes an appropriate topic selector. This scopes delivery to the resources that subscriber may receive. GraphQL already publishes to generated subscription topics. Here, the configured private field values become part of the subscription identity itself, so different partitions produce different subscription IDs and therefore different Mercure topics—even when the operation and selected fields are identical. The scope is encoded into the generated topic rather than expressed through a manually configured alternate topic. new SubscriptionCollection(
name: 'watchMessages',
mercure: [
'private' => true,
'private_fields' => ['tenant', 'chat'],
],
)For the same resource, operation, and selection
When a message in tenant 7/chat 12 changes, API Platform calculates that partition's bucket directly and publishes to its registered subscription topics. It does not scan other tenants/chats or publish that event to B or C. Different field selections within the same partition can still produce additional subscription IDs. The two settings have complementary roles: PropertyAccessorValueExtractor reads values through PropertyAccess and resolves related resources using their identifier metadata rather than assuming As with the documented restrictive-update approach, application authorization and Mercure subscriber permissions must match the intended access scope. The application still controls who may register and which generated topics their Mercure credentials authorize; partition hashing does not grant or revoke those permissions. Payload suppression and concurrencyPreviously, complete item payloads lived alongside registrations. The store now separates registry entries from per-subscription SHA-256 fingerprints. Collections retain neither payloads nor fingerprints. prepareUpdate() suppresses unchanged item payloads. For example, changing an unselected field does not publish if the selected result remains identical. Initial item results seed the fingerprint; preparing subsequent updates does not advance it. The fingerprint advances only after successful hub acceptance or Messenger dispatch acceptance. A failed item publication therefore remains eligible if a later event or explicit replay prepares it again; this does not schedule an automatic retry. Dispatch acceptance does not imply client delivery; Messenger handles subsequent retries of accepted messages. Registry registration/deletion uses a bucket lock. Fingerprint acknowledgement uses a separate per-subscription lock; normalization, hashing, and network publication happen outside those locks. Versioned comparison detects overlapping publications and invalidates conflicting suppression state so a later correction is not skipped. This is change suppression, not an exactly-once delivery or ordering guarantee. The Symfony Naming, migration, and verificationUnnamed collection subscriptions use Custom managers must implement the updated SubscriptionManagerInterface: operation-aware registration, batched The cache layout and generated IDs/topics change. Upgrading requires stopping old workers, clearing subscription state, and re-establishing subscriptions. Old and new workers must not share the registry. The focused local checks passed 165 tests / 1,197 assertions, with deprecation notices and PHPUnit notices. Formatting and PHP syntax checks passed. The scoped PHPStan run reported two missing MongoDB ODM class errors. These local results do not represent a complete hosted CI run. Coverage includes naming/schema behavior, identity and private partitions, normalization, independent REST/GraphQL delivery, hubs, failed publication, and concurrency. The shared-bucket regression tests verify read counts, distinct short-name/operation identities, partition isolation, and deletion cleanup; the functional publication test verifies the cache-read reduction through the Doctrine listener. The delete delivery regression tests cover synchronous and Messenger dispatch failures, cleanup across subsequent items and private partitions, preservation of the first delivery error, and acknowledgement after failed deletes while failed creates/updates remain unacknowledged. |
This PR makes create and delete operations issue mercure updates for GraphQL subscriptions.
It also adds the ability to configure "restrictive updates" - https://api-platform.com/docs/core/mercure/#dispatching-restrictive-updates-security-mode - implementation follows the same idea, just one that fits GraphQL subscriptions.
Some things that need feedback/work/ideas:
This is a WIP because I need feedback on the direction taken and suggestions for improving some aspects of it.
This code is being used in production, works for our use case, but it's obviously far from ideal and I want to improve it and make it standard part of ApiPlatform.
The goal for this is to make managing items on frontend easy purely via mercure updates via graphql subscriptions. In our case it's a chat like frontend that needs to not only receive updates, but also new data (new messages, new chats) and data removal messages. This puts GraphQL API on par with functionality of the REST API where you can have restrictive data updates AND you can subscribe to collections which include all 3 types of data events.