lsm: hold read lock on iterate (#557)

* lsm: hold read lock on Iterate

Otherwise, snapshots will not hold a read lock over the index when iterating.

* *: handle 0-size object storage blocks

There are sometimes errors when serializing a block. This would cause us to
create a file but not delete it. This commit deletes an object storage file on
serialization failure.

Additionally, this commit ignores empty object storage files on scans.

index/lsm.go

@@ -193,6 +193,8 @@ func (l *LSM) Prefixes(ctx context.Context, prefix string) ([]string, error) {
 }
 
 func (l *LSM) Iterate(iter func(node *Node) bool) {
+	l.RLock()
+	defer l.RUnlock()
 	l.levels.Iterate(iter)
 }
 

store.go

@@ -48,7 +48,10 @@ func (t *TableBlock) Persist() error {
 		}
 
 		if err != nil {
-			return fmt.Errorf("failed to serialize block: %v", err)
+			if deleteErr := sink.Delete(context.Background(), fileName); deleteErr != nil {
+				err = fmt.Errorf("%v failed to delete file on error: %w", err, deleteErr)
+			}
+			return fmt.Errorf("failed to serialize block: %w", err)
 		}
 	}
 
@@ -212,6 +215,14 @@ func (b *DefaultObjstoreBucket) ProcessFile(ctx context.Context, blockDir string
 
 	span.SetAttributes(attribute.Int64("size", attribs.Size))
 
+	if attribs.Size == 0 {
+		level.Debug(b.logger).Log(
+			"msg", "ignoring empty block",
+			"blockTime", blockUlid.Time(),
+		)
+		return nil
+	}
+
 	file, err := b.openBlockFile(ctx, blockName, attribs.Size)
 	if err != nil {
 		return err