feat(indexer)!: implement state sync and utxo scan api - #1549
Conversation
|
Caution Review failedThe pull request is closed. WalkthroughAdds a network-wide state synchronization subsystem (worker, committee RPC pool, block scanner), rewrites indexer storage and schemas (events, utxos, checkpoints, key-values), moves many codecs to reader-based APIs, adjusts epoch/validator RPCs for multi-checkpoint and value-filtered sync, and updates client/GraphQL/JSON-RPC surfaces and types. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
participant Main as Indexer Main
participant Bootstrap as Bootstrap
participant Epoch as EpochManager
participant Net as Networking
participant Store as SqliteIndexerStore
participant TM as TemplateManager
participant Sync as NetworkWideStateSync
Main->>Bootstrap: spawn_services(config, shutdown, keypair, global_db, consensus_constants)
Bootstrap->>Store: open DB & check_store(network)
Bootstrap->>Epoch: start epoch handles
Bootstrap->>TM: start template manager
Bootstrap->>Sync: new(epoch_manager, networking, store, template_manager, cfg)
Bootstrap->>Sync: spawn(shutdown)
Note over Sync: loop on tick or epoch change
Sync->>Epoch: get_network_description()
Sync->>Net: build ValidatorCommitteeRpcPools
alt checkpoints
Sync->>Net: fetch checkpoints (from_epoch, limit)
Sync->>Store: insert_or_ignore_epoch_checkpoint(...)
end
alt state sync
Sync->>Net: stream state updates (value_filters)
Sync->>Store: batch_insert_substate_transitions(...)
Sync->>Store: batch_insert_utxo_updates(...)
Sync->>Store: batch_insert_events(...)
Sync->>Store: upsert_substate(...)
Sync->>TM: enqueue template changes
end
sequenceDiagram
autonumber
participant Client
participant RPC as JSON-RPC Server
participant H as JsonRpcHandlers
participant SM as SubstateManager
participant DB as SqliteIndexerStore
Client->>RPC: get_utxo_updates(request)
RPC->>H: dispatch to handler
H->>SM: get_utxo_updates(resource, shard, from_ver, tag_bytes, limit)
SM->>DB: with_read_tx(get_utxo_updates(...))
DB-->>SM: Vec<UtxoUpdate>
SM-->>H: Vec<UtxoUpdate>
H-->>RPC: GetUtxoUpdatesResponse
RPC-->>Client: result
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120–180 minutes Possibly related PRs
Poem
Tip 🔌 Remote MCP (Model Context Protocol) integration is now available!Pro plan users can now connect to remote MCP servers from the Integrations page. Connect with popular remote MCPs such as Notion and Linear to add more context to your reviews and chats. ✨ Finishing Touches
🧪 Generate unit tests
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. CodeRabbit Commands (Invoked using PR/Issue comments)Type Other keywords and placeholders
CodeRabbit Configuration File (
|
cb4f5fb to
0ef0108
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 23
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (7)
applications/tari_indexer/src/storage_sqlite/models/substate.rs (2)
49-55: Avoid lossy cast: use u32::try_from for version.Casting i32 → u32 can wrap negatives. Use a checked conversion and surface an error.
Apply:
- version: row.version as u32, + version: u32::try_from(row.version) + .map_err(|_| anyhow::anyhow!("Negative version for substate {}", row.address))?,
33-44: Fix PrimitiveDateTime serialization and update downstream consumersThe switch from an
i64epoch value totime::PrimitiveDateTimemeans every layer that reads or writestimestampmust be updated. In particular:• Rust model & store
–applications/tari_indexer/src/storage_sqlite/models/substate.rsnow defines
rust pub timestamp: PrimitiveDateTime,
–applications/tari_indexer/src/storage_sqlite/store_factory.rsmaps the DB value into the DTO unchanged:
rust let timestamp = s.timestamp; Ok(ListSubstateItem { …, timestamp })• Indexer JSON-RPC → wallet SDK mapping
– Inapplications/tari_walletd/src/indexer_jrpc_impl.rs, you forwards.timestampintoSubstateListItem { timestamp }from the wallet SDK. Confirm thattari_ootle_wallet_sdk::network::SubstateListItemalso usesPrimitiveDateTime(it should), and that its Serde support is configured (e.g. via#[serde(with = "time::serde::rfc3339")]or equivalent).• Client-side API types
– Inclients/tari_indexer_client/src/types.rs,ListSubstateItemalready has
rust pub timestamp: PrimitiveDateTime,• Web UI parsing
– Inapplications/tari_indexer/web_ui/src/routes/Substates/Substates.tsxat line 254:
tsx {new Date(Number(row.timestamp) * 1000).toDateString()}
This still treatstimestampas an epoch numeric value. You must either:
1. SerializePrimitiveDateTimeas a numeric epoch in the API (add a customSerializeimpl or#[serde]attribute), or
2. Update the UI to parse the ISO‐8601 string (e.g.new Date(row.timestamp).toDateString())Action items
- Add or verify
#[serde(...)]attributes on allPrimitiveDateTimefields so Serde produces the desired format (epoch vs. RFC 3339 string).- Update the JSON-RPC layer (wallet SDK and client types) to match.
- Fix the React UI to stop using
Number(row.timestamp)*1000when parsing.This is a critical fix: without it, UIs will display
NaNdates and API clients may fail to deserialize timestamps correctly.crates/state_store_rocksdb/src/codecs/column.rs (1)
14-19: Docstring incorrectly states "32 bytes" — should be "4 bytes".The Column key encodes a u32 (4 bytes), not 32 bytes. This can mislead future maintainers.
Apply this diff:
/// A const key used to differentiate "columns" in a reused column family. -/// This hard codes 32 bytes (big-endian) from the encoded bytes. +/// This hard codes 4 bytes (big-endian) from the encoded bytes. /// It is not recommended to use this on a shared column family that uses prefix lookups, as the codec used would needapplications/tari_indexer/src/json_rpc/server.rs (2)
64-67: Avoid logging full request bodies; log metadata insteadLogging the entire JsonRpcExtractor at debug can leak sensitive payloads and will bloat logs. Prefer logging method, request id, and payload size; avoid parameters by default.
Apply this minimal change to stop logging full bodies:
- debug!(target: LOG_TARGET, "🌐 JSON-RPC body: {:?}", value); + debug!(target: LOG_TARGET, "🌐 JSON-RPC body received (method={}, id={:?})", value.method, value.id);If you want payload introspection in development, gate detailed logging behind a feature flag or env var and redact known sensitive fields.
98-126: Middleware can panic on body read and degrades performance by fully buffering the responseto_bytes(...).await.unwrap() will panic if the body stream errs. Also, the middleware buffers the entire response body for every request, which is expensive and breaks streaming semantics.
Prefer tower_http::trace for request/response tracing. It avoids full buffering and integrates with tracing. Replace the ad-hoc logger:
use tower_http::cors::CorsLayer; +use tower_http::trace::TraceLayer; @@ - .layer(middleware::from_fn(logger::middleware_fn)) + .layer(TraceLayer::new_for_http())If you must keep the custom logger temporarily, at least make it non-panicking and avoid buffering unless debug is enabled:
- let (parts, body) = res.into_parts(); - let body_bytes = to_bytes(body).await.unwrap(); - debug!(target: LOG_TARGET, "🌐 Response: {}", String::from_utf8_lossy(&body_bytes)); - Ok(Response::from_parts(parts, Body::from(body_bytes))) + let (parts, body) = res.into_parts(); + if log::log_enabled!(log::Level::Debug) { + match to_bytes(body).await { + Ok(body_bytes) => { + debug!(target: LOG_TARGET, "🌐 Response ({} bytes)", body_bytes.len()); + Ok(Response::from_parts(parts, Body::from(body_bytes))) + }, + Err(e) => { + warn!(target: LOG_TARGET, "🌐 Response body read failed: {}", e); + Ok(Response::from_parts(parts, Body::empty())) + }, + } + } else { + // Do not buffer; just return as-is when not debugging + Ok(Response::from_parts(parts, Body::empty())) + }Note: The “empty body” fallback above still alters the response. The recommended fix is to use TraceLayer and remove full-body logging.
applications/tari_indexer/src/storage_sqlite/schema.rs (1)
14-24: Add missing migration foreventsschema changesThe Diesel schema in
applications/tari_indexer/src/storage_sqlite/schema.rshas removed theversionandtimestampcolumns and introducedcreated_at, but there is no corresponding SQL migration. Without this, existing databases will fail to migrate and data integrity will be lost.Please add a new migration under
applications/tari_indexer/src/storage_sqlite/migrations/(e.g.2025-08-22-000000_alter_events_table) that:
- Adds the
created_atcolumn, backfilling fromtimestampfor existing rowsALTER TABLE events ADD COLUMN created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP; UPDATE events SET created_at = timestamp;- Drops the deprecated columns
ALTER TABLE events DROP COLUMN timestamp; ALTER TABLE events DROP COLUMN version;- Updates or adds an index if needed (e.g. on
created_at) to maintain query performance.After adding the migration, ensure:
- All existing data is preserved and backfilled correctly.
- Diesel schema (
schema.rs) matches the final table definition.- Integration tests (and any CI migration checks) pass successfully.
applications/tari_indexer/src/lib.rs (1)
210-224: Address the TODO comment for proper shutdown handling.The TODO comment on Line 218 indicates that shutdown handling during scanning needs to be implemented. This could lead to incomplete state or resource leaks if the scanner is in the middle of processing when shutdown is requested.
Would you like me to implement proper shutdown handling for the scanning loop? This would involve checking the shutdown signal within the scan operation and ensuring graceful termination.
Test Results (CI)418 tests ±0 394 ✅ ±0 1h 15m 44s ⏱️ -19s For more details on these failures, see this check. Results for commit 0ef0108. ± Comparison against base commit efc2ddf. |
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (5)
applications/tari_indexer/src/storage_sqlite/migrations/2023-02-16-145719_initial/up.sql (3)
17-29: Add FK on substate_transitions.substate_id → substates(address) with ON DELETE CASCADE.This preserves referential integrity and enables cascaded cleanup during reorgs/resyncs. The past review already requested this; the shard/state_version index is now present (nice), but the FK is still missing.
Apply this diff within the table definition:
value_hash text NULL, - created_at timestamp not null default current_timestamp + created_at timestamp not null default current_timestamp, + FOREIGN KEY (substate_id) REFERENCES substates (address) ON DELETE CASCADE );
107-115: Unify shard_group type: use INTEGER to match scanned_block_ids.Inconsistent types (TEXT here vs INTEGER in scanned_block_ids) hurt joins and index reuse. This was flagged previously and still applies.
CREATE TABLE epoch_checkpoints ( id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, epoch BIGINT NOT NULL, - shard_group TEXT NOT NULL, + shard_group INTEGER NOT NULL, json_data TEXT NOT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP );
119-136: Enforce UTXO uniqueness per (substate_id, version) and align indexes with get_utxo_updates.
- Essential: prevent duplicates for the same UTXO logical revision.
- Optional: add a lighter composite for tag-less scans if common.
create table utxos ( @@ created_at timestamp not null default current_timestamp ); -CREATE INDEX utxos_shard_tag_resource_state_version_idx - ON utxos (shard, utxo_tag_byte, resource_address, state_version); +CREATE INDEX utxos_shard_tag_resource_state_version_idx + ON utxos (shard, utxo_tag_byte, resource_address, state_version); + +-- Ensure one logical UTXO version per substate +CREATE UNIQUE INDEX utxos_substate_id_version_uniq + ON utxos (substate_id, version); + +-- Optional: when tag filter is omitted +--CREATE INDEX utxos_shard_resource_state_version_idx +-- ON utxos (shard, resource_address, state_version);applications/tari_indexer/src/network_state_sync/block_scanner.rs (1)
362-386: Fix timestamp fallback: log says UNIX_EPOCH but returns MAX; also avoid i64::MAX sentinel.Returning
PrimitiveDateTime::MAXcontradicts the log (“Using UNIX_EPOCH”) and can poison ordering/retention logic with far-future timestamps. Fall back to UNIX_EPOCH consistently and drop the i64::MAX sentinel path.-fn unix_epoch_to_primitive_date_time(timestamp: u64) -> PrimitiveDateTime { - let timestamp = i64::try_from(timestamp).unwrap_or_else(|e| { - // TODO: this is very possible because we trust that the timestamp is roughly correct, however - // it is purely informational and not enforced in consensus therefore could be any value and - // therefore cannot be relied for ordering (use (epoch,height) instead). - warn!( - target: LOG_TARGET, - "Failed to convert block timestamp to PrimitiveDateTime: {}", - e - ); - i64::MAX // = August 17, 292278994, 07:12:55.807 UTC - }); - OffsetDateTime::from_unix_timestamp(timestamp) - .map(|osdt| PrimitiveDateTime::new(osdt.date(), osdt.time())) - .unwrap_or_else(|e| { - warn!( - target: LOG_TARGET, - "Failed to convert block timestamp to OffsetDateTime: {}. Using UNIX_EPOCH", - e - ); - // An error cannot be because the timestamp is too small, because we use an u64 and a zero unix - // timestamp represents a greater date (1970 AD) than the minimum (9999 BC) - PrimitiveDateTime::MAX - }) -} +fn unix_epoch_to_primitive_date_time(ts: u64) -> PrimitiveDateTime { + let unix_epoch = { + let osdt = OffsetDateTime::UNIX_EPOCH; + PrimitiveDateTime::new(osdt.date(), osdt.time()) + }; + let Ok(ts_i64) = i64::try_from(ts) else { + warn!(target: LOG_TARGET, "Invalid block timestamp (u64->i64 overflow): {}. Using UNIX_EPOCH.", ts); + return unix_epoch; + }; + match OffsetDateTime::from_unix_timestamp(ts_i64) { + Ok(osdt) => PrimitiveDateTime::new(osdt.date(), osdt.time()), + Err(e) => { + warn!(target: LOG_TARGET, "Invalid block timestamp value {}: {}. Using UNIX_EPOCH.", ts_i64, e); + unix_epoch + } + } +}crates/rpc_state_sync/src/state_sync.rs (1)
129-131: Duplicate not_found arm from a prior revision is resolved.The earlier duplicated Err(RpcError::RequestFailed(err)) if err.is_not_found() => Ok(None) arm is gone; the match coverage is now clean.
🧹 Nitpick comments (30)
applications/tari_indexer/src/storage_sqlite/migrations/2023-02-16-145719_initial/up.sql (9)
31-33: Index for sync scans is present — minor nit: index name mismatch with covered columns.The composite index on (shard, state_version) addresses the resumable scan path. However, the unique index name suggests only (substate_id, version) while it includes is_up. Consider renaming for clarity in future maintenance.
-create unique index substate_transitions_substate_id_version_uniq on substate_transitions (substate_id, version, is_up); +create unique index substate_transitions_substate_id_version_is_up_uniq on substate_transitions (substate_id, version, is_up);
45-50: Drop redundant non-unique index on non_fungible_indexes.The UNIQUE index (resource_address, idx) already provides an index. The additional non-unique index on the exact same columns duplicates storage and maintenance cost with no benefit.
--- a/applications/tari_indexer/src/storage_sqlite/migrations/2023-02-16-145719_initial/up.sql +++ b/applications/tari_indexer/src/storage_sqlite/migrations/2023-02-16-145719_initial/up.sql @@ -- DB index for faster collection scan queries -create index nft_indexes_resource on non_fungible_indexes (resource_address, idx); +-- (removed) Redundant with uniq_nft_indexes
51-61: Consider FK on events.substate_id with ON DELETE SET NULL.Events optionally reference a substate. An FK guards against typos and keeps rows consistent when substates are pruned. Using SET NULL maintains historical events.
substate_id text NULL, - created_at timestamp not null default current_timestamp + created_at timestamp not null default current_timestamp, + FOREIGN KEY (substate_id) REFERENCES substates (address) ON DELETE SET NULL );
64-66: Add an index for events.substate_id (optional).If queries fetch events by substate, this index avoids table scans.
create index events_indexer on events (template_address, tx_hash); +create index events_substate_id_idx on events (substate_id);
78-83: Drop redundant non-unique index on scanned_block_ids.The UNIQUE (epoch, shard_group) already creates an index. The extra non-unique index on the same columns is redundant.
create unique index scanned_block_ids_unique_committee on scanned_block_ids (epoch, shard_group); - --- DB index for faster retrieval of the latest block by committee -create index scanned_block_ids_committee on scanned_block_ids (epoch, shard_group);
129-131: Normalize boolean columns to INTEGER with CHECK constraints.SQLite doesn’t have a native BOOLEAN type; enforce 0/1 to avoid surprises and accidental non-boolean values.
- is_spent boolean not NULL, - is_burnt boolean not NULL, - is_frozen boolean not NULL, + is_spent INTEGER not NULL CHECK (is_spent IN (0,1)), + is_burnt INTEGER not NULL CHECK (is_burnt IN (0,1)), + is_frozen INTEGER not NULL CHECK (is_frozen IN (0,1)),
95-102: updated_at columns won’t auto-update without triggers.key_values.updated_at and epoch_checkpoints.updated_at default only on INSERT. If you rely on last-modified timestamps, add triggers.
Example triggers to add after the table definitions:
CREATE TRIGGER key_values_set_updated_at AFTER UPDATE ON key_values FOR EACH ROW BEGIN UPDATE key_values SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END; CREATE TRIGGER epoch_checkpoints_set_updated_at AFTER UPDATE ON epoch_checkpoints FOR EACH ROW BEGIN UPDATE epoch_checkpoints SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END;Also applies to: 106-115
10-12: Consistency nit: prefer DATETIME type name across the schema.You mix timestamp/DATETIME; in SQLite both map to NUMERIC affinity, but consistent naming improves readability and tooling expectations.
- timestamp timestamp not NULL, - updated_at timestamp not null default current_timestamp, - created_at timestamp not null default current_timestamp + timestamp DATETIME not NULL, + updated_at DATETIME not null default current_timestamp, + created_at DATETIME not null default current_timestamp
119-136: Optional: add an index to accelerate UTXO lookups by substate_id.If you frequently check UTXO status by substate_id alone, a direct index helps. The unique (substate_id, version) also works for equality, but an explicit single-column index can be marginally smaller and planner-friendly if version is not involved.
CREATE INDEX utxos_shard_tag_resource_state_version_idx ON utxos (shard, utxo_tag_byte, resource_address, state_version); +CREATE INDEX utxos_substate_id_idx ON utxos (substate_id);applications/tari_indexer/src/network_state_sync/block_scanner.rs (6)
49-51: Metric/log wording and counting are inconsistent; count updates, not blocks.
scan()logs “Scanned {} events” butcountwas based on blocks. Count substate updates instead and adjust the log to avoid confusion.@@ pub async fn scan(&self) -> Result<usize, anyhow::Error> { - let mut block_count = 0; + let mut update_count = 0; @@ - "Scanned {} events", - block_count + "Scanned {} substate updates", + update_count ); - Ok(block_count) + Ok(update_count) @@ async fn scan_blocks_in_epoch(&self, epoch: Epoch) -> Result<usize, anyhow::Error> { - let mut count = 0; + let mut count = 0usize; @@ - info!( + info!( target: LOG_TARGET, "Scanned {} blocks in epoch={}", new_blocks.len(), epoch, ); - count += new_blocks.len(); for block_data in new_blocks { let timestamp = unix_epoch_to_primitive_date_time(block_data.block.timestamp()); + // Count the number of substate updates processed + count += block_data.diff.len(); self.store_substates_in_db(&block_data.diff, timestamp)?; }Also applies to: 74-81, 99-106
335-340: Extract 10,000-substate ceiling into a named constant (and consider making it configurable).Improves readability and makes future tuning simpler.
@@ - // TODO: what should this limit be? - if num_substates > 10_000 { + // TODO: consider making this configurable + const MAX_SUBSTATES_PER_BLOCK: usize = 10_000; + if num_substates > MAX_SUBSTATES_PER_BLOCK { return Err(anyhow::anyhow!( - "Exceeded maximum number of substates (10,000). Got {}", - num_substates + "Exceeded maximum number of substates ({}). Got {}", + MAX_SUBSTATES_PER_BLOCK, + num_substates, )); }
147-151: Avoid storing empty string for missing substate data; prefer NULL/Option.Persisting
""for pruned substates conflates “no data” with “empty data”. If the schema allows, changeNewSubstate.datatoOption<String>and storeNone.I can draft the schema/model tweak and the ripple changes if you want this in this PR.
155-160: Reduce log noise: don’t dump the entire substate payload at debug.
{:?}onNewSubstatelikely includes the JSON blob and increases log volume. Log the address/version and sizes instead.- debug!( - target: LOG_TARGET, - "Saving substate: {:?}", - substate_row - ); + debug!( + target: LOG_TARGET, + "Saving substate addr={}, ver={}, data_len={}", + create.substate.substate_id, + create.substate.version, + substate_row.data.len(), + );
11-16: Prefer importing time types from the time crate directly to reduce coupling.Using
tari_ootle_storage::time::{...}couples this module to storage. Importing fromtimekeeps concerns separated.- time::{OffsetDateTime, PrimitiveDateTime}, + time::{OffsetDateTime, PrimitiveDateTime},If
tari_ootle_storageintentionally re-exports time to pin versions, feel free to keep as-is.
162-163: Destroy updates are ignored; confirm if tombstoning/removal is required.If the indexer should reflect deletions, consider recording a tombstone or removing the substate on
Destroy.Would you like me to propose a
tx.delete_substate(...)ortx.upsert_tombstone(...)flow behind a feature flag?crates/rpc_state_sync/src/state_sync.rs (8)
36-36: Confirm breadth of value_filters; consider narrowing or making configurable.You’re requesting SubstateValueFilterFlags::all().bits() for every sync_state stream. That maximizes payload sizes and CPU at both ends. If only specific substate types need in-value materialization during state sync, prefer a targeted mask or a config-driven mask.
Apply either approach:
- until_epoch: Some(checkpoint.epoch().into()), - value_filters: SubstateValueFilterFlags::all().bits(), + until_epoch: Some(checkpoint.epoch().into()), + // Prefer a curated mask or inject via config + value_filters: SubstateValueFilterFlags::COMPONENT + .union(SubstateValueFilterFlags::RESOURCE) + .bits(),If COMPONENТ/RESOURCE is not the right combo, please wire a node config knob and thread it here.
Also applies to: 167-169
85-88: Connection creation is fine; consider hedged timeouts/reuse for robustness.Current code dials per peer, per attempt. If practical, add a dial timeout and/or client reuse to avoid burning attempts on slow peers and to reduce handshake overhead.
109-131: Avoid expect in checkpoint pop; fold empty handling and capture metrics.Minor: eliminate the expect by folding empty handling into the same arm. Also, recording the source (“store” vs “rpc”) and number of returned checkpoints will help observability.
- Ok(GetCheckpointsResponse { checkpoints }) if checkpoints.is_empty() => Ok(None), - Ok(GetCheckpointsResponse { mut checkpoints }) => { - match EpochCheckpoint::try_from(checkpoints.pop().expect("checked is_empty")) { + Ok(GetCheckpointsResponse { mut checkpoints }) => { + let Some(last) = checkpoints.pop() else { return Ok(None); }; + match EpochCheckpoint::try_from(last) { Ok(checkpoint) => { info!(target: LOG_TARGET, "🛜 Checkpoint: {checkpoint}"); self.validate_checkpoint(&checkpoint, prev_committee, prev_epoch)?; - self.state_store.with_write_tx(|tx| checkpoint.save(tx))?; - self.valid_checkpoints.insert(for_shard_group, checkpoint.clone()); + self.state_store.with_write_tx(|tx| checkpoint.save(tx))?; + // Optional: record that this came from RPC, and how many were returned + // self.stats.last_checkpoint_source = Some("rpc".into()); + // self.stats.last_checkpoint_count = Some(checkpoints_len); Ok(Some(checkpoint)) },
232-244: Fix “new template(s)” log to report per-batch, not cumulative; pre-compute update count.The message currently prints template_changes.len(), which is cumulative across the session and can be misleading; also compute the update count before consuming the iterator.
- info!(target: LOG_TARGET, "🛜 Buffering {} state update(s) (state version: v{})", updates_for_state_version.len(), state_version); - for result in updates_for_state_version { + let num_updates = updates_for_state_version.len(); + info!(target: LOG_TARGET, "🛜 Buffering {} state update(s) (state version: v{})", num_updates, state_version); + let prev_templates = template_changes.len(); + for result in updates_for_state_version { let update = result?; let (tree_change, template_change) = extract_tree_and_template_changes(msg_epoch, &update)?; debug!(target: LOG_TARGET, "🛜 -> state update (v{}) {}", state_version, update); template_changes.extend(template_change); tree_changes.push(tree_change); updates.push(update); } - - info!(target: LOG_TARGET, "🛜 Sync: {} state update(s), {} new template(s) (state version: v{})", updates.len(), template_changes.len(), state_version); + let new_templates = template_changes.len().saturating_sub(prev_templates); + info!(target: LOG_TARGET, "🛜 Sync: {} state update(s), {} new template(s) (state version: v{})", updates.len(), new_templates, state_version);
279-314: Early-exit the stream once the checkpoint state version is verified to avoid reading trailing items.Not critical, but breaking the loop immediately after matching the checkpoint avoids extra polling round-trips if a peer keeps the stream open.
- if state_version == checkpoint_state_version { + if state_version == checkpoint_state_version { if local_state_root != checkpoint_shard_root { error!(target: LOG_TARGET, "❌ State root mismatch ..."); // rollback! return Err(RpcStateSyncError::StateRootMismatch { expected: checkpoint_shard_root, actual: local_state_root, }); } info!(target: LOG_TARGET, "🛜 ✅ State root for {shard} matches checkpoint: {local_state_root} (v{state_version})",); - - maybe_persisted_state_version = Some(state_version); - store.set_state_version(state_version)?; - // Done - return Ok(()); + maybe_persisted_state_version = Some(state_version); + store.set_state_version(state_version)?; + // Signal to break out of the outer loop + return Ok(()); }And immediately after with_write_tx, break out if we just reached the checkpoint:
- })?; + })?; + if last_state_version == checkpoint_state_version { + break; + }
61-62: Remove unused valid_checkpoints field and related maintenance.Since you now prefer the persisted fast-path (EpochCheckpoint::get_by_shard_group), the in-memory valid_checkpoints map is no longer read anywhere in this file. Keeping it adds memory churn and dead code paths.
struct RpcStateSyncClientProtocol<TConsensusSpec: ConsensusSpec> { epoch_manager: TConsensusSpec::EpochManager, state_store: TConsensusSpec::StateStore, client_factory: TariValidatorNodeRpcClientFactory, template_manager: TemplateManagerHandle, - valid_checkpoints: HashMap<ShardGroup, EpochCheckpoint>, stats: StateSyncStats, } @@ Self { epoch_manager, state_store, client_factory, template_manager, - valid_checkpoints: HashMap::new(), stats: StateSyncStats::default(), } } @@ - self.state_store.with_write_tx(|tx| checkpoint.save(tx))?; - self.valid_checkpoints.insert(for_shard_group, checkpoint.clone()); + self.state_store.with_write_tx(|tx| checkpoint.save(tx))?; Ok(Some(checkpoint)) }, @@ - // Clear the valid checkpoints cache - self.valid_checkpoints = HashMap::new(); self.stats = StateSyncStats::default(); return Err(err); @@ - // Clear the valid checkpoints cache - self.valid_checkpoints = HashMap::new(); self.stats = StateSyncStats::default();If another module uses valid_checkpoints, keep it and read from it before hitting the store; otherwise, delete.
Also applies to: 79-81, 123-124, 645-646, 653-654
461-467: Follow-up TODO: robust “first epoch” detection.Relying on a single peer returning not_found could stall progress. Consider:
- Ask f + 1 peers before concluding “no checkpoint”.
- Query epoch_manager for “is_first_active_epoch(prev_epoch)”.
I can wire a hedged RPC (fanout to min quorum with early-cancel on first success).
525-528: Comment is outdated with current logic; clarify or adjust.Because sync_global_shard persists the checkpoint under ShardGroup::all_shards and get_or_fetch_valid_epoch_checkpoint does a store fast-path, subsequent committee iterations won’t refetch from the network. Update the note accordingly.
- // TODO: any checkpoint for the previous epoch will justify the global shard sync. - // Currently we'll fetch the checkpoint again even if we already have it if there are more than one - // shard groups. + // NOTE: Any checkpoint for the previous epoch justifies the global shard sync. + // Since we persist under ShardGroup::all_shards, subsequent committees will hit the store fast-path + // and avoid additional network fetches.crates/state_store_rocksdb/src/codecs/substate_lock.rs (7)
26-31: Make the error variant and message more specific to TransactionId.Other codecs in this crate use
MalformedData { operation, details }for fixed-size reads. Consider using that variant and naming the operation/type explicitly for clearer diagnostics.Apply:
- let buf = read_to_fixed(reader).ok_or_else(|| RocksDbStorageError::DecodeError { - source: anyhow!("SubstateLockKeyCodec: Invalid bytes for FixedHash"), - })?; + let buf = read_to_fixed(reader).ok_or_else(|| RocksDbStorageError::MalformedData { + operation: "decode TransactionId", + details: "Invalid bytes for TransactionId (expected 32)".to_string(), + })?;
33-38: Similarly, tailor the error to BlockId and preferMalformedData.This will keep errors consistent and self-explanatory across codecs.
Apply:
- let buf = read_to_fixed(reader).ok_or_else(|| RocksDbStorageError::DecodeError { - source: anyhow!("SubstateLockKeyCodec: Invalid bytes for FixedHash"), - })?; + let buf = read_to_fixed(reader).ok_or_else(|| RocksDbStorageError::MalformedData { + operation: "decode BlockId", + details: "Invalid bytes for BlockId (expected 32)".to_string(), + })?;
40-46: Nice: added explicit context when delegating toSubstateIdCodec. Preserve the source error chain.You already attach context, which is great. Tiny improvement: construct
anyhow::Errorfromeinstead of interpolating it into a string so the original error remains the source in the chain.Apply:
- self.substate_id_codec - .decode_reader(reader) - .map_err(|e| RocksDbStorageError::DecodeError { - source: anyhow!("SubstateLockKeyCodec: Failed to decode SubstateId: {}", e), - }) + self.substate_id_codec + .decode_reader(reader) + .map_err(|e| RocksDbStorageError::DecodeError { + // Keep the original error as the source; attach context here + source: anyhow::Error::new(e).context("SubstateLockKeyCodec: Failed to decode SubstateId"), + })Note: add
use anyhow::Context;at the top if it’s not already imported.
48-53: Consistency: construct NodeHeight via From and considerMalformedData.Other number codecs use
NodeHeight::from(u64)andMalformedData. Adopt the same style for uniformity.Apply:
- let height = read_to_fixed(reader).ok_or_else(|| RocksDbStorageError::DecodeError { - source: anyhow!("SubstateLockKeyCodec: Invalid bytes for NodeHeight"), - })?; - Ok(NodeHeight(u64::from_be_bytes(height))) + let height = read_to_fixed(reader).ok_or_else(|| RocksDbStorageError::MalformedData { + operation: "decode NodeHeight", + details: "Invalid bytes for NodeHeight (expected 8)".to_string(), + })?; + Ok(NodeHeight::from(u64::from_be_bytes(height)))
70-81: Optionally reject trailing bytes to catch malformed keys early.If these keys are decoded from exact RocksDB values, detecting extra bytes helps spot encode/decode mismatches sooner.
Apply:
- Ok(SubstateLockKey { - block_id, - substate_id, - transaction_id, - block_height, - }) + let key = SubstateLockKey { + block_id, + substate_id, + transaction_id, + block_height, + }; + // Optional: ensure the reader is exhausted (no trailing bytes) + let mut _tmp = [0u8; 1]; + match reader.read(&mut _tmp) { + Ok(0) => {} + Ok(_) => { + return Err(RocksDbStorageError::MalformedData { + operation: "decode SubstateLockKey", + details: "Trailing bytes after expected fields".to_string(), + }) + } + Err(e) => { + return Err(RocksDbStorageError::MalformedData { + operation: "decode SubstateLockKey", + details: format!("I/O while checking trailing bytes: {e}"), + }) + } + } + Ok(key)
98-110: Same optional trailing-bytes guard for this permutation.Apply the same pattern as above after assembling the struct:
- Ok(SubstateLockKey { - block_id, - substate_id, - transaction_id, - block_height, - }) + let key = SubstateLockKey { + block_id, + substate_id, + transaction_id, + block_height, + }; + let mut _tmp = [0u8; 1]; + match reader.read(&mut _tmp) { + Ok(0) => {} + Ok(_) => { + return Err(RocksDbStorageError::MalformedData { + operation: "decode SubstateLockKey", + details: "Trailing bytes after expected fields".to_string(), + }) + } + Err(e) => { + return Err(RocksDbStorageError::MalformedData { + operation: "decode SubstateLockKey", + details: format!("I/O while checking trailing bytes: {e}"), + }) + } + } + Ok(key)
127-139: Mirror the trailing-bytes check on the third permutation for consistency.Apply:
- Ok(SubstateLockKey { - block_id, - substate_id, - transaction_id, - block_height, - }) + let key = SubstateLockKey { + block_id, + substate_id, + transaction_id, + block_height, + }; + let mut _tmp = [0u8; 1]; + match reader.read(&mut _tmp) { + Ok(0) => {} + Ok(_) => { + return Err(RocksDbStorageError::MalformedData { + operation: "decode SubstateLockKey", + details: "Trailing bytes after expected fields".to_string(), + }) + } + Err(e) => { + return Err(RocksDbStorageError::MalformedData { + operation: "decode SubstateLockKey", + details: format!("I/O while checking trailing bytes: {e}"), + }) + } + } + Ok(key)
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (6)
applications/tari_indexer/src/lib.rs(6 hunks)applications/tari_indexer/src/network_state_sync/block_scanner.rs(1 hunks)applications/tari_indexer/src/storage_sqlite/migrations/2023-02-16-145719_initial/up.sql(1 hunks)crates/rpc_state_sync/src/state_sync.rs(10 hunks)crates/state_store_rocksdb/src/codecs/misc.rs(9 hunks)crates/state_store_rocksdb/src/codecs/substate_lock.rs(5 hunks)
🚧 Files skipped from review as they are similar to previous changes (2)
- crates/state_store_rocksdb/src/codecs/misc.rs
- applications/tari_indexer/src/lib.rs
🧰 Additional context used
🧬 Code graph analysis (3)
applications/tari_indexer/src/network_state_sync/block_scanner.rs (4)
crates/storage/src/consensus_models/substate_change.rs (1)
substate(58-63)applications/tari_indexer/src/storage_sqlite/store_factory.rs (1)
updates(619-636)crates/storage/src/consensus_models/substate.rs (1)
as_versioned_substate_id_ref(382-384)crates/engine_types/src/events.rs (1)
template_address(120-122)
crates/state_store_rocksdb/src/codecs/substate_lock.rs (2)
crates/state_store_rocksdb/src/utils.rs (1)
read_to_fixed(27-35)crates/state_store_rocksdb/src/codecs/misc.rs (8)
decode_reader(22-24)decode_reader(35-44)decode_reader(61-70)decode_reader(78-84)decode_reader(92-98)decode_reader(106-113)decode_reader(121-128)decode_reader(136-143)
crates/rpc_state_sync/src/state_sync.rs (2)
crates/storage/src/consensus_models/epoch_checkpoint.rs (1)
get_by_shard_group(181-187)crates/p2p/src/conversions/rpc.rs (7)
try_from(29-37)try_from(52-57)try_from(73-79)try_from(96-105)try_from(132-141)try_from(149-162)try_from(182-187)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (5)
- GitHub Check: check stable
- GitHub Check: test
- GitHub Check: machete
- GitHub Check: check nightly
- GitHub Check: clippy
🔇 Additional comments (7)
applications/tari_indexer/src/storage_sqlite/migrations/2023-02-16-145719_initial/up.sql (2)
1-16: Substates uniqueness/semantics look good for “latest-only” storage.Unique(address) suggests substates holds the latest value while versions live in substate_transitions. This matches typical indexer patterns.
17-33: Drop the misplaced index‐tuning suggestionThe migration in
up.sqlcreates only thesubstate_transitionstable and two indexes on(substate_id, version, is_up)and(shard, state_version). The earlier feedback about right-sizing composite indexes forget_utxo_updates(which operates on theutxostable and itsutxo_tag_byte/resource_addressfilters) does not apply here—this script neither touches theutxostable nor affects those query paths. Please ignore or relocate that comment to the relevant migration or query implementation forutxos.Likely an incorrect or invalid review comment.
crates/rpc_state_sync/src/state_sync.rs (4)
26-26: RPC request/response rename adoption looks good.The switch to GetCheckpointsRequest/GetCheckpointsResponse aligns with the new multi-checkpoint API. No issues spotted here.
97-105: Pre-RPC checkpoint fast-path via state store is a solid improvement.Replacing the in-memory cache fast-path with a persisted read avoids redundant network calls after restarts. Looks good.
696-728: New helper cleanly encapsulates tree change + template extraction.This replaces scattered logic and keeps the main loop focused. The use of Option with extend is neat and efficient.
417-425: Visibility reduction to private forsync_shardis safeNo external call sites to
sync_shardwere found outside ofcrates/rpc_state_sync/src/state_sync.rs(verified via ripgrep), so reducing its visibility to private will not break any modules or tests.crates/state_store_rocksdb/src/codecs/substate_lock.rs (1)
4-5: Confirmed: Alldecode(&[u8])calls for SubstateLockKey have been removed
I ran the grep acrosscrates/**/*.rsand found zero occurrences of.decode(forSubstateLockKey. The only remaining use is the intended.decode_reader(reader)call in
crates/state_store_rocksdb/src/codecs/substate_lock.rs, so downstream code is fully aligned with the streaming API.
Merging is good to go!
2 participants
Description
feat(indexer)!: implement resumable state sync
feat(indexer/jrpc): utxo scan api
feat(indexer)!: sync epoch checkpoints
feat(indexer)!: implement transaction and event scanning to state sync worker
feat(validator/p2p)!: allow values for specific substate types to be returned over the wire
fix(validator/storage): minor improvement in DB codecs (perf, maint)
fix(consensus): bug causing new genesis block to be created after each restart
fix(bindings): remove export_to everywhere except where needed
Motivation and Context
This PR allows clients (wallets) to retrieve changes to UTXOs (either unspent or spent) with specific UTXO tags across shards.
REQUEST:
RESPONSE:
How Has This Been Tested?
Manually
Breaking Changes
Summary by CodeRabbit
New Features
API Changes
Data / Migrations
Performance / Stability