Reduce lock contention in WAL

pkg/wal/repository.go

@@ -120,6 +120,16 @@ func (s *Repository) Open(ctx context.Context) error {
 				continue
 			}
 
+			// If the segment is only 8 bytes, that means only the segment magic header has been written and there
+			// is no data in the file.  We don't want to upload these to kusto so they can be removed.
+			if fi.Size() == 8 {
+				logger.Warnf("Removing empty segment: %s", path)
+				if err := os.Remove(path); err != nil {
+					logger.Warnf("Failed to remove empty segment: %s %s", path, err.Error())
+				}
+				continue
+			}
+
 			info := SegmentInfo{
 				Prefix:    prefix,
 				Ulid:      epoch,

pkg/wal/repository_test.go

@@ -83,6 +83,7 @@ func TestRepository_Remove(t *testing.T) {
 
 			// Add a closed segment for this WAL.
 			seg, err := wal.NewSegment(dir, "db_foo")
+			require.NoError(t, seg.Write(context.Background(), []byte("bar")))
 			require.NoError(t, err)
 			require.NoError(t, seg.Close())
 

pkg/wal/wal.go

@@ -247,17 +247,29 @@ func (w *WAL) rotate(ctx context.Context) {
 			// Rotate the segment once it's past the max segment size
 			if (w.opts.SegmentMaxSize != 0 && sz >= w.opts.SegmentMaxSize) ||
 				(w.opts.SegmentMaxAge.Seconds() != 0 && time.Since(seg.CreatedAt()).Seconds() > w.opts.SegmentMaxAge.Seconds()) {
+
 				toClose = seg
-				w.segment = nil
+				var err error
+				w.segment, err = NewSegment(w.opts.StorageDir, w.opts.Prefix)
+				if err != nil {
+					logger.Errorf("Failed to create new segment: %s", err.Error())
+					w.segment = nil
+				}
 			}
 			w.mu.Unlock()
 
 			if toClose != nil {
-				info, err := toClose.Info()
-				if err != nil {
-					logger.Errorf("Failed to get segment info: %s %s", toClose.Path(), err.Error())
+				// 8 bytes is the size of the segment magic header bytes.  If that is all we've written, we can just
+				// delete it so that we don't end up uploading empty segments to Kusto.
+				if sz > 8 {
+					info, err := toClose.Info()
+					if err != nil {
+						logger.Errorf("Failed to get segment info: %s %s", toClose.Path(), err.Error())
+					} else {
+						w.index.Add(info)
+					}
 				} else {
-					w.index.Add(info)
+					_ = os.Remove(toClose.Path())
 				}
 
 				if err := toClose.Close(); err != nil {