From 731126e2893f81192f1b001158e513dc76702408 Mon Sep 17 00:00:00 2001 From: sneak Date: Thu, 8 Oct 2026 01:48:40 +0000 Subject: [PATCH] Keep the progress line of a multi-path snapshot within 100% (closes #271) One progress reporter spans every path of a snapshot. Its counts of files and bytes processed run across all the paths, while the totals they were divided by were reset to each path's own, so a second path smaller than the first showed more than 100% and an ETA of `unknown`. The totals now add up over the paths scanned so far. The rate behind 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. SetTotalSize is renamed AddTotalSize because it now adds. The issue says the ETA went negative; it was computed negative and printed as `unknown`. Model: opus-5-5 --- TODO.md | 11 +++ internal/snapshot/progress.go | 122 ++++++++++++++---------- internal/snapshot/progress_rate_test.go | 54 +++++++++++ internal/snapshot/progress_test.go | 94 ++++++++++++++++++ internal/snapshot/scanner.go | 4 +- 5 files changed, 230 insertions(+), 55 deletions(-) create mode 100644 internal/snapshot/progress_rate_test.go create mode 100644 internal/snapshot/progress_test.go diff --git a/TODO.md b/TODO.md index 906798c..4e052b4 100644 --- a/TODO.md +++ b/TODO.md @@ -22,6 +22,17 @@ the tag exists and is exercised; what is left is merging `next` to # Completed Steps +- 2026-10-08: Kept the progress line of a snapshot with more than one + path within 100% + ([issue #271](https://git.eeqj.de/sneak/vaultik/issues/271)). The + bytes and files processed counted every path, but the totals they + 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`. + The totals now add up over the paths scanned so far. The rate behind + 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 it starts in the same second as the first ([issue #270](https://git.eeqj.de/sneak/vaultik/issues/270)). The diff --git a/internal/snapshot/progress.go b/internal/snapshot/progress.go index 9aba0f5..d58a339 100644 --- a/internal/snapshot/progress.go +++ b/internal/snapshot/progress.go @@ -56,23 +56,27 @@ const ( // ProgressStats holds atomic counters for progress tracking type ProgressStats struct { - FilesScanned atomic.Int64 // Total files seen during scan (includes skipped) - FilesProcessed atomic.Int64 // Files actually processed in phase 2 - FilesSkipped atomic.Int64 // Files skipped due to no changes - BytesScanned atomic.Int64 // Bytes from new/changed files only - BytesSkipped atomic.Int64 // Bytes from unchanged files - BytesProcessed atomic.Int64 // Actual bytes processed (for ETA calculation) - ChunksCreated atomic.Int64 - BlobsCreated atomic.Int64 - BlobsUploaded atomic.Int64 - BytesUploaded atomic.Int64 - CurrentFile atomic.Value // stores string - TotalSize atomic.Int64 // Total size to process (set after scan phase) - TotalFiles atomic.Int64 // Total files to process in phase 2 - ProcessStartTime atomic.Value // stores time.Time when processing starts - StartTime time.Time - mu sync.RWMutex - lastDetailTime time.Time + FilesScanned atomic.Int64 // Total files seen during scan (includes skipped) + FilesProcessed atomic.Int64 // Files actually processed in phase 2 + FilesSkipped atomic.Int64 // Files skipped due to no changes + BytesScanned atomic.Int64 // Bytes from new/changed files only + BytesSkipped atomic.Int64 // Bytes from unchanged files + BytesProcessed atomic.Int64 // Actual bytes processed (for ETA calculation) + ChunksCreated atomic.Int64 + BlobsCreated atomic.Int64 + BlobsUploaded atomic.Int64 + BytesUploaded atomic.Int64 + CurrentFile atomic.Value // stores string + TotalSize atomic.Int64 // Size to process in the paths scanned so far + TotalFiles atomic.Int64 // Files to process in the paths scanned so far + StartTime time.Time + mu sync.RWMutex + lastDetailTime time.Time + + // Guarded by mu: when the last scan phase ended, and + // BytesProcessed at that moment. + processStartTime time.Time + processStartBytes int64 // Upload tracking CurrentUpload atomic.Value // stores *UploadInfo @@ -148,10 +152,36 @@ func (pr *ProgressReporter) GetStats() *ProgressStats { return pr.stats } -// SetTotalSize sets the total size to process (after scan phase) -func (pr *ProgressReporter) SetTotalSize(size int64) { - pr.stats.TotalSize.Store(size) - pr.stats.ProcessStartTime.Store(time.Now().UTC()) +// AddTotalSize adds the size one path of the snapshot has to process to +// the total, once that path's scan phase is done, and starts measuring +// the processing rate again. The processed counts run across every path, +// 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) { + pr.stats.TotalSize.Add(size) + + pr.stats.mu.Lock() + defer pr.stats.mu.Unlock() + + 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 @@ -357,19 +387,12 @@ func (pr *ProgressReporter) printSummaryStatus() { // Calculate ETA if we have total size and are processing etaStr := "" - if totalSize > 0 && bytesProcessed > 0 { - processStart, ok := pr.stats.ProcessStartTime.Load().(time.Time) - if ok && !processStart.IsZero() { - processElapsed := time.Since(processStart) - - rate := float64(bytesProcessed) / processElapsed.Seconds() - if rate > 0 { - remainingBytes := totalSize - bytesProcessed - remainingSeconds := float64(remainingBytes) / rate - eta := time.Duration(remainingSeconds * float64(time.Second)) - etaStr = " | ETA: " + formatDuration(eta) - } - } + processRate := pr.stats.processRate() + if totalSize > 0 && processRate > 0 { + remainingBytes := totalSize - bytesProcessed + remainingSeconds := float64(remainingBytes) / processRate + eta := time.Duration(remainingSeconds * float64(time.Second)) + etaStr = " | ETA: " + formatDuration(eta) } rate := float64(bytesScanned+bytesSkipped) / elapsed.Seconds() @@ -421,25 +444,18 @@ func (pr *ProgressReporter) printDetailedStatus() { log.Info("Elapsed time", "duration", formatDuration(elapsed)) // Calculate and show ETA if we have data - if totalSize > 0 && bytesProcessed > 0 { - processStart, ok := pr.stats.ProcessStartTime.Load().(time.Time) - if ok && !processStart.IsZero() { - processElapsed := time.Since(processStart) - - processRate := float64(bytesProcessed) / processElapsed.Seconds() - if processRate > 0 { - remainingBytes := totalSize - bytesProcessed - remainingSeconds := float64(remainingBytes) / processRate - eta := time.Duration(remainingSeconds * float64(time.Second)) - percentComplete := float64(bytesProcessed) / float64(totalSize) * percentScale - log.Info("Overall progress", - "percent", fmt.Sprintf("%.1f%%", percentComplete), - "processed", humanize.Bytes(safeUint64(bytesProcessed)), - "total", humanize.Bytes(safeUint64(totalSize)), - "rate", humanize.Bytes(uint64(processRate))+"/s", - "eta", formatDuration(eta)) - } - } + processRate := pr.stats.processRate() + if totalSize > 0 && processRate > 0 { + remainingBytes := totalSize - bytesProcessed + remainingSeconds := float64(remainingBytes) / processRate + eta := time.Duration(remainingSeconds * float64(time.Second)) + percentComplete := float64(bytesProcessed) / float64(totalSize) * percentScale + log.Info("Overall progress", + "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", diff --git a/internal/snapshot/progress_rate_test.go b/internal/snapshot/progress_rate_test.go new file mode 100644 index 0000000..69d0689 --- /dev/null +++ b/internal/snapshot/progress_rate_test.go @@ -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) + } +} diff --git a/internal/snapshot/progress_test.go b/internal/snapshot/progress_test.go new file mode 100644 index 0000000..ed63c91 --- /dev/null +++ b/internal/snapshot/progress_test.go @@ -0,0 +1,94 @@ +package snapshot_test + +import ( + "context" + "path/filepath" + "strings" + "testing" + + "github.com/spf13/afero" + "sneak.berlin/go/vaultik/internal/database" + "sneak.berlin/go/vaultik/internal/snapshot" +) + +// TestProgressPercentWithSecondPathSmaller backs up two paths with one +// scanner, as a snapshot with two paths does, the second path smaller +// than the first. The progress line divides the bytes processed by the +// total size, so both must count both paths to stay within 100%. +func TestProgressPercentWithSecondPathSmaller(t *testing.T) { + t.Parallel() + + fs := afero.NewMemMapFs() + files := map[string]string{ + "/large/one.txt": strings.Repeat("1", 4000), + "/large/two.txt": strings.Repeat("2", 4000), + "/small/three.txt": strings.Repeat("3", 1000), + } + + for path, content := range files { + err := fs.MkdirAll(filepath.Dir(path), 0755) + if err != nil { + t.Fatalf("mkdir: %v", err) + } + + err = afero.WriteFile(fs, path, []byte(content), 0644) + if err != nil { + t.Fatalf("write %s: %v", path, err) + } + } + + db, err := database.NewTestDB() + if err != nil { + t.Fatalf("create test db: %v", err) + } + + t.Cleanup(func() { + cerr := db.Close() + if cerr != nil { + t.Errorf("close db: %v", cerr) + } + }) + + repos := database.NewRepositories(db) + + scanner := snapshot.NewScanner(snapshot.ScannerConfig{ + FS: fs, + ChunkSize: int64(1024 * 16), + Repositories: repos, + MaxBlobSize: int64(1024 * 1024), + CompressionLevel: 3, + AgeRecipients: []string{testAgePublicKey}, + EnableProgress: true, + }) + + // Never started, but Stop releases its tickers and signal handler. + progress := scanner.GetProgress() + defer progress.Stop() + + ctx := context.Background() + snapshotID := "test-snapshot-progress" + createTestSnapshotRecord(ctx, t, repos, snapshotID) + + for _, path := range []string{"/large", "/small"} { + _, err := scanner.Scan(ctx, path, snapshotID) + if err != nil { + t.Fatalf("scanning %s: %v", path, err) + } + } + + stats := progress.GetStats() + + // Directories count toward the total size but produce no chunks, so + // the percentage ends just under 100%. + percent := float64(stats.BytesProcessed.Load()) / + float64(stats.TotalSize.Load()) * 100 + if percent > 100 { + t.Errorf("progress after both paths is %.1f%%, want at most 100%%", + percent) + } + + if stats.FilesProcessed.Load() != stats.TotalFiles.Load() { + t.Errorf("progress after both paths is %d of %d files, want all", + stats.FilesProcessed.Load(), stats.TotalFiles.Load()) + } +} diff --git a/internal/snapshot/scanner.go b/internal/snapshot/scanner.go index 46765b2..79d7e33 100644 --- a/internal/snapshot/scanner.go +++ b/internal/snapshot/scanner.go @@ -407,8 +407,8 @@ func (s *Scanner) summarizeScanPhase( } if s.progress != nil { - s.progress.SetTotalSize(totalSizeToProcess) - s.progress.GetStats().TotalFiles.Store(int64(len(filesToProcess))) + s.progress.AddTotalSize(totalSizeToProcess) + s.progress.GetStats().TotalFiles.Add(int64(len(filesToProcess))) } log.Info("Phase 1 complete",