wal: clean up all segment files

When WAL failover is configured, a single logical WAL may be composed of
multiple physical segment files. The creation of these physical segment files
occurs asynchronously. This asynchronous creation may race with the closing of
the failover writer.

Specifically, an outstanding attempt to create a segment file may block.
Meanwhile, writes persisting all the necessary records may complete on the
other device. If the logical WAL is now finished, the failover writer may be
closed by higher-level code while the outstanding attempt to create a file
remains. The system progresses, accumulating a list of obsolete files based on
the set that existed at the time that the writer was closed. Eventually, the
outstanding file creation may complete, creating a new file.

Previously this race resulted in leaking the straggling file. Because we
preallocate disk space for WAL files, this logically empty file could still
consume significant disk space. This leaked file could not be discovered and
deleted until process restart.

This commit adjusts the FailoverWriter to invoke a callback on the
FailoverManager, propagating information about these abandoned segment files.
This allows the FailoverManager to propagate these obsolete files to higher
levels for deletion.

Fix #5355.

Author: jbowens

wal/failover_manager.go

@@ -5,6 +5,7 @@
 package wal
 
 import (
+	"cmp"
 	"fmt"
 	"io"
 	"math/rand/v2"
@@ -537,26 +538,22 @@ func (wm *failoverManager) init(o Options, initial Logs) error {
 func (wm *failoverManager) List() Logs {
 	wm.mu.Lock()
 	defer wm.mu.Unlock()
-	n := len(wm.mu.closedWALs)
-	if wm.mu.ww != nil {
-		n++
-	}
-	wals := make(Logs, n)
-	setLogicalLog := func(index int, llse logicalLogWithSizesEtc) {
+	wals := make(Logs, 0, len(wm.mu.closedWALs)+1)
+	setLogicalLog := func(llse logicalLogWithSizesEtc) {
 		segments := make([]segment, len(llse.segments))
 		for j := range llse.segments {
 			segments[j] = llse.segments[j].segment
 		}
-		wals[index] = LogicalLog{
+		wals = append(wals, LogicalLog{
 			Num:      llse.num,
 			segments: segments,
-		}
+		})
 	}
-	for i, llse := range wm.mu.closedWALs {
-		setLogicalLog(i, llse)
+	for _, llse := range wm.mu.closedWALs {
+		setLogicalLog(llse)
 	}
 	if wm.mu.ww != nil {
-		setLogicalLog(n-1, wm.mu.ww.getLog())
+		setLogicalLog(wm.mu.ww.getLog())
 	}
 	return wals
 }
@@ -637,6 +634,7 @@ func (wm *failoverManager) Create(wn NumWAL, jobID int) (Writer, error) {
 		stopper:                     wm.stopper,
 		failoverWriteAndSyncLatency: wm.opts.FailoverWriteAndSyncLatency,
 		writerClosed:                wm.writerClosed,
+		segmentClosed:               wm.segmentClosed,
 		writerCreatedForTest:        wm.opts.logWriterCreatedForTesting,
 		writeWALSyncOffsets:         wm.opts.WriteWALSyncOffsets,
 	}
@@ -671,6 +669,35 @@ func (wm *failoverManager) writerClosed(llse logicalLogWithSizesEtc) {
 	wm.mu.ww = nil
 }
 
+// segmentClosed is called by the failoverWriter; see
+// failoverWriterOpts.segmentClosed.
+func (wm *failoverManager) segmentClosed(num NumWAL, s segmentWithSizeEtc) {
+	wm.mu.Lock()
+	defer wm.mu.Unlock()
+	// Find the closed WAL matching the logical WAL num, if one exists. If we
+	// find one, we append the segment to the list of segments if it's not
+	// already there.
+	i, found := slices.BinarySearchFunc(wm.mu.closedWALs, num, func(llse logicalLogWithSizesEtc, num NumWAL) int {
+		return cmp.Compare(llse.num, num)
+	})
+	if found {
+		segmentIndex, segmentFound := slices.BinarySearchFunc(wm.mu.closedWALs[i].segments, s.segment.logNameIndex,
+			func(s segmentWithSizeEtc, logNameIndex LogNameIndex) int {
+				return cmp.Compare(s.segment.logNameIndex, logNameIndex)
+			})
+		if !segmentFound {
+			wm.mu.closedWALs[i].segments = slices.Insert(wm.mu.closedWALs[i].segments, segmentIndex, s)
+		}
+		return
+	}
+	// If we didn't find an existing entry in closedWALs for the provided
+	// NumWAL, append a new entry.
+	wm.mu.closedWALs = slices.Insert(wm.mu.closedWALs, i, logicalLogWithSizesEtc{
+		num:      num,
+		segments: []segmentWithSizeEtc{s},
+	})
+}
+
 // Stats implements Manager.
 func (wm *failoverManager) Stats() Stats {
 	obsoleteLogsCount, obsoleteLogSize := wm.recycler.Stats()
@@ -889,8 +916,3 @@ func (t *defaultTicker) stop() {
 func (t *defaultTicker) ch() <-chan time.Time {
 	return (*time.Ticker)(t).C
 }
-
-// Make lint happy.
-var _ = (*failoverMonitor).noWriter
-var _ = (*failoverManager).writerClosed
-var _ = (&stopper{}).shouldQuiesce

wal/failover_manager_test.go

@@ -5,8 +5,10 @@
 package wal
 
 import (
+	"bytes"
 	"container/list"
 	"fmt"
+	"math/rand/v2"
 	"os"
 	"slices"
 	"strings"
@@ -15,6 +17,7 @@ import (
 	"time"
 
 	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/pebble/internal/testutils"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/cockroachdb/pebble/vfs/errorfs"
 	"github.com/prometheus/client_golang/prometheus"
@@ -617,3 +620,111 @@ func TestFailoverManager_SecondaryIsWritable(t *testing.T) {
 // are complete. Currently this is done by waiting on various channels etc.
 // which exposes implementation detail. See concurrency_test.monitor, in
 // CockroachDB, for an alternative.
+
+// TestFailoverManager_AllFilesDeletable is a randomized test that validates
+// that all the files that are created by the manager are eventually returned by
+// FailoverManager.Obsolete for deletion.
+func TestFailoverManager_AllFilesDeletable(t *testing.T) {
+	seed := time.Now().UnixNano()
+	memFS := vfs.NewMem()
+	require.NoError(t, memFS.MkdirAll("primary", os.ModePerm))
+	require.NoError(t, memFS.MkdirAll("secondary", os.ModePerm))
+	pcg := rand.NewPCG(0, uint64(seed))
+	rng := rand.New(pcg)
+	latencySeed := rng.Int64()
+	fs := errorfs.Wrap(memFS, errorfs.RandomLatency(
+		errorfs.Randomly(0.50, latencySeed), 10*time.Millisecond, latencySeed, 0 /* no limit */))
+
+	var m failoverManager
+	require.NoError(t, m.init(Options{
+		Primary:              Dir{FS: fs, Dirname: "primary"},
+		Secondary:            Dir{FS: fs, Dirname: "secondary"},
+		Logger:               testutils.Logger{T: t},
+		MaxNumRecyclableLogs: 0,
+		PreallocateSize:      func() int { return 4 },
+		FailoverOptions: FailoverOptions{
+			PrimaryDirProbeInterval:            250 * time.Microsecond,
+			HealthyProbeLatencyThreshold:       time.Millisecond,
+			HealthyInterval:                    3 * time.Millisecond,
+			UnhealthySamplingInterval:          250 * time.Microsecond,
+			UnhealthyOperationLatencyThreshold: func() (time.Duration, bool) { return time.Millisecond, true },
+		},
+		FailoverWriteAndSyncLatency: prometheus.NewHistogram(prometheus.HistogramOpts{}),
+		WriteWALSyncOffsets:         func() bool { return false },
+	}, nil /* initial  logs */))
+
+	// Repeatedly create a new log file, write some random data to it, and then
+	// maybe ratchet the minUnflushed log number, collecting the set of obsolete
+	// log files if we do.
+	const numIters = 20
+	minUnflushed := int64(0)
+	var allDeletableLogs []DeletableLog
+	for i := 1; i <= numIters; i++ {
+		// Create a new log file and write random data to it.
+		w, err := m.Create(NumWAL(i), i)
+		require.NoError(t, err)
+		for j := 0; j < testutils.RandIntInRange(rng, 1, 4); j++ {
+			data := testutils.RandBytes(rng, testutils.RandIntInRange(rng, 1, 1024))
+			_, err = w.WriteRecord(data, SyncOptions{}, nil)
+			require.NoError(t, err)
+		}
+		_, err = w.Close()
+		require.NoError(t, err)
+
+		// Ratchet the minUnflushed log number randomly, and call Obsolete to
+		// collect the set of deletable logs.
+		minUnflushed = rng.Int64N(int64(i)-minUnflushed+1) + minUnflushed
+		toDelete, err := m.Obsolete(NumWAL(minUnflushed), false)
+		require.NoError(t, err)
+		allDeletableLogs = append(allDeletableLogs, toDelete...)
+	}
+
+	var buf bytes.Buffer
+	defer func() {
+		if t.Failed() {
+			t.Log(buf.String())
+		}
+	}()
+
+	require.Eventually(t, func() bool {
+		toDelete, err := m.Obsolete(NumWAL(minUnflushed), false)
+		require.NoError(t, err)
+		allDeletableLogs = append(allDeletableLogs, toDelete...)
+		// Delete any of the obolsete log files that we collected.
+		for _, dl := range allDeletableLogs {
+			require.NoError(t, dl.FS.Remove(dl.Path))
+		}
+		allDeletableLogs = allDeletableLogs[:0]
+
+		// Find all the remaining log files in both the primary and secondary
+		// directories.
+		var fa FileAccumulator
+		ls, err := memFS.List(m.opts.Primary.Dirname)
+		require.NoError(t, err)
+		secondaryLS, err := memFS.List(m.opts.Secondary.Dirname)
+		require.NoError(t, err)
+		for _, f := range ls {
+			_, err := fa.MaybeAccumulate(memFS, memFS.PathJoin(m.opts.Primary.Dirname, f))
+			require.NoError(t, err)
+		}
+		for _, f := range secondaryLS {
+			_, err := fa.MaybeAccumulate(memFS, memFS.PathJoin(m.opts.Secondary.Dirname, f))
+			require.NoError(t, err)
+		}
+		logs := fa.Finish()
+
+		// Remove any logs that are above the minUnflushed log. These are to be
+		// expected.
+		logs = slices.DeleteFunc(logs, func(log LogicalLog) bool {
+			return log.Num >= NumWAL(minUnflushed)
+		})
+		if len(logs) > 0 {
+			buf.Reset()
+			fmt.Fprintf(&buf, "logs with numbers beneath %d remain on the filesystem:\n", minUnflushed)
+			for _, log := range logs {
+				fmt.Fprintf(&buf, "%s remains\n", log)
+			}
+		}
+		return len(logs) == 0
+	}, time.Second, 25*time.Millisecond)
+}

wal/failover_writer.go

@@ -460,7 +460,23 @@ type failoverWriterOpts struct {
 	stopper         *stopper
 
 	failoverWriteAndSyncLatency prometheus.Histogram
-	writerClosed                func(logicalLogWithSizesEtc)
+	// writerClosed is a callback invoked by the FailoverWriter when it's
+	// closed. It notifies the FailoverManager that the writer is now closed and
+	// propagates information about the various physical segment files that have
+	// been created.
+	//
+	// Note that the asynchronous creation of physical segment files means that
+	// the writerClosed invocation is not guaranteed to include all physical
+	// segment files that will ultimately be created for this logical WAL. If a
+	// new segment file is created after writerClosed is inovked, it will be
+	// propagated to the FailoverManager via the segmentClosed callback.
+	writerClosed func(logicalLogWithSizesEtc)
+	// segmentClosed is a callback invoked by the FailoverWriter when a segment
+	// file creation completes but the writerClosed callback has already been
+	// invoked. It's used to ensure that we reclaim all physical segment files,
+	// including ones that did not complete creation before the Writer was
+	// closed.
+	segmentClosed func(NumWAL, segmentWithSizeEtc)
 
 	writerCreatedForTest chan<- struct{}
 
@@ -684,14 +700,18 @@ func (ww *failoverWriter) switchToNewDir(dir dirAndFileHandle) error {
 			// returned error.
 			ww.opts.stopper.runAsync(func() {
 				_ = w.Close()
-				// TODO(sumeer): consider deleting this file too, since
-				// failoverWriter.Close may not wait for it. This is going to be
-				// extremely rare, so the risk of garbage empty files piling up is
-				// extremely low. Say failover happens daily and of those cases we
-				// have to be very unlucky and the close happens while a failover was
-				// ongoing and the previous LogWriter successfully wrote everything
-				// (say 1% probability if we want to be pessimistic). A garbage file
-				// every 100 days. Restarts will delete that garbage.
+				// Invoke the segmentClosed callback to propagate knowledge that
+				// there's an obsolete segment file we should clean up. Note
+				// that the file may be occupying non-negligible disk space even
+				// though we never wrote to it due to preallocation.
+				ww.opts.segmentClosed(ww.opts.wn, segmentWithSizeEtc{
+					segment: segment{
+						logNameIndex: LogNameIndex(writerIndex),
+						dir:          dir.Dir,
+					},
+					approxFileSize:      initialFileSize,
+					synchronouslyClosed: false,
+				})
 			})
 		}
 	})

wal/failover_writer_test.go

@@ -285,6 +285,7 @@ func TestFailoverWriter(t *testing.T) {
 						stopper:                     stopper,
 						failoverWriteAndSyncLatency: prometheus.NewHistogram(prometheus.HistogramOpts{}),
 						writerClosed:                func(_ logicalLogWithSizesEtc) {},
+						segmentClosed:               func(_ NumWAL, _ segmentWithSizeEtc) {},
 						writerCreatedForTest:        logWriterCreated,
 						writeWALSyncOffsets:         func() bool { return false },
 					}, testDirs[dirIndex])
@@ -650,6 +651,7 @@ func TestConcurrentWritersWithManyRecords(t *testing.T) {
 		stopper:                     stopper,
 		failoverWriteAndSyncLatency: prometheus.NewHistogram(prometheus.HistogramOpts{}),
 		writerClosed:                func(_ logicalLogWithSizesEtc) {},
+		segmentClosed:               func(_ NumWAL, _ segmentWithSizeEtc) {},
 		writerCreatedForTest:        logWriterCreated,
 		writeWALSyncOffsets:         func() bool { return false },
 	}, dirs[dirIndex])
@@ -753,6 +755,7 @@ func TestFailoverWriterManyRecords(t *testing.T) {
 		stopper:                     stopper,
 		failoverWriteAndSyncLatency: prometheus.NewHistogram(prometheus.HistogramOpts{}),
 		writerClosed:                func(_ logicalLogWithSizesEtc) {},
+		segmentClosed:               func(_ NumWAL, _ segmentWithSizeEtc) {},
 		writeWALSyncOffsets:         func() bool { return false },
 	}, dir)
 	require.NoError(t, err)

wal/testdata/manager_failover

@@ -290,10 +290,11 @@ ok
 list-and-stats
 ----
 logs:
+  000001: {(sec,001)}
   000001: {(pri,002)}
 stats:
   obsolete: count 0 size 0
-  live: count 1 size 18
+  live: count 2 size 18
   failover: switches 2 pri-dur 77ms sec-dur 80ms
 
 advance-time dur=1s
@@ -318,6 +319,7 @@ obsolete min-unflushed=2
 ok
 recycler empty
 to delete:
+  wal 1: path: sec/000001-001.log size: 0
   wal 1: path: pri/000001-002.log size: 18
 
 create-writer wal-num=2