feat: detect chain reorgs #411
Conversation
WalkthroughThis pull request introduces comprehensive changes to handle blockchain reorganization events across multiple components of the system. The modifications span database schema, indexing logic, log streaming, and mock implementations. The primary focus is on enhancing the system's ability to detect and respond to blockchain state changes, particularly when block histories are modified. Key additions include a new Changes
Sequence DiagramsequenceDiagram
participant Client
participant RpcLogStreamer
participant Indexer
participant ReorgHandler
participant Database
Client->>RpcLogStreamer: ListenForContractEvent
RpcLogStreamer-->>Client: eventChannel, reorgChannel
Indexer->>RpcLogStreamer: Watch contract events
RpcLogStreamer->>Indexer: Emit log events
alt Potential Reorganization
Indexer->>ReorgHandler: FindReorgPoint
ReorgHandler->>Database: Retrieve block range
Database-->>ReorgHandler: Block information
ReorgHandler->>Indexer: Reorganization point
Indexer->>Database: Update block canonicality
end
Assessment against linked issues
Possibly related PRs
Suggested reviewers
✨ Finishing Touches
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
🧹 Nitpick comments (2)
pkg/indexer/indexer.go (1)
225-236: Potential Performance Impact Due to Frequent Block FetchingFetching the block using
client.BlockByNumberfor every event can significantly impact performance, especially when processing a high volume of events.Consider the following optimization:
- Batch Processing: Fetch block hashes in batches or at set intervals instead of per event.
- Caching: Implement a caching mechanism to store recently fetched block hashes.
Example code snippet for caching:
var blockHashCache = make(map[uint64][]byte) // Before fetching the block if hash, exists := blockHashCache[storedBlockNumber]; exists { // Use the cached hash latestBlockHash = hash } else { // Fetch and cache the block hash block, err := client.BlockByNumber(ctx, big.NewInt(int64(storedBlockNumber))) if err != nil { // Handle error } latestBlockHash = block.Hash().Bytes() blockHashCache[storedBlockNumber] = latestBlockHash }pkg/migrations/00004_update-gateway-envelopes.sql (1)
1-5: Add indexes for chain reorganization queries.Consider adding indexes on
(block_number, block_hash)andis_canonicalto optimize queries during chain reorganization handling.CREATE INDEX idx_gateway_envelopes_block ON gateway_envelopes(block_number, block_hash); CREATE INDEX idx_gateway_envelopes_canonical ON gateway_envelopes(is_canonical);
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (8)
pkg/blockchain/rpcLogStreamer.go(4 hunks)pkg/blockchain/rpcLogStreamer_test.go(2 hunks)pkg/db/queries.sql(1 hunks)pkg/db/queries/models.go(1 hunks)pkg/db/queries/queries.sql.go(5 hunks)pkg/indexer/indexer.go(8 hunks)pkg/indexer/storer/groupMessage.go(1 hunks)pkg/migrations/00004_update-gateway-envelopes.sql(1 hunks)
✅ Files skipped from review due to trivial changes (1)
- pkg/blockchain/rpcLogStreamer_test.go
🔇 Additional comments (2)
pkg/db/queries/models.go (1)
25-28: Ensure Consistency of New Fields inGatewayEnvelopeStructThe addition of
BlockNumber,BlockHash,Version, andIsCanonicalfields to theGatewayEnvelopestruct is appropriate for tracking blockchain event metadata. Ensure that these fields are consistently populated throughout the application and properly handled in database operations.Please verify that:
- The database migrations have been applied to include the new columns.
- The fields are correctly set when inserting new
GatewayEnveloperecords.- Any queries or usages of
GatewayEnvelopeare updated to accommodate the new fields.pkg/db/queries/queries.sql.go (1)
142-143: LGTM! Generated code matches schema changes.The generated code correctly reflects the database schema changes. No manual modifications needed as this is an auto-generated file.
Also applies to: 153-156, 165-168, 246-246, 280-283
fbac
force-pushed
the
feat/reorg-detection
branch
2 times, most recently
from
4a96e4f to
a99d582
Compare
There was a problem hiding this comment.
Caution
Inline review comments failed to post. This is likely due to GitHub's limits when posting large numbers of comments.
Actionable comments posted: 5
🔭 Outside diff range comments (2)
pkg/indexer/storer/groupMessage_test.go (1)
Line range hint
1-143: Add dedicated test cases for chain reorganization scenarios.The test suite should include dedicated test cases to verify the system's behavior during chain reorganizations. Consider adding these test cases:
Chain Reorg Detection
- Test detection of non-canonical blocks
- Test handling of orphaned blocks
- Test recovery from deep reorgs
Edge Cases
- Test concurrent reorgs
- Test invalid block sequences
- Test missing block information
Would you like me to provide example implementations for these test cases?
pkg/db/queries/queries.sql.go (1)
Line range hint
165-191: Reconsider the conflict resolution strategy.The current
ON CONFLICT DO NOTHINGstrategy might not be optimal for handling chain reorganizations. Consider usingON CONFLICT DO UPDATEto update the block information, version, and canonical status when conflicts occur, ensuring we don't silently ignore potential reorg updates.Example modification:
-ON CONFLICT - DO NOTHING +ON CONFLICT (originator_node_id, originator_sequence_id) + DO UPDATE SET + block_number = EXCLUDED.block_number, + block_hash = EXCLUDED.block_hash, + version = EXCLUDED.version, + is_canonical = EXCLUDED.is_canonical + WHERE gateway_envelopes.version < EXCLUDED.version
♻️ Duplicate comments (1)
pkg/indexer/indexer.go (1)
224-259:⚠️ Potential issueCritical: Incomplete reorg detection and handling
The current implementation has several critical issues:
- The safe block distance check
event.BlockNumber-storedBlockNumber < safeBlockDistancemight miss reorgs that happen within the safe distance.- The reorg channel send is commented out, leaving the reorg handling incomplete.
- The TODO items outline critical steps that need to be implemented.
Consider implementing the following changes:
if !bytes.Equal(storedBlockHash, latestBlock.Hash().Bytes()) { logger.Warn("blockchain reorg detected", zap.Uint64("storedBlockNumber", storedBlockNumber), zap.String("storedBlockHash", hex.EncodeToString(storedBlockHash)), zap.String("onchainBlockHash", latestBlock.Hash().String()), ) - // TODO: Implement reorg handling: - // reorgChannel <- event.BlockNumber + // Send the reorg event + select { + case reorgChannel <- storedBlockNumber: + logger.Info("sent reorg event", + zap.Uint64("blockNumber", storedBlockNumber)) + default: + logger.Warn("reorg channel full, skipping reorg event", + zap.Uint64("blockNumber", storedBlockNumber)) + } }
🧹 Nitpick comments (2)
pkg/db/queries/queries.sql.go (2)
243-262: Enhance error handling for envelope invalidation.The function could benefit from:
- Returning the number of rows affected to verify if any envelope was actually invalidated
- Explicit handling of the case where no rows are updated (might indicate a race condition)
-func (q *Queries) InvalidateEnvelope(ctx context.Context, arg InvalidateEnvelopeParams) error { - _, err := q.db.ExecContext(ctx, invalidateEnvelope, arg.OriginatorNodeID, arg.OriginatorSequenceID) - return err +func (q *Queries) InvalidateEnvelope(ctx context.Context, arg InvalidateEnvelopeParams) (int64, error) { + result, err := q.db.ExecContext(ctx, invalidateEnvelope, arg.OriginatorNodeID, arg.OriginatorSequenceID) + if err != nil { + return 0, err + } + return result.RowsAffected() +}
Line range hint
290-327: Consider additional filtering options for envelope selection.The query could benefit from optional filters for:
is_canonicalstatus to filter out invalidated envelopesblock_numberrange to help with reorg analysisversionto track envelope historyThis would require updating the stored procedure
select_gateway_envelopesto accept these additional parameters.
🛑 Comments failed to post (5)
pkg/indexer/storer/groupMessage_test.go (2)
62-64: 🛠️ Refactor suggestion
Add test coverage for chain reorganization fields.
The test should be updated to verify the new fields (
block_number,block_hash,version,is_canonical) that were added to handle chain reorganizations. Currently, it only validates the envelope content and sequence ID.Add assertions to verify these fields:
require.NoError(t, err) +require.NotNil(t, firstEnvelope.BlockNumber, "Block number should be set") +require.NotEmpty(t, firstEnvelope.BlockHash, "Block hash should be set") +require.Equal(t, int32(1), firstEnvelope.Version, "Initial version should be 1") +require.True(t, firstEnvelope.IsCanonical, "Message should be marked as canonical")Committable suggestion skipped: line range outside the PR's diff.
111-113: 🛠️ Refactor suggestion
Enhance duplicate message test for chain reorganization scenarios.
The duplicate message test should be expanded to verify the version increment and canonicality updates when the same message is received from different blocks (chain reorg scenario).
Consider adding these test cases:
require.NoError(t, queryErr) require.Equal(t, len(envelopes), 1) + +// Verify version and canonicality handling +envelope := envelopes[0] +require.Equal(t, int32(2), envelope.Version, "Version should be incremented for duplicate") +require.True(t, envelope.IsCanonical, "Message should remain canonical") + +// Test reorg scenario with different block +reorgLog := logMessage +reorgLog.BlockNumber += 1 +reorgLog.BlockHash = common.HexToHash("0xdifferent") +err = storer.StoreLog(ctx, reorgLog) +require.NoError(t, err) + +reorgEnvelopes, err := storer.db.SelectGatewayEnvelopes(ctx, queries.SelectGatewayEnvelopesParams{ + OriginatorNodeIds: []int32{0}, +}) +require.NoError(t, err) +require.Equal(t, 1, len(reorgEnvelopes), "Should still have one envelope") + +reorgEnvelope := reorgEnvelopes[0] +require.Equal(t, int32(3), reorgEnvelope.Version, "Version should be incremented") +require.Equal(t, reorgLog.BlockNumber, reorgEnvelope.BlockNumber, "Block number should be updated") +require.Equal(t, reorgLog.BlockHash.Hex(), reorgEnvelope.BlockHash, "Block hash should be updated")Committable suggestion skipped: line range outside the PR's diff.
pkg/indexer/storer/logAppend.go (1)
12-49:
⚠️ Potential issueWrap database operations in a transaction for atomicity.
The function performs multiple database operations (get version and invalidate envelope) that should be atomic to prevent race conditions and ensure data consistency during chain reorganizations.
Consider wrapping the operations in a transaction:
func GetVersionForAppend( ctx context.Context, querier *queries.Queries, logger *zap.Logger, originatorNodeID int32, sequenceID int64, ) (sql.NullInt32, error) { var version sql.NullInt32 + + // Start transaction + tx, err := querier.Begin(ctx) + if err != nil { + logger.Error("Error starting transaction", zap.Error(err)) + return version, err + } + defer tx.Rollback() + + qtx := querier.WithTx(tx) - currentVersion, err := querier.GetEnvelopeVersion(ctx, queries.GetEnvelopeVersionParams{ + currentVersion, err := qtx.GetEnvelopeVersion(ctx, queries.GetEnvelopeVersionParams{ OriginatorNodeID: originatorNodeID, OriginatorSequenceID: sequenceID, }) if err != nil && !errors.Is(err, sql.ErrNoRows) { logger.Error("Error getting current version", zap.Error(err)) return version, err } if errors.Is(err, sql.ErrNoRows) { return version, err } if err == nil { - if err = querier.InvalidateEnvelope(ctx, queries.InvalidateEnvelopeParams{ + if err = qtx.InvalidateEnvelope(ctx, queries.InvalidateEnvelopeParams{ OriginatorNodeID: originatorNodeID, OriginatorSequenceID: sequenceID, }); err != nil { logger.Error("Error invalidating old envelope", zap.Error(err)) return version, err } version = sql.NullInt32{ Int32: currentVersion.Int32 + 1, Valid: true, } } + if err = tx.Commit(); err != nil { + logger.Error("Error committing transaction", zap.Error(err)) + return version, err + } + return version, nil }Committable suggestion skipped: line range outside the PR's diff.
pkg/indexer/storer/identityUpdate.go (2)
174-195: 🛠️ Refactor suggestion
Simplify error handling logic.
The error handling has redundant checks and suboptimal log placement. Consider simplifying:
version := sql.NullInt32{Int32: 1, Valid: true} if appendLog { version, err = GetVersionForAppend( ctx, querier, s.logger, IDENTITY_UPDATE_ORIGINATOR_ID, int64(msgSent.SequenceId), ) if err != nil { - if !errors.Is(err, sql.ErrNoRows) { - return NewLogStorageError(err, true) - } - if errors.Is(err, sql.ErrNoRows) { - s.logger.Debug("No rows found for envelope, inserting new", - zap.Int("originator_node_id", IDENTITY_UPDATE_ORIGINATOR_ID), - zap.Int64("originator_sequence_id", int64(msgSent.SequenceId)), - ) - } + if errors.Is(err, sql.ErrNoRows) { + s.logger.Debug("No rows found for envelope, inserting new", + zap.Int("originator_node_id", IDENTITY_UPDATE_ORIGINATOR_ID), + zap.Int64("originator_sequence_id", int64(msgSent.SequenceId)), + ) + } else { + return NewLogStorageError(err, true) + } } }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.version := sql.NullInt32{Int32: 1, Valid: true} if appendLog { version, err = GetVersionForAppend( ctx, querier, s.logger, IDENTITY_UPDATE_ORIGINATOR_ID, int64(msgSent.SequenceId), ) if err != nil { if errors.Is(err, sql.ErrNoRows) { s.logger.Debug("No rows found for envelope, inserting new", zap.Int("originator_node_id", IDENTITY_UPDATE_ORIGINATOR_ID), zap.Int64("originator_sequence_id", int64(msgSent.SequenceId)), ) } else { return NewLogStorageError(err, true) } } }
52-56: 💡 Codebase verification
Breaking change in StoreLog requires updates in multiple files
The following files are using outdated 2-parameter version of
StoreLogand need to be updated with the newappendLogparameter:
pkg/mocks/storer/mock_LogStorer.gopkg/indexer/indexer_test.gopkg/indexer/storer/groupMessage.gopkg/indexer/storer/identityUpdate_test.gopkg/indexer/storer/groupMessage_test.go🔗 Analysis chain
Verify all callers of StoreLog are updated.
The method signature has been updated to include the
appendLogparameter. This is a breaking change that requires updates to all callers.🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Search for all calls to StoreLog to ensure they've been updated rg -A 2 "StoreLog\(" --type goLength of output: 2632
fbac
force-pushed
the
feat/reorg-detection
branch
from
a99d582 to
7e6a14e
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (5)
pkg/indexer/storer/identityUpdate.go (2)
27-27: Consider makingIDENTITY_UPDATE_ORIGINATOR_IDconfigurable.The constant
IDENTITY_UPDATE_ORIGINATOR_IDis hardcoded to1, and the comment suggests that this may need to vary for different smart contracts. To enhance flexibility and maintainability, consider externalizing this value to a configuration file or environment variable.
175-196: Simplify error handling logic for clarity.The current error handling for
sql.ErrNoRowsinvolves a redundantifcheck. Simplifying this logic improves readability and maintainability.Apply this diff to simplify the error handling:
if err != nil { - if !errors.Is(err, sql.ErrNoRows) { + if errors.Is(err, sql.ErrNoRows) { + s.logger.Debug("No rows found for envelope, inserting new", + zap.Int("originator_node_id", IDENTITY_UPDATE_ORIGINATOR_ID), + zap.Int64("originator_sequence_id", int64(msgSent.SequenceId)), + ) + } else { return NewLogStorageError(err, true) } - if errors.Is(err, sql.ErrNoRows) { - s.logger.Debug("No rows found for envelope, inserting new", - zap.Int("originator_node_id", IDENTITY_UPDATE_ORIGINATOR_ID), - zap.Int64("originator_sequence_id", int64(msgSent.SequenceId)), - ) - } }pkg/indexer/indexer_test.go (1)
74-81: Add test cases for retry-reorg interaction.The test verifies retryable error handling but doesn't test how retries interact with chain reorganizations. Consider adding test cases that:
- Simulate a reorg during a retry
- Verify retry behavior when a reorg occurs
- Test error handling during reorg processing
Example test case:
func TestIndexLogsRetryDuringReorg(t *testing.T) { // Setup channels and context... // Simulate error and reorg logStorer.EXPECT(). StoreLog(mock.Anything, event, false). RunAndReturn(func(ctx context.Context, log types.Log, isCanonical bool) storer.LogStorageError { // Trigger reorg during retry reorgChannel <- uint64(5) return storer.NewLogStorageError(errors.New("retryable error"), true) }) // Verify handling... }Also applies to: 94-123
pkg/indexer/storer/groupMessage.go (1)
17-20: Consider making the originator ID configurable.The TODO comment suggests not hardcoding the originator ID. Consider:
- Making it configurable via constructor
- Using a registry pattern for contract-specific IDs
Example:
type Config struct { OriginatorID int } func NewGroupMessageStorer( queries *queries.Queries, logger *zap.Logger, contract *groupmessages.GroupMessages, config Config, ) *GroupMessageStorer { return &GroupMessageStorer{ queries: queries, logger: logger.Named("GroupMessageStorer"), contract: contract, originatorID: config.OriginatorID, } }pkg/db/queries/queries.sql.go (1)
243-262: Consider returning affected rows count.While the implementation is correct, returning the number of affected rows would help verify whether the invalidation actually occurred.
-func (q *Queries) InvalidateEnvelope(ctx context.Context, arg InvalidateEnvelopeParams) error { - _, err := q.db.ExecContext(ctx, invalidateEnvelope, arg.OriginatorNodeID, arg.OriginatorSequenceID) - return err +func (q *Queries) InvalidateEnvelope(ctx context.Context, arg InvalidateEnvelopeParams) (int64, error) { + result, err := q.db.ExecContext(ctx, invalidateEnvelope, arg.OriginatorNodeID, arg.OriginatorSequenceID) + if err != nil { + return 0, err + } + return result.RowsAffected() }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (35)
pkg/proto/identity/api/v1/identity.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/identity/associations/association.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/identity/associations/signature.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/identity/credential.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/keystore_api/v1/keystore.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_api/v1/authn.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_api/v1/message_api.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/ciphertext.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/composite.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/contact.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/content.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/conversation_reference.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/ecies.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/frames.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/invitation.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/message.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/private_key.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/private_preferences.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/public_key.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/signature.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/signed_payload.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls/api/v1/mls.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls/database/intents.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls/message_contents/content.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls/message_contents/content_types/reaction.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls/message_contents/group_membership.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls/message_contents/group_metadata.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls/message_contents/group_mutable_metadata.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls/message_contents/group_permissions.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls/message_contents/transcript_messages.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls_validation/v1/service.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/xmtpv4/envelopes/envelopes.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/xmtpv4/message_api/message_api.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/xmtpv4/message_api/misbehavior_api.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/xmtpv4/payer_api/payer_api.pb.gois excluded by!**/*.pb.go,!pkg/proto/**
📒 Files selected for processing (21)
pkg/blockchain/interface.go(1 hunks)pkg/blockchain/rpcLogStreamer.go(5 hunks)pkg/blockchain/rpcLogStreamer_test.go(3 hunks)pkg/config/options.go(1 hunks)pkg/db/queries.sql(2 hunks)pkg/db/queries/models.go(1 hunks)pkg/db/queries/queries.sql.go(7 hunks)pkg/indexer/indexer.go(8 hunks)pkg/indexer/indexer_test.go(1 hunks)pkg/indexer/storer/groupMessage.go(4 hunks)pkg/indexer/storer/groupMessage_test.go(3 hunks)pkg/indexer/storer/identityUpdate.go(4 hunks)pkg/indexer/storer/identityUpdate_test.go(1 hunks)pkg/indexer/storer/interface.go(1 hunks)pkg/indexer/storer/logAppend.go(1 hunks)pkg/interceptors/server/auth_test.go(4 hunks)pkg/migrations/00004_add_blockchain_columns.down.sql(1 hunks)pkg/migrations/00004_add_blockchain_columns.up.sql(1 hunks)pkg/mocks/authn/mock_JWTVerifier.go(2 hunks)pkg/mocks/blockchain/mock_ChainClient.go(4 hunks)pkg/mocks/storer/mock_LogStorer.go(3 hunks)
🚧 Files skipped from review as they are similar to previous changes (10)
- pkg/blockchain/interface.go
- pkg/indexer/storer/interface.go
- pkg/config/options.go
- pkg/db/queries/models.go
- pkg/blockchain/rpcLogStreamer_test.go
- pkg/migrations/00004_add_blockchain_columns.down.sql
- pkg/migrations/00004_add_blockchain_columns.up.sql
- pkg/indexer/storer/logAppend.go
- pkg/indexer/storer/groupMessage_test.go
- pkg/blockchain/rpcLogStreamer.go
🔇 Additional comments (25)
pkg/indexer/storer/identityUpdate.go (1)
53-57: Ensure all calls toStoreLogare updated with the newappendLogparameter.The
StoreLogmethod signature now includes an additionalappendLogparameter. Please verify that all implementations and usages of this method have been updated accordingly to prevent potential runtime errors.Run the following script to identify all calls to
StoreLogthat may need updating:✅ Verification successful
All
StoreLogcalls are properly updated with the new parameterAll implementations and usages of
StoreLoghave been verified to include the newappendLogparameter, with consistent usage offalseas the default value. No outdated calls were found.🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Description: Find all calls to StoreLog and check for the correct number of arguments. rg 'StoreLog\(' -t go -A 1Length of output: 2043
pkg/mocks/storer/mock_LogStorer.go (4)
27-37: Update mockStoreLogmethod to reflect new parameter.The
StoreLogmethod inMockLogStorerhas been updated to include theappendLogparameter, ensuring consistency with the interface. Verify that all mock expectations and return values are appropriately adjusted.
55-57: Ensure test expectations includeappendLogparameter.The
StoreLogexpectation in the mock now includes theappendLogparameter. Please verify that all tests setting expectations onStoreLogprovide the correct arguments.
60-62: PassappendLogparameter in mockRunfunction.The
Runmethod now correctly passes theappendLogparameter to the provided function, ensuring accurate simulation of method behavior in tests.
72-72: IncludeappendLoginRunAndReturnfor completeness.The
RunAndReturnmethod is updated to accept theappendLogparameter, maintaining consistency in the mock's behavior.pkg/indexer/storer/identityUpdate_test.go (1)
82-82: UpdateStoreLogcall withappendLogparameter.In
TestStoreIdentityUpdate, the call toStoreLognow includesfalsefor theappendLogparameter. Ensure that this aligns with the intended test scenario and that other tests are updated accordingly.pkg/mocks/blockchain/mock_ChainClient.go (1)
32-89: LGTM! The mock implementation is complete and correct.The auto-generated mock file correctly implements all the required blockchain client methods with proper:
- Method signatures
- Return value handling
- Error cases
- Helper methods for expectations
Also applies to: 91-148, 501-558, 560-615, 617-675
pkg/db/queries.sql (3)
16-19: Enhance conflict handling for chain reorganizations.The current
ON CONFLICT DO NOTHINGmight not handle chain reorganizations correctly. Consider updating theis_canonicalflag when conflicts occur based on block metadata.
125-134: LGTM! The version retrieval query is well-designed.The query correctly:
- Retrieves the version for canonical envelopes only
- Uses proper parameter binding
135-144: LGTM! The envelope invalidation query is well-designed.The query correctly:
- Updates only canonical envelopes
- Uses proper parameter binding
- Has appropriate WHERE clause conditions
pkg/indexer/storer/groupMessage.go (2)
88-89: Implement dynamic version and canonical status handling.The version and canonical status are hardcoded. Consider:
- Implementing version based on protocol or message format version
- Determining canonical status based on block confirmation depth
Also applies to: 114-117
90-109: LGTM! The version retrieval logic is well-implemented.The code correctly:
- Handles version retrieval for appending
- Properly handles the no-rows case
- Includes appropriate logging
pkg/indexer/indexer.go (5)
138-138: LGTM! Well-structured channel declarations.The send-only channels (
chan<- uint64) are appropriately typed for their purpose of signaling reorganization events.Also applies to: 140-140
162-162: LGTM! Proper initialization of reorg channels.The reorg channels are correctly initialized and returned from the builder, maintaining consistency with the struct updates.
Also applies to: 184-184, 195-197
212-215: LGTM! Enhanced function signature for reorg handling.The addition of blockchain client, reorg channel, and safe block distance parameters provides the necessary components for reorg detection and handling.
92-95: LGTM! Proper parameter passing for reorg handling.The function correctly passes the blockchain client, reorg channels, and safe block distance to both message and identity update indexers.
Also applies to: 116-120
224-259:⚠️ Potential issueComplete the reorg handling implementation.
The reorg detection logic is well-structured, but several critical items need attention:
- The reorgChannel send is commented out, leaving the reorg handling incomplete
- The safeBlockDistance check could be more robust by also verifying the minimum safe distance
- The TODO comments outline important steps that need implementation
Let's verify the reorg detection logic:
Consider implementing the following improvements:
if event.BlockNumber > storedBlockNumber && - event.BlockNumber-storedBlockNumber < safeBlockDistance { + event.BlockNumber-storedBlockNumber < safeBlockDistance && + event.BlockNumber-storedBlockNumber >= MIN_SAFE_DISTANCE { latestBlock, err := client.BlockByNumber(ctx, big.NewInt(int64(storedBlockNumber))) // ... existing code ... if !bytes.Equal(storedBlockHash, latestBlock.Hash().Bytes()) { // ... existing code ... - // reorgChannel <- event.BlockNumber + reorgChannel <- event.BlockNumber } }pkg/db/queries/queries.sql.go (3)
78-99: LGTM! Well-implemented version retrieval.The query correctly filters for canonical envelopes and properly handles NULL versions using sql.NullInt32.
165-166: LGTM! Comprehensive envelope storage.The changes properly track block information, versioning, and canonicity status, which are essential for reorg handling.
Also applies to: 176-179, 188-191
290-290: LGTM! Consistent envelope retrieval.The SELECT clause and scan operations properly include all new fields, maintaining consistency with the updated struct.
Also applies to: 324-327
pkg/mocks/authn/mock_JWTVerifier.go (4)
28-37: LGTM: Proper handling of uint32 return valueThe mock correctly handles the uint32 return value through type assertions and function calls.
39-45: LGTM: Error handling remains consistentThe error handling logic is maintained while accommodating the new return type structure.
66-74: LGTM: Mock helper methods updated correctlyThe Return and RunAndReturn methods are properly updated to handle both uint32 and error return values.
Line range hint
28-74: Verify the semantic meaning of uint32 return valueThe introduction of a uint32 return value suggests it carries semantic meaning (e.g., status code, identifier). This should be documented to ensure correct usage.
Let's check for any documentation or usage patterns:
✅ Verification successful
The uint32 return value represents the source node ID in the node-to-node authentication system
The uint32 return value from Verify() has clear semantic meaning: it represents the ID of the source node that created the token. This is used in the system's node-to-node authentication mechanism, where tokens are created by one node (the source) targeting another node (the audience), and the verification process returns the source node's ID upon successful validation.
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Search for documentation or usage patterns of the uint32 return value rg -g '!vendor' -g '*.go' "type JWTVerifier interface" -A 5Length of output: 205
Script:
#!/bin/bash # Search for implementations and usage of JWTVerifier.Verify echo "=== Searching for implementations ===" ast-grep --pattern 'type $_ struct { $$$ } func ($_ $_) Verify(tokenString string) (uint32, error) { $$$ }' echo -e "\n=== Searching for method usage ===" rg -g '!vendor' -g '*.go' -B 3 -A 3 '\.Verify\(' echo -e "\n=== Searching for related documentation ===" rg -g '!vendor' -g '*.go' -B 3 "(?i)jwt.*verify"Length of output: 9853
Script:
#!/bin/bash # Search for VERIFIER_NODE_ID constant and related configuration echo "=== Searching for node ID constants and configuration ===" rg -g '!vendor' -g '*.go' "VERIFIER_NODE_ID|CreateToken|type.*Claims" -B 2 -A 2 echo -e "\n=== Searching for token creation implementation ===" ast-grep --pattern 'func ($_ $_) CreateToken(nodeId uint32) ($_, error) { $$$ }'Length of output: 8525
pkg/interceptors/server/auth_test.go (1)
39-39: Verify the interceptor's handling of uint32 valuesThe tests don't verify if the uint32 value is properly propagated through the interceptor chain.
Let's check how the interceptor uses this value:
Also applies to: 74-74, 134-134, 159-159
fbac
force-pushed
the
feat/reorg-detection
branch
from
cf48f8c to
743cfa8
Compare
pkg/indexer/indexer.go
Outdated
|
|
||
| // TODO: Calculate the blocks safe distance in the L3 or risk tolerance we assume | ||
| if event.BlockNumber > storedBlockNumber && | ||
| event.BlockNumber-storedBlockNumber < safeBlockDistance { |
There was a problem hiding this comment.
We need to balance timeliness with safety here. A lot of these blockchain events are the kind of thing a human is going to be waiting for on their phone. I think that means we need to be optimistic and rely on preconfirmations, and then fix later in the rare emergency case.
There was a problem hiding this comment.
I modified the indexer to be more performant by skipping certain IO tasks when they're not needed:
- Only
blockTracker.GetLatestBlockwhen really needed. - Check for reorgs only after a certain period has passed. The
safeBlockDistancehas been removed in favor of this. Also, I'm not comfortable leaving this open to configurations so I dropped the option to do it. - When a reorg is detected the indexer will drop the messages that are in the eventChannel, until it starts receiving from the first reorganized block.
pkg/indexer/storer/identityUpdate.go
Outdated
| func (s *IdentityUpdateStorer) StoreLog( | ||
| ctx context.Context, | ||
| event types.Log, | ||
| appendLog bool, |
There was a problem hiding this comment.
Maybe we can add a comment about when you would and wouldn't want to use this flag.
pkg/db/queries/queries.sql.go
Outdated
| @@ -150,6 +173,10 @@ type InsertGatewayEnvelopeParams struct { | |||
| OriginatorSequenceID int64 | |||
| Topic []byte | |||
| OriginatorEnvelope []byte | |||
| BlockNumber sql.NullInt64 | |||
| BlockHash []byte | |||
There was a problem hiding this comment.
Just thinking out loud, but what if we did this relationally (like with a table called block_versions) and joined to that rather than denormalizing?
Would use less storage and still be easy to find impacted messages in a re-org.
Not a suggestion, more of a question.
There was a problem hiding this comment.
I made a similar proposal in slack :)
There was a problem hiding this comment.
I like that approach, too. My biggest concern is that for each group/identity message we'd be inserting in db twice and there would be additional storage overhead.
Would use less storage
I'm not sure I follow, could you expand this?
There was a problem hiding this comment.
Just that we only have to store the block hash and number once per block, and store a foreign key on each message in the block. The block number and foreign key are the same size so they cancel out, but the block hash is pure savings and bigger.
I'm assuming most blocks have > 1 message in them.
There was a problem hiding this comment.
SQL schemas updated. It still needs some fine tuning, and better thinking on the queries, but captures the idea.
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (5)
pkg/indexer/indexer.go (2)
228-228: Consider making reorgCheckInterval configurable.The fixed interval of 10 blocks might not be optimal for all network conditions. Consider making this configurable through the ContractsOptions.
255-279: Consider adding a maximum reorg depth limit.While the reorg detection logic is sound, consider adding a maximum depth limit to prevent processing very deep reorgs. This would help balance safety with timeliness, as mentioned in the feedback.
if storedBlockNumber > 0 && event.BlockNumber > storedBlockNumber && - event.BlockNumber >= reorgLastCheckAt+reorgCheckInterval { + event.BlockNumber >= reorgLastCheckAt+reorgCheckInterval && + event.BlockNumber-storedBlockNumber < maxReorgDepth {pkg/migrations/00004_add_blockchain_columns.up.sql (3)
2-10: Add table and column comments for better documentation.Consider adding comments to document the purpose of the table and its columns. This improves maintainability and helps other developers understand the schema design.
Consider adding these comments:
CREATE TABLE blockchain_messages( block_number BIGINT NOT NULL, block_hash BYTEA NOT NULL, originator_node_id INT NOT NULL, originator_sequence_id BIGINT NOT NULL, is_canonical BOOLEAN NOT NULL DEFAULT TRUE, PRIMARY KEY (block_number, block_hash), FOREIGN KEY (originator_node_id, originator_sequence_id) REFERENCES gateway_envelopes(originator_node_id, originator_sequence_id) -); +); + +COMMENT ON TABLE blockchain_messages IS 'Tracks blockchain messages for chain reorganization detection'; +COMMENT ON COLUMN blockchain_messages.block_number IS 'The block number where the message was recorded'; +COMMENT ON COLUMN blockchain_messages.block_hash IS 'The hash of the block where the message was recorded'; +COMMENT ON COLUMN blockchain_messages.is_canonical IS 'Indicates if this block is part of the canonical chain';
12-14: Add an index for the foreign key columns.Consider adding an index on the foreign key columns to improve the performance of joins and foreign key constraint checks.
Add this index:
CREATE INDEX idx_blockchain_messages_fk ON blockchain_messages(originator_node_id, originator_sequence_id);
12-14: Consider a more descriptive index name.The current index name
idx_blockchain_messages_canonicalcould be more descriptive about its purpose and sort order.Consider renaming to:
-CREATE INDEX idx_blockchain_messages_canonical ON blockchain_messages(block_number DESC, block_hash) +CREATE INDEX idx_blockchain_messages_latest_canonical ON blockchain_messages(block_number DESC, block_hash) WHERE is_canonical = TRUE;
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (4)
pkg/indexer/indexer.go(9 hunks)pkg/indexer/indexer_test.go(1 hunks)pkg/migrations/00004_add_blockchain_columns.down.sql(1 hunks)pkg/migrations/00004_add_blockchain_columns.up.sql(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (2)
- pkg/migrations/00004_add_blockchain_columns.down.sql
- pkg/indexer/indexer_test.go
🔇 Additional comments (6)
pkg/indexer/indexer.go (5)
49-62: LGTM! Clean resource management.The Close method properly handles cleanup of resources in the correct order, with appropriate nil checks.
Line range hint
98-124: LGTM! Symmetric handling of messages and identity updates.The changes consistently pass the required parameters for reorg detection to both indexing operations.
141-144: LGTM! Clear channel definitions.The struct properly defines typed channels for reorganization events.
Line range hint
166-201: LGTM! Proper initialization of reorg detection.The changes symmetrically configure reorg detection for both contracts.
235-247: LGTM! Efficient event handling during reorgs.The implementation properly discards events during reorg and resumes processing afterward, which aligns well with the need to balance timeliness with safety.
pkg/migrations/00004_add_blockchain_columns.up.sql (1)
1-15: Provide the corresponding DOWN migration.Please ensure there's a corresponding
00004_add_blockchain_columns.down.sqlfile that cleanly reverts these changes. This is crucial for database maintainability and rollback scenarios.Here's a script to check for the DOWN migration:
✅ Verification successful
DOWN migration is properly implemented
The corresponding
00004_add_blockchain_columns.down.sqlfile exists and correctly reverts all changes in the proper order.🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Check if the corresponding down migration exists if [ ! -f "pkg/migrations/00004_add_blockchain_columns.down.sql" ]; then echo "Missing DOWN migration file" exit 1 fiLength of output: 69
Script:
#!/bin/bash # List the migration file if it exists ls -l pkg/migrations/00004_add_blockchain_columns.down.sql 2>/dev/null echo "=== Down Migration Content ===" cat pkg/migrations/00004_add_blockchain_columns.down.sql 2>/dev/null || echo "File not found"Length of output: 434
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (1)
pkg/db/queries.sql (1)
125-127:⚠️ Potential issueAdd conflict handling for chain reorganizations.
The current INSERT statement lacks conflict handling, which is crucial for managing chain reorganizations. When a block is reorganized, we need to update the canonicity of existing records.
Add conflict handling similar to the gateway_envelopes suggestion:
INSERT INTO blockchain_messages(block_number, block_hash, originator_node_id, originator_sequence_id, is_canonical) - VALUES (@block_number, @block_hash, @originator_node_id, @originator_sequence_id, @is_canonical); + VALUES (@block_number, @block_hash, @originator_node_id, @originator_sequence_id, @is_canonical) +ON CONFLICT (block_number, originator_node_id, originator_sequence_id) + DO UPDATE SET + is_canonical = EXCLUDED.is_canonical, + block_hash = EXCLUDED.block_hash + WHERE blockchain_messages.block_hash != EXCLUDED.block_hash;
🧹 Nitpick comments (3)
pkg/db/queries.sql (2)
129-134: Fix incorrect comment about sort order.The comment states "Returns blocks in descending order" but the query uses
ORDER BY block_number ASC.--- name: GetBlocksInRange :many -// Returns blocks in descending order (newest to oldest) +// Returns blocks in ascending order (oldest to newest) // StartBlock should be the lower bound (older block) // EndBlock should be the upper bound (newer block) -// Example: GetBlocksInRange(1000, 2000) gets blocks 2000 down to 1000 +// Example: GetBlocksInRange(1000, 2000) gets blocks from 1000 up to 2000
134-145: Optimize query performance with proper indexing.The query performs filtering and ordering on multiple columns. Consider adding appropriate indexes to improve performance.
Add the following index:
CREATE INDEX idx_blockchain_messages_block_range ON blockchain_messages (block_number, is_canonical) WHERE block_hash IS NOT NULL;Also consider creating a materialized view for frequently accessed canonical blocks:
CREATE MATERIALIZED VIEW canonical_blocks AS SELECT DISTINCT ON (block_number) block_number, block_hash FROM blockchain_messages WHERE block_hash IS NOT NULL AND is_canonical = TRUE ORDER BY block_number ASC, block_hash;pkg/indexer/indexer.go (1)
235-235: Consider making reorgCheckInterval configurable.The hardcoded value of 10 for
reorgCheckIntervalmight not be optimal for all chains or load conditions.Consider making this configurable through the config package:
- reorgCheckInterval uint64 = 10 // TODO: Adapt based on blocks per batch + reorgCheckInterval = cfg.ReorgCheckInterval // Default to 10 if not set
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
pkg/blockchain/rpcLogStreamer.go(5 hunks)pkg/db/queries.sql(1 hunks)pkg/db/queries/models.go(1 hunks)pkg/db/queries/queries.sql.go(2 hunks)pkg/indexer/indexer.go(9 hunks)pkg/indexer/storer/interface.go(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (2)
- pkg/indexer/storer/interface.go
- pkg/db/queries/queries.sql.go
🧰 Additional context used
📓 Learnings (1)
pkg/blockchain/rpcLogStreamer.go (1)
Learnt from: fbac
PR: xmtp/xmtpd#411
File: pkg/blockchain/rpcLogStreamer.go:51-65
Timestamp: 2025-01-16T15:20:37.460Z
Learning: In Go, returning a send-only channel (chan<-) from a public API while using it bidirectionally internally is a valid pattern. It allows the implementation to have full channel access while restricting callers to only writing to the channel. This pattern is used in the RpcLogStreamBuilder.ListenForContractEvent method where reorgChannel is returned as chan<- uint64.
🪛 golangci-lint (1.62.2)
pkg/indexer/indexer.go
143-143: undefined: ChainReorgHandler
(typecheck)
199-199: undefined: NewChainReorgHandler
(typecheck)
228-228: undefined: ChainReorgHandler
(typecheck)
🔇 Additional comments (10)
pkg/db/queries/models.go (1)
19-25: Consider adding hash length validation for BlockHash field.While the struct is well-designed for tracking chain reorganizations, consider adding validation to ensure BlockHash is the correct length for your blockchain (e.g., 32 bytes for Ethereum).
Run this script to verify hash length usage:
pkg/blockchain/rpcLogStreamer.go (4)
51-65: LGTM! Well-designed channel setup for reorg notifications.The method signature change and channel setup are well-implemented:
- Buffer size of 1 prevents blocking while allowing queuing of a single reorg notification
- Channel direction constraints properly enforce write-only access for callers
77-78: LGTM! Proper channel direction constraints in struct fields.The struct fields correctly use channel direction constraints that match the public API.
138-143: LGTM! Clean implementation of reorg notification handling.The reorg handling logic properly:
- Resets the block number to resume from the reorg point
- Includes informative logging
230-232: LGTM! Clean getter implementation.The Client method provides necessary access to the underlying chain client for reorg detection.
pkg/indexer/indexer.go (5)
49-62: LGTM! Proper cleanup sequence in Close method.The cleanup sequence is well-implemented:
- Nil checks prevent panics
- Channels are closed before stopping the streamer
222-228: LGTM! Well-structured parameter additions for reorg handling.The new parameters provide all necessary dependencies for reorg detection and handling.
🧰 Tools
🪛 golangci-lint (1.62.2)
228-228: undefined: ChainReorgHandler
(typecheck)
230-241: LGTM! Well-structured state management for reorg handling.The state variables are well-organized and clearly named, making the reorg state machine easy to understand.
279-325: LGTM! Robust reorg detection implementation.The reorg detection logic is well-implemented:
- Efficient checking at intervals
- Proper ancestor finding
- Clear error handling
- Appropriate logging at each step
329-329: LGTM! Proper reorg state handling in storage.The
reorgInProgressflag is correctly passed to the storage layer, allowing proper handling of reorged events.
There was a problem hiding this comment.
Actionable comments posted: 5
♻️ Duplicate comments (2)
pkg/indexer/storer/groupMessage.go (1)
87-89:⚠️ Potential issueEmpty if block needs implementation.
The
appendLogbranch is currently empty. This could lead to confusion about the intended behavior.Based on the previous review comments about dynamic version handling, consider implementing the version retrieval logic here:
if appendLog { - // placeholder + version, err := GetVersionForAppend(ctx, s.queries, s.logger, GROUP_MESSAGE_ORIGINATOR_ID, int64(msgSent.SequenceId)) + if err != nil { + s.logger.Error("Error getting version for append", zap.Error(err)) + return NewLogStorageError(err, true) + } + // Use the version for subsequent operations }🧰 Tools
🪛 GitHub Check: Lint-Go
[failure] 87-87:
SA9003: empty branch (staticcheck)pkg/indexer/indexer_test.go (1)
20-27: 🛠️ Refactor suggestionAdd test cases for chain reorganization handling.
The test setup includes reorg channels but lacks test coverage for reorganization scenarios. Consider adding test cases that:
- Simulate chain reorganization events
- Verify the system's response to reorgs
- Test different block distances
Would you like me to create a GitHub issue to track this task?
🧹 Nitpick comments (11)
pkg/indexer/reorgHandler.go (3)
30-31: Consider makingBLOCK_RANGE_SIZEconfigurableThe constant
BLOCK_RANGE_SIZEis currently hardcoded to1000. Making it configurable would enhance flexibility and allow adjustments based on different chain characteristics or operational requirements.
132-134: Handle unexpected cases insearchInRangemethod more gracefullyThe
searchInRangemethod returns an error stating "reorg point not found" if it fails to locate the reorganization point. Since the comment suggests this scenario should never happen, consider enhancing error handling by providing more detailed logging or by implementing a fallback mechanism.
137-144: Review the logic inblockRangefunction for edge casesIn the
blockRangefunction, whenendBlockis less than or equal toBLOCK_RANGE_SIZE,startBlockremains0. Ensure that this behavior is intended and that it correctly handles scenarios at the beginning of the chain.pkg/indexer/reorgHandler_test.go (1)
9-43: Enhance test coverage forblockRangefunctionThe current test cases cover general scenarios. Consider adding test cases for edge conditions, such as:
- When
fromis exactly equal toBLOCK_RANGE_SIZE.- When
fromis just one less thanBLOCK_RANGE_SIZE.- Extremely large values of
from.This will ensure the
blockRangefunction behaves correctly across all possible inputs.pkg/mocks/indexer/mock_ChainReorgHandler.go (1)
24-26: Avoid panicking in mock implementationsIn the
FindReorgPointmock method, invokingpanic("no return value specified for FindReorgPoint")can make tests brittle and harder to debug. Consider returning default values or an error instead to allow tests to handle this scenario gracefully.pkg/db/queries/models.go (1)
19-25: Add documentation for the BlockchainMessage struct.While the struct is well-designed, adding documentation would improve code maintainability. Consider adding comments explaining the purpose of the struct and its fields.
+// BlockchainMessage represents a message stored with its blockchain context type BlockchainMessage struct { + // BlockNumber is the block height where the message was included BlockNumber uint64 + // BlockHash is the hash of the block containing the message BlockHash []byte + // OriginatorNodeID identifies the node that originated the message OriginatorNodeID int32 + // OriginatorSequenceID is the sequence number assigned by the originator OriginatorSequenceID int64 + // IsCanonical indicates if the message is on the main chain IsCanonical bool }pkg/indexer/storer/groupMessage.go (2)
16-19: Consider making originator ID configurable.The comment suggests that hardcoding to 0 might not be ideal. Consider:
- Making this configurable through constructor parameters
- Using an enum or constants for different contract types
-const ( - // We may not want to hardcode this to 0 and have an originator ID for each smart contract? - GROUP_MESSAGE_ORIGINATOR_ID = 0 -) +// OriginatorID represents the source of a message in the system +type OriginatorID int32 + +const ( + // GroupMessageOriginatorID represents messages from the group messages contract + GroupMessageOriginatorID OriginatorID = 0 + // Add other originator IDs as needed +)
94-94: Use the new constant type if implemented.If you implement the suggested OriginatorID type, update this usage accordingly.
- OriginatorNodeID: GROUP_MESSAGE_ORIGINATOR_ID, + OriginatorNodeID: int32(GroupMessageOriginatorID),pkg/indexer/indexer.go (1)
236-236: Consider making reorgCheckInterval configurable.The hardcoded
reorgCheckIntervalmight need adjustment based on network conditions or block times.- reorgCheckInterval uint64 = 10 // TODO: Adapt based on blocks per batch + reorgCheckInterval = config.GetReorgCheckInterval() // Default: 10pkg/db/queries/queries.sql.go (2)
78-91: Consider optimizing the query performance.The current query uses
DISTINCT ONwith an additionalORDER BYclause. For better performance, consider using a subquery withROW_NUMBER()or a CTE.-SELECT DISTINCT ON (block_number) - block_number, - block_hash -FROM - blockchain_messages -WHERE - block_number BETWEEN $1 AND $2 - AND block_hash IS NOT NULL - AND is_canonical = TRUE -ORDER BY - block_number ASC, - block_hash +WITH RankedBlocks AS ( + SELECT + block_number, + block_hash, + ROW_NUMBER() OVER (PARTITION BY block_number ORDER BY block_hash) as rn + FROM blockchain_messages + WHERE block_number BETWEEN $1 AND $2 + AND block_hash IS NOT NULL + AND is_canonical = TRUE +) +SELECT block_number, block_hash +FROM RankedBlocks +WHERE rn = 1 +ORDER BY block_number ASC
206-215: Return affected rows count for better error handling.Consider returning the number of affected rows to help callers verify if the insertion was successful.
-func (q *Queries) InsertBlockchainMessage(ctx context.Context, arg InsertBlockchainMessageParams) error { - _, err := q.db.ExecContext(ctx, insertBlockchainMessage, +func (q *Queries) InsertBlockchainMessage(ctx context.Context, arg InsertBlockchainMessageParams) (int64, error) { + result, err := q.db.ExecContext(ctx, insertBlockchainMessage, arg.BlockNumber, arg.BlockHash, arg.OriginatorNodeID, arg.OriginatorSequenceID, arg.IsCanonical, ) - return err + if err != nil { + return 0, err + } + return result.RowsAffected() }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (16)
.mockery.yaml(1 hunks)pkg/blockchain/rpcLogStreamer.go(6 hunks)pkg/db/queries.sql(1 hunks)pkg/db/queries/models.go(1 hunks)pkg/db/queries/queries.sql.go(2 hunks)pkg/indexer/indexer.go(9 hunks)pkg/indexer/indexer_test.go(1 hunks)pkg/indexer/reorgHandler.go(1 hunks)pkg/indexer/reorgHandler_test.go(1 hunks)pkg/indexer/storer/groupMessage.go(3 hunks)pkg/indexer/storer/identityUpdate.go(3 hunks)pkg/indexer/storer/interface.go(1 hunks)pkg/indexer/storer/logAppend.go(1 hunks)pkg/migrations/00004_add_blockchain_columns.up.sql(1 hunks)pkg/mocks/indexer/mock_ChainReorgHandler.go(1 hunks)sqlc.yaml(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (3)
- pkg/indexer/storer/interface.go
- pkg/db/queries.sql
- pkg/migrations/00004_add_blockchain_columns.up.sql
🧰 Additional context used
📓 Learnings (1)
pkg/blockchain/rpcLogStreamer.go (1)
Learnt from: fbac
PR: xmtp/xmtpd#411
File: pkg/blockchain/rpcLogStreamer.go:51-65
Timestamp: 2025-01-16T15:20:37.460Z
Learning: In Go, returning a send-only channel (chan<-) from a public API while using it bidirectionally internally is a valid pattern. It allows the implementation to have full channel access while restricting callers to only writing to the channel. This pattern is used in the RpcLogStreamBuilder.ListenForContractEvent method where reorgChannel is returned as chan<- uint64.
🪛 GitHub Check: Lint-Go
pkg/indexer/storer/groupMessage.go
[failure] 87-87:
SA9003: empty branch (staticcheck)
pkg/indexer/storer/identityUpdate.go
[failure] 174-174:
SA9003: empty branch (staticcheck)
🔇 Additional comments (10)
pkg/indexer/reorgHandler.go (2)
46-89: Verify loop termination conditions inFindReorgPointThe loop starting at line 49 in the
FindReorgPointmethod may potentially run indefinitely if the reorganization point is not found andstartBlockrepeatedly resets to0. Please verify that there is a definitive exit condition to prevent an infinite loop under all scenarios.
77-84: Ensure proper block range adjustment when blocks do not matchWhen the oldest stored block's hash does not match the chain block's hash, the code recalculates the block range. Confirm that the new
startBlockandendBlockvalues progress the search correctly and that the loop will eventually terminate even in edge cases.sqlc.yaml (1)
10-12: LGTM! Good choice of type for block numbers.Using
uint64for block numbers is appropriate as it:
- Provides sufficient range for blockchain block numbers
- Prevents negative values which are invalid in this context
.mockery.yaml (1)
23-23: LGTM! Mock configuration added correctly.Adding
ChainReorgHandlerto mockery configuration will enable proper mock generation for testing chain reorganization scenarios.pkg/blockchain/rpcLogStreamer.go (2)
51-65: LGTM! Well-designed channel direction pattern.The implementation correctly uses
chan<- uint64for the public API while maintaining bidirectional access internally. This is a good pattern for enforcing channel direction at API boundaries.
138-143: LGTM! Clean reorg event handling.The implementation properly handles reorganization events by:
- Resetting the fromBlock to the reorg point
- Logging the event for debugging
pkg/indexer/indexer.go (3)
246-264: LGTM! Robust reorg state handling.The implementation properly handles active reorganization states by:
- Discarding future events during reorg
- Maintaining clear state transitions
- Proper logging for debugging
280-329: LGTM! Well-implemented reorg detection.The implementation includes:
- Efficient periodic checks
- Proper block hash comparison
- Clear logging
- Proper error handling
- Correct reorg point calculation
333-333: LGTM! Proper canonical state tracking.The
StoreLogcall correctly passes thereorgInProgressflag to track canonical state during reorganizations.pkg/db/queries/queries.sql.go (1)
193-215: Consider using a relational approach with block_versions table.Based on the previous discussion in the PR comments, consider implementing a relational approach using a separate
block_versionstable. This would reduce storage overhead when multiple messages reference the same block.Let's verify the current storage approach:
fbac
force-pushed
the
feat/reorg-detection
branch
from
f81d6cd to
f3da803
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (3)
pkg/indexer/reorgHandler.go (2)
15-17: Add documentation for the public interface.The
ChainReorgHandlerinterface lacks documentation explaining its purpose and contract.Add this documentation:
+// ChainReorgHandler defines the interface for detecting blockchain reorganizations. +// Implementations should efficiently find the point where the chain diverged from +// the previously known state. type ChainReorgHandler interface { + // FindReorgPoint takes a block number where a reorg was detected and returns + // the actual reorganization point (block number and hash) or an error. FindReorgPoint(detectedAt uint64) (uint64, []byte, error) }
30-31: Make block range size configurable.The
BLOCK_RANGE_SIZEshould be configurable through environment variables or configuration files to allow tuning based on network characteristics.pkg/indexer/indexer.go (1)
231-242: Consider making reorgCheckInterval configurable.The
reorgCheckIntervalis hardcoded to 10 blocks. Consider making it configurable based on network characteristics.-reorgCheckInterval uint64 = 10 // TODO: Adapt based on blocks per batch +reorgCheckInterval = cfg.ReorgCheckInterval // Default to 10 if not configured
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (35)
pkg/proto/identity/api/v1/identity.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/identity/associations/association.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/identity/associations/signature.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/identity/credential.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/keystore_api/v1/keystore.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_api/v1/authn.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_api/v1/message_api.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/ciphertext.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/composite.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/contact.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/content.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/conversation_reference.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/ecies.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/frames.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/invitation.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/message.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/private_key.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/private_preferences.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/public_key.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/signature.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/message_contents/signed_payload.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls/api/v1/mls.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls/database/intents.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls/message_contents/content.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls/message_contents/content_types/reaction.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls/message_contents/group_membership.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls/message_contents/group_metadata.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls/message_contents/group_mutable_metadata.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls/message_contents/group_permissions.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls/message_contents/transcript_messages.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/mls_validation/v1/service.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/xmtpv4/envelopes/envelopes.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/xmtpv4/message_api/message_api.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/xmtpv4/message_api/misbehavior_api.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/xmtpv4/payer_api/payer_api.pb.gois excluded by!**/*.pb.go,!pkg/proto/**
📒 Files selected for processing (28)
.mockery.yaml(1 hunks)pkg/blockchain/interface.go(1 hunks)pkg/blockchain/rpcLogStreamer.go(6 hunks)pkg/blockchain/rpcLogStreamer_test.go(3 hunks)pkg/db/queries.sql(1 hunks)pkg/db/queries/db.go(1 hunks)pkg/db/queries/models.go(2 hunks)pkg/db/queries/queries.sql.go(4 hunks)pkg/indexer/indexer.go(9 hunks)pkg/indexer/indexer_test.go(2 hunks)pkg/indexer/reorgHandler.go(1 hunks)pkg/indexer/reorgHandler_test.go(1 hunks)pkg/indexer/storer/groupMessage.go(4 hunks)pkg/indexer/storer/identityUpdate.go(3 hunks)pkg/interceptors/server/auth_test.go(4 hunks)pkg/migrations/00004_add_blockchain_columns.down.sql(1 hunks)pkg/migrations/00004_add_blockchain_columns.up.sql(1 hunks)pkg/mocks/authn/mock_JWTVerifier.go(1 hunks)pkg/mocks/blockchain/mock_ChainClient.go(4 hunks)pkg/mocks/blockchain/mock_IBlockchainPublisher.go(1 hunks)pkg/mocks/indexer/mock_ChainReorgHandler.go(1 hunks)pkg/mocks/indexer/mock_IBlockTracker.go(1 hunks)pkg/mocks/mls_validationv1/mock_ValidationApiClient.go(1 hunks)pkg/mocks/mlsvalidate/mock_MLSValidationService.go(1 hunks)pkg/mocks/registry/mock_NodeRegistry.go(1 hunks)pkg/mocks/registry/mock_NodesContract.go(1 hunks)pkg/mocks/storer/mock_LogStorer.go(1 hunks)sqlc.yaml(1 hunks)
✅ Files skipped from review due to trivial changes (9)
- pkg/mocks/registry/mock_NodesContract.go
- pkg/mocks/authn/mock_JWTVerifier.go
- pkg/mocks/storer/mock_LogStorer.go
- pkg/db/queries/db.go
- pkg/mocks/indexer/mock_IBlockTracker.go
- pkg/mocks/blockchain/mock_IBlockchainPublisher.go
- pkg/mocks/mlsvalidate/mock_MLSValidationService.go
- pkg/mocks/mls_validationv1/mock_ValidationApiClient.go
- pkg/mocks/registry/mock_NodeRegistry.go
🚧 Files skipped from review as they are similar to previous changes (10)
- .mockery.yaml
- pkg/blockchain/interface.go
- pkg/migrations/00004_add_blockchain_columns.down.sql
- sqlc.yaml
- pkg/indexer/storer/identityUpdate.go
- pkg/blockchain/rpcLogStreamer_test.go
- pkg/db/queries/models.go
- pkg/indexer/indexer_test.go
- pkg/indexer/reorgHandler_test.go
- pkg/mocks/indexer/mock_ChainReorgHandler.go
🧰 Additional context used
📓 Learnings (1)
pkg/blockchain/rpcLogStreamer.go (1)
Learnt from: fbac
PR: xmtp/xmtpd#411
File: pkg/blockchain/rpcLogStreamer.go:51-65
Timestamp: 2025-01-16T15:20:37.460Z
Learning: In Go, returning a send-only channel (chan<-) from a public API while using it bidirectionally internally is a valid pattern. It allows the implementation to have full channel access while restricting callers to only writing to the channel. This pattern is used in the RpcLogStreamBuilder.ListenForContractEvent method where reorgChannel is returned as chan<- uint64.
🔇 Additional comments (22)
pkg/migrations/00004_add_blockchain_columns.up.sql (2)
1-9: Specify ON DELETE/UPDATE behavior for the foreign key constraint.The foreign key constraint should explicitly specify the behavior when referenced rows are updated or deleted to prevent data inconsistencies.
Apply this diff:
CREATE TABLE blockchain_messages( block_number BIGINT NOT NULL, block_hash BYTEA NOT NULL, originator_node_id INT NOT NULL, originator_sequence_id BIGINT NOT NULL, is_canonical BOOLEAN NOT NULL DEFAULT TRUE, PRIMARY KEY (block_number, block_hash, originator_node_id, originator_sequence_id), - FOREIGN KEY (originator_node_id, originator_sequence_id) REFERENCES gateway_envelopes(originator_node_id, originator_sequence_id) + FOREIGN KEY (originator_node_id, originator_sequence_id) + REFERENCES gateway_envelopes(originator_node_id, originator_sequence_id) + ON DELETE CASCADE );
11-11: LGTM! Index design supports efficient block range queries.The index on
(block_number, is_canonical)will efficiently support theGetBlocksInRangequery that filters on these columns.pkg/indexer/storer/groupMessage.go (1)
16-19: Consider making originator ID configurable.The TODO comment suggests not hardcoding this value. Consider making it configurable through environment variables or configuration files.
pkg/db/queries.sql (1)
125-129: 🛠️ Refactor suggestionConsider updating existing records on conflict.
The
ON CONFLICT DO NOTHINGclause might miss important updates. Consider updating theis_canonicalflag when conflicts occur.Apply this modification:
INSERT INTO blockchain_messages(block_number, block_hash, originator_node_id, originator_sequence_id, is_canonical) VALUES (@block_number, @block_hash, @originator_node_id, @originator_sequence_id, @is_canonical) -ON CONFLICT - DO NOTHING; +ON CONFLICT (block_number, block_hash, originator_node_id, originator_sequence_id) + DO UPDATE SET + is_canonical = EXCLUDED.is_canonical + WHERE blockchain_messages.is_canonical != EXCLUDED.is_canonical;Likely invalid or redundant comment.
pkg/indexer/reorgHandler.go (1)
100-146: LGTM! Efficient binary search implementation.The binary search implementation efficiently finds the reorganization point while minimizing RPC calls to the blockchain client.
pkg/interceptors/server/auth_test.go (1)
39-39: LGTM! Interface adaptation changes.The changes correctly update the mock verifier calls to match the new interface.
Also applies to: 74-74, 134-134, 159-159
pkg/blockchain/rpcLogStreamer.go (5)
51-65: LGTM! The method signature change properly supports reorg notifications.The addition of
chan<- uint64as a return type is a good design choice:
- It allows callers to receive reorg notifications
- The send-only channel type (
chan<-) correctly restricts callers to only writing to the channel
77-78: LGTM! The struct field addition aligns with the method signature change.The
reorgChannelfield incontractConfigproperly stores the channel returned byListenForContractEvent.
138-143: LGTM! The reorg handling logic is well-implemented.The code properly:
- Updates the
fromBlockto resume from the reorg point- Logs the reorg detection with appropriate context
230-232: LGTM! The Client method provides necessary access to the underlying client.This getter method is essential for reorg detection as it allows access to the chain client for block verification.
131-132:⚠️ Potential issueAvoid closing reorgChannel from the receiver side.
The
watchContractmethod is receiving fromreorgChannel, but the channel should be closed by the sender to prevent panics if the sender attempts to send on a closed channel.defer close(watcher.eventChannel) -defer close(watcher.reorgChannel)Likely invalid or redundant comment.
pkg/indexer/indexer.go (6)
50-63: LGTM! The Close method properly cleans up resources.The method correctly:
- Checks for nil before closing channels
- Closes reorg channels before stopping the streamer
- Maintains proper shutdown order
99-105: LGTM! The StartIndexer method properly initializes reorg handling.The changes correctly pass the reorg channels and handler to the indexLogs function.
Also applies to: 123-134
144-148: LGTM! The builtStreamer struct properly encapsulates reorg handling.The new fields properly support the reorg functionality:
reorgHandlerfor managing reorgsmessagesReorgChannelandidentityUpdatesReorgChannelfor notifications
246-264: LGTM! The reorg state handling is well-implemented.The code properly:
- Discards future events during reorg
- Updates block tracking state
- Logs appropriate messages
266-272: LGTM! The reorg state deactivation is properly handled.The code correctly deactivates the reorg state when processing reaches beyond the reorg finish point.
280-329: LGTM! The reorg detection logic is comprehensive.The code effectively:
- Checks for reorgs at appropriate intervals
- Verifies block hashes
- Updates block tracking
- Notifies via reorg channel
pkg/db/queries/queries.sql.go (4)
78-91: LGTM! The GetBlocksInRange query is well-structured.The query correctly:
- Uses DISTINCT ON for unique blocks
- Filters for non-null hashes and canonical blocks
- Orders by block number ASC
488-495: LGTM! The UpdateBlocksCanonicalityInRange query is correct.The query properly updates the canonicality of blocks from the reorg point onwards.
103-106:⚠️ Potential issueFix incorrect documentation.
The comment states blocks are returned in descending order, but the query uses
ORDER BY block_number ASC.-// Returns blocks in descending order (newest to oldest) +// Returns blocks in ascending order (oldest to newest) // StartBlock should be the lower bound (older block) // EndBlock should be the upper bound (newer block) // Example: GetBlocksInRange(1000, 2000), returns 1000, 1001, 1002, ..., 2000Likely invalid or redundant comment.
193-198: 🛠️ Refactor suggestionAdd ON CONFLICT handling to prevent duplicate entries.
The current query lacks conflict handling for duplicate entries.
INSERT INTO blockchain_messages(block_number, block_hash, originator_node_id, originator_sequence_id, is_canonical) VALUES ($1, $2, $3, $4, $5) -ON CONFLICT +ON CONFLICT (block_number, originator_node_id, originator_sequence_id) + DO UPDATE SET + block_hash = EXCLUDED.block_hash, + is_canonical = EXCLUDED.is_canonical DO NOTHINGLikely invalid or redundant comment.
pkg/mocks/blockchain/mock_ChainClient.go (1)
Line range hint
1-676: LGTM! The mock implementation is complete.The auto-generated mock correctly implements all required methods for chain reorganization testing:
- BlockByHash/Number for block verification
- HeaderByHash/Number for header access
- TransactionCount/InBlock for transaction verification
- SubscribeNewHead for new block notifications
| if err = s.queries.InsertBlockchainMessage(ctx, queries.InsertBlockchainMessageParams{ | ||
| BlockNumber: event.BlockNumber, | ||
| BlockHash: event.BlockHash.Bytes(), | ||
| OriginatorNodeID: GROUP_MESSAGE_ORIGINATOR_ID, | ||
| OriginatorSequenceID: int64(msgSent.SequenceId), | ||
| IsCanonical: true, // New messages are always canonical | ||
| }); err != nil { | ||
| s.logger.Error("Error inserting blockchain message", zap.Error(err)) | ||
| return NewLogStorageError(err, true) | ||
| } |
There was a problem hiding this comment.
Wrap database operations in a transaction.
The gateway envelope and blockchain message insertions should be atomic to maintain data consistency. If the gateway envelope insert fails, the blockchain message could become orphaned.
Apply this modification:
+ tx, err := s.queries.BeginTx(ctx)
+ if err != nil {
+ s.logger.Error("Error beginning transaction", zap.Error(err))
+ return NewLogStorageError(err, true)
+ }
+ defer tx.Rollback()
+
if _, err = s.queries.InsertGatewayEnvelope(ctx, queries.InsertGatewayEnvelopeParams{
OriginatorNodeID: GROUP_MESSAGE_ORIGINATOR_ID,
OriginatorSequenceID: int64(msgSent.SequenceId),
Topic: topicStruct.Bytes(),
OriginatorEnvelope: originatorEnvelopeBytes,
}); err != nil {
s.logger.Error("Error inserting envelope from smart contract", zap.Error(err))
return NewLogStorageError(err, true)
}
if err = s.queries.InsertBlockchainMessage(ctx, queries.InsertBlockchainMessageParams{
BlockNumber: event.BlockNumber,
BlockHash: event.BlockHash.Bytes(),
OriginatorNodeID: GROUP_MESSAGE_ORIGINATOR_ID,
OriginatorSequenceID: int64(msgSent.SequenceId),
IsCanonical: true, // New messages are always canonical
}); err != nil {
s.logger.Error("Error inserting blockchain message", zap.Error(err))
return NewLogStorageError(err, true)
}
+
+ if err = tx.Commit(); err != nil {
+ s.logger.Error("Error committing transaction", zap.Error(err))
+ return NewLogStorageError(err, true)
+ }Committable suggestion skipped: line range outside the PR's diff.
There was a problem hiding this comment.
I feel like the bunny is up to something, but is not quite right. If the gateway envelope insert succeeds, but the blockchain insert fails (or the process dies in between) we will have the gateway envelope with no corresponding blockchain pointer. I suspect that this could lead to the reorg handler missing the gateway envelope.
There was a problem hiding this comment.
@mkysel You're absolutely right! Looking at the migration file, I see that blockchain_messages has a foreign key to gateway_envelopes, so we must maintain the current insert order. Here's the corrected solution:
+ tx, err := s.queries.BeginTx(ctx)
+ if err != nil {
+ s.logger.Error("Error beginning transaction", zap.Error(err))
+ return NewLogStorageError(err, true)
+ }
+ defer func() {
+ if err != nil {
+ if rbErr := tx.Rollback(); rbErr != nil {
+ s.logger.Error("Error rolling back transaction", zap.Error(rbErr))
+ }
+ }
+ }()
if _, err = s.queries.InsertGatewayEnvelope(ctx, queries.InsertGatewayEnvelopeParams{
OriginatorNodeID: GROUP_MESSAGE_ORIGINATOR_ID,
OriginatorSequenceID: int64(msgSent.SequenceId),
Topic: topicStruct.Bytes(),
OriginatorEnvelope: originatorEnvelopeBytes,
}); err != nil {
s.logger.Error("Error inserting envelope from smart contract", zap.Error(err))
return NewLogStorageError(err, true)
}
if err = s.queries.InsertBlockchainMessage(ctx, queries.InsertBlockchainMessageParams{
BlockNumber: event.BlockNumber,
BlockHash: event.BlockHash.Bytes(),
OriginatorNodeID: GROUP_MESSAGE_ORIGINATOR_ID,
OriginatorSequenceID: int64(msgSent.SequenceId),
IsCanonical: true, // New messages are always canonical
}); err != nil {
s.logger.Error("Error inserting blockchain message", zap.Error(err))
return NewLogStorageError(err, true)
}
+ if err = tx.Commit(); err != nil {
+ s.logger.Error("Error committing transaction", zap.Error(err))
+ return NewLogStorageError(err, true)
+ }This ensures that either both inserts succeed or neither does, preventing the scenario where we have an orphaned gateway envelope that the reorg handler might miss.
(_/)<
(•-•)
/> 📝
🔗 Analysis chain
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Check table structure and constraints
cat pkg/migrations/00004_add_blockchain_columns.up.sql
Length of output: 595
✏️ Learnings added
Learnt from: mkysel
PR: xmtp/xmtpd#411
File: pkg/indexer/storer/groupMessage.go:98-107
Timestamp: 2025-01-28T15:04:15.240Z
Learning: The `blockchain_messages` table has a foreign key constraint to `gateway_envelopes` on `(originator_node_id, originator_sequence_id)`. When inserting both records, they must be wrapped in a transaction with gateway_envelope inserted first to maintain referential integrity.
Note: Learnings are effective only in the context of similar code segments. To apply general coding guidelines, please configure review instructions. You can manage existing learnings in the app.
| -- name: UpdateBlocksCanonicalityInRange :exec | ||
| UPDATE | ||
| blockchain_messages | ||
| SET | ||
| is_canonical = FALSE | ||
| WHERE | ||
| block_number >= @reorg_block_number; |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Set transaction isolation level for canonicality updates.
The UpdateBlocksCanonicalityInRange query should run with SERIALIZABLE isolation to prevent race conditions during reorganizations.
Add this comment before the query:
-- Set transaction isolation level to SERIALIZABLE to prevent race conditions
SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;There was a problem hiding this comment.
@fbac is bunny has the right intuition here. Are we holding a lock somewhere to prevent the insertion of new canonical blocks?
There was a problem hiding this comment.
Seems like the humans are having a chat. I'll hop back into my burrow for now. If you need me again, just tag @coderabbitai in a new comment, and I'll come hopping out!
| return 0, nil, ErrNoBlocksFound | ||
| } | ||
|
|
||
| startBlock, endBlock = blockRange(startBlock) |
There was a problem hiding this comment.
I'm not sure, but is it not possible to get into an infinite loop if the returned startBlock is 0, but there are blocks in the storedBlocks results for that range?
I don't know how it would happen in practice in a blockchain, but we might want to have an explicit exit if startBlock is 0 and we don't have the reorg block inside the range.
There was a problem hiding this comment.
In every path we check for startBlock == 0 and return, I don't see how we could get into an infinite loop.
|
|
||
| if len(storedBlocks) == 0 || errors.Is(err, sql.ErrNoRows) { | ||
| if startBlock == 0 { | ||
| return 0, nil, ErrNoBlocksFound |
There was a problem hiding this comment.
I worry we are overloading the ErrNoBlocksFound error. It seems to mean slightly different things in different places (no blocks in the DB, reorg happened earlier in the chain, empty range)
There was a problem hiding this comment.
The idea around it is using it when the range starts at 0 and we still didn't find the reorg point, which means the reorg happened since block 0... very unlikely.
Removing it from searchInRange, it's true that it doesn't make sense there as we'll be always passing ranges at least length == 1.
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (5)
pkg/indexer/reorgHandler.go (5)
15-24: Add godoc comments for the interface and struct.The interface and struct are well-designed, but they would benefit from godoc comments explaining their purpose and responsibilities.
Add documentation like this:
+// ChainReorgHandler defines the interface for detecting blockchain reorganization points type ChainReorgHandler interface { + // FindReorgPoint returns the block number and hash where the chain reorganization occurred + // detectedAt is the block number where the reorganization was first detected FindReorgPoint(detectedAt uint64) (uint64, []byte, error) } +// ReorgHandler implements ChainReorgHandler interface to detect chain reorganizations type ReorgHandler struct { ctx context.Context client blockchain.ChainClient queries *queries.Queries }
25-28: Consider splitting ErrNoBlocksFound into more specific errors.Based on previous discussions,
ErrNoBlocksFoundis being used in different contexts (no blocks in DB, reorg before genesis, empty range). Consider creating separate errors for better error handling:var ( - ErrNoBlocksFound = errors.New("no blocks found") + ErrNoBlocksInDB = errors.New("no blocks found in database") + ErrReorgBeforeGenesis = errors.New("reorganization occurred before genesis block") + ErrEmptyBlockRange = errors.New("no blocks found in specified range") ErrGetBlock = errors.New("failed to get block") )
30-32: Make BLOCK_RANGE_SIZE configurable via environment variable or config file.The TODO comment suggests making this configurable. Consider implementing configuration through environment variables or a config file to allow tuning without recompilation.
Example implementation:
type Config struct { BlockRangeSize uint64 `env:"BLOCK_RANGE_SIZE" default:"1000"` }
45-98: Add detailed documentation and improve error handling in FindReorgPoint.While the implementation is solid, consider these improvements:
- Add detailed documentation explaining the algorithm
- Add explicit logging for better observability
- Consider adding a maximum iteration limit as a safety measure
Example improvements:
+// FindReorgPoint determines where a chain reorganization occurred by comparing stored blocks +// with the current blockchain state. It uses a binary search approach within block ranges +// to efficiently locate the point of divergence. +// +// Parameters: +// - detectedAt: The block number where the reorganization was first detected +// +// Returns: +// - uint64: The block number where the reorganization occurred +// - []byte: The hash of the last canonical block +// - error: Any error encountered during the process func (r *ReorgHandler) FindReorgPoint(detectedAt uint64) (uint64, []byte, error) { + const maxIterations = 100 // Prevent infinite loops + iterations := 0 startBlock, endBlock := blockRange(detectedAt) for { + iterations++ + if iterations > maxIterations { + return 0, nil, fmt.Errorf("exceeded maximum iterations while searching for reorg point") + } + + log.Debug(). + Uint64("startBlock", startBlock). + Uint64("endBlock", endBlock). + Msg("Searching for reorg point in range")
100-142: Add documentation and edge case handling to searchInRange.The binary search implementation is correct, but consider:
- Adding documentation explaining the algorithm
- Adding invariant checks
- Improving the fallback case documentation
+// searchInRange performs a binary search to find the exact block where the chain +// reorganization occurred. It compares stored block hashes with current chain state +// to identify the point of divergence. func (r *ReorgHandler) searchInRange(blocks []queries.GetBlocksInRangeRow) (uint64, []byte, error) { + if len(blocks) == 0 { + return 0, nil, ErrEmptyBlockRange + } left, right := 0, len(blocks)-1
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
pkg/indexer/reorgHandler.go(1 hunks)
🔇 Additional comments (2)
pkg/indexer/reorgHandler.go (2)
1-14: LGTM! Package structure and imports are well-organized.
1-152: Verify similar reorg handling patterns across the codebase.Let's check for similar patterns and potential inconsistencies in reorg handling across the codebase.
✅ Verification successful
Reorg handling implementation is consistent and well-integrated
The reorg handling follows a uniform pattern across the codebase with proper integration between the indexer, streamer, and database layers. The implementation is particularly well-suited for L2 chains with low reorg risk.
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Description: Check for similar reorg handling patterns and potential issues # Look for other reorg-related code echo "Checking for other reorg handling patterns..." rg -l "reorg" --type go # Check for potential uint64 underflow patterns echo "Checking for potential uint64 underflow patterns..." ast-grep --pattern 'if $x > $y { $x = $x - $y }' # Look for similar binary search implementations echo "Checking for similar binary search implementations..." rg -A 5 "left.*right.*mid.*:=.*\(left.*right\).*2" # Check for error handling patterns echo "Checking for similar error definitions and usage..." rg "errors\.New.*block.*found"Length of output: 824
Script:
#!/bin/bash # Examine reorg handling across files echo "Checking reorg handling in indexer.go..." rg -A 10 "reorg" pkg/indexer/indexer.go echo -e "\nChecking reorg handling in rpcLogStreamer.go..." rg -A 10 "reorg" pkg/blockchain/rpcLogStreamer.go echo -e "\nChecking reorg-related database operations..." rg -A 5 "Canonicality|BlockHash" pkg/db/queries/queries.sql.go echo -e "\nChecking for interface implementations..." ast-grep --pattern 'type $_ struct { $$$ } func ($x $_) FindReorgPoint($_) ($_, $_, $_) { $$$ }'Length of output: 8924
| func blockRange(from uint64) (startBlock uint64, endBlock uint64) { | ||
| endBlock = from | ||
|
|
||
| if endBlock > BLOCK_RANGE_SIZE { | ||
| startBlock = endBlock - BLOCK_RANGE_SIZE | ||
| } | ||
|
|
||
| return startBlock, endBlock | ||
| } |
There was a problem hiding this comment.
Add overflow protection to blockRange function.
The function needs protection against uint64 overflow when calculating the start block.
func blockRange(from uint64) (startBlock uint64, endBlock uint64) {
endBlock = from
- if endBlock > BLOCK_RANGE_SIZE {
+ // Protect against uint64 underflow
+ if endBlock >= BLOCK_RANGE_SIZE {
startBlock = endBlock - BLOCK_RANGE_SIZE
+ } else {
+ startBlock = 0
}
return startBlock, endBlock
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| func blockRange(from uint64) (startBlock uint64, endBlock uint64) { | |
| endBlock = from | |
| if endBlock > BLOCK_RANGE_SIZE { | |
| startBlock = endBlock - BLOCK_RANGE_SIZE | |
| } | |
| return startBlock, endBlock | |
| } | |
| func blockRange(from uint64) (startBlock uint64, endBlock uint64) { | |
| endBlock = from | |
| // Protect against uint64 underflow | |
| if endBlock >= BLOCK_RANGE_SIZE { | |
| startBlock = endBlock - BLOCK_RANGE_SIZE | |
| } else { | |
| startBlock = 0 | |
| } | |
| return startBlock, endBlock | |
| } |
| @@ -13,6 +13,11 @@ import ( | |||
| "google.golang.org/protobuf/proto" | |||
| ) | |||
|
|
|||
| const ( | |||
| // We may not want to hardcode this to 0 and have an originator ID for each smart contract? | |||
| GROUP_MESSAGE_ORIGINATOR_ID = 0 | |||
There was a problem hiding this comment.
I thought it was supposted to be >=1 ID < 100...
There was a problem hiding this comment.
we should come up with some term for the blockchain originator. Its not only group messages, its welcome messages as well. And its not ALL group messages...
| @@ -177,6 +180,17 @@ func (s *IdentityUpdateStorer) StoreLog(ctx context.Context, event types.Log) Lo | |||
| return NewLogStorageError(err, true) | |||
| } | |||
|
|
|||
| if err = querier.InsertBlockchainMessage(ctx, queries.InsertBlockchainMessageParams{ | |||
There was a problem hiding this comment.
same atomicity comment here
| FOREIGN KEY (originator_node_id, originator_sequence_id) REFERENCES gateway_envelopes(originator_node_id, originator_sequence_id) | ||
| ); | ||
|
|
||
| CREATE INDEX idx_blockchain_messages_block_canonical ON blockchain_messages(block_number, is_canonical); |
There was a problem hiding this comment.
the GetBlocksInRange can't actually use this index because it is missing parts of the query. You need a composite index such as:
CREATE INDEX idx_blockchain_messages_block_hash_canonical
ON blockchain_messages(block_number, is_canonical, block_hash);
There was a problem hiding this comment.
maybe the index without the hash is sufficient and the table scans won't be too bad. Hard to predict without seeing a query plan.
| ErrGetBlock = errors.New("failed to get block") | ||
| ) | ||
|
|
||
| // TODO: Make this configurable? |
There was a problem hiding this comment.
Rich always says that TODO should have your name in them :) So we know who thought that there is some work left. You will see TODO(rich) and TODO(mkysel) through the codebase. I don't have strong opinions on this.
| // 1.1 Handle active reorg state first | ||
| if reorgDetectedAt > 0 { | ||
| // Under a reorg, future events are no-op | ||
| if event.BlockNumber >= reorgDetectedAt { | ||
| logger.Debug("discarding future event due to reorg", | ||
| zap.Uint64("eventBlockNumber", event.BlockNumber), | ||
| zap.Uint64("reorgBlockNumber", reorgBeginsAt)) | ||
| continue | ||
| } | ||
| logger.Info("starting processing reorg", | ||
| zap.Uint64("eventBlockNumber", event.BlockNumber), | ||
| zap.Uint64("reorgBlockNumber", reorgBeginsAt)) | ||
|
|
||
| // When all future events have been discarded, it means we've reached the reorg point | ||
| storedBlockNumber, storedBlockHash = blockTracker.GetLatestBlock() | ||
| lastBlockSeen = event.BlockNumber | ||
| reorgDetectedAt = 0 | ||
| reorgInProgress = true | ||
| } |
There was a problem hiding this comment.
how does this work in HA? There are multiple indexer processes running. Are they somehow guaranteed to see the same reorg?
|
Once the reorg handler marks the secondary table as |
Closes #322
Summary by CodeRabbit
Release Notes
New Features
Database Changes
blockchain_messagestable to track block-level metadata.Performance Improvements
Testing
Infrastructure