Start storing congestion per minute #598
Conversation
WalkthroughThis pull request introduces concurrent execution within the Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant Function as InsertGatewayEnvelopeAndIncrementUnsettledUsage
participant TxQueries
Client->>Function: Call InsertGatewayEnvelopeAndIncrementUnsettledUsage
par Concurrent Operations
Function->>TxQueries: IncrementUnsettledUsage
Function->>TxQueries: IncrementOriginatorCongestion
end
TxQueries-->>Function: Return increment results (unsettled usage & congestion)
Function->>Client: Return inserted rows or error based on results
Possibly related PRs
Suggested reviewers
Warning There were issues while running some tools. Please review the errors and either fix the tool’s configuration or disable the tool if it’s a critical failure. 🔧 golangci-lint (1.62.2)Error: can't load config: the Go language version (go1.23) used to build golangci-lint is lower than the targeted Go version (1.24) ✨ Finishing Touches
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
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: 0
🧹 Nitpick comments (2)
pkg/db/gatewayEnvelope_test.go (2)
69-74: Consider additional test cases for congestion metricsThe test verifies that congestion is tracked, but it doesn't explicitly verify that the minutes_since_epoch is stored correctly or how multiple messages in the same minute are handled.
Consider adding tests that:
- Verify the correct minute is used (matching incrementParams.MinutesSinceEpoch)
- Test multiple insertions within the same minute to ensure num_messages increments correctly
143-148: Verify congestion behavior with multiple insertionsWhile this parallel test checks for race conditions, it only verifies a single successful insertion. Consider extending this test to verify how congestion is tracked when multiple insertions succeed.
Since totalInserted is 1 (due to sequence ID uniqueness constraint), but you're making multiple insertion attempts, it would be valuable to test a scenario where multiple distinct messages are successfully inserted to verify congestion increments correctly.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (7)
pkg/db/gatewayEnvelope.go(2 hunks)pkg/db/gatewayEnvelope_test.go(3 hunks)pkg/db/queries.sql(1 hunks)pkg/db/queries/models.go(1 hunks)pkg/db/queries/queries.sql.go(5 hunks)pkg/migrations/00009_originator-congestion.down.sql(1 hunks)pkg/migrations/00009_originator-congestion.up.sql(1 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)
pkg/migrations/00009_originator-congestion.down.sql (1)
1-2: SQL migration looks good!This down migration correctly drops the originator_congestion table to revert the changes made by the up migration.
pkg/db/queries/models.go (1)
53-57: Data model looks good for tracking congestion metricsThe
OriginatorCongestionstruct is well-structured and matches the schema defined in the migration files, with appropriate types for each field.pkg/migrations/00009_originator-congestion.up.sql (1)
1-6: Table schema looks good!The table design with a composite primary key on (originator_id, minutes_since_epoch) is appropriate for tracking congestion per minute. The default value of 0 for num_messages is also a sensible choice.
pkg/db/gatewayEnvelope_test.go (1)
9-10: Appropriate import alias addedAdding the explicit xmtpd_db alias makes it clear where the functions are coming from.
pkg/db/gatewayEnvelope.go (1)
34-63: Great implementation of concurrent operations!The introduction of concurrent execution for incrementing unsettled usage and originator congestion using goroutines and WaitGroup is a good performance improvement. The implementation correctly:
- Uses
sync.WaitGroupto coordinate both operations- Captures errors from each goroutine individually
- Properly handles error cases in sequence after both operations complete
pkg/db/queries.sql (2)
234-251: SQL queries for originator congestion look good.The new SQL queries for tracking and retrieving originator congestion are well structured and follow the same patterns as existing queries. The implementation will correctly:
- Insert or update congestion metrics in the
originator_congestiontable- Retrieve congestion metrics with appropriate filtering based on time range
203-232: Formatting improvements enhance readability.The formatting adjustments to the SQL queries improve code readability while maintaining the same functionality.
pkg/db/queries/queries.sql.go (3)
292-316: Well-implemented function for retrieving originator congestion.The implementation follows the established pattern for database query functions and correctly handles parameter passing and result processing.
344-360: Correctly implemented function for incrementing originator congestion.The function correctly passes the parameters to the SQL query and properly handles errors.
274-282: Formatting enhancements to SQL queries improve readability.The formatting changes to the constant SQL strings align with the changes in the queries.sql file and improve readability without changing functionality.
TL;DR
Added originator congestion tracking to measure message volume per originator over time.
What changed?
originator_congestiontable to track message counts per originatorOther ideas
I thought about including message counts on the
unsettled_usagetable. I decided against because I'm worried that the lifecycle of the rows might be different. If we get a payer report, we will delete allunsettled_usagefor that payer from before the report. That could be in the last 5 minutes, where we would still need the congestion information.It would also mean we would need to store congestion per-payer, instead of a single value per originator.
How to test?
TestInsertAndIncrementandTestInsertAndIncrementParallelpassWhy make this change?
This change enables tracking message volume per originator, which is essential for implementing rate limiting and preventing potential abuse of the system. The concurrent updates ensure efficient processing while maintaining accurate metrics for both usage and congestion.