feat(traffic): 流量列表新增「清空全部」，并借空库时机压实索引
一次删除全部流量记录，忽略当前筛选条件。与按 host 删除的两条路径相比多做 两件事： - 清理索引已不再记录的历史 host 目录。按 host 删除只能从索引反查目录， 行已消失的孤立目录会一直留着；「清空全部」既然语义是不留残余，就直接 扫顶层目录（跳过 _index/_blobs/_ca/_delete_staging）。 - 收尾做一次全量压实（optimize + VACUUM + wal_checkpoint(TRUNCATE)），把 索引占用的磁盘空间还给系统，并把实际释放量返回给前端提示。 VACUUM 的耗时与临时空间只和它保留的内容成正比，所以空库是做全量压实唯一 免费的时机——也正因如此，这一步顺带成为已有实例的转换途径：auto_vacuum 只能通过 VACUUM 生效，清空一次之后旧库即转为增量回收模式，此后日常按 host 删除就能自行回收空间（TestDeleteAllConvertsLegacyIndex 覆盖）。 已绑定到漏洞的流量证据保存在独立的证据库中，不受影响——绑定时已拷出独立 副本，正是为了让热流量可以随时清空而不带走证据。确认弹窗里明确写了这点。 前端把「清空全部」做成 outline 样式而非第二个 destructive 按钮：它忽略一切 筛选条件，不该看起来和「删除该目标」只差一次误点；无流量时禁用。 测试：新增 TestDeleteAllPurgesAndCompacts（清空后索引、正文、blob 引用与 孤立目录全清，空间回收到 1/8 以下，清空后仍可继续录制）与 TestDeleteAllConvertsLegacyIndex（旧库经清空后转为增量回收，随后普通删除 即可回收）。
This commit is contained in:
Autumn-27 committed
1 parent
61d486160e
commit
2f5f7c0227
6 files changed
+314
-24
No files matched your search
@@ -154,3 +154,104 @@ func TestReclaimOnLegacyIndexIsHarmless(t *testing.T) {
|
||||
t.Fatal("旧库居然回收了空闲页，说明测试没有真的构造出旧库")
|
||||
}
|
||||
}
|
||||
|
||||
// TestDeleteAllPurgesAndCompacts covers the page's clear-everything action: it
|
||||
// must leave nothing behind — including host directories the index no longer
|
||||
// knows about — and it must hand the index space back, since an emptied index is
|
||||
// the one moment a full rewrite is cheap.
|
||||
func TestDeleteAllPurgesAndCompacts(t *testing.T) {
|
||||
tr, dir := openTraffic(t)
|
||||
bulkRecord(tr, "a.example.com", 10, 200*1024)
|
||||
bulkRecord(tr, "b.example.com", 10, 200*1024)
|
||||
// A text body so the full-text index has real content, and a spilled one so a
|
||||
// blob exists to collect.
|
||||
tr.record(newFlow("c.example.com", "GET", "/page", nil, []byte(strings.Repeat("secret-token ", 500))))
|
||||
tr.record(newFlow("c.example.com", "GET", "/big", nil,
|
||||
[]byte(strings.Repeat("B", maxInlineBody+1024)), withRespType("application/sql")))
|
||||
// An orphaned legacy directory: no index row points at it, so only a
|
||||
// clear-everything should take it.
|
||||
orphan := filepath.Join(dir, "orphan.example.com")
|
||||
if err := os.MkdirAll(orphan, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
grown := tr.indexBytes()
|
||||
if grown < 5<<20 {
|
||||
t.Fatalf("索引只有 %d 字节，样本不足", grown)
|
||||
}
|
||||
|
||||
deleted, reclaimed, err := tr.DeleteAll()
|
||||
if err != nil {
|
||||
t.Fatalf("DeleteAll: %v", err)
|
||||
}
|
||||
if deleted != 22 {
|
||||
t.Fatalf("deleted=%d，应为 22", deleted)
|
||||
}
|
||||
tr.reaping.Wait()
|
||||
|
||||
if reclaimed < grown/2 {
|
||||
t.Fatalf("只回收了 %d 字节（删除前索引 %d）", reclaimed, grown)
|
||||
}
|
||||
if after := tr.indexBytes(); after > grown/8 {
|
||||
t.Fatalf("清空后索引仍占 %d 字节（删除前 %d）", after, grown)
|
||||
}
|
||||
for _, q := range []string{
|
||||
`SELECT COUNT(*) FROM exchanges`,
|
||||
`SELECT COUNT(*) FROM exchange_bodies`,
|
||||
`SELECT COUNT(*) FROM blob_refs`,
|
||||
} {
|
||||
var c int
|
||||
if err := tr.DB().QueryRow(q).Scan(&c); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if c != 0 {
|
||||
t.Fatalf("%s = %d，应为 0", q, c)
|
||||
}
|
||||
}
|
||||
if _, err := os.Stat(orphan); !os.IsNotExist(err) {
|
||||
t.Fatalf("孤立的历史 host 目录未被清理：%v", err)
|
||||
}
|
||||
// Recording must keep working against the freshly rewritten file.
|
||||
tr.record(newFlow("d.example.com", "GET", "/after", nil, []byte("清空后仍可录制")))
|
||||
if n, err := tr.Count(); err != nil || n != 1 {
|
||||
t.Fatalf("清空后 Count=(%d,%v)，应为 (1,nil)", n, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestDeleteAllConvertsLegacyIndex is why the purge compacts rather than just
|
||||
// deleting: auto_vacuum cannot be switched on after the fact except through a
|
||||
// VACUUM, and an emptied index is the cheapest place to pay for one. After this,
|
||||
// ordinary deletions reclaim space on their own.
|
||||
func TestDeleteAllConvertsLegacyIndex(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
old := openLegacyIndex(t, dir)
|
||||
if err := old.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("旧库不应报告已启用增量回收")
|
||||
}
|
||||
|
||||
bulkRecord(tr, "legacy.example.com", 10, 200*1024)
|
||||
if _, _, err := tr.DeleteAll(); err != nil {
|
||||
t.Fatalf("DeleteAll: %v", err)
|
||||
}
|
||||
if !tr.incrementalVacuum {
|
||||
t.Fatal("清空后旧库未被转换为增量回收模式")
|
||||
}
|
||||
|
||||
// The converted database now reclaims on an ordinary host deletion.
|
||||
bulkRecord(tr, "again.example.com", 10, 200*1024)
|
||||
grown := tr.indexBytes()
|
||||
if _, err := tr.DeleteHostsExact([]string{"again.example.com"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tr.reaping.Wait()
|
||||
if after := tr.indexBytes(); after > grown/4 {
|
||||
t.Fatalf("转换后普通删除仍未回收：%d 字节（删除前 %d）", after, grown)
|
||||
}
|
||||
}
|
||||
@@ -1066,6 +1066,116 @@ func (t *Traffic) DeleteHost(host string) (int64, error) {
|
||||
return n, nil
|
||||
}
|
||||
|
||||
// DeleteAll removes every recorded exchange — the page's clear-everything
|
||||
// action. It differs from the host-scoped deletions in two ways. It sweeps host
|
||||
// directories without consulting the index, because "clear everything" should
|
||||
// leave nothing behind and a directory whose rows are already gone would
|
||||
// otherwise survive. And it ends with a full compaction: VACUUM costs what it
|
||||
// keeps, so an emptied index is the one moment it is free, and it is also the
|
||||
// only way to switch on auto_vacuum for a database created without it.
|
||||
//
|
||||
// Evidence already bound to a finding is untouched. Those bodies were copied
|
||||
// into the evidence store when they were bound, precisely so that disposable
|
||||
// traffic could be cleared without taking proof with it.
|
||||
//
|
||||
// Returns the number of exchanges deleted and how many bytes of index the
|
||||
// compaction handed back.
|
||||
func (t *Traffic) DeleteAll() (deleted int64, reclaimed int64, err error) {
|
||||
t.wmu.Lock()
|
||||
defer t.wmu.Unlock()
|
||||
before := t.indexBytes()
|
||||
trees, err := t.allHostTrees()
|
||||
if err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
tx, err := t.db.Begin()
|
||||
if err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
defer tx.Rollback() //nolint:errcheck // no-op once committed
|
||||
if deleted, err = t.deleteWhere(tx, `1=1`); err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
// Staged before the commit so a filesystem failure can still abort the whole
|
||||
// deletion, exactly as in DeleteHost.
|
||||
stageDir, moves, err := t.stageTrees(trees)
|
||||
if err != nil {
|
||||
return 0, 0, errors.Join(err, restoreTrees(stageDir, moves))
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return 0, 0, errors.Join(fmt.Errorf("提交流量索引删除: %w", err), restoreTrees(stageDir, moves))
|
||||
}
|
||||
t.reapStage(stageDir)
|
||||
if err := t.gcBlobs(); err != nil {
|
||||
return deleted, 0, err
|
||||
}
|
||||
if err := t.compactIndex(); err != nil {
|
||||
// The deletion is already durable; compaction is disk space, not
|
||||
// correctness, so it must not turn a completed purge into a failed one.
|
||||
log.Printf("[traffic] 压实索引失败：%v", err)
|
||||
return deleted, 0, nil
|
||||
}
|
||||
return deleted, before - t.indexBytes(), nil
|
||||
}
|
||||
|
||||
// allHostTrees lists every host directory left by the pre-SQLite layout. Unlike
|
||||
// hostTrees it does not go through the index, so it also finds directories whose
|
||||
// rows are already gone. Underscore-prefixed entries are the store's own
|
||||
// (_index, _blobs, _ca, _delete_staging) and are never hosts.
|
||||
func (t *Traffic) allHostTrees() ([]string, error) {
|
||||
entries, err := os.ReadDir(t.dir)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var dirs []string
|
||||
for _, e := range entries {
|
||||
if e.IsDir() && !strings.HasPrefix(e.Name(), "_") {
|
||||
dirs = append(dirs, filepath.Join(t.dir, e.Name()))
|
||||
}
|
||||
}
|
||||
return dirs, nil
|
||||
}
|
||||
|
||||
// compactIndex rewrites the index into a fresh file, which is what actually
|
||||
// returns its pages to the filesystem. VACUUM's runtime and temporary space
|
||||
// scale with the content it keeps, so this is only reached right after DeleteAll
|
||||
// has emptied the index — never as a routine step. It doubles as the conversion
|
||||
// path for a database created before auto_vacuum=incremental: that pragma only
|
||||
// takes effect through the VACUUM that follows it, and both must run on the same
|
||||
// connection. Callers must hold wmu.
|
||||
func (t *Traffic) compactIndex() error {
|
||||
ctx := context.Background()
|
||||
conn, err := t.db.Conn(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer conn.Close()
|
||||
if t.fts {
|
||||
// A full merge, not the bounded one reclaim uses: with the index emptied
|
||||
// there is nothing left to merge, so this only discards the tombstones.
|
||||
if _, err := conn.ExecContext(ctx, `INSERT INTO ex_fts(ex_fts) VALUES('optimize')`); err != nil {
|
||||
return fmt.Errorf("合并全文索引: %w", err)
|
||||
}
|
||||
}
|
||||
if _, err := conn.ExecContext(ctx, `PRAGMA auto_vacuum=incremental`); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := conn.ExecContext(ctx, `VACUUM`); err != nil {
|
||||
return fmt.Errorf("压实索引: %w", err)
|
||||
}
|
||||
var mode int
|
||||
if err := conn.QueryRowContext(ctx, `PRAGMA auto_vacuum`).Scan(&mode); err != nil {
|
||||
return err
|
||||
}
|
||||
// A legacy database has just been converted, so later deletions can reclaim
|
||||
// space on their own instead of waiting for another purge.
|
||||
t.incrementalVacuum = mode == autoVacuumIncremental
|
||||
if _, err := conn.ExecContext(ctx, `PRAGMA wal_checkpoint(TRUNCATE)`); err != nil {
|
||||
return fmt.Errorf("截断 WAL: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// deleteWhere removes every trace of the exchanges matching the condition:
|
||||
// full-text index rows, bodies, blob references, and finally the index rows.
|
||||
// Order matters — each sub-select reads exchanges, so that table is emptied last.
|
||||
|
||||
Reference in new issue
Block a user