feat(backup): 归档先写 .partial 再 fsync 改名并记录 sha256,删除原件前回读校验,reaper 抽样巡检与清扫半截归档,保留权限与 mtime,面板标出损坏与已校验
This commit is contained in:
25 files changed
+1489
-125
No files matched your search
+50
-11
@@ -53,13 +53,14 @@ func (s *PGStore) FreshBackup(ctx context.Context, server string, since time.Tim
|
||||
// Only the reaper's own archives count, and only those taken since the
|
||||
// current owner claimed the server: a manual backup may predate a panel edit
|
||||
// that did not move last_active_at, and an archive from before the claim is
|
||||
// the previous owner's world.
|
||||
const q = `SELECT b.backup_ref, b.offsite_at IS NOT NULL FROM world_backups b
|
||||
// the previous owner's world. One found corrupt is never reused.
|
||||
const q = `SELECT b.id, b.backup_ref, COALESCE(b.sha256, ''), b.offsite_at IS NOT NULL FROM world_backups b
|
||||
WHERE b.server_name = $1 AND b.status = 'present' AND b.reason = 'inactive_15d' AND b.created_at >= $2
|
||||
AND b.corrupt_at IS NULL
|
||||
AND b.created_at >= COALESCE((SELECT s.claimed_at FROM servers s WHERE s.name = $1 AND s.deleted_at IS NULL), '-infinity')
|
||||
ORDER BY b.offsite_at IS NOT NULL DESC, b.created_at DESC LIMIT 1`
|
||||
var f Fresh
|
||||
switch err := s.db.QueryRowContext(ctx, q, server, since).Scan(&f.Ref, &f.Offsite); {
|
||||
switch err := s.db.QueryRowContext(ctx, q, server, since).Scan(&f.ID, &f.Ref, &f.SHA256, &f.Offsite); {
|
||||
case err == sql.ErrNoRows:
|
||||
return Fresh{}, false, nil
|
||||
case err != nil:
|
||||
@@ -70,10 +71,11 @@ func (s *PGStore) FreshBackup(ctx context.Context, server string, since time.Tim
|
||||
|
||||
func (s *PGStore) InsertBackup(ctx context.Context, rec BackupRecord) error {
|
||||
const q = `INSERT INTO world_backups
|
||||
(id, server_name, former_owner, backup_ref, size_bytes, reason, status, created_at, expires_at)
|
||||
VALUES ($1, $2, NULLIF($3, ''), $4, $5, $6, 'present', now(), $7)`
|
||||
(id, server_name, former_owner, backup_ref, size_bytes, reason, status, created_at, expires_at, sha256, skipped_entries)
|
||||
VALUES ($1, $2, NULLIF($3, ''), $4, $5, $6, 'present', now(), $7, NULLIF($8, ''), $9)`
|
||||
_, err := s.db.ExecContext(ctx, q,
|
||||
rec.ID, rec.ServerName, rec.FormerOwner, rec.BackupRef, rec.SizeBytes, rec.Reason, rec.ExpiresAt)
|
||||
rec.ID, rec.ServerName, rec.FormerOwner, rec.BackupRef, rec.SizeBytes, rec.Reason, rec.ExpiresAt,
|
||||
rec.SHA256, rec.SkippedEntries)
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -107,7 +109,7 @@ func (s *PGStore) PresentBackupBytes(ctx context.Context) (int64, error) {
|
||||
}
|
||||
|
||||
func (s *PGStore) EvictableBackups(ctx context.Context) ([]StoredBackup, error) {
|
||||
const q = `SELECT id, server_name, backup_ref, size_bytes, reason FROM world_backups
|
||||
const q = `SELECT id, server_name, backup_ref, size_bytes, reason, COALESCE(sha256, '') FROM world_backups
|
||||
WHERE status = 'present' AND (reason <> 'inactive_15d' OR offsite_at IS NOT NULL)
|
||||
ORDER BY reason = 'inactive_15d', created_at ASC`
|
||||
return s.queryBackups(ctx, q)
|
||||
@@ -118,9 +120,9 @@ func (s *PGStore) EvictableBackups(ctx context.Context) ([]StoredBackup, error)
|
||||
// set, is a backup id left out of the list whatever its age (the one a chained
|
||||
// restore is about to extract).
|
||||
func (s *PGStore) ExcessBackups(ctx context.Context, server, reason string, keep int, protect string) ([]StoredBackup, error) {
|
||||
const q = `SELECT id, server_name, backup_ref, size_bytes, reason FROM world_backups
|
||||
const q = `SELECT id, server_name, backup_ref, size_bytes, reason, COALESCE(sha256, '') FROM world_backups
|
||||
WHERE server_name = $1 AND status = 'present' AND reason = $2 AND id <> $4
|
||||
ORDER BY created_at DESC OFFSET $3`
|
||||
ORDER BY corrupt_at IS NULL DESC, created_at DESC OFFSET $3`
|
||||
out, err := s.queryBackups(ctx, q, server, reason, keep, protect)
|
||||
for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 {
|
||||
out[i], out[j] = out[j], out[i]
|
||||
@@ -129,7 +131,7 @@ func (s *PGStore) ExcessBackups(ctx context.Context, server, reason string, keep
|
||||
}
|
||||
|
||||
func (s *PGStore) ListExpiredBackups(ctx context.Context, now time.Time) ([]StoredBackup, error) {
|
||||
const q = `SELECT id, server_name, backup_ref, size_bytes, reason FROM world_backups
|
||||
const q = `SELECT id, server_name, backup_ref, size_bytes, reason, COALESCE(sha256, '') FROM world_backups
|
||||
WHERE status = 'present' AND expires_at < $1 ORDER BY expires_at ASC`
|
||||
return s.queryBackups(ctx, q, now)
|
||||
}
|
||||
@@ -143,7 +145,7 @@ func (s *PGStore) queryBackups(ctx context.Context, q string, args ...any) ([]St
|
||||
var out []StoredBackup
|
||||
for rows.Next() {
|
||||
var b StoredBackup
|
||||
if err := rows.Scan(&b.ID, &b.ServerName, &b.BackupRef, &b.SizeBytes, &b.Reason); err != nil {
|
||||
if err := rows.Scan(&b.ID, &b.ServerName, &b.BackupRef, &b.SizeBytes, &b.Reason, &b.SHA256); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, b)
|
||||
@@ -151,6 +153,43 @@ func (s *PGStore) queryBackups(ctx context.Context, q string, args ...any) ([]St
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func (s *PGStore) BackupsToVerify(ctx context.Context, checkedBefore time.Time, limit int) ([]StoredBackup, error) {
|
||||
const q = `SELECT id, server_name, backup_ref, size_bytes, reason, COALESCE(sha256, '') FROM world_backups
|
||||
WHERE status = 'present' AND corrupt_at IS NULL AND (verified_at IS NULL OR verified_at < $1)
|
||||
ORDER BY verified_at NULLS FIRST, created_at LIMIT $2`
|
||||
return s.queryBackups(ctx, q, checkedBefore, limit)
|
||||
}
|
||||
|
||||
func (s *PGStore) MarkBackupVerified(ctx context.Context, id, sha256 string, at time.Time) error {
|
||||
_, err := s.db.ExecContext(ctx,
|
||||
`UPDATE world_backups SET verified_at = $3, sha256 = COALESCE(sha256, NULLIF($2, '')) WHERE id = $1`,
|
||||
id, sha256, at)
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *PGStore) MarkBackupCorrupt(ctx context.Context, id string, at time.Time) error {
|
||||
_, err := s.db.ExecContext(ctx,
|
||||
`UPDATE world_backups SET corrupt_at = $2 WHERE id = $1 AND corrupt_at IS NULL`, id, at)
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *PGStore) LiveBackupRefs(ctx context.Context) ([]string, error) {
|
||||
rows, err := s.db.QueryContext(ctx, `SELECT backup_ref FROM world_backups WHERE status <> 'deleted'`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []string
|
||||
for rows.Next() {
|
||||
var ref string
|
||||
if err := rows.Scan(&ref); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, ref)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func (s *PGStore) MarkBackupDeleted(ctx context.Context, id string, at time.Time) error {
|
||||
_, err := s.db.ExecContext(ctx,
|
||||
`UPDATE world_backups SET status = 'deleted', deleted_at = $2 WHERE id = $1`, id, at)
|
||||
|
||||
+173
-19
@@ -97,6 +97,14 @@ type Config struct {
|
||||
// world is deleted on the first run after the copy lands, normally the
|
||||
// next day.
|
||||
RequireOffsite bool
|
||||
// VerifyEvery is how often each stored archive is read back in full, and
|
||||
// VerifyPerRun caps how many one run reads (the ones checked longest ago
|
||||
// first), so bit rot in a backup is found before a restore needs it.
|
||||
VerifyEvery time.Duration
|
||||
VerifyPerRun int
|
||||
// PartialAfter is how old an unfinished archive file must be before it is
|
||||
// swept: longer than any archive takes to write.
|
||||
PartialAfter time.Duration
|
||||
}
|
||||
|
||||
// DefaultConfig is the spec's §24 default window set.
|
||||
@@ -110,6 +118,10 @@ func DefaultConfig() Config {
|
||||
ManualRetention: 30 * Day,
|
||||
ManualKeep: 5,
|
||||
ManualCooldown: 10 * time.Minute,
|
||||
|
||||
VerifyEvery: 7 * Day,
|
||||
VerifyPerRun: 10,
|
||||
PartialAfter: 6 * time.Hour,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -149,11 +161,17 @@ type BackupRecord struct {
|
||||
SizeBytes int64
|
||||
Reason string
|
||||
ExpiresAt time.Time
|
||||
// SHA256 is the archive's digest as written ("" when the backend keeps none)
|
||||
// and SkippedEntries the world entries it could not hold.
|
||||
SHA256 string
|
||||
SkippedEntries int
|
||||
}
|
||||
|
||||
// Fresh is the backup FreshBackup found.
|
||||
type Fresh struct {
|
||||
Ref string
|
||||
ID string
|
||||
Ref string
|
||||
SHA256 string
|
||||
// Offsite reports that the archive has its off-site copy (offsite_at).
|
||||
Offsite bool
|
||||
}
|
||||
@@ -166,6 +184,7 @@ type StoredBackup struct {
|
||||
BackupRef string
|
||||
SizeBytes int64
|
||||
Reason string
|
||||
SHA256 string // "" when none was recorded
|
||||
}
|
||||
|
||||
// AuditRecord is a reaper-sourced audit_logs entry. The PG binding fills
|
||||
@@ -186,9 +205,10 @@ type Store interface {
|
||||
// FreshBackup reports an existing present reaper archive (reason
|
||||
// inactive_15d) for server whose world is still current — created at or
|
||||
// after since (the world's last_active_at) and after the current claim,
|
||||
// preferring one already copied off-site. It makes a reap idempotent across
|
||||
// a DeletePVC failure, and across the wait for the off-site copy: the retry
|
||||
// reuses the archive instead of writing a duplicate.
|
||||
// preferring one already copied off-site, and never one found corrupt. It
|
||||
// makes a reap idempotent across a DeletePVC failure, and across the wait for
|
||||
// the off-site copy: the retry reuses the archive instead of writing a
|
||||
// duplicate.
|
||||
FreshBackup(ctx context.Context, server string, since time.Time) (b Fresh, ok bool, err error)
|
||||
|
||||
// InsertBackup records a world_backups row (status=present).
|
||||
@@ -218,6 +238,19 @@ type Store interface {
|
||||
// MarkBackupDeleted flips a backup to status=deleted, deleted_at=at.
|
||||
MarkBackupDeleted(ctx context.Context, id string, at time.Time) error
|
||||
|
||||
// BackupsToVerify lists up to limit present backups not found corrupt and
|
||||
// not read back since checkedBefore, the ones never read back first, then
|
||||
// the ones read back longest ago.
|
||||
BackupsToVerify(ctx context.Context, checkedBefore time.Time, limit int) ([]StoredBackup, error)
|
||||
// MarkBackupVerified records a read-back that matched at `at`, and the
|
||||
// archive's digest when none was recorded yet.
|
||||
MarkBackupVerified(ctx context.Context, id, sha256 string, at time.Time) error
|
||||
// MarkBackupCorrupt records a read-back that failed: the backup is no
|
||||
// longer reused for a reap or offered for a restore.
|
||||
MarkBackupCorrupt(ctx context.Context, id string, at time.Time) error
|
||||
// LiveBackupRefs lists the backup_ref of every backup not deleted.
|
||||
LiveBackupRefs(ctx context.Context) ([]string, error)
|
||||
|
||||
// Audit appends a reaper-sourced audit_logs row.
|
||||
Audit(ctx context.Context, rec AuditRecord) error
|
||||
}
|
||||
@@ -276,12 +309,27 @@ type Summary struct {
|
||||
// AwaitingOffsite are idle worlds that are archived and kept until the
|
||||
// archive's off-site copy lands.
|
||||
AwaitingOffsite int
|
||||
// Verified are archives read back in full and found matching; Corrupt are
|
||||
// the ones that were not (marked, and never reused or restored from);
|
||||
// VerifyFailed are the ones that could not be read back at all this run.
|
||||
Verified int
|
||||
Corrupt int
|
||||
VerifyFailed int
|
||||
// Swept are leftover files of interrupted archives removed; OrphanArchives
|
||||
// are finished archives no backup records, kept for now (see
|
||||
// backup.Swept); SweepFailed reports that the sweep did not complete.
|
||||
Swept int
|
||||
OrphanArchives int
|
||||
SweepFailed bool
|
||||
}
|
||||
|
||||
// Failed reports whether the run left work undone: a server it could not
|
||||
// process, or an expired backup it could not remove. The world is safe either
|
||||
// way, but the run did not do its job and whoever operates it must hear.
|
||||
func (s Summary) Failed() bool { return s.Skipped > 0 || s.ExpireFailed > 0 }
|
||||
// Failed reports whether the run left work undone or found damage: a server it
|
||||
// could not process, an expired backup it could not remove, an archive that did
|
||||
// not read back or could not be read, or a sweep that did not complete. The
|
||||
// worlds are safe either way, but whoever operates the run must hear.
|
||||
func (s Summary) Failed() bool {
|
||||
return s.Skipped > 0 || s.ExpireFailed > 0 || s.Corrupt > 0 || s.VerifyFailed > 0 || s.SweepFailed
|
||||
}
|
||||
|
||||
func (r *Reaper) now() time.Time {
|
||||
if r.Now != nil {
|
||||
@@ -340,6 +388,8 @@ func (r *Reaper) RunOnce(ctx context.Context) (Summary, error) {
|
||||
}
|
||||
|
||||
r.expireBackups(ctx, now, &sum)
|
||||
r.verifyBackups(ctx, now, &sum)
|
||||
r.sweepArchives(ctx, now, &sum)
|
||||
return sum, nil
|
||||
}
|
||||
|
||||
@@ -393,32 +443,49 @@ func (r *Reaper) reap(ctx context.Context, now time.Time, c Candidate, crd Serve
|
||||
if err != nil {
|
||||
return fmt.Errorf("lookup fresh backup: %w", err)
|
||||
}
|
||||
if ok {
|
||||
// The archive may have sat on disk for days (a delete that failed, the
|
||||
// wait for its off-site copy); it is about to become the only copy, so it
|
||||
// is read back first. One that fails is marked and replaced by a fresh
|
||||
// archive of the world, which is still there.
|
||||
good, err := r.verifyOne(ctx, now, fresh.ID, fresh.Ref, fresh.SHA256, sum)
|
||||
if err != nil {
|
||||
return fmt.Errorf("read back archive %s: %w", fresh.ID, err)
|
||||
}
|
||||
ok = good
|
||||
}
|
||||
ref, offsite := fresh.Ref, fresh.Offsite
|
||||
if !ok {
|
||||
aref, size, err := r.Archiver.Archive(ctx, c.Name, crd.PVC)
|
||||
a, err := r.Archiver.Archive(ctx, c.Name, crd.PVC)
|
||||
if err != nil {
|
||||
// Red line ④: archive failed → the PVC is untouched, the world
|
||||
// survives, and this server is retried next run.
|
||||
return fmt.Errorf("archive: %w", err)
|
||||
}
|
||||
if len(a.Skipped) > 0 {
|
||||
r.log().Warn("reaper: archive leaves out entries that are not plain files or directories",
|
||||
"server", c.Name, "count", len(a.Skipped), "first", a.Skipped[:min(len(a.Skipped), 5)])
|
||||
}
|
||||
rec := BackupRecord{
|
||||
ID: r.id(),
|
||||
ServerName: c.Name,
|
||||
FormerOwner: c.OwnerID,
|
||||
BackupRef: string(aref),
|
||||
SizeBytes: size,
|
||||
Reason: ReasonInactive,
|
||||
ExpiresAt: now.Add(r.Cfg.Retention),
|
||||
ID: r.id(),
|
||||
ServerName: c.Name,
|
||||
FormerOwner: c.OwnerID,
|
||||
BackupRef: string(a.Ref),
|
||||
SizeBytes: a.Size,
|
||||
Reason: ReasonInactive,
|
||||
ExpiresAt: now.Add(r.Cfg.Retention),
|
||||
SHA256: a.SHA256,
|
||||
SkippedEntries: len(a.Skipped),
|
||||
}
|
||||
if err := r.Store.InsertBackup(ctx, rec); err != nil {
|
||||
// The archive exists but is untracked. Delete the orphan so it does
|
||||
// not leak, then fail without touching the PVC.
|
||||
if derr := r.Archiver.Delete(ctx, aref); derr != nil {
|
||||
r.log().Error("reaper: orphan archive cleanup failed", "server", c.Name, "ref", aref, "err", derr)
|
||||
if derr := r.Archiver.Delete(ctx, a.Ref); derr != nil {
|
||||
r.log().Error("reaper: orphan archive cleanup failed", "server", c.Name, "ref", a.Ref, "err", derr)
|
||||
}
|
||||
return fmt.Errorf("insert backup: %w", err)
|
||||
}
|
||||
ref, offsite = string(aref), false
|
||||
ref, offsite = string(a.Ref), false
|
||||
}
|
||||
|
||||
// With an off-site bucket configured, the archive on this node's disk is
|
||||
@@ -570,6 +637,93 @@ func (r *Reaper) expireBackups(ctx context.Context, now time.Time, sum *Summary)
|
||||
}
|
||||
}
|
||||
|
||||
// verifyBackups is the read-back pass: up to VerifyPerRun present archives not
|
||||
// read back within VerifyEvery are read end to end and checked against their
|
||||
// recorded digest. A backend that cannot read its archives back skips it.
|
||||
func (r *Reaper) verifyBackups(ctx context.Context, now time.Time, sum *Summary) {
|
||||
if _, ok := r.Archiver.(backup.Verifier); !ok || r.Cfg.VerifyPerRun <= 0 {
|
||||
return
|
||||
}
|
||||
list, err := r.Store.BackupsToVerify(ctx, now.Add(-r.Cfg.VerifyEvery), r.Cfg.VerifyPerRun)
|
||||
if err != nil {
|
||||
r.log().Error("reaper: list backups to read back", "err", err)
|
||||
sum.VerifyFailed++
|
||||
return
|
||||
}
|
||||
for _, b := range list {
|
||||
if _, err := r.verifyOne(ctx, now, b.ID, b.BackupRef, b.SHA256, sum); err != nil {
|
||||
r.log().Error("reaper: read back archive", "id", b.ID, "server", b.ServerName, "err", err)
|
||||
sum.VerifyFailed++
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// verifyOne reads one archive back and records the outcome. It reports whether
|
||||
// the archive is good; an error means it could not be read back at all (the
|
||||
// store is not mounted, the run was cancelled), which says nothing about the
|
||||
// archive. A backend that cannot read its archives back reports every archive
|
||||
// good, as before read-backs existed.
|
||||
func (r *Reaper) verifyOne(ctx context.Context, now time.Time, id, ref, want string, sum *Summary) (bool, error) {
|
||||
v, ok := r.Archiver.(backup.Verifier)
|
||||
if !ok {
|
||||
return true, nil
|
||||
}
|
||||
got, err := v.Verify(ctx, backup.ArchiveRef(ref), want)
|
||||
switch {
|
||||
case errors.Is(err, backup.ErrCorrupt):
|
||||
sum.Corrupt++
|
||||
r.log().Error("reaper: archive is corrupt; it will not be reused or offered for restore", "id", id, "ref", ref, "err", err)
|
||||
if err := r.Store.MarkBackupCorrupt(ctx, id, now); err != nil {
|
||||
return false, fmt.Errorf("mark corrupt: %w", err)
|
||||
}
|
||||
return false, nil
|
||||
case err != nil:
|
||||
return false, err
|
||||
}
|
||||
if err := r.Store.MarkBackupVerified(ctx, id, got, now); err != nil {
|
||||
r.log().Error("reaper: record read-back", "id", id, "err", err)
|
||||
}
|
||||
sum.Verified++
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// sweepArchives removes what interrupted archives left in the store (see
|
||||
// backup.Sweeper): unfinished files older than PartialAfter, and finished ones
|
||||
// no backup records once they are older than Retention, the longest any backup
|
||||
// is kept. Younger unrecorded archives are reported and kept.
|
||||
func (r *Reaper) sweepArchives(ctx context.Context, now time.Time, sum *Summary) {
|
||||
sw, ok := r.Archiver.(backup.Sweeper)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
refs, err := r.Store.LiveBackupRefs(ctx)
|
||||
if err != nil {
|
||||
// Without the list every archive would look unclaimed.
|
||||
r.log().Error("reaper: list recorded archives; sweep skipped", "err", err)
|
||||
sum.SweepFailed = true
|
||||
return
|
||||
}
|
||||
live := make(map[backup.ArchiveRef]bool, len(refs))
|
||||
for _, ref := range refs {
|
||||
live[backup.ArchiveRef(ref)] = true
|
||||
}
|
||||
res, err := sw.Sweep(ctx, func(ref backup.ArchiveRef) bool { return live[ref] },
|
||||
now.Add(-r.Cfg.PartialAfter), now.Add(-r.Cfg.Retention))
|
||||
if err != nil {
|
||||
r.log().Error("reaper: sweep archive store", "err", err)
|
||||
sum.SweepFailed = true
|
||||
}
|
||||
for _, p := range res.Removed {
|
||||
r.log().Info("reaper: removed leftover of an interrupted archive", "path", p)
|
||||
}
|
||||
sum.Swept += len(res.Removed)
|
||||
sum.OrphanArchives += len(res.Orphans)
|
||||
if len(res.Orphans) > 0 {
|
||||
r.log().Warn("reaper: archives no backup records are kept until they are older than the retention",
|
||||
"count", len(res.Orphans), "bytes", res.OrphanBytes, "first", res.Orphans[:min(len(res.Orphans), 5)])
|
||||
}
|
||||
}
|
||||
|
||||
// formatRemaining renders an offset as the human-facing time left before reap.
|
||||
func formatRemaining(d time.Duration) string {
|
||||
if d%Day == 0 {
|
||||
|
||||
@@ -40,14 +40,48 @@ type fakeArchiver struct {
|
||||
seq int
|
||||
}
|
||||
|
||||
func (f *fakeArchiver) Archive(_ context.Context, server, _ string) (backup.ArchiveRef, int64, error) {
|
||||
func (f *fakeArchiver) Archive(_ context.Context, server, _ string) (backup.Archived, error) {
|
||||
if f.archiveErr != nil {
|
||||
return "", 0, f.archiveErr
|
||||
return backup.Archived{}, f.archiveErr
|
||||
}
|
||||
f.archives++
|
||||
f.seq++
|
||||
f.rec.add("archive")
|
||||
return backup.ArchiveRef(fmt.Sprintf("ref-%s-%d", server, f.seq)), 10, nil
|
||||
ref := fmt.Sprintf("ref-%s-%d", server, f.seq)
|
||||
return backup.Archived{Ref: backup.ArchiveRef(ref), Size: 10, SHA256: "sha-" + ref}, nil
|
||||
}
|
||||
|
||||
// checkingArchiver is a fakeArchiver that also reads archives back
|
||||
// (backup.Verifier) and sweeps its store (backup.Sweeper).
|
||||
type checkingArchiver struct {
|
||||
*fakeArchiver
|
||||
corrupt map[string]bool // refs that no longer read back
|
||||
verifyErr error // every read-back fails with this
|
||||
verified []string
|
||||
|
||||
sweepErr error
|
||||
sweepLive func(backup.ArchiveRef) bool
|
||||
sweepCutoffs []time.Time
|
||||
sweepRemoved []string
|
||||
sweepOrphaned []string
|
||||
}
|
||||
|
||||
func (c *checkingArchiver) Verify(_ context.Context, ref backup.ArchiveRef, want string) (string, error) {
|
||||
if c.verifyErr != nil {
|
||||
return "", c.verifyErr
|
||||
}
|
||||
c.verified = append(c.verified, string(ref))
|
||||
c.rec.add("verify")
|
||||
if c.corrupt[string(ref)] {
|
||||
return "", fmt.Errorf("%w: %s", backup.ErrCorrupt, ref)
|
||||
}
|
||||
return "sha-" + string(ref), nil
|
||||
}
|
||||
|
||||
func (c *checkingArchiver) Sweep(_ context.Context, live func(backup.ArchiveRef) bool, partialBefore, orphanBefore time.Time) (backup.Swept, error) {
|
||||
c.sweepLive = live
|
||||
c.sweepCutoffs = []time.Time{partialBefore, orphanBefore}
|
||||
return backup.Swept{Removed: c.sweepRemoved, Orphans: c.sweepOrphaned, OrphanBytes: int64(len(c.sweepOrphaned))}, c.sweepErr
|
||||
}
|
||||
|
||||
func (f *fakeArchiver) Restore(context.Context, backup.ArchiveRef, string) error { return nil }
|
||||
@@ -109,6 +143,10 @@ type fakeBackup struct {
|
||||
status string // present | deleted
|
||||
createdAt, expires time.Time
|
||||
offsite bool
|
||||
sha string
|
||||
skipped int
|
||||
verifiedAt time.Time
|
||||
corruptAt time.Time
|
||||
}
|
||||
|
||||
type fakeStore struct {
|
||||
@@ -121,6 +159,7 @@ type fakeStore struct {
|
||||
released []string
|
||||
listErr error
|
||||
insertErr error
|
||||
liveErr error
|
||||
idn int
|
||||
}
|
||||
|
||||
@@ -138,7 +177,7 @@ func (s *fakeStore) ListActiveServers(context.Context) ([]Candidate, error) {
|
||||
func (s *fakeStore) FreshBackup(_ context.Context, server string, since time.Time) (Fresh, bool, error) {
|
||||
var found *fakeBackup
|
||||
for _, b := range s.backups {
|
||||
if b.server == server && b.status == "present" && b.reason == ReasonInactive && !b.createdAt.Before(since) {
|
||||
if b.server == server && b.status == "present" && b.reason == ReasonInactive && !b.createdAt.Before(since) && b.corruptAt.IsZero() {
|
||||
if found == nil || (b.offsite && !found.offsite) {
|
||||
found = b
|
||||
}
|
||||
@@ -147,7 +186,7 @@ func (s *fakeStore) FreshBackup(_ context.Context, server string, since time.Tim
|
||||
if found == nil {
|
||||
return Fresh{}, false, nil
|
||||
}
|
||||
return Fresh{Ref: found.ref, Offsite: found.offsite}, true, nil
|
||||
return Fresh{ID: found.id, Ref: found.ref, SHA256: found.sha, Offsite: found.offsite}, true, nil
|
||||
}
|
||||
|
||||
func (s *fakeStore) InsertBackup(_ context.Context, rec BackupRecord) error {
|
||||
@@ -156,7 +195,7 @@ func (s *fakeStore) InsertBackup(_ context.Context, rec BackupRecord) error {
|
||||
}
|
||||
s.backups = append(s.backups, &fakeBackup{
|
||||
id: rec.ID, server: rec.ServerName, ref: rec.BackupRef, reason: rec.Reason, size: rec.SizeBytes,
|
||||
status: "present", createdAt: s.clock, expires: rec.ExpiresAt,
|
||||
status: "present", createdAt: s.clock, expires: rec.ExpiresAt, sha: rec.SHA256, skipped: rec.SkippedEntries,
|
||||
})
|
||||
s.rec.add("insert")
|
||||
return nil
|
||||
@@ -233,6 +272,73 @@ func (s *fakeStore) MarkBackupDeleted(_ context.Context, id string, at time.Time
|
||||
return fmt.Errorf("no backup %s", id)
|
||||
}
|
||||
|
||||
func (s *fakeStore) BackupsToVerify(_ context.Context, checkedBefore time.Time, limit int) ([]StoredBackup, error) {
|
||||
var ps []*fakeBackup
|
||||
for _, b := range s.backups {
|
||||
if b.status == "present" && b.corruptAt.IsZero() && b.verifiedAt.Before(checkedBefore) {
|
||||
ps = append(ps, b)
|
||||
}
|
||||
}
|
||||
sort.SliceStable(ps, func(i, j int) bool {
|
||||
if !ps[i].verifiedAt.Equal(ps[j].verifiedAt) {
|
||||
return ps[i].verifiedAt.Before(ps[j].verifiedAt)
|
||||
}
|
||||
return ps[i].createdAt.Before(ps[j].createdAt)
|
||||
})
|
||||
var out []StoredBackup
|
||||
for _, b := range ps {
|
||||
if len(out) == limit {
|
||||
break
|
||||
}
|
||||
out = append(out, StoredBackup{ID: b.id, ServerName: b.server, BackupRef: b.ref, SizeBytes: b.size, Reason: b.reason, SHA256: b.sha})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *fakeStore) find(id string) (*fakeBackup, error) {
|
||||
for _, b := range s.backups {
|
||||
if b.id == id {
|
||||
return b, nil
|
||||
}
|
||||
}
|
||||
return nil, fmt.Errorf("no backup %s", id)
|
||||
}
|
||||
|
||||
func (s *fakeStore) MarkBackupVerified(_ context.Context, id, sha string, at time.Time) error {
|
||||
b, err := s.find(id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
b.verifiedAt = at
|
||||
if b.sha == "" {
|
||||
b.sha = sha
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *fakeStore) MarkBackupCorrupt(_ context.Context, id string, at time.Time) error {
|
||||
b, err := s.find(id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
b.corruptAt = at
|
||||
s.rec.add("corrupt")
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *fakeStore) LiveBackupRefs(context.Context) ([]string, error) {
|
||||
if s.liveErr != nil {
|
||||
return nil, s.liveErr
|
||||
}
|
||||
var out []string
|
||||
for _, b := range s.backups {
|
||||
if b.status != "deleted" {
|
||||
out = append(out, b.ref)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *fakeStore) Audit(_ context.Context, rec AuditRecord) error {
|
||||
s.audits = append(s.audits, rec)
|
||||
s.rec.add("audit:" + rec.Action)
|
||||
@@ -808,3 +914,146 @@ func TestListErrorAbortsBatch(t *testing.T) {
|
||||
t.Fatal("expected a hard error when listing servers fails")
|
||||
}
|
||||
}
|
||||
|
||||
// checking swaps the fixture's archiver for one that reads archives back and
|
||||
// sweeps.
|
||||
func checking(r *Reaper, ar *fakeArchiver) *checkingArchiver {
|
||||
c := &checkingArchiver{fakeArchiver: ar, corrupt: map[string]bool{}}
|
||||
r.Archiver = c
|
||||
return c
|
||||
}
|
||||
|
||||
// An archive left from an earlier run is read back before the world it holds is
|
||||
// deleted, and the read-back is recorded.
|
||||
func TestReapReadsBackReusedArchive(t *testing.T) {
|
||||
r, st, cl, ar := newReaper(DefaultConfig(),
|
||||
Candidate{Name: "kilo", OwnerID: "user-1", LastActiveAt: idleBy(20 * Day)})
|
||||
ca := checking(r, ar)
|
||||
st.backups = []*fakeBackup{{id: "old", server: "kilo", ref: "ref-old", reason: ReasonInactive, size: 5,
|
||||
status: "present", createdAt: idleBy(Day), expires: testNow.Add(89 * Day), sha: "sha-ref-old"}}
|
||||
|
||||
sum := mustRun(t, r)
|
||||
if sum.WorldsReaped != 1 || ar.archives != 0 || len(cl.deletedPVCs) != 1 {
|
||||
t.Fatalf("summary %+v archives=%d deleted=%v: want the checked archive reused", sum, ar.archives, cl.deletedPVCs)
|
||||
}
|
||||
if st.rec.events[0] != "verify" || st.rec.events[1] != "deletePVC" {
|
||||
t.Errorf("events = %v, want the read-back before the delete", st.rec.events)
|
||||
}
|
||||
if !st.backups[0].verifiedAt.Equal(testNow) || !reflect.DeepEqual(ca.verified, []string{"ref-old"}) {
|
||||
t.Errorf("read-back not recorded: %+v", st.backups[0])
|
||||
}
|
||||
}
|
||||
|
||||
// A reused archive that no longer reads back is marked corrupt and the world,
|
||||
// still there, is archived afresh before it is deleted. The run reports it.
|
||||
func TestReapReplacesCorruptArchive(t *testing.T) {
|
||||
r, st, cl, ar := newReaper(DefaultConfig(),
|
||||
Candidate{Name: "lima", OwnerID: "user-1", LastActiveAt: idleBy(20 * Day)})
|
||||
ca := checking(r, ar)
|
||||
ca.corrupt["ref-rotten"] = true
|
||||
st.backups = []*fakeBackup{{id: "rotten", server: "lima", ref: "ref-rotten", reason: ReasonInactive, size: 5,
|
||||
status: "present", createdAt: idleBy(Day), expires: testNow.Add(89 * Day)}}
|
||||
|
||||
sum := mustRun(t, r)
|
||||
if sum.WorldsReaped != 1 || sum.Corrupt != 1 || !sum.Failed() {
|
||||
t.Fatalf("summary = %+v, want reaped with one corrupt archive reported", sum)
|
||||
}
|
||||
want := []string{"verify", "corrupt", "archive", "insert", "deletePVC"}
|
||||
if !reflect.DeepEqual(st.rec.events[:len(want)], want) {
|
||||
t.Errorf("events = %v, want %v first", st.rec.events, want)
|
||||
}
|
||||
if st.backups[0].corruptAt.IsZero() || len(st.backups) != 2 || st.backups[1].sha != "sha-ref-lima-1" {
|
||||
t.Errorf("backups = %+v %+v", st.backups[0], st.backups[len(st.backups)-1])
|
||||
}
|
||||
if len(cl.deletedPVCs) != 1 {
|
||||
t.Errorf("deleted = %v", cl.deletedPVCs)
|
||||
}
|
||||
}
|
||||
|
||||
// When the archive cannot be read back at all (the store is not mounted), the
|
||||
// world is kept and nothing is marked: the error says nothing about the archive.
|
||||
func TestReapKeepsWorldWhenReadBackFails(t *testing.T) {
|
||||
r, st, cl, ar := newReaper(DefaultConfig(),
|
||||
Candidate{Name: "mike", OwnerID: "user-1", LastActiveAt: idleBy(20 * Day)})
|
||||
ca := checking(r, ar)
|
||||
ca.verifyErr = errors.New("input/output error")
|
||||
st.backups = []*fakeBackup{{id: "b", server: "mike", ref: "ref-b", reason: ReasonInactive, size: 5,
|
||||
status: "present", createdAt: idleBy(Day), expires: testNow.Add(89 * Day)}}
|
||||
|
||||
sum := mustRun(t, r)
|
||||
if sum.WorldsReaped != 0 || sum.Skipped != 1 || cl.deletePVCCalls != 0 || ar.archives != 0 {
|
||||
t.Fatalf("summary %+v deletes=%d archives=%d: want the world kept", sum, cl.deletePVCCalls, ar.archives)
|
||||
}
|
||||
if !st.backups[0].corruptAt.IsZero() {
|
||||
t.Error("an unreadable store marked the archive corrupt")
|
||||
}
|
||||
}
|
||||
|
||||
// The read-back pass takes VerifyPerRun archives, never-checked ones first, and
|
||||
// records what it finds; a corrupt one is marked and fails the run once.
|
||||
func TestVerifyPassSamplesArchives(t *testing.T) {
|
||||
cfg := DefaultConfig()
|
||||
cfg.VerifyPerRun = 2
|
||||
r, st, _, ar := newReaper(cfg)
|
||||
ca := checking(r, ar)
|
||||
ca.corrupt["ref-c"] = true
|
||||
st.backups = []*fakeBackup{
|
||||
{id: "a", ref: "ref-a", status: "present", createdAt: idleBy(9 * Day), verifiedAt: idleBy(8 * Day), expires: testNow.Add(30 * Day)},
|
||||
{id: "b", ref: "ref-b", status: "present", createdAt: idleBy(9 * Day), verifiedAt: idleBy(Day), expires: testNow.Add(30 * Day)},
|
||||
{id: "c", ref: "ref-c", status: "present", createdAt: idleBy(5 * Day), expires: testNow.Add(30 * Day)},
|
||||
{id: "d", ref: "ref-d", status: "deleted", createdAt: idleBy(5 * Day), expires: testNow.Add(30 * Day)},
|
||||
}
|
||||
|
||||
sum := mustRun(t, r)
|
||||
if !reflect.DeepEqual(ca.verified, []string{"ref-c", "ref-a"}) {
|
||||
t.Errorf("read back %v, want [ref-c ref-a]", ca.verified)
|
||||
}
|
||||
if sum.Verified != 1 || sum.Corrupt != 1 || !sum.Failed() {
|
||||
t.Errorf("summary = %+v", sum)
|
||||
}
|
||||
if st.backups[2].corruptAt.IsZero() || !st.backups[0].verifiedAt.Equal(testNow) || st.backups[0].sha != "sha-ref-a" {
|
||||
t.Errorf("outcomes not recorded: %+v %+v", st.backups[0], st.backups[2])
|
||||
}
|
||||
|
||||
// The next run skips the corrupt one and what was checked within
|
||||
// VerifyEvery; a week on, b and a are due again, b first.
|
||||
ca.verified = nil
|
||||
if sum := mustRun(t, r); sum.Corrupt != 0 || len(ca.verified) != 0 {
|
||||
t.Errorf("second run read back %v (%+v), want nothing", ca.verified, sum)
|
||||
}
|
||||
r.Now = func() time.Time { return testNow.Add(7*Day + time.Hour) }
|
||||
if mustRun(t, r); !reflect.DeepEqual(ca.verified, []string{"ref-b", "ref-a"}) {
|
||||
t.Errorf("a week later read back %v, want [ref-b ref-a]", ca.verified)
|
||||
}
|
||||
}
|
||||
|
||||
// The sweep claims every archive a backup records, removes leftovers past
|
||||
// PartialAfter and orphans past the retention, and never runs without the list
|
||||
// of recorded archives.
|
||||
func TestSweepPass(t *testing.T) {
|
||||
r, st, _, ar := newReaper(DefaultConfig())
|
||||
ca := checking(r, ar)
|
||||
ca.sweepRemoved = []string{"/archives/.x-1.tar.gz.partial"}
|
||||
ca.sweepOrphaned = []string{"/archives/x-2.tar.gz"}
|
||||
st.backups = []*fakeBackup{
|
||||
{id: "a", ref: "ref-a", status: "present", expires: testNow.Add(Day), verifiedAt: testNow},
|
||||
{id: "d", ref: "ref-d", status: "deleted", expires: testNow.Add(Day)},
|
||||
}
|
||||
sum := mustRun(t, r)
|
||||
if sum.Swept != 1 || sum.OrphanArchives != 1 || sum.Failed() {
|
||||
t.Errorf("summary = %+v", sum)
|
||||
}
|
||||
if !ca.sweepLive("ref-a") || ca.sweepLive("ref-d") {
|
||||
t.Error("sweep did not get the recorded archives as live")
|
||||
}
|
||||
cfg := DefaultConfig()
|
||||
if want := []time.Time{testNow.Add(-cfg.PartialAfter), testNow.Add(-cfg.Retention)}; !reflect.DeepEqual(ca.sweepCutoffs, want) {
|
||||
t.Errorf("cutoffs = %v, want %v", ca.sweepCutoffs, want)
|
||||
}
|
||||
|
||||
ca.sweepLive = nil
|
||||
st.liveErr = errors.New("db down")
|
||||
if sum := mustRun(t, r); !sum.SweepFailed || ca.sweepLive != nil {
|
||||
t.Errorf("sweep ran without the recorded archives: %+v", sum)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user