Merge pull request #180 from Autumn-27/fix/traffic-index-reclaim

fix(traffic): 删除流量后把索引空间真正还给操作系统

CHANGELOG.md

@@ -4,6 +4,14 @@
 
 ## [Unreleased]
 
+### 流量
+
+#### 修复的问题
+
+- **修复删除流量后磁盘空间不被释放**：删除只让数据不可见，空间一直留在索引文件里。SQLite 删行仅把页挂到 freelist，而索引库建库时没有启用 `auto_vacuum`，文件永不收缩；同时 `ex_fts` 是 `contentless_delete` 全文索引，`DELETE` 只写 tombstone 而不回收原 postings，不合并就永久累积——删流量反而让索引变大。由于 256KB 以下的正文全部内联在这个库里，加上 trigram 索引约为正文体积的 2 倍，抓包量大的实例会为早已删掉的流量长期占用数倍磁盘（实测抓 6MB 正文 → 索引 16MB，删光后仍是 16MB）。现在新建索引库直接启用 `auto_vacuum=incremental`，每次删除提交后在后台分块执行「全文索引增量合并 + `incremental_vacuum` + `wal_checkpoint(TRUNCATE)`」，逐步把空间还给操作系统（同一场景删除后回落到 104KB）。回收分块进行并在块间释放写锁，不阻塞流量录制；进程退出时立即让出，剩余工作在下次删除时续做。
+
+> 升级说明：`auto_vacuum` 只能在建库时设定，因此**已有实例的索引库仍是旧模式**，`incremental_vacuum` 在其上是空操作——启动时会打印一行提示。这类库升级后 tombstone 的持续增长已经止住（全文索引合并照常执行），但已占用的体积需要后续提供的「存储压缩」入口一次性回收。
+
 ### LLM
 
 #### 修复的问题

traffic/reclaim_test.go

@@ -0,0 +1,156 @@
+package traffic
+
+import (
+	"database/sql"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+// bulkRecord fills the index with inline bodies — the ones that actually make
+// index.sqlite grow. Binary content type keeps them out of the full-text index so
+// the test stays fast; the FTS side is covered by TestReclaimMergesFTSTombstones.
+func bulkRecord(tr *Traffic, host string, n, size int) {
+	body := []byte(strings.Repeat("A", size))
+	for i := 0; i < n; i++ {
+		tr.record(newFlow(host, "GET", fmt.Sprintf("/blob/%d", i), nil, body,
+			withRespType("application/octet-stream")))
+	}
+}
+
+func TestNewIndexEnablesIncrementalVacuum(t *testing.T) {
+	tr, _ := openTraffic(t)
+	if !tr.incrementalVacuum {
+		t.Fatal("新建索引库未启用增量回收")
+	}
+	var mode int
+	if err := tr.DB().QueryRow(`PRAGMA auto_vacuum`).Scan(&mode); err != nil {
+		t.Fatal(err)
+	}
+	if mode != autoVacuumIncremental {
+		t.Fatalf("auto_vacuum=%d，应为 %d", mode, autoVacuumIncremental)
+	}
+}
+
+// TestDeleteReclaimsIndexSpace is the regression: deleting traffic used to leave
+// index.sqlite at its high-water mark forever, because SQLite only chains freed
+// pages onto its freelist and nothing ever returned them to the filesystem.
+func TestDeleteReclaimsIndexSpace(t *testing.T) {
+	tr, _ := openTraffic(t)
+	const host = "bulk.example.com"
+	// 30 × 200KB stays under maxInlineBody, so every body lands in the database
+	// itself rather than the blob store — that is where the growth was invisible.
+	bulkRecord(tr, host, 30, 200*1024)
+	grown := tr.indexBytes()
+	if grown < 5<<20 {
+		t.Fatalf("索引只有 %d 字节，样本不足以验证回收", grown)
+	}
+
+	if n, err := tr.DeleteHostsExact([]string{host}); err != nil || n != 30 {
+		t.Fatalf("DeleteHostsExact=(%d,%v)，应为 (30,nil)", n, err)
+	}
+	tr.reaping.Wait() // 回收在后台分块进行
+
+	after := tr.indexBytes()
+	if after > grown/4 {
+		t.Fatalf("删除后索引仍占 %d 字节（删除前 %d），空间没有还给文件系统", after, grown)
+	}
+	// A handful of pages incremental_vacuum could not move to the end of the file
+	// is a normal residual; the ~1500 that the deletion freed must be gone.
+	var free int
+	if err := tr.DB().QueryRow(`PRAGMA freelist_count`).Scan(&free); err != nil {
+		t.Fatal(err)
+	}
+	if free > 64 {
+		t.Fatalf("仍有 %d 个空闲页未回收", free)
+	}
+}
+
+// TestReclaimMergesFTSTombstones covers the second half of the leak: ex_fts is a
+// contentless_delete index, so a DELETE only writes tombstones. Without a merge
+// the index keeps growing on every deletion — deleting traffic made it bigger.
+func TestReclaimMergesFTSTombstones(t *testing.T) {
+	tr, _ := openTraffic(t)
+	if !tr.fts {
+		t.Skip("驱动未启用 FTS5")
+	}
+	// Deleted in batches, which is what leaves tombstones spread over many
+	// segments rather than emptying the index in one shot.
+	for round := 0; round < 4; round++ {
+		host := fmt.Sprintf("fts%d.example.com", round)
+		for i := 0; i < 20; i++ {
+			tr.record(newFlow(host, "GET", fmt.Sprintf("/p/%d", i), nil,
+				[]byte(strings.Repeat("secret token 中文正文 padding ", 200))))
+		}
+		if _, err := tr.DeleteHostsExact([]string{host}); err != nil {
+			t.Fatal(err)
+		}
+		tr.reaping.Wait()
+	}
+
+	var exchanges, segments int
+	if err := tr.DB().QueryRow(`SELECT COUNT(*) FROM exchanges`).Scan(&exchanges); err != nil {
+		t.Fatal(err)
+	}
+	if err := tr.DB().QueryRow(`SELECT COUNT(*) FROM ex_fts_data`).Scan(&segments); err != nil {
+		t.Fatal(err)
+	}
+	if exchanges != 0 {
+		t.Fatalf("还剩 %d 条流量", exchanges)
+	}
+	// A fully merged, empty contentless index keeps only its structure rows.
+	if segments > 8 {
+		t.Fatalf("全文索引残留 %d 行段数据，tombstone 未被合并回收", segments)
+	}
+}
+
+// TestReclaimOnLegacyIndexIsHarmless covers installs created before
+// auto_vacuum=incremental became the default: incremental_vacuum is a silent
+// no-op there, so reclamation must report the situation and finish rather than
+// spin or fail. Only a full compaction can convert such a file.
+func TestReclaimOnLegacyIndexIsHarmless(t *testing.T) {
+	dir := t.TempDir()
+	if err := os.MkdirAll(filepath.Join(dir, "_index"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	// Create the tables first, with auto_vacuum left at its default 0 — exactly the
+	// shape Open used to leave behind.
+	legacy, err := sql.Open("sqlite", filepath.Join(dir, "_index", "index.sqlite"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := legacy.Exec(indexSchema); err != nil {
+		t.Fatal(err)
+	}
+	if err := legacy.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	tr, err := Open(dir, "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { tr.Close() })
+	if tr.incrementalVacuum {
+		t.Fatal("旧库不应报告已启用增量回收")
+	}
+
+	const host = "legacy.example.com"
+	bulkRecord(tr, host, 8, 200*1024)
+	if n, err := tr.DeleteHostsExact([]string{host}); err != nil || n != 8 {
+		t.Fatalf("DeleteHostsExact=(%d,%v)，应为 (8,nil)", n, err)
+	}
+	tr.reaping.Wait() // 必须收敛，不能卡在预算里
+
+	// The freelist stays populated: that is the whole reason a compaction entry
+	// point is needed for pre-existing databases.
+	var free int
+	if err := tr.DB().QueryRow(`PRAGMA freelist_count`).Scan(&free); err != nil {
+		t.Fatal(err)
+	}
+	if free == 0 {
+		t.Fatal("旧库居然回收了空闲页，说明测试没有真的构造出旧库")
+	}
+}

traffic/traffic.go

@@ -102,6 +102,30 @@ const (
 	// maxBlobRead caps one traffic_blob read so paging through a large body never
 	// floods the agent's context.
 	maxBlobRead = 8 * 1024
+	// autoVacuumIncremental is SQLite's numeric value for auto_vacuum=incremental.
+	autoVacuumIncremental = 2
+	// reclaimChunkPages bounds how much index space one locked step returns to the
+	// filesystem (pages are 4KiB, so ~32MB). Reclamation holds the write lock, and
+	// record() runs before go-mitmproxy replies to the client, so an unbounded
+	// pass would stall the very requests being recorded.
+	reclaimChunkPages = 8192
+	// reclaimMergePages bounds the full-text merge done alongside each chunk.
+	// Deleting from a contentless_delete index only writes tombstones; merging is
+	// what discards them, and left undone the index grows on every deletion.
+	reclaimMergePages = 256
+	// reclaimMergeSteps bounds how many merges one reclamation performs. fts5
+	// offers no way to ask whether an index has settled — its special INSERT
+	// registers a row change of its own, so change counting cannot tell a real
+	// merge from a no-op — so the budget is simply spent down, and whatever is
+	// left over falls to the next deletion.
+	reclaimMergeSteps = 16
+	// reclaimMaxSteps backstops the loop. Both merging and incremental_vacuum are
+	// documented to stop making progress eventually, but a reclamation that cannot
+	// converge must give up rather than spin holding the write lock.
+	reclaimMaxSteps = 512
+	// reclaimBudget caps one background reclamation. Deleting a busy host can free
+	// gigabytes, and the next deletion resumes wherever this one stopped.
+	reclaimBudget = 5 * time.Minute
 )
 
 // TrafficSearchDescription is persisted into the tool catalog for new and
@@ -121,9 +145,22 @@ type Traffic struct {
 	// build without FTS5: recording and metadata search still work, body search
 	// degrades to unsupported rather than erroring.
 	fts bool
-	// reaping tracks background reclamation of legacy trees, so shutdown and tests
-	// can wait for it instead of racing it.
+	// reaping tracks background reclamation of legacy trees and index space, so
+	// shutdown and tests can wait for it instead of racing it.
 	reaping sync.WaitGroup
+	// incrementalVacuum reports whether the index can return freed pages to the
+	// filesystem on its own. False on a database created before this was the
+	// default: PRAGMA incremental_vacuum is a silent no-op there, so deletions
+	// reclaim nothing until an explicit full compaction converts the file.
+	incrementalVacuum bool
+	// reclaiming keeps a single background reclamation in flight. Concurrent ones
+	// would only contend for wmu, never finish sooner.
+	reclaiming atomic.Bool
+	// closed is shut by Close so a reclamation abandons its remaining budget
+	// instead of holding shutdown open for minutes. Nil on the zero value, which
+	// stopping() treats as "not closing".
+	closed    chan struct{}
+	closeOnce sync.Once
 	// pass is the set of hosts whose MITM interception failed for a proxy/protocol
 	// reason; connections to them are tunneled transparently (fail-open) so the
 	// request still reaches the target — unrecorded — instead of being killed.
@@ -142,26 +179,27 @@ func Open(dir, addr string) (*Traffic, error) {
 			return nil, err
 		}
 	}
-	db, err := sql.Open("sqlite", filepath.Join(dir, "_index", "index.sqlite"))
+	// busy_timeout is a per-connection setting, so it belongs in the DSN rather
+	// than in a one-off Exec: the pool opens connections on demand, and an Exec
+	// only configures whichever one happened to serve it — leaving every other
+	// connection to fail instantly the moment a writer holds the database.
+	// The driver splits the DSN at the first '?', so a data directory containing
+	// one would silently name a different file; that path falls back to a bare
+	// DSN, where initIndex still applies the pragmas to its own connection.
+	index := filepath.Join(dir, "_index", "index.sqlite")
+	dsn := index
+	if !strings.ContainsRune(index, '?') {
+		dsn = "file:" + index + "?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)"
+	}
+	db, err := sql.Open("sqlite", dsn)
 	if err != nil {
 		return nil, err
 	}
-	for _, p := range []string{"PRAGMA journal_mode=WAL", "PRAGMA busy_timeout=5000"} {
-		if _, err := db.Exec(p); err != nil {
-			db.Close()
-			return nil, err
-		}
-	}
-	if _, err := db.Exec(indexSchema); err != nil {
+	t := &Traffic{dir: dir, addr: addr, db: db, closed: make(chan struct{})}
+	if err := t.initIndex(); err != nil {
 		db.Close()
 		return nil, err
 	}
-	t := &Traffic{dir: dir, addr: addr, db: db}
-	if _, err := db.Exec(ftsSchema); err != nil {
-		log.Printf("[traffic] 全文索引不可用，正文搜索将被禁用（元数据搜索不受影响）：%v", err)
-	} else {
-		t.fts = true
-	}
 
 	p, err := mproxy.NewProxy(&mproxy.Options{
 		Addr:        addr,
@@ -193,6 +231,66 @@ func Open(dir, addr string) (*Traffic, error) {
 	return t, nil
 }
 
+// initIndex applies the schema on a single pinned connection. Pinning is what
+// makes auto_vacuum reliable: it can only be set while the database still holds
+// no tables, and only the VACUUM that follows writes it into the file header —
+// two steps a pooled *sql.DB is free to route to different connections, which
+// would silently drop the setting.
+//
+// auto_vacuum=incremental is what lets a deletion hand freed pages back to the
+// filesystem. Without it SQLite merely chains them onto its freelist, so the
+// index file never shrinks no matter how much traffic is deleted — and since
+// every body below maxInlineBody lives in that file, plus a trigram index
+// roughly twice the size of the text it covers, a capture-heavy install ends up
+// holding gigabytes for traffic it no longer has. On a database that already has
+// tables the pragma is a documented no-op, so installs created before this keep
+// auto_vacuum=0 until a full compaction converts the file; incrementalVacuum
+// records that so reclaim can say so instead of pretending to reclaim.
+func (t *Traffic) initIndex() error {
+	ctx := context.Background()
+	conn, err := t.db.Conn(ctx)
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+	// Re-applied rather than left to the DSN so the bare-DSN fallback in Open is
+	// still correct: journal_mode persists in the file header, which is what every
+	// later connection reads.
+	for _, p := range []string{`PRAGMA journal_mode=WAL`, `PRAGMA busy_timeout=5000`} {
+		if _, err := conn.ExecContext(ctx, p); err != nil {
+			return err
+		}
+	}
+	var tables int
+	if err := conn.QueryRowContext(ctx, `SELECT count(*) FROM sqlite_master WHERE type='table'`).Scan(&tables); err != nil {
+		return err
+	}
+	if tables == 0 {
+		for _, p := range []string{`PRAGMA auto_vacuum=incremental`, `VACUUM`} {
+			if _, err := conn.ExecContext(ctx, p); err != nil {
+				return err
+			}
+		}
+	}
+	var mode int
+	if err := conn.QueryRowContext(ctx, `PRAGMA auto_vacuum`).Scan(&mode); err != nil {
+		return err
+	}
+	t.incrementalVacuum = mode == autoVacuumIncremental
+	if !t.incrementalVacuum {
+		log.Printf("[traffic] 索引库未启用增量回收（auto_vacuum=%d）：删除流量不会缩小 index.sqlite，需要执行一次存储压缩来转换", mode)
+	}
+	if _, err := conn.ExecContext(ctx, indexSchema); err != nil {
+		return err
+	}
+	if _, err := conn.ExecContext(ctx, ftsSchema); err != nil {
+		log.Printf("[traffic] 全文索引不可用，正文搜索将被禁用（元数据搜索不受影响）：%v", err)
+		return nil
+	}
+	t.fts = true
+	return nil
+}
+
 // hostOnly strips an optional :port, so passthrough keys match whether the host
 // arrives as "example.com:443" (CONNECT) or "example.com" (request URL).
 func hostOnly(hostport string) string {
@@ -262,11 +360,27 @@ func (t *Traffic) Start() error { return t.proxy.Start() }
 
 // Close waits for background tree reclamation to finish before closing the
 // index, so shutdown never leaves a goroutine unlinking files out from under a
-// removed data directory.
+// removed data directory. Index-space reclamation is signalled to stop first:
+// it holds a whole minutes-long budget, and finishing it is never worth delaying
+// shutdown for — the next deletion resumes it.
 func (t *Traffic) Close() error {
+	if t.closed != nil {
+		t.closeOnce.Do(func() { close(t.closed) })
+	}
 	t.reaping.Wait()
 	return t.db.Close()
 }
+
+// stopping reports whether Close has been called. A nil channel (the zero value)
+// is never ready, so this reads as "not closing" without a separate guard.
+func (t *Traffic) stopping() bool {
+	select {
+	case <-t.closed:
+		return true
+	default:
+		return false
+	}
+}
 func (t *Traffic) DB() *sql.DB { return t.db }
 
 // sink is the go-mitmproxy addon that records completed exchanges.
@@ -944,6 +1058,7 @@ func (t *Traffic) DeleteHost(host string) (int64, error) {
 	}
 	t.reapStage(stageDir)
 	if n > 0 {
+		t.reclaim()
 		if err := t.gcBlobs(); err != nil {
 			return n, err
 		}
@@ -1345,6 +1460,7 @@ func (t *Traffic) RecoverHostDeleteStages(archiveCommitted func(int64, int64) (b
 		needsGC = true
 	}
 	if needsGC {
+		t.reclaim()
 		if err := t.gcBlobs(); err != nil {
 			errs = append(errs, err)
 		}
@@ -1436,6 +1552,7 @@ func (s *HostDeleteStage) Commit() error {
 	// it now runs in the background, while collection below only needs them to be
 	// out of the live tree — which the staging rename already guaranteed.
 	s.traffic.reapStage(s.stageDir)
+	s.traffic.reclaim()
 	var errs []error
 	if err := s.traffic.gcBlobs(); err != nil {
 		errs = append(errs, fmt.Errorf("回收流量 blob: %w", err))
@@ -1445,6 +1562,127 @@ func (s *HostDeleteStage) Commit() error {
 	return errors.Join(errs...)
 }
 
+// indexBytes is the index's real footprint on disk: the database plus its
+// write-ahead log and shared-memory file, since those are what an operator sees
+// the data directory holding.
+func (t *Traffic) indexBytes() int64 {
+	base := filepath.Join(t.dir, "_index", "index.sqlite")
+	var total int64
+	for _, p := range []string{base, base + "-wal", base + "-shm"} {
+		if st, err := os.Stat(p); err == nil {
+			total += st.Size()
+		}
+	}
+	return total
+}
+
+// reclaim hands space freed by a deletion back to the filesystem. Deleting rows
+// only makes them invisible: SQLite chains their pages onto a freelist, and
+// deleting from the contentless_delete full-text index writes tombstones rather
+// than removing the postings they hide. Neither shrinks a single byte on disk,
+// and since bodies below maxInlineBody live in that same file, an install that
+// captures heavily ends up holding far more than the traffic it still has.
+//
+// The work is done in the background, in bounded steps that drop the write lock
+// between them, because it is proportional to what was deleted — freeing several
+// gigabytes under one lock would stall record(), which go-mitmproxy calls before
+// it replies to the client, and so would stall the requests being recorded.
+// Safe to call with wmu held: the background pass simply waits for it.
+//
+// Best-effort throughout. Failing to reclaim costs disk space, never
+// correctness, so errors are logged and the next deletion resumes the work.
+func (t *Traffic) reclaim() {
+	if !t.reclaiming.CompareAndSwap(false, true) {
+		return // one pass at a time; a second would only contend for the lock
+	}
+	t.reaping.Go(func() {
+		defer t.reclaiming.Store(false)
+		// One pinned connection for the whole pass: a long reclamation issues
+		// hundreds of statements, and letting the pool hand each one a different
+		// connection would both churn connections and split the freelist readings
+		// that decide when to stop away from the vacuum they measure.
+		ctx := context.Background()
+		conn, err := t.db.Conn(ctx)
+		if err != nil {
+			log.Printf("[traffic] 回收索引空间失败（获取连接）：%v", err)
+			return
+		}
+		defer conn.Close()
+		deadline := time.Now().Add(reclaimBudget)
+		merges := reclaimMergeSteps
+		for step := 0; ; step++ {
+			t.wmu.Lock()
+			progressed, err := t.reclaimChunk(ctx, conn, &merges)
+			t.wmu.Unlock()
+			if err != nil {
+				log.Printf("[traffic] 回收索引空间失败：%v", err)
+				return
+			}
+			if !progressed {
+				break
+			}
+			if t.stopping() {
+				return // shutdown must not wait out the remaining budget
+			}
+			if step+1 >= reclaimMaxSteps {
+				log.Printf("[traffic] 索引空间回收未做完（已用满 %d 步上限），下次删除时继续", reclaimMaxSteps)
+				return
+			}
+			if time.Now().After(deadline) {
+				log.Printf("[traffic] 索引空间回收未做完（已用满 %s 预算），下次删除时继续", reclaimBudget)
+				return
+			}
+		}
+		// Truncating the log is what makes the reclamation visible on disk: in WAL
+		// mode the freed pages are recorded there first, and a PASSIVE checkpoint
+		// would leave the log itself sitting at its high-water mark.
+		t.wmu.Lock()
+		defer t.wmu.Unlock()
+		if _, err := conn.ExecContext(ctx, `PRAGMA wal_checkpoint(TRUNCATE)`); err != nil {
+			log.Printf("[traffic] 截断 WAL 失败：%v", err)
+		}
+	})
+}
+
+// reclaimChunk does one bounded step and reports whether it made progress, which
+// is what the caller loops on. merges carries the remaining full-text merge
+// budget and is spent down here. Callers must hold wmu.
+func (t *Traffic) reclaimChunk(ctx context.Context, conn *sql.Conn, merges *int) (bool, error) {
+	progressed := false
+	if t.fts && *merges > 0 {
+		// A negative rank is fts5's page budget for one incremental merge.
+		if _, err := conn.ExecContext(ctx, `INSERT INTO ex_fts(ex_fts, rank) VALUES('merge', ?)`, -reclaimMergePages); err != nil {
+			return false, fmt.Errorf("合并全文索引: %w", err)
+		}
+		*merges--
+		progressed = true
+	}
+	if !t.incrementalVacuum {
+		// incremental_vacuum is a silent no-op on a database created with
+		// auto_vacuum=0; only a full compaction can convert one. Merging the
+		// full-text index above still pays off, so stop here rather than earlier.
+		return progressed, nil
+	}
+	var before, after int
+	if err := conn.QueryRowContext(ctx, `PRAGMA freelist_count`).Scan(&before); err != nil {
+		return false, err
+	}
+	if before == 0 {
+		return progressed, nil
+	}
+	// The budget is a constant and PRAGMA arguments cannot be bound as parameters.
+	if _, err := conn.ExecContext(ctx, fmt.Sprintf(`PRAGMA incremental_vacuum(%d)`, reclaimChunkPages)); err != nil {
+		return false, fmt.Errorf("回收索引空闲页: %w", err)
+	}
+	if err := conn.QueryRowContext(ctx, `PRAGMA freelist_count`).Scan(&after); err != nil {
+		return false, err
+	}
+	// Progress, not an empty freelist, is the right stopping condition:
+	// incremental_vacuum can only release pages it manages to move to the end of
+	// the file, so a residual freelist it cannot shrink is a normal outcome.
+	return progressed || after < before, nil
+}
+
 // gcBlobs removes blobs that no remaining exchange references. Live references
 // come from blob_refs, so collection is one query plus a sweep of the blob
 // directory — no exchange body is ever read. Emptied bucket directories are

traffic/upgrade_test.go

@@ -0,0 +1,164 @@
+package traffic
+
+import (
+	"database/sql"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+// openLegacyIndex builds the index exactly as the pre-reclamation Open did: a
+// plain-path DSN, pragmas via the pool, and auto_vacuum left at its default 0.
+func openLegacyIndex(t *testing.T, dir string) *sql.DB {
+	t.Helper()
+	if err := os.MkdirAll(filepath.Join(dir, "_index"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	old, err := sql.Open("sqlite", filepath.Join(dir, "_index", "index.sqlite"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, p := range []string{"PRAGMA journal_mode=WAL", "PRAGMA busy_timeout=5000"} {
+		if _, err := old.Exec(p); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := old.Exec(indexSchema); err != nil {
+		t.Fatal(err)
+	}
+	return old
+}
+
+// TestUpgradeFromOldInstall guards the upgrade path. Open now names the database
+// through a file: URI so per-connection pragmas can ride in the DSN, and a
+// driver that did not treat that as a URI would quietly open a file literally
+// named "file:/…" — an empty index, with every recorded exchange apparently
+// gone. The assertions below are what prove that does not happen.
+func TestUpgradeFromOldInstall(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "_index", "index.sqlite")
+	old := openLegacyIndex(t, dir)
+	if _, err := old.Exec(ftsSchema); err != nil {
+		t.Fatal(err)
+	}
+	// 三条历史流量，含一条 legacy path<>'' 的行
+	for i, row := range [][]any{
+		{"1700000000-0001", "old.example.com", ""},
+		{"1700000000-0002", "old.example.com", ""},
+		{"1700000000-0003", "legacy.example.com", "legacy.example.com/GET/x"},
+	} {
+		if _, err := old.Exec(`INSERT INTO exchanges(id,ts,host,method,url_template,url,status,content_type,req_len,resp_len,path)
+VALUES(?,?,?,'GET','/x','http://x/x',200,'text/html',0,9,?)`, row[0], 1700000000+i, row[1], row[2]); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := old.Exec(`INSERT INTO exchange_bodies(id,req_head,req_body,resp_head,resp_body)
+VALUES(?,'GET /x','','HTTP 200','老数据正文')`, row[0]); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := old.Exec(`INSERT INTO ex_fts(rowid,content) VALUES(?,?)`, i+1, "老数据正文 secret-token"); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := old.Close(); err != nil {
+		t.Fatal(err)
+	}
+	stat, err := os.Stat(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// ---- 新版本接管
+	tr, err := Open(dir, "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("新版本无法打开旧库: %v", err)
+	}
+	defer tr.Close()
+
+	// 1. 必须是同一个文件，不能悄悄开了个新空库
+	if st2, err := os.Stat(path); err != nil || st2.Size() == 0 {
+		t.Fatalf("原索引文件异常: size=%v err=%v", st2, err)
+	}
+	if entries, _ := os.ReadDir(filepath.Join(dir, "_index")); len(entries) > 3 {
+		for _, e := range entries {
+			t.Logf("_index 下: %s", e.Name())
+		}
+		t.Fatal("_index 下出现了预期外的文件，DSN 可能指向了别的库")
+	}
+	t.Logf("旧库 %d 字节，新版本接管后仍是同一文件", stat.Size())
+
+	// 2. 历史数据全部可见
+	n, err := tr.Count()
+	if err != nil || n != 3 {
+		t.Fatalf("Count=(%d,%v)，应为 (3,nil) —— 历史流量丢失", n, err)
+	}
+	// 3. 历史全文索引仍可搜
+	if tr.fts {
+		rows, err := tr.query("old.example.com", "", "secret-token", 0, 10)
+		if err != nil {
+			t.Fatalf("历史全文搜索失败: %v", err)
+		}
+		if len(rows) != 2 {
+			t.Fatalf("历史全文搜索命中 %d 条，应为 2", len(rows))
+		}
+	}
+	// 4. 历史正文仍可读
+	if _, resp, err := tr.Get("1700000000-0001"); err != nil {
+		t.Fatalf("读取历史正文失败: %v", err)
+	} else if resp == "" {
+		t.Fatal("历史响应为空")
+	}
+	// 5. 旧库不会被误判为已启用增量回收
+	if tr.incrementalVacuum {
+		t.Fatal("旧库被误判为已启用增量回收")
+	}
+	// 6. 删除仍然正常工作，且回收流程在旧库上能收敛
+	deleted, err := tr.DeleteHostsExact([]string{"old.example.com"})
+	if err != nil || deleted != 2 {
+		t.Fatalf("DeleteHostsExact=(%d,%v)，应为 (2,nil)", deleted, err)
+	}
+	tr.reaping.Wait()
+	if n, err := tr.Count(); err != nil || n != 1 {
+		t.Fatalf("删除后 Count=(%d,%v)，应为 (1,nil)", n, err)
+	}
+	// 7. legacy path<>'' 的行没被牵连
+	var legacyPath string
+	if err := tr.DB().QueryRow(`SELECT path FROM exchanges`).Scan(&legacyPath); err != nil {
+		t.Fatal(err)
+	}
+	if legacyPath == "" {
+		t.Fatal("legacy 行的 path 被清空了")
+	}
+}
+
+// TestDowngradeToOldBinary covers a rollback: a database created with
+// auto_vacuum=incremental must stay readable and writable by a build that knows
+// nothing about it. auto_vacuum only changes where SQLite tracks free pages, so
+// the old binary simply goes back to never returning them.
+func TestDowngradeToOldBinary(t *testing.T) {
+	dir := t.TempDir()
+	tr, err := Open(dir, "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !tr.incrementalVacuum {
+		t.Fatal("新库应启用增量回收")
+	}
+	bulkRecord(tr, "keep.example.com", 5, 100*1024)
+	if err := tr.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	old := openLegacyIndex(t, dir) // 旧版本二进制接管
+	defer old.Close()
+	var n int
+	if err := old.QueryRow(`SELECT COUNT(*) FROM exchanges`).Scan(&n); err != nil || n != 5 {
+		t.Fatalf("旧版本读到 (%d,%v)，应为 (5,nil)", n, err)
+	}
+	if _, err := old.Exec(`INSERT INTO exchanges(id,ts,host,method,url_template,url,status,content_type,req_len,resp_len,path)
+VALUES('x',1,'new.example.com','GET','/x','http://x/x',200,'',0,0,'')`); err != nil {
+		t.Fatalf("旧版本写入失败: %v", err)
+	}
+	if _, err := old.Exec(`DELETE FROM exchanges WHERE host='keep.example.com'`); err != nil {
+		t.Fatalf("旧版本删除失败: %v", err)
+	}
+}