From 840e0e271d28584200d7d6304bad6cb5b993beac Mon Sep 17 00:00:00 2001
From: Cristian Klein <cristian.klein@elastisys.com>
Date: Wed, 1 Jan 2020 20:44:06 +0100
Subject: [PATCH] fix(tiller): improve handling of corrupted storage

For some reason, many users experince corrupted storage with the
ConfigMaps storage backend. Specifically, several Releases are marked as
DEPLOYED. This patch improved handling of such situations, by taking the latest
DEPLOYED Release. Eventually, the storage will clean itself out, after
the corrupted Releases are deleted due to --history-max.

Closes #6031

Signed-off-by: Cristian Klein <cristian.klein@elastisys.com>
---
 pkg/storage/storage.go      |  3 +++
 pkg/storage/storage_test.go | 40 +++++++++++++++++++++++++++++++++++++
 2 files changed, 43 insertions(+)

diff --git a/pkg/storage/storage.go b/pkg/storage/storage.go
index e79cacc8d..d319c0c93 100644
--- a/pkg/storage/storage.go
+++ b/pkg/storage/storage.go
@@ -136,6 +136,9 @@ func (s *Storage) Deployed(name string) (*rspb.Release, error) {
 		return nil, fmt.Errorf("%q %s", name, NoReleasesErr)
 	}
 
+	// Sometimes Tiller's database gets corrupted and multiple releases are DEPLOYED. Take the latest.
+	relutil.Reverse(ls, relutil.SortByRevision)
+
 	return ls[0], err
 }
 
diff --git a/pkg/storage/storage_test.go b/pkg/storage/storage_test.go
index f7f3a86c7..acf6b3572 100644
--- a/pkg/storage/storage_test.go
+++ b/pkg/storage/storage_test.go
@@ -205,6 +205,46 @@ func TestStorageDeployed(t *testing.T) {
 	}
 }
 
+func TestStorageDeployedWithCorruption(t *testing.T) {
+	storage := Init(driver.NewMemory())
+
+	const name = "angry-bird"
+	const vers = int32(4)
+
+	// setup storage with test releases
+	setup := func() {
+		// release records (notice odd order and corruption)
+		rls0 := ReleaseTestData{Name: name, Version: 1, Status: rspb.Status_SUPERSEDED}.ToRelease()
+		rls1 := ReleaseTestData{Name: name, Version: 4, Status: rspb.Status_DEPLOYED}.ToRelease()
+		rls2 := ReleaseTestData{Name: name, Version: 3, Status: rspb.Status_SUPERSEDED}.ToRelease()
+		rls3 := ReleaseTestData{Name: name, Version: 2, Status: rspb.Status_DEPLOYED}.ToRelease()
+
+		// create the release records in the storage
+		assertErrNil(t.Fatal, storage.Create(rls0), "Storing release 'angry-bird' (v1)")
+		assertErrNil(t.Fatal, storage.Create(rls1), "Storing release 'angry-bird' (v2)")
+		assertErrNil(t.Fatal, storage.Create(rls2), "Storing release 'angry-bird' (v3)")
+		assertErrNil(t.Fatal, storage.Create(rls3), "Storing release 'angry-bird' (v4)")
+	}
+
+	setup()
+
+	rls, err := storage.Deployed(name)
+	if err != nil {
+		t.Fatalf("Failed to query for deployed release: %s\n", err)
+	}
+
+	switch {
+	case rls == nil:
+		t.Fatalf("Release is nil")
+	case rls.Name != name:
+		t.Fatalf("Expected release name %q, actual %q\n", name, rls.Name)
+	case rls.Version != vers:
+		t.Fatalf("Expected release version %d, actual %d\n", vers, rls.Version)
+	case rls.Info.Status.Code != rspb.Status_DEPLOYED:
+		t.Fatalf("Expected release status 'DEPLOYED', actual %s\n", rls.Info.Status.Code)
+	}
+}
+
 func TestStorageHistory(t *testing.T) {
 	storage := Init(driver.NewMemory())