Add verified snapshots and harden archive identity
This commit is contained in:
DonVoo
parent
c6de214ac8
commit
68b9cee880
9 changed files with 1372 additions and 52 deletions
711
backend/14-snapshot.go
Normal file
711
backend/14-snapshot.go
Normal file
|
|
@ -0,0 +1,711 @@
|
|||
package backend
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"database/sql"
|
||||
"encoding/binary"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
const snapshotFormatVersion = 1
|
||||
|
||||
type SnapshotManifest struct {
|
||||
FormatVersion int `json:"format_version"`
|
||||
CreatedAt string `json:"created_at"`
|
||||
DatabasePath string `json:"database_path"`
|
||||
DatabaseSHA256 string `json:"database_sha256"`
|
||||
IndexSHA256 string `json:"index_sha256"`
|
||||
IndexRecords int64 `json:"index_records"`
|
||||
PreciseCopied int64 `json:"precise_copied"`
|
||||
PendingMbox int64 `json:"pending_mbox"`
|
||||
PendingTarget int64 `json:"pending_target"`
|
||||
InvalidStageOrder int64 `json:"invalid_stage_order"`
|
||||
MissingArchiveRecords int64 `json:"missing_archive_records"`
|
||||
Files []SnapshotFile `json:"files"`
|
||||
}
|
||||
|
||||
type SnapshotFile struct {
|
||||
Path string `json:"path"`
|
||||
Bytes int64 `json:"bytes"`
|
||||
SHA256 string `json:"sha256"`
|
||||
AccountID int64 `json:"account_id"`
|
||||
Account string `json:"account"`
|
||||
Folder string `json:"folder"`
|
||||
IndexRecords int64 `json:"index_records"`
|
||||
}
|
||||
|
||||
type snapshotWatermark struct {
|
||||
AccountID int64
|
||||
Account string
|
||||
MboxDir string
|
||||
Folder string
|
||||
Bytes int64
|
||||
IndexRecords int64
|
||||
SourcePath string
|
||||
RelativePath string
|
||||
}
|
||||
|
||||
// CreateSnapshot creates a self-consistent DB + mbox-prefix snapshot without
|
||||
// stopping append-only archive writers. The DB snapshot is taken first; every
|
||||
// copied mbox prefix is then bounded exclusively by offsets from that snapshot.
|
||||
func CreateSnapshot(target string) error {
|
||||
return createSnapshot(target, nil)
|
||||
}
|
||||
|
||||
func createSnapshot(target string, afterDatabaseSnapshot func() error) (retErr error) {
|
||||
if DB == nil {
|
||||
return fmt.Errorf("snapshot: database is not connected")
|
||||
}
|
||||
target = strings.TrimSpace(target)
|
||||
if target == "" {
|
||||
return fmt.Errorf("snapshot: target directory is empty")
|
||||
}
|
||||
targetAbs, err := filepath.Abs(target)
|
||||
if err != nil {
|
||||
return fmt.Errorf("snapshot target: %w", err)
|
||||
}
|
||||
if _, err := os.Stat(targetAbs); err == nil {
|
||||
return fmt.Errorf("snapshot target already exists: %s", targetAbs)
|
||||
} else if !os.IsNotExist(err) {
|
||||
return fmt.Errorf("snapshot target stat: %w", err)
|
||||
}
|
||||
if err := os.MkdirAll(filepath.Dir(targetAbs), 0o700); err != nil {
|
||||
return fmt.Errorf("snapshot target parent: %w", err)
|
||||
}
|
||||
partial := fmt.Sprintf("%s.partial-%d-%d", targetAbs, os.Getpid(), time.Now().UnixNano())
|
||||
if err := os.Mkdir(partial, 0o700); err != nil {
|
||||
return fmt.Errorf("snapshot staging directory: %w", err)
|
||||
}
|
||||
complete := false
|
||||
defer func() {
|
||||
if !complete {
|
||||
log.Printf("snapshot incomplete; staging data retained at %s", partial)
|
||||
if retErr != nil {
|
||||
retErr = fmt.Errorf("%w (incomplete snapshot retained at %s)", retErr, partial)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
databaseRel := filepath.ToSlash(filepath.Join("data", "mail-graveyard.db"))
|
||||
databasePath := filepath.Join(partial, filepath.FromSlash(databaseRel))
|
||||
if err := os.MkdirAll(filepath.Dir(databasePath), 0o700); err != nil {
|
||||
return fmt.Errorf("snapshot database directory: %w", err)
|
||||
}
|
||||
if _, err := DB.Exec(`VACUUM INTO ?`, databasePath); err != nil {
|
||||
return fmt.Errorf("snapshot database VACUUM INTO: %w", err)
|
||||
}
|
||||
if err := chmodPrivateFile(databasePath); err != nil {
|
||||
return fmt.Errorf("snapshot database permissions: %w", err)
|
||||
}
|
||||
if err := syncFile(databasePath); err != nil {
|
||||
return fmt.Errorf("snapshot database sync: %w", err)
|
||||
}
|
||||
if afterDatabaseSnapshot != nil {
|
||||
if err := afterDatabaseSnapshot(); err != nil {
|
||||
return fmt.Errorf("snapshot test hook: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
snapshotDB, err := openSnapshotDatabase(databasePath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
watermarks, err := collectSnapshotWatermarks(snapshotDB, Cfg.MboxRoot)
|
||||
if err != nil {
|
||||
_ = snapshotDB.Close()
|
||||
return err
|
||||
}
|
||||
stats, err := snapshotDatabaseStats(snapshotDB)
|
||||
if err == nil {
|
||||
stats.IndexSHA256, err = indexDigest(snapshotDB)
|
||||
}
|
||||
if closeErr := snapshotDB.Close(); err == nil && closeErr != nil {
|
||||
err = closeErr
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if stats.MissingArchiveRecords != 0 {
|
||||
return fmt.Errorf("snapshot refused: %d mbox_done identities have no archive index", stats.MissingArchiveRecords)
|
||||
}
|
||||
if stats.InvalidStageOrder != 0 {
|
||||
return fmt.Errorf("snapshot refused: %d precise identities have target_done=1 before mbox_done", stats.InvalidStageOrder)
|
||||
}
|
||||
var watermarkRecords int64
|
||||
for _, watermark := range watermarks {
|
||||
watermarkRecords += watermark.IndexRecords
|
||||
}
|
||||
if watermarkRecords != stats.IndexRecords {
|
||||
return fmt.Errorf("snapshot refused: indexed records=%d but mapped mbox records=%d", stats.IndexRecords, watermarkRecords)
|
||||
}
|
||||
|
||||
manifest := SnapshotManifest{
|
||||
FormatVersion: snapshotFormatVersion,
|
||||
CreatedAt: time.Now().UTC().Format(time.RFC3339Nano),
|
||||
DatabasePath: databaseRel,
|
||||
IndexSHA256: stats.IndexSHA256,
|
||||
IndexRecords: stats.IndexRecords,
|
||||
PreciseCopied: stats.PreciseCopied,
|
||||
PendingMbox: stats.PendingMbox,
|
||||
PendingTarget: stats.PendingTarget,
|
||||
InvalidStageOrder: stats.InvalidStageOrder,
|
||||
MissingArchiveRecords: stats.MissingArchiveRecords,
|
||||
Files: make([]SnapshotFile, 0, len(watermarks)),
|
||||
}
|
||||
manifest.DatabaseSHA256, err = hashFile(databasePath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("snapshot database hash: %w", err)
|
||||
}
|
||||
for _, watermark := range watermarks {
|
||||
destinationRel := filepath.ToSlash(filepath.Join("backup", watermark.RelativePath))
|
||||
destination := filepath.Join(partial, filepath.FromSlash(destinationRel))
|
||||
hash, err := copyFilePrefix(watermark.SourcePath, destination, watermark.Bytes)
|
||||
if err != nil {
|
||||
return fmt.Errorf("snapshot mbox %s: %w", destinationRel, err)
|
||||
}
|
||||
manifest.Files = append(manifest.Files, SnapshotFile{
|
||||
Path: destinationRel,
|
||||
Bytes: watermark.Bytes,
|
||||
SHA256: hash,
|
||||
AccountID: watermark.AccountID,
|
||||
Account: watermark.Account,
|
||||
Folder: watermark.Folder,
|
||||
IndexRecords: watermark.IndexRecords,
|
||||
})
|
||||
}
|
||||
sortSnapshotFiles(manifest.Files)
|
||||
manifestPath := filepath.Join(partial, "snapshot-manifest.json")
|
||||
manifestJSON, err := json.MarshalIndent(manifest, "", " ")
|
||||
if err != nil {
|
||||
return fmt.Errorf("snapshot manifest encode: %w", err)
|
||||
}
|
||||
manifestJSON = append(manifestJSON, '\n')
|
||||
if err := os.WriteFile(manifestPath, manifestJSON, 0o600); err != nil {
|
||||
return fmt.Errorf("snapshot manifest write: %w", err)
|
||||
}
|
||||
if err := syncFile(manifestPath); err != nil {
|
||||
return fmt.Errorf("snapshot manifest sync: %w", err)
|
||||
}
|
||||
if err := os.Rename(partial, targetAbs); err != nil {
|
||||
return fmt.Errorf("snapshot finalize: %w", err)
|
||||
}
|
||||
complete = true
|
||||
log.Printf("snapshot complete target=%s files=%d bytes=%d index_records=%d precise_copied=%d pending_mbox=%d pending_target=%d",
|
||||
targetAbs, len(manifest.Files), snapshotBytes(manifest.Files), manifest.IndexRecords, manifest.PreciseCopied, manifest.PendingMbox, manifest.PendingTarget)
|
||||
return nil
|
||||
}
|
||||
|
||||
type snapshotStats struct {
|
||||
IndexRecords int64
|
||||
PreciseCopied int64
|
||||
PendingMbox int64
|
||||
PendingTarget int64
|
||||
MissingArchiveRecords int64
|
||||
InvalidStageOrder int64
|
||||
IndexSHA256 string
|
||||
}
|
||||
|
||||
func snapshotDatabaseStats(db *sql.DB) (snapshotStats, error) {
|
||||
var stats snapshotStats
|
||||
queries := []struct {
|
||||
query string
|
||||
dest *int64
|
||||
}{
|
||||
{`SELECT count(*) FROM mbox_index`, &stats.IndexRecords},
|
||||
{`SELECT count(*) FROM copied WHERE body_sha256<>''`, &stats.PreciseCopied},
|
||||
{`SELECT count(*) FROM copied WHERE body_sha256<>'' AND mbox_done=0`, &stats.PendingMbox},
|
||||
{`SELECT count(*) FROM copied WHERE body_sha256<>'' AND target_done=0`, &stats.PendingTarget},
|
||||
{`SELECT count(*) FROM copied WHERE body_sha256<>'' AND target_done=1 AND mbox_done=0`, &stats.InvalidStageOrder},
|
||||
{`SELECT count(*) FROM copied c WHERE c.body_sha256<>'' AND c.mbox_done=1 AND NOT EXISTS (
|
||||
SELECT 1 FROM mbox_index i
|
||||
WHERE i.account_id=c.account_id AND i.folder=c.folder
|
||||
AND i.message_id=c.message_id AND i.body_sha256=c.body_sha256
|
||||
)`, &stats.MissingArchiveRecords},
|
||||
}
|
||||
for _, item := range queries {
|
||||
if err := db.QueryRow(item.query).Scan(item.dest); err != nil {
|
||||
return snapshotStats{}, fmt.Errorf("snapshot database statistics: %w", err)
|
||||
}
|
||||
}
|
||||
return stats, nil
|
||||
}
|
||||
|
||||
func indexDigest(db *sql.DB) (string, error) {
|
||||
rows, err := db.Query(`SELECT account_id, folder, seq, message_id, body_sha256
|
||||
FROM mbox_index ORDER BY account_id, folder, seq`)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("snapshot index digest query: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
hash := sha256.New()
|
||||
for rows.Next() {
|
||||
var accountID, seq int64
|
||||
var folder, messageID, bodyHash string
|
||||
if err := rows.Scan(&accountID, &folder, &seq, &messageID, &bodyHash); err != nil {
|
||||
return "", fmt.Errorf("snapshot index digest row: %w", err)
|
||||
}
|
||||
for _, value := range []int64{accountID, seq} {
|
||||
if err := binary.Write(hash, binary.LittleEndian, value); err != nil {
|
||||
return "", err
|
||||
}
|
||||
}
|
||||
for _, value := range []string{folder, messageID, bodyHash} {
|
||||
if err := binary.Write(hash, binary.LittleEndian, uint64(len(value))); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if _, err := io.WriteString(hash, value); err != nil {
|
||||
return "", err
|
||||
}
|
||||
}
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return "", fmt.Errorf("snapshot index digest: %w", err)
|
||||
}
|
||||
return hex.EncodeToString(hash.Sum(nil)), nil
|
||||
}
|
||||
|
||||
func collectSnapshotWatermarks(db *sql.DB, mboxRoot string) ([]snapshotWatermark, error) {
|
||||
rootAbs, err := filepath.Abs(mboxRoot)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("snapshot mbox root: %w", err)
|
||||
}
|
||||
rows, err := db.Query(`SELECT i.account_id, a.name, a.mbox_dir, i.folder,
|
||||
MAX(i.file_offset+i.frame_len), count(*)
|
||||
FROM mbox_index i JOIN accounts a ON a.id=i.account_id
|
||||
GROUP BY i.account_id, a.name, a.mbox_dir, i.folder
|
||||
ORDER BY lower(a.name), a.name, i.folder`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("snapshot watermarks: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []snapshotWatermark
|
||||
seenPaths := map[string]bool{}
|
||||
archivePaths := map[string]map[string]string{}
|
||||
for rows.Next() {
|
||||
var watermark snapshotWatermark
|
||||
if err := rows.Scan(&watermark.AccountID, &watermark.Account, &watermark.MboxDir, &watermark.Folder, &watermark.Bytes, &watermark.IndexRecords); err != nil {
|
||||
return nil, fmt.Errorf("snapshot watermark row: %w", err)
|
||||
}
|
||||
watermark.MboxDir = strings.TrimSpace(watermark.MboxDir)
|
||||
if watermark.MboxDir == "" {
|
||||
return nil, fmt.Errorf("snapshot account %s has no mbox_dir", watermark.Account)
|
||||
}
|
||||
paths, ok := archivePaths[watermark.MboxDir]
|
||||
if !ok {
|
||||
paths = map[string]string{}
|
||||
for _, path := range archiveMboxPaths(filepath.Join(rootAbs, watermark.MboxDir)) {
|
||||
paths[folderNameFromMboxPath(path)] = path
|
||||
}
|
||||
archivePaths[watermark.MboxDir] = paths
|
||||
}
|
||||
watermark.SourcePath = paths[watermark.Folder]
|
||||
if watermark.SourcePath == "" {
|
||||
return nil, fmt.Errorf("snapshot indexed mbox missing for account=%s folder=%q", watermark.Account, watermark.Folder)
|
||||
}
|
||||
pathAbs, err := filepath.Abs(watermark.SourcePath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("snapshot mbox path: %w", err)
|
||||
}
|
||||
relative, err := filepath.Rel(rootAbs, pathAbs)
|
||||
if err != nil || relative == ".." || strings.HasPrefix(relative, ".."+string(os.PathSeparator)) || filepath.IsAbs(relative) {
|
||||
return nil, fmt.Errorf("snapshot mbox escapes configured root: %s", pathAbs)
|
||||
}
|
||||
watermark.SourcePath = pathAbs
|
||||
watermark.RelativePath = relative
|
||||
pathKey := snapshotPathKey(relative)
|
||||
if seenPaths[pathKey] {
|
||||
return nil, fmt.Errorf("snapshot multiple index groups resolve to the same mbox: %s", relative)
|
||||
}
|
||||
seenPaths[pathKey] = true
|
||||
info, err := os.Stat(pathAbs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("snapshot mbox stat %s: %w", pathAbs, err)
|
||||
}
|
||||
if !info.Mode().IsRegular() {
|
||||
return nil, fmt.Errorf("snapshot mbox is not a regular file: %s", pathAbs)
|
||||
}
|
||||
if watermark.Bytes <= 0 || info.Size() < watermark.Bytes {
|
||||
return nil, fmt.Errorf("snapshot invalid watermark account=%s folder=%q boundary=%d size=%d",
|
||||
watermark.Account, watermark.Folder, watermark.Bytes, info.Size())
|
||||
}
|
||||
out = append(out, watermark)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("snapshot watermarks: %w", err)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func copyFilePrefix(source, destination string, length int64) (string, error) {
|
||||
if length < 0 {
|
||||
return "", fmt.Errorf("negative prefix length %d", length)
|
||||
}
|
||||
if err := os.MkdirAll(filepath.Dir(destination), 0o700); err != nil {
|
||||
return "", err
|
||||
}
|
||||
src, err := os.Open(source)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer src.Close()
|
||||
dst, err := os.OpenFile(destination, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
hash := sha256.New()
|
||||
written, copyErr := io.CopyN(io.MultiWriter(dst, hash), src, length)
|
||||
if copyErr == nil && written != length {
|
||||
copyErr = io.ErrShortWrite
|
||||
}
|
||||
if copyErr == nil {
|
||||
copyErr = dst.Sync()
|
||||
}
|
||||
closeErr := dst.Close()
|
||||
if copyErr != nil {
|
||||
return "", copyErr
|
||||
}
|
||||
if closeErr != nil {
|
||||
return "", closeErr
|
||||
}
|
||||
return hex.EncodeToString(hash.Sum(nil)), nil
|
||||
}
|
||||
|
||||
func openSnapshotDatabase(path string) (*sql.DB, error) {
|
||||
db, err := sql.Open("sqlite", path)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("open snapshot database: %w", err)
|
||||
}
|
||||
db.SetMaxOpenConns(1)
|
||||
if _, err := db.Exec(`PRAGMA query_only=ON; PRAGMA foreign_keys=ON;`); err != nil {
|
||||
_ = db.Close()
|
||||
return nil, fmt.Errorf("open snapshot database read-only: %w", err)
|
||||
}
|
||||
return db, nil
|
||||
}
|
||||
|
||||
// VerifySnapshot validates hashes and DB/file boundaries, then copies only the
|
||||
// small DB into a temporary restore workspace and performs a full reindex there.
|
||||
// The snapshot mbox files and any live database remain untouched.
|
||||
func VerifySnapshot(root string) error {
|
||||
root = strings.TrimSpace(root)
|
||||
if root == "" {
|
||||
return fmt.Errorf("verify snapshot: directory is empty")
|
||||
}
|
||||
rootAbs, err := filepath.Abs(root)
|
||||
if err != nil {
|
||||
return fmt.Errorf("verify snapshot path: %w", err)
|
||||
}
|
||||
manifestBytes, err := os.ReadFile(filepath.Join(rootAbs, "snapshot-manifest.json"))
|
||||
if err != nil {
|
||||
return fmt.Errorf("verify snapshot manifest: %w", err)
|
||||
}
|
||||
var manifest SnapshotManifest
|
||||
if err := json.Unmarshal(manifestBytes, &manifest); err != nil {
|
||||
return fmt.Errorf("verify snapshot manifest decode: %w", err)
|
||||
}
|
||||
if manifest.FormatVersion != snapshotFormatVersion {
|
||||
return fmt.Errorf("verify snapshot format=%d, supported=%d", manifest.FormatVersion, snapshotFormatVersion)
|
||||
}
|
||||
databasePath, err := pathInsideSnapshot(rootAbs, manifest.DatabasePath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if runtime.GOOS != "windows" {
|
||||
databaseInfo, err := os.Stat(databasePath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("verify snapshot database permissions: %w", err)
|
||||
}
|
||||
if permissions := databaseInfo.Mode().Perm(); permissions&0o077 != 0 {
|
||||
return fmt.Errorf("verify snapshot database permissions=%#o, want no group/other access", permissions)
|
||||
}
|
||||
}
|
||||
databaseHash, err := hashFile(databasePath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("verify snapshot database hash: %w", err)
|
||||
}
|
||||
if databaseHash != manifest.DatabaseSHA256 {
|
||||
return fmt.Errorf("verify snapshot database hash mismatch: got %s want %s", databaseHash, manifest.DatabaseSHA256)
|
||||
}
|
||||
for _, file := range manifest.Files {
|
||||
path, err := pathInsideSnapshot(rootAbs, file.Path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
info, err := os.Stat(path)
|
||||
if err != nil {
|
||||
return fmt.Errorf("verify snapshot file %s: %w", file.Path, err)
|
||||
}
|
||||
if info.Size() != file.Bytes {
|
||||
return fmt.Errorf("verify snapshot file size %s: got %d want %d", file.Path, info.Size(), file.Bytes)
|
||||
}
|
||||
hash, err := hashFile(path)
|
||||
if err != nil {
|
||||
return fmt.Errorf("verify snapshot file hash %s: %w", file.Path, err)
|
||||
}
|
||||
if hash != file.SHA256 {
|
||||
return fmt.Errorf("verify snapshot file hash mismatch %s: got %s want %s", file.Path, hash, file.SHA256)
|
||||
}
|
||||
}
|
||||
|
||||
snapshotDB, err := openSnapshotDatabase(databasePath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var integrity string
|
||||
if err := snapshotDB.QueryRow(`PRAGMA integrity_check`).Scan(&integrity); err != nil {
|
||||
_ = snapshotDB.Close()
|
||||
return fmt.Errorf("verify snapshot integrity: %w", err)
|
||||
}
|
||||
if integrity != "ok" {
|
||||
_ = snapshotDB.Close()
|
||||
return fmt.Errorf("verify snapshot integrity: %s", integrity)
|
||||
}
|
||||
stats, err := snapshotDatabaseStats(snapshotDB)
|
||||
if err == nil {
|
||||
stats.IndexSHA256, err = indexDigest(snapshotDB)
|
||||
}
|
||||
if err != nil {
|
||||
_ = snapshotDB.Close()
|
||||
return err
|
||||
}
|
||||
watermarks, err := collectSnapshotWatermarks(snapshotDB, filepath.Join(rootAbs, "backup"))
|
||||
if closeErr := snapshotDB.Close(); err == nil && closeErr != nil {
|
||||
err = closeErr
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if stats.IndexRecords != manifest.IndexRecords || stats.PreciseCopied != manifest.PreciseCopied ||
|
||||
stats.PendingMbox != manifest.PendingMbox || stats.PendingTarget != manifest.PendingTarget ||
|
||||
stats.InvalidStageOrder != manifest.InvalidStageOrder || stats.MissingArchiveRecords != manifest.MissingArchiveRecords ||
|
||||
stats.IndexSHA256 != manifest.IndexSHA256 {
|
||||
return fmt.Errorf("verify snapshot manifest/database statistics differ")
|
||||
}
|
||||
if stats.MissingArchiveRecords != 0 {
|
||||
return fmt.Errorf("verify snapshot has %d mbox_done identities without archive index", stats.MissingArchiveRecords)
|
||||
}
|
||||
if stats.InvalidStageOrder != 0 {
|
||||
return fmt.Errorf("verify snapshot has %d precise identities with invalid stage order", stats.InvalidStageOrder)
|
||||
}
|
||||
if err := compareManifestWatermarks(manifest.Files, watermarks); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := verifySnapshotReindex(rootAbs, databasePath, manifest.IndexRecords, manifest.IndexSHA256); err != nil {
|
||||
return err
|
||||
}
|
||||
log.Printf("snapshot verified root=%s files=%d bytes=%d index_records=%d precise_copied=%d pending_mbox=%d pending_target=%d",
|
||||
rootAbs, len(manifest.Files), snapshotBytes(manifest.Files), manifest.IndexRecords, manifest.PreciseCopied, manifest.PendingMbox, manifest.PendingTarget)
|
||||
return nil
|
||||
}
|
||||
|
||||
func compareManifestWatermarks(files []SnapshotFile, watermarks []snapshotWatermark) error {
|
||||
manifestByPath := make(map[string]SnapshotFile, len(files))
|
||||
for _, file := range files {
|
||||
key := snapshotPathKey(filepath.FromSlash(file.Path))
|
||||
if _, exists := manifestByPath[key]; exists {
|
||||
return fmt.Errorf("verify snapshot duplicate manifest path: %s", file.Path)
|
||||
}
|
||||
manifestByPath[key] = file
|
||||
}
|
||||
if len(manifestByPath) != len(watermarks) {
|
||||
return fmt.Errorf("verify snapshot file/watermark count differs: manifest=%d database=%d", len(manifestByPath), len(watermarks))
|
||||
}
|
||||
for _, watermark := range watermarks {
|
||||
path := filepath.Join("backup", watermark.RelativePath)
|
||||
key := snapshotPathKey(path)
|
||||
file, ok := manifestByPath[key]
|
||||
if !ok {
|
||||
return fmt.Errorf("verify snapshot manifest missing DB watermark: %s", filepath.ToSlash(path))
|
||||
}
|
||||
if file.Bytes != watermark.Bytes || file.IndexRecords != watermark.IndexRecords || file.AccountID != watermark.AccountID || file.Folder != watermark.Folder {
|
||||
return fmt.Errorf("verify snapshot watermark mismatch: %s", file.Path)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func verifySnapshotReindex(root, databasePath string, expectedIndexRecords int64, expectedIndexSHA256 string) error {
|
||||
temporaryRoot, err := os.MkdirTemp("", "mail-graveyard-snapshot-verify-")
|
||||
if err != nil {
|
||||
return fmt.Errorf("verify snapshot temporary restore: %w", err)
|
||||
}
|
||||
defer os.RemoveAll(temporaryRoot)
|
||||
temporaryDB := filepath.Join(temporaryRoot, "mail-graveyard.db")
|
||||
databaseInfo, err := os.Stat(databasePath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("verify snapshot temporary database stat: %w", err)
|
||||
}
|
||||
if _, err := copyFilePrefix(databasePath, temporaryDB, databaseInfo.Size()); err != nil {
|
||||
return fmt.Errorf("verify snapshot temporary database copy: %w", err)
|
||||
}
|
||||
oldCfg, oldDB := Cfg, DB
|
||||
Cfg.DBPath = temporaryDB
|
||||
Cfg.MboxRoot = filepath.Join(root, "backup")
|
||||
DB = nil
|
||||
defer func() {
|
||||
if DB != nil {
|
||||
_ = DB.Close()
|
||||
}
|
||||
Cfg, DB = oldCfg, oldDB
|
||||
}()
|
||||
if err := ConnectDB(false); err != nil {
|
||||
return fmt.Errorf("verify snapshot restore database: %w", err)
|
||||
}
|
||||
if _, err := DB.Exec(`CREATE TABLE snapshot_verify_expected_index AS
|
||||
SELECT account_id, folder, seq, message_id, body_sha256 FROM mbox_index`); err != nil {
|
||||
return fmt.Errorf("verify snapshot preserve expected index: %w", err)
|
||||
}
|
||||
if err := ReindexArchives("all"); err != nil {
|
||||
return fmt.Errorf("verify snapshot restore reindex: %w", err)
|
||||
}
|
||||
var reindexed int64
|
||||
if err := DB.QueryRow(`SELECT count(*) FROM mbox_index`).Scan(&reindexed); err != nil {
|
||||
return fmt.Errorf("verify snapshot restore index count: %w", err)
|
||||
}
|
||||
if reindexed != expectedIndexRecords {
|
||||
return fmt.Errorf("verify snapshot restore index count=%d, want %d", reindexed, expectedIndexRecords)
|
||||
}
|
||||
reindexedDigest, err := indexDigest(DB)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if reindexedDigest != expectedIndexSHA256 {
|
||||
difference, differenceErr := snapshotIndexDifference(DB)
|
||||
if differenceErr != nil {
|
||||
return fmt.Errorf("verify snapshot restore index digest=%s, want %s (difference query: %v)", reindexedDigest, expectedIndexSHA256, differenceErr)
|
||||
}
|
||||
return fmt.Errorf("verify snapshot restore index digest=%s, want %s (%s)", reindexedDigest, expectedIndexSHA256, difference)
|
||||
}
|
||||
stats, err := snapshotDatabaseStats(DB)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if stats.MissingArchiveRecords != 0 {
|
||||
return fmt.Errorf("verify snapshot restore has %d copied identities without archive index", stats.MissingArchiveRecords)
|
||||
}
|
||||
if stats.InvalidStageOrder != 0 {
|
||||
return fmt.Errorf("verify snapshot restore has %d precise identities with invalid stage order", stats.InvalidStageOrder)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func snapshotIndexDifference(db *sql.DB) (string, error) {
|
||||
type differenceRow struct {
|
||||
accountID int64
|
||||
folder string
|
||||
seq int64
|
||||
messageID string
|
||||
bodyHash string
|
||||
}
|
||||
queries := []struct {
|
||||
label string
|
||||
left string
|
||||
right string
|
||||
}{
|
||||
{"missing after reindex", "snapshot_verify_expected_index", "mbox_index"},
|
||||
{"unexpected after reindex", "mbox_index", "snapshot_verify_expected_index"},
|
||||
}
|
||||
parts := make([]string, 0, len(queries))
|
||||
for _, query := range queries {
|
||||
statement := fmt.Sprintf(`SELECT account_id, folder, seq, message_id, body_sha256 FROM %s
|
||||
EXCEPT SELECT account_id, folder, seq, message_id, body_sha256 FROM %s`, query.left, query.right)
|
||||
var count int64
|
||||
if err := db.QueryRow(`SELECT count(*) FROM (` + statement + `)`).Scan(&count); err != nil {
|
||||
return "", err
|
||||
}
|
||||
rows, err := db.Query(statement + ` LIMIT 3`)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
var samples []string
|
||||
for rows.Next() {
|
||||
var row differenceRow
|
||||
if err := rows.Scan(&row.accountID, &row.folder, &row.seq, &row.messageID, &row.bodyHash); err != nil {
|
||||
_ = rows.Close()
|
||||
return "", err
|
||||
}
|
||||
hash := row.bodyHash
|
||||
if len(hash) > 16 {
|
||||
hash = hash[:16]
|
||||
}
|
||||
samples = append(samples, fmt.Sprintf("account=%d folder=%q seq=%d id=%q hash=%s", row.accountID, row.folder, row.seq, row.messageID, hash))
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
_ = rows.Close()
|
||||
return "", err
|
||||
}
|
||||
if err := rows.Close(); err != nil {
|
||||
return "", err
|
||||
}
|
||||
parts = append(parts, fmt.Sprintf("%s=%d samples=[%s]", query.label, count, strings.Join(samples, "; ")))
|
||||
}
|
||||
return strings.Join(parts, ", "), nil
|
||||
}
|
||||
|
||||
func pathInsideSnapshot(root, relative string) (string, error) {
|
||||
if relative == "" || filepath.IsAbs(relative) {
|
||||
return "", fmt.Errorf("verify snapshot invalid relative path: %q", relative)
|
||||
}
|
||||
path := filepath.Join(root, filepath.FromSlash(relative))
|
||||
pathAbs, err := filepath.Abs(path)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
rel, err := filepath.Rel(root, pathAbs)
|
||||
if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(os.PathSeparator)) {
|
||||
return "", fmt.Errorf("verify snapshot path escapes root: %q", relative)
|
||||
}
|
||||
return pathAbs, nil
|
||||
}
|
||||
|
||||
func hashFile(path string) (string, error) {
|
||||
f, err := os.Open(path)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer f.Close()
|
||||
hash := sha256.New()
|
||||
if _, err := io.Copy(hash, f); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return hex.EncodeToString(hash.Sum(nil)), nil
|
||||
}
|
||||
|
||||
func syncFile(path string) error {
|
||||
f, err := os.OpenFile(path, os.O_RDWR, 0)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
return f.Sync()
|
||||
}
|
||||
|
||||
func snapshotBytes(files []SnapshotFile) int64 {
|
||||
var total int64
|
||||
for _, file := range files {
|
||||
total += file.Bytes
|
||||
}
|
||||
return total
|
||||
}
|
||||
|
||||
func snapshotPathKey(path string) string {
|
||||
path = filepath.Clean(path)
|
||||
if runtime.GOOS == "windows" {
|
||||
return strings.ToLower(path)
|
||||
}
|
||||
return path
|
||||
}
|
||||
|
||||
func sortSnapshotFiles(files []SnapshotFile) {
|
||||
sort.Slice(files, func(i, j int) bool { return files[i].Path < files[j].Path })
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue