refactor!: collapse postgres/sqlite repositories into a shared generic layer - #111
Conversation
|
Warning Review limit reached
Next review available in: 19 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (30)
📝 WalkthroughWalkthroughThis PR consolidates Postgres and SQLite repositories onto a shared ChangesSQLx Backend Unification
Estimated code review effort: 5 (Critical) | ~120 minutes Outbox Store Paging
Estimated code review effort: 2 (Simple) | ~15 minutes Batched Snapshot Hydration
Estimated code review effort: 3 (Moderate) | ~25 minutes EventRecord payload_codec Cow Refactor
Estimated code review effort: 2 (Simple) | ~10 minutes Commit Batch Validation Centralization
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant Caller
participant SqlxRepository
participant Transaction
participant Database
Caller->>SqlxRepository: commit_batch(batch)
SqlxRepository->>Transaction: begin
Transaction->>Database: check stream versions
Database-->>Transaction: current versions
Transaction->>Database: insert events (chunked)
Transaction->>Database: insert outbox messages
Transaction->>Database: apply read-model plans
Transaction->>Database: upsert snapshots
Transaction->>Database: insert inbox receipts
Transaction-->>SqlxRepository: commit
SqlxRepository-->>Caller: committed batch
sequenceDiagram
participant AggregateRepository
participant SnapshotPolicy
participant SnapshotStore
AggregateRepository->>SnapshotPolicy: hydrate_all(entities)
SnapshotPolicy->>SnapshotStore: get_snapshots(identities)
SnapshotStore-->>SnapshotPolicy: Vec<SnapshotRecord>
SnapshotPolicy->>SnapshotPolicy: hydrate_with_optional_snapshot per entity
SnapshotPolicy-->>AggregateRepository: hydrated entities
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ 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 |
613c7d4 to
5660350
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@migrations/postgres/0001_initial.sql`:
- Around line 6-9: The widened Postgres column types were changed in the initial
migration, but existing databases will not pick that up because SQLx tracks
applied migrations in _sqlx_migrations. Add a new follow-up migration that ALTER
TABLEs the existing tables to change event_version, payload_codec_version,
snapshot_version, and attempts to bigint, and leave 0001_initial.sql unchanged
so fresh installs and already-deployed databases stay compatible.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 3016c1e6-4540-41ef-ab34-57859a4ba076
📒 Files selected for processing (32)
Cargo.tomlmigrations/postgres/0001_initial.sqlsrc/aggregate/repository.rssrc/entity/event_record.rssrc/hashmap_repo/repository.rssrc/microsvc/runtime.rssrc/outbox/commit.rssrc/outbox_worker/outbox_source.rssrc/outbox_worker/store.rssrc/postgres_repo/mod.rssrc/queued_repo/repository.rssrc/repository/mod.rssrc/repository/traits.rssrc/repository/validation.rssrc/snapshot/in_memory.rssrc/snapshot/repository.rssrc/sqlite_repo/mod.rssrc/sqlx_repo/mod.rssrc/sqlx_repo/read_model.rssrc/sqlx_repo/repo.rstests/bomberman/main.rstests/distributed_read_model/main.rstests/durable_enqueue_sqlite/main.rstests/microsvc/convention.rstests/persistent_repository_conformance/inbox.rstests/persistent_repository_conformance/outbox.rstests/postgres_repository/main.rstests/postgres_transport/main.rstests/sourced_snapshot/main.rstests/sqlite_repository/main.rstests/todos/main.rstests/transport_conformance/mod.rs
| event_version bigint NOT NULL DEFAULT 1, | ||
| payload bytea NOT NULL, | ||
| payload_codec text NOT NULL, | ||
| payload_codec_version integer NOT NULL, | ||
| payload_codec_version bigint NOT NULL, |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
sqlx postgres try_get i64 on INT4 column type mismatch error
💡 Result:
The error occurs because sqlx strictly maps the PostgreSQL INT4 (INTEGER) type to the Rust i32 type, while i64 is exclusively reserved for the BIGINT (INT8) type [1][2][3]. When you attempt to decode an INT4 column into an i64 using try_get, sqlx detects a type mismatch between the database's reported type and the requested Rust type, resulting in a decoding error [4][5]. To resolve this issue, use one of the following approaches: 1. Use i32 in Rust: If the data in your database column is guaranteed to fit within a 32-bit integer, change your Rust code to use i32 instead of i64 [2][3]. 2. Explicitly cast in SQL: If you must use i64 in Rust, you can cast the column value in your SQL query to bigint. This promotes the type at the database level before it reaches sqlx [5]. For example: SELECT your_column::bigint FROM your_table; 3. Change the database schema: If the values in the column may exceed the range of an i32, consider altering your PostgreSQL table schema to use BIGINT instead of INT4/INTEGER [2][3].
Citations:
- 1: https://docs.rs/sqlx/latest/sqlx/postgres/types/index.html
- 2: https://docs.rs/sqlx-postgres/latest/sqlx_postgres/types/index.html
- 3: https://github.com/launchbadge/sqlx/blob/1dd526a2/sqlx-postgres/src/types/mod.rs
- 4:
Row.try_get()should check the decoding type against the type reported by the server transact-rs/sqlx#8 - 5: https://wiki.charleschen.ai/Review/Research/postgresql-rust-integer-type-mismatch-resolution
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf 'Repo root: '; pwd
# Find relevant migration and decoding code.
git ls-files 'migrations/postgres/*' 'src/*' 'crates/*' | sed -n '1,200p'
printf '\n--- search for widened columns and sqlx decoders ---\n'
rg -n "event_version|payload_codec_version|snapshot_version|attempts|try_get::<i64>|repository_u64_from_i64|repository_u16_from_i64|u32::try_from|Migrator|sqlx::migrate|CREATE TABLE IF NOT EXISTS" . --glob '!target' --glob '!node_modules' | sed -n '1,240p'Repository: hops-ops/distributed
Length of output: 22782
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the initial migration and the migration runner.
for f in migrations/postgres/0001_initial.sql $(git ls-files | rg 'migrate|migration|sqlx'); do
[ -f "$f" ] || continue
printf '\n### %s (lines, if small enough)\n' "$f"
wc -l "$f"
done
# If migration runner exists, show structure.
cand=$(rg -l "Migrator|migrate!|sqlx::migrate" . --glob '!target' --glob '!node_modules' | head -n 20)
printf '\nCandidates:\n%s\n' "$cand"Repository: hops-ops/distributed
Length of output: 895
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Read the relevant SQL migration in a bounded way.
nl -ba migrations/postgres/0001_initial.sql | sed -n '1,180p'Repository: hops-ops/distributed
Length of output: 198
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Probe for the migration implementation and runtime decoding paths.
rg -n "try_get::<i64>|repository_u64_from_i64|repository_u16_from_i64|u32::try_from|CREATE TABLE IF NOT EXISTS|ALTER TABLE .*ALTER COLUMN .*TYPE bigint|sqlx::migrate|Migrator" . --glob '!target' --glob '!node_modules' | sed -n '1,260p'Repository: hops-ops/distributed
Length of output: 4525
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Show the initial migration and the migration runner file paths only.
printf 'Initial migration:\n'
nl -ba migrations/postgres/0001_initial.sql | sed -n '1,120p'
printf '\nMigration-related files:\n'
rg -l "sqlx::migrate|Migrator|migrate!" . --glob '!target' --glob '!node_modules' | sed -n '1,50p'Repository: hops-ops/distributed
Length of output: 217
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf 'Postgres migration files:\n'
git ls-files 'migrations/postgres/*' | sort
printf '\n--- migrations/postgres/0001_initial.sql (selected lines) ---\n'
sed -n '1,120p' migrations/postgres/0001_initial.sql
printf '\n--- src/postgres_repo/mod.rs ---\n'
sed -n '1,140p' src/postgres_repo/mod.rs
printf '\n--- src/sqlx_repo/repo.rs (migrator-related) ---\n'
sed -n '1,120p' src/sqlx_repo/repo.rsRepository: hops-ops/distributed
Length of output: 15165
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf 'Postgres migration files:\n'
git ls-files 'migrations/postgres/*' | sort
printf '\n--- migrations/postgres/0001_initial.sql (selected lines) ---\n'
sed -n '1,120p' migrations/postgres/0001_initial.sql
printf '\n--- src/postgres_repo/mod.rs ---\n'
sed -n '1,140p' src/postgres_repo/mod.rs
printf '\n--- src/sqlx_repo/repo.rs (migrator-related) ---\n'
sed -n '1,140p' src/sqlx_repo/repo.rsRepository: hops-ops/distributed
Length of output: 16403
Add a follow-up migration for the widened columns
Changing 0001_initial.sql only affects fresh installs. SQLx records applied migrations in _sqlx_migrations, so already-deployed Postgres databases will keep the old integer columns and fail to decode these fields as i64. Add a new ALTER TABLE ... ALTER COLUMN ... TYPE bigint migration for event_version, payload_codec_version, snapshot_version, and attempts instead of editing 0001.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@migrations/postgres/0001_initial.sql` around lines 6 - 9, The widened
Postgres column types were changed in the initial migration, but existing
databases will not pick that up because SQLx tracks applied migrations in
_sqlx_migrations. Add a new follow-up migration that ALTER TABLEs the existing
tables to change event_version, payload_codec_version, snapshot_version, and
attempts to bigint, and leave 0001_initial.sql unchanged so fresh installs and
already-deployed databases stay compatible.
…ayer Extends the existing SqlxReadModelBackend / lock/sqlx_common dialect-trait pattern to the event-store/snapshot/outbox/inbox layers. postgres_repo and sqlite_repo shrink from ~1700 lines each to ~460-line dialect shims over a shared SqlxRepository/SqlxOutboxStore in src/sqlx_repo/repo.rs. Outbox `claim` stays per-backend (postgres SKIP LOCKED CTE vs sqlite scan-loop). Squashed from 14 commits for a single rebase reconciliation onto main (after - fix: surface malformed sqlite timestamps as errors (was silent UNIX_EPOCH); align null-bind error message with postgres (includes field_name). - refactor: share commit-batch validation across all backends (fixes hashmap vs SQL snapshot-identity drift). - feat!: widen postgres integer columns to BIGINT so both backends decode i64; deletes the width-conversion helpers. - perf: batch snapshot loads for get_all hydration (fixes N+1); get_streams is a GetStream default method. - perf: batch the commit_batch concurrency pre-check into one query. - perf!: borrow events in PreparedEventAppend instead of cloning. - feat!: bound outbox status listings (messages_by_status/pending) with a limit. - perf!: store EventRecord.payload_codec as Cow<'static, str>. - feat: run migrations through sqlx's Migrator with a _sqlx_migrations ledger. - fix: chunk postgres event/outbox inserts under the 65535 bind-param cap. - fix: classify postgres 57P01/57P02/57P03 and class-08 SQLSTATEs as transient. Merge-reconciliation notes: - table/ is now the canonical vocabulary (#109); backend shims import the renamed types via local aliases (TableColumn as ColumnDef, TableStoreError as ReadModelError) to keep the collapsed bodies unchanged. - #106's batched OutboxStore::complete_many overrides lived in the old backend files and were collapsed away; SqlxOutboxStore inherits the serial default (correct, conformance-tested). Re-adding a batched override to the shared layer is a follow-up (see tasks/outbox-sqlx-batched-complete-many). REBASED onto main+#110 (9f34f0a): #110 already merged, so its fault-injection test that pinned the OLD 57P01 (mis)classification is flipped here to `assert!(err.is_retryable())` to match this PR's classification fix; #110's shared outbox test helpers are updated for the bounded messages_by_status(limit) API. Main stays green on merge. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DzYSVLas93c7LbgHJWsW7n
5660350 to
3d7fea3
Compare
Summary
Collapses the duplicated postgres/sqlite event-store repositories (1705 + 1644 lines, 876 identical) into a shared generic layer, extending the proven
SqlxReadModelBackendpattern (lock/sqlx_common.rs,sqlx_repo/read_model.rs) to the whole repository surface, plus the P0 drift fixes, API/perf riders, and a real migration ledger.Backend files after:
src/postgres_repo/mod.rs461 lines,src/sqlite_repo/mod.rs463 lines (trait impl + claim + timestamp codec + the pre-existing read-model backend impl). Shared layer:src/sqlx_repo/repo.rs(1820 lines, one copy of everything).What moved where
validate_commit_batchinsrc/repository/validation.rs— now also validates snapshot identity on all backends (fixes hashmap/SQL drift)event_from_row/snapshot_from_row/outbox_message_from_rowsrc/sqlx_repo/repo.rscomplete/release/fail(6 copies) +ensure_outbox_update_applied+ the 18-column SELECT repeated 4xtransition_claimed_outbox_message+OUTBOX_SELECTtrait constinsert_events_in_tx/insert_outbox_messages_in_tx(+EventRow/OutboxRowstructs declared twice)MAX_BIND_PARAMS; postgres now gets its real 65535-param cap — the old unchunked insert failed outright above ~6500 events)stream_version_in_txvsstream_version_pool(intra-postgres dup)stream_versiongeneric overExecutorconcurrent_write_from_conflictover&mut DB::Connection;CONFLICT_REREAD_IN_TXselects in-tx (sqlite, tx survives) vs fresh-connection (postgres, tx aborted)get_stream/get_streams/get_stream_tailincl. the byte-identical slicing loopEVENT_SELECT+push_id_filter(= ANY($n)vsIN (...)) per backendmigrate/bootstrap_table_schema_for_dev(4 copies of the table-schema trio)SqlxRepository<DB>/SqlxOutboxStore<DB>Kept per-backend (genuine divergence): outbox
claim(postgres CTE +FOR UPDATE SKIP LOCKEDvs sqlite candidate-scan loop), timestamp codec, unique-violation predicate, schema SQL, pool sizing.Public API shape unchanged:
PostgresRepository,PostgresOutboxStore,SqliteRepository,SqliteOutboxStoreare now type aliases of the generic structs.Per-step summary (all plan items landed)
system_time_from_storageno longer silently turns corrupt rows into 1970-dated events (returnsRepositoryError::Model, matching postgres); sqlitepush_null_bindmessage includes the field name.validate_commit_batch(see table).feat!) — postgresevent_version/payload_codec_version/snapshot_version/attemptswidened toBIGINTin the initial migration (edited in place, pre-release); both backends bind/decodei64; the postgres-only width-conversion helpers are deleted.get_streamsis aGetStreamdefault method (hashmap loop + snapshot test stub deleted).SnapshotStore::get_snapshots(grouped single query on SQL backends, single-lock in-memory impls,QueuedRepositoryforward — one-line delegation outside the listed scope, flagged here deliberately);get_allhydration reads all snapshot records in one round trip via a newhydrate_allpolicy hook (fixes the N+1).MAX(sequence)query per batch (stream_versions_in_tx) instead of one per stream; semantics unchanged (kept for zero-event appends too).PreparedEventAppend<'a>borrowsevents: &'a [EventRecord](breaking: gains a lifetime) — no more per-commit event clone.messages_by_status/pendingtake a mandatorylimit(breaking); SQL backends push a boundLIMIT, clamped toi64::MAX.EventRecord.payload_codecisCow<'static, str>(breaking); constructors borrow the codec constant and the row mapper compares before allocating.sqlx::migrate::Migrator(sqlx/migratefeature):_sqlx_migrationsledger with checksums, whole-file execution (the old runner split on;, which breaks on function bodies/string literals). The embedded migrator is built with the publicMigrator::with_migrationsfrominclude_str!files rather than themigrate!macro, so consumers don't pay for the sqlx proc-macro stack.Rider from the test-coverage agent:
is_sqlx_transientnow classifies postgres57P01/57P02/57P03and class-08SQLSTATEs as transient (killed backend / failover / connection loss), with unit tests. The fault-injection test onreview/tests-dedup-coveragepins the old (wrong) behavior with a "flip this assertion" note — flip it when both branches merge.Review pass
An 8-angle review (line-by-line, removed-behavior, cross-file, reuse, simplification, efficiency, altitude, conventions) was run over the final diff; accepted findings are folded in (unified conflict recovery, by-reference timestamp binds removing per-row sqlite allocations,
Migrator::with_migrationsinstead of semver-exempt struct fields, sharedids_by_type, single epoch-seconds conversion). Notes for reviewers:commit_batchchanged: snapshot-identity validation now runs in the shared preamble (before the version pre-check) on all backends — intentional per the drift fix.CREATE TABLE IF NOT EXISTSwill not ALTER an old-width schema, and the ledger will record v1 as applied). Called out in thefeat!commit.repo.rsfollow the existingread_model.rsconvention; trait-level where-clauses are not implied bounds in Rust (verified empirically), so they cannot be hoisted ontoSqlxRepoBackend.complete_manytoOutboxStore; merge conflicts to be resolved on whichever lands second.Tests
cargo fmtclean,cargo clippy --workspace --all-features --all-targetsclean (two pre-existing warnings untouched), and:🤖 Generated with Claude Code
https://claude.ai/code/session_01DzYSVLas93c7LbgHJWsW7n
Summary by CodeRabbit
New Features
Bug Fixes