if a volume containing incomplete metadata is found during a ListVolumes call, clean up and continue

Signed-off-by: James Munnelly <[email protected]>

Author: munnerz

storage/filesystem.go

@@ -111,9 +111,16 @@ func (f *Filesystem) ListVolumes() ([]string, error) {
 	var vols []string
 	for _, dir := range dirs {
 		file, err := f.fs.Open(f.metadataPathForVolumeID(dir.Name()))
-		if err != nil {
+		switch {
+		case errors.Is(err, fs.ErrNotExist):
+			f.log.Info("Directory exists but does not contain a metadata file - deleting directory and its contents", "volume_id", dir.Name())
+			if err := f.RemoveVolume(dir.Name()); err != nil {
+				return nil, fmt.Errorf("deleting stale volume: %v", err)
+			}
+			// continue to skip this loop iteration
+			continue
+		case err != nil:
 			// discovered a volume/directory that does not contain a metadata file
-			// TODO: log this error to allow startup to continue
 			return nil, err
 		}
 		// immediately close the file as we just need to verify it exists

storage/filesystem_test.go

@@ -118,6 +118,25 @@ func TestFilesystem_ListVolumes(t *testing.T) {
 	}
 }
 
+func TestFilesystem_ListVolumes_CleansUpCorruptVolumes(t *testing.T) {
+	backend := &Filesystem{
+		fs: fstest.MapFS{
+			"inmemfs/fake-volume/metadata.json": &fstest.MapFile{Data: []byte{}},
+			"inmemfs/fake-emptyvolume/nothing":  &fstest.MapFile{Data: []byte{}},
+		},
+	}
+
+	vols, err := backend.ListVolumes()
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+	if len(vols) != 1 {
+		t.Errorf("expected 1 volume to be returned but got: %+v", vols)
+	}
+	if vols[0] != "fake-volume" {
+		t.Errorf("expected only entry to be 'fake-volume' but got: %s", vols[0])
+	}
+}
 func Test_fsGroupForMetadata(t *testing.T) {
 	intPtr := func(i int64) *int64 {
 		return &i