fix(store): persist continuation affinity before delivery

Remove the asynchronous continuation queue so Core waits for the bounded Redis write before exposing a reusable response ID. Keep Redis index maintenance bounded per record to protect terminal response latency.

Co-authored-by: Codex <noreply@openai.com>

backend/crates/gateway-store/src/lib.rs

@@ -332,6 +332,7 @@ pub async fn initialize(mut config: StoreConfig) -> StoreResult<StoreBundle> {
             REDIS_NAMESPACE,
             redis::ProviderCircuitPolicy::default(),
         )?);
+    // Continuation affinity 是下一轮请求的路由事实，Core 必须直接等待 Redis 确认。
     let continuation: Arc<dyn gateway_core::engine::continuation::NativeContinuationPort> =
         Arc::new(redis::RedisNativeContinuationRepository::new(
             redis_connection.clone(),
@@ -340,8 +341,6 @@ pub async fn initialize(mut config: StoreConfig) -> StoreResult<StoreBundle> {
     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<StoreBundle> {
             )),
         ),
         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<StoreBundle> {
         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<postgres::PgExecutionStore>,
     admission_release_writer: redis::ClientAdmissionReleaseWriter,
     circuit_feedback_writer: redis::ProviderCircuitFeedbackWriter,
-    continuation_writer: redis::NativeContinuationWriter,
     retention: Arc<postgres::PgRetentionRepository>,
 ) -> StoreResult<Vec<WorkerContribution>> {
     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)?,
-        ),
     ])
 }
 

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<dyn NativeContinuationPort>,
-    sender: mpsc::Sender<NativeContinuationPin>,
-}
-
-impl BufferedNativeContinuationPort {
-    #[must_use]
-    pub fn new(inner: Arc<dyn NativeContinuationPort>) -> (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<dyn NativeContinuationPort>,
-        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<Option<NativeContinuationPin>, 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<dyn NativeContinuationPort>,
-    receiver: Mutex<mpsc::Receiver<NativeContinuationPin>>,
-}
-
-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 后台写入失败");
-                }
-            }
-        })
-    }
-}

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::<i64>(&mut connection)
                 .await
                 .map_err(|_| NativeContinuationStoreError)?;

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<Option<NativeContinuationPin>, 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<T>(
@@ -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"),
-    )
-}

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::<i64>(&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::<usize>(&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<String> {
     redis::cmd("KEYS")
         .arg(format!("{namespace}:*"))