Compare commits
7
Commits
b4284b2170
..
next
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
fa56165fe4 | ||
|
|
34f61fb408 | ||
|
|
05bf73c48a | ||
|
|
79a73fa122 | ||
|
|
0d0368df81 | ||
|
|
c06c4d201b | ||
|
|
70f008a21a |
@@ -359,7 +359,8 @@ on the destination in one go, use `vaultik remote nuke --force`.
|
|||||||
* `--local-only`: Skip remote cleanup; only touch the local index
|
* `--local-only`: Skip remote cleanup; only touch the local index
|
||||||
* `--dry-run`: Show what would be deleted without deleting
|
* `--dry-run`: Show what would be deleted without deleting
|
||||||
* `--force`: Skip confirmation prompt
|
* `--force`: Skip confirmation prompt
|
||||||
* `--json`: Output result as JSON
|
* `--json`: Output result as JSON. Also skips the confirmation prompt, as
|
||||||
|
`--force` does.
|
||||||
|
|
||||||
**`snapshot restore`**: Restore files from a backup snapshot.
|
**`snapshot restore`**: Restore files from a backup snapshot.
|
||||||
* Requires `VAULTIK_AGE_SECRET_KEY` environment variable
|
* Requires `VAULTIK_AGE_SECRET_KEY` environment variable
|
||||||
@@ -383,7 +384,8 @@ manifests — network cost scales with the number of snapshots. `snapshot
|
|||||||
create --prune` runs the same cleanup automatically; this is the
|
create --prune` runs the same cleanup automatically; this is the
|
||||||
manual entry point for the same work.
|
manual entry point for the same work.
|
||||||
* `--force`: Skip confirmation prompt
|
* `--force`: Skip confirmation prompt
|
||||||
* `--json`: Output stats as JSON
|
* `--json`: Output stats as JSON. Also skips the confirmation prompt, as
|
||||||
|
`--force` does.
|
||||||
|
|
||||||
**`info`**: Display system configuration, storage settings, encryption
|
**`info`**: Display system configuration, storage settings, encryption
|
||||||
recipients, and local database statistics.
|
recipients, and local database statistics.
|
||||||
@@ -396,7 +398,8 @@ key is skipped with a warning and is not printed. If a listed
|
|||||||
orphaned blob figures are reported as unknown; `--json` gives them as
|
orphaned blob figures are reported as unknown; `--json` gives them as
|
||||||
`null`, lists the remote key of each unreadable manifest in
|
`null`, lists the remote key of each unreadable manifest in
|
||||||
`unreadable_manifests` and counts the manifests under skipped names in
|
`unreadable_manifests` and counts the manifests under skipped names in
|
||||||
`skipped_manifest_count`.
|
`skipped_manifest_count`. An unreadable manifest also leaves its
|
||||||
|
snapshot's blob count and blob size unknown, `null` in `--json`.
|
||||||
* `--json`: Output as JSON
|
* `--json`: Output as JSON
|
||||||
|
|
||||||
**`remote nuke`**: Delete every snapshot's metadata and every blob from the
|
**`remote nuke`**: Delete every snapshot's metadata and every blob from the
|
||||||
|
|||||||
@@ -22,14 +22,58 @@ the tag exists and is exercised; what is left is merging `next` to
|
|||||||
|
|
||||||
# Completed Steps
|
# Completed Steps
|
||||||
|
|
||||||
|
- 2026-10-08: Made a backup cancelled under `--skip-errors` stop at once
|
||||||
|
([issue #286](https://git.eeqj.de/sneak/vaultik/issues/286)). After
|
||||||
|
Ctrl-C or SIGTERM, phase 2 treated the cancellation error from each
|
||||||
|
remaining file like an unreadable file: it opened the file, printed an
|
||||||
|
error line for it, counted it as failed and went on to the next one.
|
||||||
|
The run now stops at the first file after the cancellation, and no
|
||||||
|
file is counted as failed because of it.
|
||||||
|
|
||||||
|
- 2026-10-08: Made `remote nuke` delete the `.partial` files that
|
||||||
|
uploads cut off part-way leave on the destination store
|
||||||
|
([issue #281](https://git.eeqj.de/sneak/vaultik/issues/281)). The
|
||||||
|
file and rclone listings skip such a file, so the command left it in
|
||||||
|
place and still reported the store empty. It now removes them under
|
||||||
|
`metadata/` and `blobs/` as its last step.
|
||||||
|
|
||||||
|
- 2026-10-08: Made a local index error while recording a directory or
|
||||||
|
symlink stop a backup under `--skip-errors`
|
||||||
|
([issue #284](https://git.eeqj.de/sneak/vaultik/issues/284)). Phase 2
|
||||||
|
only records such an entry in the local index, since it has no data
|
||||||
|
to open or read, but an error doing so was skipped like an unreadable
|
||||||
|
file. The snapshot then completed without the entry, and a restore
|
||||||
|
did not recreate it. The error now stops the backup, as it does
|
||||||
|
without the flag.
|
||||||
|
|
||||||
|
- 2026-10-08: Counted a file that a backup could not store as failed
|
||||||
|
([issue #280](https://git.eeqj.de/sneak/vaultik/issues/280)). A file
|
||||||
|
that phase 1 counted and phase 2 could not open, because it was
|
||||||
|
unreadable under `--skip-errors` or removed in between, was reported
|
||||||
|
in the summary as unchanged with its bytes as backed up, and the
|
||||||
|
`snapshots` row's `file_count` and `total_size` included it. The
|
||||||
|
summary now counts it as failed, and its data total and the row leave
|
||||||
|
it out.
|
||||||
|
|
||||||
|
- 2026-10-08: Made `remote info` report a snapshot's blob count and
|
||||||
|
blob size as unknown when its manifest cannot be read
|
||||||
|
([issue #272](https://git.eeqj.de/sneak/vaultik/issues/272)). The
|
||||||
|
orphan figures were already unknown in that case, but the snapshot's
|
||||||
|
row still gave 0 blobs and 0 B, in the table and in `--json`. The row
|
||||||
|
now reads `unknown` and `--json` gives `null`. A directory with no
|
||||||
|
manifest still shows 0, since the orphan figures count its blobs as
|
||||||
|
orphaned.
|
||||||
|
|
||||||
- 2026-10-08: Kept the progress line of a snapshot with more than one
|
- 2026-10-08: Kept the progress line of a snapshot with more than one
|
||||||
path within 100%
|
path within 100%
|
||||||
([issue #271](https://git.eeqj.de/sneak/vaultik/issues/271)). The
|
([issue #271](https://git.eeqj.de/sneak/vaultik/issues/271)). The
|
||||||
bytes and files processed counted every path, but the totals they
|
bytes and files processed counted every path, but the totals they
|
||||||
were divided by held only the current path's, so a second path
|
were divided by held only the current path's, so a second path
|
||||||
smaller than the first showed more than 100% and an ETA of `unknown`.
|
smaller than the first showed more than 100% and an ETA of `unknown`.
|
||||||
The totals now add up over the paths scanned so far, and the rate is
|
The totals now add up over the paths scanned so far. The rate behind
|
||||||
measured from when the first path's processing started.
|
the ETA restarts when each path's scan phase ends, so that phase does
|
||||||
|
not lower the rate while the path is processed; until then the rate
|
||||||
|
is the previous path's.
|
||||||
|
|
||||||
- 2026-10-08: Made a second `snapshot create` of one name succeed when
|
- 2026-10-08: Made a second `snapshot create` of one name succeed when
|
||||||
it starts in the same second as the first
|
it starts in the same second as the first
|
||||||
@@ -39,6 +83,14 @@ the tag exists and is exercised; what is left is merging `next` to
|
|||||||
snapshots.id`. When the local index already has a snapshot with the
|
snapshots.id`. When the local index already has a snapshot with the
|
||||||
ID, the create now waits a second and takes a new timestamp.
|
ID, the create now waits a second and takes a new timestamp.
|
||||||
|
|
||||||
|
- 2026-10-07: Documented that `--json` skips the confirmation prompt of
|
||||||
|
`snapshot remove` and `prune`
|
||||||
|
([issue #268](https://git.eeqj.de/sneak/vaultik/issues/268)). Both
|
||||||
|
commands delete without asking under `--json`, since a prompt on stdout
|
||||||
|
would break the JSON document, but the help and the README described
|
||||||
|
only `--force` as skipping it. The `--json` help of both commands and
|
||||||
|
their README entries now say so.
|
||||||
|
|
||||||
- 2026-10-07: Made a symlink whose target cannot be read stop the backup
|
- 2026-10-07: Made a symlink whose target cannot be read stop the backup
|
||||||
([issue #269](https://git.eeqj.de/sneak/vaultik/issues/269)). It was
|
([issue #269](https://git.eeqj.de/sneak/vaultik/issues/269)). It was
|
||||||
left out of the snapshot with only a debug log line, even without
|
left out of the snapshot with only a debug log line, even without
|
||||||
|
|||||||
@@ -0,0 +1,24 @@
|
|||||||
|
package cli //nolint:testpackage // exercises the unexported command constructor
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/spf13/cobra"
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestJSONHelpSaysConfirmationPromptIsSkipped checks that the --json help
|
||||||
|
// of `snapshot remove` and `prune` says the flag skips the confirmation
|
||||||
|
// prompt. Both delete without asking under --json, since a prompt on
|
||||||
|
// stdout would break the JSON document.
|
||||||
|
func TestJSONHelpSaysConfirmationPromptIsSkipped(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
for _, cmd := range []*cobra.Command{
|
||||||
|
newSnapshotRemoveCommand(),
|
||||||
|
NewPruneCommand(),
|
||||||
|
} {
|
||||||
|
assert.Contains(t, cmd.Flags().Lookup("json").Usage,
|
||||||
|
"skips the confirmation prompt", cmd.Name())
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -57,7 +57,8 @@ referenced.`,
|
|||||||
}
|
}
|
||||||
|
|
||||||
cmd.Flags().BoolVar(&opts.Force, "force", false, "Skip confirmation prompt")
|
cmd.Flags().BoolVar(&opts.Force, "force", false, "Skip confirmation prompt")
|
||||||
cmd.Flags().BoolVar(&opts.JSON, "json", false, "Output pruning stats as JSON")
|
cmd.Flags().BoolVar(&opts.JSON, "json", false,
|
||||||
|
"Output pruning stats as JSON; skips the confirmation prompt, as --force does")
|
||||||
|
|
||||||
return cmd
|
return cmd
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -280,7 +280,8 @@ nuke --force' — it is the single supported entry point for that.`,
|
|||||||
cmd.Flags().BoolVarP(&opts.Force, "force", "f", false, "Skip confirmation prompt")
|
cmd.Flags().BoolVarP(&opts.Force, "force", "f", false, "Skip confirmation prompt")
|
||||||
cmd.Flags().BoolVar(&opts.DryRun, "dry-run", false,
|
cmd.Flags().BoolVar(&opts.DryRun, "dry-run", false,
|
||||||
"Show what would be removed without removing")
|
"Show what would be removed without removing")
|
||||||
cmd.Flags().BoolVar(&opts.JSON, "json", false, "Output result as JSON")
|
cmd.Flags().BoolVar(&opts.JSON, "json", false,
|
||||||
|
"Output result as JSON; skips the confirmation prompt, as --force does")
|
||||||
cmd.Flags().BoolVar(&opts.LocalOnly, "local-only", false,
|
cmd.Flags().BoolVar(&opts.LocalOnly, "local-only", false,
|
||||||
"Skip remote cleanup; only touch the local index")
|
"Skip remote cleanup; only touch the local index")
|
||||||
|
|
||||||
|
|||||||
@@ -56,23 +56,27 @@ const (
|
|||||||
|
|
||||||
// ProgressStats holds atomic counters for progress tracking
|
// ProgressStats holds atomic counters for progress tracking
|
||||||
type ProgressStats struct {
|
type ProgressStats struct {
|
||||||
FilesScanned atomic.Int64 // Total files seen during scan (includes skipped)
|
FilesScanned atomic.Int64 // Total files seen during scan (includes skipped)
|
||||||
FilesProcessed atomic.Int64 // Files actually processed in phase 2
|
FilesProcessed atomic.Int64 // Files actually processed in phase 2
|
||||||
FilesSkipped atomic.Int64 // Files skipped due to no changes
|
FilesSkipped atomic.Int64 // Files skipped due to no changes
|
||||||
BytesScanned atomic.Int64 // Bytes from new/changed files only
|
BytesScanned atomic.Int64 // Bytes from new/changed files only
|
||||||
BytesSkipped atomic.Int64 // Bytes from unchanged files
|
BytesSkipped atomic.Int64 // Bytes from unchanged files
|
||||||
BytesProcessed atomic.Int64 // Actual bytes processed (for ETA calculation)
|
BytesProcessed atomic.Int64 // Actual bytes processed (for ETA calculation)
|
||||||
ChunksCreated atomic.Int64
|
ChunksCreated atomic.Int64
|
||||||
BlobsCreated atomic.Int64
|
BlobsCreated atomic.Int64
|
||||||
BlobsUploaded atomic.Int64
|
BlobsUploaded atomic.Int64
|
||||||
BytesUploaded atomic.Int64
|
BytesUploaded atomic.Int64
|
||||||
CurrentFile atomic.Value // stores string
|
CurrentFile atomic.Value // stores string
|
||||||
TotalSize atomic.Int64 // Size to process in the paths scanned so far
|
TotalSize atomic.Int64 // Size to process in the paths scanned so far
|
||||||
TotalFiles atomic.Int64 // Files to process in the paths scanned so far
|
TotalFiles atomic.Int64 // Files to process in the paths scanned so far
|
||||||
ProcessStartTime atomic.Value // stores time.Time; set by the first path
|
StartTime time.Time
|
||||||
StartTime time.Time
|
mu sync.RWMutex
|
||||||
mu sync.RWMutex
|
lastDetailTime time.Time
|
||||||
lastDetailTime time.Time
|
|
||||||
|
// Guarded by mu: when the last scan phase ended, and
|
||||||
|
// BytesProcessed at that moment.
|
||||||
|
processStartTime time.Time
|
||||||
|
processStartBytes int64
|
||||||
|
|
||||||
// Upload tracking
|
// Upload tracking
|
||||||
CurrentUpload atomic.Value // stores *UploadInfo
|
CurrentUpload atomic.Value // stores *UploadInfo
|
||||||
@@ -149,16 +153,35 @@ func (pr *ProgressReporter) GetStats() *ProgressStats {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// AddTotalSize adds the size one path of the snapshot has to process to
|
// AddTotalSize adds the size one path of the snapshot has to process to
|
||||||
// the total, once that path's scan phase is done. The processed counts
|
// the total, once that path's scan phase is done, and starts measuring
|
||||||
// run across every path, so the total does too, and the processing start
|
// the processing rate again. The processed counts run across every path,
|
||||||
// time, which the rate is measured from, is the first path's.
|
// so the total does too. Restarting the rate keeps a path's scan phase,
|
||||||
|
// which processes nothing, out of the rate while that path is processed.
|
||||||
|
// During a later path's scan phase the rate is still the previous
|
||||||
|
// path's, and falls as that scan goes on.
|
||||||
func (pr *ProgressReporter) AddTotalSize(size int64) {
|
func (pr *ProgressReporter) AddTotalSize(size int64) {
|
||||||
pr.stats.TotalSize.Add(size)
|
pr.stats.TotalSize.Add(size)
|
||||||
|
|
||||||
_, started := pr.stats.ProcessStartTime.Load().(time.Time)
|
pr.stats.mu.Lock()
|
||||||
if !started {
|
defer pr.stats.mu.Unlock()
|
||||||
pr.stats.ProcessStartTime.Store(time.Now().UTC())
|
|
||||||
|
pr.stats.processStartTime = time.Now().UTC()
|
||||||
|
pr.stats.processStartBytes = pr.stats.BytesProcessed.Load()
|
||||||
|
}
|
||||||
|
|
||||||
|
// processRate returns the bytes processed per second since the last scan
|
||||||
|
// phase ended, or 0 before the first path's scan phase is done.
|
||||||
|
func (s *ProgressStats) processRate() float64 {
|
||||||
|
s.mu.RLock()
|
||||||
|
defer s.mu.RUnlock()
|
||||||
|
|
||||||
|
if s.processStartTime.IsZero() {
|
||||||
|
return 0
|
||||||
}
|
}
|
||||||
|
|
||||||
|
processed := s.BytesProcessed.Load() - s.processStartBytes
|
||||||
|
|
||||||
|
return float64(processed) / time.Since(s.processStartTime).Seconds()
|
||||||
}
|
}
|
||||||
|
|
||||||
// Helper functions
|
// Helper functions
|
||||||
@@ -364,19 +387,12 @@ func (pr *ProgressReporter) printSummaryStatus() {
|
|||||||
// Calculate ETA if we have total size and are processing
|
// Calculate ETA if we have total size and are processing
|
||||||
etaStr := ""
|
etaStr := ""
|
||||||
|
|
||||||
if totalSize > 0 && bytesProcessed > 0 {
|
processRate := pr.stats.processRate()
|
||||||
processStart, ok := pr.stats.ProcessStartTime.Load().(time.Time)
|
if totalSize > 0 && processRate > 0 {
|
||||||
if ok && !processStart.IsZero() {
|
remainingBytes := totalSize - bytesProcessed
|
||||||
processElapsed := time.Since(processStart)
|
remainingSeconds := float64(remainingBytes) / processRate
|
||||||
|
eta := time.Duration(remainingSeconds * float64(time.Second))
|
||||||
rate := float64(bytesProcessed) / processElapsed.Seconds()
|
etaStr = " | ETA: " + formatDuration(eta)
|
||||||
if rate > 0 {
|
|
||||||
remainingBytes := totalSize - bytesProcessed
|
|
||||||
remainingSeconds := float64(remainingBytes) / rate
|
|
||||||
eta := time.Duration(remainingSeconds * float64(time.Second))
|
|
||||||
etaStr = " | ETA: " + formatDuration(eta)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
rate := float64(bytesScanned+bytesSkipped) / elapsed.Seconds()
|
rate := float64(bytesScanned+bytesSkipped) / elapsed.Seconds()
|
||||||
@@ -428,25 +444,18 @@ func (pr *ProgressReporter) printDetailedStatus() {
|
|||||||
log.Info("Elapsed time", "duration", formatDuration(elapsed))
|
log.Info("Elapsed time", "duration", formatDuration(elapsed))
|
||||||
|
|
||||||
// Calculate and show ETA if we have data
|
// Calculate and show ETA if we have data
|
||||||
if totalSize > 0 && bytesProcessed > 0 {
|
processRate := pr.stats.processRate()
|
||||||
processStart, ok := pr.stats.ProcessStartTime.Load().(time.Time)
|
if totalSize > 0 && processRate > 0 {
|
||||||
if ok && !processStart.IsZero() {
|
remainingBytes := totalSize - bytesProcessed
|
||||||
processElapsed := time.Since(processStart)
|
remainingSeconds := float64(remainingBytes) / processRate
|
||||||
|
eta := time.Duration(remainingSeconds * float64(time.Second))
|
||||||
processRate := float64(bytesProcessed) / processElapsed.Seconds()
|
percentComplete := float64(bytesProcessed) / float64(totalSize) * percentScale
|
||||||
if processRate > 0 {
|
log.Info("Overall progress",
|
||||||
remainingBytes := totalSize - bytesProcessed
|
"percent", fmt.Sprintf("%.1f%%", percentComplete),
|
||||||
remainingSeconds := float64(remainingBytes) / processRate
|
"processed", humanize.Bytes(safeUint64(bytesProcessed)),
|
||||||
eta := time.Duration(remainingSeconds * float64(time.Second))
|
"total", humanize.Bytes(safeUint64(totalSize)),
|
||||||
percentComplete := float64(bytesProcessed) / float64(totalSize) * percentScale
|
"rate", humanize.Bytes(uint64(processRate))+"/s",
|
||||||
log.Info("Overall progress",
|
"eta", formatDuration(eta))
|
||||||
"percent", fmt.Sprintf("%.1f%%", percentComplete),
|
|
||||||
"processed", humanize.Bytes(safeUint64(bytesProcessed)),
|
|
||||||
"total", humanize.Bytes(safeUint64(totalSize)),
|
|
||||||
"rate", humanize.Bytes(uint64(processRate))+"/s",
|
|
||||||
"eta", formatDuration(eta))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
log.Info("Files processed",
|
log.Info("Files processed",
|
||||||
|
|||||||
@@ -0,0 +1,54 @@
|
|||||||
|
//nolint:testpackage // exercises the unexported processRate helper
|
||||||
|
package snapshot
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestProcessRateNotLoweredByLaterScanPhase feeds the reporter two paths
|
||||||
|
// that each take 10 seconds to process, the second after a 30-second scan
|
||||||
|
// phase. That scan phase processes nothing, so the rate while the second
|
||||||
|
// path is processed must be that path's own.
|
||||||
|
func TestProcessRateNotLoweredByLaterScanPhase(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
const (
|
||||||
|
pathSize = 1000
|
||||||
|
processingTime = 10 * time.Second
|
||||||
|
scanTime = 30 * time.Second
|
||||||
|
)
|
||||||
|
|
||||||
|
// Never started, but Stop releases its tickers and signal handler.
|
||||||
|
progress := NewProgressReporter()
|
||||||
|
defer progress.Stop()
|
||||||
|
|
||||||
|
stats := progress.GetStats()
|
||||||
|
|
||||||
|
// Moves the processing start time back instead of sleeping.
|
||||||
|
elapse := func(d time.Duration) {
|
||||||
|
stats.mu.Lock()
|
||||||
|
defer stats.mu.Unlock()
|
||||||
|
|
||||||
|
stats.processStartTime = stats.processStartTime.Add(-d)
|
||||||
|
}
|
||||||
|
|
||||||
|
progress.AddTotalSize(pathSize)
|
||||||
|
stats.BytesProcessed.Add(pathSize)
|
||||||
|
elapse(processingTime)
|
||||||
|
|
||||||
|
elapse(scanTime)
|
||||||
|
progress.AddTotalSize(pathSize)
|
||||||
|
stats.BytesProcessed.Add(pathSize)
|
||||||
|
elapse(processingTime)
|
||||||
|
|
||||||
|
want := pathSize / processingTime.Seconds()
|
||||||
|
got := stats.processRate()
|
||||||
|
|
||||||
|
// The test's own run time adds to the elapsed time, so got is a hair
|
||||||
|
// under want.
|
||||||
|
if got > want || got < want*0.99 {
|
||||||
|
t.Errorf("rate while the second path is processed is %.1f bytes/s, "+
|
||||||
|
"want %.1f", got, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -133,10 +133,13 @@ type ScannerConfig struct {
|
|||||||
|
|
||||||
// ScanResult contains the results of a scan operation. Files and bytes
|
// ScanResult contains the results of a scan operation. Files and bytes
|
||||||
// are counted per file: BytesScanned is the size of the new and changed
|
// are counted per file: BytesScanned is the size of the new and changed
|
||||||
// files, BytesSkipped that of the unchanged ones.
|
// files, BytesSkipped that of the unchanged ones. FilesFailed counts the
|
||||||
|
// new and changed files that phase 2 could not store; FilesScanned
|
||||||
|
// includes them and BytesScanned does not.
|
||||||
type ScanResult struct {
|
type ScanResult struct {
|
||||||
FilesScanned int
|
FilesScanned int
|
||||||
FilesSkipped int
|
FilesSkipped int
|
||||||
|
FilesFailed int
|
||||||
FilesDeleted int
|
FilesDeleted int
|
||||||
BytesScanned int64
|
BytesScanned int64
|
||||||
BytesSkipped int64
|
BytesSkipped int64
|
||||||
@@ -1309,6 +1312,13 @@ func (s *Scanner) processPhase(
|
|||||||
|
|
||||||
// Process each file
|
// Process each file
|
||||||
for _, fileToProcess := range filesToProcess {
|
for _, fileToProcess := range filesToProcess {
|
||||||
|
// Check context cancellation
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return ctx.Err()
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
|
||||||
// Update progress
|
// Update progress
|
||||||
if s.progress != nil {
|
if s.progress != nil {
|
||||||
s.progress.GetStats().CurrentFile.Store(fileToProcess.Path)
|
s.progress.GetStats().CurrentFile.Store(fileToProcess.Path)
|
||||||
@@ -1345,13 +1355,32 @@ func (s *Scanner) processPhase(
|
|||||||
return s.finalizeProcessPhase(ctx, result)
|
return s.finalizeProcessPhase(ctx, result)
|
||||||
}
|
}
|
||||||
|
|
||||||
// processFileWithErrorHandling wraps processFileStreaming with error recovery for
|
// processFileWithErrorHandling records a directory or symlink, or wraps
|
||||||
// deleted files and skip-errors mode. Returns (skipped, error).
|
// processFileStreaming for a regular file with error recovery for deleted
|
||||||
|
// files and skip-errors mode. Returns (skipped, error).
|
||||||
func (s *Scanner) processFileWithErrorHandling(
|
func (s *Scanner) processFileWithErrorHandling(
|
||||||
ctx context.Context, fileToProcess *FileToProcess, result *ScanResult,
|
ctx context.Context, fileToProcess *FileToProcess, result *ScanResult,
|
||||||
) (bool, error) {
|
) (bool, error) {
|
||||||
|
// A directory or symlink has no data to open or read; it is only
|
||||||
|
// recorded in the local index. An error recording it stops the run
|
||||||
|
// even under --skip-errors, or the snapshot would complete without it.
|
||||||
|
mode := os.FileMode(fileToProcess.File.Mode)
|
||||||
|
if mode&os.ModeSymlink != 0 || mode.IsDir() {
|
||||||
|
err := s.recordNonRegularFile(ctx, fileToProcess)
|
||||||
|
if err != nil {
|
||||||
|
return false, fmt.Errorf("processing file %s: %w", fileToProcess.Path, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return false, nil
|
||||||
|
}
|
||||||
|
|
||||||
err := s.processFileStreaming(ctx, fileToProcess, result)
|
err := s.processFileStreaming(ctx, fileToProcess, result)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
// A cancelled run stops here even under --skip-errors, rather than
|
||||||
|
// counting this file as failed and going on to the next one.
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
return false, fmt.Errorf("processing file %s: %w", fileToProcess.Path, err)
|
||||||
|
}
|
||||||
// A packer/database/encryption/upload failure means the chunk's data
|
// A packer/database/encryption/upload failure means the chunk's data
|
||||||
// may not have been stored. Skipping the file would let the snapshot
|
// may not have been stored. Skipping the file would let the snapshot
|
||||||
// record a file whose chunk is in no blob and cannot be restored, so
|
// record a file whose chunk is in no blob and cannot be restored, so
|
||||||
@@ -1365,7 +1394,7 @@ func (s *Scanner) processFileWithErrorHandling(
|
|||||||
log.Warn("File was deleted during backup, skipping",
|
log.Warn("File was deleted during backup, skipping",
|
||||||
"path", fileToProcess.Path)
|
"path", fileToProcess.Path)
|
||||||
|
|
||||||
result.FilesSkipped++
|
countFailedFile(fileToProcess, result)
|
||||||
|
|
||||||
return true, nil
|
return true, nil
|
||||||
}
|
}
|
||||||
@@ -1376,7 +1405,7 @@ func (s *Scanner) processFileWithErrorHandling(
|
|||||||
s.ui.Errorf("Failed to process %s: %v. Skipping (--skip-errors).",
|
s.ui.Errorf("Failed to process %s: %v. Skipping (--skip-errors).",
|
||||||
s.ui.Path(fileToProcess.Path), err)
|
s.ui.Path(fileToProcess.Path), err)
|
||||||
|
|
||||||
result.FilesSkipped++
|
countFailedFile(fileToProcess, result)
|
||||||
|
|
||||||
return true, nil
|
return true, nil
|
||||||
}
|
}
|
||||||
@@ -1387,6 +1416,13 @@ func (s *Scanner) processFileWithErrorHandling(
|
|||||||
return false, nil
|
return false, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// countFailedFile counts a file that phase 2 could not store as failed
|
||||||
|
// and takes its size back out of BytesScanned, where phase 1 put it.
|
||||||
|
func countFailedFile(fileToProcess *FileToProcess, result *ScanResult) {
|
||||||
|
result.FilesFailed++
|
||||||
|
result.BytesScanned -= fileToProcess.FileInfo.Size()
|
||||||
|
}
|
||||||
|
|
||||||
// printProcessingProgress prints a periodic progress line during the process phase,
|
// printProcessingProgress prints a periodic progress line during the process phase,
|
||||||
// showing files processed, bytes transferred, throughput, and ETA
|
// showing files processed, bytes transferred, throughput, and ETA
|
||||||
func (s *Scanner) printProcessingProgress(
|
func (s *Scanner) printProcessingProgress(
|
||||||
@@ -1755,16 +1791,11 @@ func (e *packerError) Error() string { return e.err.Error() }
|
|||||||
|
|
||||||
func (e *packerError) Unwrap() error { return e.err }
|
func (e *packerError) Unwrap() error { return e.err }
|
||||||
|
|
||||||
// processFileStreaming processes a file by streaming chunks directly to the packer
|
// processFileStreaming processes a regular file by streaming chunks directly
|
||||||
|
// to the packer
|
||||||
func (s *Scanner) processFileStreaming(
|
func (s *Scanner) processFileStreaming(
|
||||||
ctx context.Context, fileToProcess *FileToProcess, result *ScanResult,
|
ctx context.Context, fileToProcess *FileToProcess, result *ScanResult,
|
||||||
) error {
|
) error {
|
||||||
// Symlinks and directories have no data to chunk — just record them in the DB.
|
|
||||||
mode := os.FileMode(fileToProcess.File.Mode)
|
|
||||||
if mode&os.ModeSymlink != 0 || mode.IsDir() {
|
|
||||||
return s.recordNonRegularFile(ctx, fileToProcess)
|
|
||||||
}
|
|
||||||
|
|
||||||
file, err := s.fs.Open(fileToProcess.Path)
|
file, err := s.fs.Open(fileToProcess.Path)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("opening file: %w", wrapPermissionError(fileToProcess.Path, err))
|
return fmt.Errorf("opening file: %w", wrapPermissionError(fileToProcess.Path, err))
|
||||||
|
|||||||
@@ -82,6 +82,32 @@ func (f *readFailFs) Open(name string) (afero.File, error) {
|
|||||||
return file, nil
|
return file, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// cancelOnOpenFs cancels the run when the scanner opens the target file to
|
||||||
|
// back it up, as Ctrl-C partway through processing would, and records each
|
||||||
|
// file opened after that.
|
||||||
|
type cancelOnOpenFs struct {
|
||||||
|
afero.Fs
|
||||||
|
|
||||||
|
target string
|
||||||
|
cancel context.CancelFunc
|
||||||
|
cancelled bool
|
||||||
|
openedAfterCancel []string
|
||||||
|
}
|
||||||
|
|
||||||
|
//nolint:ireturn // afero.Fs.Open is defined to return the interface.
|
||||||
|
func (f *cancelOnOpenFs) Open(name string) (afero.File, error) {
|
||||||
|
if f.cancelled {
|
||||||
|
f.openedAfterCancel = append(f.openedAfterCancel, name)
|
||||||
|
}
|
||||||
|
|
||||||
|
if name == f.target {
|
||||||
|
f.cancel()
|
||||||
|
f.cancelled = true
|
||||||
|
}
|
||||||
|
|
||||||
|
return f.Fs.Open(name)
|
||||||
|
}
|
||||||
|
|
||||||
// linkRemovedAfterLstatFs is the real filesystem, except that the symlink at
|
// linkRemovedAfterLstatFs is the real filesystem, except that the symlink at
|
||||||
// target is removed right after the walk lstats it, as happens when a link is
|
// target is removed right after the walk lstats it, as happens when a link is
|
||||||
// deleted during a backup. The scanner's readlink of it then fails.
|
// deleted during a backup. The scanner's readlink of it then fails.
|
||||||
@@ -128,15 +154,16 @@ func writeSkipErrorTestFile(t *testing.T, fs afero.Fs, path, content string) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// runSkipErrorScan scans source on fs with the given skip-errors setting,
|
// runSkipErrorScan scans source on fs under ctx with the given skip-errors
|
||||||
// printing user-facing messages to uiw (nil discards them), and returns the
|
// setting, printing user-facing messages to uiw (nil discards them), and
|
||||||
// repositories (for inspection) and the scan error.
|
// returns the repositories (for inspection) and the scan error.
|
||||||
func runSkipErrorScan(
|
func runSkipErrorScan(
|
||||||
t *testing.T, fs afero.Fs, source string, skipErrors bool, uiw *ui.Writer,
|
ctx context.Context, t *testing.T, fs afero.Fs, source string,
|
||||||
|
skipErrors bool, uiw *ui.Writer,
|
||||||
) (*database.Repositories, error) {
|
) (*database.Repositories, error) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
|
||||||
db, err := database.NewTestDB()
|
db, err := database.New(ctx, ":memory:")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("create test db: %v", err)
|
t.Fatalf("create test db: %v", err)
|
||||||
}
|
}
|
||||||
@@ -161,7 +188,6 @@ func runSkipErrorScan(
|
|||||||
SkipErrors: skipErrors,
|
SkipErrors: skipErrors,
|
||||||
})
|
})
|
||||||
|
|
||||||
ctx := context.Background()
|
|
||||||
snapshotID := "test-snapshot-skip-errors"
|
snapshotID := "test-snapshot-skip-errors"
|
||||||
createTestSnapshotRecord(ctx, t, repos, snapshotID)
|
createTestSnapshotRecord(ctx, t, repos, snapshotID)
|
||||||
|
|
||||||
@@ -185,7 +211,7 @@ func TestScannerPackingFailureAbortsUnderSkipErrors(t *testing.T) {
|
|||||||
writeSkipErrorTestFile(t, fs, "/source/file1.txt", "first file content")
|
writeSkipErrorTestFile(t, fs, "/source/file1.txt", "first file content")
|
||||||
writeSkipErrorTestFile(t, fs, "/source/file2.txt", "second file content")
|
writeSkipErrorTestFile(t, fs, "/source/file2.txt", "second file content")
|
||||||
|
|
||||||
repos, err := runSkipErrorScan(t, fs, "/source", true, nil)
|
repos, err := runSkipErrorScan(context.Background(), t, fs, "/source", true, nil)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
t.Fatal("expected scan to abort on the packer error, got nil")
|
t.Fatal("expected scan to abort on the packer error, got nil")
|
||||||
}
|
}
|
||||||
@@ -212,7 +238,7 @@ func TestScannerReadErrorAbortsWithoutSkipErrors(t *testing.T) {
|
|||||||
fs := &readFailFs{Fs: afero.NewMemMapFs(), target: target}
|
fs := &readFailFs{Fs: afero.NewMemMapFs(), target: target}
|
||||||
writeSkipErrorTestFile(t, fs, target, "content that cannot be read")
|
writeSkipErrorTestFile(t, fs, target, "content that cannot be read")
|
||||||
|
|
||||||
_, err := runSkipErrorScan(t, fs, "/source", false, nil)
|
_, err := runSkipErrorScan(context.Background(), t, fs, "/source", false, nil)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
t.Fatal("expected scan to fail on the read error, got nil")
|
t.Fatal("expected scan to fail on the read error, got nil")
|
||||||
}
|
}
|
||||||
@@ -228,7 +254,7 @@ func TestScannerReadErrorSkippedWithSkipErrors(t *testing.T) {
|
|||||||
fs := &readFailFs{Fs: afero.NewMemMapFs(), target: target}
|
fs := &readFailFs{Fs: afero.NewMemMapFs(), target: target}
|
||||||
writeSkipErrorTestFile(t, fs, target, "content that cannot be read")
|
writeSkipErrorTestFile(t, fs, target, "content that cannot be read")
|
||||||
|
|
||||||
repos, err := runSkipErrorScan(t, fs, "/source", true, nil)
|
repos, err := runSkipErrorScan(context.Background(), t, fs, "/source", true, nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("expected scan to complete with --skip-errors, got %v", err)
|
t.Fatalf("expected scan to complete with --skip-errors, got %v", err)
|
||||||
}
|
}
|
||||||
@@ -267,7 +293,7 @@ func TestScannerUnreadableSymlinkAbortsWithoutSkipErrors(t *testing.T) {
|
|||||||
sourceDir, linkPath := writeSymlinkSource(t)
|
sourceDir, linkPath := writeSymlinkSource(t)
|
||||||
fs := &linkRemovedAfterLstatFs{t: t, target: linkPath}
|
fs := &linkRemovedAfterLstatFs{t: t, target: linkPath}
|
||||||
|
|
||||||
_, err := runSkipErrorScan(t, fs, sourceDir, false, nil)
|
_, err := runSkipErrorScan(context.Background(), t, fs, sourceDir, false, nil)
|
||||||
if !errors.Is(err, os.ErrNotExist) {
|
if !errors.Is(err, os.ErrNotExist) {
|
||||||
t.Fatalf("expected scan to fail on the removed symlink, got %v", err)
|
t.Fatalf("expected scan to fail on the removed symlink, got %v", err)
|
||||||
}
|
}
|
||||||
@@ -283,7 +309,7 @@ func TestScannerUnreadableSymlinkSkippedWithSkipErrors(t *testing.T) {
|
|||||||
fs := &linkRemovedAfterLstatFs{t: t, target: linkPath}
|
fs := &linkRemovedAfterLstatFs{t: t, target: linkPath}
|
||||||
uiw := ui.NewWithColor(io.Discard, false)
|
uiw := ui.NewWithColor(io.Discard, false)
|
||||||
|
|
||||||
repos, err := runSkipErrorScan(t, fs, sourceDir, true, uiw)
|
repos, err := runSkipErrorScan(context.Background(), t, fs, sourceDir, true, uiw)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("expected scan to complete with --skip-errors, got %v", err)
|
t.Fatalf("expected scan to complete with --skip-errors, got %v", err)
|
||||||
}
|
}
|
||||||
@@ -302,3 +328,56 @@ func TestScannerUnreadableSymlinkSkippedWithSkipErrors(t *testing.T) {
|
|||||||
t.Fatalf("expected %s not to be recorded", linkPath)
|
t.Fatalf("expected %s not to be recorded", linkPath)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestScannerCancelStopsSkipErrorsRun checks that a --skip-errors backup
|
||||||
|
// cancelled partway through processing returns the cancellation error,
|
||||||
|
// opens no further file, and reports no file as failed.
|
||||||
|
func TestScannerCancelStopsSkipErrorsRun(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
targetContent string
|
||||||
|
}{
|
||||||
|
// The cancellation lands while the target is being read.
|
||||||
|
{name: "while reading a file", targetContent: "first file content"},
|
||||||
|
// An empty target has no chunks, so the cancellation goes unnoticed
|
||||||
|
// until the run moves on to the next file.
|
||||||
|
{name: "between files", targetContent: ""},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
const target = "/source/a.txt"
|
||||||
|
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
fs := &cancelOnOpenFs{
|
||||||
|
Fs: afero.NewMemMapFs(), target: target, cancel: cancel,
|
||||||
|
}
|
||||||
|
writeSkipErrorTestFile(t, fs, target, tt.targetContent)
|
||||||
|
writeSkipErrorTestFile(t, fs, "/source/b.txt", "second file content")
|
||||||
|
writeSkipErrorTestFile(t, fs, "/source/c.txt", "third file content")
|
||||||
|
|
||||||
|
uiw := ui.NewWithColor(io.Discard, false)
|
||||||
|
|
||||||
|
_, err := runSkipErrorScan(ctx, t, fs, "/source", true, uiw)
|
||||||
|
if !errors.Is(err, context.Canceled) {
|
||||||
|
t.Fatalf("expected the cancellation error, got %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if len(fs.openedAfterCancel) != 0 {
|
||||||
|
t.Fatalf("expected no file opened after the cancellation, got %v",
|
||||||
|
fs.openedAfterCancel)
|
||||||
|
}
|
||||||
|
|
||||||
|
if uiw.ErrorCount() != 0 {
|
||||||
|
t.Fatalf("expected no file reported as failed, got %d error lines",
|
||||||
|
uiw.ErrorCount())
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -892,7 +892,7 @@ func (sm *SnapshotManager) getFileSize(path string) int64 {
|
|||||||
// BackupStats contains statistics from a backup operation
|
// BackupStats contains statistics from a backup operation
|
||||||
type BackupStats struct {
|
type BackupStats struct {
|
||||||
FilesScanned int
|
FilesScanned int
|
||||||
TotalSize int64 // Total size of all files examined
|
TotalSize int64 // Total size of the files in the snapshot
|
||||||
ChunksCreated int
|
ChunksCreated int
|
||||||
BlobsCreated int
|
BlobsCreated int
|
||||||
BytesUploaded int64
|
BytesUploaded int64
|
||||||
|
|||||||
@@ -171,6 +171,11 @@ func (f *Storer) ListStream(
|
|||||||
return f.inner.ListStream(ctx, prefix)
|
return f.inner.ListStream(ctx, prefix)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DeletePartialUploads delegates unchanged.
|
||||||
|
func (f *Storer) DeletePartialUploads(ctx context.Context, prefix string) error {
|
||||||
|
return f.inner.DeletePartialUploads(ctx, prefix)
|
||||||
|
}
|
||||||
|
|
||||||
// Info delegates unchanged.
|
// Info delegates unchanged.
|
||||||
func (f *Storer) Info() storage.Info {
|
func (f *Storer) Info() storage.Info {
|
||||||
return f.inner.Info()
|
return f.inner.Info()
|
||||||
|
|||||||
@@ -242,6 +242,42 @@ func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-chan Objec
|
|||||||
return ch
|
return ch
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DeletePartialUploads removes every file under prefix whose name ends in
|
||||||
|
// tempSuffix. A missing prefix has none to remove.
|
||||||
|
func (f *FileStorer) DeletePartialUploads(ctx context.Context, prefix string) error {
|
||||||
|
basePath := f.fullPath(prefix)
|
||||||
|
|
||||||
|
exists, err := afero.Exists(f.fs, basePath)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("checking path: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if !exists {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
err = afero.Walk(f.fs, basePath, func(path string, info os.FileInfo, err error) error {
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
return ctx.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
if info.IsDir() || !strings.HasSuffix(info.Name(), tempSuffix) {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
return f.fs.Remove(path)
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("walking directory: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// Info returns human-readable storage location information.
|
// Info returns human-readable storage location information.
|
||||||
func (f *FileStorer) Info() Info {
|
func (f *FileStorer) Info() Info {
|
||||||
return Info{
|
return Info{
|
||||||
|
|||||||
@@ -120,3 +120,45 @@ func TestFileStorer_ListSkipsPartialFiles(t *testing.T) {
|
|||||||
t.Fatalf("ListStream should return only the real key, got %v", streamed)
|
t.Fatalf("ListStream should return only the real key, got %v", streamed)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestFileStorer_DeletePartialUploads checks that a leftover temp file is
|
||||||
|
// removed and the object at the real key is kept.
|
||||||
|
func TestFileStorer_DeletePartialUploads(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
base := t.TempDir()
|
||||||
|
|
||||||
|
f, err := storage.NewFileStorer(base)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("NewFileStorer: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
err = f.Put(ctx, testBlobKey, strings.NewReader("blob-bytes"))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Put: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
leftover := filepath.Join(base, testBlobKey+"-123456.partial")
|
||||||
|
|
||||||
|
err = os.WriteFile(leftover, []byte("half"), 0o600)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("writing leftover temp file: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
err = f.DeletePartialUploads(ctx, "blobs/")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("DeletePartialUploads: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err = os.Stat(leftover)
|
||||||
|
if !os.IsNotExist(err) {
|
||||||
|
t.Errorf("leftover temp file was not removed: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err = f.Stat(ctx, testBlobKey)
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("Stat of the real key: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -213,6 +213,31 @@ func (r *RcloneStorer) ListStream(
|
|||||||
return ch
|
return ch
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DeletePartialUploads removes every object under prefix whose name ends
|
||||||
|
// in tempSuffix.
|
||||||
|
func (r *RcloneStorer) DeletePartialUploads(ctx context.Context, prefix string) error {
|
||||||
|
var partial []fs.Object
|
||||||
|
|
||||||
|
err := operations.ListFn(ctx, r.fsys, func(obj fs.Object) {
|
||||||
|
key := obj.Remote()
|
||||||
|
if strings.HasPrefix(key, prefix) && strings.HasSuffix(key, tempSuffix) {
|
||||||
|
partial = append(partial, obj)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("listing objects: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, obj := range partial {
|
||||||
|
err = obj.Remove(ctx)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("removing object: %w", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// Info returns human-readable storage location information.
|
// Info returns human-readable storage location information.
|
||||||
func (r *RcloneStorer) Info() Info {
|
func (r *RcloneStorer) Info() Info {
|
||||||
location := r.remote
|
location := r.remote
|
||||||
|
|||||||
@@ -198,6 +198,47 @@ func TestRcloneStorerListSkipsPartialFiles(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestRcloneStorerDeletePartialUploads checks that a temporary file left
|
||||||
|
// by a killed upload is removed and the object at the real key is kept.
|
||||||
|
//
|
||||||
|
//nolint:paralleltest // NewRcloneStorer installs the process-global rclone config
|
||||||
|
func TestRcloneStorerDeletePartialUploads(t *testing.T) {
|
||||||
|
dir := t.TempDir()
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
s, err := storage.NewRcloneStorer(ctx, ":local", dir)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("NewRcloneStorer: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
err = s.Put(ctx, testBlobKey, strings.NewReader("blob-bytes"))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Put: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
leftover := filepath.Join(dir, testBlobKey+"-123456.partial")
|
||||||
|
|
||||||
|
err = os.WriteFile(leftover, []byte("half"), 0o600)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("writing leftover temp file: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
err = s.DeletePartialUploads(ctx, "blobs/")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("DeletePartialUploads: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err = os.Stat(leftover)
|
||||||
|
if !os.IsNotExist(err) {
|
||||||
|
t.Errorf("leftover temp file was not removed: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err = s.Stat(ctx, testBlobKey)
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("Stat of the real key: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// newRcloneStorerOnWrappedLocal registers name as rclone's local backend
|
// newRcloneStorerOnWrappedLocal registers name as rclone's local backend
|
||||||
// wrapped by wrap, and builds an rclone backend on it rooted at a fresh
|
// wrapped by wrap, and builds an rclone backend on it rooted at a fresh
|
||||||
// temp directory. wrap changes the features the local backend reports, so
|
// temp directory. wrap changes the features the local backend reports, so
|
||||||
|
|||||||
@@ -99,6 +99,12 @@ func (s *S3Storer) ListStream(ctx context.Context, prefix string) <-chan ObjectI
|
|||||||
return ch
|
return ch
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DeletePartialUploads has nothing to remove: S3 shows an object only once
|
||||||
|
// its upload has completed, so an upload cut off part-way leaves none.
|
||||||
|
func (s *S3Storer) DeletePartialUploads(_ context.Context, _ string) error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// Info returns human-readable storage location information.
|
// Info returns human-readable storage location information.
|
||||||
func (s *S3Storer) Info() Info {
|
func (s *S3Storer) Info() Info {
|
||||||
return Info{
|
return Info{
|
||||||
|
|||||||
@@ -71,6 +71,12 @@ type Storer interface {
|
|||||||
// If an error occurs during listing, the final item will have Err set.
|
// If an error occurs during listing, the final item will have Err set.
|
||||||
ListStream(ctx context.Context, prefix string) <-chan ObjectInfo
|
ListStream(ctx context.Context, prefix string) <-chan ObjectInfo
|
||||||
|
|
||||||
|
// DeletePartialUploads removes every object under prefix that an
|
||||||
|
// upload cut off part-way left under a temporary name ending in
|
||||||
|
// `.partial`. The file and rclone backends' List and ListStream skip
|
||||||
|
// such an object.
|
||||||
|
DeletePartialUploads(ctx context.Context, prefix string) error
|
||||||
|
|
||||||
// Info returns human-readable storage location information.
|
// Info returns human-readable storage location information.
|
||||||
Info() Info
|
Info() Info
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -257,6 +257,10 @@ func (s *stubLister) List(_ context.Context, _ string) ([]string, error) {
|
|||||||
return nil, errStubUnused
|
return nil, errStubUnused
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (s *stubLister) DeletePartialUploads(_ context.Context, _ string) error {
|
||||||
|
return errStubUnused
|
||||||
|
}
|
||||||
|
|
||||||
func (s *stubLister) Info() storage.Info {
|
func (s *stubLister) Info() storage.Info {
|
||||||
return storage.Info{}
|
return storage.Info{}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -181,8 +181,11 @@ type SnapshotMetadataInfo struct {
|
|||||||
ManifestSize int64 `json:"manifest_size"`
|
ManifestSize int64 `json:"manifest_size"`
|
||||||
DatabaseSize int64 `json:"database_size"`
|
DatabaseSize int64 `json:"database_size"`
|
||||||
TotalSize int64 `json:"total_size"`
|
TotalSize int64 `json:"total_size"`
|
||||||
BlobCount int `json:"blob_count"`
|
|
||||||
BlobsSize int64 `json:"blobs_size"`
|
// Both stay nil (null in the JSON) when the snapshot's manifest was
|
||||||
|
// listed but could not be read.
|
||||||
|
BlobCount *int `json:"blob_count"`
|
||||||
|
BlobsSize *int64 `json:"blobs_size"`
|
||||||
|
|
||||||
// Set when the listing holds this snapshot's manifest.json.zst. A
|
// Set when the listing holds this snapshot's manifest.json.zst. A
|
||||||
// backup interrupted before its manifest upload leaves a directory
|
// backup interrupted before its manifest upload leaves a directory
|
||||||
@@ -380,6 +383,10 @@ func (v *Vaultik) collectReferencedBlobsFromManifests(
|
|||||||
for _, snapshotID := range snapshotIDs {
|
for _, snapshotID := range snapshotIDs {
|
||||||
info := snapshotMetadata[snapshotID]
|
info := snapshotMetadata[snapshotID]
|
||||||
if !info.hasManifest {
|
if !info.hasManifest {
|
||||||
|
// The orphan figures count this directory's blobs as
|
||||||
|
// orphaned, so it references none.
|
||||||
|
info.BlobCount, info.BlobsSize = new(int), new(int64)
|
||||||
|
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -395,7 +402,7 @@ func (v *Vaultik) collectReferencedBlobsFromManifests(
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
info.BlobCount = manifest.BlobCount
|
blobCount := manifest.BlobCount
|
||||||
|
|
||||||
var blobsSize int64
|
var blobsSize int64
|
||||||
|
|
||||||
@@ -404,7 +411,8 @@ func (v *Vaultik) collectReferencedBlobsFromManifests(
|
|||||||
blobsSize += blob.CompressedSize
|
blobsSize += blob.CompressedSize
|
||||||
}
|
}
|
||||||
|
|
||||||
info.BlobsSize = blobsSize
|
info.BlobCount = &blobCount
|
||||||
|
info.BlobsSize = &blobsSize
|
||||||
}
|
}
|
||||||
|
|
||||||
return referencedBlobs, unreadable
|
return referencedBlobs, unreadable
|
||||||
@@ -516,13 +524,23 @@ func (v *Vaultik) printRemoteInfoTable(result *RemoteInfoResult) {
|
|||||||
v.stdoutf("%s", separator)
|
v.stdoutf("%s", separator)
|
||||||
|
|
||||||
for _, info := range result.Snapshots {
|
for _, info := range result.Snapshots {
|
||||||
|
blobCount := unknownText
|
||||||
|
if info.BlobCount != nil {
|
||||||
|
blobCount = humanize.Comma(int64(*info.BlobCount))
|
||||||
|
}
|
||||||
|
|
||||||
|
blobsSize := unknownText
|
||||||
|
if info.BlobsSize != nil {
|
||||||
|
blobsSize = ubytes(*info.BlobsSize)
|
||||||
|
}
|
||||||
|
|
||||||
v.stdoutf(rowFormat,
|
v.stdoutf(rowFormat,
|
||||||
truncateString(info.SnapshotID, snapshotIDColWidth),
|
truncateString(info.SnapshotID, snapshotIDColWidth),
|
||||||
ubytes(info.ManifestSize),
|
ubytes(info.ManifestSize),
|
||||||
ubytes(info.DatabaseSize),
|
ubytes(info.DatabaseSize),
|
||||||
ubytes(info.TotalSize),
|
ubytes(info.TotalSize),
|
||||||
humanize.Comma(int64(info.BlobCount)),
|
blobCount,
|
||||||
ubytes(info.BlobsSize),
|
blobsSize,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -155,6 +155,12 @@ func (m *MockStorer) ListStream(
|
|||||||
return ch
|
return ch
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DeletePartialUploads has nothing to remove: Put stores each object
|
||||||
|
// under its key at once.
|
||||||
|
func (m *MockStorer) DeletePartialUploads(_ context.Context, _ string) error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
func (m *MockStorer) Info() storage.Info {
|
func (m *MockStorer) Info() storage.Info {
|
||||||
return storage.Info{
|
return storage.Info{
|
||||||
Type: "mock",
|
Type: "mock",
|
||||||
|
|||||||
@@ -0,0 +1,64 @@
|
|||||||
|
package vaultik_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"io/fs"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
|
"sneak.berlin/go/vaultik/internal/snapshot"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestNukeRemoteLeavesNoFiles checks that remote nuke leaves no file
|
||||||
|
// under a file:// destination, including the `.partial` files uploads
|
||||||
|
// killed part-way leave next to a snapshot's metadata and next to where a
|
||||||
|
// blob would have been.
|
||||||
|
func TestNukeRemoteLeavesNoFiles(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
storeDir := filepath.Join(t.TempDir(), "store")
|
||||||
|
v, repos, _ := backUpToFileDestination(ctx, t, storeDir)
|
||||||
|
|
||||||
|
snapshots, err := repos.Snapshots.ListRecent(ctx, listRecentTestLimit)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.Len(t, snapshots, 1)
|
||||||
|
|
||||||
|
snapshotKey := snapshot.RemoteSnapshotKey(snapshots[0].ID.String())
|
||||||
|
hash := testBlobHashA
|
||||||
|
leftovers := []string{
|
||||||
|
filepath.Join(storeDir, "metadata", snapshotKey,
|
||||||
|
"db.zst.age-123456.partial"),
|
||||||
|
filepath.Join(storeDir, "blobs", hash[:2], hash[2:4],
|
||||||
|
hash+"-123456.partial"),
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, leftover := range leftovers {
|
||||||
|
require.NoError(t, os.MkdirAll(filepath.Dir(leftover), 0o750))
|
||||||
|
require.NoError(t, os.WriteFile(leftover, []byte("half an upload"), 0o600))
|
||||||
|
}
|
||||||
|
|
||||||
|
require.NoError(t, v.NukeRemote(true))
|
||||||
|
|
||||||
|
var files []string
|
||||||
|
|
||||||
|
err = filepath.WalkDir(storeDir,
|
||||||
|
func(path string, entry fs.DirEntry, err error) error {
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if !entry.IsDir() {
|
||||||
|
files = append(files, path)
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Empty(t, files)
|
||||||
|
}
|
||||||
@@ -24,8 +24,9 @@ var errNukeRequiresForce = errors.New(
|
|||||||
const metadataDirName = "metadata"
|
const metadataDirName = "metadata"
|
||||||
|
|
||||||
// NukeRemote deletes every snapshot's metadata and every blob from remote
|
// NukeRemote deletes every snapshot's metadata and every blob from remote
|
||||||
// storage. After this returns successfully the bucket prefix is empty and
|
// storage, along with any object an upload cut off part-way left under a
|
||||||
// the next backup starts from scratch.
|
// temporary `.partial` name. After this returns successfully the bucket
|
||||||
|
// prefix is empty and the next backup starts from scratch.
|
||||||
//
|
//
|
||||||
// Refuses to run unless force is true. The caller is responsible for
|
// Refuses to run unless force is true. The caller is responsible for
|
||||||
// confirming with the user.
|
// confirming with the user.
|
||||||
@@ -48,6 +49,15 @@ func (v *Vaultik) NukeRemote(force bool) error {
|
|||||||
return fmt.Errorf("pruning blobs: %w", err)
|
return fmt.Errorf("pruning blobs: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// The file and rclone listings skip `.partial` objects, so the two
|
||||||
|
// steps above never delete them.
|
||||||
|
for _, prefix := range []string{"metadata/", "blobs/"} {
|
||||||
|
err = v.Storage.DeletePartialUploads(v.ctx, prefix)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("deleting partial uploads: %w", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
v.UI.Completef("Backup destination store is now empty.")
|
v.UI.Completef("Backup destination store is now empty.")
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
@@ -42,7 +42,7 @@ func TestTableCountForReportSurfacesReadFailure(t *testing.T) {
|
|||||||
assert.Nil(t, missing, "a failed read is unknown, not a count")
|
assert.Nil(t, missing, "a failed read is unknown, not a count")
|
||||||
|
|
||||||
// The rendered count for a failed read must say unknown, never 0.
|
// The rendered count for a failed read must say unknown, never 0.
|
||||||
assert.Equal(t, countUnknown, countText(missing))
|
assert.Equal(t, unknownText, countText(missing))
|
||||||
assert.NotEqual(t, "0", countText(missing))
|
assert.NotEqual(t, "0", countText(missing))
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -57,7 +57,7 @@ func TestCountTextDistinguishesEmptyFromUnknown(t *testing.T) {
|
|||||||
|
|
||||||
assert.Equal(t, "0", countText(&zero))
|
assert.Equal(t, "0", countText(&zero))
|
||||||
assert.Equal(t, "7", countText(&seven))
|
assert.Equal(t, "7", countText(&seven))
|
||||||
assert.Equal(t, countUnknown, countText(nil))
|
assert.Equal(t, unknownText, countText(nil))
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestCountDiffUnknownWhenEitherSideUnknown checks that a delta computed
|
// TestCountDiffUnknownWhenEitherSideUnknown checks that a delta computed
|
||||||
|
|||||||
@@ -0,0 +1,40 @@
|
|||||||
|
package vaultik_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
|
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestPruneBlobs_JSONDeletesWithoutAsking checks that prune with --json
|
||||||
|
// and without --force deletes an unreferenced blob without the
|
||||||
|
// confirmation prompt. Stdin is empty, so a prompt would read no answer
|
||||||
|
// and cancel, and its text would come before the JSON document.
|
||||||
|
func TestPruneBlobs_JSONDeletesWithoutAsking(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
env := newListEnv(t)
|
||||||
|
|
||||||
|
// The store holds no manifest, so nothing references this blob.
|
||||||
|
addBlob(t, env.store.testStorer, testBlobHashA)
|
||||||
|
|
||||||
|
err := env.v.PruneBlobs(&vaultik.PruneOptions{JSON: true})
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
blobKey := "blobs/" + testBlobHashA[:2] + "/" + testBlobHashA[2:4] +
|
||||||
|
"/" + testBlobHashA
|
||||||
|
assert.False(t, env.store.hasKey(blobKey),
|
||||||
|
"the unreferenced blob must be deleted")
|
||||||
|
|
||||||
|
var result vaultik.PruneBlobsResult
|
||||||
|
|
||||||
|
require.NoError(t, json.Unmarshal(env.stdout.Bytes(), &result),
|
||||||
|
"stdout must hold only the JSON document, got:\n%s",
|
||||||
|
env.stdout.String())
|
||||||
|
assert.Equal(t, 1, result.BlobsDeleted)
|
||||||
|
}
|
||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"bytes"
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -69,6 +70,74 @@ func TestRemoteInfo_UnreadableManifestLeavesOrphansUnknown(t *testing.T) {
|
|||||||
assert.Equal(t, []any{unreadableKey}, doc["unreadable_manifests"])
|
assert.Equal(t, []any{unreadableKey}, doc["unreadable_manifests"])
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestRemoteInfo_UnreadableManifestLeavesSnapshotBlobsUnknown checks
|
||||||
|
// that the row of a snapshot whose manifest cannot be read gives its
|
||||||
|
// blob count and blob size as unknown in the table and as null in
|
||||||
|
// --json, not as 0. A directory without a manifest still shows 0: the
|
||||||
|
// orphan figures count its blobs as orphaned, so it references none.
|
||||||
|
func TestRemoteInfo_UnreadableManifestLeavesSnapshotBlobsUnknown(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
env := newListEnv(t)
|
||||||
|
readableKey := env.addRemote(t, listRemoteID,
|
||||||
|
time.Date(2026, 3, 2, 0, 0, 0, 0, time.UTC))
|
||||||
|
|
||||||
|
unreadableKey := snapshot.RemoteSnapshotKey(listLocalID)
|
||||||
|
require.NoError(t, env.store.Put(context.Background(),
|
||||||
|
"metadata/"+unreadableKey+"/manifest.json.zst",
|
||||||
|
bytes.NewReader([]byte("not a valid manifest"))))
|
||||||
|
|
||||||
|
noManifestKey := snapshot.RemoteSnapshotKey("testhost_home_2026-03-03T10:00:00Z")
|
||||||
|
require.NoError(t, env.store.Put(context.Background(),
|
||||||
|
"metadata/"+noManifestKey+"/db.zst.age",
|
||||||
|
bytes.NewReader([]byte("not a valid database"))))
|
||||||
|
|
||||||
|
require.NoError(t, env.v.RemoteInfo(false))
|
||||||
|
|
||||||
|
// The table truncates the remote key, so a row is found by a prefix.
|
||||||
|
wantUnknown := map[string]int{readableKey: 0, unreadableKey: 2, noManifestKey: 0}
|
||||||
|
for key, want := range wantUnknown {
|
||||||
|
var row string
|
||||||
|
|
||||||
|
for line := range strings.SplitSeq(env.stdout.String(), "\n") {
|
||||||
|
if strings.HasPrefix(line, key[:16]) {
|
||||||
|
row = line
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
require.NotEmpty(t, row, "no table row for %s", key)
|
||||||
|
assert.Equal(t, want, strings.Count(row, "unknown"), "row: %q", row)
|
||||||
|
}
|
||||||
|
|
||||||
|
env.stdout.Reset()
|
||||||
|
require.NoError(t, env.v.RemoteInfo(true))
|
||||||
|
|
||||||
|
var doc struct {
|
||||||
|
Snapshots []map[string]any `json:"snapshots"`
|
||||||
|
}
|
||||||
|
|
||||||
|
require.NoError(t, json.Unmarshal(env.stdout.Bytes(), &doc))
|
||||||
|
require.Len(t, doc.Snapshots, len(wantUnknown))
|
||||||
|
|
||||||
|
for _, entry := range doc.Snapshots {
|
||||||
|
switch entry["snapshot_id"] {
|
||||||
|
case readableKey:
|
||||||
|
assert.InDelta(t, 1, entry["blob_count"], 0)
|
||||||
|
assert.InDelta(t, fiveMegabytes, entry["blobs_size"], 0)
|
||||||
|
case noManifestKey:
|
||||||
|
assert.InDelta(t, 0, entry["blob_count"], 0)
|
||||||
|
assert.InDelta(t, 0, entry["blobs_size"], 0)
|
||||||
|
default:
|
||||||
|
assert.Equal(t, unreadableKey, entry["snapshot_id"])
|
||||||
|
assert.Contains(t, entry, "blob_count")
|
||||||
|
assert.Nil(t, entry["blob_count"])
|
||||||
|
assert.Contains(t, entry, "blobs_size")
|
||||||
|
assert.Nil(t, entry["blobs_size"])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// TestRemoteInfo_SkipsNonConformingMetadataName checks that a directory
|
// TestRemoteInfo_SkipsNonConformingMetadataName checks that a directory
|
||||||
// under metadata/ whose name is not a remote key is left out of the
|
// under metadata/ whose name is not a remote key is left out of the
|
||||||
// report, and that the orphan figures are unknown when it holds a
|
// report, and that the orphan figures are unknown when it holds a
|
||||||
|
|||||||
@@ -125,6 +125,12 @@ func (s *testStorer) ListStream(
|
|||||||
return ch
|
return ch
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DeletePartialUploads has nothing to remove: Put stores each object
|
||||||
|
// under its key at once.
|
||||||
|
func (s *testStorer) DeletePartialUploads(_ context.Context, _ string) error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
func (s *testStorer) Info() storage.Info {
|
func (s *testStorer) Info() storage.Info {
|
||||||
return storage.Info{
|
return storage.Info{
|
||||||
Type: testLabel,
|
Type: testLabel,
|
||||||
|
|||||||
@@ -185,6 +185,7 @@ type snapshotStats struct {
|
|||||||
totalBlobs int
|
totalBlobs int
|
||||||
totalBytesSkipped int64
|
totalBytesSkipped int64
|
||||||
totalFilesSkipped int
|
totalFilesSkipped int
|
||||||
|
totalFilesFailed int
|
||||||
totalFilesDeleted int
|
totalFilesDeleted int
|
||||||
totalBytesDeleted int64
|
totalBytesDeleted int64
|
||||||
totalBytesUploaded int64
|
totalBytesUploaded int64
|
||||||
@@ -315,6 +316,7 @@ func (v *Vaultik) scanAllDirectories(
|
|||||||
stats.totalChunks += result.ChunksCreated
|
stats.totalChunks += result.ChunksCreated
|
||||||
stats.totalBlobs += result.BlobsCreated
|
stats.totalBlobs += result.BlobsCreated
|
||||||
stats.totalFilesSkipped += result.FilesSkipped
|
stats.totalFilesSkipped += result.FilesSkipped
|
||||||
|
stats.totalFilesFailed += result.FilesFailed
|
||||||
stats.totalBytesSkipped += result.BytesSkipped
|
stats.totalBytesSkipped += result.BytesSkipped
|
||||||
stats.totalFilesDeleted += result.FilesDeleted
|
stats.totalFilesDeleted += result.FilesDeleted
|
||||||
stats.totalBytesDeleted += result.BytesDeleted
|
stats.totalBytesDeleted += result.BytesDeleted
|
||||||
@@ -326,6 +328,7 @@ func (v *Vaultik) scanAllDirectories(
|
|||||||
"path", dir,
|
"path", dir,
|
||||||
"files", result.FilesScanned,
|
"files", result.FilesScanned,
|
||||||
"files_skipped", result.FilesSkipped,
|
"files_skipped", result.FilesSkipped,
|
||||||
|
"files_failed", result.FilesFailed,
|
||||||
"bytes", result.BytesScanned,
|
"bytes", result.BytesScanned,
|
||||||
"bytes_skipped", result.BytesSkipped,
|
"bytes_skipped", result.BytesSkipped,
|
||||||
"chunks", result.ChunksCreated,
|
"chunks", result.ChunksCreated,
|
||||||
@@ -359,9 +362,11 @@ func (v *Vaultik) finalizeSnapshotMetadata(
|
|||||||
return fmt.Errorf("getting snapshot blob sizes: %w", err)
|
return fmt.Errorf("getting snapshot blob sizes: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// file_count and total_size leave out the files that could not be
|
||||||
|
// stored; stats.totalBytes already does.
|
||||||
extStats := snapshot.ExtendedBackupStats{
|
extStats := snapshot.ExtendedBackupStats{
|
||||||
BackupStats: snapshot.BackupStats{
|
BackupStats: snapshot.BackupStats{
|
||||||
FilesScanned: stats.totalFiles,
|
FilesScanned: stats.totalFiles - stats.totalFilesFailed,
|
||||||
TotalSize: stats.totalBytes + stats.totalBytesSkipped,
|
TotalSize: stats.totalBytes + stats.totalBytesSkipped,
|
||||||
ChunksCreated: stats.totalChunks,
|
ChunksCreated: stats.totalChunks,
|
||||||
BlobsCreated: stats.totalBlobs,
|
BlobsCreated: stats.totalBlobs,
|
||||||
@@ -409,7 +414,8 @@ func (v *Vaultik) printSnapshotSummary(
|
|||||||
snapshotID string, startTime time.Time, stats *snapshotStats,
|
snapshotID string, startTime time.Time, stats *snapshotStats,
|
||||||
) {
|
) {
|
||||||
snapshotDuration := time.Since(startTime)
|
snapshotDuration := time.Since(startTime)
|
||||||
totalFilesChanged := stats.totalFiles - stats.totalFilesSkipped
|
totalFilesChanged := stats.totalFiles - stats.totalFilesSkipped -
|
||||||
|
stats.totalFilesFailed
|
||||||
totalBytesAll := stats.totalBytes + stats.totalBytesSkipped
|
totalBytesAll := stats.totalBytes + stats.totalBytesSkipped
|
||||||
|
|
||||||
var compressionRatio float64
|
var compressionRatio float64
|
||||||
@@ -426,6 +432,10 @@ func (v *Vaultik) printSnapshotSummary(
|
|||||||
v.UI.Count(stats.totalFiles),
|
v.UI.Count(stats.totalFiles),
|
||||||
v.UI.Count(totalFilesChanged),
|
v.UI.Count(totalFilesChanged),
|
||||||
v.UI.Count(stats.totalFilesSkipped))
|
v.UI.Count(stats.totalFilesSkipped))
|
||||||
|
if stats.totalFilesFailed > 0 {
|
||||||
|
filesMsg += fmt.Sprintf(", %s failed", v.UI.Count(stats.totalFilesFailed))
|
||||||
|
}
|
||||||
|
|
||||||
if stats.totalFilesDeleted > 0 {
|
if stats.totalFilesDeleted > 0 {
|
||||||
filesMsg += fmt.Sprintf(", %s deleted", v.UI.Count(stats.totalFilesDeleted))
|
filesMsg += fmt.Sprintf(", %s deleted", v.UI.Count(stats.totalFilesDeleted))
|
||||||
}
|
}
|
||||||
@@ -1756,9 +1766,9 @@ func (v *Vaultik) PruneDatabase() (*PruneResult, error) {
|
|||||||
return result, nil
|
return result, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// countUnknown is what a count reads as when its query could not be run,
|
// unknownText is what a count or size reads as when it could not be
|
||||||
// distinct from "0", which means the table really was empty.
|
// determined, distinct from "0", which is a real zero.
|
||||||
const countUnknown = "unknown"
|
const unknownText = "unknown"
|
||||||
|
|
||||||
// tableCountForReport returns the row count of a table for the prune
|
// tableCountForReport returns the row count of a table for the prune
|
||||||
// summary, or nil if the count could not be read. A read failure is
|
// summary, or nil if the count could not be read. A read failure is
|
||||||
@@ -1795,7 +1805,7 @@ func countDiff(before, after *int64) *int64 {
|
|||||||
// one that could not be queried.
|
// one that could not be queried.
|
||||||
func countText(count *int64) string {
|
func countText(count *int64) string {
|
||||||
if count == nil {
|
if count == nil {
|
||||||
return countUnknown
|
return unknownText
|
||||||
}
|
}
|
||||||
|
|
||||||
return strconv.FormatInt(*count, 10)
|
return strconv.FormatInt(*count, 10)
|
||||||
|
|||||||
@@ -0,0 +1,123 @@
|
|||||||
|
package vaultik_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/spf13/afero"
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/config"
|
||||||
|
"sneak.berlin/go/vaultik/internal/database"
|
||||||
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
|
"sneak.berlin/go/vaultik/internal/storage"
|
||||||
|
"sneak.berlin/go/vaultik/internal/ui"
|
||||||
|
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||||
|
)
|
||||||
|
|
||||||
|
// openFailFs is the real filesystem, except that opening path fails
|
||||||
|
// with err. Phase 1 of a backup only lstats a file, so it still counts
|
||||||
|
// path; phase 2 is the first to open it.
|
||||||
|
type openFailFs struct {
|
||||||
|
afero.OsFs
|
||||||
|
|
||||||
|
path string
|
||||||
|
err error
|
||||||
|
}
|
||||||
|
|
||||||
|
//nolint:ireturn // afero.Fs.Open is defined to return the interface.
|
||||||
|
func (f *openFailFs) Open(name string) (afero.File, error) {
|
||||||
|
if name == f.path {
|
||||||
|
return nil, &os.PathError{Op: "open", Path: name, Err: f.err}
|
||||||
|
}
|
||||||
|
|
||||||
|
return f.OsFs.Open(name)
|
||||||
|
}
|
||||||
|
|
||||||
|
// A file that phase 2 cannot open is reported as failed, not as
|
||||||
|
// unchanged, and neither the summary's data total nor the snapshots row
|
||||||
|
// counts it. See https://git.eeqj.de/sneak/vaultik/issues/280.
|
||||||
|
func TestSnapshotSummaryCountsFileNotStoredAsFailed(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
openErr error
|
||||||
|
skipErrors bool
|
||||||
|
}{
|
||||||
|
// What a normal user gets opening a file with mode 000.
|
||||||
|
{name: "unopenable under skip-errors",
|
||||||
|
openErr: os.ErrPermission, skipErrors: true},
|
||||||
|
// What opening a file removed after phase 1 gives.
|
||||||
|
{name: "removed between the phases",
|
||||||
|
openErr: os.ErrNotExist, skipErrors: false},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
// The scan walks the source path with symlinks resolved, so
|
||||||
|
// failedPath must be spelled the same way to match.
|
||||||
|
tempDir, err := filepath.EvalSymlinks(t.TempDir())
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
srcDir := filepath.Join(tempDir, "src")
|
||||||
|
failedPath := filepath.Join(srcDir, "failed.txt")
|
||||||
|
storedContent := []byte("this file is backed up")
|
||||||
|
storedSize := int64(len(storedContent))
|
||||||
|
|
||||||
|
fs := &openFailFs{path: failedPath, err: tt.openErr}
|
||||||
|
require.NoError(t, fs.MkdirAll(srcDir, 0o755))
|
||||||
|
require.NoError(t, afero.WriteFile(fs,
|
||||||
|
filepath.Join(srcDir, "stored.txt"), storedContent, 0o644))
|
||||||
|
require.NoError(t, afero.WriteFile(fs,
|
||||||
|
failedPath, []byte("this file cannot be opened"), 0o644))
|
||||||
|
|
||||||
|
cfg := faultTestConfig()
|
||||||
|
cfg.IndexPath = filepath.Join(tempDir, "index.sqlite")
|
||||||
|
cfg.Snapshots = map[string]config.SnapshotConfig{
|
||||||
|
"src": {Paths: []string{srcDir}},
|
||||||
|
}
|
||||||
|
|
||||||
|
store, err := storage.NewFileStorer(filepath.Join(tempDir, "remote"))
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
db, err := database.New(ctx, cfg.IndexPath)
|
||||||
|
require.NoError(t, err)
|
||||||
|
t.Cleanup(func() { _ = db.Close() })
|
||||||
|
|
||||||
|
repos := database.NewRepositories(db)
|
||||||
|
out := &bytes.Buffer{}
|
||||||
|
v := newBackupVaultik(ctx, cfg, store, repos, db, fs)
|
||||||
|
v.UI = ui.NewWithColor(out, false)
|
||||||
|
|
||||||
|
require.NoError(t, v.CreateSnapshot(&vaultik.SnapshotCreateOptions{
|
||||||
|
SkipErrors: tt.skipErrors,
|
||||||
|
Snapshots: []string{"src"},
|
||||||
|
}))
|
||||||
|
|
||||||
|
summary := out.String()
|
||||||
|
assert.Contains(t, summary,
|
||||||
|
"Files: 2 examined, 1 backed up, 0 unchanged, 1 failed.")
|
||||||
|
assert.Contains(t, summary,
|
||||||
|
fmt.Sprintf("Data: %s total (%s backed up).",
|
||||||
|
v.UI.Size(storedSize), v.UI.Size(storedSize)))
|
||||||
|
|
||||||
|
snap, err := repos.Snapshots.GetByID(ctx,
|
||||||
|
localSnapshotID(ctx, t, repos, "src"))
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.NotNil(t, snap)
|
||||||
|
|
||||||
|
assert.Equal(t, int64(1), snap.FileCount)
|
||||||
|
assert.Equal(t, storedSize, snap.TotalSize)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,75 @@
|
|||||||
|
package vaultik_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/spf13/afero"
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/config"
|
||||||
|
"sneak.berlin/go/vaultik/internal/database"
|
||||||
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
|
"sneak.berlin/go/vaultik/internal/storage"
|
||||||
|
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||||
|
)
|
||||||
|
|
||||||
|
// --skip-errors skips only a file that cannot be opened or read. A local
|
||||||
|
// index error while a directory is recorded stops the backup, and the
|
||||||
|
// snapshot is not recorded as complete. See
|
||||||
|
// https://git.eeqj.de/sneak/vaultik/issues/284.
|
||||||
|
func TestSkipErrorsBackupStopsOnIndexErrorRecordingDirectory(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
// The scan walks the source path with symlinks resolved, so dirPath
|
||||||
|
// must be spelled the same way to match the row the scan inserts.
|
||||||
|
tempDir, err := filepath.EvalSymlinks(t.TempDir())
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
srcDir := filepath.Join(tempDir, "src")
|
||||||
|
dirPath := filepath.Join(srcDir, "dir")
|
||||||
|
|
||||||
|
fs := afero.NewOsFs()
|
||||||
|
require.NoError(t, fs.MkdirAll(dirPath, 0o755))
|
||||||
|
require.NoError(t, afero.WriteFile(fs,
|
||||||
|
filepath.Join(dirPath, "file.txt"), []byte("file content"), 0o644))
|
||||||
|
|
||||||
|
cfg := faultTestConfig()
|
||||||
|
cfg.IndexPath = filepath.Join(tempDir, "index.sqlite")
|
||||||
|
cfg.Snapshots = map[string]config.SnapshotConfig{
|
||||||
|
"tree": {Paths: []string{srcDir}},
|
||||||
|
}
|
||||||
|
|
||||||
|
store, err := storage.NewFileStorer(filepath.Join(tempDir, "remote"))
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
db, err := database.New(ctx, cfg.IndexPath)
|
||||||
|
require.NoError(t, err)
|
||||||
|
t.Cleanup(func() { _ = db.Close() })
|
||||||
|
|
||||||
|
// The local index refuses the directory's files row.
|
||||||
|
_, err = db.Conn().ExecContext(ctx, fmt.Sprintf(`
|
||||||
|
CREATE TRIGGER refuse_directory BEFORE INSERT ON files
|
||||||
|
WHEN NEW.path = '%s'
|
||||||
|
BEGIN SELECT RAISE(ABORT, 'simulated index error'); END`, dirPath))
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
repos := database.NewRepositories(db)
|
||||||
|
v := newBackupVaultik(ctx, cfg, store, repos, db, fs)
|
||||||
|
|
||||||
|
err = v.CreateSnapshot(&vaultik.SnapshotCreateOptions{
|
||||||
|
SkipErrors: true,
|
||||||
|
Snapshots: []string{"tree"},
|
||||||
|
})
|
||||||
|
require.ErrorContains(t, err, "simulated index error")
|
||||||
|
|
||||||
|
snapshots, err := repos.Snapshots.ListRecent(ctx, listRecentTestLimit)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.Len(t, snapshots, 1)
|
||||||
|
assert.Nil(t, snapshots[0].CompletedAt)
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user