(MOT-4280) feat(database): native row-change capture for postgres, sqlite, and mysql - #640
Merged
Conversation
Opt-in per-database mode: worker-installed triggers + LISTEN/NOTIFY on a dedicated connection make database::row-changed fire for committed writes from any client, not just this worker. Table-scoped bindings only; sqlite/mysql rejected at config load; self-writes leave the SQL classification path so nothing fires twice.
New pg_native_db handle (same postgres, capture: native) and a case group that proves the cross-client claim end-to-end: a write entering through the pg_db pool is heard by the native subscriber via trigger+NOTIFY, both capture paths attribute correctly, own writes fire exactly once, zero-row writes stay silent, and table-less bindings are rejected (sentinel binding proves the silence is not vacuous). Schema fixture regenerated for the new capture field.
Extends capture: native to sqlite. No notification bus exists in an embedded database, so capture is triggers appending to a _iii_row_changes changelog (commit-gated by construction — rollbacks take their rows with them), drained by a per-database watcher thread on a dedicated connection: inotify wake-up via the notify crate, PRAGMA data_version as the other-connection-committed gate, a 2s fallback tick as missed-event insurance, and run-length coalescing so a bulk UPDATE is one event. Cursor starts at the changelog head (at-most-once, postgres parity); :memory: is rejected at config load. E2e: sqlite_native_db handle on the same file as sqlite_db, native cases parameterized over both drivers.
capture: native on a mysql database connects as a replica (mysql_async's binlog API, feature-gated in the dep) from the current position and decodes row events into database::row-changed. Nothing is installed in the schema; commit gating is the binlog's own. Chunked row events merge until the next non-rows event, so one statement stays one event. Registration verifies log_bin=ON, binlog_format=ROW, and position visibility, failing with the exact GRANT the user is missing. Reconnect re-snapshots the position (at-most-once, parity with pg/sqlite). E2e: mysql_native_db handle, warmup-write readiness (no triggers to probe), replication grants baked into the compose mysql init.
Registering without metadata left the console's workers view showing the SDK fallback hostname:pid — the worker was listed but unfindable. Set name (III_WORKER_NAME still wins, preserving the managed-spawn identity contract and letting workers-dev tag instances per worktree) and a description for engine::workers::list.
Replaces the generic schema-driven editor for the database entry on the Workers tab (host.configForms): one card per database with a live driver badge, connection url, capture mode with driver-aware guidance (grants hint for mysql, :memory: warning for sqlite), TLS controls hidden where they are meaningless, pool knobs behind a disclosure, add/remove/rename of handles (rename commits on blur so mid-typing collisions cannot swallow a sibling entry). Save/validation/dirty tracking stay host-owned; values round-trip minimal (defaults are omitted, unknown keys preserved).
database::testConnection takes a candidate url + tls straight from the request (not a configured handle), opens one throwaway connection outside every pool, and reports ok/driver/latency/server-version as data — failures are a response, not an error. Credentials are scrubbed from any echoed driver error; a missing sqlite file is reported as news (created on save), never created by the probe. The form gets a per-card button via host.iii; results are dropped on any edit of that handle so a stale 'connected' can't vouch for a changed url.
Three new cases per driver (nine runs): interactive-transaction commit gating (rollback is absolute silence, commit delivers in order — on a native handle this exercises the database's own gating, since the worker's staging path is off); multi-subscriber fan-out with an ops filter, registered second so the DDL reinstall happens under a live binding, plus writes-succeed-and-nobody-hears after unregister; and bulk coalescing — 100-row insert/update/delete arrive as exactly three events with true counts, proving all three per-driver coalescers (pg statement triggers, sqlite run-length, mysql chunk merge).
…nstall-hint misdirection From the manual trigger test report: a pg native binding spelled in the wrong case fails DDL install (quoted identifiers are case-sensitive) but the error came dressed in privilege advice — the hint is now matched to the failure (undefined table → casing guidance, permission → grants). New e2e: statements-path filters match case-insensitively (×3); native casing behaves per driver contract (sqlite/mysql capture, pg rejects, sentinel proves flow); two-table binding isolation (×3); registration against a missing table (pg/sqlite reject terminally — silent even after the table appears — mysql accepts and captures once it exists). Also: the tx-gating case now asserts the committed event SET, not arrival order (Void dispatch carries no cross-event ordering guarantee), and the sqlite trigger probe compares tbl_name case-insensitively (a reinstall from a differently-cased binding stores that spelling).
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
Contributor
skill-check — worker0 verified, 49 skipped (no docs/).
Four for four. Nicely done. |
|
Warning Review limit reached
Next review available in: 37 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: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (14)
📝 WalkthroughWalkthroughAdds native row-change capture for Postgres, SQLite, and MySQL, with configurable capture modes, listener synchronization, connection probing, a database configuration UI, and comprehensive unit, integration, and end-to-end tests. ChangesNative capture configuration
Connection probing and configuration UI
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant ConfigUI
participant TestConnection
participant DatabaseDriver
participant RowChangeBus
ConfigUI->>TestConnection: database::testConnection request
TestConnection->>DatabaseDriver: probe URL with bounded timeout
DatabaseDriver-->>TestConnection: server version or scrubbed error
TestConnection-->>ConfigUI: structured response
DatabaseDriver->>RowChangeBus: native row-change event
Possibly related PRs
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches 💡 1🛠️ Fix failing CI checks 💡
📝 Generate docstrings
🧪 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 |
andersonleal
changed the title
rustfmt over the branch's new files; reword a doc line that wrapped onto a markdown list bullet (doc_lazy_continuation) and alias the sqlite drain tuple (type_complexity).
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 12
🧹 Nitpick comments (4)
database/src/triggers/native.rs (2)
309-317: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse
CHANNELin the LISTEN statement.The channel name is hardcoded here while the DDL (Line 48) and the log line (Line 317) use
CHANNEL. A future rename of the const would silently detach the listener from the notifications.♻️ Proposed refactor
- let listen = client.batch_execute("LISTEN iii_row_changed"); + let listen = client.batch_execute(&format!("LISTEN {CHANNEL}"));🤖 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 `@database/src/triggers/native.rs` around lines 309 - 317, Update the LISTEN statement in the native capture setup around the `listen` future to build the SQL using the `CHANNEL` constant instead of the hardcoded channel name, keeping it consistent with the DDL and `tracing::info!` usage.
261-276: 🩺 Stability & Availability | 🔵 TrivialDocument or expose the LISTEN connection liveness contract.
listen_oncesits inpoll_messageuntil the connection reports EOF/error; if the listener relies only ontokio-postgresdefaults, inactivity can survive for up to the default keepalive boundary before reconnecting. Since native capture is at-most-once, either document that behavior explicitly or shorten the recovery window with a shorter keepalive/periodic health check on this dedicated listener.🤖 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 `@database/src/triggers/native.rs` around lines 261 - 276, Document the liveness and at-most-once recovery contract for run_listener/listen_once, including that inactivity may delay EOF detection under tokio-postgres defaults, or configure this dedicated LISTEN connection with a shorter keepalive or periodic health check. Prefer the smallest targeted change that makes the reconnect window explicit and bounded.database/src/triggers/sqlite_watch.rs (1)
106-128: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winUnbounded drain materializes the whole backlog.
draincollects every changelog row past the cursor intorawbefore coalescing. A single bulk load (or a burst while the watcher was blocked) makes that allocation proportional to the row count — millions of rows means millions ofStrings on a watcher thread. ALIMITper pass, looped until fewer than the limit come back, keeps the footprint flat without changing the event stream.🤖 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 `@database/src/triggers/sqlite_watch.rs` around lines 106 - 128, Update drain to fetch changelog rows in bounded batches using a fixed LIMIT, repeatedly advancing the cursor and coalescing each batch until a batch returns fewer than the limit. Preserve ordering, unknown-operation skipping, missing-table handling, and the existing coalesced event stream while avoiding accumulation of all raw rows in memory.database/ui/src/configuration/index.tsx (1)
352-370: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winPool number fields accept non-integer values.
Number(e.target.value)accepts fractional input (e.g.10.5) formax/idle_timeout_ms/acquire_timeout_ms, which areu32/u64server-side — a fractional value will only be caught later as a backend deserialization error on save, instead of being prevented client-side.♻️ Proposed integer clamp
onChange={(e) => setBlock('pool', (block) => { if (e.target.value.trim() === '') delete block[f.key] - else if (!Number.isNaN(Number(e.target.value))) block[f.key] = Number(e.target.value) + else { + const n = Number(e.target.value) + if (Number.isInteger(n) && n >= 0) block[f.key] = n + } }) }🤖 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 `@database/ui/src/configuration/index.tsx` around lines 352 - 370, Update the pool field change handler in the POOL_FIELDS mapping to accept only positive integer values, rejecting fractional input before assigning Number(e.target.value) to block[f.key]. Preserve the existing behavior that removes the field for blank input and keeps invalid values from being stored.
🤖 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 `@database/src/configuration.rs`:
- Around line 91-93: Update the listener replacement flow around
listeners.sync(&cfg) so native listeners removed or replaced by a configuration
reload remain alive while transactions pinned to the old database are in flight.
Retire affected listeners only after those transactions finish, or reject/retry
the reload when active transactions prevent safe replacement, ensuring committed
row-change events are not lost.
In `@database/src/triggers/handler.rs`:
- Around line 181-189: Update the comment in the native capture error branch to
reflect that finalize() supports MySQL as well as PostgreSQL and SQLite;
describe only the missing-pool/configuration drift invariant, without claiming
native databases are restricted to those two engines.
In `@database/src/triggers/mysql_binlog.rs`:
- Around line 142-146: Update the binlog stream setup around schema
initialization and its event filter to reject configurations where
opts.db_name() is None before connecting or consuming events, preventing
server-wide capture; preserve schema-scoped filtering for URLs with a database
name.
- Around line 42-48: Update server_id() to incorporate the database handle name
alongside the process ID, producing distinct stable IDs for native MySQL handles
on the same worker. Change the BinlogStreamRequest::new(...) call site to pass
db_name into server_id(), while preserving the existing range and u32 behavior.
- Around line 157-215: Update the binlog row aggregation loop around pending and
EventData::TableMapEvent so same-table, same-operation rows are flushed at
statement/query boundaries rather than merged across consecutive statements.
Preserve TableMapEvent handling needed to decode the following row event, and
align the native-capture E2E expectations with the resulting statement-level
events if required.
In `@database/src/triggers/native.rs`:
- Around line 129-147: Update TaskHandle::Thread and its sqlite watcher
lifecycle to retain both the stop signal and the thread JoinHandle, plus a wake
sender that causes the watcher’s recv_timeout loop to exit immediately. Make
stop signal the flag and wake the thread, then have the sync replacement path
join the old watcher before spawning its replacement so sqlite watchers never
overlap.
- Around line 229-239: Handle the `std::thread::Builder::spawn` error in the
native capture setup instead of discarding it with `.ok()?`. Log a warning
containing the spawn failure details, matching the warning behavior of the
sibling `sqlite_file_path` branch, while preserving the existing early-return
behavior after logging.
In `@database/src/triggers/sqlite_watch.rs`:
- Around line 54-78: Make trigger installation atomic in install_sql and its
caller install_native_triggers by ensuring the complete changelog-table creation
and trigger drop/recreate batch executes within one SQLite transaction, rather
than relying on execute_batch’s per-statement implicit transactions. Preserve
the existing trigger definitions and reinstall behavior while preventing writes
from occurring between trigger removal and recreation.
In `@database/tests/e2e/workers/harness/src/cases-native-capture.ts`:
- Around line 1-15: Update the module header documentation for the native
change-capture cases to include MySQL and its binlog capture mechanism alongside
PostgreSQL and file-backed SQLite. Also mention the mysql_native_db ↔ mysql_db
sibling relationship so the documented scope matches TARGETS.
- Around line 752-778: Add upper.expectDrained() and exact.expectDrained() after
their respective event assertions in the capture case so duplicate deliveries
fail on SQLite. Also update sqlite_watch::install_sql to normalize table
spelling consistently for trigger names and the logged tbl value, while
preserving the existing subscriber matching behavior.
In `@database/tests/e2e/workers/harness/src/cases-row-changed.ts`:
- Around line 42-56: The row-changed case must wait for trigger propagation
before issuing the INSERT. After registerTrigger in the surrounding test flow,
poll engine::registered-triggers::list until the newly registered triggerRef is
present, following the sibling native-case pattern, then execute the existing
INSERT and event assertions.
In `@database/ui/src/configuration/index.tsx`:
- Around line 89-110: Guard async completion in runTest with a per-handle
generation token that is invalidated by clearTest and by handle edits, renames,
or removals. Capture the token before awaiting props.host.iii.trigger, and only
apply the final setTestResults update when the token still matches the current
generation for that name; otherwise discard the stale result.
---
Nitpick comments:
In `@database/src/triggers/native.rs`:
- Around line 309-317: Update the LISTEN statement in the native capture setup
around the `listen` future to build the SQL using the `CHANNEL` constant instead
of the hardcoded channel name, keeping it consistent with the DDL and
`tracing::info!` usage.
- Around line 261-276: Document the liveness and at-most-once recovery contract
for run_listener/listen_once, including that inactivity may delay EOF detection
under tokio-postgres defaults, or configure this dedicated LISTEN connection
with a shorter keepalive or periodic health check. Prefer the smallest targeted
change that makes the reconnect window explicit and bounded.
In `@database/src/triggers/sqlite_watch.rs`:
- Around line 106-128: Update drain to fetch changelog rows in bounded batches
using a fixed LIMIT, repeatedly advancing the cursor and coalescing each batch
until a batch returns fewer than the limit. Preserve ordering, unknown-operation
skipping, missing-table handling, and the existing coalesced event stream while
avoiding accumulation of all raw rows in memory.
In `@database/ui/src/configuration/index.tsx`:
- Around line 352-370: Update the pool field change handler in the POOL_FIELDS
mapping to accept only positive integer values, rejecting fractional input
before assigning Number(e.target.value) to block[f.key]. Preserve the existing
behavior that removes the field for blank input and keeps invalid values from
being stored.
🪄 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: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 5a1bd630-5db8-4f8d-be57-97f8993f20cc
⛔ Files ignored due to path filters (1)
database/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (28)
database/Cargo.tomldatabase/config.yaml.exampledatabase/src/config.rsdatabase/src/configuration.rsdatabase/src/handlers/mod.rsdatabase/src/handlers/rollback_transaction.rsdatabase/src/handlers/test_connection.rsdatabase/src/handlers/transaction_execute.rsdatabase/src/main.rsdatabase/src/triggers/bus.rsdatabase/src/triggers/handler.rsdatabase/src/triggers/mod.rsdatabase/src/triggers/mysql_binlog.rsdatabase/src/triggers/native.rsdatabase/src/triggers/sqlite_watch.rsdatabase/src/ui.rsdatabase/tests/e2e/README.mddatabase/tests/e2e/docker-compose.ymldatabase/tests/e2e/mysql-init/grant-replication.sqldatabase/tests/e2e/workers/harness/fixtures/database.schema.jsondatabase/tests/e2e/workers/harness/src/cases-native-capture.tsdatabase/tests/e2e/workers/harness/src/cases-row-changed.tsdatabase/tests/e2e/workers/harness/src/database-config.tsdatabase/tests/e2e/workers/harness/src/runner.tsdatabase/tests/integration.rsdatabase/ui/page.tsxdatabase/ui/src/configuration/index.tsxdatabase/ui/styles.css
Comment on lines
+142
to
+146
| // Events for OTHER databases on the same server are not this handle's | ||
| // business — a binding names a db handle, and the handle names a schema. | ||
| let schema = opts.db_name().map(str::to_string); | ||
| let mut conn = Conn::new(opts).await.map_err(|e| e.to_string())?; | ||
| let (file, pos) = binlog_position(&mut conn).await?; |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟡 Minor | ⚡ Quick win
A URL without a database name captures the whole server.
schema is None when the MySQL URL carries no default database, and the filter at Line 193 then admits row events from every schema on the instance — subscribers bound to db: <handle> receive table names and change volumes from unrelated databases. Prefer failing the stream (or at least warning loudly once) when opts.db_name() is None.
Also applies to: 193-196
🤖 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 `@database/src/triggers/mysql_binlog.rs` around lines 142 - 146, Update the
binlog stream setup around schema initialization and its event filter to reject
configurations where opts.db_name() is None before connecting or consuming
events, preventing server-wide capture; preserve schema-scoped filtering for
URLs with a database name.
- mysql: server_id now hashes the handle name — two native handles on one server registered as the same replica and evicted each other into a reconnect-thrash loop. TableMapEvent is now a flush boundary, so consecutive same-op statements in one transaction emit one event each (statement re-maps its table; chunks of one statement share the map). A schema-less mysql url is rejected at config load — the binlog filter would otherwise admit every database on the server. - sqlite: the install script runs in one explicit transaction (execute_batch autocommits per statement — a reinstall had a window with triggers dropped); trigger names and the recorded tbl are lowercased so differently-cased bindings converge on one trigger set, pinned by a live unit test. Stopping a watcher now pokes its wake channel, closing the up-to-2s overlap where old and new watchers double-drained one changelog. Thread-spawn failure is logged. - console form: probe completions are token-guarded so an edit, rename, or remove mid-flight can no longer resurrect a stale result. - e2e: casing case asserts exactly-one-event per subscriber; the statements-path case waits for engine-visible registration; docs and a stale match-arm comment updated. - reload + in-flight-transaction event loss stays as documented at-most-once behavior (doorbell, not ledger) — noted in triggers/mod.rs rather than building listener-retirement machinery.
ytallo
approved these changes
Jul 30, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Adds
capture: native— a per-database opt-in that makesdatabase::row-changedfire for committed writes from any client (psql, other processes, other services), not just SQL executed through this worker. The default (statements) is unchanged; existing configs round-trip byte-identical.pg_notify, one dedicated LISTEN connection per db_iii_row_changeschangelog, drained by a watcher thread (inotify wake-up vianotify,PRAGMA data_versionconfirm gate, 2s fallback tick):memory:rejected at config load)mysql_asyncbinlogfeature)log_bin=ON,binlog_format=ROW,REPLICATION SLAVE, REPLICATION CLIENTShared semantics, deliberately uniform: commit-gated (rollbacks are silence), at-most-once across worker downtime (a doorbell, not a ledger), table-scoped bindings only, bulk statements coalesce into single events with true counts, and self-writes leave the classification path so nothing fires twice. Registration fails loudly with actionable errors (including the exact missing GRANT on mysql, and a casing hint on pg where quoted identifiers are case-sensitive).
Also in this branch:
databaseentry (host.configForms) — per-database cards, driver badges, capture-mode guidance, TLS/pool controls — plus a test connection button backed by the newdatabase::testConnectionfunction (probes a candidate url/tls with one throwaway connection; failures are data; credentials scrubbed from echoed errors).WorkerMetadata(namedatabase+ description) so the console's workers view stops showing thehostname:pidfallback.III_WORKER_NAMEstill wins.Validation
tests/e2e): newpg_native_db/sqlite_native_db/mysql_native_dbhandles sharing the statements-path siblings' physical databases — a write through the sibling pool is a genuine external client from the native handle's perspective. 8 native cases per driver: cross-client delivery + attribution, no double-fire, table-less rejection (sentinel-proven), interactive-tx commit/rollback gating, fan-out + ops filter across a live DDL reinstall, bulk coalescing (100 rows = 1 event), per-driver casing contract, missing-table registration contract (incl. the deliberate mysql accept-then-capture asymmetry). Local full runs: pg 56/56, sqlite 51/51, mysql 51/51.tests/e2e/mysql-init/(first volume init;docker compose down -von older volumes).Notes for review
notify(sqlite fs wake-up),futures-util(binlog StreamExt), and thebinlogfeature on the existingmysql_async.capture).Summary by CodeRabbit