cli: implemented 'snapshot gc' and moved verify to be snapshot verify

`snapshot gc` marks contents not reachable from the root of any snapshot
as soft-deleted

The algorithm is a mark-and-sweep with parallel iteration of objects.
Currently it stores content IDs and object IDs in a map, so won't scale
to huge repositories, but this can be fixed in the future.

This fixes #110 at least for reasonable repository sizes.

cli/command_snapshot_gc.go

@@ -0,0 +1,130 @@
+package cli
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/pkg/errors"
+
+	"github.com/kopia/kopia/internal/units"
+	"github.com/kopia/kopia/repo"
+	"github.com/kopia/kopia/repo/content"
+	"github.com/kopia/kopia/repo/manifest"
+	"github.com/kopia/kopia/repo/object"
+	"github.com/kopia/kopia/snapshot"
+	"github.com/kopia/kopia/snapshot/snapshotfs"
+)
+
+var (
+	snapshotGCCommand       = snapshotCommands.Command("gc", "Remove contents not used by any snapshot")
+	snapshotGCMinContentAge = snapshotGCCommand.Flag("min-age", "Minimum content age to allow deletion").Default("24h").Duration()
+	snapshotGCDelete        = snapshotGCCommand.Flag("delete", "Delete unreferenced contents").Bool()
+)
+
+func findInUseContentIDs(ctx context.Context, rep *repo.Repository, used *sync.Map) error {
+	w := snapshotfs.NewTreeWalker()
+
+	ids, err := snapshot.ListSnapshotManifests(ctx, rep, nil)
+	if err != nil {
+		return errors.Wrap(err, "unable to list snapshot manifest IDs")
+	}
+
+	manifests, err := snapshot.LoadSnapshots(ctx, rep, ids)
+	if err != nil {
+		return errors.Wrap(err, "unable to load manifest IDs")
+	}
+
+	for _, m := range manifests {
+		root, err := snapshotfs.SnapshotRoot(rep, m)
+		if err != nil {
+			return errors.Wrap(err, "unable to get snapshot root")
+		}
+		w.RootEntries = append(w.RootEntries, root)
+	}
+
+	w.ObjectCallback = func(oid object.ID) error {
+		_, contentIDs, err := rep.Objects.VerifyObject(ctx, oid)
+		if err != nil {
+			return errors.Wrapf(err, "error verifying %v", oid)
+		}
+		for _, cid := range contentIDs {
+			used.Store(cid, nil)
+		}
+		return nil
+	}
+
+	log.Infof("looking for active contents")
+	if err := w.Run(ctx); err != nil {
+		return errors.Wrap(err, "error walking snapshot tree")
+	}
+
+	return nil
+}
+
+func runSnapshotGCCommand(ctx context.Context, rep *repo.Repository) error {
+	var used sync.Map
+	if err := findInUseContentIDs(ctx, rep, &used); err != nil {
+		return errors.Wrap(err, "unable to find in-use content ID")
+	}
+
+	var unusedCount, inUseCount, systemCount, tooRecentCount int32
+	var totalUnusedBytes, totalInUseBytes, totalSystemBytes, totalTooRecentBytes int64
+
+	log.Infof("looking for unreferenced contents")
+	if err := rep.Content.IterateContents(content.IterateOptions{}, func(ci content.Info) error {
+		if manifest.ContentPrefix == ci.ID.Prefix() {
+			atomic.AddInt32(&systemCount, 1)
+			atomic.AddInt64(&totalSystemBytes, int64(ci.Length))
+			return nil
+		}
+
+		if _, ok := used.Load(ci.ID); !ok {
+			if time.Since(ci.Timestamp()) < *snapshotGCMinContentAge {
+				log.Debugf("recent unreferenced content %v (%v bytes, modified %v)", ci.ID, ci.Length, ci.Timestamp())
+				atomic.AddInt32(&tooRecentCount, 1)
+				atomic.AddInt64(&totalTooRecentBytes, int64(ci.Length))
+				return nil
+			}
+			log.Debugf("unreferenced %v (%v bytes, modified %v)", ci.ID, ci.Length, ci.Timestamp())
+			cnt := atomic.AddInt32(&unusedCount, 1)
+			totalSize := atomic.AddInt64(&totalUnusedBytes, int64(ci.Length))
+			if *snapshotGCDelete {
+				if err := rep.Content.DeleteContent(ci.ID); err != nil {
+					return errors.Wrap(err, "error deleting content")
+				}
+			}
+
+			if cnt%100000 == 0 {
+				log.Infof("... found %v unused contents so far (%v bytes)", cnt, units.BytesStringBase2(totalSize))
+				if *snapshotGCDelete {
+					if err := rep.Flush(ctx); err != nil {
+						return errors.Wrap(err, "flush error")
+					}
+				}
+			}
+		} else {
+			atomic.AddInt32(&inUseCount, 1)
+			atomic.AddInt64(&totalInUseBytes, int64(ci.Length))
+		}
+		return nil
+	}); err != nil {
+		return errors.Wrap(err, "error iterating contents")
+	}
+
+	log.Infof("found %v unused contents (%v bytes)", unusedCount, units.BytesStringBase2(totalUnusedBytes))
+	log.Infof("found %v unused contents that are too recent to delete (%v bytes)", tooRecentCount, units.BytesStringBase2(totalTooRecentBytes))
+	log.Infof("found %v in-use contents (%v bytes)", inUseCount, units.BytesStringBase2(totalInUseBytes))
+	log.Infof("found %v in-use system-contents (%v bytes)", systemCount, units.BytesStringBase2(totalSystemBytes))
+
+	if unusedCount > 0 && !*snapshotGCDelete {
+		return errors.Errorf("Not deleting because '--delete' flag was not set.")
+	}
+
+	return nil
+}
+
+func init() {
+	snapshotGCCommand.Action(repositoryAction(runSnapshotGCCommand))
+}

cli/command_snapshot_verify.go

@@ -21,7 +21,7 @@ import (
 )
 
 var (
-	verifyCommand               = app.Command("verify", "Verify the contents of stored object")
+	verifyCommand               = snapshotCommands.Command("verify", "Verify the contents of stored snapshot")
 	verifyCommandErrorThreshold = verifyCommand.Flag("max-errors", "Maximum number of errors before stopping").Default("0").Int()
 	verifyCommandDirObjectIDs   = verifyCommand.Flag("directory-id", "Directory object IDs to verify").Strings()
 	verifyCommandFileObjectIDs  = verifyCommand.Flag("file-id", "File object IDs to verify").Strings()
@@ -96,8 +96,8 @@ func (v *verifier) enqueueVerifyDirectory(ctx context.Context, oid object.ID, pa
 	if !v.shouldEnqueue(oid) {
 		return
 	}
-	v.workQueue.EnqueueFront(func() {
-		v.doVerifyDirectory(ctx, oid, path)
+	v.workQueue.EnqueueFront(func() error {
+		return v.doVerifyDirectory(ctx, oid, path)
 	})
 }
 
@@ -106,19 +106,19 @@ func (v *verifier) enqueueVerifyObject(ctx context.Context, oid object.ID, path
 	if !v.shouldEnqueue(oid) {
 		return
 	}
-	v.workQueue.EnqueueBack(func() {
-		v.doVerifyObject(ctx, oid, path, expectedLength)
+	v.workQueue.EnqueueBack(func() error {
+		return v.doVerifyObject(ctx, oid, path, expectedLength)
 	})
 }
 
-func (v *verifier) doVerifyDirectory(ctx context.Context, oid object.ID, path string) {
+func (v *verifier) doVerifyDirectory(ctx context.Context, oid object.ID, path string) error {
 	log.Debugf("verifying directory %q (%v)", path, oid)
 
 	d := snapshotfs.DirectoryEntry(v.rep, oid, nil)
 	entries, err := d.Readdir(ctx)
 	if err != nil {
 		v.reportError(path, errors.Wrapf(err, "error reading %v", oid))
-		return
+		return nil
 	}
 
 	for _, e := range entries {
@@ -134,9 +134,11 @@ func (v *verifier) doVerifyDirectory(ctx context.Context, oid object.ID, path st
 			v.enqueueVerifyObject(ctx, objectID, childPath, e.Size())
 		}
 	}
+
+	return nil
 }
 
-func (v *verifier) doVerifyObject(ctx context.Context, oid object.ID, path string, expectedLength int64) {
+func (v *verifier) doVerifyObject(ctx context.Context, oid object.ID, path string, expectedLength int64) error {
 	if expectedLength < 0 {
 		log.Debugf("verifying object %v", oid)
 	} else {
@@ -160,6 +162,8 @@ func (v *verifier) doVerifyObject(ctx context.Context, oid object.ID, path strin
 			v.reportError(path, errors.Wrapf(err, "error reading object %v", oid))
 		}
 	}
+
+	return nil
 }
 
 func (v *verifier) readEntireObject(ctx context.Context, oid object.ID, path string) error {
@@ -191,7 +195,9 @@ func runVerifyCommand(ctx context.Context, rep *repo.Repository) error {
 	}
 
 	v.workQueue.ProgressCallback = v.progressCallback
-	v.workQueue.Process(*verifyCommandParallel)
+	if err := v.workQueue.Process(*verifyCommandParallel); err != nil {
+		return errors.Wrap(err, "error processing work queue")
+	}
 
 	if len(v.errors) == 0 {
 		return nil

internal/parallelwork/parallel_work_queue.go

@@ -20,17 +20,20 @@ type Queue struct {
 	ProgressCallback func(enqueued, active, completed int64)
 }
 
+// CallbackFunc is a callback function
+type CallbackFunc func() error
+
 // EnqueueFront adds the work to the front of the queue.
-func (v *Queue) EnqueueFront(callback func()) {
+func (v *Queue) EnqueueFront(callback CallbackFunc) {
 	v.enqueue(true, callback)
 }
 
 // EnqueueBack adds the work to the back of the queue.
-func (v *Queue) EnqueueBack(callback func()) {
+func (v *Queue) EnqueueBack(callback CallbackFunc) {
 	v.enqueue(false, callback)
 }
 
-func (v *Queue) enqueue(front bool, callback func()) {
+func (v *Queue) enqueue(front bool, callback CallbackFunc) {
 	v.monitor.L.Lock()
 	defer v.monitor.L.Unlock()
 
@@ -48,8 +51,9 @@ func (v *Queue) enqueue(front bool, callback func()) {
 
 // Process starts N workers, which will be processing elements in the queue until the queue
 // is empty and all workers are idle.
-func (v *Queue) Process(workers int) {
+func (v *Queue) Process(workers int) error {
 	var wg sync.WaitGroup
+	errors := make(chan error, workers)
 
 	for i := 0; i < workers; i++ {
 		wg.Add(1)
@@ -63,16 +67,23 @@ func (v *Queue) Process(workers int) {
 				if callback == nil {
 					break
 				}
-				callback()
+				if err := callback(); err != nil {
+					errors <- err
+					break
+				}
 				v.completed()
 			}
 		}(i)
 	}
 
 	wg.Wait()
+	close(errors)
+
+	// return first error or nil
+	return <-errors
 }
 
-func (v *Queue) dequeue() func() {
+func (v *Queue) dequeue() CallbackFunc {
 	v.monitor.L.Lock()
 	defer v.monitor.L.Unlock()
 
@@ -91,7 +102,7 @@ func (v *Queue) dequeue() func() {
 
 	front := v.queueItems.Front()
 	v.queueItems.Remove(front)
-	return front.Value.(func())
+	return front.Value.(CallbackFunc)
 }
 
 func (v *Queue) completed() {

repo/manifest/manifest_manager.go

@@ -23,7 +23,7 @@ var log = repologging.Logger("kopia/manifest")
 // ErrNotFound is returned when the metadata item is not found.
 var ErrNotFound = errors.New("not found")
 
-const manifestContentPrefix = "m"
+const ContentPrefix = "m"
 const autoCompactionContentCount = 16
 
 type contentManager interface {
@@ -228,7 +228,7 @@ func (m *Manager) flushPendingEntriesLocked(ctx context.Context) (content.ID, er
 	mustSucceed(gz.Flush())
 	mustSucceed(gz.Close())
 
-	contentID, err := m.b.WriteContent(ctx, buf.Bytes(), manifestContentPrefix)
+	contentID, err := m.b.WriteContent(ctx, buf.Bytes(), ContentPrefix)
 	if err != nil {
 		return "", err
 	}
@@ -287,7 +287,7 @@ func (m *Manager) loadCommittedContentsLocked(ctx context.Context) error {
 		manifests = map[content.ID]manifest{}
 
 		err := m.b.IterateContents(content.IterateOptions{
-			Prefix:   manifestContentPrefix,
+			Prefix:   ContentPrefix,
 			Parallel: 8,
 		}, func(ci content.Info) error {
 			man, err := m.loadManifestContent(ctx, ci.ID)

repo/manifest/manifest_manager_test.go

@@ -99,7 +99,7 @@ func TestManifest(t *testing.T) {
 
 	foundContents := 0
 	if err := mgr.b.IterateContents(
-		content.IterateOptions{Prefix: manifestContentPrefix},
+		content.IterateOptions{Prefix: ContentPrefix},
 		func(ci content.Info) error {
 			foundContents++
 			return nil

snapshot/snapshotfs/snapshot_tree_walker.go

@@ -0,0 +1,74 @@
+package snapshotfs
+
+import (
+	"context"
+	"runtime"
+	"sync"
+
+	"github.com/pkg/errors"
+
+	"github.com/kopia/kopia/fs"
+	"github.com/kopia/kopia/internal/parallelwork"
+	"github.com/kopia/kopia/repo/object"
+)
+
+type TreeWalker struct {
+	Parallelism    int
+	RootEntries    []fs.Entry
+	ObjectCallback func(oid object.ID) error
+
+	enqueued sync.Map
+	queue    *parallelwork.Queue
+}
+
+func oidOf(entry fs.Entry) object.ID {
+	return entry.(object.HasObjectID).ObjectID()
+}
+
+func (w *TreeWalker) enqueueEntry(ctx context.Context, entry fs.Entry) {
+	oid := oidOf(entry)
+	if _, existing := w.enqueued.LoadOrStore(oid, w); existing {
+		return
+	}
+
+	w.queue.EnqueueBack(func() error { return w.processEntry(ctx, entry) })
+}
+
+func (w *TreeWalker) processEntry(ctx context.Context, entry fs.Entry) error {
+	err := w.ObjectCallback(oidOf(entry))
+	if err != nil {
+		return err
+	}
+
+	if dir, ok := entry.(fs.Directory); ok {
+		entries, err := dir.Readdir(ctx)
+		if err != nil {
+			return errors.Wrap(err, "error reading directory")
+		}
+
+		for _, ent := range entries {
+			w.enqueueEntry(ctx, ent)
+		}
+	}
+
+	return nil
+}
+
+func (w *TreeWalker) Run(ctx context.Context) error {
+	for _, root := range w.RootEntries {
+		w.enqueueEntry(ctx, root)
+	}
+	w.queue.ProgressCallback = func(enqueued, active, completed int64) {
+		log.Infof("processed(%v/%v) active %v", completed, enqueued, active)
+
+	}
+	return w.queue.Process(w.Parallelism)
+}
+
+// NewTreeWalker creates new tree walker.
+func NewTreeWalker() *TreeWalker {
+	return &TreeWalker{
+		Parallelism: 4 * runtime.NumCPU(),
+		queue:       parallelwork.NewQueue(),
+	}
+}

tests/end_to_end_test/end_to_end_test.go

@@ -222,6 +222,65 @@ func TestEndToEnd(t *testing.T) {
 	})
 }
 
+func TestSnapshotGC(t *testing.T) {
+	e := newTestEnv(t)
+	defer e.cleanup(t)
+	defer e.runAndExpectSuccess(t, "repo", "disconnect")
+
+	e.runAndExpectSuccess(t, "repo", "create", "filesystem", "--path", e.repoDir)
+
+	expectedContentCount := len(e.runAndExpectSuccess(t, "content", "list"))
+
+	dataDir := filepath.Join(e.dataDir, "dir1")
+	assertNoError(t, os.MkdirAll(dataDir, 0777))
+	assertNoError(t, ioutil.WriteFile(filepath.Join(dataDir, "some-file1"), []byte(`
+hello world
+how are you
+`), 0600))
+
+	// take a snapshot of a directory with 1 file
+	e.runAndExpectSuccess(t, "snap", "create", dataDir)
+
+	// data block + directory block + manifest block
+	expectedContentCount += 3
+	e.runAndVerifyOutputLineCount(t, expectedContentCount, "content", "list")
+
+	// now delete all manifests, making the content unreachable
+	for _, line := range e.runAndExpectSuccess(t, "snap", "list", "-m") {
+		p := strings.Index(line, "manifest:")
+		if p >= 0 {
+			manifestID := strings.TrimPrefix(strings.Split(line[p:], " ")[0], "manifest:")
+			t.Logf("manifestID: %v", manifestID)
+			e.runAndExpectSuccess(t, "manifest", "rm", manifestID)
+		}
+	}
+
+	// deletion of manifests creates a new manifest
+	expectedContentCount++
+
+	// run verification
+	e.runAndExpectSuccess(t, "snapshot", "verify", "--all-sources")
+
+	// garbage-collect in dry run mode
+	e.runAndExpectSuccess(t, "snapshot", "gc")
+
+	// data block + directory block + manifest block + manifest block from manifest deletion
+	e.runAndVerifyOutputLineCount(t, expectedContentCount, "content", "list")
+
+	// garbage-collect for real, but contents are too recent so won't be deleted
+	e.runAndExpectSuccess(t, "snapshot", "gc", "--delete")
+
+	// data block + directory block + manifest block + manifest block from manifest deletion
+	e.runAndVerifyOutputLineCount(t, expectedContentCount, "content", "list")
+
+	// garbage-collect for real, this time without age limit
+	e.runAndExpectSuccess(t, "snapshot", "gc", "--delete", "--min-age", "0s")
+
+	// two contents are deleted
+	expectedContentCount -= 2
+	e.runAndVerifyOutputLineCount(t, expectedContentCount, "content", "list")
+}
+
 func TestDiff(t *testing.T) {
 	e := newTestEnv(t)
 	defer e.cleanup(t)