added 'storage sweep' subcommand which removes unreferenced storage blocks

block/block_manager.go

@@ -560,6 +560,15 @@ func (bm *Manager) WriteBlock(data []byte, prefix string) (string, error) {
 	return blockID, err
 }
 
+// IsStorageBlockInUse determines whether given storage block is in use by currently loaded pack indexes.
+func (bm *Manager) IsStorageBlockInUse(storageBlockID string) bool {
+	bm.lock()
+	defer bm.unlock()
+
+	bm.ensurePackIndexesLoaded()
+	return bm.packBlockIDToIndex[storageBlockID] != nil
+}
+
 // Repackage reorganizes all pack blocks belonging to a given group that are not bigger than given size.
 func (bm *Manager) Repackage(maxLength uint64) error {
 	bm.lock()

cli/command_block_index_show.go

@@ -21,7 +21,24 @@ func runShowBlockIndexesAction(context *kingpin.ParseContext) error {
 	rep := mustOpenRepository(nil)
 	defer rep.Close()
 
-	for _, blockID := range *blockIndexShowIDs {
+	blockIDs := *blockIndexShowIDs
+	if len(blockIDs) == 1 || blockIDs[0] == "active" {
+		b, err := rep.Blocks.ActiveIndexBlocks()
+		if err != nil {
+			return err
+		}
+
+		sort.Slice(b, func(i, j int) bool {
+			return b[i].Timestamp.Before(b[j].Timestamp)
+		})
+
+		blockIDs = nil
+		for _, bi := range b {
+			blockIDs = append(blockIDs, bi.BlockID)
+		}
+	}
+
+	for _, blockID := range blockIDs {
 		data, err := rep.Blocks.GetBlock(blockID)
 		if err != nil {
 			return fmt.Errorf("can't read block %q: %v", blockID, err)
@@ -33,7 +50,7 @@ func runShowBlockIndexesAction(context *kingpin.ParseContext) error {
 		}
 
 		for _, ndx := range d.Indexes {
-			fmt.Printf("pack %v len: %v created %v\n", ndx.PackBlockId, ndx.PackLength, time.Unix(0, int64(ndx.CreateTimeNanos)).Local())
+			fmt.Printf("pack:%v len:%v created:%v\n", ndx.PackBlockId, ndx.PackLength, time.Unix(0, int64(ndx.CreateTimeNanos)).Local())
 			type unpacked struct {
 				blockID string
 				offset  uint32

cli/command_storage_sweep.go

@@ -0,0 +1,60 @@
+package cli
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/kopia/kopia/repo"
+
+	kingpin "gopkg.in/alecthomas/kingpin.v2"
+)
+
+var (
+	storageSweepCommand      = storageCommands.Command("sweep", "Remove unused storage blocks")
+	storageSweepAgeThreshold = storageSweepCommand.Flag("min-age", "Minimum age of storage block to delete").Default("24h").Duration()
+	storageSweepPrintUnused  = storageSweepCommand.Flag("print", "Do not delete blocks, but print which ones would be deleted").Short('n').Bool()
+)
+
+func runStorageSweepAction(context *kingpin.ParseContext) error {
+	rep := mustOpenRepository(nil)
+	defer rep.Close()
+
+	indexBlocks, err := rep.Blocks.ActiveIndexBlocks()
+	if err != nil {
+		return err
+	}
+
+	inUseIndexBlocks := map[string]bool{}
+	for _, ib := range indexBlocks {
+		inUseIndexBlocks[ib.BlockID] = true
+	}
+
+	ch, cancel := rep.Storage.ListBlocks("")
+	defer cancel()
+
+	for bm := range ch {
+		if bm.Error != nil {
+			return bm.Error
+		}
+
+		age := time.Since(bm.TimeStamp)
+		inUse := rep.Blocks.IsStorageBlockInUse(bm.BlockID) || inUseIndexBlocks[bm.BlockID] || bm.BlockID == repo.FormatBlockID
+		keep := inUse || age < *storageSweepAgeThreshold
+		if !keep {
+			if *storageSweepPrintUnused {
+				fmt.Printf("unused block %v age %v\n", bm.BlockID, age)
+			} else {
+				fmt.Printf("deleting unused block %v age %v\n", bm.BlockID, age)
+				if err := rep.Storage.DeleteBlock(bm.BlockID); err != nil {
+					return err
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
+func init() {
+	storageSweepCommand.Action(runStorageSweepAction)
+}

repo/connect.go

@@ -32,7 +32,7 @@ type ConnectOptions struct {
 
 // Connect connects to the repository in the specified storage and persists the configuration and credentials in the file provided.
 func Connect(ctx context.Context, configFile string, st storage.Storage, creds auth.Credentials, opt ConnectOptions) error {
-	formatBytes, err := st.GetBlock(formatBlockID, 0, -1)
+	formatBytes, err := st.GetBlock(FormatBlockID, 0, -1)
 	if err != nil {
 		return fmt.Errorf("unable to read format block: %v", err)
 	}

repo/format_block.go

@@ -14,9 +14,8 @@ import (
 	"github.com/kopia/kopia/storage"
 )
 
-const (
-	formatBlockID = "kopia.repository"
-)
+// FormatBlockID is the identifier of a storage block that describes repository format.
+const FormatBlockID = "kopia.repository"
 
 var (
 	purposeAESKey   = []byte("AES")
@@ -59,7 +58,7 @@ func writeFormatBlock(st storage.Storage, f *formatBlock) error {
 		return fmt.Errorf("unable to marshal format block: %v", err)
 	}
 
-	if err := st.PutBlock(formatBlockID, buf.Bytes()); err != nil {
+	if err := st.PutBlock(FormatBlockID, buf.Bytes()); err != nil {
 		return fmt.Errorf("unable to write format block: %v", err)
 	}
 

repo/open.go

@@ -172,7 +172,7 @@ func readAndCacheFormatBlock(st storage.Storage, cacheDirectory string) (*format
 		}
 	}
 
-	b, err := st.GetBlock(formatBlockID, 0, -1)
+	b, err := st.GetBlock(FormatBlockID, 0, -1)
 	if err != nil {
 		return nil, err
 	}