From 2122dc9469646352e1495d1809f95ec5a78b8f9d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E7=8E=8B=E5=BE=B7=E5=AE=87?= <kvtodev@outlook.com>
Date: Sat, 3 Oct 2026 19:28:39 +0800
Subject: [PATCH] =?UTF-8?q?=E5=90=8E=E7=AB=AF=EF=BC=9A=E5=85=B3=E5=81=9C?=
 =?UTF-8?q?=20flush=20=E4=B8=8D=E5=86=8D=E5=81=87=E6=89=BF=E8=AF=BA?=
 =?UTF-8?q?=E9=87=8D=E8=AF=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- game_play_counter_worker.rs：flush_deltas 增加 requeue_on_build；worker 循环仍把 Build 失败放回下一轮，关停路径改为直接记丢失量，不再把增量放回一个不会再被 flush 的缓冲
---
 .../src/game_play_counter_worker.rs           | 22 +++++++++++++++----
 1 file changed, 18 insertions(+), 4 deletions(-)

diff --git a/server-rs/crates/api-server/src/game_play_counter_worker.rs b/server-rs/crates/api-server/src/game_play_counter_worker.rs
index 6df0607e6..f740a1fab 100644
--- a/server-rs/crates/api-server/src/game_play_counter_worker.rs
+++ b/server-rs/crates/api-server/src/game_play_counter_worker.rs
@@ -28,7 +28,7 @@ pub(crate) fn spawn_game_play_counter_worker(state: AppState) {
             let counter = state.game_play_counter();
             counter.prune_expired(now);
             if let Some(deltas) = counter.take_pending_if_due(now) {
-                flush_deltas(&state, deltas).await;
+                flush_deltas(&state, deltas, true).await;
             }
         }
     });
@@ -40,16 +40,20 @@ pub(crate) async fn flush_game_play_counter_for_shutdown(state: &AppState) {
     if deltas.is_empty() {
         return;
     }
-    flush_deltas(state, deltas).await;
+    flush_deltas(state, deltas, false).await;
 }
 
-async fn flush_deltas(state: &AppState, deltas: Vec<GamePlayCountDelta>) {
+/// 把一批增量写库。
+///
+/// `requeue_on_build` 控制 `Build`（确定未发出）失败时是否放回：worker 循环为 `true`，等待下一轮；
+/// 关停路径为 `false`——`finalize_shutdown` 之后不再重试，放回只会随进程退出丢失，应直接记丢失量。
+async fn flush_deltas(state: &AppState, deltas: Vec<GamePlayCountDelta>, requeue_on_build: bool) {
     let mut accepted = 0usize;
     let mut dropped = 0u64;
     for chunk in deltas.chunks(GAME_PLAY_COUNT_FLUSH_BATCH_SIZE) {
         match write_batch(state, chunk).await {
             Ok(()) => accepted += chunk.len(),
-            Err(SpacetimeClientError::Build(message)) => {
+            Err(SpacetimeClientError::Build(message)) if requeue_on_build => {
                 state.game_play_counter().requeue(chunk);
                 warn!(
                     error = %message,
@@ -57,6 +61,16 @@ async fn flush_deltas(state: &AppState, deltas: Vec<GamePlayCountDelta>) {
                     "游戏游玩计数写入连接未建立，已放回等待下一轮"
                 );
             }
+            Err(SpacetimeClientError::Build(message)) => {
+                let lost = chunk.iter().map(|delta| delta.delta).sum::<u64>();
+                dropped = dropped.saturating_add(lost);
+                warn!(
+                    error = %message,
+                    games = chunk.len(),
+                    lost,
+                    "游戏游玩计数关停 flush 连接未建立，未落库增量丢弃"
+                );
+            }
             Err(error) => {
                 let lost = chunk.iter().map(|delta| delta.delta).sum::<u64>();
                 dropped = dropped.saturating_add(lost);