Metadata Service postgres db implementation#4846
Conversation
…g storage backend.
…etadata_service files.
…te tests accordingly
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
PR SummaryIntroduces a PostgreSQL-backed metadata store with block-ordered updates, notifications, and comprehensive tests.
Written by Cursor Bugbot for commit 388106d. This will update automatically on new commits. Configure here. |
📝 WalkthroughWalkthroughAdds a PostgreSQL-backed metadata service: new SQL migrations for versioned, block-serialized metadata tables and notifications; a PostgresMetadataServiceStore implementing block-based updates, listing, counts, LISTEN/NOTIFY and validation; and comprehensive tests for store behavior. Changes
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested reviewers
🚥 Pre-merge checks | ✅ 1 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (1 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
🧪 Generate unit tests (beta)
📜 Recent review detailsConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro 📒 Files selected for processing (2)
🧰 Additional context used📓 Path-based instructions (8)core/**/*.go📄 CodeRabbit inference engine (core/AGENTS.md)
Files:
core/**/*_test.go📄 CodeRabbit inference engine (core/AGENTS.md)
Files:
core/node/storage/pg_*.go📄 CodeRabbit inference engine (core/AGENTS.md)
Files:
core/node/storage/**/*.go📄 CodeRabbit inference engine (core/AGENTS.md)
Files:
**/core/**/*.go📄 CodeRabbit inference engine (AGENTS.md)
Files:
**/core/**/*_test.go📄 CodeRabbit inference engine (AGENTS.md)
Files:
**/core/node/**/*.go📄 CodeRabbit inference engine (AGENTS.md)
Files:
**/core/node/storage/**/*_test.go📄 CodeRabbit inference engine (AGENTS.md)
Files:
🧠 Learnings (26)📓 Common learnings📚 Learning: 2025-12-17T02:40:57.825ZApplied to files:
📚 Learning: 2026-01-08T18:27:30.866ZApplied to files:
📚 Learning: 2025-12-17T02:40:57.825ZApplied to files:
📚 Learning: 2025-12-17T02:40:57.825ZApplied to files:
📚 Learning: 2025-11-25T08:48:08.084ZApplied to files:
📚 Learning: 2025-11-25T08:48:08.084ZApplied to files:
📚 Learning: 2025-11-25T08:48:08.084ZApplied to files:
📚 Learning: 2025-11-25T08:47:37.754ZApplied to files:
📚 Learning: 2025-11-25T08:48:08.084ZApplied to files:
📚 Learning: 2025-11-25T08:48:08.084ZApplied to files:
📚 Learning: 2025-11-25T08:48:08.084ZApplied to files:
📚 Learning: 2025-11-25T08:48:08.084ZApplied to files:
📚 Learning: 2025-11-25T08:48:08.084ZApplied to files:
📚 Learning: 2025-11-25T08:48:08.084ZApplied to files:
📚 Learning: 2025-11-25T08:47:37.754ZApplied to files:
📚 Learning: 2025-12-17T02:40:57.825ZApplied to files:
📚 Learning: 2025-11-25T08:47:01.034ZApplied to files:
📚 Learning: 2025-11-25T08:48:08.084ZApplied to files:
📚 Learning: 2025-11-25T08:48:08.084ZApplied to files:
📚 Learning: 2025-11-25T08:48:08.084ZApplied to files:
📚 Learning: 2025-11-25T08:48:08.084ZApplied to files:
📚 Learning: 2025-11-25T08:48:08.084ZApplied to files:
📚 Learning: 2025-11-25T08:48:08.084ZApplied to files:
📚 Learning: 2025-11-25T08:48:08.084ZApplied to files:
📚 Learning: 2025-11-25T08:48:08.084ZApplied to files:
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (7)
🔇 Additional comments (20)
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. Comment |
![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)
core/node/storage/metadata_migrations/000001_create_initial_schema.up.sql (1)
27-47: LGTM!Composite primary key
(block_num, block_slot)is appropriate for ordered block retrieval. Comment at line 31-32 has a typo: "event not updated" should likely be "even if not updated".core/node/storage/pg_metadata_service.go (1)
912-939: Variable shadowing at line 926.
blockNumis redeclared in the inner scope, which shadows the outerblockNum. While the behavior is correct (returns immediately), this can be confusing.♻️ Suggested fix
blockNum, err := strconv.ParseInt(payload, 10, 64) if err != nil { - blockNum, lastErr := s.GetLastRecordBlockNum(ctx) + fallbackBlockNum, lastErr := s.GetLastRecordBlockNum(ctx) if lastErr != nil { return 0, RiverErrorWithBase( protocol.Err_INVALID_ARGUMENT, "invalid block notification payload", err, "payload", payload, ).Func("parseRecordBlockNotification") } - return blockNum, nil + return fallbackBlockNum, nil }
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (4)
core/node/storage/metadata_migrations/000001_create_initial_schema.down.sqlcore/node/storage/metadata_migrations/000001_create_initial_schema.up.sqlcore/node/storage/pg_metadata_service.gocore/node/storage/pg_metadata_service_test.go
🧰 Additional context used
📓 Path-based instructions (8)
core/**/*.go
📄 CodeRabbit inference engine (core/AGENTS.md)
core/**/*.go: Format all modified Go files using ./fmt.sh from the /core directory before committing
Run the linter using ./lint.sh from the /core directory to check code quality, which runs both golangci-lint and staticcheck
Use RiverError type from node/base/error.go for structured errors
Use RiverErrorWithBase to construct a river error derived from an error of arbitrary or unknown type
Wrap blockchain errors with context using crypto/ utilities
Log errors with structured fields using zap logger
Use node/crypto/blockchain.go for Ethereum client operations
Use node/crypto/chain_monitor.go for transaction monitoring
Use xchain/entitlement/ for cross-chain entitlement checks
Do not run ./fmt.sh and ./lint.sh at the same time; run them as separate commands
Use go.uber.org/zap for structured logging with custom extensions from node/logging/
Use go.opentelemetry.io for distributed tracing across the application
Use prometheus for OpenMetrics in application metrics collection
Files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
core/**/*_test.go
📄 CodeRabbit inference engine (core/AGENTS.md)
core/**/*_test.go: Use testify/require and testify/assert for test assertions in Go test files
Database tests should use testutils/dbtestutils for test DB lifecycle management
Do not set GOCACHE or GOMODCACHE when running Go test commands; run go commands directly without setting extra environment variables
Set RIVER_TEST_LOG=info and RIVER_TEST_PRINT=1 environment variables for debug output in tests
Mock implementations should be used from testutils/mocks/ in tests
Files:
core/node/storage/pg_metadata_service_test.go
core/node/storage/pg_*.go
📄 CodeRabbit inference engine (core/AGENTS.md)
core/node/storage/pg_*.go: Use PostgreSQL transactions for consistency in all storage operations
All storage operations should be in node/storage/pg_*.go files
Files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
core/node/storage/**/*.go
📄 CodeRabbit inference engine (core/AGENTS.md)
core/node/storage/**/*.go: Use PostgreSQL transactions in node/storage/ for all database operations to ensure consistency
Use jackc/pgx/v5 for PostgreSQL connection pooling and database driver operations
Files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
**/core/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
**/core/**/*.go: Format all Go files using the fmt.sh script from the /core directory
Run the linter using the lint.sh script from the /core directory for all Go code changes
Use structured logging with zap for Go development
Implement proper error handling with RiverError types in Go code
Implement proper context cancellation for graceful shutdowns in Go code
Files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
**/core/**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
**/core/**/*_test.go: Use testify for test assertions in Go tests
Make tests re-runnable by using unique identifiers (e.g., UUIDs) instead of hardcoded values
Files:
core/node/storage/pg_metadata_service_test.go
**/core/node/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
Use structured logging with zap in Go backend River nodes for observability
Files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
**/core/node/storage/**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
Storage tests should verify data persistence and constraints for database schema changes
Files:
core/node/storage/pg_metadata_service_test.go
🧠 Learnings (19)
📓 Common learnings
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Make changes to storage interfaces first, then implement PostgreSQL-specific logic, write comprehensive tests, benchmark critical paths, and update migrations if schema changes are needed
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/AGENTS.md:0-0
Timestamp: 2025-12-17T02:40:57.825Z
Learning: Applies to core/node/storage/**/*.go : Use PostgreSQL transactions in node/storage/ for all database operations to ensure consistency
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/AGENTS.md:0-0
Timestamp: 2025-12-17T02:40:57.825Z
Learning: Applies to core/node/storage/pg_*.go : Use PostgreSQL transactions for consistency in all storage operations
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/AGENTS.md:0-0
Timestamp: 2025-12-17T02:40:57.825Z
Learning: Applies to core/node/storage/pg_*.go : All storage operations should be in node/storage/pg_*.go files
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: AGENTS.md:0-0
Timestamp: 2026-01-08T18:27:30.866Z
Learning: Applies to **/core/node/storage/**/*_test.go : Storage tests should verify data persistence and constraints for database schema changes
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/app_registry/AGENTS.md:0-0
Timestamp: 2025-11-25T08:47:01.034Z
Learning: Applies to core/node/app_registry/**/app_registry/**/*.go : When adding storage methods, implement in three layers: AppRegistryStore interface, PostgresAppRegistryStore implementation, and CachedEncryptedMessageQueue delegation
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/AGENTS.md:0-0
Timestamp: 2025-12-17T02:40:57.825Z
Learning: Applies to core/node/storage/**/*.go : Use jackc/pgx/v5 for PostgreSQL connection pooling and database driver operations
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*.go : Use `COPY FROM` for atomic batch miniblock writes instead of individual insert statements in loops
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/migrations/*.sql : Migrations use golang-migrate with embedded SQL files; always test migrations on a copy of production data and consider partition count impacts
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*.go : Always include stream ID in queries to leverage table partitioning (256 partitions by default)
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Make changes to storage interfaces first, then implement PostgreSQL-specific logic, write comprehensive tests, benchmark critical paths, and update migrations if schema changes are needed
Applied to files:
core/node/storage/metadata_migrations/000001_create_initial_schema.up.sql
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*.go : Enforce miniblock requirement constraint: if a stream exists in the `es` (event streams) table, it MUST have at least one miniblock; empty streams are not allowed
Applied to files:
core/node/storage/metadata_migrations/000001_create_initial_schema.up.sqlcore/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*.go : Always include stream ID in queries to leverage table partitioning (256 partitions by default)
Applied to files:
core/node/storage/metadata_migrations/000001_create_initial_schema.up.sqlcore/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
📚 Learning: 2026-01-08T18:27:30.866Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: AGENTS.md:0-0
Timestamp: 2026-01-08T18:27:30.866Z
Learning: Applies to **/core/node/storage/**/*_test.go : Storage tests should verify data persistence and constraints for database schema changes
Applied to files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*_test.go : Implement table-driven tests when testing multiple scenarios to improve maintainability and coverage
Applied to files:
core/node/storage/pg_metadata_service_test.go
📚 Learning: 2025-12-17T02:40:57.825Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/AGENTS.md:0-0
Timestamp: 2025-12-17T02:40:57.825Z
Learning: Applies to core/node/storage/pg_*.go : All storage operations should be in node/storage/pg_*.go files
Applied to files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
📚 Learning: 2025-12-17T02:40:57.825Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/AGENTS.md:0-0
Timestamp: 2025-12-17T02:40:57.825Z
Learning: Applies to core/node/storage/**/*.go : Use PostgreSQL transactions in node/storage/ for all database operations to ensure consistency
Applied to files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
📚 Learning: 2025-12-17T02:40:57.825Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/AGENTS.md:0-0
Timestamp: 2025-12-17T02:40:57.825Z
Learning: Applies to core/node/storage/pg_*.go : Use PostgreSQL transactions for consistency in all storage operations
Applied to files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*_test.go : Always test edge cases including: empty/nil values, non-existent entities (NOT_FOUND errors), duplicate entries (ALREADY_EXISTS errors), concurrent operations, and transaction rollbacks
Applied to files:
core/node/storage/pg_metadata_service_test.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*_test.go : Use helper functions to generate test data consistently: `safeAddress(t)` for Ethereum addresses, `testAppMetadataWithName(name string)` for app metadata, `testutils.FakeStreamId(streamType)` for stream IDs
Applied to files:
core/node/storage/pg_metadata_service_test.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*_test.go : All storage tests should follow the consistent setup pattern using helper functions: `setupStreamStorageTest(t)`, `setupAppRegistryStorageTest(t)`, or `prepareNotificationsDB(ctx)` depending on storage type, with automatic cleanup via `t.Cleanup(params.closer)`
Applied to files:
core/node/storage/pg_metadata_service_test.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*.go : Use `lockStream(ctx, tx, streamId, true)` with `FOR UPDATE` for write operations and `lockStream(ctx, tx, streamId, false)` with `FOR SHARE` for read operations to ensure proper row-level locking
Applied to files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*_test.go : Use `require.New(t)` at the beginning of test functions and check error codes with `base.IsRiverErrorCode(err, code)` for specific error type verification
Applied to files:
core/node/storage/pg_metadata_service_test.go
📚 Learning: 2025-11-25T08:47:37.754Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/rpc/AGENTS.md:0-0
Timestamp: 2025-11-25T08:47:37.754Z
Learning: Applies to core/node/rpc/**/*_test.go : Enable River logs with `enableRiverLogs: true` in tester options when debugging test failures
Applied to files:
core/node/storage/pg_metadata_service_test.go
📚 Learning: 2025-12-17T02:40:57.825Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/AGENTS.md:0-0
Timestamp: 2025-12-17T02:40:57.825Z
Learning: Applies to core/node/storage/**/*.go : Use jackc/pgx/v5 for PostgreSQL connection pooling and database driver operations
Applied to files:
core/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*.go : Use `COPY FROM` for atomic batch miniblock writes instead of individual insert statements in loops
Applied to files:
core/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:47:01.034Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/app_registry/AGENTS.md:0-0
Timestamp: 2025-11-25T08:47:01.034Z
Learning: Applies to core/node/app_registry/**/app_registry/**/*.go : When adding storage methods, implement in three layers: AppRegistryStore interface, PostgresAppRegistryStore implementation, and CachedEncryptedMessageQueue delegation
Applied to files:
core/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*.go : Use the `txRunner` function for all database operations to provide automatic retry for serialization failures and deadlocks, exponential backoff with jitter, transaction tracking, debugging, and comprehensive metrics and logging
Applied to files:
core/node/storage/pg_metadata_service.go
🧬 Code graph analysis (1)
core/node/storage/pg_metadata_service_test.go (7)
core/node/storage/pg_metadata_service.go (3)
PostgresMetadataServiceStore(149-153)PostgresMetadataServiceStore(941-949)NewPostgresMetadataServiceStore(161-172)core/config/config.go (1)
DatabaseConfig(259-287)core/node/base/test/context.go (1)
NewTestContext(15-23)core/node/testutils/dbtestutils/db.go (1)
ConfigureDbWithPrefix(60-70)core/node/storage/pg_storage.go (1)
CreateAndValidatePgxPool(243-254)core/node/shared/stream_id.go (2)
STREAM_CHANNEL_BIN(16-16)StreamId(41-41)core/node/protocol/protocol.pb.go (1)
Err_BAD_PREV_MINIBLOCK_HASH(457-457)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (7)
- GitHub Check: Multinode_Ent
- GitHub Check: Multinode_Ent_Legacy
- GitHub Check: Multinode
- GitHub Check: Go_Tests
- GitHub Check: Common_CI
- GitHub Check: Cursor Bugbot
- GitHub Check: XChain_Integration
🔇 Additional comments (27)
core/node/storage/pg_metadata_service_test.go (7)
30-72: LGTM!Test setup follows established patterns: uses
require.New(t),dbtestutils.ConfigureDbWithPrefix, proper cleanup viat.Cleanup. Based on learnings, this aligns with storage test conventions.
78-207: LGTM!Comprehensive test covering initial state, batch inserts, pagination, node filtering, record retrieval, counts, and block retrieval with event mask validation.
209-333: LGTM!Good coverage of mixed update types, partial batch success with per-update errors, and EventMask composition verification.
335-402: LGTM!Correctly validates block trimming behavior with
numBlocksToKeep=2.
404-471: LGTM!Correctly validates that
numBlocksToKeep=0preserves all blocks.
473-484: LGTM!Input validation tests for
GetRecordBlocksare appropriate.
486-540: LGTM!Good coverage of
OnNewRecordBlocknotification behavior and channel lifecycle. The 50ms sleep for listener setup is acceptable for test synchronization.core/node/storage/metadata_migrations/000001_create_initial_schema.up.sql (4)
1-14: LGTM!Good schema with appropriate constraints for stream_id length, hash length, and non-negative miniblock numbers.
16-25: LGTM!Denormalization for placement is a good design choice to avoid GIN index updates on frequent hash changes.
49-60: LGTM!Singleton pattern with
CHECK (singleton_key = TRUE)is a clean approach for the global block counter and serialization lock.
62-79: LGTM!Trigger function and trigger setup are correct.
DROP TRIGGER IF EXISTSbeforeCREATE TRIGGERensures migration idempotency.core/node/storage/pg_metadata_service.go (16)
30-85: LGTM!Clean type definitions with proper bit flags for event masks and clear separation of insert/update operations.
87-147: LGTM!Interface is well-documented with clear contracts for pagination, batch updates with per-update errors, and async notifications.
161-198: LGTM!Constructor and initialization correctly embed migrations and use
AsRiverErrorfor error wrapping.
200-262: LGTM!Correct use of
RepeatableReadisolation for snapshot consistency. Cursor-based pagination is efficient.
264-367: LGTM!Query variants are correctly structured with parameterized queries and proper ordering.
369-448: LGTM!Serialization via
FOR UPDATEon singleton row is correct. Per-update error handling allows partial batch success. Trimming logic is sound.
450-483: LGTM!Clean validation ensuring exactly one operation per update and proper dispatch.
485-582: LGTM!Thorough input validation and defensive handling of unique violations. Correctly inserts into both
md_stream_recordsandmd_stream_placement.
584-671: LGTM!Correctly handles all placement update combinations and only updates
md_stream_placementwhen nodes actually change.
673-740: LGTM!Correct precondition checks for miniblock updates: sealed stream check, hash chain validation, and sequential block number enforcement.
742-799: LGTM!Correct use of transactions for read consistency with appropriate logging options.
818-879: LGTM!Well-structured async listener with proper cleanup, backoff retry for resilience, and correct connection release before reconnection.
881-910: LGTM!Correct backoff pattern with proper resource cleanup on partial failures.
951-1040: LGTM!Correct range validation, query construction, and ordered block retrieval.
1071-1099: LGTM!Correct use of
FOR UPDATEfor write operations as per coding guidelines.
1174-1210: LGTM!Error classification and node index validation are well-implemented.
…lace of separate hash and num fields.
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
| if err := s.insertRecordBlockTx(ctx, tx, blockNum, blockSlot, mask, record); err != nil { | ||
| return err | ||
| } | ||
| blockSlot++ |
There was a problem hiding this comment.
Stale errors retained in errs slice after transaction retry
Medium Severity
The errs slice is created once at line 377 outside the txRunner callback, but the callback can be retried on serialization failures or deadlocks. When a retry occurs, the errs slice is not reset. If update i fails with a validation error on attempt 1 (e.g., Err_NOT_FOUND) but succeeds on retry attempt 2, errs[i] will still contain the stale error from attempt 1. The caller would incorrectly believe the update failed when it actually succeeded. The fix would be to reset the errs slice at the start of the callback or explicitly set errs[i] = nil on successful updates.
🔬 Verification Test
Why verification test was not possible: This bug requires triggering a PostgreSQL serialization failure or deadlock during the transaction, followed by a successful retry where a previously failed update now succeeds. This requires coordinating concurrent database transactions with precise timing to induce the serialization conflict, which cannot be reliably reproduced in a unit test without mocking the database layer or using specialized concurrency testing tools. The bug is evident from code inspection: the errs slice is captured by closure and never cleared inside the callback that txRunner may retry.
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (3)
core/node/storage/pg_metadata_service_test.go (1)
490-493: Potential test flakiness with sleep-based synchronization.The 50ms sleep assumes the listener goroutine has started and subscribed. Under CI load, this could occasionally fail. Consider adding a ready signal or retry logic.
core/node/storage/pg_metadata_service.go (2)
272-293: Consider cursor-based pagination for large datasets.Currently loads all records into memory before paginating. For very large stream counts, this could cause memory pressure. Database-level pagination with LIMIT/OFFSET or cursors would be more scalable.
832-858: Silent fallback on parse error may hide issues.If notification payload is malformed, the code falls back to
GetLastRecordBlockNumwithout logging. Consider logging a warning for unexpected payload formats to aid debugging.
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
core/node/storage/metadata_migrations/000001_create_initial_schema.up.sqlcore/node/storage/pg_metadata_service.gocore/node/storage/pg_metadata_service_test.go
🧰 Additional context used
📓 Path-based instructions (8)
core/**/*.go
📄 CodeRabbit inference engine (core/AGENTS.md)
core/**/*.go: Format all modified Go files using ./fmt.sh from the /core directory before committing
Run the linter using ./lint.sh from the /core directory to check code quality, which runs both golangci-lint and staticcheck
Use RiverError type from node/base/error.go for structured errors
Use RiverErrorWithBase to construct a river error derived from an error of arbitrary or unknown type
Wrap blockchain errors with context using crypto/ utilities
Log errors with structured fields using zap logger
Use node/crypto/blockchain.go for Ethereum client operations
Use node/crypto/chain_monitor.go for transaction monitoring
Use xchain/entitlement/ for cross-chain entitlement checks
Do not run ./fmt.sh and ./lint.sh at the same time; run them as separate commands
Use go.uber.org/zap for structured logging with custom extensions from node/logging/
Use go.opentelemetry.io for distributed tracing across the application
Use prometheus for OpenMetrics in application metrics collection
Files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
core/**/*_test.go
📄 CodeRabbit inference engine (core/AGENTS.md)
core/**/*_test.go: Use testify/require and testify/assert for test assertions in Go test files
Database tests should use testutils/dbtestutils for test DB lifecycle management
Do not set GOCACHE or GOMODCACHE when running Go test commands; run go commands directly without setting extra environment variables
Set RIVER_TEST_LOG=info and RIVER_TEST_PRINT=1 environment variables for debug output in tests
Mock implementations should be used from testutils/mocks/ in tests
Files:
core/node/storage/pg_metadata_service_test.go
core/node/storage/pg_*.go
📄 CodeRabbit inference engine (core/AGENTS.md)
core/node/storage/pg_*.go: Use PostgreSQL transactions for consistency in all storage operations
All storage operations should be in node/storage/pg_*.go files
Files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
core/node/storage/**/*.go
📄 CodeRabbit inference engine (core/AGENTS.md)
core/node/storage/**/*.go: Use PostgreSQL transactions in node/storage/ for all database operations to ensure consistency
Use jackc/pgx/v5 for PostgreSQL connection pooling and database driver operations
Files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
**/core/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
**/core/**/*.go: Format all Go files using the fmt.sh script from the /core directory
Run the linter using the lint.sh script from the /core directory for all Go code changes
Use structured logging with zap for Go development
Implement proper error handling with RiverError types in Go code
Implement proper context cancellation for graceful shutdowns in Go code
Files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
**/core/**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
**/core/**/*_test.go: Use testify for test assertions in Go tests
Make tests re-runnable by using unique identifiers (e.g., UUIDs) instead of hardcoded values
Files:
core/node/storage/pg_metadata_service_test.go
**/core/node/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
Use structured logging with zap in Go backend River nodes for observability
Files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
**/core/node/storage/**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
Storage tests should verify data persistence and constraints for database schema changes
Files:
core/node/storage/pg_metadata_service_test.go
🧠 Learnings (27)
📓 Common learnings
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Make changes to storage interfaces first, then implement PostgreSQL-specific logic, write comprehensive tests, benchmark critical paths, and update migrations if schema changes are needed
📚 Learning: 2025-12-17T02:40:57.825Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/AGENTS.md:0-0
Timestamp: 2025-12-17T02:40:57.825Z
Learning: Applies to core/node/storage/pg_*.go : All storage operations should be in node/storage/pg_*.go files
Applied to files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
📚 Learning: 2026-01-08T18:27:30.866Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: AGENTS.md:0-0
Timestamp: 2026-01-08T18:27:30.866Z
Learning: Applies to **/core/node/storage/**/*_test.go : Storage tests should verify data persistence and constraints for database schema changes
Applied to files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
📚 Learning: 2025-12-17T02:40:57.825Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/AGENTS.md:0-0
Timestamp: 2025-12-17T02:40:57.825Z
Learning: Applies to core/node/storage/**/*.go : Use PostgreSQL transactions in node/storage/ for all database operations to ensure consistency
Applied to files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
📚 Learning: 2025-12-17T02:40:57.825Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/AGENTS.md:0-0
Timestamp: 2025-12-17T02:40:57.825Z
Learning: Applies to core/node/storage/pg_*.go : Use PostgreSQL transactions for consistency in all storage operations
Applied to files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*_test.go : Implement table-driven tests when testing multiple scenarios to improve maintainability and coverage
Applied to files:
core/node/storage/pg_metadata_service_test.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*_test.go : Use helper functions to generate test data consistently: `safeAddress(t)` for Ethereum addresses, `testAppMetadataWithName(name string)` for app metadata, `testutils.FakeStreamId(streamType)` for stream IDs
Applied to files:
core/node/storage/pg_metadata_service_test.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*_test.go : Always test edge cases including: empty/nil values, non-existent entities (NOT_FOUND errors), duplicate entries (ALREADY_EXISTS errors), concurrent operations, and transaction rollbacks
Applied to files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:47:37.754Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/rpc/AGENTS.md:0-0
Timestamp: 2025-11-25T08:47:37.754Z
Learning: Applies to core/node/rpc/**/*_test.go : Generate test wallets and metadata using helper functions like `safeNewWallet()` and `appMetadataForBot()` for realistic test data
Applied to files:
core/node/storage/pg_metadata_service_test.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*.go : Extract transaction logic into separate `*Tx` methods for better organization and testability in PostgreSQL storage implementations
Applied to files:
core/node/storage/pg_metadata_service_test.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*_test.go : Use standard imports in storage tests: context, testing, github.com/ethereum/go-ethereum/common, github.com/stretchr/testify/require, and relevant River protocol packages
Applied to files:
core/node/storage/pg_metadata_service_test.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*.go : Enforce miniblock requirement constraint: if a stream exists in the `es` (event streams) table, it MUST have at least one miniblock; empty streams are not allowed
Applied to files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/metadata_migrations/000001_create_initial_schema.up.sqlcore/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*_test.go : All storage tests should follow the consistent setup pattern using helper functions: `setupStreamStorageTest(t)`, `setupAppRegistryStorageTest(t)`, or `prepareNotificationsDB(ctx)` depending on storage type, with automatic cleanup via `t.Cleanup(params.closer)`
Applied to files:
core/node/storage/pg_metadata_service_test.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*_test.go : Use `require.New(t)` at the beginning of test functions and check error codes with `base.IsRiverErrorCode(err, code)` for specific error type verification
Applied to files:
core/node/storage/pg_metadata_service_test.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*.go : Use `lockStream(ctx, tx, streamId, true)` with `FOR UPDATE` for write operations and `lockStream(ctx, tx, streamId, false)` with `FOR SHARE` for read operations to ensure proper row-level locking
Applied to files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*.go : Always include stream ID in queries to leverage table partitioning (256 partitions by default)
Applied to files:
core/node/storage/pg_metadata_service_test.gocore/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:47:37.754Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/rpc/AGENTS.md:0-0
Timestamp: 2025-11-25T08:47:37.754Z
Learning: Applies to core/node/rpc/**/*_test.go : Enable River logs with `enableRiverLogs: true` in tester options when debugging test failures
Applied to files:
core/node/storage/pg_metadata_service_test.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Make changes to storage interfaces first, then implement PostgreSQL-specific logic, write comprehensive tests, benchmark critical paths, and update migrations if schema changes are needed
Applied to files:
core/node/storage/metadata_migrations/000001_create_initial_schema.up.sqlcore/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:47:01.034Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/app_registry/AGENTS.md:0-0
Timestamp: 2025-11-25T08:47:01.034Z
Learning: Applies to core/node/app_registry/**/app_registry/**/*.go : When adding storage methods, implement in three layers: AppRegistryStore interface, PostgresAppRegistryStore implementation, and CachedEncryptedMessageQueue delegation
Applied to files:
core/node/storage/pg_metadata_service.go
📚 Learning: 2025-12-17T02:40:57.825Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/AGENTS.md:0-0
Timestamp: 2025-12-17T02:40:57.825Z
Learning: Applies to core/node/storage/**/*.go : Use jackc/pgx/v5 for PostgreSQL connection pooling and database driver operations
Applied to files:
core/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*.go : Use `COPY FROM` for atomic batch miniblock writes instead of individual insert statements in loops
Applied to files:
core/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*.go : The `last_snapshot_miniblock_num` in the `es` table MUST always point to one of the miniblocks within the last continuous miniblock sequence, and the pointer only grows forward
Applied to files:
core/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*.go : Enforce minipool generation sequencing constraint: the minipool generation number MUST always be set to the last miniblock number + 1
Applied to files:
core/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*.go : Use the `txRunner` function for all database operations to provide automatic retry for serialization failures and deadlocks, exponential backoff with jitter, transaction tracking, debugging, and comprehensive metrics and logging
Applied to files:
core/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*.go : Keep transactions short-lived and use appropriate access modes (ReadOnly vs ReadWrite) in txRunner calls
Applied to files:
core/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*.go : Add transaction tags for debugging in txRunner calls to improve transaction tracking and monitoring
Applied to files:
core/node/storage/pg_metadata_service.go
📚 Learning: 2025-11-25T08:48:08.084Z
Learnt from: CR
Repo: towns-protocol/towns PR: 0
File: core/node/storage/AGENTS.md:0-0
Timestamp: 2025-11-25T08:48:08.084Z
Learning: Applies to core/node/storage/storage/**/*.go : Handle context cancellation properly; disable context cancellation for write operations in txRunner
Applied to files:
core/node/storage/pg_metadata_service.go
🧬 Code graph analysis (2)
core/node/storage/pg_metadata_service_test.go (8)
core/node/storage/pg_metadata_service.go (3)
PostgresMetadataServiceStore(145-149)PostgresMetadataServiceStore(861-869)NewPostgresMetadataServiceStore(157-168)core/config/config.go (1)
DatabaseConfig(259-287)core/node/base/test/context.go (1)
NewTestContext(15-23)core/node/testutils/dbtestutils/db.go (1)
ConfigureDbWithPrefix(60-70)core/node/storage/pg_storage.go (1)
CreateAndValidatePgxPool(243-254)core/node/shared/stream_id.go (2)
STREAM_CHANNEL_BIN(16-16)StreamId(41-41)core/node/protocol/protocol.pb.go (1)
Err_BAD_PREV_MINIBLOCK_HASH(457-457)core/node/base/error.go (1)
AsRiverError(273-355)
core/node/storage/pg_metadata_service.go (4)
core/node/shared/stream_id.go (2)
StreamId(41-41)StreamIdFromBytes(77-85)core/node/base/error.go (2)
RiverError(74-86)RiverErrorWithBase(88-94)core/node/protocol/protocol.pb.go (6)
Err_INVALID_ARGUMENT(407-407)Err_ALREADY_EXISTS(416-416)Err_FAILED_PRECONDITION(426-426)Err_BAD_PREV_MINIBLOCK_HASH(457-457)Err_BAD_BLOCK_NUMBER(473-473)Err_NOT_FOUND(413-413)core/node/base/backoff_tracker.go (1)
BackoffTracker(11-34)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (7)
- GitHub Check: Common_CI
- GitHub Check: Go_Tests
- GitHub Check: Multinode_Ent_Legacy
- GitHub Check: Multinode
- GitHub Check: Multinode_Ent
- GitHub Check: Cursor Bugbot
- GitHub Check: XChain_Integration
🔇 Additional comments (20)
core/node/storage/pg_metadata_service_test.go (7)
31-73: LGTM!Setup helper follows established patterns with proper cleanup registration and unique DB prefix.
75-84: LGTM!Clean test data generators.
86-211: LGTM!Comprehensive test covering listing, pagination, filtering by node, counting, and block retrieval with proper event mask validation.
213-333: LGTM!Good coverage of partial batch failures, error code mapping, and event mask differentiation between update types.
335-400: LGTM!Properly validates block trimming behavior with numBlocksToKeep=2.
402-467: LGTM!Confirms zero-trim parameter preserves all blocks.
469-480: LGTM!Validates error handling for invalid block range inputs.
core/node/storage/metadata_migrations/000001_create_initial_schema.up.sql (3)
1-60: LGTM!Well-structured schema with proper constraints. The denormalized placement table for GIN index efficiency is a sensible optimization.
62-77: LGTM!Clean NOTIFY trigger implementation. The DROP IF EXISTS ensures idempotent migrations.
79-79: LGTM!Initial block_num of -1 correctly indicates no updates have occurred.
core/node/storage/pg_metadata_service.go (10)
30-82: LGTM!Types are well-structured. The event mask bits and update union pattern are clean.
83-143: LGTM!Well-documented interface with clear contracts.
145-194: LGTM!Initialization follows the established pattern with embedded PostgresEventStore.
301-380: LGTM!Batch update with proper serialization via row lock, partial failure handling, and correct trimming logic.
382-415: LGTM!Clean union validation and dispatch.
417-509: LGTM!Thorough validation with proper duplicate handling via both pre-check and UniqueViolation.
511-598: LGTM!Placement update handles all field combinations correctly with proper constraint validation.
600-660: LGTM!Strict miniblock sequence validation with proper sealed state handling.
871-962: LGTM!Proper range validation and ordered block retrieval.
964-1138: LGTM!Helper functions are well-organized with proper locking differentiation and comprehensive validation.
1 participant
No description provided.