fsck: detect problems with _format file, and remove stream if empty

A bulkdata dir may get created for a new stream with an empty or
corrupted _format, before any data gets actually written.  In that
case, we can just delete the new stream; worst case, we lose some
metadata.

Note: The info in _format should really get moved into the database.
This was born when bulkdata switched from PyTables to a custom storage
system, and was probably stored this way to avoid tying the main DB
to specific implementation details while they were still in flux.

nilmdb/fsck/fsck.py

@@ -39,6 +39,10 @@ class RetryFsck(FsckError):
     pass
 
 
+class FsckFormatError(FsckError):
+    pass
+
+
 def log(format, *args):
     printf(format, *args)
 
@@ -238,16 +242,29 @@ class Fsck(object):
                                             "set: %s\nnew: %s",
                                             path, str(posiset), str(new))
 
-                # check bulkdata
-                self.check_bulkdata(sid, path, bulk)
-
-                # Check that we can open bulkdata
                 try:
+                    # Check bulkdata
+                    self.check_bulkdata(sid, path, bulk)
+
+                    # Check that we can open bulkdata
                     tab = nilmdb.server.bulkdata.Table(bulk)
-                except Exception as e:  # pragma: no cover --
-                    # No coverage here because, in the current code,
-                    # everything that would cause the bulkdata to fail
-                    # has been already checked.
+                except FsckFormatError:
+                    # If there are no files except _format, try deleting
+                    # the entire stream; this may remove metadata, but
+                    # it's probably unimportant.
+                    files = list(os.listdir(bulk))
+                    if len(files) > 1:
+                        raise
+                    if len(files) == 1 and files[0] != b'_format':
+                        raise
+
+                    # Since the stream was empty, just remove it
+                    self.fix_remove_stream(sid, path, bulk,
+                                           "empty, with corrupted format file")
+
+                except FsckError:
+                    raise
+                except Exception as e:
                     raise FsckError("%s: can't open bulkdata: %s",
                                     path, str(e))
                 tab.close()
@@ -256,21 +273,24 @@ class Fsck(object):
 
     @retry_if_raised(RetryFsck)
     def check_bulkdata(self, sid, path, bulk):
-        with open(os.path.join(bulk, b"_format"), "rb") as f:
-            fmt = pickle.load(f)
+        try:
+            with open(os.path.join(bulk, b"_format"), "rb") as f:
+                fmt = pickle.load(f)
+        except Exception as e:
+            raise FsckFormatError(f"{path}: can't load _format file ({e})")
         if fmt["version"] != 3:
-            raise FsckError("%s: bad or unsupported bulkdata version %d",
-                            path, fmt["version"])
+            raise FsckFormatError("%s: bad or unsupported bulkdata version %d",
+                                  path, fmt["version"])
         rows_per_file = int(fmt["rows_per_file"])
         if rows_per_file < 1:
-            raise FsckError(f"{path}: bad rows_per_file {rows_per_file}")
+            raise FsckFormatError(f"{path}: bad rows_per_file {rows_per_file}")
         files_per_dir = int(fmt["files_per_dir"])
         if files_per_dir < 1:
-            raise FsckError(f"{path}: bad files_per_dir {files_per_dir}")
+            raise FsckFormatError(f"{path}: bad files_per_dir {files_per_dir}")
         layout = fmt["layout"]
         if layout != self.stream_layout[sid]:
-            raise FsckError("%s: layout mismatch %s != %s", path,
-                            layout, self.stream_layout[sid])
+            raise FsckFormatError("%s: layout mismatch %s != %s", path,
+                                  layout, self.stream_layout[sid])
 
         # Every file should have a size that's the multiple of the row size
         rkt = nilmdb.server.rocket.Rocket(layout, None)