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

@@ -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
 }
 

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())