Splitted read storage from write storage #3164

xgreenx · 2025-12-09T20:19:43Z

Splitted read storage from write storage
Instead of using channels, now we query data directly from the storage adapter
Removed cloning of blocks by suing Arc

Instead of using channels, now we query data directly from the storage adapter.

cursor · 2025-12-09T20:19:51Z

PR Summary

Major refactor of the block-aggregator: introduces separate read/write storage layers and a direct storage-backed service (no channels), updates RPC wiring, and adapts tests and storage utilities.

Block Aggregator API (service/task):
- Introduce service and task modules with SharedState, Task, and StorageOrRemote{DB,BlocksProvider}.
- Define BlocksProvider (read) and BlocksStorage (write) traits; enforce contiguous height writes.
- Replace channel-based API with direct adapter calls via BlocksAggregatorApi in protobuf_adapter.
- Remove integration and importer_and_db_source modules; add old_block_source with iterator-based block fetching and ConvertorAdapter.
RPC wiring and adapters:
- Switch to fuel_core_block_aggregator_api::service::{Config, StorageMethod} across CLI, config, and sub-services.
- Initialize RPC with OldBlocksSource<ConvertorAdapter,...>; use broadcast-backed new-block subscriptions.
- Make TxReceipts::get_receipts sync; update ReceiptSource and tests.
Storage changes:
- Add AsStructuredStorage and refactor storage DBs (storage_db, remote_cache) to use it; provide StorageBlocksProvider and RemoteBlocksProvider for reads.
- Update contiguous height checks and metadata updates; simplify rollback via latest_height_from_metadata and rollback_last_block.
Core tweaks:
- Adjust Database<BlockAggregatorDatabase> commit behavior; remove custom rollback method.
- Refine CombinedDatabase::rollback_to with <= checks and block-aggregator storage rollback path.
Tests and tooling:
- Update RPC and S3 tests for new receipts shape (nested vectors) and API; remove flaky concurrency test.
- Add mocks (mockall) and itertools; tweak tokio features.

^{Written by Cursor Bugbot for commit 6635578. This will update automatically on new commits. Configure here.}

crates/services/block_aggregator_api/src/service.rs

MitchTurner · 2025-12-09T23:24:58Z

crates/fuel-core/src/p2p_test_helpers.rs

    subscribe_to_transactions: Option<bool>,
    #[cfg(feature = "rpc")]
-    rpc_config: Option<fuel_core_block_aggregator_api::integration::Config>,
+    rpc_config: Option<fuel_core_block_aggregator_api::service::Config>,


We don't like integration over service :P ?

MitchTurner · 2025-12-09T23:28:13Z

crates/services/block_aggregator_api/src/api/protobuf_adapter.rs

-    query_sender:
-        tokio::sync::mpsc::Sender<BlockAggregatorQuery<BlockRangeResponse, ProtoBlock>>,
+#[cfg_attr(test, mockall::automock)]
+pub trait BlocksAggregatorApi: Send + Sync + 'static {


In reality, we could just depend on the BlocksProvider trait here and directly on the broadcast::Sender that we have in the SharedStorage. This would parallel what we are doing in the aggregator service already, since it takes a dep on BlocksStorage and broadcast::Sender.

I think that's the right pattern.

MitchTurner · 2025-12-09T23:32:34Z

crates/services/block_aggregator_api/src/api/protobuf_adapter.rs

-        tokio::sync::mpsc::Receiver<BlockAggregatorQuery<BlockRangeResponse, ProtoBlock>>,
+pub struct UninitializedTask<B> {
+    addr: SocketAddr,
+    api: B,


See above comment. Maybe rename something like block_aggregator

MitchTurner · 2025-12-09T23:35:14Z

crates/services/block_aggregator_api/src/api/protobuf_adapter.rs

+        let server = Server::new(self.api);
        let router = tonic::transport::Server::builder()
            .add_service(ProtoBlockAggregatorServer::new(server));
-        self.router = Some(router);
-        Ok(())
-    }
-
-    fn get_router(&mut self) -> anyhow::Result<Router> {
-        self.router
-            .take()
-            .ok_or_else(|| anyhow!("Router has not been initialized yet"))
+        let task = Task {
+            addr: self.addr,
+            router: Some(router),
+        };
+        Ok(task)


Do we really need an UninitializedTask just for this? What's the point of into_task for anyway haha?

MitchTurner · 2025-12-09T23:37:21Z

crates/services/block_aggregator_api/src/api/protobuf_adapter.rs

-impl BlockAggregatorApi for ProtobufAPI {
-    type BlockRangeResponse = BlockRangeResponse;
-    type Block = ProtoBlock;
+pub type APIService<B> = ServiceRunner<UninitializedTask<B>>;


Nit: I'm not a fan of aliases like this, as it adds misdirection. If I publically saw an APIService, I wouldn't assume that it was in fact the ServiceRunner.

MitchTurner · 2025-12-10T00:12:29Z

crates/services/block_aggregator_api/src/service.rs

+    }
+}
+
+impl<S> BlocksAggregatorApi for SharedState<S>


We shouldn't be iimplementing a foreign port like this here. See above comment on moving renaming and moving SharedState.

MitchTurner · 2025-12-10T00:17:02Z

crates/services/block_aggregator_api/src/task.rs

+
+/// The Block Aggregator service, which aggregates blocks from a source and stores them in a database
+/// Queries can be made to the service to retrieve data from the `DB`
+pub struct Task<S1, S2, Blocks>


Why S1 and S2? Can we use S and P? or Storage and PRovider?

MitchTurner · 2025-12-10T00:34:40Z

crates/services/block_aggregator_api/src/db.rs

+pub trait BlocksProvider: Send + Sync + 'static {
+    type Block: Send + Sync + 'static;
    /// The type used to report a range of blocks
    type BlockRangeResponse;

-    /// Stores a block with the given ID
-    fn store_block(
-        &mut self,
-        block: BlockSourceEvent<Self::Block>,
-    ) -> impl Future<Output = Result<()>> + Send;
-
    /// Retrieves a range of blocks from the database
    fn get_block_range(
        &self,
        first: BlockHeight,
        last: BlockHeight,
-    ) -> impl Future<Output = Result<Self::BlockRangeResponse>> + Send;
+    ) -> Result<Self::BlockRangeResponse>;

    /// Retrieves the current height of the aggregated blocks If there is a break in the blocks,
    /// i.e. the blocks are being aggregated out of order, return the height of the last
    /// contiguous block
-    fn get_current_height(
-        &self,
-    ) -> impl Future<Output = Result<Option<BlockHeight>>> + Send;
+    fn get_current_height(&self) -> Result<Option<BlockHeight>>;
+}


This can move the the protobuf adapter code and replace BlocksAggregatorAPI port, like I mentioned above.

MitchTurner · 2025-12-10T00:38:17Z

crates/services/block_aggregator_api/src/db/storage_db.rs

    }
+}
+
+impl<S> BlocksProvider for StorageBlocksProvider<S>


If we move this port to the protobuf adapter code, this impl should go with it.

MitchTurner · 2025-12-10T00:39:28Z

crates/services/block_aggregator_api/src/db/storage_or_remote_db.rs

+    S: 'static,
+    S: KeyValueInspect<Column = Column>,
+    S: AtomicView,
+    S::LatestView: Unpin + Send + Sync + KeyValueInspect<Column = Column> + 'static,


IDK how you remember which traits to use to keep it this clean :P

… blocks (#3166)

cursor · 2025-12-11T01:51:20Z

crates/services/block_aggregator_api/src/task.rs

+            let take = match result {
+                Ok((block_height, _)) => {
+                    if let Some(importer_height) = importer_height {
+                        *block_height < importer_height


Bug: Block at importer height lost during stream restart

The take_while condition uses *block_height < importer_height which excludes the block at exactly last_seen_importer_height from being fetched from the old blocks source. When restart_blocks_stream() is called, a new broadcast subscription is created after the block at that height was already broadcasted. Since the block is neither taken from old blocks (excluded by <) nor received by the new subscription (broadcast already happened), it gets permanently lost. The condition should use <= instead of < to ensure the boundary block is recovered from the old blocks source.

Splitted read storage from write storage

0bb0afa

Instead of using channels, now we query data directly from the storage adapter.

xgreenx self-assigned this Dec 9, 2025

xgreenx requested review from a team, Dentosal, MitchTurner and rymnc as code owners December 9, 2025 20:19

xgreenx added the no changelog Skip the CI check of the changelog modification label Dec 9, 2025

Make CI happy

7095811

cursor bot reviewed Dec 9, 2025

View reviewed changes

crates/services/block_aggregator_api/src/service.rs Outdated Show resolved Hide resolved

Make CI happy

eba8c17

MitchTurner reviewed Dec 10, 2025

View reviewed changes

Unified new and old block in one stream to simplify handling of these…

6635578

… blocks (#3166)

cursor bot reviewed Dec 11, 2025

View reviewed changes

Splitted read storage from write storage #3164

Are you sure you want to change the base?

Splitted read storage from write storage #3164

Conversation

xgreenx commented Dec 9, 2025

Uh oh!

cursor bot commented Dec 9, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

PR Summary

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

MitchTurner Dec 9, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cursor bot Dec 11, 2025

Choose a reason for hiding this comment

Bug: Block at importer height lost during stream restart

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

cursor bot commented Dec 9, 2025 •

edited

Loading

MitchTurner Dec 9, 2025 •

edited

Loading