diff --git a/backend/crates/gateway-store/src/lib.rs b/backend/crates/gateway-store/src/lib.rs index 36a6cb1c1..858259945 100644 --- a/backend/crates/gateway-store/src/lib.rs +++ b/backend/crates/gateway-store/src/lib.rs @@ -332,6 +332,7 @@ pub async fn initialize(mut config: StoreConfig) -> StoreResult { REDIS_NAMESPACE, redis::ProviderCircuitPolicy::default(), )?); + // Continuation affinity 是下一轮请求的路由事实，Core 必须直接等待 Redis 确认。 let continuation: Arc = Arc::new(redis::RedisNativeContinuationRepository::new( redis_connection.clone(), @@ -340,8 +341,6 @@ pub async fn initialize(mut config: StoreConfig) -> StoreResult { let (admissions, admission_release_writer) = redis::BufferedClientAdmissionPort::new(admissions); let (circuits, circuit_feedback_writer) = redis::BufferedProviderCircuitPort::new(circuits); - let (continuation, continuation_writer) = - redis::BufferedNativeContinuationPort::new(continuation); let core_ports = CoreStorePorts::new( execution, ( @@ -351,7 +350,7 @@ pub async fn initialize(mut config: StoreConfig) -> StoreResult { )), ), Arc::new(circuits), - Arc::new(continuation), + continuation, ( Arc::new(postgres::PgRuntimeSnapshotRepository::new(pool.clone())), Arc::new(redis::RedisRuntimeChangeRepository::new( @@ -390,7 +389,6 @@ pub async fn initialize(mut config: StoreConfig) -> StoreResult { execution_writer, admission_release_writer, circuit_feedback_writer, - continuation_writer, retention, )?; Ok(StoreBundle { @@ -408,7 +406,6 @@ fn store_worker_contributions( execution_writer: postgres::ExecutionObservationWriter, admission_release_writer: redis::ClientAdmissionReleaseWriter, circuit_feedback_writer: redis::ProviderCircuitFeedbackWriter, - continuation_writer: redis::NativeContinuationWriter, retention: Arc, ) -> StoreResult> { let stale_id = WorkerId::try_new(WorkerKind::StaleModelRequestRecovery, "postgres") @@ -421,8 +418,6 @@ fn store_worker_contributions( .map_err(worker_definition_error)?; let circuit_flush_id = WorkerId::try_new(WorkerKind::OpsFlush, "redis_circuit") .map_err(worker_definition_error)?; - let continuation_flush_id = WorkerId::try_new(WorkerKind::OpsFlush, "redis_continuation") - .map_err(worker_definition_error)?; let ops_flush_restart = DaemonRestartPolicy::try_new(Duration::from_secs(1), Duration::from_secs(60)) .map_err(worker_definition_error)?; @@ -467,16 +462,6 @@ fn store_worker_contributions( ) .map_err(worker_definition_error)?, ), - WorkerContribution::Registration( - WorkerRegistration::try_new( - continuation_flush_id, - WorkerRunnable::Daemon { - restart: ops_flush_restart, - task: Box::new(continuation_writer), - }, - ) - .map_err(worker_definition_error)?, - ), ]) } diff --git a/backend/crates/gateway-store/src/redis/coordination_buffer.rs b/backend/crates/gateway-store/src/redis/coordination_buffer.rs index 3bc9574b2..9fbce86ef 100644 --- a/backend/crates/gateway-store/src/redis/coordination_buffer.rs +++ b/backend/crates/gateway-store/src/redis/coordination_buffer.rs @@ -1,4 +1,7 @@ -//! 数据面 Redis 写副作用的有界非阻塞队列。 +//! 可由 TTL 或后续反馈收敛的 Redis 写副作用队列。 +//! +//! Continuation affinity 决定下一轮 Provider 与账号，必须在响应 ID 可复用前 +//! 获得 Redis 确认，因此不属于本模块的可丢失副作用。 use std::num::NonZeroUsize; use std::sync::Arc; @@ -8,9 +11,6 @@ use gateway_core::engine::admission::{ ClientAdmissionDecision, ClientAdmissionError, ClientAdmissionPort, ClientAdmissionRecovery, ClientAdmissionRequest, ClientAdmissionRestoreResult, }; -use gateway_core::engine::continuation::{ - NativeContinuationPin, NativeContinuationPort, NativeContinuationStoreError, PreviousResponseId, -}; use gateway_core::engine::execution::{ ProviderCircuitDecision, ProviderCircuitError, ProviderCircuitPort, }; @@ -252,94 +252,3 @@ impl DaemonTask for ProviderCircuitFeedbackWriter { }) } } - -/// Continuation resolve 仍直读 Redis；新 pin 只做有界入队。 -#[derive(Clone)] -pub struct BufferedNativeContinuationPort { - inner: Arc, - sender: mpsc::Sender, -} - -impl BufferedNativeContinuationPort { - #[must_use] - pub fn new(inner: Arc) -> (Self, NativeContinuationWriter) { - Self::with_capacity( - inner, - NonZeroUsize::new(DEFAULT_QUEUE_CAPACITY).expect("queue capacity is non-zero"), - ) - } - - #[must_use] - pub fn with_capacity( - inner: Arc, - capacity: NonZeroUsize, - ) -> (Self, NativeContinuationWriter) { - let (sender, receiver) = mpsc::channel(capacity.get()); - ( - Self { - inner: Arc::clone(&inner), - sender, - }, - NativeContinuationWriter { - inner, - receiver: Mutex::new(receiver), - }, - ) - } - - fn enqueue(&self, pin: NativeContinuationPin) { - if let Err(error) = self.sender.try_send(pin) { - let reason = match error { - mpsc::error::TrySendError::Full(_) => "full", - mpsc::error::TrySendError::Closed(_) => "closed", - }; - tracing::warn!( - operation = "record_native_continuation", - reason, - "Redis 协调队列不可用，已丢弃本次 continuation affinity" - ); - } - } -} - -impl NativeContinuationPort for BufferedNativeContinuationPort { - fn resolve<'a>( - &'a self, - previous_response_id: &'a PreviousResponseId, - ) -> BoxFuture<'a, Result, NativeContinuationStoreError>> { - self.inner.resolve(previous_response_id) - } - - fn record<'a>( - &'a self, - pin: NativeContinuationPin, - ) -> BoxFuture<'a, Result<(), NativeContinuationStoreError>> { - self.enqueue(pin); - Box::pin(ready(Ok(()))) - } -} - -pub struct NativeContinuationWriter { - inner: Arc, - receiver: Mutex>, -} - -impl DaemonTask for NativeContinuationWriter { - fn run(&self, cancellation: CancellationToken) -> BoxFuture<'_, Result<(), WorkerTaskError>> { - Box::pin(async move { - let mut receiver = self.receiver.lock().await; - loop { - let pin = tokio::select! { - () = cancellation.cancelled() => return Ok(()), - pin = receiver.recv() => pin, - }; - let Some(pin) = pin else { - return Err(WorkerTaskError::safe("native continuation queue closed")); - }; - if let Err(error) = self.inner.record(pin).await { - tracing::warn!(%error, "Continuation affinity 后台写入失败"); - } - } - }) - } -} diff --git a/backend/crates/gateway-store/src/redis/native_continuation.rs b/backend/crates/gateway-store/src/redis/native_continuation.rs index b32864239..cfd9fc163 100644 --- a/backend/crates/gateway-store/src/redis/native_continuation.rs +++ b/backend/crates/gateway-store/src/redis/native_continuation.rs @@ -24,6 +24,7 @@ use super::{namespace, resource_fingerprint}; const CONTINUATION_TTL: Duration = Duration::from_secs(4 * 60 * 60); const MAX_CONTINUATIONS: usize = 65_536; const MAX_RECORD_BYTES: usize = 64 * 1024; +const MAX_INDEX_CLEANUP_PER_RECORD: usize = 64; const RECORD_SCRIPT: &str = r#" local entry_key = KEYS[1] @@ -35,12 +36,13 @@ local ttl_seconds = tonumber(ARGV[4]) local cutoff = tonumber(ARGV[5]) local max_entries = tonumber(ARGV[6]) local entry_prefix = ARGV[7] +local cleanup_limit = tonumber(ARGV[8]) redis.call('SET', entry_key, payload, 'EX', ttl_seconds) redis.call('ZADD', index_key, score, entry_id) redis.call('EXPIRE', index_key, ttl_seconds) -local stale = redis.call('ZRANGEBYSCORE', index_key, '-inf', cutoff) +local stale = redis.call('ZRANGEBYSCORE', index_key, '-inf', cutoff, 'LIMIT', 0, cleanup_limit) for _, stale_id in ipairs(stale) do redis.call('UNLINK', entry_prefix .. '{' .. stale_id .. '}') redis.call('ZREM', index_key, stale_id) @@ -48,6 +50,9 @@ end local count = redis.call('ZCARD', index_key) local excess = count - max_entries +if excess > cleanup_limit then + excess = cleanup_limit +end if excess > 0 then local oldest = redis.call('ZRANGE', index_key, 0, excess - 1) for _, old_id in ipairs(oldest) do @@ -158,6 +163,7 @@ impl NativeContinuationPort for RedisNativeContinuationRepository { .arg(cutoff) .arg(MAX_CONTINUATIONS) .arg(self.entry_prefix()) + .arg(MAX_INDEX_CLEANUP_PER_RECORD) .invoke_async::(&mut connection) .await .map_err(|_| NativeContinuationStoreError)?; diff --git a/backend/crates/gateway-store/tests/coordination_buffer.rs b/backend/crates/gateway-store/tests/coordination_buffer.rs index c5995e3ab..f3a137e3d 100644 --- a/backend/crates/gateway-store/tests/coordination_buffer.rs +++ b/backend/crates/gateway-store/tests/coordination_buffer.rs @@ -7,19 +7,13 @@ use gateway_core::engine::admission::{ ClientAdmissionDecision, ClientAdmissionError, ClientAdmissionPort, ClientAdmissionRecovery, ClientAdmissionRequest, ClientAdmissionRestoreResult, }; -use gateway_core::engine::continuation::{ - NativeContinuationPin, NativeContinuationPort, NativeContinuationStoreError, PreviousResponseId, -}; -use gateway_core::engine::credential::ProviderAccountId; use gateway_core::engine::execution::{ ProviderCircuitDecision, ProviderCircuitError, ProviderCircuitPort, }; use gateway_core::engine::{CancellationToken, ModelRequestId}; use gateway_core::policy::ClientApiKeyId; use gateway_core::routing::ProviderKind; -use gateway_store::redis::{ - BufferedClientAdmissionPort, BufferedNativeContinuationPort, BufferedProviderCircuitPort, -}; +use gateway_store::redis::{BufferedClientAdmissionPort, BufferedProviderCircuitPort}; #[derive(Default)] struct RecordingCoordination { @@ -98,27 +92,8 @@ impl ProviderCircuitPort for RecordingCoordination { } } -impl NativeContinuationPort for RecordingCoordination { - fn resolve<'a>( - &'a self, - _: &'a PreviousResponseId, - ) -> BoxFuture<'a, Result, NativeContinuationStoreError>> { - Box::pin(async { Ok(None) }) - } - - fn record<'a>( - &'a self, - _: NativeContinuationPin, - ) -> BoxFuture<'a, Result<(), NativeContinuationStoreError>> { - Box::pin(async move { - self.record("continuation"); - Ok(()) - }) - } -} - #[tokio::test] -async fn full_redis_coordination_queues_should_drop_writes_without_waiting() { +async fn full_recoverable_coordination_queues_should_drop_writes_without_waiting() { let inner = Arc::new(RecordingCoordination::default()); let (admissions, _admission_writer) = BufferedClientAdmissionPort::with_capacity( inner.clone(), @@ -128,10 +103,6 @@ async fn full_redis_coordination_queues_should_drop_writes_without_waiting() { inner.clone(), NonZeroUsize::new(1).expect("capacity"), ); - let (continuation, _continuation_writer) = BufferedNativeContinuationPort::with_capacity( - inner.clone(), - NonZeroUsize::new(1).expect("capacity"), - ); let client = ClientApiKeyId::new("key_buffer_test").expect("client key"); let request = ModelRequestId::new("req_buffer_test").expect("request ID"); let provider = ProviderKind::new("openai").expect("provider"); @@ -153,14 +124,6 @@ async fn full_redis_coordination_queues_should_drop_writes_without_waiting() { .observe_failure(&provider) .await .expect("full circuit queue remains fail-open"); - continuation - .record(continuation_pin()) - .await - .expect("first continuation enqueue"); - continuation - .record(continuation_pin()) - .await - .expect("full continuation queue remains fail-open"); }) .await .expect("coordination enqueue must never wait for Redis"); @@ -179,10 +142,6 @@ async fn redis_coordination_writers_should_flush_each_side_effect() { inner.clone(), NonZeroUsize::new(8).expect("capacity"), ); - let (continuation, continuation_writer) = BufferedNativeContinuationPort::with_capacity( - inner.clone(), - NonZeroUsize::new(8).expect("capacity"), - ); let client = ClientApiKeyId::new("key_writer_test").expect("client key"); let request = ModelRequestId::new("req_writer_test").expect("request ID"); let provider = ProviderKind::new("openai").expect("provider"); @@ -194,20 +153,14 @@ async fn redis_coordination_writers_should_flush_each_side_effect() { .observe_success(&provider) .await .expect("enqueue circuit feedback"); - continuation - .record(continuation_pin()) - .await - .expect("enqueue continuation"); - let cancellation = CancellationToken::new(); let tasks = [ spawn_writer(Arc::new(admission_writer), cancellation.clone()), spawn_writer(Arc::new(circuit_writer), cancellation.clone()), - spawn_writer(Arc::new(continuation_writer), cancellation.clone()), ]; tokio::time::timeout(Duration::from_secs(1), async { loop { - if inner.operations().len() == 3 { + if inner.operations().len() == 2 { break; } tokio::task::yield_now().await; @@ -224,7 +177,7 @@ async fn redis_coordination_writers_should_flush_each_side_effect() { let mut operations = inner.operations(); operations.sort_unstable(); - assert_eq!(operations, ["admission", "circuit_success", "continuation"]); + assert_eq!(operations, ["admission", "circuit_success"]); } fn spawn_writer( @@ -236,12 +189,3 @@ where { tokio::spawn(async move { writer.run(cancellation).await }) } - -fn continuation_pin() -> NativeContinuationPin { - NativeContinuationPin::new( - PreviousResponseId::new("resp_buffer_test"), - PreviousResponseId::new("resp_buffer_test"), - ProviderKind::new("openai").expect("provider"), - ProviderAccountId::new("acct_buffer_test").expect("account"), - ) -} diff --git a/backend/crates/gateway-store/tests/redis/native_continuation.rs b/backend/crates/gateway-store/tests/redis/native_continuation.rs index 006ebdd45..3a7f9d6c3 100644 --- a/backend/crates/gateway-store/tests/redis/native_continuation.rs +++ b/backend/crates/gateway-store/tests/redis/native_continuation.rs @@ -115,6 +115,51 @@ async fn native_continuation_rejects_provider_state_for_a_different_provider() { assert!(namespace_keys(&mut connection, &namespace).await.is_empty()); } +#[tokio::test] +async fn native_continuation_record_should_bound_index_cleanup_work() { + const STALE_ENTRY_COUNT: usize = 100; + const CLEANUP_BATCH: usize = 64; + + let Some((repository, mut connection, namespace)) = repository().await else { + return; + }; + repository + .record(pin("resp-cleanup-anchor")) + .await + .expect("record cleanup anchor"); + let index_key = namespace_keys(&mut connection, &namespace) + .await + .into_iter() + .find(|key| key.ends_with(":global:index")) + .expect("continuation index key"); + let mut seed = redis::cmd("ZADD"); + seed.arg(&index_key); + for index in 0..STALE_ENTRY_COUNT { + seed.arg(0).arg(format!("stale-{index}")); + } + assert_eq!( + seed.query_async::(&mut connection) + .await + .expect("seed stale continuation index entries"), + i64::try_from(STALE_ENTRY_COUNT).expect("stale entry count fits i64") + ); + + repository + .record(pin("resp-cleanup-trigger")) + .await + .expect("record continuation with bounded cleanup"); + + assert_eq!( + redis::cmd("ZCARD") + .arg(index_key) + .query_async::(&mut connection) + .await + .expect("count continuation index entries"), + STALE_ENTRY_COUNT + 2 - CLEANUP_BATCH + ); + delete_namespace_keys(&mut connection, &namespace).await; +} + async fn repository() -> Option<(RedisNativeContinuationRepository, ConnectionManager, String)> { let redis_url = crate::support::test_env("CPR_TEST_REDIS_URL")?; let client = redis::Client::open(redis_url).expect("valid CPR_TEST_REDIS_URL"); @@ -128,6 +173,15 @@ async fn repository() -> Option<(RedisNativeContinuationRepository, ConnectionMa Some((repository, connection, namespace)) } +fn pin(response_id: &str) -> NativeContinuationPin { + NativeContinuationPin::new( + PreviousResponseId::new(response_id), + PreviousResponseId::new(response_id), + ProviderKind::new("openai").expect("provider"), + ProviderAccountId::new("acct_primary").expect("account"), + ) +} + async fn namespace_keys(connection: &mut ConnectionManager, namespace: &str) -> Vec { redis::cmd("KEYS") .arg(format!("{namespace}:*"))