归档

traffic/archive.go

@@ -0,0 +1,237 @@
+package traffic
+
+import (
+	"crypto/sha256"
+	"database/sql"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+)
+
+// ArchiveSnapshot is the portable traffic subset embedded in one task archive.
+// Large content-addressed bodies are copied alongside this manifest rather than
+// base64-encoded, preserving deduplication and allowing the tar writer to stream.
+type ArchiveSnapshot struct {
+	Version   int               `json:"version"`
+	Exchanges []ArchiveExchange `json:"exchanges"`
+	Blobs     []string          `json:"blobs"`
+}
+
+type ArchiveExchange struct {
+	ID          string `json:"id"`
+	TS          int64  `json:"ts"`
+	Host        string `json:"host"`
+	Method      string `json:"method"`
+	URLTemplate string `json:"url_template"`
+	URL         string `json:"url"`
+	Status      int    `json:"status"`
+	ContentType string `json:"content_type"`
+	ReqLen      int    `json:"req_len"`
+	RespLen     int    `json:"resp_len"`
+	ReqHead     string `json:"req_head"`
+	ReqBody     []byte `json:"req_body,omitempty"`
+	ReqBlob     string `json:"req_blob,omitempty"`
+	RespHead    string `json:"resp_head"`
+	RespBody    []byte `json:"resp_body,omitempty"`
+	RespBlob    string `json:"resp_blob,omitempty"`
+}
+
+// ExportHosts writes an exact-host snapshot to dir. It holds the traffic writer
+// lock while reading SQLite and blobs, so each body and its index row come from
+// one consistent point in time.
+func (t *Traffic) ExportHosts(hosts []string, dir string) (int64, error) {
+	if t == nil || len(hosts) == 0 {
+		return 0, nil
+	}
+	if err := os.MkdirAll(filepath.Join(dir, "blobs"), 0o700); err != nil {
+		return 0, err
+	}
+	t.wmu.Lock()
+	defer t.wmu.Unlock()
+	unique := uniqueArchiveHosts(hosts)
+	placeholders := make([]string, len(unique))
+	args := make([]any, len(unique))
+	for i, host := range unique {
+		placeholders[i] = "?"
+		args[i] = host
+	}
+	rows, err := t.db.Query(`SELECT id,ts,host,method,url_template,url,status,content_type,req_len,resp_len,path
+FROM exchanges WHERE host IN (`+strings.Join(placeholders, ",")+`) ORDER BY ts,id`, args...)
+	if err != nil {
+		return 0, err
+	}
+	defer rows.Close()
+	snapshot := ArchiveSnapshot{Version: 1}
+	blobs := map[string]struct{}{}
+	for rows.Next() {
+		var item ArchiveExchange
+		var legacyPath string
+		if err := rows.Scan(&item.ID, &item.TS, &item.Host, &item.Method, &item.URLTemplate,
+			&item.URL, &item.Status, &item.ContentType, &item.ReqLen, &item.RespLen, &legacyPath); err != nil {
+			return 0, err
+		}
+		var reqBlob, respBlob sql.NullString
+		err := t.db.QueryRow(`SELECT req_head,req_body,req_blob,resp_head,resp_body,resp_blob
+FROM exchange_bodies WHERE id=?`, item.ID).Scan(&item.ReqHead, &item.ReqBody, &reqBlob, &item.RespHead, &item.RespBody, &respBlob)
+		if errors.Is(err, sql.ErrNoRows) && strings.TrimSpace(legacyPath) != "" {
+			req, readErr := os.ReadFile(filepath.Join(t.dir, legacyPath, "request.http"))
+			if readErr != nil && !os.IsNotExist(readErr) {
+				return 0, readErr
+			}
+			resp, readErr := os.ReadFile(filepath.Join(t.dir, legacyPath, "response.http"))
+			if readErr != nil && !os.IsNotExist(readErr) {
+				return 0, readErr
+			}
+			item.ReqHead, item.RespHead = string(req), string(resp)
+		} else if err != nil {
+			return 0, err
+		}
+		item.ReqBlob, item.RespBlob = reqBlob.String, respBlob.String
+		for _, hash := range []string{item.ReqBlob, item.RespBlob} {
+			if hash != "" {
+				blobs[hash] = struct{}{}
+			}
+		}
+		snapshot.Exchanges = append(snapshot.Exchanges, item)
+	}
+	if err := rows.Err(); err != nil {
+		return 0, err
+	}
+	for hash := range blobs {
+		source, err := t.blobPath(hash)
+		if err != nil {
+			return 0, err
+		}
+		data, err := os.ReadFile(source)
+		if err != nil {
+			return 0, err
+		}
+		if err := os.WriteFile(filepath.Join(dir, "blobs", hash+".bin"), data, 0o600); err != nil {
+			return 0, err
+		}
+		snapshot.Blobs = append(snapshot.Blobs, hash)
+	}
+	sort.Strings(snapshot.Blobs)
+	raw, err := json.Marshal(snapshot)
+	if err != nil {
+		return 0, err
+	}
+	if err := os.WriteFile(filepath.Join(dir, "traffic.json"), raw, 0o600); err != nil {
+		return 0, err
+	}
+	return int64(len(snapshot.Exchanges)), nil
+}
+
+// ImportArchive imports only missing exchange IDs. Current hot rows always win,
+// and repeated restore attempts are safe after a partial external failure.
+func (t *Traffic) ImportArchive(dir string) (int64, error) {
+	if t == nil {
+		return 0, nil
+	}
+	raw, err := os.ReadFile(filepath.Join(dir, "traffic.json"))
+	if os.IsNotExist(err) {
+		return 0, nil
+	}
+	if err != nil {
+		return 0, err
+	}
+	var snapshot ArchiveSnapshot
+	if err := json.Unmarshal(raw, &snapshot); err != nil {
+		return 0, err
+	}
+	if snapshot.Version != 1 {
+		return 0, fmt.Errorf("unsupported traffic archive version %d", snapshot.Version)
+	}
+	t.wmu.Lock()
+	defer t.wmu.Unlock()
+	for _, hash := range snapshot.Blobs {
+		if !blobHashRe.MatchString(hash) {
+			return 0, fmt.Errorf("invalid archived traffic blob %q", hash)
+		}
+		data, err := os.ReadFile(filepath.Join(dir, "blobs", hash+".bin"))
+		if err != nil {
+			return 0, err
+		}
+		sum := sha256.Sum256(data)
+		if hex.EncodeToString(sum[:]) != hash {
+			return 0, fmt.Errorf("traffic blob checksum mismatch: %s", hash)
+		}
+		destination := filepath.Join(t.dir, "_blobs", "sha256", hash[:2], hash+".bin")
+		if _, err := os.Stat(destination); os.IsNotExist(err) {
+			if err := os.MkdirAll(filepath.Dir(destination), 0o755); err != nil {
+				return 0, err
+			}
+			if err := os.WriteFile(destination, data, 0o644); err != nil {
+				return 0, err
+			}
+		}
+	}
+	tx, err := t.db.Begin()
+	if err != nil {
+		return 0, err
+	}
+	defer tx.Rollback() //nolint:errcheck
+	var imported int64
+	for _, item := range snapshot.Exchanges {
+		res, err := tx.Exec(`INSERT OR IGNORE INTO exchanges(id,ts,host,method,url_template,url,status,content_type,req_len,resp_len,path)
+VALUES(?,?,?,?,?,?,?,?,?,?,'')`, item.ID, item.TS, item.Host, item.Method, item.URLTemplate,
+			item.URL, item.Status, item.ContentType, item.ReqLen, item.RespLen)
+		if err != nil {
+			return imported, err
+		}
+		n, _ := res.RowsAffected()
+		if n == 0 {
+			continue
+		}
+		if _, err := tx.Exec(`INSERT INTO exchange_bodies(id,req_head,req_body,req_blob,resp_head,resp_body,resp_blob)
+VALUES(?,?,?,?,?,?,?)`, item.ID, item.ReqHead, item.ReqBody, nullIfEmpty(item.ReqBlob),
+			item.RespHead, item.RespBody, nullIfEmpty(item.RespBlob)); err != nil {
+			return imported, err
+		}
+		for _, hash := range []string{item.ReqBlob, item.RespBlob} {
+			if hash != "" {
+				if _, err := tx.Exec(`INSERT OR IGNORE INTO blob_refs(hash,exchange_id) VALUES(?,?)`, hash, item.ID); err != nil {
+					return imported, err
+				}
+			}
+		}
+		if t.fts {
+			var rowID int64
+			if err := tx.QueryRow(`SELECT rowid FROM exchanges WHERE id=?`, item.ID).Scan(&rowID); err != nil {
+				return imported, err
+			}
+			content := strings.Join([]string{item.URL, item.ReqHead, string(item.ReqBody), item.RespHead, string(item.RespBody)}, "\n")
+			if _, err := tx.Exec(`INSERT INTO ex_fts(rowid,content) VALUES(?,?)`, rowID, content); err != nil {
+				return imported, err
+			}
+		}
+		imported++
+	}
+	if err := tx.Commit(); err != nil {
+		return imported, err
+	}
+	return imported, nil
+}
+
+func uniqueArchiveHosts(hosts []string) []string {
+	seen := make(map[string]struct{}, len(hosts))
+	out := make([]string, 0, len(hosts))
+	for _, host := range hosts {
+		host = strings.ToLower(strings.TrimSpace(host))
+		if host == "" {
+			continue
+		}
+		if _, exists := seen[host]; exists {
+			continue
+		}
+		seen[host] = struct{}{}
+		out = append(out, host)
+	}
+	sort.Strings(out)
+	return out
+}

traffic/archive_test.go

@@ -0,0 +1,158 @@
+package traffic
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+func TestTrafficArchiveRoundTripAndCurrentRowWins(t *testing.T) {
+	trafficDir := t.TempDir()
+	tr, err := Open(trafficDir, "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer tr.Close()
+	const (
+		id   = "archive-exchange-1"
+		host = "archive.example"
+	)
+	requestHead := "POST /v1/test HTTP/1.1\nHost: archive.example"
+	responseHead := "HTTP/1.1 200 OK\nContent-Type: application/json"
+	if _, err := tr.DB().Exec(`INSERT INTO exchanges(id,ts,host,method,url_template,url,status,content_type,req_len,resp_len,path)
+VALUES(?,?,?,?,?,?,?,?,?,?,?)`, id, 1, host, "POST", "/v1/test", "https://archive.example/v1/test", 200,
+		"application/json", 7, 11, ""); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := tr.DB().Exec(`INSERT INTO exchange_bodies(id,req_head,req_body,resp_head,resp_body)
+VALUES(?,?,?,?,?)`, id, requestHead, []byte("payload"), responseHead, []byte(`{"ok":true}`)); err != nil {
+		t.Fatal(err)
+	}
+
+	archiveDir := filepath.Join(t.TempDir(), "traffic")
+	count, err := tr.ExportHosts([]string{host, host}, archiveDir)
+	if err != nil || count != 1 {
+		t.Fatalf("ExportHosts count=%d err=%v", count, err)
+	}
+	if _, err := os.Stat(filepath.Join(archiveDir, "traffic.json")); err != nil {
+		t.Fatal(err)
+	}
+	stage, err := tr.StageDeleteHostsExact([]string{host})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if stage.Deleted() != 1 {
+		t.Fatalf("staged deleted=%d, want 1", stage.Deleted())
+	}
+	if err := stage.Commit(); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, err := tr.Get(id); err == nil {
+		t.Fatal("deleted exchange remained readable")
+	}
+	imported, err := tr.ImportArchive(archiveDir)
+	if err != nil || imported != 1 {
+		t.Fatalf("ImportArchive imported=%d err=%v", imported, err)
+	}
+	req, resp, err := tr.Get(id)
+	if err != nil || !strings.Contains(req, "payload") || !strings.Contains(resp, `{"ok":true}`) {
+		t.Fatalf("restored exchange req=%q resp=%q err=%v", req, resp, err)
+	}
+	if _, err := tr.DB().Exec(`UPDATE exchange_bodies SET resp_body=? WHERE id=?`, []byte(`{"current":true}`), id); err != nil {
+		t.Fatal(err)
+	}
+	imported, err = tr.ImportArchive(archiveDir)
+	if err != nil || imported != 0 {
+		t.Fatalf("idempotent import imported=%d err=%v", imported, err)
+	}
+	_, resp, err = tr.Get(id)
+	if err != nil || !strings.Contains(resp, `{"current":true}`) || strings.Contains(resp, `{"ok":true}`) {
+		t.Fatalf("archive overwrote current exchange resp=%q err=%v", resp, err)
+	}
+}
+
+func TestRecoverArchiveHostDeleteStageRollsBackBeforePostgresCommit(t *testing.T) {
+	trafficDir := t.TempDir()
+	tr, err := Open(trafficDir, "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer tr.Close()
+	const host = "rollback-archive.example"
+	insertArchiveRecoveryExchange(t, tr, "rollback-exchange", host)
+	legacyDir := filepath.Join(trafficDir, sanitize(host))
+	if err := os.MkdirAll(legacyDir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(legacyDir, "request.http"), []byte("request"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	stage, err := tr.StageDeleteHostsExactForArchive([]string{host}, 17, 42)
+	if err != nil {
+		t.Fatal(err)
+	}
+	simulateTrafficStageProcessExit(t, stage)
+	if err := tr.RecoverHostDeleteStages(func(id, taskID int64) (bool, error) {
+		if id != 17 || taskID != 42 {
+			t.Fatalf("archive id=%d task=%d, want 17/42", id, taskID)
+		}
+		return false, nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := os.Stat(filepath.Join(legacyDir, "request.http")); err != nil {
+		t.Fatalf("legacy traffic tree not restored: %v", err)
+	}
+	var count int
+	if err := tr.DB().QueryRow(`SELECT count(*) FROM exchanges WHERE host=?`, host).Scan(&count); err != nil || count != 1 {
+		t.Fatalf("exchange count=%d err=%v, want 1", count, err)
+	}
+}
+
+func TestRecoverArchiveHostDeleteStageCompletesAfterPostgresCommit(t *testing.T) {
+	trafficDir := t.TempDir()
+	tr, err := Open(trafficDir, "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer tr.Close()
+	const host = "committed-archive.example"
+	insertArchiveRecoveryExchange(t, tr, "committed-exchange", host)
+	legacyDir := filepath.Join(trafficDir, sanitize(host))
+	if err := os.MkdirAll(legacyDir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	stage, err := tr.StageDeleteHostsExactForArchive([]string{host}, 18, 43)
+	if err != nil {
+		t.Fatal(err)
+	}
+	simulateTrafficStageProcessExit(t, stage)
+	if err := tr.RecoverHostDeleteStages(func(id, taskID int64) (bool, error) { return id == 18 && taskID == 43, nil }); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := os.Stat(legacyDir); !os.IsNotExist(err) {
+		t.Fatalf("committed legacy traffic tree remains: %v", err)
+	}
+	var count int
+	if err := tr.DB().QueryRow(`SELECT count(*) FROM exchanges WHERE host=?`, host).Scan(&count); err != nil || count != 0 {
+		t.Fatalf("exchange count=%d err=%v, want 0", count, err)
+	}
+}
+
+func insertArchiveRecoveryExchange(t *testing.T, tr *Traffic, id, host string) {
+	t.Helper()
+	if _, err := tr.DB().Exec(`INSERT INTO exchanges(id,ts,host,method,url_template,url,status,content_type,req_len,resp_len,path)
+VALUES(?,?,?,?,?,?,?,?,?,?,?)`, id, 1, host, "GET", "/", "https://"+host+"/", 200, "text/plain", 0, 0, ""); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func simulateTrafficStageProcessExit(t *testing.T, stage *HostDeleteStage) {
+	t.Helper()
+	if err := stage.tx.Rollback(); err != nil {
+		t.Fatal(err)
+	}
+	stage.done = true
+	stage.traffic.wmu.Unlock()
+}

traffic/traffic.go

@@ -205,7 +205,7 @@ func (t *Traffic) Close() error {
 	t.reaping.Wait()
 	return t.db.Close()
 }
-func (t *Traffic) DB() *sql.DB  { return t.db }
+func (t *Traffic) DB() *sql.DB { return t.db }
 
 // sink is the go-mitmproxy addon that records completed exchanges.
 type sink struct {
@@ -870,6 +870,21 @@ type stagedTrafficPath struct {
 	staged string
 }
 
+type hostDeleteStageJournal struct {
+	Version   int                   `json:"version"`
+	ArchiveID int64                 `json:"archive_id,omitempty"`
+	TaskID    int64                 `json:"task_id,omitempty"`
+	Hosts     []string              `json:"hosts,omitempty"`
+	Moves     []hostDeleteStageMove `json:"moves"`
+}
+
+type hostDeleteStageMove struct {
+	Source string `json:"source"`
+	Staged string `json:"staged"`
+}
+
+const hostDeleteStageJournalName = "journal.json"
+
 // stageTrees moves host directories aside onto the same filesystem. The rename is
 // atomic and instant, which buys two things the unlink cannot: the deletion stays
 // reversible until the transaction commits, and the tree stops being visible to
@@ -877,7 +892,12 @@ type stagedTrafficPath struct {
 // directories, so anything under _delete_staging is already out of the live set).
 // An empty stageDir is returned when there was nothing to stage.
 func (t *Traffic) stageTrees(dirs []string) (stageDir string, moves []stagedTrafficPath, err error) {
+	return t.stageTreesForArchive(dirs, nil, 0, 0)
+}
+
+func (t *Traffic) stageTreesForArchive(dirs, hosts []string, archiveID, taskID int64) (stageDir string, moves []stagedTrafficPath, err error) {
 	seen := make(map[string]struct{}, len(dirs))
+	planned := make([]stagedTrafficPath, 0, len(dirs))
 	for _, source := range dirs {
 		if _, dup := seen[source]; dup {
 			continue
@@ -889,16 +909,29 @@ func (t *Traffic) stageTrees(dirs []string) (stageDir string, moves []stagedTraf
 			}
 			return stageDir, moves, fmt.Errorf("检查历史流量目录 %s: %w", source, err)
 		}
-		if stageDir == "" {
-			parent := filepath.Join(t.dir, "_delete_staging")
-			if err := os.MkdirAll(parent, 0o700); err != nil {
-				return stageDir, moves, fmt.Errorf("创建流量暂存目录: %w", err)
-			}
-			if stageDir, err = os.MkdirTemp(parent, "hosts-"); err != nil {
-				return stageDir, moves, fmt.Errorf("创建流量暂存目录: %w", err)
-			}
-		}
-		staged := filepath.Join(stageDir, fmt.Sprintf("%d-%s", len(moves), filepath.Base(source)))
+		planned = append(planned, stagedTrafficPath{source: source})
+	}
+	if len(planned) == 0 {
+		return "", nil, nil
+	}
+	parent := filepath.Join(t.dir, "_delete_staging")
+	if err := os.MkdirAll(parent, 0o700); err != nil {
+		return "", nil, fmt.Errorf("创建流量暂存目录: %w", err)
+	}
+	if stageDir, err = os.MkdirTemp(parent, "hosts-"); err != nil {
+		return "", nil, fmt.Errorf("创建流量暂存目录: %w", err)
+	}
+	journal := hostDeleteStageJournal{Version: 1, ArchiveID: archiveID, TaskID: taskID, Hosts: uniqueArchiveHosts(hosts)}
+	for i := range planned {
+		planned[i].staged = filepath.Join(stageDir, fmt.Sprintf("%d-%s", i, filepath.Base(planned[i].source)))
+		journal.Moves = append(journal.Moves, hostDeleteStageMove{Source: planned[i].source, Staged: planned[i].staged})
+	}
+	if err := writeHostDeleteStageJournal(filepath.Join(stageDir, hostDeleteStageJournalName), journal); err != nil {
+		_ = os.RemoveAll(stageDir)
+		return "", nil, err
+	}
+	for _, move := range planned {
+		source, staged := move.source, move.staged
 		if err := os.Rename(source, staged); err != nil {
 			return stageDir, moves, fmt.Errorf("移出历史流量目录 %s: %w", source, err)
 		}
@@ -907,6 +940,26 @@ func (t *Traffic) stageTrees(dirs []string) (stageDir string, moves []stagedTraf
 	return stageDir, moves, nil
 }
 
+func writeHostDeleteStageJournal(path string, journal hostDeleteStageJournal) error {
+	raw, err := json.Marshal(journal)
+	if err != nil {
+		return err
+	}
+	file, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
+	if err != nil {
+		return err
+	}
+	if _, err := file.Write(raw); err != nil {
+		_ = file.Close()
+		return err
+	}
+	if err := file.Sync(); err != nil {
+		_ = file.Close()
+		return err
+	}
+	return file.Close()
+}
+
 // restoreTrees puts staged directories back where they came from, newest move
 // first, and drops the staging directory once everything is home.
 func restoreTrees(stageDir string, moves []stagedTrafficPath) error {
@@ -992,6 +1045,20 @@ func (s *HostDeleteStage) Deleted() int64 {
 // StageDeleteHostsExact prepares a reversible exact-host deletion. Callers must
 // finish every successful stage with Commit or Rollback.
 func (t *Traffic) StageDeleteHostsExact(hosts []string) (*HostDeleteStage, error) {
+	return t.stageDeleteHostsExact(hosts, 0, 0)
+}
+
+// StageDeleteHostsExactForArchive ties the reversible traffic deletion to a
+// persistent task archive. Startup recovery uses archiveCommitted to decide
+// whether an interrupted stage must be completed or rolled back.
+func (t *Traffic) StageDeleteHostsExactForArchive(hosts []string, archiveID, taskID int64) (*HostDeleteStage, error) {
+	if archiveID <= 0 || taskID <= 0 {
+		return nil, errors.New("archive and task ids must be positive")
+	}
+	return t.stageDeleteHostsExact(hosts, archiveID, taskID)
+}
+
+func (t *Traffic) stageDeleteHostsExact(hosts []string, archiveID, taskID int64) (*HostDeleteStage, error) {
 	t.wmu.Lock()
 	stage := &HostDeleteStage{traffic: t}
 	fail := func(cause error) (*HostDeleteStage, error) {
@@ -1037,13 +1104,156 @@ func (t *Traffic) StageDeleteHostsExact(hosts []string) (*HostDeleteStage, error
 		stage.deleted += n
 	}
 
-	stage.stageDir, stage.moves, err = t.stageTrees(legacy)
+	stage.stageDir, stage.moves, err = t.stageTreesForArchive(legacy, unique, archiveID, taskID)
 	if err != nil {
 		return fail(err)
 	}
 	return stage, nil
 }
 
+// RecoverHostDeleteStages resolves filesystem moves left by a process exit.
+// SQLite rolls open transactions back on restart, while the supplied callback
+// identifies task archives whose PostgreSQL compaction already committed.
+func (t *Traffic) RecoverHostDeleteStages(archiveCommitted func(int64, int64) (bool, error)) error {
+	if t == nil {
+		return nil
+	}
+	t.wmu.Lock()
+	defer t.wmu.Unlock()
+	parent := filepath.Join(t.dir, "_delete_staging")
+	entries, err := os.ReadDir(parent)
+	if os.IsNotExist(err) {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	var errs []error
+	needsGC := false
+	for _, entry := range entries {
+		if !entry.IsDir() {
+			continue
+		}
+		stageDir := filepath.Join(parent, entry.Name())
+		raw, err := os.ReadFile(filepath.Join(stageDir, hostDeleteStageJournalName))
+		if os.IsNotExist(err) {
+			// Older versions did not persist enough information for lossless
+			// recovery. Keep the directory for manual inspection.
+			continue
+		}
+		if err != nil {
+			errs = append(errs, err)
+			continue
+		}
+		var journal hostDeleteStageJournal
+		if err := json.Unmarshal(raw, &journal); err != nil {
+			errs = append(errs, fmt.Errorf("读取流量暂存日志 %s: %w", stageDir, err))
+			continue
+		}
+		if journal.Version != 1 {
+			errs = append(errs, fmt.Errorf("流量暂存日志 %s 的版本 %d 不受支持", stageDir, journal.Version))
+			continue
+		}
+		moves := make([]stagedTrafficPath, 0, len(journal.Moves))
+		for _, move := range journal.Moves {
+			if !pathWithin(t.dir, move.Source) || !pathWithin(stageDir, move.Staged) {
+				errs = append(errs, fmt.Errorf("流量暂存日志包含越界路径: %s", stageDir))
+				moves = nil
+				break
+			}
+			if _, err := os.Lstat(move.Staged); err == nil {
+				moves = append(moves, stagedTrafficPath{source: move.Source, staged: move.Staged})
+			} else if !os.IsNotExist(err) {
+				errs = append(errs, err)
+				moves = nil
+				break
+			}
+		}
+		if moves == nil {
+			continue
+		}
+		committed := false
+		if journal.ArchiveID > 0 {
+			if archiveCommitted == nil {
+				errs = append(errs, fmt.Errorf("流量归档 %d 无状态解析器", journal.ArchiveID))
+				continue
+			}
+			committed, err = archiveCommitted(journal.ArchiveID, journal.TaskID)
+			if err != nil {
+				errs = append(errs, err)
+				continue
+			}
+		} else if len(journal.Hosts) > 0 {
+			committed, err = t.hostsHaveNoExchanges(journal.Hosts)
+			if err != nil {
+				errs = append(errs, err)
+				continue
+			}
+		}
+		if !committed {
+			if err := restoreTrees(stageDir, moves); err != nil {
+				errs = append(errs, err)
+			}
+			continue
+		}
+		if err := t.deleteArchivedHosts(journal.Hosts); err != nil {
+			errs = append(errs, err)
+			continue
+		}
+		if err := os.RemoveAll(stageDir); err != nil {
+			errs = append(errs, err)
+			continue
+		}
+		needsGC = true
+	}
+	if needsGC {
+		if err := t.gcBlobs(); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	return errors.Join(errs...)
+}
+
+func (t *Traffic) hostsHaveNoExchanges(hosts []string) (bool, error) {
+	for _, host := range hosts {
+		var count int
+		if err := t.db.QueryRow(`SELECT count(*) FROM exchanges WHERE host=?`, host).Scan(&count); err != nil {
+			return false, err
+		}
+		if count > 0 {
+			return false, nil
+		}
+	}
+	return true, nil
+}
+
+func (t *Traffic) deleteArchivedHosts(hosts []string) error {
+	tx, err := t.db.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback() //nolint:errcheck
+	for _, host := range hosts {
+		if _, err := t.deleteWhere(tx, `host=?`, host); err != nil {
+			return err
+		}
+	}
+	return tx.Commit()
+}
+
+func pathWithin(root, candidate string) bool {
+	rootAbs, err := filepath.Abs(root)
+	if err != nil {
+		return false
+	}
+	candidateAbs, err := filepath.Abs(candidate)
+	if err != nil {
+		return false
+	}
+	relative, err := filepath.Rel(rootAbs, candidateAbs)
+	return err == nil && relative != ".." && !strings.HasPrefix(relative, ".."+string(filepath.Separator))
+}
+
 // Rollback discards the staged deletion, leaving the traffic store untouched.
 func (s *HostDeleteStage) Rollback() error {
 	if s == nil || s.done {