Skip to content

[Data Availability] Introduction of a new subscription package #8131

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
illia-malachyn wants to merge 10 commits into
base: feature/optimistic-sync
Choose a base branch
from
Open

[Data Availability] Introduction of a new subscription package #8131

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
ab6f2f3
Rename subscription to subscription_old
illia-malachyn Nov 6, 2025
2a078ff
add new subscription package
illia-malachyn Nov 6, 2025
56068d1
integrate new subscription package to backends
illia-malachyn Nov 7, 2025
034b9d3
init exec data backend
illia-malachyn Nov 7, 2025
837bc23
fix tests in rpc backends
illia-malachyn Nov 10, 2025
b316674
use new subscription package in websockets
illia-malachyn Nov 10, 2025
3d8b749
use new subscription type in legacy ws handler
illia-malachyn Nov 10, 2025
a3c5979
fix build errors
illia-malachyn Nov 10, 2025
af45ca6
run go mod tidy
illia-malachyn Nov 10, 2025
9d79b68
fix state stream backend tests
illia-malachyn Nov 11, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion .mockery.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,8 @@ packages:
github.com/onflow/flow-go/engine/access/rpc/connection:
github.com/onflow/flow-go/engine/access/state_stream:
github.com/onflow/flow-go/engine/access/subscription:
github.com/onflow/flow-go/engine/access/subscription/tracker:
github.com/onflow/flow-go/engine/access/subscription_old:
github.com/onflow/flow-go/engine/access/subscription_old/tracker:
github.com/onflow/flow-go/engine/access/wrapper:
config:
dir: "engine/access/mock"
Expand Down
236 changes: 136 additions & 100 deletions access/api.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@ type TransactionStreamAPI interface {
ctx context.Context,
txID flow.Identifier,
requiredEventEncodingVersion entities.EventEncodingVersion,
) subscription.Subscription
) subscription.Subscription[[]*accessmodel.TransactionResult]

// SendAndSubscribeTransactionStatuses sends a transaction to the execution node and subscribes to its status updates.
// Monitoring begins from the reference block saved in the transaction itself and streams status updates until the transaction
Expand All @@ -88,7 +88,140 @@ type TransactionStreamAPI interface {
ctx context.Context,
tx *flow.TransactionBody,
requiredEventEncodingVersion entities.EventEncodingVersion,
) subscription.Subscription
) subscription.Subscription[[]*accessmodel.TransactionResult]
}

type BlockStreamAPI interface {
// SubscribeBlocksFromStartBlockID subscribes to the finalized or sealed blocks starting at the
// requested start block id, up until the latest available block. Once the latest is reached,
// the stream will remain open and responses are sent for each new block as it becomes available.
//
// Each block is filtered by the provided block status, and only blocks that match blockStatus
// are returned. blockStatus must be BlockStatusSealed or BlockStatusFinalized.
//
// If invalid parameters are supplied, a failed subscription is returned.
SubscribeBlocksFromStartBlockID(
ctx context.Context,
startBlockID flow.Identifier,
blockStatus flow.BlockStatus,
) subscription.Subscription[*flow.Block]

// SubscribeBlocksFromStartHeight subscribes to the finalized or sealed blocks starting at the requested
// start block height, up until the latest available block. Once the latest is
// reached, the stream will remain open and responses are sent for each new
// block as it becomes available.
//
// Each block is filtered by the provided block status, and only blocks that match blockStatus
// are returned. blockStatus must be BlockStatusSealed or BlockStatusFinalized.
//
// If invalid parameters are supplied, a failed subscription is returned.
SubscribeBlocksFromStartHeight(
ctx context.Context,
startHeight uint64,
blockStatus flow.BlockStatus,
) subscription.Subscription[*flow.Block]

// SubscribeBlocksFromLatest subscribes to the finalized or sealed blocks starting at the latest sealed block,
// up until the latest available block. Once the latest is
// reached, the stream will remain open and responses are sent for each new
// block as it becomes available.
//
// Each block is filtered by the provided block status, and only blocks that match blockStatus
// are returned. blockStatus must be BlockStatusSealed or BlockStatusFinalized.
//
// If invalid parameters are supplied, a failed subscription is returned.
SubscribeBlocksFromLatest(
ctx context.Context,
blockStatus flow.BlockStatus,
) subscription.Subscription[*flow.Block]

// SubscribeBlockHeadersFromStartBlockID streams finalized or sealed block headers starting at the requested
// start block id, up until the latest available block header. Once the latest is
// reached, the stream will remain open and responses are sent for each new
// block header as it becomes available.
//
// Each block is filtered by the provided block status, and only blocks that match blockStatus
// are returned. blockStatus must be BlockStatusSealed or BlockStatusFinalized.
//
// If invalid parameters are supplied, a failed subscription is returned.
SubscribeBlockHeadersFromStartBlockID(
ctx context.Context,
startBlockID flow.Identifier,
blockStatus flow.BlockStatus,
) subscription.Subscription[*flow.Header]

// SubscribeBlockHeadersFromStartHeight streams finalized or sealed block headers starting at the requested
// start block height, up until the latest available block header. Once the latest is
// reached, the stream will remain open and responses are sent for each new
// block header as it becomes available.
//
// Each block is filtered by the provided block status, and only blocks that match blockStatus
// are returned. blockStatus must be BlockStatusSealed or BlockStatusFinalized.
//
// If invalid parameters are supplied, a failed subscription is returned.
SubscribeBlockHeadersFromStartHeight(
ctx context.Context,
startHeight uint64,
blockStatus flow.BlockStatus,
) subscription.Subscription[*flow.Header]

// SubscribeBlockHeadersFromLatest streams finalized or sealed block headers starting at the latest sealed block,
// up until the latest available block header. Once the latest is
// reached, the stream will remain open and responses are sent for each new
// block header as it becomes available.
//
// Each block is filtered by the provided block status, and only blocks that match blockStatus
// are returned. blockStatus must be BlockStatusSealed or BlockStatusFinalized.
//
// If invalid parameters are supplied, a failed subscription is returned.
SubscribeBlockHeadersFromLatest(
ctx context.Context,
blockStatus flow.BlockStatus,
) subscription.Subscription[*flow.Header]

// SubscribeBlockDigestsFromStartBlockID streams finalized or sealed lightweight block starting at the requested
// start block id, up until the latest available block. Once the latest is
// reached, the stream will remain open and responses are sent for each new
// block as it becomes available.
//
// Each block is filtered by the provided block status, and only blocks that match blockStatus
// are returned. blockStatus must be BlockStatusSealed or BlockStatusFinalized.
//
// If invalid parameters are supplied, a failed subscription is returned.
SubscribeBlockDigestsFromStartBlockID(
ctx context.Context,
startBlockID flow.Identifier,
blockStatus flow.BlockStatus,
) subscription.Subscription[*flow.BlockDigest]

// SubscribeBlockDigestsFromStartHeight streams finalized or sealed lightweight block starting at the requested
// start block height, up until the latest available block. Once the latest is
// reached, the stream will remain open and responses are sent for each new
// block as it becomes available.
//
// Each block is filtered by the provided block status, and only blocks that match blockStatus
// are returned. blockStatus must be BlockStatusSealed or BlockStatusFinalized.
//
// If invalid parameters are supplied, a failed subscription is returned.
SubscribeBlockDigestsFromStartHeight(
ctx context.Context,
startHeight uint64,
blockStatus flow.BlockStatus,
) subscription.Subscription[*flow.BlockDigest]

// SubscribeBlockDigestsFromLatest streams finalized or sealed lightweight block starting at the latest sealed block,
// up until the latest available block. Once the latest is
// reached, the stream will remain open and responses are sent for each new
// block as it becomes available.
//
// Each block is filtered by the provided block status, and only blocks that match blockStatus
// are returned. blockStatus must be BlockStatusSealed or BlockStatusFinalized.
//
// If invalid parameters are supplied, a failed subscription is returned.
SubscribeBlockDigestsFromLatest(
ctx context.Context,
blockStatus flow.BlockStatus,
) subscription.Subscription[*flow.BlockDigest]
}

// API provides all public-facing functionality of the Flow Access API.
Expand All @@ -103,6 +236,7 @@ type API interface {
ScriptsAPI
TransactionsAPI
TransactionStreamAPI
BlockStreamAPI

// Ping responds to requests when the server is up.
//
Expand Down Expand Up @@ -249,102 +383,4 @@ type API interface {
// Expected sentinel errors providing details to clients about failed requests:
// - access.DataNotFoundError - No execution result with the given ID was found
GetExecutionResultByID(ctx context.Context, id flow.Identifier) (*flow.ExecutionResult, error)

// SubscribeBlocksFromStartBlockID subscribes to the finalized or sealed blocks starting at the
// requested start block id, up until the latest available block. Once the latest is reached,
// the stream will remain open and responses are sent for each new block as it becomes available.
//
// Each block is filtered by the provided block status, and only blocks that match blockStatus
// are returned. blockStatus must be BlockStatusSealed or BlockStatusFinalized.
//
// If invalid parameters are supplied, a failed subscription is returned.
SubscribeBlocksFromStartBlockID(ctx context.Context, startBlockID flow.Identifier, blockStatus flow.BlockStatus) subscription.Subscription

// SubscribeBlocksFromStartHeight subscribes to the finalized or sealed blocks starting at the requested
// start block height, up until the latest available block. Once the latest is
// reached, the stream will remain open and responses are sent for each new
// block as it becomes available.
//
// Each block is filtered by the provided block status, and only blocks that match blockStatus
// are returned. blockStatus must be BlockStatusSealed or BlockStatusFinalized.
//
// If invalid parameters are supplied, a failed subscription is returned.
SubscribeBlocksFromStartHeight(ctx context.Context, startHeight uint64, blockStatus flow.BlockStatus) subscription.Subscription

// SubscribeBlocksFromLatest subscribes to the finalized or sealed blocks starting at the latest sealed block,
// up until the latest available block. Once the latest is
// reached, the stream will remain open and responses are sent for each new
// block as it becomes available.
//
// Each block is filtered by the provided block status, and only blocks that match blockStatus
// are returned. blockStatus must be BlockStatusSealed or BlockStatusFinalized.
//
// If invalid parameters are supplied, a failed subscription is returned.
SubscribeBlocksFromLatest(ctx context.Context, blockStatus flow.BlockStatus) subscription.Subscription

// SubscribeBlockHeadersFromStartBlockID streams finalized or sealed block headers starting at the requested
// start block id, up until the latest available block header. Once the latest is
// reached, the stream will remain open and responses are sent for each new
// block header as it becomes available.
//
// Each block is filtered by the provided block status, and only blocks that match blockStatus
// are returned. blockStatus must be BlockStatusSealed or BlockStatusFinalized.
//
// If invalid parameters are supplied, a failed subscription is returned.
SubscribeBlockHeadersFromStartBlockID(ctx context.Context, startBlockID flow.Identifier, blockStatus flow.BlockStatus) subscription.Subscription

// SubscribeBlockHeadersFromStartHeight streams finalized or sealed block headers starting at the requested
// start block height, up until the latest available block header. Once the latest is
// reached, the stream will remain open and responses are sent for each new
// block header as it becomes available.
//
// Each block is filtered by the provided block status, and only blocks that match blockStatus
// are returned. blockStatus must be BlockStatusSealed or BlockStatusFinalized.
//
// If invalid parameters are supplied, a failed subscription is returned.
SubscribeBlockHeadersFromStartHeight(ctx context.Context, startHeight uint64, blockStatus flow.BlockStatus) subscription.Subscription

// SubscribeBlockHeadersFromLatest streams finalized or sealed block headers starting at the latest sealed block,
// up until the latest available block header. Once the latest is
// reached, the stream will remain open and responses are sent for each new
// block header as it becomes available.
//
// Each block is filtered by the provided block status, and only blocks that match blockStatus
// are returned. blockStatus must be BlockStatusSealed or BlockStatusFinalized.
//
// If invalid parameters are supplied, a failed subscription is returned.
SubscribeBlockHeadersFromLatest(ctx context.Context, blockStatus flow.BlockStatus) subscription.Subscription

// SubscribeBlockDigestsFromStartBlockID streams finalized or sealed lightweight block starting at the requested
// start block id, up until the latest available block. Once the latest is
// reached, the stream will remain open and responses are sent for each new
// block as it becomes available.
//
// Each block is filtered by the provided block status, and only blocks that match blockStatus
// are returned. blockStatus must be BlockStatusSealed or BlockStatusFinalized.
//
// If invalid parameters are supplied, a failed subscription is returned.
SubscribeBlockDigestsFromStartBlockID(ctx context.Context, startBlockID flow.Identifier, blockStatus flow.BlockStatus) subscription.Subscription

// SubscribeBlockDigestsFromStartHeight streams finalized or sealed lightweight block starting at the requested
// start block height, up until the latest available block. Once the latest is
// reached, the stream will remain open and responses are sent for each new
// block as it becomes available.
//
// Each block is filtered by the provided block status, and only blocks that match blockStatus
// are returned. blockStatus must be BlockStatusSealed or BlockStatusFinalized.
//
// If invalid parameters are supplied, a failed subscription is returned.
SubscribeBlockDigestsFromStartHeight(ctx context.Context, startHeight uint64, blockStatus flow.BlockStatus) subscription.Subscription

// SubscribeBlockDigestsFromLatest streams finalized or sealed lightweight block starting at the latest sealed block,
// up until the latest available block. Once the latest is
// reached, the stream will remain open and responses are sent for each new
// block as it becomes available.
//
// Each block is filtered by the provided block status, and only blocks that match blockStatus
// are returned. blockStatus must be BlockStatusSealed or BlockStatusFinalized.
//
// If invalid parameters are supplied, a failed subscription is returned.
SubscribeBlockDigestsFromLatest(ctx context.Context, blockStatus flow.BlockStatus) subscription.Subscription
}
Loading
Loading