feat(toolkit-db): support configurable outbox table prefixes#4163
Merged
MikeFalcon77 merged 1 commit intoJul 7, 2026
Merged
Conversation
📝 WalkthroughWalkthroughAdds validated outbox table-prefix support across SQL generation, migrations, runtime wiring, tests, benchmarks, and an example. The outbox now derives table families from prefixes, compiles table-aware statements once, and routes processing, maintenance, and verification through configured store-backed execution. ChangesCustom outbox table prefix
Estimated code review effort: 5 (Critical) | ~120 minutes Suggested reviewers: 🚥 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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 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 `@libs/toolkit-db/benches/outbox_throughput.rs`:
- Around line 283-288: The validation in validate_table_prefixes currently
allows duplicate entries, which lets an “independent” benchmark still share the
same table family. Update the check in
OutboxThroughputBench::validate_table_prefixes to require that any non-empty
self.table_prefixes list has one entry per instance and that all prefixes are
distinct, using the existing table_prefixes/num_instances fields to reject
duplicates before the benchmark runs.
- Around line 776-783: The independent-table-families branch in
outbox_throughput’s verification only checks the total consumed count and then
returns, which can miss duplicate/drop pairs inside a single partition bucket.
Update the validation in the `profile.independent_table_families()` path to also
keep the per-partition payload check used elsewhere, leveraging
`partition_key_offset`-based partition buckets and the existing
`state.consumed`/bucket assertions so each bucket still validates its expected
message set before returning.
In `@libs/toolkit-db/src/outbox/store.rs`:
- Around line 69-85: The MySQL batch insert logic in the outbox store currently
reconstructs IDs with `first_id + i`, but that only works when the backend
guarantees contiguous auto-increment allocation. Update the body insert path in
the outbox store to either explicitly enforce the MySQL/InnoDB batch-ID
precondition before using arithmetic reconstruction, or switch to a per-row ID
retrieval strategy for MySQL. Keep the fix localized around the batch insert
branch that calls `last_insert_id()` and builds the returned ID range so
`incoming.body_id` always maps to the correct body row.
In `@libs/toolkit-db/src/outbox/tables.rs`:
- Around line 141-149: The chained replace logic in `Tables::format_sql` is
rewriting identifiers multiple times when a custom prefix contains a default
`toolkit_outbox_*` token, which produces malformed table names. Update
`format_sql` to use non-overlapping placeholders or construct the SQL from the
`Tables::*` accessors directly so each identifier is substituted exactly once,
and add a regression test covering a prefix like `toolkit_outbox_body` to verify
names such as `dead_letters` and `body_dead_letters` stay correct.
🪄 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: 7b376c64-738e-4a73-8d25-c044f7210046
📒 Files selected for processing (19)
libs/toolkit-db/benches/outbox_throughput.rslibs/toolkit-db/examples/outbox_custom_prefix.rslibs/toolkit-db/src/outbox/README.mdlibs/toolkit-db/src/outbox/builder.rslibs/toolkit-db/src/outbox/core.rslibs/toolkit-db/src/outbox/dead_letter.rslibs/toolkit-db/src/outbox/dialect.rslibs/toolkit-db/src/outbox/integration_tests.rslibs/toolkit-db/src/outbox/manager.rslibs/toolkit-db/src/outbox/migrations.rslibs/toolkit-db/src/outbox/mod.rslibs/toolkit-db/src/outbox/store.rslibs/toolkit-db/src/outbox/strategy.rslibs/toolkit-db/src/outbox/tables.rslibs/toolkit-db/src/outbox/types.rslibs/toolkit-db/src/outbox/workers/processor.rslibs/toolkit-db/src/outbox/workers/reconciler.rslibs/toolkit-db/src/outbox/workers/sequencer.rslibs/toolkit-db/src/outbox/workers/vacuum.rs
striped-zebra-dev
force-pushed
the
configurable-outbox-table-prefix
branch
2 times, most recently
from
June 30, 2026 13:41
7d4b3a4 to
79e7a58
Compare
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
libs/toolkit-db/benches/outbox_throughput.rs (1)
765-797: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winAlign message counts to the expanded independent partition space.
For independent families, verification uses
total_partitions * num_instances, butaligned_message_count()still rounds only tototal_partitions. The added2i_independent_prefix_batchprofile rounds 100,000 to 100,032 for 64 partitions, then verifies against 128 buckets, so the total-count and per-bucket assertions cannot both pass.Proposed fix
impl BenchProfile { + fn verification_partitions(&self) -> usize { + let total_parts = self.total_partitions(); + if self.independent_table_families() { + total_parts * self.num_instances + } else { + total_parts + } + } + /// Round message_count up to be evenly divisible by total_partitions. fn aligned_message_count(&self) -> usize { - let tp = self.total_partitions(); + let tp = self.verification_partitions(); let remainder = self.message_count % tp; if remainder == 0 { self.message_count } else { self.message_count + tp - remainder @@ fn verify_results(state: &BenchState, profile: &BenchProfile) { let msg_count = profile.aligned_message_count(); - let total_parts = profile.total_partitions(); - let verification_parts = if profile.independent_table_families() { - total_parts * profile.num_instances - } else { - total_parts - }; + let verification_parts = profile.verification_partitions();🤖 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 `@libs/toolkit-db/benches/outbox_throughput.rs` around lines 765 - 797, The message-count alignment is still based on the non-expanded partition total, which breaks verification when independent table families multiply the bucket space. Update the message-alignment logic used by the benchmark (including aligned_message_count and the verification path in outbox_throughput) so it rounds against verification_parts when profile.independent_table_families() is enabled, and keep the total consumed assertion and per-partition checks consistent with that expanded count.libs/toolkit-db/src/outbox/integration_tests.rs (1)
1808-1815: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winDon’t drop decode failures while collecting vacuum IDs.
filter_map(...ok())can turn a brokenselect_outgoing_chunkprojection into empty/partial ID lists, letting this helper delete the wrong subset and still reset the counter. Fail the test on decode errors instead.Proposed fix
- let outgoing_ids: Vec<i64> = rows - .iter() - .filter_map(|r| r.try_get_by_index::<i64>(0).ok()) - .collect(); - let body_ids: Vec<i64> = rows - .iter() - .filter_map(|r| r.try_get_by_index::<i64>(1).ok()) - .collect(); + let outgoing_ids: Vec<i64> = rows + .iter() + .map(|r| r.try_get_by_index::<i64>(0)) + .collect::<Result<_, _>>() + .expect("outgoing id column"); + let body_ids: Vec<i64> = rows + .iter() + .map(|r| r.try_get_by_index::<i64>(1)) + .collect::<Result<_, _>>() + .expect("body id column");🤖 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 `@libs/toolkit-db/src/outbox/integration_tests.rs` around lines 1808 - 1815, The vacuum ID collection in the integration test is silently skipping decode failures because `filter_map(...ok())` on `rows` can hide a broken `select_outgoing_chunk` projection and produce partial ID lists. Update the `outgoing_ids` and `body_ids` collection logic in this test to fail immediately on any `try_get_by_index` decode error instead of dropping it, so the helper cannot proceed with incorrect IDs or reset the counter after a bad projection.libs/toolkit-db/src/outbox/core.rs (1)
390-398: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winWrap MySQL single enqueue in a transaction.
exec_insert_body_and_incomingis two statements on MySQL. Whenenqueue()is called with a plain connection, a failure after the body insert can leave an orphaned body row; mirror the MySQL transaction wrapper already used byinsert_batch.🐛 Proposed fix outline
async fn insert_body_and_incoming( runner: &SeaOrmRunner<'_>, statements: &OutboxStatements, partition_id: i64, payload: Vec<u8>, payload_type: &str, ) -> Result<i64, OutboxError> { - let (conn, backend): (&dyn ConnectionTrait, DbBackend) = match runner { - SeaOrmRunner::Conn(c) => (*c, c.get_database_backend()), - SeaOrmRunner::Tx(t) => (*t, t.get_database_backend()), - }; + let backend = match runner { + SeaOrmRunner::Conn(c) => c.get_database_backend(), + SeaOrmRunner::Tx(t) => t.get_database_backend(), + }; debug_assert_eq!(backend, statements.backend()); - let store = OutboxStore::new(statements); - let incoming_id = store - .exec_insert_body_and_incoming(conn, partition_id, payload, payload_type) - .await?; + if backend == DbBackend::MySql { + return match runner { + SeaOrmRunner::Conn(c) => { + let txn = c.begin().await?; + let store = OutboxStore::new(statements); + let incoming_id = store + .exec_insert_body_and_incoming(&txn, partition_id, payload, payload_type) + .await?; + txn.commit().await?; + Ok(incoming_id) + } + SeaOrmRunner::Tx(t) => { + let store = OutboxStore::new(statements); + let incoming_id = store + .exec_insert_body_and_incoming(*t, partition_id, payload, payload_type) + .await?; + Ok(incoming_id) + } + }; + } - Ok(incoming_id) + let conn: &dyn ConnectionTrait = match runner { + SeaOrmRunner::Conn(c) => *c, + SeaOrmRunner::Tx(t) => *t, + }; + let store = OutboxStore::new(statements); + Ok(store + .exec_insert_body_and_incoming(conn, partition_id, payload, payload_type) + .await?) }🤖 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 `@libs/toolkit-db/src/outbox/core.rs` around lines 390 - 398, The MySQL single-enqueue path in OutboxStore::enqueue currently calls exec_insert_body_and_incoming on a plain connection, so a mid-call failure can leave an orphaned body row. Update the enqueue flow to mirror insert_batch by wrapping the MySQL case in a transaction when SeaOrmRunner::Conn is used, while keeping SeaOrmRunner::Tx unchanged, and then run exec_insert_body_and_incoming through that transactional runner.
🤖 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 `@libs/toolkit-db/src/contention.rs`:
- Around line 44-47: Tighten the MySQL retry matcher in the contention logic by
updating the constants and matcher in contention::is_mysql_retryable so it only
recognizes the specific wsrep certification conflict phrases instead of
standalone wsrep or certification. Keep the existing deadlock and try restarting
transaction handling, but narrow the wsrep/certification checks to the
documented Galera conflict text. Add a negative test alongside the existing
retry tests to verify an unrelated message containing wsrep or certification is
not treated as retryable.
In `@libs/toolkit-db/src/outbox/migrations.rs`:
- Around line 536-555: The MySQL ID sequence tables are being created inside an
existing migration body, so already-applied databases will skip them and later
fail when runtime batch inserts call the ID reservation path. Add a separate
follow-up migration and ensure it is included from both outbox_migrations() and
outbox_migrations_with_prefix(...), using the existing mysql table setup helpers
in migrations.rs. Also update the seeding logic for
create_mysql_id_sequence_tables so next_id starts from MAX(id) + 1 for the body
and incoming tables when rows already exist, instead of always seeding 1.
In `@libs/toolkit-db/src/outbox/store.rs`:
- Around line 212-239: Single-row MySQL enqueue inserts still use the
AUTO_INCREMENT/LAST_INSERT_ID path instead of the new sequence reservation flow.
Update exec_insert_body and exec_insert_incoming in outbox::store to follow the
same explicit-ID logic used by the MySQL batch helpers, reusing the batch insert
code path or shared helper that reserves IDs from {body,incoming}_id_sequence
before inserting. Keep the existing non-MySQL behavior unchanged and make sure
these methods no longer depend on insert_returning_id for MySQL single inserts.
---
Outside diff comments:
In `@libs/toolkit-db/benches/outbox_throughput.rs`:
- Around line 765-797: The message-count alignment is still based on the
non-expanded partition total, which breaks verification when independent table
families multiply the bucket space. Update the message-alignment logic used by
the benchmark (including aligned_message_count and the verification path in
outbox_throughput) so it rounds against verification_parts when
profile.independent_table_families() is enabled, and keep the total consumed
assertion and per-partition checks consistent with that expanded count.
In `@libs/toolkit-db/src/outbox/core.rs`:
- Around line 390-398: The MySQL single-enqueue path in OutboxStore::enqueue
currently calls exec_insert_body_and_incoming on a plain connection, so a
mid-call failure can leave an orphaned body row. Update the enqueue flow to
mirror insert_batch by wrapping the MySQL case in a transaction when
SeaOrmRunner::Conn is used, while keeping SeaOrmRunner::Tx unchanged, and then
run exec_insert_body_and_incoming through that transactional runner.
In `@libs/toolkit-db/src/outbox/integration_tests.rs`:
- Around line 1808-1815: The vacuum ID collection in the integration test is
silently skipping decode failures because `filter_map(...ok())` on `rows` can
hide a broken `select_outgoing_chunk` projection and produce partial ID lists.
Update the `outgoing_ids` and `body_ids` collection logic in this test to fail
immediately on any `try_get_by_index` decode error instead of dropping it, so
the helper cannot proceed with incorrect IDs or reset the counter after a bad
projection.
🪄 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: 6743a70a-55c6-4a24-9351-8830565412a6
📒 Files selected for processing (21)
libs/toolkit-db/benches/outbox_throughput.rslibs/toolkit-db/examples/outbox_custom_prefix.rslibs/toolkit-db/src/contention.rslibs/toolkit-db/src/outbox/README.mdlibs/toolkit-db/src/outbox/builder.rslibs/toolkit-db/src/outbox/core.rslibs/toolkit-db/src/outbox/dead_letter.rslibs/toolkit-db/src/outbox/dialect.rslibs/toolkit-db/src/outbox/integration_tests.rslibs/toolkit-db/src/outbox/manager.rslibs/toolkit-db/src/outbox/migrations.rslibs/toolkit-db/src/outbox/mod.rslibs/toolkit-db/src/outbox/statements.rslibs/toolkit-db/src/outbox/store.rslibs/toolkit-db/src/outbox/strategy.rslibs/toolkit-db/src/outbox/tables.rslibs/toolkit-db/src/outbox/types.rslibs/toolkit-db/src/outbox/workers/processor.rslibs/toolkit-db/src/outbox/workers/reconciler.rslibs/toolkit-db/src/outbox/workers/sequencer.rslibs/toolkit-db/src/outbox/workers/vacuum.rs
✅ Files skipped from review due to trivial changes (1)
- libs/toolkit-db/src/outbox/README.md
🚧 Files skipped from review as they are similar to previous changes (6)
- libs/toolkit-db/examples/outbox_custom_prefix.rs
- libs/toolkit-db/src/outbox/types.rs
- libs/toolkit-db/src/outbox/mod.rs
- libs/toolkit-db/src/outbox/workers/sequencer.rs
- libs/toolkit-db/src/outbox/strategy.rs
- libs/toolkit-db/src/outbox/dialect.rs
striped-zebra-dev
force-pushed
the
configurable-outbox-table-prefix
branch
2 times, most recently
from
June 30, 2026 14:30
458222d to
1a089aa
Compare
Add validated outbox table-family configuration for migrations and runtime SQL. Introduce OutboxStore to bind backend, dialect, and table names together, and update workers, dead letters, tests, docs, examples, and benchmarks for custom table prefixes. Signed-off-by: Striped Zebra <striped.zebra.dev@gmail.com>
striped-zebra-dev
force-pushed
the
configurable-outbox-table-prefix
branch
from
July 1, 2026 08:23
1a089aa to
62c0d2b
Compare
There was a problem hiding this comment.
🧹 Nitpick comments (1)
libs/toolkit-db/src/outbox/integration_tests.rs (1)
2864-2864: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winMake this prove a custom-prefix migration mismatch, not a generic empty DB failure.
Line 2864 starts from an empty DB, so the test would pass even if startup simply fails because all outbox tables are missing. Running default migrations first better verifies that a custom-prefix instance does not fall back to default tables when its matching migrations are absent.
🧪 Proposed test tightening
- let db = setup_empty_db("ch10_custom_prefix_missing_migration").await; + let db = setup_db("ch10_custom_prefix_missing_migration").await;🤖 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 `@libs/toolkit-db/src/outbox/integration_tests.rs` at line 2864, The test setup in the custom-prefix outbox migration case is too weak because it starts from an empty database, so it only proves generic startup failure. Update the test around the ch10_custom_prefix_missing_migration scenario to run the default migration setup first, then validate that the custom-prefix path fails because its matching migrations are absent and does not fall back to default outbox tables. Use the existing test helper/fixture flow in integration_tests.rs and the custom-prefix startup/migration code paths to keep the assertion specific to the prefix mismatch.
🤖 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.
Nitpick comments:
In `@libs/toolkit-db/src/outbox/integration_tests.rs`:
- Line 2864: The test setup in the custom-prefix outbox migration case is too
weak because it starts from an empty database, so it only proves generic startup
failure. Update the test around the ch10_custom_prefix_missing_migration
scenario to run the default migration setup first, then validate that the
custom-prefix path fails because its matching migrations are absent and does not
fall back to default outbox tables. Use the existing test helper/fixture flow in
integration_tests.rs and the custom-prefix startup/migration code paths to keep
the assertion specific to the prefix mismatch.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: d1d95b48-26f6-4083-ac5f-47e77c38092f
📒 Files selected for processing (22)
libs/toolkit-db/Cargo.tomllibs/toolkit-db/benches/outbox_throughput.rslibs/toolkit-db/examples/outbox_custom_prefix.rslibs/toolkit-db/src/contention.rslibs/toolkit-db/src/outbox/README.mdlibs/toolkit-db/src/outbox/builder.rslibs/toolkit-db/src/outbox/core.rslibs/toolkit-db/src/outbox/dead_letter.rslibs/toolkit-db/src/outbox/dialect.rslibs/toolkit-db/src/outbox/integration_tests.rslibs/toolkit-db/src/outbox/manager.rslibs/toolkit-db/src/outbox/migrations.rslibs/toolkit-db/src/outbox/mod.rslibs/toolkit-db/src/outbox/statements.rslibs/toolkit-db/src/outbox/store.rslibs/toolkit-db/src/outbox/strategy.rslibs/toolkit-db/src/outbox/tables.rslibs/toolkit-db/src/outbox/types.rslibs/toolkit-db/src/outbox/workers/processor.rslibs/toolkit-db/src/outbox/workers/reconciler.rslibs/toolkit-db/src/outbox/workers/sequencer.rslibs/toolkit-db/src/outbox/workers/vacuum.rs
💤 Files with no reviewable changes (8)
- libs/toolkit-db/src/outbox/workers/reconciler.rs
- libs/toolkit-db/src/outbox/workers/processor.rs
- libs/toolkit-db/src/outbox/workers/sequencer.rs
- libs/toolkit-db/src/outbox/types.rs
- libs/toolkit-db/src/outbox/tables.rs
- libs/toolkit-db/src/outbox/store.rs
- libs/toolkit-db/src/outbox/workers/vacuum.rs
- libs/toolkit-db/src/outbox/strategy.rs
✅ Files skipped from review due to trivial changes (1)
- libs/toolkit-db/src/outbox/README.md
🚧 Files skipped from review as they are similar to previous changes (10)
- libs/toolkit-db/examples/outbox_custom_prefix.rs
- libs/toolkit-db/src/outbox/builder.rs
- libs/toolkit-db/src/outbox/mod.rs
- libs/toolkit-db/src/outbox/manager.rs
- libs/toolkit-db/src/contention.rs
- libs/toolkit-db/src/outbox/statements.rs
- libs/toolkit-db/src/outbox/core.rs
- libs/toolkit-db/benches/outbox_throughput.rs
- libs/toolkit-db/src/outbox/dead_letter.rs
- libs/toolkit-db/src/outbox/dialect.rs
MikeFalcon77
approved these changes
Jul 7, 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.
Add validated outbox table-family configuration for migrations and runtime SQL. Introduce OutboxStore to bind backend, dialect, and table names together, and update workers, dead letters, tests, docs, examples, and benchmarks for custom table prefixes.
Summary by CodeRabbit