530 lines
20 KiB
Go
530 lines
20 KiB
Go
// direct_pacing_test.go — 上傳節奏／新檔優先／單輪上限／斷點續傳（2026-08-07）。
|
||
//
|
||
// 背景見 direct_pacing.go 檔頭：封測事故實測 1,070 次寫入撞上免費上限 1,000，
|
||
// 690 個檔逐檔萃取上傳、daemon 這一半完全沒有節奏。這裡驗四件事：
|
||
// 1. 一輪掃到的多個事件依 mtime 新到舊處理（今天寫的最優先）。
|
||
// 2. 兩次觸發雲端之間有節流間隔。
|
||
// 3. 單輪上限存在——巨量積壓不會一次湧完。
|
||
// 4. 已處理的事件會立刻落地 manifest，process 被殺掉重開也不會重做。
|
||
package collector
|
||
|
||
import (
|
||
"encoding/json"
|
||
"io"
|
||
"net/http"
|
||
"net/http/httptest"
|
||
"os"
|
||
"path/filepath"
|
||
"strings"
|
||
"sync"
|
||
"testing"
|
||
"time"
|
||
)
|
||
|
||
// init：測試環境預設把節流歸零，不然每個既有測試都要多等 700ms×N，拖慢整個套件。
|
||
// 需要驗證節流本身的測試（見下）自行 save/restore 成一個很小的非零值。
|
||
func init() {
|
||
directPaceInterval = 0
|
||
}
|
||
|
||
// ── 1) sortEventsNewestFirst ─────────────────────────────────────────────
|
||
|
||
func TestSortEventsNewestFirst_NewestGoesFirst(t *testing.T) {
|
||
root := t.TempDir()
|
||
old := baseTime
|
||
mid := baseTime.Add(time.Hour)
|
||
newest := baseTime.Add(2 * time.Hour)
|
||
writeFile(t, root, "old.md", "old", old)
|
||
writeFile(t, root, "mid.md", "mid", mid)
|
||
writeFile(t, root, "newest.md", "newest", newest)
|
||
|
||
events := []Event{
|
||
{Type: "added", Path: "old.md"},
|
||
{Type: "added", Path: "newest.md"},
|
||
{Type: "added", Path: "mid.md"},
|
||
}
|
||
got := sortEventsNewestFirst(root, events)
|
||
want := []string{"newest.md", "mid.md", "old.md"}
|
||
for i, w := range want {
|
||
if got[i].Path != w {
|
||
t.Fatalf("順序=%v，want %v", pathsOf(got), want)
|
||
}
|
||
}
|
||
}
|
||
|
||
func pathsOf(evs []Event) []string {
|
||
out := make([]string, len(evs))
|
||
for i, e := range evs {
|
||
out[i] = e.Path
|
||
}
|
||
return out
|
||
}
|
||
|
||
// removed 事件沒有 mtime 可排，一律排在 added/modified/renamed 之後（下架不急）。
|
||
func TestSortEventsNewestFirst_RemovedGoesLast(t *testing.T) {
|
||
root := t.TempDir()
|
||
writeFile(t, root, "new.md", "x", baseTime.Add(time.Hour))
|
||
events := []Event{
|
||
{Type: "removed", Path: "gone.md"},
|
||
{Type: "added", Path: "new.md"},
|
||
}
|
||
got := sortEventsNewestFirst(root, events)
|
||
if got[0].Path != "new.md" || got[1].Path != "gone.md" {
|
||
t.Fatalf("順序=%v", pathsOf(got))
|
||
}
|
||
}
|
||
|
||
// ── 2) 節流間隔 ────────────────────────────────────────────────────────────
|
||
|
||
func TestPace_RespectsInterval(t *testing.T) {
|
||
old := directPaceInterval
|
||
directPaceInterval = 30 * time.Millisecond
|
||
defer func() { directPaceInterval = old }()
|
||
|
||
start := time.Now()
|
||
pace()
|
||
pace()
|
||
elapsed := time.Since(start)
|
||
if elapsed < 55*time.Millisecond { // 兩次 pace，留一點餘裕
|
||
t.Fatalf("兩次 pace() 只花了 %v，節流間隔沒有生效", elapsed)
|
||
}
|
||
}
|
||
|
||
// ── 端到端：積壓一次湧上去 vs 有節奏 ─────────────────────────────────────────
|
||
//
|
||
// 模擬「大量積壓」場景：8 個檔、單輪上限設 3——驗證①一輪只處理 3 個
|
||
// ②被延後的檔案有交代（不是安靜消失）③下一輪接著處理下一批、依然新到舊。
|
||
|
||
func TestDirect_LargeBacklog_ProcessedInNewestFirstBatches(t *testing.T) {
|
||
root := t.TempDir()
|
||
// 8 個檔，mtime 依檔名反向遞增（a 最舊，h 最新）
|
||
names := []string{"a", "b", "c", "d", "e", "f", "g", "h"}
|
||
for i, n := range names {
|
||
writeFile(t, root, n+".md", "內容 "+n, baseTime.Add(time.Duration(i)*time.Minute))
|
||
}
|
||
|
||
var mu sync.Mutex
|
||
var order []string
|
||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||
// InkStoneCo#44：資料夾樹走 portal 登記端點，body 沒有 page_name。
|
||
// 不先擋掉，下面那行型別斷言會 panic，而測試會變成掛住 8 分鐘（見該 helper 的說明）。
|
||
if answeredFolderTreePost(w, r) {
|
||
return
|
||
}
|
||
body, _ := io.ReadAll(r.Body)
|
||
var m map[string]any
|
||
_ = json.Unmarshal(body, &m)
|
||
mu.Lock()
|
||
order = append(order, m["page_name"].(string))
|
||
mu.Unlock()
|
||
_ = json.NewEncoder(w).Encode(map[string]any{"success": true})
|
||
}))
|
||
defer srv.Close()
|
||
defer gemmaStub(t, func(w http.ResponseWriter, r *http.Request) {
|
||
// 這條測試只在意「觸發了幾次、依什麼順序送到假 cypher」，卡片內容固定即可。
|
||
_ = json.NewEncoder(w).Encode(map[string]any{
|
||
"candidates": []map[string]any{{
|
||
"content": map[string]any{"parts": []map[string]any{{"text": cardFixture("卡", "測試")}}},
|
||
}},
|
||
})
|
||
})()
|
||
|
||
cfg := &DirectConfig{
|
||
WatchFolders: []string{root},
|
||
Manifest: filepath.Join(t.TempDir(), "m.json"),
|
||
CypherURL: srv.URL, Namespace: "demo", APIKey: "demo",
|
||
Library: "kb", Extractor: "gemma", ExtractorExplicit: true, GeminiAPIKey: "k-test",
|
||
CardIngestWF: "rag_ingest_card", MaxRemoved: DefaultMaxRemovedRatio,
|
||
MaxEventsPerRun: 3,
|
||
}
|
||
|
||
// 第一輪：只處理 3 個（上限），且依 mtime 新到舊＝h, g, f
|
||
results1, exit1, _ := RunDirectOnce(cfg, false)
|
||
if exit1 != 0 {
|
||
t.Fatalf("exit=%d results=%+v", exit1, results1)
|
||
}
|
||
ingested1 := ingestedPaths(results1)
|
||
if len(ingested1) != 3 {
|
||
t.Fatalf("第一輪應處理 3 個，got %d: %+v", len(ingested1), results1)
|
||
}
|
||
wantFirst := []string{"h.md", "g.md", "f.md"}
|
||
for i, w := range wantFirst {
|
||
if ingested1[i] != w {
|
||
t.Fatalf("第一輪順序=%v，want %v（新改的檔要優先）", ingested1, wantFirst)
|
||
}
|
||
}
|
||
// 有交代被延後了幾筆，不是安靜消失
|
||
if !hasDeferredNotice(results1) {
|
||
t.Errorf("應該要交代還有事件被延後，results=%+v", results1)
|
||
}
|
||
|
||
// 第二輪：接著處理下一批 3 個（e, d, c）
|
||
results2, exit2, _ := RunDirectOnce(cfg, false)
|
||
if exit2 != 0 {
|
||
t.Fatalf("exit=%d results=%+v", exit2, results2)
|
||
}
|
||
ingested2 := ingestedPaths(results2)
|
||
wantSecond := []string{"e.md", "d.md", "c.md"}
|
||
if len(ingested2) != 3 {
|
||
t.Fatalf("第二輪應處理 3 個，got %d: %+v", len(ingested2), results2)
|
||
}
|
||
for i, w := range wantSecond {
|
||
if ingested2[i] != w {
|
||
t.Fatalf("第二輪順序=%v，want %v", ingested2, wantSecond)
|
||
}
|
||
}
|
||
|
||
// 第三輪：剩下 2 個（b, a）全部處理完，沒有再延後
|
||
results3, exit3, _ := RunDirectOnce(cfg, false)
|
||
if exit3 != 0 {
|
||
t.Fatalf("exit=%d", exit3)
|
||
}
|
||
ingested3 := ingestedPaths(results3)
|
||
if len(ingested3) != 2 || ingested3[0] != "b.md" || ingested3[1] != "a.md" {
|
||
t.Fatalf("第三輪=%v，want [b.md a.md]", ingested3)
|
||
}
|
||
if hasDeferredNotice(results3) {
|
||
t.Error("全部處理完不該再有延後交代")
|
||
}
|
||
|
||
// 全部 8 個都真的送到雲端了（沒有一個被永久漏掉）；總覽卡（結構先行）另計
|
||
mu.Lock()
|
||
defer mu.Unlock()
|
||
var contentCards []string
|
||
for _, pn := range order {
|
||
if !strings.HasPrefix(pn, "資料夾總覽") {
|
||
contentCards = append(contentCards, pn)
|
||
}
|
||
}
|
||
if len(contentCards) != 8 {
|
||
t.Fatalf("雲端總共應收到 8 次卡片，got %d: %v", len(contentCards), order)
|
||
}
|
||
}
|
||
|
||
// ── 4) 退避中的檔案不該永久佔滿單輪名額（arcrun-rag#104 comment 4480，t217）──
|
||
//
|
||
// 背景：leo 實測 leo21c 帳號的積壓「1691→1880→1936 筆從不減少」，
|
||
// 「已經好久沒有加過任何檔案，哪來的這些筆數？那就是之前卡住的，就是你要解決的問題」
|
||
// 「佇列就是問題本身」。
|
||
//
|
||
// 根因：舊版把 perRunCap 套在「排序後的原始清單」上。mtime 最新的幾個檔如果持續
|
||
// 失敗（進入退避），它們不會因為在退避就往後排，於是每一輪都繼續佔著最前面的
|
||
// perRunCap 個名額——即使這一輪根本不會被嘗試，只是被跳過。排在它們後面、
|
||
// 從沒被嘗試過的健康檔案因此永遠排不到，不管跑幾輪都一樣。
|
||
//
|
||
// 這個測試重現該情境：3 個 mtime 最新的檔一直失敗（模擬持續性錯誤，例如票上量到的
|
||
// 雲端 subrequest 上限或萃取回傳格式錯誤），5 個 mtime較舊、原本會成功的健康檔案
|
||
// 排在它們後面。單輪上限=3。
|
||
//
|
||
// - 第一輪：全部檔案都還沒失敗過（FailCount=0），cap 選中 mtime 最新的 3 個
|
||
// （也就是那 3 個會一直失敗的檔），全部失敗，記下退避（下次重試在 60 秒後）。
|
||
// - 第二輪（緊接著呼叫，真實時間遠不到 60 秒）：那 3 個檔仍在退避中。
|
||
// 舊版行為：cap 依然套在原始排序上，選中的還是同一批退避中的檔案 ⇒
|
||
// 這一輪 0 個健康檔案被嘗試，健康檔案永遠排不到。
|
||
// 修好後的行為：退避中的檔案被分流到 waiting、不佔 ready 的名額，
|
||
// cap 改套用在 ready 上 ⇒ 健康檔案的前 3 名遞補上來，這一輪就會被嘗試並成功。
|
||
func TestDirect_StarvedBacklog_HealthyFilesEventuallyGetATurn(t *testing.T) {
|
||
root := t.TempDir()
|
||
|
||
// 3 個「一直失敗」的檔，mtime 最新（若 bug 還在，會永久佔滿 cap）。
|
||
blockers := []string{"blocker-a", "blocker-b", "blocker-c"}
|
||
for i, n := range blockers {
|
||
writeFile(t, root, n+".md", "持續失敗的內容 "+n, baseTime.Add(time.Duration(10+i)*time.Minute))
|
||
}
|
||
// 5 個「健康」的檔，mtime 較舊（排在後面，理應遞補上來）。
|
||
healthy := []string{"h5", "h4", "h3", "h2", "h1"}
|
||
for i, n := range healthy {
|
||
writeFile(t, root, n+".md", "健康內容 "+n, baseTime.Add(time.Duration(4-i)*time.Minute))
|
||
}
|
||
|
||
failingPages := map[string]bool{"blocker-a": true, "blocker-b": true, "blocker-c": true}
|
||
var mu sync.Mutex
|
||
var succeededPages []string
|
||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||
if answeredFolderTreePost(w, r) {
|
||
return
|
||
}
|
||
body, _ := io.ReadAll(r.Body)
|
||
var m map[string]any
|
||
_ = json.Unmarshal(body, &m)
|
||
pageName, _ := m["page_name"].(string)
|
||
if strings.HasPrefix(pageName, "資料夾總覽") {
|
||
_ = json.NewEncoder(w).Encode(map[string]any{"success": true})
|
||
return
|
||
}
|
||
if failingPages[pageName] {
|
||
w.WriteHeader(http.StatusInternalServerError)
|
||
_, _ = w.Write([]byte(`{"success":false,"error":"boom"}`))
|
||
return
|
||
}
|
||
mu.Lock()
|
||
succeededPages = append(succeededPages, pageName)
|
||
mu.Unlock()
|
||
_ = json.NewEncoder(w).Encode(map[string]any{"success": true})
|
||
}))
|
||
defer srv.Close()
|
||
defer gemmaStub(t, func(w http.ResponseWriter, r *http.Request) {
|
||
_ = json.NewEncoder(w).Encode(map[string]any{
|
||
"candidates": []map[string]any{{
|
||
"content": map[string]any{"parts": []map[string]any{{"text": cardFixture("卡", "測試")}}},
|
||
}},
|
||
})
|
||
})()
|
||
|
||
cfg := &DirectConfig{
|
||
WatchFolders: []string{root},
|
||
Manifest: filepath.Join(t.TempDir(), "m.json"),
|
||
CypherURL: srv.URL, Namespace: "demo", APIKey: "demo",
|
||
Library: "kb", Extractor: "gemma", ExtractorExplicit: true, GeminiAPIKey: "k-test",
|
||
CardIngestWF: "rag_ingest_card", MaxRemoved: DefaultMaxRemovedRatio,
|
||
MaxEventsPerRun: 3,
|
||
}
|
||
|
||
// 第一輪：8 個檔都還沒失敗過，cap 選中 mtime 最新的 3 個（blocker-a/b/c），全部失敗。
|
||
results1, _, _ := RunDirectOnce(cfg, false)
|
||
if got := ingestedPaths(results1); len(got) != 0 {
|
||
t.Fatalf("第一輪不該有任何成功（cap 選中的 3 個全會失敗），got %v", got)
|
||
}
|
||
var failedCount int
|
||
for _, r := range results1 {
|
||
if r.Status == "failed" {
|
||
failedCount++
|
||
}
|
||
}
|
||
if failedCount != 3 {
|
||
t.Fatalf("第一輪應該有 3 筆真的被嘗試且失敗（blocker-a/b/c），got %d：%+v", failedCount, results1)
|
||
}
|
||
|
||
// 第二輪：緊接著呼叫（真實時間遠不到 60 秒退避窗口）。blocker-a/b/c 仍在退避中。
|
||
results2, _, _ := RunDirectOnce(cfg, false)
|
||
ingested2 := ingestedPaths(results2)
|
||
if len(ingested2) != 3 {
|
||
t.Fatalf("🔴 第二輪應該有 3 個健康檔案遞補上來被嘗試並成功——"+
|
||
"如果這裡是 0，代表退避中的 blocker-a/b/c 又佔滿了本輪名額，"+
|
||
"健康檔案永遠排不到（這正是 leo 實測「佇列從不減少」的那個 bug）。got %d：%+v",
|
||
len(ingested2), results2)
|
||
}
|
||
wantSecond := []string{"h5.md", "h4.md", "h3.md"}
|
||
for i, w := range wantSecond {
|
||
if ingested2[i] != w {
|
||
t.Fatalf("第二輪順序=%v，want %v（健康檔案仍照 mtime 新到舊遞補）", ingested2, wantSecond)
|
||
}
|
||
}
|
||
// blocker-a/b/c 這一輪不該再被真的嘗試（還在退避中）——它們只會以「skipped」出現。
|
||
for _, r := range results2 {
|
||
if r.Path == "blocker-a.md" || r.Path == "blocker-b.md" || r.Path == "blocker-c.md" {
|
||
if r.Status != "skipped" {
|
||
t.Fatalf("退避中的 %s 這一輪不該被真的嘗試，got status=%s", r.Path, r.Status)
|
||
}
|
||
}
|
||
}
|
||
|
||
mu.Lock()
|
||
defer mu.Unlock()
|
||
if len(succeededPages) != 3 {
|
||
t.Fatalf("雲端應該收到 3 筆健康卡片，got %d: %v", len(succeededPages), succeededPages)
|
||
}
|
||
}
|
||
|
||
func ingestedPaths(results []DirectResult) []string {
|
||
var out []string
|
||
for _, r := range results {
|
||
// 只數「檔案事件」。總覽卡（inventory）與資料夾樹（folder_tree，InkStoneCo#44）
|
||
// 都是每輪的結構回報，不是使用者的檔案——同 countsAsDocument 的那條線。
|
||
if r.Status == "ingested" && r.Type != "inventory" && r.Type != "folder_tree" {
|
||
out = append(out, r.Path)
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
func hasDeferredNotice(results []DirectResult) bool {
|
||
for _, r := range results {
|
||
if r.Type == "info" && strings.Contains(r.Error, "已排入佇列") {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
// ── 3) 斷點續傳：process 被殺掉重開，不從頭來 ────────────────────────────────
|
||
//
|
||
// 用「單輪上限=1」模擬中斷：一次只做一件事就等同「這一刻的程序被殺掉」，
|
||
// 用全新的 RunDirectOnce 呼叫（不共用任何記憶體狀態，manifest 完全從磁碟重讀）
|
||
// 模擬「程序重開」。驗證：已經成功的不會被重送，且每一步都真的落地磁碟
|
||
// （不是等到全部做完才存檔）。
|
||
func TestDirect_ResumeAfterInterruption(t *testing.T) {
|
||
root := t.TempDir()
|
||
writeFile(t, root, "one.md", "內容一", baseTime)
|
||
writeFile(t, root, "two.md", "內容二", baseTime.Add(time.Minute))
|
||
writeFile(t, root, "three.md", "內容三", baseTime.Add(2*time.Minute))
|
||
|
||
var mu sync.Mutex
|
||
var posted []string
|
||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||
// InkStoneCo#44：資料夾樹的 body 沒有 path ⇒ 不擋掉的話會被記成一筆空字串，
|
||
// 本測就會看到「送了 4 次」而誤判斷點續傳失效。
|
||
if answeredFolderTreePost(w, r) {
|
||
return
|
||
}
|
||
body, _ := io.ReadAll(r.Body)
|
||
var m map[string]any
|
||
_ = json.Unmarshal(body, &m)
|
||
mu.Lock()
|
||
if p, _ := m["path"].(string); !strings.HasPrefix(p, ".arcrun-rag/") {
|
||
posted = append(posted, p) // 結構先行：總覽卡（合成路徑）另計
|
||
}
|
||
mu.Unlock()
|
||
_ = json.NewEncoder(w).Encode(map[string]any{"success": true})
|
||
}))
|
||
defer srv.Close()
|
||
defer gemmaStub(t, func(w http.ResponseWriter, r *http.Request) {
|
||
_ = json.NewEncoder(w).Encode(map[string]any{
|
||
"candidates": []map[string]any{{
|
||
"content": map[string]any{"parts": []map[string]any{{"text": cardFixture("卡", "測試")}}},
|
||
}},
|
||
})
|
||
})()
|
||
|
||
manifestPath := filepath.Join(t.TempDir(), "m.json")
|
||
newCfg := func() *DirectConfig {
|
||
return &DirectConfig{
|
||
WatchFolders: []string{root},
|
||
Manifest: manifestPath,
|
||
CypherURL: srv.URL, Namespace: "demo", APIKey: "demo",
|
||
Library: "kb", Extractor: "gemma", ExtractorExplicit: true, GeminiAPIKey: "k-test",
|
||
CardIngestWF: "rag_ingest_card", MaxRemoved: DefaultMaxRemovedRatio,
|
||
MaxEventsPerRun: 1, // 模擬「只做一件事就被打斷」
|
||
}
|
||
}
|
||
|
||
// 「跑到一半」：只完成第一批（three.md，最新）
|
||
if _, exit, _ := RunDirectOnce(newCfg(), false); exit != 0 {
|
||
t.Fatal("第一批失敗")
|
||
}
|
||
|
||
// 直接讀磁碟上的 manifest（不透過任何記憶體物件）——驗證真的已經落地，
|
||
// 不是要等三批都跑完才存檔。
|
||
absRoot, _ := filepath.Abs(root)
|
||
mp := newCfg().manifestPathFor(absRoot)
|
||
m1, err := LoadManifest(mp, absRoot)
|
||
if err != nil {
|
||
t.Fatalf("讀 manifest 失敗：%v", err)
|
||
}
|
||
if m1.Entries["three.md"].IngestedHash == "" {
|
||
t.Fatal("「跑到一半」之後，已完成的那份應該已經落地 manifest（斷點續傳的前提）")
|
||
}
|
||
if m1.Entries["two.md"].IngestedHash != "" || m1.Entries["one.md"].IngestedHash != "" {
|
||
t.Fatal("還沒輪到的不該被誤標成已完成")
|
||
}
|
||
|
||
// 「重開程序」：全新呼叫，模擬 process 重啟——應該接著做 two.md，不是重做 three.md
|
||
if _, exit, _ := RunDirectOnce(newCfg(), false); exit != 0 {
|
||
t.Fatal("第二批失敗")
|
||
}
|
||
if _, exit, _ := RunDirectOnce(newCfg(), false); exit != 0 {
|
||
t.Fatal("第三批失敗")
|
||
}
|
||
|
||
mu.Lock()
|
||
defer mu.Unlock()
|
||
if len(posted) != 3 {
|
||
t.Fatalf("三個檔應該總共只被送出 3 次（不重送已完成的），got %d: %v", len(posted), posted)
|
||
}
|
||
seen := map[string]bool{}
|
||
for _, p := range posted {
|
||
if seen[p] {
|
||
t.Fatalf("%q 被重複送出——斷點續傳失效，重開後從頭來了", p)
|
||
}
|
||
seen[p] = true
|
||
}
|
||
|
||
// 第四輪：三個都做完了，不該再有任何事件
|
||
results4, _, _ := RunDirectOnce(newCfg(), false)
|
||
if len(results4) != 0 {
|
||
t.Fatalf("全部完成後應該零事件：%+v", results4)
|
||
}
|
||
}
|
||
|
||
// removed 事件在「已放回但下架失敗」時不能被提早存檔清掉——否則下一輪兩邊都找不到
|
||
// 這個路徑，永遠不會再重試下架。驗證：POST 失敗時，manifest 磁碟版本仍保留該筆，
|
||
// 下一輪會重新產生 removed 事件。
|
||
func TestDirect_RemovedRetriesOnFailureEvenWithIncrementalSave(t *testing.T) {
|
||
root := t.TempDir()
|
||
writeFile(t, root, "keep.md", "留著", baseTime)
|
||
writeFile(t, root, "gone.md", "要刪的", baseTime.Add(time.Minute))
|
||
|
||
var failRemoved = true
|
||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||
if strings.Contains(r.URL.Path, "rag_takedown_direct") && failRemoved {
|
||
w.WriteHeader(http.StatusInternalServerError)
|
||
_, _ = w.Write([]byte(`{"error":"boom"}`))
|
||
return
|
||
}
|
||
_ = json.NewEncoder(w).Encode(map[string]any{"success": true})
|
||
}))
|
||
defer srv.Close()
|
||
defer gemmaStub(t, func(w http.ResponseWriter, r *http.Request) {
|
||
_ = json.NewEncoder(w).Encode(map[string]any{
|
||
"candidates": []map[string]any{{
|
||
"content": map[string]any{"parts": []map[string]any{{"text": cardFixture("卡", "測試")}}},
|
||
}},
|
||
})
|
||
})()
|
||
|
||
cfg := &DirectConfig{
|
||
WatchFolders: []string{root},
|
||
Manifest: filepath.Join(t.TempDir(), "m.json"),
|
||
CypherURL: srv.URL, Namespace: "demo", APIKey: "demo",
|
||
Library: "kb", Extractor: "gemma", ExtractorExplicit: true, GeminiAPIKey: "k-test",
|
||
CardIngestWF: "rag_ingest_card", RemovedWF: "rag_takedown_direct",
|
||
MaxRemoved: 1.0,
|
||
}
|
||
if _, exit, _ := RunDirectOnce(cfg, false); exit != 0 {
|
||
t.Fatal("第一輪（兩份都新增）失敗")
|
||
}
|
||
if err := os.Remove(filepath.Join(root, "gone.md")); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
|
||
// 第二輪：下架 POST 會失敗（且這輪也會處理 keep.md 的事件？不會，keep.md 已 ingest 過、
|
||
// 沒有變化，不會再產生事件——但仍會呼叫 saveManifest() 若有其他觸發）。
|
||
results, exit, _ := RunDirectOnce(cfg, false)
|
||
if exit != 1 {
|
||
t.Fatalf("下架失敗應該 exit=1，got %d results=%+v", exit, results)
|
||
}
|
||
|
||
absRoot, _ := filepath.Abs(root)
|
||
mp := cfg.manifestPathFor(absRoot)
|
||
m, err := LoadManifest(mp, absRoot)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, ok := m.Entries["gone.md"]; !ok {
|
||
t.Fatal("下架失敗時，manifest 不該把這個路徑永久丟掉——否則永遠不會再重試下架")
|
||
}
|
||
|
||
// 第三輪：換伺服器成功 → 應該重新補發並成功下架
|
||
failRemoved = false
|
||
results3, exit3, _ := RunDirectOnce(cfg, false)
|
||
if exit3 != 0 {
|
||
t.Fatalf("第三輪應成功下架：exit=%d results=%+v", exit3, results3)
|
||
}
|
||
var removedOK bool
|
||
for _, r := range results3 {
|
||
if r.Type == "removed" && r.Status == "removed" && r.Path == "gone.md" {
|
||
removedOK = true
|
||
}
|
||
}
|
||
if !removedOK {
|
||
t.Fatalf("第三輪應該重試下架成功：%+v", results3)
|
||
}
|
||
m2, err := LoadManifest(mp, absRoot)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, ok := m2.Entries["gone.md"]; ok {
|
||
t.Fatal("下架成功後這個路徑應該真的從 manifest 消失")
|
||
}
|
||
}
|