diff --git a/apis/src/openai/conversations/filter.rs b/apis/src/openai/conversations/filter.rs index 726d4d636..e23863a78 100644 --- a/apis/src/openai/conversations/filter.rs +++ b/apis/src/openai/conversations/filter.rs @@ -92,6 +92,22 @@ impl OpenaiConversationsFilter { } } + /// Build a filter around a pre-initialized store for tests. + /// + /// Pre-seeding the `OnceCell` lets tests inject a fault-injecting store + /// (e.g. one whose `create_conversation_items` fails) without standing up a + /// real database, so append-back error handling can be exercised directly. + #[cfg(test)] + pub(super) fn with_store_for_test(config: ConversationsConfig, store: Arc) -> Self { + Self { + config, + // Pre-initialize the cell so `get_or_init_store` returns this store + // without touching a real backend. The outer `Some` marks the cell + // initialized; the inner `Some` is the stored (available) store. + store: OnceCell::new_with(Some(Some(store))), + } + } + /// Build the configured store backend. async fn build_store(&self) -> Result, StoreError> { let responses_table = self.config.responses_table(); @@ -474,9 +490,22 @@ impl HttpFilter for OpenaiConversationsFilter { }; let conv_id = items.conversation_id; - if let Err(e) = self.append_items_blocking(&items.tenant_id, &conv_id, ctx, items.all_items) { - warn!(error = %e, conversation_id = %conv_id, "conversation append-back failed"); - } + // Fail closed on lost items. Append-back runs at end-of-stream while the + // completed response body is still buffered (StreamBuffer), before any + // byte is released downstream. Under the default `failure_mode: closed`, + // the buffered body is never released after a persistence failure, so the + // client cannot observe a clean success that hides items which never + // persisted (#837). The exact downstream outcome is Pingora-timing-dependent + // — a not-yet-flushed header yields a clean 500, an already-committed one + // yields a 2xx followed by a reset — but either way the body is withheld. + // `failure_mode: open` is an explicit operator opt-out of that guarantee: + // the pipeline logs this error and converts it to Continue, releasing the + // body even though items were lost. Only pre-commit failures (item insertion + // and earlier) reach this `?`: a post-commit message-cache refresh failure + // is tolerated inside `persist_items` because the cache is a self-healing + // projection (see `refresh_message_cache`). + self.append_items_blocking(&items.tenant_id, &conv_id, ctx, items.all_items) + .inspect_err(|e| warn!(error = %e, conversation_id = %conv_id, "conversation append-back failed"))?; Ok(FilterAction::Continue) } @@ -590,6 +619,12 @@ async fn persist_items( .await .map_err(|e| -> FilterError { Box::new(e) })?; + // Item rows are durable past this point. The message cache is a self-healing + // projection rebuilt from the items table on the next successful sync, so a + // refresh failure is not data loss and must not fail the turn: propagating it + // would abort a request whose items already committed and drive a client retry + // that re-appends the same (typically id-less) input items as duplicates + // (#837). Log and continue; a later successful cache-refresh re-syncs the cache. refresh_message_cache(store, tenant_id, conversation_id).await; debug!( conversation_id, @@ -600,6 +635,18 @@ async fn persist_items( } /// Refresh the denormalized conversation message cache after item mutation. +/// +/// The cache is a projection of the items table that `openai_responses_rehydrate` +/// replays on the next turn. It runs *after* the item rows have committed, so it +/// is deliberately best-effort: the sync is idempotent — it rebuilds `messages` +/// from the durable items table — so a later successful sync re-syncs whatever this +/// attempt left behind (a later *append* is not sufficient: its own refresh may +/// also fail). Failing the turn here would abort a request whose +/// items already persisted and push the client into a retry that re-appends the +/// same items as duplicates, so a refresh failure is logged and swallowed rather +/// than propagated (#837). A transient stale-cache window remains until the next +/// successful sync; closing it fully (atomic item-insert + projection) is tracked +/// separately. async fn refresh_message_cache(store: &dyn ConversationItemStore, tenant_id: &str, conversation_id: &str) { let record = ConversationRecord { conversation_id: conversation_id.to_owned(), @@ -609,7 +656,12 @@ async fn refresh_message_cache(store: &dyn ConversationItemStore, tenant_id: &st messages: Value::Null, }; if let Err(e) = handlers::sync_conversation_messages(store, record).await { - warn!(error = %e, conversation_id, "conversation message sync failed after append-back"); + warn!( + error = %e, + conversation_id, + "conversation message cache refresh failed after append-back; items are durable and the \ + cache re-syncs on a later successful refresh (#837)" + ); } } diff --git a/apis/src/openai/conversations/tests.rs b/apis/src/openai/conversations/tests.rs index 4f346e75c..6f7ba04f1 100644 --- a/apis/src/openai/conversations/tests.rs +++ b/apis/src/openai/conversations/tests.rs @@ -1,7 +1,7 @@ // SPDX-License-Identifier: Apache-2.0 // Copyright (c) 2026 Praxis Contributors -use std::collections::BTreeMap; +use std::{collections::BTreeMap, sync::Arc}; use bytes::Bytes; use http::Method; @@ -16,6 +16,7 @@ use super::{ }; use crate::{ openai::responses::state::ResponsesState, + store::{ConversationItemRecord, ConversationItemStore, ConversationRecord, StoreError}, test_utils::{make_filter_context, make_request, make_response}, }; @@ -3425,6 +3426,89 @@ async fn on_response_body_appends_completed_response() { assert_eq!(items.len(), 2, "append-back should persist both input and output items"); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn on_response_body_surfaces_item_insert_failure() { + let filter = build_failing_filter(FailingItemStore { + fail_create_items: true, + fail_message_sync: false, + }); + + let req = make_request(Method::POST, "/v1/responses"); + let mut ctx = make_filter_context(&req); + ctx.current_filter_id = Some(0); + set_append_back_metadata(&mut ctx); + ctx.extensions.insert(ResponsesState { + input: vec![serde_json::json!({"type": "message", "role": "user", "content": "hello from append"})], + ..ResponsesState::default() + }); + + let mut resp = make_response(); + resp.headers + .insert(http::header::CONTENT_TYPE, "application/json".parse().unwrap()); + ctx.response_header = Some(&mut resp); + drop(filter.on_response(&mut ctx).await.unwrap()); + + // A completed response whose items cannot be persisted must not be swallowed: + // the filter propagates the append-back failure as an error (#837). The + // client-visible outcome is then the pipeline's failure-mode decision — a + // withheld body under the default `failure_mode: closed`, or a logged release + // under `open`. + let response_json = serde_json::json!({ + "status": "completed", + "output": [{"type": "message", "role": "assistant", "content": "hi from model"}] + }); + let mut body = Some(Bytes::from(serde_json::to_vec(&response_json).unwrap())); + let result = filter.on_response_body(&mut ctx, &mut body, true); + + assert!( + result.is_err(), + "item-insert failure during append-back must propagate as an error, not a silent success" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn on_response_body_tolerates_message_cache_failure() { + let filter = build_failing_filter(FailingItemStore { + fail_create_items: false, + fail_message_sync: true, + }); + + let req = make_request(Method::POST, "/v1/responses"); + let mut ctx = make_filter_context(&req); + ctx.current_filter_id = Some(0); + set_append_back_metadata(&mut ctx); + ctx.extensions.insert(ResponsesState { + input: vec![serde_json::json!({"type": "message", "role": "user", "content": "hello from append"})], + ..ResponsesState::default() + }); + + let mut resp = make_response(); + resp.headers + .insert(http::header::CONTENT_TYPE, "application/json".parse().unwrap()); + ctx.response_header = Some(&mut resp); + drop(filter.on_response(&mut ctx).await.unwrap()); + + // Items committed, but the denormalized message-cache refresh failed. The + // cache is a self-healing projection of the durable items table, so this is + // not data loss: failing the turn would drive a client retry that re-appends + // the same items as duplicates. The turn must succeed (Continue); the cache + // re-syncs on a later successful cache-refresh/sync attempt — a later append is + // not sufficient, since its own refresh may also fail (#837, durability boundary). + let response_json = serde_json::json!({ + "status": "completed", + "output": [{"type": "message", "role": "assistant", "content": "hi from model"}] + }); + let mut body = Some(Bytes::from(serde_json::to_vec(&response_json).unwrap())); + let action = filter + .on_response_body(&mut ctx, &mut body, true) + .expect("message-cache refresh failure must not fail the turn"); + + assert!( + matches!(action, FilterAction::Continue), + "message-cache refresh failure must be tolerated (Continue), not surfaced as an error" + ); +} + #[test] fn conformance_conversations_routes_match_runtime_registry() { let spec = generated_openapi_spec(); @@ -3530,6 +3614,132 @@ async fn create_test_conversation(filter: &dyn HttpFilter, metadata: Value) -> S resp["id"].as_str().unwrap().to_owned() } +/// Build a conversations filter backed by a fault-injecting store. +fn build_failing_filter(store: FailingItemStore) -> OpenaiConversationsFilter { + let yaml: serde_yaml::Value = serde_yaml::from_str( + r#" + backend: sqlite + database_url: "sqlite::memory:" + conversations_table: test_conversations + items_table: test_items + "#, + ) + .unwrap(); + let cfg: ConversationsConfig = parse_filter_config("openai_conversations", &yaml).unwrap(); + OpenaiConversationsFilter::with_store_for_test(cfg, Arc::new(store)) +} + +/// A [`ConversationItemStore`] that fails selected operations on demand. +/// +/// Benign methods return empty/default results; the two boolean knobs force the +/// item-insert and message-cache-sync paths to error so append-back failure +/// handling can be tested without a real database. +struct FailingItemStore { + /// Force `create_conversation_items` to return a database error. + fail_create_items: bool, + /// Force the message-cache sync (`update_conversation_messages`) to error. + fail_message_sync: bool, +} + +#[async_trait::async_trait] +impl ConversationItemStore for FailingItemStore { + async fn upsert_conversation(&self, _record: &ConversationRecord) -> Result<(), StoreError> { + Ok(()) + } + + async fn update_conversation_messages( + &self, + _tenant_id: &str, + _conversation_id: &str, + _messages: &Value, + ) -> Result { + if self.fail_message_sync { + return Err(StoreError::Database("mock message sync failure".to_owned())); + } + Ok(true) + } + + async fn compare_and_swap_conversation_messages( + &self, + _tenant_id: &str, + _conversation_id: &str, + _expected_messages: &Value, + _messages: &Value, + ) -> Result { + Ok(true) + } + + async fn get_conversation( + &self, + _tenant_id: &str, + _conversation_id: &str, + ) -> Result, StoreError> { + Ok(None) + } + + async fn delete_conversation(&self, _tenant_id: &str, _conversation_id: &str) -> Result { + Ok(false) + } + + async fn create_conversation_items(&self, _items: &[ConversationItemRecord]) -> Result<(), StoreError> { + if self.fail_create_items { + return Err(StoreError::Database("mock item insert failure".to_owned())); + } + Ok(()) + } + + async fn list_conversation_items( + &self, + _tenant_id: &str, + _conversation_id: &str, + _after_item_id: Option<&str>, + _limit: u32, + _ascending: bool, + ) -> Result, StoreError> { + Ok(Vec::new()) + } + + async fn get_existing_conversation_item_ids( + &self, + _tenant_id: &str, + _conversation_id: &str, + _item_ids: &[&str], + ) -> Result, StoreError> { + Ok(Vec::new()) + } + + async fn get_conversation_item( + &self, + _tenant_id: &str, + _conversation_id: &str, + _item_id: &str, + ) -> Result, StoreError> { + Ok(None) + } + + async fn delete_conversation_item( + &self, + _tenant_id: &str, + _conversation_id: &str, + _item_id: &str, + ) -> Result { + Ok(false) + } + + async fn conversation_item_position( + &self, + _tenant_id: &str, + _conversation_id: &str, + _item_id: &str, + ) -> Result, StoreError> { + Ok(None) + } + + async fn max_item_position(&self, _tenant_id: &str, _conversation_id: &str) -> Result { + Ok(0) + } +} + #[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)] struct OperationKey { method: String,