fix(conversations): atomic item mutations with transactional message cache sync - #570
Conversation
|
PR too large: 815 lines added (limit: 750, excludes Cargo files, tests, docs, examples, and benchmarks). Please split into smaller PRs. Add |
franciscojavierarceo
left a comment
There was a problem hiding this comment.
Requesting changes for two merge-blocking correctness issues and one SQLite performance regression. The existing mixed-batch API thread at filter.rs:581 also remains valid and unresolved.\n\nI reproduced the legacy-duplicate initialization failure on SQLite. Targeted transactional tests otherwise passed locally, while the PR is currently also failing nightly rustfmt and the coverage threshold.
praxis-bot
left a comment
There was a problem hiding this comment.
Review: fix(conversations): atomic item mutations with transactional message cache sync
| Severity | Count |
|---|---|
| Large | 1 |
| Medium | 1 |
Solid architectural improvement. The transactional approach correctly eliminates the read-compute-write race on max_item_position that caused position collisions under concurrent appends. SELECT ... FOR UPDATE (PostgreSQL) and BEGIN IMMEDIATE (SQLite) are the right serialization strategies for each backend. The removal of the bogus ConversationRecord construction in refresh_message_cache / sync_conversation_messages is a clean simplification. Test coverage for the new methods is thorough, including the concurrent SQLite test that proves the locking works.
praxis-bot
left a comment
There was a problem hiding this comment.
Review: fix(conversations): atomic item mutations with transactional message cache sync
| Severity | Count |
|---|---|
| Medium | 1 |
The transactional approach is well-designed and the serialization strategies are correct for each backend (SELECT ... FOR UPDATE for PostgreSQL, BEGIN IMMEDIATE for SQLite). Clean removal of the ConversationRecord construction hack in refresh_message_cache / sync_conversation_messages. The concurrent SQLite test is a good addition. Handler refactoring correctly simplifies the existence-check and mutation calls.
praxis-bot
left a comment
There was a problem hiding this comment.
Review: fix(conversations): atomic item mutations with transactional message cache sync
| Severity | Count |
|---|---|
| (none new) | 0 |
Strong improvement. The transactional approach correctly eliminates the TOCTOU race on max_item_position that caused position collisions under concurrent appends. Serialization strategies are appropriate for each backend: SELECT ... FOR UPDATE for PostgreSQL and BEGIN IMMEDIATE for SQLite. The *_rebuild_messages helpers properly detect concurrent conversation deletion via rows_affected() == 0 and propagate the error, rolling back the entire transaction so no orphaned items persist.
Handler refactoring is clean. The move from two separate operations (create_conversation_items + sync_conversation_messages) to a single create_items_and_sync_messages call removes the bogus ConversationRecord construction that existed only to thread tenant_id/conversation_id through to the sync logic. The handle_create_items and handle_delete_item existence checks are simplified to discard the record since it is no longer needed for the sync call.
Test coverage is thorough: position assignment, multi-batch continuation, empty batch no-op, missing conversation rollback (verifying items are cleaned up), deletion with cache sync, nonexistent item returns false, and a concurrent write test verifying distinct positions under contention. The PostgreSQL tests mirror the SQLite ones for the core paths. The rollback verification tests (*_missing_conversation_errors) are well-designed.
The unique index on (tenant_id, conversation_id, position) is good defense-in-depth. Previous review feedback about migration risk on existing data with duplicate positions remains the main outstanding consideration for the upgrade path.
No new findings at Medium or above.
|
Hi @aslakknutsen @franciscojavierarceo, I have addressed your requested changes. PTAL when you get a chance |
praxis-bot
left a comment
There was a problem hiding this comment.
Review: fix(conversations): atomic item mutations with transactional message cache sync
| Severity | Count |
|---|---|
| Medium | 1 |
The transactional redesign is solid and the test coverage for the new methods is thorough (position assignment, multi-batch continuation, empty batch no-op, missing conversation rollback, deletion with cache sync, concurrent writes). The prior review findings remain the main outstanding items.
One new finding: the coverage exclusion added for store/postgres.rs is overly broad and suppresses coverage measurement for all existing and future PostgreSQL store code.
praxis-bot
left a comment
There was a problem hiding this comment.
Round 6 review -- 1 new finding (Medium).
The transactional approach is sound and the test coverage for the new methods is solid. One scope-consistency concern in the INSERT path.
praxis-bot
left a comment
There was a problem hiding this comment.
Review: fix(conversations): atomic item mutations with transactional message cache sync
| Severity | Count |
|---|---|
| (none new) | 0 |
Round 7 review -- no new findings at Medium or above.
The iterative work since the last review has been good: commit da27ad4 restored the existence guard in *_rebuild_messages via rows_affected() checks, and f747003 eliminated the empty-messages fallback in handle_create_conversation. The transactional design is sound, serialization strategies are correct per backend, and the test suite is thorough (position assignment, multi-batch continuation, concurrent writes, missing-conversation rollback for both create and delete paths).
The following prior findings remain the outstanding items for consideration before merge:
- [Large] Unique index migration risk on existing data with duplicate positions (
schemas.rs) - [Medium] PostgreSQL lock query result discarded without checking row existence --
pg_rebuild_messagescatches it downstream, but early-out onlocked.is_none()would avoid inserting items that get rolled back (postgres.rs) - [Medium] INSERT binds
item.tenant_id/item.conversation_idinstead of the function parameters in both PostgreSQL and SQLite -- scope fields can silently diverge (postgres.rs,sqlite.rs) - [Medium] Coverage exclusion suppresses the entire
store/postgres.rsfile, not just the new transactional methods (Makefile)
|
@aslakknutsen @franciscojavierarceo PTAL to unblock this, thanks |
|
@aslakknutsen @franciscojavierarceo can you please review again to unblock this?
|
|
@eoinfennessy please rebase |
…cache sync Move position assignment, item insert/delete, and message cache rebuild into single database transactions to eliminate races under concurrent appends and stale cache overwrites. This replaces refresh_message_cache and sync_conversation_messages, which constructed a ConversationRecord with bogus field values just to thread tenant_id and conversation_id through to the sync logic. Closes praxis-proxy#358 Signed-off-by: Eoin Fennessy <efenness@redhat.com>
Spawns two tasks that each append one item to the same conversation concurrently, then asserts distinct positions and a coherent two-item message cache. Consolidates make_file_store helpers to accept an optional items_table parameter. Signed-off-by: Eoin Fennessy <efenness@redhat.com>
…ssages Add debug_assert! to SQLite and PostgreSQL implementations to validate that all items share the same tenant_id and conversation_id, preventing silent position misassignment from a future internal caller. Signed-off-by: Eoin Fennessy <efenness@redhat.com>
Replace manual BEGIN IMMEDIATE / COMMIT / ROLLBACK raw queries with
pool.begin_with("BEGIN IMMEDIATE") to get SQLx's RAII rollback-on-drop
behavior. If the future is cancelled after BEGIN, or an error propagates
before COMMIT, the Transaction guard now automatically rolls back when
dropped — preventing a pooled connection from being returned with an
open write lock.
Aligns helper function signatures with the Postgres side by accepting
&mut Transaction instead of &mut PoolConnection.
Signed-off-by: Eoin Fennessy <efenness@redhat.com>
Signed-off-by: Eoin Fennessy <efenness@redhat.com>
…te_items_and_sync_messages The method already assumed all items shared the same tenant_id and conversation_id (extracting them from items.first() with a debug_assert). Making them explicit parameters aligns the signature with delete_item_and_sync_messages and removes the need for the runtime assertion. Signed-off-by: Eoin Fennessy <efenness@redhat.com>
The existing `conversation_item_methods_fail_without_items_table` test verified six `ConversationItemStore` methods but omitted the two new transactional methods added by this branch. Add assertions for `create_items_and_sync_messages` and `delete_item_and_sync_messages` to close the gap. Signed-off-by: Eoin Fennessy <efenness@redhat.com>
The PostgreSQL store implementation mirrors the SQLite store but its 27 tests are all `#[ignore]` because they require a running database that CI does not provide. Including postgres.rs in the coverage report counts ~300 lines of untestable code against the threshold, masking the actual test quality of the exercised SQLite path. Signed-off-by: Eoin Fennessy <efenness@redhat.com>
…xis-proxy#578) The CallTool error variant used a raw String URL and included the third-party source error in its Display output, both of which could leak credentials stored in URL query strings. Switch to McpDisplayUrl (which strips query, fragment, and userinfo) and drop the source from the formatted message, matching the existing pattern on Connection and ListTools variants. Signed-off-by: Sébastien Han <seb@redhat.com>
* fix(token_usage): Gemini candidatesTokenCount optional
Gemini omits candidatesTokenCount in safety-filtered responses where
no output candidates are generated. I expect the required u64 field
could cause deserialization to fail, silently losing the input token
count as well. Now defaults to zero when absent.
Signed-off-by: Shane Utt <shaneutt@linux.com>
* fix(build): fix package name in filter discovery
Signed-off-by: Shane Utt <shaneutt@linux.com>
* fix(a2a): flush pending SSE state before clearing at end-of-stream
When a provider omits the trailing blank line before closing the
connection the SSE buffers may contain a complete but undispatched
event. The filter cleared this state without processing it, silently
dropping the last task route. This updates it to flush the state
and prior to clearing so these buffered SSE payloads will get
dispatched before we clean up.
Signed-off-by: Shane Utt <shaneutt@linux.com>
* perf(apis): replace allocs in URL path matching
Multiple sites collected path.split('/') into a heap-allocated
Vec<&str> just to pattern-match fixed-shape URL segments. These
run on every request for matched routes. This patch replaces those
with strip_prefix and strip_suffix chains that avoid the heap
allocation entirely.
Signed-off-by: Shane Utt <shaneutt@linux.com>
* fix(mcp): block unique-local IPv6 in SSRF checks
is_ssrf_sensitive did not check unique-local IPv6 (fc00::/7), so
addresses like fd00:ec2::23 (AWS ECS credential endpoint over IPv6)
bypassed SSRF protection.
Signed-off-by: Shane Utt <shaneutt@linux.com>
* fix(mcp): block cookie and fwd headers in mcp req
is_blocked_mcp_header did not block cookie, set-cookie, or
forwarded/x-forwarded-* headers. While these come from operator
configs rather than end-user HTTP requests, accidental pass-through
could leak stuff.
Signed-off-by: Shane Utt <shaneutt@linux.com>
* refactor: extract utils from store and rehydrate
There were some function pairs that were duplicated across filters, this
consolidates those into a single implementation both use.
Signed-off-by: Shane Utt <shaneutt@linux.com>
---------
Signed-off-by: Shane Utt <shaneutt@linux.com>
The refactor of sync_conversation_messages into create_items_and_sync_messages / delete_item_and_sync_messages dropped the pre-refactor existence guard: the message-cache UPDATE result was discarded, so a conversation deleted concurrently between the handler's existence check and the transaction would silently succeed with a zero-row update. Restore the guard at a single choke point per backend by checking rows_affected on the UPDATE in pg_rebuild_messages and sqlite_rebuild_messages, returning the original "conversation disappeared during message sync" error. The enclosing transaction rolls back on the propagated error, so no item mutations persist. This covers both create and delete paths and works for the SQLite path, which uses BEGIN IMMEDIATE and has no row lock to inspect. Add SQLite tests for both paths and mirrored ignored PostgreSQL tests. Signed-off-by: Eoin Fennessy <efenness@redhat.com> rh-pre-commit.version: 2.4.0 rh-pre-commit.check-secrets: ENABLED
create_items_and_sync_messages binds item.tenant_id/item.conversation_id in the per-item INSERT while the lock, MAX(position), and rebuild queries in the same transaction use the tenant_id/conversation_id function parameters. If a caller ever passed items whose scope fields diverged from the parameters, rows would be inserted outside the locked/synced scope. Bind the function parameters instead so the invariant can't be violated. Signed-off-by: Eoin Fennessy <efenness@redhat.com> rh-pre-commit.version: 2.4.0 rh-pre-commit.check-secrets: ENABLED
17a8c9c to
ad90f9e
Compare
Signed-off-by: Eoin Fennessy <efenness@redhat.com>
Signed-off-by: Eoin Fennessy <efenness@redhat.com>
ad90f9e to
7b39391
Compare
Signed-off-by: Eoin Fennessy <efenness@redhat.com>
94e21ef to
e4d62ca
Compare
praxis-bot
left a comment
There was a problem hiding this comment.
Review: fix(conversations): atomic item mutations with transactional message cache sync
| Severity | Count |
|---|---|
| (none new) | 0 |
Round 8 review -- no new findings at Medium or above.
The two actionable findings from Round 7 have been addressed:
- [Fixed]
rows_affected()check restored inpg_rebuild_messagesandsqlite_rebuild_messages(9c9e8ff). A conversation deleted concurrently between the handler's existence check and the transaction now produces"conversation disappeared during message sync"and rolls back item mutations. Both create and delete paths are covered, with SQLite tests for each and mirrored#[ignore]d PostgreSQL tests. - [Fixed] INSERT in
create_items_and_sync_messagesnow binds the function parameterstenant_id/conversation_idinstead of the item record fields (345ad3d). The scope invariant is enforced at the query level across both backends.
Remaining prior findings (unchanged from Round 7):
- [Large] Unique index migration risk on existing data with duplicate positions (
schemas.rs) - [Medium] Coverage exclusion suppresses the entire
store/postgres.rsfile (Makefile)
Summary
create_items_and_sync_messagesanddelete_item_and_sync_messagesto theConversationItemStoretrait, combining position assignment, item mutation, and message cache rebuild into single database transactions. This eliminate races under concurrent appends and stale cache overwritesrefresh_message_cacheandsync_conversation_messages, which constructed aConversationRecordwith bogus field values just to threadtenant_idandconversation_idthrough to the sync logic(tenant_id, conversation_id, position)as defense-in-depth against position collisionsBEGIN IMMEDIATE(SQLite) andSELECT ... FOR UPDATE(PostgreSQL) to serialize concurrent writesTest plan
duplicate_positionstest replaced withduplicate_position_rejected_by_unique_constraintCloses #358