From 1f0582cadc1ba222c5d7a7f9977a091edd55f6af 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(helm): improve handling of corrupted storage

Helm does not yet properly handle concurrent executions (see #7322),
and invoking Helm concurrently on the same release lead to corrupted storage.
Specifically, several Releases may be 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.

This is a port to Helm v3 of #7319.

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

diff --git a/pkg/storage/storage.go b/pkg/storage/storage.go
index 56881493b..89183355f 100644
--- a/pkg/storage/storage.go
+++ b/pkg/storage/storage.go
@@ -119,6 +119,10 @@ func (s *Storage) Deployed(name string) (*rspb.Release, error) {
 		return nil, errors.Errorf("%q has no deployed releases", name)
 	}
 
+	// If executed concurrently, Helm's database gets corrupted
+	// and multiple releases are DEPLOYED. Take the latest.
+	relutil.Reverse(ls, relutil.SortByRevision)
+
 	return ls[0], nil
 }
 
diff --git a/pkg/storage/storage_test.go b/pkg/storage/storage_test.go
index b8c333d24..ee9c68b80 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 = int(4)
+
+	// setup storage with test releases
+	setup := func() {
+		// release records (notice odd order and corruption)
+		rls0 := ReleaseTestData{Name: name, Version: 1, Status: rspb.StatusSuperseded}.ToRelease()
+		rls1 := ReleaseTestData{Name: name, Version: 4, Status: rspb.StatusDeployed}.ToRelease()
+		rls2 := ReleaseTestData{Name: name, Version: 3, Status: rspb.StatusSuperseded}.ToRelease()
+		rls3 := ReleaseTestData{Name: name, Version: 2, Status: rspb.StatusDeployed}.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 != rspb.StatusDeployed:
+		t.Fatalf("Expected release status 'DEPLOYED', actual %s\n", rls.Info.Status.String())
+	}
+}
+
 func TestStorageHistory(t *testing.T) {
 	storage := Init(driver.NewMemory())