Read your own commits - Stage 1 #438
Conversation
|
Warning Rate limit exceeded@mkysel has exceeded the limit for the number of commits or files that can be reviewed per hour. Please wait 24 minutes and 44 seconds before requesting another review. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. 📒 Files selected for processing (1)
WalkthroughThis pull request introduces a new metadata service for the XMTPD system. The changes include adding environment variable exports to enable metadata and reflection, creating a new Changes
Sequence DiagramsequenceDiagram
participant Client
participant MetadataService
participant CursorUpdater
participant Database
Client->>MetadataService: SubscribeSyncCursor()
MetadataService->>CursorUpdater: AddSubscriber()
CursorUpdater->>Database: Periodically read latest cursor
Database-->>CursorUpdater: Return latest cursor data
CursorUpdater->>CursorUpdater: Update internal cursor
CursorUpdater->>Client: Send cursor updates
Possibly related PRs
Suggested reviewers
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: 3
🧹 Nitpick comments (5)
pkg/api/metadata/service.go (3)
15-20: Consider clarifying field naming.It might be more descriptive to rename the
cufield in theServicestruct to something likecursorUpdaterorupdaterto reflect its purpose more clearly.
22-32: Add error-handling forNewCursorUpdater.The constructor
Func NewMetadataApiServicereturns aServiceafter callingNewCursorUpdater, but any errors from creating or initializing the cursor updater might be lost ifNewCursorUpdaterwere to return an error in future changes. Consider handling potential future errors in this flow or making the constructor safer to reflect initialization failures.
44-83: Evaluate subscriber-based streaming load.While this subscription model is straightforward and functional, consider situations with a large number of clients subscribing concurrently. The current approach spawns a unique channel per subscriber, potentially leading to high memory usage if the number of subscribers grows. A shared publish/subscribe mechanism or event bus might be more efficient for large-scale scenarios.
pkg/api/metadata/cursorUpdater.go (1)
75-89: Optimize subscriber notifications.In
notifySubscribers(), every subscriber gets notified on every read cycle. If you plan to increase the ticker frequency or have many subscribers, consider batching or combining notifications (e.g., only send if the cursor value changed). This can help reduce load on the many channels summoned here.pkg/db/queries.sql (1)
Line range hint
100-136: Consider caching strategy for cursor updates.For the read-your-own-writes guarantee mentioned in the PR objectives, consider the following architectural aspects:
- Implement appropriate caching with a configurable TTL to balance freshness vs performance
- Add appropriate locking mechanisms to prevent race conditions during cursor updates
- Consider adding a mechanism to force cursor refresh when a client writes data, ensuring immediate visibility of their own writes
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (4)
pkg/proto/openapi/xmtpv4/metadata_api/metadata_api.swagger.jsonis excluded by!pkg/proto/**pkg/proto/xmtpv4/metadata_api/metadata_api.pb.gois excluded by!**/*.pb.go,!pkg/proto/**pkg/proto/xmtpv4/metadata_api/metadata_api.pb.gw.gois excluded by!**/*.pb.gw.go,!pkg/proto/**pkg/proto/xmtpv4/metadata_api/metadata_api_grpc.pb.gois excluded by!**/*.pb.go,!pkg/proto/**
📒 Files selected for processing (9)
dev/gen-protos(1 hunks)dev/run(1 hunks)pkg/api/metadata/cursorUpdater.go(1 hunks)pkg/api/metadata/service.go(1 hunks)pkg/config/options.go(2 hunks)pkg/config/validation.go(1 hunks)pkg/db/queries.sql(1 hunks)pkg/db/queries/queries.sql.go(1 hunks)pkg/server/server.go(4 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (2)
- GitHub Check: Test (Node)
- GitHub Check: Build pre-baked anvil-xmtpd
🔇 Additional comments (10)
dev/gen-protos (1)
4-4: Validate branch availability.Referencing the
mkysel/metadata-endpointbranch is fine, but ensure it’s consistently updated or merged to avoid version mismatches during protobuf generation.dev/run (1)
7-12: LGTM! Environment variables are properly configured.The addition of
XMTPD_METADATA_ENABLEaligns with the PR objectives to enable the metadata service, andXMTPD_REFLECTION_ENABLEis beneficial for development and debugging.pkg/config/validation.go (1)
62-66: LGTM! Validation logic is consistent and complete.The validation ensures that the required database connection is available when the metadata service is enabled, following the same pattern as other services.
pkg/server/server.go (2)
19-19: LGTM! Required imports are properly added.The metadata-related package imports are correctly added.
Also applies to: 29-29
129-129: LGTM! API server condition is properly updated.The condition is correctly extended to include the metadata service.
pkg/config/options.go (2)
54-56: LGTM! MetadataOptions type is well-defined.The type follows the established pattern and includes clear documentation.
87-87: LGTM! ServerOptions is properly updated.The Metadata field is correctly added with appropriate group and namespace, maintaining the struct's organization.
pkg/db/queries/queries.sql.go (2)
110-113: LGTM! Implementation follows best practices.The Go implementation follows proper error handling patterns and resource management.
Also applies to: 115-136
100-108: LGTM! Query implementation looks correct.The SQL query is well-structured and uses appropriate type casting.
Let's verify the query performance on large datasets:
✅ Verification successful
Query performance verified - optimal execution plan expected
The query will efficiently use the primary key index (originator_node_id, originator_sequence_id) for both grouping and MAX aggregation, ensuring good performance even on large datasets.
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Description: Check the query execution plan and table statistics # to ensure efficient execution on large datasets # Check if there's an index on originator_node_id rg -A 5 "CREATE.*INDEX.*originator_node_id" . # Look for table creation to understand the schema rg -A 10 "CREATE.*TABLE.*gateway_envelopes" .Length of output: 922
pkg/db/queries.sql (1)
125-132: LGTM! SQL query definition is well-structured.The query follows the established patterns in the file and uses appropriate type annotations.
pkg/api/metadata/cursorUpdater.go
Outdated
| func (cu *CursorUpdater) start() { | ||
| ticker := time.NewTicker(100 * time.Millisecond) // Adjust the period as needed | ||
| defer ticker.Stop() | ||
| for { | ||
| select { | ||
| case <-cu.ctx.Done(): | ||
| return | ||
| case <-ticker.C: | ||
| err := cu.read() | ||
| if err != nil { | ||
| //TODO proper error handling | ||
| return | ||
| } | ||
| cu.notifySubscribers() | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Provide resiliency to transient read failures.
The start method stops processing at line 48 if cu.read() returns an error. Consider logging the error and continuing to retry instead of ceasing the entire updater. In production, transient database failures could cause the service to terminate prematurely.
|
| @@ -124,7 +126,7 @@ func NewReplicationServer( | |||
| log.Info("Indexer service enabled") | |||
| } | |||
|
|
|||
| if options.Payer.Enable || options.Replication.Enable { | |||
| if options.Payer.Enable || options.Replication.Enable || options.Metadata.Enable { | |||
There was a problem hiding this comment.
Should we simplify this by including the metadata api every time payer and/or replication are enabled? I think we might be overcomplicating this.
Co-authored-by: Francisco de Borja Aranda Castillejo <[email protected]>
Co-authored-by: Francisco de Borja Aranda Castillejo <[email protected]>
|
Relates to #418 |
Depends on xmtp/proto#241
This adds a new metadata service to the XMTPD GRPC API which exposes the current last synced vector/cursor.
The next PR will teach the payer how to subscribe to this to have some sort of read-your-own-commits guarantees. At least for early stages of development while the client expects these.
Summary by CodeRabbit
New Features
Configuration
Backend Improvements