[graphql-stream] Add Address.asTransactionObject [12/n] - #26495
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
amnn
left a comment
There was a problem hiding this comment.
I was expecting to see the scope update in code related to Transaction/TransactionEffects/Event but I'm not seeing that -- where is the scope being set?
Could you also add tests for the non-subscription case?
| return Ok(None); | ||
| }; | ||
|
|
||
| let contents = EffectsContents::empty(self.scope.clone()) |
There was a problem hiding this comment.
Is it possible to cache the whole TransactionEffects in the scope, so that in case you have already lazily fetched the contents, you don't need to refetch it?
There was a problem hiding this comment.
The TransactionContents is already cached on Scope, and that's what streamed_transaction_by_digest returns (see the comment above), backed by the ProcessedCheckpoint payload Scope::for_streamed_checkpoint puts there.
However, there is actually an inefficiency though: ProcessedTransaction.contents was being deep-cloned per subscriber in . I just updated to wrap it in Arc so the deep clone happens once at ingestion.
There was a problem hiding this comment.
That's true for streaming, but not for queries.
If instead of tx_digest_viewed_at we had Scope::effects_viewed_at(&self): TransactionEffects, we could still instantiate that using just a digest to create an instance whose contents were not fetched yet, but if we already have TransactionEffects in our hand, we can reuse that, which also means re-using its potentially cached contents.
- In the subscriptions case, we would instantiate the
effects_viewed_atusing just the digest (as you are doing now), and if/when its contents are fetched, it will look in theScope's data source first (note that this is not a cache). - In the query case, if
asTransactionObjectwas nested under some effects, then we would already have the transaction and its contents in our hands, and we would initialize the scope withselfin the context ofScope. - In either case, if
asTransactionObjectwas nested under an event, we would again initialise the effects being viewed with just a digest (if they are not already set), because we may not have fetched the event's transaction.
There was a problem hiding this comment.
Yeah makes sense! I anchored Scope at the Transaction level. Every Transaction and TransactionEffects constructor (with_contents, fetch, from_executed_transaction) sets Scope::active_transaction, so descendants like Address.asTransactionObject resolve consistently regardless of entry point.Scope::with_active_transaction_digest preserves an existing same-digest anchor, so re-entering the same transaction (e.g. via Event::transaction) doesn't drop hydrated contents.
941ea01 to
1042ab2
Compare
0d6799b to
cf16c47
Compare
## Summary Replace the WebSocket subscription transport with Server-Sent Events. SSE is the standard transport for GraphQL subscriptions over HTTP. WebSocket was added in 1/n (#26019) as a starting point and is not in production yet, so this PR removes it outright. ## Why For one-way streaming (server pushes events, client mostly listens — exactly our subscription shape), SSE is closer to current industry best practice than WebSocket: simpler infra (no upgrade dance, plays naturally with HTTP/2/3, standard HTTP caching/proxy semantics), debuggable with curl, native browser support. The same pattern is used by OpenAI/Anthropic completion streaming, Cloudflare Workers AI, GitHub live activity, Vercel AI SDK, etc. ## GraphiQL story Upstream GraphiQL has no SSE fetcher today. Rather than block on the in-flight upstream PR (graphql/graphiql#4218), this PR ships a small static HTML template (`assets/graphiql.html`) that loads GraphiQL via UMD plus the `graphql-sse` UMD client and wires a custom fetcher: subscriptions go through `graphqlSse.iterate()`, queries through `fetch`. CDN deps pinned; subscription detection uses `parse` + `getOperationAST` so it handles multi-operation documents with an explicit `operationName`. Once #4218 lands, this collapses to a one-line `createGraphiQLFetcher({ url, sseUrl })` call. ## Test plan How did you test the new or updated feature? - unit + e2e tests - Start local GraphQL Server ## Stack - #26019 - #26094 - #26117 - #26170 - #26194 - #26202 - #26414 - #26453 - #26476 - #26487 - #26495 - #26425 --- ## Release notes Check each box that your changes affect. If none of the boxes relate to your changes, release notes aren't required. For each box you select, include information after the relevant heading that describes the impact of your changes that a user might notice and any actions they must take to implement updates. - [ ] Protocol: - [ ] Nodes (Validators and Full nodes): - [ ] gRPC: - [ ] JSON-RPC: - [ ] GraphQL: - [ ] CLI: - [ ] Rust SDK: - [ ] Indexing Framework: Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
cf16c47 to
b1fd018
Compare
There was a problem hiding this comment.
Thanks @amnn! For your points
I was expecting to see the scope update in code related to Transaction/TransactionEffects/Event but I'm not seeing that -- where is the scope being set?
The scope plumbing landed in #26487 (10/n) — that PR added Scope::streamed_transaction_by_digest plus the streaming fast paths in TransactionContents::fetch / EffectsContents::fetch.
Concretely, the flow in this PR for an event-subscription query that hits asTransactionObject is:
- as_transaction_object resolves the target digest
- EffectsContents::empty(scope).fetch(ctx, digest)
- scope.streamed_transaction_by_digest(digest) (O(1) HashMap lookup in the cached ProcessedCheckpoint) -> contents = Arc::new(tx.contents.clone()) // in-memory only, no DB
- content.effects() — deserializes TransactionEffects from the cached BCS.
Could you also add tests for the non-subscription case?
Sure added.
| return Ok(None); | ||
| }; | ||
|
|
||
| let contents = EffectsContents::empty(self.scope.clone()) |
There was a problem hiding this comment.
The TransactionContents is already cached on Scope, and that's what streamed_transaction_by_digest returns (see the comment above), backed by the ProcessedCheckpoint payload Scope::for_streamed_checkpoint puts there.
However, there is actually an inefficiency though: ProcessedTransaction.contents was being deep-cloned per subscriber in . I just updated to wrap it in Arc so the deep clone happens once at ingestion.
Streaming consumers want to take an object ID surfaced in an event payload and ask "how was this object referenced by the transaction that emitted the event?" without an extra round trip to the indexed Query API. The TL flagged that the relevant object is sometimes an unchanged shared input (e.g. a clock or registry), so it never appears in `objectChanges`; the resolver therefore needs to expose both the change case and the read-only consensus input case. This PR adds `Address.asTransactionObject(transactionDigest)` returning a new `TransactionObjectRef` union of `ObjectChange | ConsensusObjectRead`. In an `events` subscription, omitting the argument defaults to the parent event's transaction. The resolver scans the transaction's `object_changes` first, then the `unchanged_consensus_objects` filtered to the `ReadOnlyRoot` variant; non-object marker variants (cancelled, stream-ended, per-epoch) resolve to null since they do not house an object the user can navigate to. Tests cover both variants in streaming mode: a `mutate_and_emit` flow demonstrates the `ObjectChange` path with input/output state transitions, and an `emit_with_clock` flow demonstrates the `ConsensusObjectRead` path against the read-only shared clock object. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Covers three sub-cases against an indexed transaction: an address present in the tx as a newly created object resolves to the `ObjectChange` variant; an address with no `transactionDigest` argument outside subscription scope returns null; an address not referenced by the tx returns null.
Uses the transactional test runner's deterministic simulator and `@{digest_N}` / `@{obj_N_M}` placeholders, so no Move package or live cluster is needed.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Two follow-up changes to address reviewer feedback on this PR: Rename `TransactionObjectRef` to `TransactionObject`. In Sui's vocabulary `ObjectRef` typically refers to a tuple of id + version + digest, which is narrower than what the union here returns (a full `ObjectChange` or `ConsensusObjectRead`). The shorter name reads more accurately for what the type is. Wrap `ProcessedTransaction.contents` in `Arc<TransactionContents>`. Previously each subscriber's resolver call was deep-cloning the whole `TransactionContents` (including boxed effects, transaction data, and proto fields) when populating `EffectsContents`. With N subscribers on the same checkpoint, that adds up. Now the deep clone happens once at checkpoint ingestion and each subscriber pays only a refcount bump. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
| return Ok(None); | ||
| }; | ||
|
|
||
| let contents = EffectsContents::empty(self.scope.clone()) |
There was a problem hiding this comment.
That's true for streaming, but not for queries.
If instead of tx_digest_viewed_at we had Scope::effects_viewed_at(&self): TransactionEffects, we could still instantiate that using just a digest to create an instance whose contents were not fetched yet, but if we already have TransactionEffects in our hand, we can reuse that, which also means re-using its potentially cached contents.
- In the subscriptions case, we would instantiate the
effects_viewed_atusing just the digest (as you are doing now), and if/when its contents are fetched, it will look in theScope's data source first (note that this is not a cache). - In the query case, if
asTransactionObjectwas nested under some effects, then we would already have the transaction and its contents in our hands, and we would initialize the scope withselfin the context ofScope. - In either case, if
asTransactionObjectwas nested under an event, we would again initialise the effects being viewed with just a digest (if they are not already set), because we may not have fetched the event's transaction.
Establish a uniform rule: every Transaction and TransactionEffects constructor anchors its scope to that transaction via Scope::active_transaction. Add Transaction::with_contents and TransactionEffects::with_contents (read digest from contents, anchor with contents). Route Transaction::fetch, TransactionEffects::fetch, from_executed_transaction, paginate_preloaded_transactions, and the streaming-transactions subscription through these constructors; the events subscription anchors scope via with_active_transaction_contents directly. Scope::with_active_transaction_digest preserves an existing same-digest anchor so Event::transaction and Transaction::with_digest do not strip hydrated contents when re-entering the same transaction. With anchoring uniform across both indexed and streaming paths, *Contents::fetch needs only one short-circuit (scope.active_transaction_contents_for); the previous streaming fast path (ProcessedCheckpoint::transaction_by_digest and Scope::streamed_transaction_by_digest) is removed as dead code. Extend the indexed transactional test for Address.asTransactionObject with two new run-graphql blocks covering implicit-digest resolution through transactionEffects(...).events.* and transaction(...).effects.events.*. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
7a17df4 to
12ed41a
Compare
tpham-mysten
left a comment
There was a problem hiding this comment.
Thanks @amnn! I have refactored the Scope a bit to integrate the effects_viewed_at ideas, but with different name
| return Ok(None); | ||
| }; | ||
|
|
||
| let contents = EffectsContents::empty(self.scope.clone()) |
There was a problem hiding this comment.
Yeah makes sense! I anchored Scope at the Transaction level. Every Transaction and TransactionEffects constructor (with_contents, fetch, from_executed_transaction) sets Scope::active_transaction, so descendants like Address.asTransactionObject resolve consistently regardless of entry point.Scope::with_active_transaction_digest preserves an existing same-digest anchor, so re-entering the same transaction (e.g. via Event::transaction) doesn't drop hydrated contents.
amnn
left a comment
There was a problem hiding this comment.
Awesome, thanks @tpham-mysten
## Description - Expose cursor: String! on Checkpoint — encoded JsonCursor<u64> of the sequence number, matching the format Query.checkpoints pagination already uses. - Clients (especially streaming subscribers) read this on each yielded checkpoint and pass it back as afterCursor to resume from this point. ## Test plan How did you test the new or updated feature? - e2e + unit tests --- ## Stack - #26019 - #26094 - #26117 - #26170 - #26194 - #26202 - #26414 - #26453 - #26476 - #26487 - #26425 - #26495 - #26714 - #26731 ## Release notes Check each box that your changes affect. If none of the boxes relate to your changes, release notes aren't required. For each box you select, include information after the relevant heading that describes the impact of your changes that a user might notice and any actions they must take to implement updates. - [ ] Protocol: - [ ] Nodes (Validators and Full nodes): - [ ] gRPC: - [ ] JSON-RPC: - [x] GraphQL: Add cursor field to Checkpoint object - [ ] CLI: - [ ] Rust SDK: - [ ] Indexing Framework:
## Description - Add `afterCursor` and `afterCheckpoint` to the checkpoints subscription so clients can resume from a known point. On subscribe, a LedgerService scan covers the gap to the live tip in parallel batches, then the live broadcast takes over. - Gaps or Lagged events during the live phase re-enter the scan, bounded by `resume_max_recovery_attempts` consecutive retries before disconnecting the subscriber. ## Test plan How did you test the new or updated feature? - cargo nextest run --features staging -p sui-indexer-alt-graphql --lib - Schema snapshot regenerated and checked in. - Manual: connect to a streaming server with afterCheckpoint set --- ## Stack - #26019 - #26094 - #26117 - #26170 - #26194 - #26202 - #26414 - #26453 - #26476 - #26487 - #26425 - #26495 - #26731 - #26714 ## Release notes Check each box that your changes affect. If none of the boxes relate to your changes, release notes aren't required. For each box you select, include information after the relevant heading that describes the impact of your changes that a user might notice and any actions they must take to implement updates. - [ ] Protocol: - [ ] Nodes (Validators and Full nodes): - [ ] gRPC: - [ ] JSON-RPC: - [x] GraphQL: Support resumable checkpoints subscription in features `staging` - [ ] CLI: - [ ] Rust SDK: - [ ] Indexing Framework:
## Description Part of the GraphQL streaming series (follows #26714, checkpoints [14/n]). Adds a resumable `transactions` subscription: it backfills the filter's matching transactions from a resume point via the scanning API, then hands off to the live checkpoint broadcast at a pinned seam, so delivery is contiguous with no gap or duplicate. The backfill scan retries transient errors (e.g. a rolling indexer deploy) with bounded exponential backoff before it gives up and disconnects. Staging-gated and not yet exposed. Each payload is a batch of transaction edges (`[TransactionEdge!]!`) rather than a single edge: the backfill packs matches up to a fixed batch size so a deep scan coalesces its navigation reads, while live delivers each checkpoint's matches as one batch. Per-transaction cursors keep resume exact even when a batch splits a checkpoint. The scan-then-live handoff, sparse-filter coverage, cursor unification, batching, and retry are documented at the top of `api/subscription/transactions.rs`. ## Test plan - e2e parity test (`test_transaction_subscription_live_backfill_parity`): the same transactions resolve identically whether delivered live or through the backfill scan, plus the existing streaming/ordering/resume subscription tests. - Batching e2e tests: `test_transaction_subscription_backfill_batches_matches` (a deep backfill packs matches into multi-edge payloads) and `test_transaction_subscription_live_batches_per_checkpoint` (each live payload is one checkpoint's matches). - Unit tests for the scan retry: recovers within the budget, gives up once it is exhausted. --- ## Stack - #26019 - #26094 - #26117 - #26170 - #26194 - #26202 - #26414 - #26453 - #26476 - #26487 - #26425 - #26495 - #26731 - #26714 - #27140 ## Release notes Check each box that your changes affect. If none of the boxes relate to your changes, release notes aren't required. For each box you select, include information after the relevant heading that describes the impact of your changes that a user might notice and any actions they must take to implement updates. - [ ] Protocol: - [ ] Nodes (Validators and Full nodes): - [ ] gRPC: - [ ] JSON-RPC: - [x] GraphQL: Support transaction subscription from historical cursor - [ ] CLI: - [ ] Rust SDK: - [ ] Indexing Framework:
…27537) ## Description Resolve subscription payloads concurrently rather than one at a time. A single `max_concurrent_resolutions` config drives both the resolve window (async-graphql's `subscription_resolution_concurrency`, now [merged](amnn/async-graphql#2)) and the backfill scan page, so a batch's matches resolve within the concurrency budget and coalesce their content reads into one `KvLoader` round trip. ## Test plan - e2e parity test (`test_transaction_subscription_live_backfill_parity`): the same transactions resolve identically whether delivered live or through the backfill scan, plus the existing streaming/ordering/resume subscription tests. - Batching e2e tests: `test_transaction_subscription_backfill_batches_matches` (a deep backfill packs matches into multi-edge payloads) and `test_transaction_subscription_live_batches_per_checkpoint` (each live payload is one checkpoint's matches). - Unit tests for the scan retry: recovers within the budget, gives up once it is exhausted. --- ## Stack - #26019 - #26094 - #26117 - #26170 - #26194 - #26202 - #26414 - #26453 - #26476 - #26487 - #26425 - #26495 - #26731 - #26714 - #27140 - #27537 ## Release notes Check each box that your changes affect. If none of the boxes relate to your changes, release notes aren't required. For each box you select, include information after the relevant heading that describes the impact of your changes that a user might notice and any actions they must take to implement updates. - [ ] Protocol: - [ ] Nodes (Validators and Full nodes): - [ ] gRPC: - [ ] JSON-RPC: - [ ] GraphQL: - [ ] CLI: - [ ] Rust SDK: - [ ] Indexing Framework:
…17/n] (#27564) ## Description A per-subscriber delivery throttle for GraphQL subscriptions. The SSE handler paces each payload by its cost (`cost / rate` seconds), so a subscriber's sustained delivery stays within `per_subscriber_max_output_nodes_per_second` output nodes per second (default ~off, an operator opts in by lowering it). A payload's cost is its output nodes plus a query-depth surcharge. The throttle lives in the handler, so it wraps every subscription type identically. ## Test plan Unit tests cover the cost model and `Throttle::wrap` pacing (deterministic via tokio's paused clock). `throttle_subscription.rs` adds e2e for checkpoint pacing/ordering, the disable path, richer-payload cost, and the transaction backfill path. ## Stack - #26019 - #26094 - #26117 - #26170 - #26194 - #26202 - #26414 - #26453 - #26476 - #26487 - #26425 - #26495 - #26731 - #26714 - #27140 - #27537 - #27564 ## Release notes - [ ] Protocol: - [ ] Nodes (Validators and Full nodes): - [ ] gRPC: - [ ] JSON-RPC: - [ ] GraphQL: - [ ] CLI: - [ ] Rust SDK: - [ ] Indexing Framework:
Description
Streaming subscribers want to take an object ID that surfaces inside an event payload and immediately ask "how was this object referenced by the transaction that emitted the event?" without round-tripping to the indexed Query API. The object is also sometimes an unchanged shared input (e.g. a clock or registry), so it never appears in
objectChanges; the resolver therefore needs to expose both the change case and the read-only consensus input case.This PR adds:
ObjectChangeandConsensusObjectReadare reused as-is from the schema; no duplication. The resolver scanseffects.object_changesfirst, theneffects.unchanged_consensus_objectsfiltered to the ReadOnlyRoot variant. Other unchanged-input markers (cancelled, stream-ended, per-epoch) resolve to null since they do not house an object the user can navigate to.In an events subscription, the transactionDigest argument may be omitted; the field then resolves against the transaction that emitted the parent event, via a new Scope::tx_digest_viewed_at() helper. Passing an explicit transactionDigest other than the parent event's transaction in subscription context is intentionally not supported (the document states this); use the indexed Query API for arbitrary transaction lookups.
Test plan
How did you test the new or updated feature?
Stack
Release notes
Check each box that your changes affect. If none of the boxes relate to your changes, release notes aren't required.
For each box you select, include information after the relevant heading that describes the impact of your changes that a user might notice and any actions they must take to implement updates.