mirror of
https://github.com/zyycn/codex-proxy-rs.git
synced 2026-11-15 08:06:21 +00:00
fix(telemetry): preserve timed-out record writes
This commit is contained in:
@@ -25,6 +25,8 @@ pub struct Recorder {
|
||||
pub enum RecorderError {
|
||||
#[error("invalid success usage fact")]
|
||||
InvalidUsageFact,
|
||||
#[error("telemetry persistence task failed: {0}")]
|
||||
Task(#[from] tokio::task::JoinError),
|
||||
#[error(transparent)]
|
||||
Usage(#[from] PgUsageRecordStoreError),
|
||||
#[error(transparent)]
|
||||
@@ -62,13 +64,16 @@ impl Recorder {
|
||||
return Err(RecorderError::InvalidUsageFact);
|
||||
}
|
||||
apply_capture_body_policy(&mut event.metadata, self.capture_body);
|
||||
self.usage_records.append(&event).await?;
|
||||
let usage_records = self.usage_records.clone();
|
||||
// Controller timeouts must stop waiting without cancelling an in-flight fact write.
|
||||
tokio::spawn(async move { usage_records.append(&event).await }).await??;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn record_error(&self, mut event: OpsErrorLog) -> Result<(), RecorderError> {
|
||||
apply_capture_body_policy(&mut event.metadata, self.capture_body);
|
||||
self.ops_errors.append(&event).await?;
|
||||
let ops_errors = self.ops_errors.clone();
|
||||
tokio::spawn(async move { ops_errors.append(&event).await }).await??;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -730,6 +730,24 @@ async fn responses_stream_should_preserve_body_metadata_when_capture_body_enable
|
||||
|
||||
let (app, api_key, pool, _dir) =
|
||||
test_app_with_account_pool_and_telemetry_capture_body(server.uri()).await;
|
||||
// Force telemetry past the controller wait budget; persistence must outlive that timeout.
|
||||
sqlx::raw_sql(
|
||||
r#"
|
||||
create function delay_usage_record_insert() returns trigger as $$
|
||||
begin
|
||||
perform pg_sleep(0.2);
|
||||
return new;
|
||||
end;
|
||||
$$ language plpgsql;
|
||||
|
||||
create trigger delay_usage_record_insert
|
||||
before insert on usage_records
|
||||
for each row execute function delay_usage_record_insert();
|
||||
"#,
|
||||
)
|
||||
.execute(&pool)
|
||||
.await
|
||||
.unwrap();
|
||||
let response = app
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
|
||||
Reference in New Issue
Block a user