feat: add orphan stream cleanup to remove DB streams not in registry#4850
feat: add orphan stream cleanup to remove DB streams not in registry#4850bas-vk wants to merge 1 commit into
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
Adds CleanupOrphanedStreams function that identifies and deletes streams from the database that are no longer registered on this node in the stream registry. Called during StreamCache.Start after loading registry streams. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
bas-vk
force-pushed
the
basvk/compare_streamd_db_stream_registry
branch
from
0be7e63 to
5a0ef52
Compare
📝 WalkthroughWalkthroughThis PR introduces orphaned stream cleanup functionality. A new Changes
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Suggested reviewers
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
🧪 Generate unit tests (beta)
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: 1
🤖 Fix all issues with AI agents
In @core/node/storage/orphan_stream_cleaner.go:
- Around line 17-67: CleanupOrphanedStreams performs multiple store.DeleteStream
calls without a DB transaction, risking partial deletes on context cancel;
change CleanupOrphanedStreams to start a PostgreSQL transaction (use the store's
Begin/WithTx/RunInTx pattern), perform all DeleteStream calls within that
transaction, and Commit only if all deletes succeed (Rollback on error or ctx
cancellation), return an error from CleanupOrphanedStreams instead of void, and
update callers (e.g., stream_cache.go) to check the returned error and log it
(e.g., "Orphan stream cleanup failed", "error", err).
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (4)
core/node/events/stream_cache.gocore/node/events/stream_ephemeral.gocore/node/storage/orphan_stream_cleaner.gocore/node/storage/orphan_stream_cleaner_test.go
🧰 Additional context used
📓 Path-based instructions (7)
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/events/stream_cache.gocore/node/storage/orphan_stream_cleaner_test.gocore/node/storage/orphan_stream_cleaner.gocore/node/events/stream_ephemeral.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/events/stream_cache.gocore/node/storage/orphan_stream_cleaner_test.gocore/node/storage/orphan_stream_cleaner.gocore/node/events/stream_ephemeral.go
**/core/node/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
Use structured logging with zap in Go backend River nodes for observability
Files:
core/node/events/stream_cache.gocore/node/storage/orphan_stream_cleaner_test.gocore/node/storage/orphan_stream_cleaner.gocore/node/events/stream_ephemeral.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/orphan_stream_cleaner_test.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/orphan_stream_cleaner_test.gocore/node/storage/orphan_stream_cleaner.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/orphan_stream_cleaner_test.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/orphan_stream_cleaner_test.go
🧠 Learnings (18)
📚 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/events/stream_cache.gocore/node/storage/orphan_stream_cleaner.gocore/node/events/stream_ephemeral.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/events/stream_cache.gocore/node/storage/orphan_stream_cleaner_test.gocore/node/storage/orphan_stream_cleaner.gocore/node/events/stream_ephemeral.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/events/stream_cache.gocore/node/storage/orphan_stream_cleaner.gocore/node/events/stream_ephemeral.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/events/stream_cache.gocore/node/storage/orphan_stream_cleaner.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/events/stream_cache.gocore/node/storage/orphan_stream_cleaner.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/events/stream_cache.gocore/node/storage/orphan_stream_cleaner_test.gocore/node/storage/orphan_stream_cleaner.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/events/stream_cache.gocore/node/storage/orphan_stream_cleaner.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/events/stream_cache.gocore/node/storage/orphan_stream_cleaner.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/orphan_stream_cleaner_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/orphan_stream_cleaner_test.gocore/node/storage/orphan_stream_cleaner.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/orphan_stream_cleaner_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 : Implement table-driven tests when testing multiple scenarios to improve maintainability and coverage
Applied to files:
core/node/storage/orphan_stream_cleaner_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/orphan_stream_cleaner_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/orphan_stream_cleaner_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 : Use `t.Cleanup()` or defer with tester cleanup functions to handle automatic cleanup and resource management
Applied to files:
core/node/storage/orphan_stream_cleaner_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/orphan_stream_cleaner_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 `ConfigureDB(ctx)` with default prefix 'tst', or `ConfigureDbWithPrefix(ctx, 'b_')` for app registry tests in the dbtestutils package
Applied to files:
core/node/storage/orphan_stream_cleaner_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 : Add transaction tags for debugging in txRunner calls to improve transaction tracking and monitoring
Applied to files:
core/node/storage/orphan_stream_cleaner.go
🧬 Code graph analysis (3)
core/node/events/stream_cache.go (2)
core/node/storage/pg_stream_store.go (1)
PostgresStreamStore(41-67)core/node/storage/orphan_stream_cleaner.go (1)
CleanupOrphanedStreams(17-67)
core/node/storage/orphan_stream_cleaner_test.go (2)
core/node/shared/stream_id.go (2)
StreamId(41-41)STREAM_CHANNEL_BIN(16-16)core/node/storage/orphan_stream_cleaner.go (1)
CleanupOrphanedStreams(17-67)
core/node/storage/orphan_stream_cleaner.go (2)
core/node/storage/pg_stream_store.go (1)
PostgresStreamStore(41-67)core/node/shared/stream_id.go (1)
StreamId(41-41)
🔇 Additional comments (5)
core/node/events/stream_ephemeral.go (1)
64-67: LGTM! Clear documentation of deletion strategy.The comment accurately explains the design decision to defer DB cleanup to boot time rather than the event loop.
core/node/events/stream_cache.go (1)
191-194: Handle cleanup errors once function signature is updated.After
CleanupOrphanedStreamsreturns an error (per the critical issue flagged in orphan_stream_cleaner.go), add error handling here to log failures.core/node/storage/orphan_stream_cleaner_test.go (2)
13-19: LGTM! Clean helper for registry test data.The helper simplifies test setup and is reused across all test cases.
21-140: LGTM! Comprehensive test coverage.The test suite covers all critical scenarios: partial orphans, no orphans, empty database, all orphaned, and both empty. Tests follow the established pattern with
setupStreamStorageTestand proper assertions.core/node/storage/orphan_stream_cleaner.go (1)
42-50: DeleteStream properly uses row-level locking for safe concurrent operations.The implementation acquires an exclusive lock with
FOR UPDATEon theestable vialockStream(ctx, tx, streamId, true)before deletion, ensuring concurrent modifications are safely serialized. The function's behavior with concurrent operations (including newly added or concurrently updated streams) is protected by this transaction-level locking mechanism.Note: The TOCTOU window between collecting DB streams and deleting orphaned streams is acknowledged in the function's documentation and is acceptable per design—placement updates emitted after cleanup returns can recreate streams if needed.
| // is collected and this function returns. | ||
| func CleanupOrphanedStreams( | ||
| ctx context.Context, | ||
| store *PostgresStreamStore, |
There was a problem hiding this comment.
Idiomatically it seems it should be a method on PostgresStreamStore and interface - then there is no need to cast in steam_cache.go
| ) | ||
| failed++ | ||
| } else { | ||
| log.Debugw("Deleted orphaned stream", "streamId", dbStreamId) |
| s.appliedBlockNum.Store(uint64(s.params.AppliedBlockNum)) | ||
|
|
||
| // delete streams from the DB that are no longer placed on this node | ||
| if store, ok := s.params.Storage.(*storage.PostgresStreamStore); ok { |
There was a problem hiding this comment.
lets add a config setting to disable it - if node takes too long to start, operator can disable this functionality and start the node without it.
| ) { | ||
| start := time.Now() | ||
| log := logging.FromCtx(ctx) | ||
|
|
There was a problem hiding this comment.
let log here and mention env var for disable setting.
2 participants
Summary
CleanupOrphanedStreamsfunction instoragepackage that identifies and deletes streams from the database that are no longer registered on this nodeStreamCache.Startafter loading registry streamsTest plan
go test ./node/storage/... -run TestCleanupOrphanedStreams)🤖 Generated with Claude Code