feat(cli): improved safety of v1->v2 index format upgrade (#2223)

When upgrading from legacy to epoch manager-based index, we will write
an intentionally-corrupted index blob, such that old clients won't be
able to understand it when they read the repository index using legacy
format.

The error message emitted by very old clients is not great, but it's
safer to do that rather than corrupt the repository.

Note that this additional safety has a delay of up to 15 minutes
which is the time required for old clients to stop relying on index list
cache in case of very long-running snapshots, server or KopiaUI.

cli/command_blob_list.go

@@ -57,7 +57,7 @@ func (c *commandBlobList) run(ctx context.Context, rep repo.DirectRepository) er
 
 func (c *commandBlobList) shouldInclude(b blob.Metadata) bool {
 	if c.dataOnly {
-		if strings.HasPrefix(string(b.BlobID), content.IndexBlobPrefix) {
+		if strings.HasPrefix(string(b.BlobID), content.LegacyIndexBlobPrefix) {
 			return false
 		}
 

cli/command_index_recover.go

@@ -58,7 +58,7 @@ func (c *commandIndexRecover) run(ctx context.Context, rep repo.DirectRepository
 	}()
 
 	if c.deleteIndexes {
-		if err := rep.BlobReader().ListBlobs(ctx, content.IndexBlobPrefix, func(bm blob.Metadata) error {
+		if err := rep.BlobReader().ListBlobs(ctx, content.LegacyIndexBlobPrefix, func(bm blob.Metadata) error {
 			if c.commit {
 				log(ctx).Infof("deleting old index blob: %v", bm.BlobID)
 				return errors.Wrap(rep.BlobStorage().DeleteBlob(ctx, bm.BlobID), "error deleting index blob")

cli/command_repository_set_parameters.go

@@ -172,6 +172,12 @@ func (c *commandRepositorySetParameters) run(ctx context.Context, rep repo.Direc
 		return errors.Wrap(err, "error setting parameters")
 	}
 
+	if upgradeToEpochManager {
+		if err := content.WriteLegacyIndexPoisonBlob(ctx, rep.BlobStorage()); err != nil {
+			log(ctx).Errorf("unable to write legacy index poison blob: %v", err)
+		}
+	}
+
 	log(ctx).Infof("NOTE: Repository parameters updated, you must disconnect and re-connect all other Kopia clients.")
 
 	return nil

cli/command_repository_upgrade.go

@@ -269,6 +269,12 @@ func (c *commandRepositoryUpgrade) upgrade(ctx context.Context, rep repo.DirectR
 	if err := rep.SetParameters(ctx, mp, rep.BlobCfg()); err != nil {
 		return errors.Wrap(err, "error setting parameters")
 	}
+
+	// poison V0 index so that old readers won't be able to open it.
+	if err := content.WriteLegacyIndexPoisonBlob(ctx, rep.BlobStorage()); err != nil {
+		log(ctx).Errorf("unable to write legacy index poison blob: %v", err)
+	}
+
 	// we need to reopen the repository after this point
 
 	log(ctx).Infof("Repository indices have been upgraded.")

repo/content/committed_read_manager.go

@@ -42,7 +42,7 @@ const (
 
 // nolint:gochecknoglobals
 var cachedIndexBlobPrefixes = []blob.ID{
-	IndexBlobPrefix,
+	LegacyIndexBlobPrefix,
 	compactionLogBlobPrefix,
 	cleanupBlobPrefix,
 
@@ -54,7 +54,7 @@ var cachedIndexBlobPrefixes = []blob.ID{
 
 // nolint:gochecknoglobals
 var allIndexBlobPrefixes = []blob.ID{
-	IndexBlobPrefix,
+	LegacyIndexBlobPrefix,
 	epoch.UncompactedIndexBlobPrefix,
 	epoch.SingleEpochCompactionBlobPrefix,
 	epoch.RangeCheckpointIndexBlobPrefix,

repo/content/content_index_recovery_test.go

@@ -28,7 +28,7 @@ func (s *contentManagerSuite) TestContentIndexRecovery(t *testing.T) {
 	}
 
 	// delete all index blobs
-	require.NoError(t, bm.st.ListBlobs(ctx, IndexBlobPrefix, func(bi blob.Metadata) error {
+	require.NoError(t, bm.st.ListBlobs(ctx, LegacyIndexBlobPrefix, func(bi blob.Metadata) error {
 		t.Logf("deleting %v", bi.BlobID)
 		return bm.st.DeleteBlob(ctx, bi.BlobID)
 	}))

repo/content/content_manager.go

@@ -50,9 +50,6 @@ var PackBlobIDPrefixes = []blob.ID{
 	PackBlobIDPrefixSpecial,
 }
 
-// IndexBlobPrefix is the prefix for all index blobs.
-const IndexBlobPrefix = "n"
-
 const (
 	parallelFetches          = 5                // number of parallel reads goroutines
 	flushPackIndexTimeout    = 10 * time.Minute // time after which all pending indexes are flushes

repo/content/content_manager_test.go

@@ -550,7 +550,7 @@ func validateIndexCount(t *testing.T, data map[blob.ID][]byte, wantIndexCount, w
 	var indexCnt, compactionLogCnt int
 
 	for blobID := range data {
-		if strings.HasPrefix(string(blobID), IndexBlobPrefix) || strings.HasPrefix(string(blobID), "x") {
+		if strings.HasPrefix(string(blobID), LegacyIndexBlobPrefix) || strings.HasPrefix(string(blobID), "x") {
 			indexCnt++
 		}
 
@@ -1202,7 +1202,7 @@ func (s *contentManagerSuite) TestFlushWaitsForAllPendingWriters(t *testing.T) {
 	bm.Flush(ctx)
 	t.Logf("<<< end of flushing")
 
-	indexBlobPrefix := blob.ID(IndexBlobPrefix)
+	indexBlobPrefix := blob.ID(LegacyIndexBlobPrefix)
 	if s.mutableParameters.EpochParameters.Enabled {
 		indexBlobPrefix = "x"
 	}

repo/content/index_blob_manager_v0.go

@@ -14,7 +14,12 @@ import (
 	"github.com/kopia/kopia/repo/logging"
 )
 
+// LegacyIndexBlobPrefix is the prefix for all legacy (v0) index blobs.
+const LegacyIndexBlobPrefix = "n"
+
 const (
+	legacyIndexPoisonBlobID = "n00000000000000000000000000000000-repository_unreadable_by_this_kopia_version_upgrade_required"
+
 	defaultEventualConsistencySettleTime = 1 * time.Hour
 	compactionLogBlobPrefix              = "m"
 	cleanupBlobPrefix                    = "l"
@@ -75,7 +80,7 @@ func (m *indexBlobManagerV0) listActiveIndexBlobs(ctx context.Context) ([]IndexB
 	})
 
 	eg.Go(func() error {
-		v, err := blob.ListAllBlobs(ctx, m.st, IndexBlobPrefix)
+		v, err := blob.ListAllBlobs(ctx, m.st, LegacyIndexBlobPrefix)
 		storageIndexBlobs = v
 
 		return errors.Wrap(err, "error listing index blobs")
@@ -183,7 +188,7 @@ func (m *indexBlobManagerV0) writeIndexBlobs(ctx context.Context, dataShards []g
 	var result []blob.Metadata
 
 	for _, data := range dataShards {
-		bm, err := m.enc.encryptAndWriteBlob(ctx, data, IndexBlobPrefix, sessionID)
+		bm, err := m.enc.encryptAndWriteBlob(ctx, data, LegacyIndexBlobPrefix, sessionID)
 		if err != nil {
 			return nil, errors.Wrap(err, "error writing index blbo")
 		}
@@ -517,6 +522,18 @@ func (m *indexBlobManagerV0) dropContentsFromBuilder(bld index.Builder, opt Comp
 	}
 }
 
+// WriteLegacyIndexPoisonBlob writes a "poison blob" that will prevent old kopia clients
+// that have not been upgraded from being able to open the repository after its format
+// has been upgraded.
+func WriteLegacyIndexPoisonBlob(ctx context.Context, st blob.Storage) error {
+	// nolint:wrapcheck
+	return st.PutBlob(
+		ctx,
+		legacyIndexPoisonBlobID,
+		gather.FromSlice([]byte("The format of this repository has been upgraded and cannot be read by old clients")),
+		blob.PutOptions{})
+}
+
 func addIndexBlobsToBuilder(ctx context.Context, enc *encryptedBlobMgr, bld index.Builder, indexBlobID blob.ID) error {
 	var data gather.WriteBuffer
 	defer data.Close()

repo/content/index_blob_manager_v0_test.go

@@ -699,7 +699,7 @@ func getAllFakeContentsInternal(ctx context.Context, t *testing.T, m *indexBlobM
 func assertBlobCounts(t *testing.T, data blobtesting.DataMap, wantN, wantM, wantL int) {
 	t.Helper()
 	require.Len(t, keysWithPrefix(data, compactionLogBlobPrefix), wantM)
-	require.Len(t, keysWithPrefix(data, IndexBlobPrefix), wantN)
+	require.Len(t, keysWithPrefix(data, LegacyIndexBlobPrefix), wantN)
 	require.Len(t, keysWithPrefix(data, "l"), wantL)
 }
 
@@ -783,7 +783,7 @@ func newIndexBlobManagerForTesting(t *testing.T, st blob.Storage, localTimeNow f
 	st = ownwrites.NewWrapper(
 		st,
 		blobtesting.NewMapStorage(blobtesting.DataMap{}, nil, nil),
-		[]blob.ID{IndexBlobPrefix, compactionLogBlobPrefix, cleanupBlobPrefix},
+		[]blob.ID{LegacyIndexBlobPrefix, compactionLogBlobPrefix, cleanupBlobPrefix},
 		15*time.Minute,
 	)
 

repo/open.go

@@ -389,7 +389,7 @@ func wrapLockingStorage(st blob.Storage, r content.BlobCfgBlob) blob.Storage {
 		prefixes = append(prefixes, string(prefix))
 	}
 
-	prefixes = append(prefixes, content.IndexBlobPrefix, epoch.EpochManagerIndexUberPrefix, FormatBlobID,
+	prefixes = append(prefixes, content.LegacyIndexBlobPrefix, epoch.EpochManagerIndexUberPrefix, FormatBlobID,
 		BlobCfgBlobID)
 
 	return beforeop.NewWrapper(st, nil, nil, nil, func(ctx context.Context, id blob.ID, opts *blob.PutOptions) error {

repo/repository_test.go

@@ -565,7 +565,7 @@ func TestObjectWritesWithRetention(t *testing.T) {
 		prefixesWithRetention = append(prefixesWithRetention, string(prefix))
 	}
 
-	prefixesWithRetention = append(prefixesWithRetention, content.IndexBlobPrefix, epoch.EpochManagerIndexUberPrefix,
+	prefixesWithRetention = append(prefixesWithRetention, content.LegacyIndexBlobPrefix, epoch.EpochManagerIndexUberPrefix,
 		repo.FormatBlobID, repo.BlobCfgBlobID)
 
 	// make sure that we cannot set mtime on the kopia objects created due to the

tests/compat_test/compat_test.go

@@ -44,9 +44,13 @@ func TestRepoCreatedWith08CanBeOpenedWithCurrent(t *testing.T) {
 		"--io-drain-timeout", "1s", "--allow-unsafe-upgrade",
 		"--status-poll-interval", "1s")
 
-	// now 0.8 can't open it anymore
+	// now 0.8 client can't open it anymore because they won't understand format V2
 	e3 := testenv.NewCLITest(t, testenv.RepoFormatNotImportant, runner08)
 	e3.RunAndExpectFailure(t, "repo", "connect", "filesystem", "--path", e1.RepoDir)
+
+	// old 0.8 client who has cached the format blob and never disconnected
+	// can't open the repository because of the poison blob
+	e1.RunAndExpectFailure(t, "snap", "ls")
 }
 
 func TestRepoCreatedWithCurrentWithFormatVersion1CanBeOpenedWith08(t *testing.T) {