t195：失敗重試加指數退避＋上限——一個壞檔不再拖住整個資料夾
leo 2026-08-05：「有封測者在等，一直出錯」「已經等了幾天了」 ## 病（leo 實撞，log 實證） `小果被AFTEE詐貸.pdf` 雲端 401 失敗後，**manifest 完全不記失敗** ⇒ 下輪掃描又當「新檔」⇒ **1387 輪、跨 11 小時**，每輪 3.2~3.8 秒全在撞同一面牆； 且它排在佇列前面 ⇒ **整個資料夾的同步被一個壞檔拖住**。 leo：「原先萃檔案速度也快，現在花了十幾分才萃完」——**萃取沒變慢，慢的是重試** （log 的 `FOREACH 所有 5 項目均失敗` 證明卡早就萃好了）。 **這是獨立於 401 的架構缺陷**：401 修好了，下次換別的錯照樣卡死。 ## 修 · ManifestEntry 加 FailCount / LastFailAt / NextRetry · MarkFailed()：退避階梯 1m→5m→15m→1h→6h（之後維持 6h） · ShouldRetry()：退避窗口內跳過；連續失敗 8 次暫停自動重試 force（使用者按「立刻同步」）忽略退避與上限——**人明確要求不該被機器擋住** · MarkIngestedBy() 成功時清空失敗狀態（下次再壞從第一階重算） · direct.go 三個失敗出口都記退避（讀檔失敗／萃取失敗／上傳失敗） · retrySkipReason()：跳過時說人話，不靜默（同 t195 燈號誠實原則） ## 🔴 真兇其實有兩層——第二層才是關鍵 只加退避欄位**沒有用**：`scan.go` 每輪都**重建** ManifestEntry， 原本只 carry IngestedHash/IngestedAt ⇒ 我寫進去的 fail_count 下一輪就被抹掉 ⇒ 退避永遠停在「第 1 次失敗」＝等同沒有退避。 （順帶發現 ExtractedBy(t73「誰萃的」)原本也一直悄悄丟失。） ⇒ scan.go carry 補齊四個欄位，並留註解：**日後新增跨輪欄位必須加在這裡**。 ## 驗（真實跑，非只有單元測試） 單元測試 6 項全過（退避窗口／指數遞增 60/300/900/3600/21600/21600／ 上限停止／force 忽略／成功清空／壞檔不連累同輪其他檔） 實跑（cypher_url 指向不存在主機製造必失敗）： 第 1 輪 failed 第 2 輪 skipped「上次失敗（第 1 次），1m0s 後重試」 第 3 輪 skipped（同上） 第 4 輪 skipped（同上） 模擬退避到期 → 確實重送、fail_count=2、退避升為 300s ✅ go test ./... 全綠；go vet 通過。 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -82,6 +82,11 @@ type DirectConfig struct {
|
||||
CardIngestWF string `json:"card_ingest_workflow,omitempty"` // 收卡 workflow（空＝rag_ingest_card）
|
||||
PollSec int `json:"poll_interval_sec"` // 輪詢間隔秒（空/0＝5）
|
||||
MaxRemoved float64 `json:"max_removed_ratio"` // 大量刪除防呆門檻（空/0＝0.4）
|
||||
|
||||
// ForceSync＝這一輪是使用者按「立刻同步」觸發的（t195）。
|
||||
// 為真時忽略失敗退避與次數上限，一律重送——**人明確要求時不該被機器的退避擋住**。
|
||||
// 不寫進 config.json（`json:"-"`）：它是單次執行旗標，不是使用者設定。
|
||||
ForceSync bool `json:"-"`
|
||||
}
|
||||
|
||||
// librarySlug 把資料夾名轉成合法庫名（A-Za-z0-9_-；中文等非 ASCII 轉為底線分段）。
|
||||
@@ -681,6 +686,23 @@ func saveDirectConfig(configPath string, cfg *DirectConfig) error {
|
||||
}
|
||||
|
||||
// runDirectOnceRoot 對單一根掃一輪、直送 added/modified/renamed、下架 removed，2xx 後回寫該根 manifest。
|
||||
// retrySkipReason 產生「為什麼這輪跳過」的人話（t195）。
|
||||
// 靜默跳過會讓使用者以為檔案被忽略了——狀態要說得出理由（t195 燈號誠實原則同源）。
|
||||
func retrySkipReason(m *Manifest, path string, now int64) string {
|
||||
e, ok := m.Entries[path]
|
||||
if !ok {
|
||||
return "暫時跳過"
|
||||
}
|
||||
if e.FailCount >= MaxFailBeforeSkip {
|
||||
return fmt.Sprintf("連續失敗 %d 次，已暫停自動重試（改檔或按「立刻同步」會再試）", e.FailCount)
|
||||
}
|
||||
wait := e.NextRetry - now
|
||||
if wait < 0 {
|
||||
wait = 0
|
||||
}
|
||||
return fmt.Sprintf("上次失敗（第 %d 次），%s 後重試", e.FailCount, (time.Duration(wait) * time.Second).String())
|
||||
}
|
||||
|
||||
func runDirectOnceRoot(cfg *DirectConfig, root string, dryRun bool) ([]DirectResult, int, *TriggerPayload) {
|
||||
results := []DirectResult{}
|
||||
exit := 0
|
||||
@@ -719,10 +741,24 @@ func runDirectOnceRoot(cfg *DirectConfig, root string, dryRun bool) ([]DirectRes
|
||||
// renamed 在 direct 模式視同 added：內容未變但為求 kbdb 有這頁名的卡，重送一次萃取
|
||||
//（頁名可能改變＝要新頁名的卡）。冪等由 kbdb 端承擔（同頁名覆蓋語意）。
|
||||
res := DirectResult{Type: ev.Type, Path: ev.Path}
|
||||
// 🔴 t195 止血點：這個檔剛失敗過且還在退避窗口內 → 這輪跳過。
|
||||
// 沒有這道閘時的實測災情：`小果被AFTEE詐貸.pdf` 因雲端 401 失敗，
|
||||
// 每輪重掃又被當成新檔 ⇒ **1387 輪、跨 11 小時**，且它排在佇列前面，
|
||||
// **整個資料夾的同步被一個壞檔拖住**（leo：「原先萃檔案速度也快，
|
||||
// 現在也花了十幾分才萃完」——萃取沒變慢，慢的是重試）。
|
||||
// 退避階梯 1m→5m→15m→1h→6h；連續失敗 8 次後暫停自動重試。
|
||||
// 使用者改檔（hash 變）或按「立刻同步」時仍會重試，不會永久卡死。
|
||||
if !m.ShouldRetry(ev.Path, now, cfg.ForceSync) {
|
||||
res.Status = "skipped"
|
||||
res.Error = retrySkipReason(m, ev.Path, now)
|
||||
results = append(results, res)
|
||||
continue
|
||||
}
|
||||
full := filepath.Join(absRoot, filepath.FromSlash(ev.Path))
|
||||
content, rerr := os.ReadFile(full)
|
||||
if rerr != nil {
|
||||
res.Status, res.Error = "failed", "讀檔失敗："+rerr.Error()
|
||||
m.MarkFailed(ev.Path, now) // t195：讀不到的檔也退避（權限／被鎖／壞掉的外接碟）
|
||||
results = append(results, res)
|
||||
exit = 1
|
||||
continue
|
||||
@@ -756,6 +792,9 @@ func runDirectOnceRoot(cfg *DirectConfig, root string, dryRun bool) ([]DirectRes
|
||||
}
|
||||
if xerr != nil {
|
||||
res.Status, res.Error = "failed", "本地萃取失敗："+xerr.Error()
|
||||
// t195：萃取階段失敗同樣要記退避。**這條路徑比上傳更早**，
|
||||
// 漏記的話（連不上知識庫、金鑰壞、模型錯）照樣每輪重撞。
|
||||
m.MarkFailed(ev.Path, now)
|
||||
results = append(results, res)
|
||||
exit = 1
|
||||
continue
|
||||
@@ -789,6 +828,9 @@ func runDirectOnceRoot(cfg *DirectConfig, root string, dryRun bool) ([]DirectRes
|
||||
// 記下是誰萃的（t73/leo 07-27）：換萃取器時才分辨得出哪些卡是舊的。
|
||||
m.MarkIngestedBy(ev.Path, ev.SourceHash, now, cfg.Extractor)
|
||||
} else {
|
||||
// t195：記下失敗並排定退避，否則下輪又把它當新檔重試
|
||||
//（實撞：1387 輪 × 11 小時全在撞同一面 401 的牆，還拖住整個佇列）。
|
||||
m.MarkFailed(ev.Path, now)
|
||||
exit = 1
|
||||
}
|
||||
results = append(results, res)
|
||||
|
||||
+69
@@ -9,6 +9,7 @@ import (
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ManifestEntry 是單一檔案在 manifest 裡的狀態。
|
||||
@@ -32,8 +33,31 @@ type ManifestEntry struct {
|
||||
// 本欄位**不改變任何現有行為**（純記錄），但它是「換萃取器要不要重萃」這個決定的前提：
|
||||
// 沒有它，之後想分辨就永遠分辨不了（舊資料補不回來）。
|
||||
ExtractedBy string `json:"extracted_by,omitempty"`
|
||||
|
||||
// ── 失敗退避（t195，2026-08-05）────────────────────────────────────────────
|
||||
// 病（leo 實撞）：`小果被AFTEE詐貸.pdf` 因雲端 401 失敗後，**manifest 完全不記失敗**
|
||||
// ⇒ 下一輪掃描又把它當「新檔」⇒ 無退避、無上限地重試。
|
||||
// 實測：**1387 輪、跨 11 小時**，每輪 3.2~3.8 秒全花在撞同一面牆上；
|
||||
// 而且它排在佇列前面 ⇒ **一個壞檔就把整個資料夾的同步拖住**。
|
||||
// leo：「不只推上去慢，原先萃檔案速度也快，現在也花了十幾分才萃完」——
|
||||
// **萃取根本沒變慢，慢的是重試**（log 的 `FOREACH 所有 5 項目均失敗` 證明卡早就萃好了）。
|
||||
//
|
||||
// 解：記下失敗次數與下次可重試時間，指數退避＋次數上限。
|
||||
// 這是**獨立於當次錯誤**的架構缺陷——401 修好了，下次換別的錯照樣卡死，所以要修這裡。
|
||||
FailCount int `json:"fail_count,omitempty"` // 連續失敗次數（成功即歸零）
|
||||
LastFailAt int64 `json:"last_fail_at,omitempty"` // 最後一次失敗的 unix 秒
|
||||
NextRetry int64 `json:"next_retry,omitempty"` // 早於這個時間不重試（0＝可立即重試）
|
||||
}
|
||||
|
||||
// retryBackoff 退避階梯：1m → 5m → 15m → 1h → 6h，之後每次 6h。
|
||||
var retryBackoff = []time.Duration{
|
||||
1 * time.Minute, 5 * time.Minute, 15 * time.Minute, 1 * time.Hour, 6 * time.Hour,
|
||||
}
|
||||
|
||||
// MaxFailBeforeSkip 連續失敗達此次數 → 該檔暫停自動重試，不再拖住佇列。
|
||||
// 使用者改檔（content hash 變）或按「立刻同步」時仍會重試（見 MarkFailed／ShouldRetry）。
|
||||
const MaxFailBeforeSkip = 8
|
||||
|
||||
// Manifest 對應一個被勾選的資料夾。
|
||||
type Manifest struct {
|
||||
FolderID string `json:"folder_id"`
|
||||
@@ -107,9 +131,54 @@ func (m *Manifest) MarkIngestedBy(path, sourceHash string, at int64, extractor s
|
||||
e.IngestedHash = sourceHash
|
||||
e.IngestedAt = at
|
||||
e.ExtractedBy = extractor
|
||||
// 成功即清掉失敗狀態（t195）：下次再壞會從第一階退避重新算起。
|
||||
e.FailCount, e.LastFailAt, e.NextRetry = 0, 0, 0
|
||||
return true
|
||||
}
|
||||
|
||||
// MarkFailed 記錄一次失敗並排定下次可重試時間（t195 指數退避）。
|
||||
//
|
||||
// 為什麼要記在 manifest 而不是記憶體：collector 每輪是獨立 process
|
||||
// （`direct --once` 由看守器反覆拉起），記憶體狀態一輪就沒了——
|
||||
// 這正是原本「1387 輪重試同一個檔」的原因：每輪都以為自己是第一次。
|
||||
func (m *Manifest) MarkFailed(path string, at int64) bool {
|
||||
e, ok := m.Entries[path]
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
e.FailCount++
|
||||
e.LastFailAt = at
|
||||
idx := e.FailCount - 1
|
||||
if idx >= len(retryBackoff) {
|
||||
idx = len(retryBackoff) - 1
|
||||
}
|
||||
e.NextRetry = at + int64(retryBackoff[idx].Seconds())
|
||||
return true
|
||||
}
|
||||
|
||||
// ShouldRetry 回報「這個檔現在該不該送」。
|
||||
//
|
||||
// - 從沒失敗過 → true（行為與 t195 前一字不變）
|
||||
// - 還在退避窗口內 → false（**這是止血點：壞檔不再每 4 秒撞一次牆拖住整個佇列**）
|
||||
// - 連續失敗 ≥ MaxFailBeforeSkip → false（暫停自動重試）
|
||||
//
|
||||
// force＝使用者按「立刻同步」：忽略退避與上限一律重送
|
||||
// （人明確要求時不該被機器的退避擋住）。
|
||||
// 另：使用者改檔會讓 content hash 變 → 走的是「內容變更」路徑，本函式不介入。
|
||||
func (m *Manifest) ShouldRetry(path string, now int64, force bool) bool {
|
||||
e, ok := m.Entries[path]
|
||||
if !ok || e.FailCount == 0 {
|
||||
return true
|
||||
}
|
||||
if force {
|
||||
return true
|
||||
}
|
||||
if e.FailCount >= MaxFailBeforeSkip {
|
||||
return false
|
||||
}
|
||||
return now >= e.NextRetry
|
||||
}
|
||||
|
||||
// Save 原子寫入（temp + rename），避免掃描中斷留半個 JSON。
|
||||
func (m *Manifest) Save(path string) error {
|
||||
data, err := json.MarshalIndent(m, "", " ")
|
||||
|
||||
@@ -0,0 +1,106 @@
|
||||
// manifest_retry_test.go — t195 失敗退避（leo 2026-08-05：「有封測者在等，一直出錯」）。
|
||||
//
|
||||
// 這組測試守的是一個**架構缺陷**，不是某次的 401：
|
||||
// 原本 manifest 只記成功不記失敗 ⇒ 每輪重掃都把失敗檔當「新檔」
|
||||
// ⇒ 實測 1387 輪 × 11 小時撞同一面牆，且壞檔排在前面**拖住整個資料夾**。
|
||||
// 401 修好了，下次換別的錯照樣卡死——所以要測的是「退避本身」。
|
||||
package main
|
||||
|
||||
import "testing"
|
||||
|
||||
func newRetryTestManifest(path string) *Manifest {
|
||||
return &Manifest{
|
||||
FolderID: "t", Root: "/tmp/t",
|
||||
Entries: map[string]*ManifestEntry{path: {ContentHash: "h1", Size: 1, Mtime: 1}},
|
||||
}
|
||||
}
|
||||
|
||||
// 沒失敗過的檔行為與 t195 前一字不變。
|
||||
func TestShouldRetry_NeverFailed(t *testing.T) {
|
||||
m := newRetryTestManifest("a.pdf")
|
||||
if !m.ShouldRetry("a.pdf", 1000, false) {
|
||||
t.Fatal("沒失敗過的檔應該要送")
|
||||
}
|
||||
// manifest 裡沒有的檔（全新檔）也一樣要送
|
||||
if !m.ShouldRetry("never-seen.pdf", 1000, false) {
|
||||
t.Fatal("全新檔應該要送")
|
||||
}
|
||||
}
|
||||
|
||||
// 核心止血：失敗後在退避窗口內不再重試。
|
||||
func TestShouldRetry_BackoffWindow(t *testing.T) {
|
||||
m := newRetryTestManifest("a.pdf")
|
||||
m.MarkFailed("a.pdf", 1000) // 第 1 次失敗 → 退避 60s
|
||||
|
||||
if m.ShouldRetry("a.pdf", 1030, false) {
|
||||
t.Fatal("退避窗口內（+30s）不該重試——這正是 1387 輪的病根")
|
||||
}
|
||||
if !m.ShouldRetry("a.pdf", 1060, false) {
|
||||
t.Fatal("退避結束（+60s）應該要重試")
|
||||
}
|
||||
}
|
||||
|
||||
// 退避要遞增，不是固定間隔（否則壞檔仍會高頻撞牆）。
|
||||
func TestMarkFailed_ExponentialBackoff(t *testing.T) {
|
||||
m := newRetryTestManifest("a.pdf")
|
||||
want := []int64{60, 300, 900, 3600, 21600, 21600} // 1m,5m,15m,1h,6h,之後維持 6h
|
||||
at := int64(1000)
|
||||
for i, w := range want {
|
||||
m.MarkFailed("a.pdf", at)
|
||||
got := m.Entries["a.pdf"].NextRetry - at
|
||||
if got != w {
|
||||
t.Fatalf("第 %d 次失敗：退避 %ds，預期 %ds", i+1, got, w)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 連續失敗達上限 → 暫停自動重試（不再永遠佔用每一輪）。
|
||||
func TestShouldRetry_MaxFailStops(t *testing.T) {
|
||||
m := newRetryTestManifest("a.pdf")
|
||||
at := int64(1000)
|
||||
for i := 0; i < MaxFailBeforeSkip; i++ {
|
||||
m.MarkFailed("a.pdf", at)
|
||||
}
|
||||
// 就算等再久也不自動重試
|
||||
if m.ShouldRetry("a.pdf", at+999999, false) {
|
||||
t.Fatalf("連續失敗 %d 次後應暫停自動重試", MaxFailBeforeSkip)
|
||||
}
|
||||
// 但使用者按「立刻同步」仍要送——人明確要求不該被機器退避擋住
|
||||
if !m.ShouldRetry("a.pdf", at+1, true) {
|
||||
t.Fatal("force（立刻同步）應忽略退避與上限")
|
||||
}
|
||||
}
|
||||
|
||||
// 成功後要清掉失敗狀態，否則下次再壞會從高階退避起跳（等太久）。
|
||||
func TestMarkIngested_ResetsFailState(t *testing.T) {
|
||||
m := newRetryTestManifest("a.pdf")
|
||||
m.MarkFailed("a.pdf", 1000)
|
||||
m.MarkFailed("a.pdf", 1100)
|
||||
if m.Entries["a.pdf"].FailCount != 2 {
|
||||
t.Fatal("失敗次數應累計")
|
||||
}
|
||||
|
||||
m.MarkIngestedBy("a.pdf", "h1", 1200, "gemma")
|
||||
e := m.Entries["a.pdf"]
|
||||
if e.FailCount != 0 || e.NextRetry != 0 || e.LastFailAt != 0 {
|
||||
t.Fatalf("成功後應清空失敗狀態，got FailCount=%d NextRetry=%d", e.FailCount, e.NextRetry)
|
||||
}
|
||||
// 清空後立刻可再送
|
||||
if !m.ShouldRetry("a.pdf", 1201, false) {
|
||||
t.Fatal("成功後應恢復可送")
|
||||
}
|
||||
}
|
||||
|
||||
// 壞檔不該擋住同一輪的其他檔（leo 實撞：整個資料夾被一個 PDF 拖住）。
|
||||
func TestShouldRetry_OtherFilesUnaffected(t *testing.T) {
|
||||
m := newRetryTestManifest("bad.pdf")
|
||||
m.Entries["good.md"] = &ManifestEntry{ContentHash: "h2", Size: 1, Mtime: 1}
|
||||
m.MarkFailed("bad.pdf", 1000)
|
||||
|
||||
if m.ShouldRetry("bad.pdf", 1010, false) {
|
||||
t.Fatal("壞檔應在退避中")
|
||||
}
|
||||
if !m.ShouldRetry("good.md", 1010, false) {
|
||||
t.Fatal("同一輪的其他檔不該被壞檔連累——這是「拖住整個佇列」的正解")
|
||||
}
|
||||
}
|
||||
@@ -276,6 +276,15 @@ func Scan(root string, m *Manifest, opts ScanOptions) (*TriggerPayload, error) {
|
||||
if carry != nil {
|
||||
ne.IngestedHash = carry.IngestedHash
|
||||
ne.IngestedAt = carry.IngestedAt
|
||||
// 🔴 t195：掃描每輪都**重建** entry，原本只 carry 上面兩欄 ⇒ 其餘欄位靜默歸零。
|
||||
// 實撞：失敗退避（fail_count/next_retry）寫進去了，下一輪掃描卻被抹掉
|
||||
// ⇒ 退避永遠停在「第 1 次失敗」，等同沒有退避（1387 輪的病根之一）。
|
||||
// ExtractedBy（t73 記的「誰萃的」）原本也一樣悄悄丟失。
|
||||
// ⚠️ 之後在 ManifestEntry 新增任何「跨輪要記住」的欄位，都必須加在這裡。
|
||||
ne.ExtractedBy = carry.ExtractedBy
|
||||
ne.FailCount = carry.FailCount
|
||||
ne.LastFailAt = carry.LastFailAt
|
||||
ne.NextRetry = carry.NextRetry
|
||||
}
|
||||
newEntries[p] = ne
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user