[graphql-stream] Support previousTransaction with cached streamed transactions [20/n]

[tpham-mysten]

Description

On a live subscription, Object.previousTransaction resolved only its digest: the contents (kind, sender, effects) came back empty. The lookup falls through to the KV backend, which lags the streamed tip, so a transaction that was just streamed is not yet readable there.

The fix caches each streamed checkpoint's transactions in memory (StreamedTransactionStore), consulted before the KV backend so a just-streamed previousTransaction resolves from memory and hands off to the backend once it catches up. This mirrors the existing StreamedPackageStore, so the shared cache-and-evict logic is pulled into a generic StreamedStore, and both stores are now flushed by a single eviction task keyed on each one's backing watermark.

Test plan

• e2e + unit tests

Stack

• [graphql-stream] WebSocket subscription infrastructure [1/n] #26019
• [graphql-stream] Implement checkpoint subscription streaming [2/n] #26094
• [graphql-stream] Propagate transactions fields for checkpoint subscription streaming [3/n] #26117
• [graphql-stream] Add object resolution to checkpoint subscription transactions [4/n] #26170
• [graphql-stream] Add package caching from streamed checkpoints [5/n] #26194
• [graphql-stream] Add top-level transaction subscription [6/n] #26202
• [graphql-stream] Recover from upstream gRPC gaps via kv-rpc [7/n] #26414
• [graphql-stream] Add top-level events subscription [8/n] #26453
• [graphql-stream] Lift streamed checkpoint payload onto Scope [9/n] #26476
• [graphql-stream] Resolve Event.transaction in streaming mode [10/n] #26487
• [graphql-stream] SSE transport + SSE-aware GraphiQL [11/n] #26425
• [graphql-stream] Add Address.asTransactionObject [12/n] #26495
• [graphql-stream] Add cursor field to Checkpoint [13/n] #26731
• [graphql-stream] Resumable checkpoints subscription [14/n] #26714
• [graphql-stream] Resumable transactions subscription [15/n] #27140
• [graphql-stream] resolve subscription payloads concurrently [16/n] #27537
• [graphql-stream] per-subscriber delivery throttle for subscriptions [17/n] #27564
• [graphql-stream] Resumable event subscriptions on a shared scan-then-live driver [18/n] #27580
• [graphql-stream] Reject subscriptions that start too far ahead of the tip [19/n] #27635
• [graphql-stream] Support previousTransaction with cached streamed transactions [20/n] #27709

Release notes

Check each box that your changes affect. If none of the boxes relate to your changes, release notes aren't required.

• Protocol:
• Nodes (Validators and Full nodes):
• gRPC:
• JSON-RPC:
• GraphQL:
• CLI:
• Rust SDK:
• Indexing Framework:

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

Project	Deployment	Actions	Updated (UTC)
sui-docs	Building	Preview	Aug 20, 2026 3:31pm
sui-kiosk	Building	Preview	Aug 20, 2026 3:31pm

1 Skipped Deployment

Project	Deployment	Actions	Updated (UTC)
multisig-toolkit	Ignored	Preview	Aug 20, 2026 3:31pm

[wlmyng]

I think it's worth moving checkpoint_viewed_at on Scope into DataSource:

  enum DataSource {
      Indexed { viewed_at: u64 },   // normal query path
      Backfill,                     // indexed reads, deliberately unbounded
      Executed { execution_objects: ExecutionObjectMap },
      Streamed { checkpoint: Arc<ProcessedCheckpoint> },
  }

so that fetch bodies can match on the datasource and decide where to retrieve data accordingly

  match scope.data_source() {
      Executed { .. } => Ok(None),               
      Streamed { .. } | Backfill => {
          if let Some(c) = streamed_store.get(&digest) { return Ok(Some(c)); }
          kv.load_one_transaction(digest).await
      }
      Indexed { viewed_at } => {
          let tx = kv.load_one_transaction(digest).await?;
          Ok(tx.filter(|t| t.cp_sequence_number() <= viewed_at))  
      }
  }

At least so that the logic in L578-602 are bundled together, as the current scope (execution, stream, or indexed data) impacts where we can read from, and given that Executed and Streamed already hold some relevant data.

Building on this, what do you think about having some Arc<StreamedOverlays> that bundles all the streamed and cached data, such that CheckpointStreamTask writes to it through index_and_broadcast, eviction task drains it, and these overlays get handed to the resolvers as we do with package_store today through matching_edges:

S::matching_edges(checkpoint, &overlays, ...)

Then we can slap it into scope

Scope::for_streamed_checkpoint(overlays, limits, checkpoint)
    → package_store field = overlays.packages
    → DataSource::Streamed { checkpoint, overlays }

culminating in

  Streamed { overlays, .. } => overlays.transactions.get(&digest)
      .map_or_else(|| kv.load(digest), ...),
  Backfill => kv.load(digest), 

[tpham-mysten]

I think it's worth moving checkpoint_viewed_at on Scope into DataSource:

Yes, this is what we discussed at standup. But I think it's orthogonal: we can get the match style on the existing enum, without moving checkpoint_viewed_at, like:

match scope.data_source() {
    DataSource::Executed { .. } => Ok(None),
    DataSource::Streamed { .. } => {
        if let Some(c) = streamed_store.get(&digest) { return Ok(Some(c)); }
        kv.load_one_transaction(digest).await
    }
    DataSource::Indexed => match scope.checkpoint_viewed_at() {
        None => {                         // backfill: unbounded indexed read
            if let Some(c) = streamed_store.get(&digest) { return Ok(Some(c)); }
            kv.load_one_transaction(digest).await
        }
        Some(viewed_at) => {              // bounded query
            let tx = kv.load_one_transaction(digest).await?;
            Ok(tx.filter(|t| t.cp_sequence_number() <= viewed_at))
        }
    },
}

At least so that the logic in L578-602 are bundled together, as the current scope (execution, stream, or indexed data) impacts where we can read from, and given that Executed and Streamed already hold some relevant data.

I get the motivation for this, and I think it's more a style/standard call than a correctness one, so whichever we pick, we should use everywhere, and it can go in a follow-up PR.

Here's how I see the two:

match on DataSource

• Pro: makes it clear the mode is what decides where we read from, and it's exhaustive, so a new variant forces us to handle it.
• Con: the kv.load_one_transaction and the streamed-store check get repeated in several cases.

if-chain (current)

• Pro: reads as a step-by-step lookup: try the nearest place we might already have it (already hydrated, then on the scope, then the streamed store), then fall back to KV. Same flow in every mode; the mode only affects two steps (is the store usable, does the KV read need a cutoff). It's the usual "do we have it, else fetch it" shape we use in a lot of other lookups.
• Con: the mode's effect is spread across a few guards instead of one place, so "mode decides source" is implicit.

I slightly lean 60:40 to the if-chain, since the main flow is this step-by-step lookup and the mode only tweaks two steps, so a match ends up repeating the same logic arms. But it's close and mostly taste, so happy to bring it to standup. Either way it doesn't block this PR.

Building on this, what do you think about having some Arc that bundles all the streamed and cached data, such that CheckpointStreamTask writes to it through index_and_broadcast, eviction task drains it, and these overlays get handed to the resolvers as we do with package_store today through matching_edges:

Yeah I think it's nice wrapper. I have implemented that StreamedOverlays

Actually, I found the name StreamedOverlays is a bit confusing as it's basically an object holding shared cached store. WDYT about StreamedCaches?

[tpham-mysten]

Thanks @wlmyng!

[tpham-mysten]

I think it's worth moving checkpoint_viewed_at on Scope into DataSource:

Yes, this is what we discussed at standup. But I think it's orthogonal: we can get the match style on the existing enum, without moving checkpoint_viewed_at, like:

match scope.data_source() {
    DataSource::Executed { .. } => Ok(None),
    DataSource::Streamed { .. } => {
        if let Some(c) = streamed_store.get(&digest) { return Ok(Some(c)); }
        kv.load_one_transaction(digest).await
    }
    DataSource::Indexed => match scope.checkpoint_viewed_at() {
        None => {                         // backfill: unbounded indexed read
            if let Some(c) = streamed_store.get(&digest) { return Ok(Some(c)); }
            kv.load_one_transaction(digest).await
        }
        Some(viewed_at) => {              // bounded query
            let tx = kv.load_one_transaction(digest).await?;
            Ok(tx.filter(|t| t.cp_sequence_number() <= viewed_at))
        }
    },
}

At least so that the logic in L578-602 are bundled together, as the current scope (execution, stream, or indexed data) impacts where we can read from, and given that Executed and Streamed already hold some relevant data.

I get the motivation for this, and I think it's more a style/standard call than a correctness one, so whichever we pick, we should use everywhere, and it can go in a follow-up PR.

Here's how I see the two:

match on DataSource

• Pro: makes it clear the mode is what decides where we read from, and it's exhaustive, so a new variant forces us to handle it.
• Con: the kv.load_one_transaction and the streamed-store check get repeated in several cases.

if-chain (current)

• Pro: reads as a step-by-step lookup: try the nearest place we might already have it (already hydrated, then on the scope, then the streamed store), then fall back to KV. Same flow in every mode; the mode only affects two steps (is the store usable, does the KV read need a cutoff). It's the usual "do we have it, else fetch it" shape we use in a lot of other lookups.
• Con: the mode's effect is spread across a few guards instead of one place, so "mode decides source" is implicit.

I slightly lean 60:40 to the if-chain, since the main flow is this step-by-step lookup and the mode only tweaks two steps, so a match ends up repeating the same logic arms. But it's close and mostly taste, so happy to bring it to standup. Either way it doesn't block this PR.

Building on this, what do you think about having some Arc that bundles all the streamed and cached data, such that CheckpointStreamTask writes to it through index_and_broadcast, eviction task drains it, and these overlays get handed to the resolvers as we do with package_store today through matching_edges:

Yeah I think it's nice wrapper. I have implemented that StreamedOverlays

Actually, I found the name StreamedOverlays is a bit confusing as it's basically an object holding shared cached store. WDYT about StreamedCaches?