fix(store): acknowledge analytics versions after projection commits (#404)

The analytics worker acknowledged projected queue versions inside its
repeatable-read projection transaction. Request updates keep rewriting
pending queue rows through the source triggers, and once such a change
commits after the worker snapshot, the acknowledgement row lock fails with
a serialization error instead of skipping the stale version. Under steady
traffic every cycle rolled back the whole projection batch, the queue kept
growing and the worker stayed in backoff.

Commit the projection and publication first, then acknowledge the exact
versions in a separate read-committed transaction. Row locks there
re-check the latest committed version, so rewritten versions no longer
match and stay queued for replay, while locked rows are still skipped.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

backend/crates/gateway-store/src/postgres/observability/analytics/worker.rs

@@ -97,11 +97,16 @@ impl PgRequestAnalyticsRepository {
             if result.projected_requests + result.backfilled_requests > 0 {
                 publish_generation(&mut transaction, true).await?;
             }
+            transaction.commit().await.map_err(|error| postgres_unavailable("publish analytics maintenance", error))?;
+            // 快照后提交的源版本会让 RR 下的行锁直接序列化失败，持续写入时整批投影会反复回滚
+            // ACK 在投影提交后使用 READ COMMITTED，行锁按最新提交版本重新比较，已改写的版本不再匹配
+            let mut transaction = connection.begin().await.map_err(|error| postgres_unavailable("begin analytics acknowledgement", error))?;
+            sqlx::query("set local statement_timeout = '10s'").execute(&mut *transaction).await.map_err(|error| postgres_unavailable("set analytics query deadline", error))?;
             // 源删除会持有队列行再修改 generation，ACK 不能等待这些源事务形成反向锁依赖
             // 被跳过的版本保留在持久队列中重放，只有精确匹配且已取得行锁的版本才确认
             sqlx::query("with acknowledged as materialized (select pending.request_id,pending.change_version from request_analytics_pending_requests pending join unnest($1::text[],$2::bigint[]) job(id,version) on pending.request_id=job.id and pending.change_version=job.version for update of pending skip locked) delete from request_analytics_pending_requests pending using acknowledged where pending.request_id=acknowledged.request_id and pending.change_version=acknowledged.change_version")
                 .bind(ids).bind(versions).execute(&mut *transaction).await.map_err(|error| postgres_unavailable("confirm analytics request versions", error))?;
-            transaction.commit().await.map_err(|error| postgres_unavailable("publish analytics maintenance", error))?;
+            transaction.commit().await.map_err(|error| postgres_unavailable("complete analytics acknowledgement", error))?;
             // 每个阶段使用独立短事务，聚合只锁自己的队列，不把读放大传递到请求终结
             let mut transaction = begin_snapshot(&mut connection).await?;
             let acquired: bool = sqlx::query_scalar("select pg_try_advisory_xact_lock(hashtextextended(current_schema()::text,382032))")

backend/crates/gateway-store/tests/postgres/analytics.rs

@@ -897,7 +897,7 @@ async fn worker_skips_locked_versions_and_preserves_a_concurrent_source_deletion
 }
 
 #[tokio::test]
-async fn committed_queue_versions_abort_old_ack_and_replay_without_losing_contributions() {
+async fn committed_queue_versions_skip_old_ack_and_replay_without_losing_contributions() {
     use sqlx::Connection;
     for backfill in [false, true] {
         let Some(database) = TestDatabase::create("analytics_committed_ack").await else {
@@ -948,32 +948,15 @@ async fn committed_queue_versions_abort_old_ack_and_replay_without_losing_contri
             .execute(&mut publication)
             .await
             .unwrap();
-        let error = running
+        // 已提交的新版本不能让整批投影回滚，旧版本 ACK 不匹配并保留给下一轮重放
+        let maintenance = running
             .await
             .unwrap()
-            .expect_err("RR cannot acknowledge a queue row changed after its snapshot");
-        let gateway_store::StoreError::Unavailable {
-            message,
-            source: Some(source),
-            ..
-        } = error
-        else {
-            panic!("expected a PostgreSQL serialization failure");
-        };
-        assert_eq!(message, "confirm analytics request versions");
-        assert_eq!(
-            source
-                .downcast_ref::<sqlx::Error>()
-                .expect("preserve original sqlx error")
-                .as_database_error()
-                .expect("preserve PostgreSQL error")
-                .code()
-                .as_deref(),
-            Some("40001")
-        );
+            .expect("committed queue versions cannot abort the projected batch");
+        assert_eq!(maintenance.projected_requests, 1);
         let preserved: (i64, i64) = sqlx::query_as("select pending.change_version,facts.total_tokens from request_analytics_pending_requests pending join request_analytics_facts facts on facts.id=pending.request_id where pending.request_id='committed_ack'")
             .fetch_one(&database.pool).await.unwrap();
-        assert_eq!(preserved, (committed_version, 120));
+        assert_eq!(preserved, (committed_version, 240));
         drain_analytics(&database.pool).await;
         let summary = observability_repository(&database.pool)
             .usage_summary(