wal: deflake TestManagerFailover

The invocation of the new segmentClosed callback introduced in #5388 occurs
asynchronously with respect to the manager and the progression through logical
log numbers. This test was flaky in two ways: If the segmentClosed callback was
invoked /before/ the writerClosed callback for the same WAL, writerClosed would
append a second record for the same logical log containing the set of WALs
other than the one inserted by the segmentClosed callback. Conversely, if the
segmentClosed callback was sufficiently delayed relative to the closing of the
writer, it might not be invoked until after the test listed the set of obsolete
logs.

The first race is fixed by a refactoring of the segmentClosed and writerClosed
callbacks, adapting them to share the same logic for merging logs. The second
race is fixed through use of (datadriven.TestData).Retry to account for the
nondeterminism.

Fix #5401.

Author: jbowens

wal/failover_manager.go

@@ -440,6 +440,10 @@ type segmentWithSizeEtc struct {
 	synchronouslyClosed bool
 }
 
+func cmpSegmentWithSizeEtc(a, b segmentWithSizeEtc) int {
+	return cmp.Compare(a.segment.logNameIndex, b.segment.logNameIndex)
+}
+
 type failoverManager struct {
 	opts Options
 	// initialObsolete holds the set of DeletableLogs that formed the logs
@@ -655,41 +659,41 @@ func (wm *failoverManager) ElevateWriteStallThresholdForFailover() bool {
 	return wm.monitor.elevateWriteStallThresholdForFailover()
 }
 
+// writerClosed is called by the failoverWriter; see
+// failoverWriterOpts.writerClosed.
 func (wm *failoverManager) writerClosed(llse logicalLogWithSizesEtc) {
 	wm.monitor.noWriter()
 	wm.mu.Lock()
 	defer wm.mu.Unlock()
-	wm.mu.closedWALs = append(wm.mu.closedWALs, llse)
+	wm.recordClosedWALLocked(llse)
 	wm.mu.ww = nil
 }
 
 // segmentClosed is called by the failoverWriter; see
 // failoverWriterOpts.segmentClosed.
-func (wm *failoverManager) segmentClosed(num NumWAL, s segmentWithSizeEtc) {
+func (wm *failoverManager) segmentClosed(llse logicalLogWithSizesEtc) {
 	wm.mu.Lock()
 	defer wm.mu.Unlock()
+	wm.recordClosedWALLocked(llse)
+}
+
+func (wm *failoverManager) recordClosedWALLocked(llse logicalLogWithSizesEtc) {
 	// 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)
-		}
+	// find one, we merge the segments into the existing list.
+	i, found := slices.BinarySearchFunc(wm.mu.closedWALs, llse.num,
+		func(llse logicalLogWithSizesEtc, num NumWAL) int {
+			return cmp.Compare(llse.num, num)
+		})
+	if !found {
+		// 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, llse)
 		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},
-	})
+	wm.mu.closedWALs[i].segments = append(wm.mu.closedWALs[i].segments, llse.segments...)
+	slices.SortFunc(wm.mu.closedWALs[i].segments, cmpSegmentWithSizeEtc)
+	wm.mu.closedWALs[i].segments = slices.CompactFunc(wm.mu.closedWALs[i].segments,
+		func(a, b segmentWithSizeEtc) bool { return a.logNameIndex == b.logNameIndex })
 }
 
 // Stats implements Manager.

wal/failover_manager_test.go

@@ -427,27 +427,29 @@ func TestManagerFailover(t *testing.T) {
 				return b.String()
 
 			case "list-and-stats":
-				logs := fm.List()
-				stats := fm.Stats()
-				var b strings.Builder
-				if len(logs) > 0 {
-					fmt.Fprintf(&b, "logs:\n")
-					for _, f := range logs {
-						fmt.Fprintf(&b, "  %s\n", f.String())
+				return td.Retry(t, func() string {
+					logs := fm.List()
+					stats := fm.Stats()
+					var b strings.Builder
+					if len(logs) > 0 {
+						fmt.Fprintf(&b, "logs:\n")
+						for _, f := range logs {
+							fmt.Fprintf(&b, "  %s\n", f.String())
+						}
 					}
-				}
-				fmt.Fprintf(&b, "stats:\n")
-				fmt.Fprintf(&b, "  obsolete: count %d size %d\n", stats.ObsoleteFileCount, stats.ObsoleteFileSize)
-				fmt.Fprintf(&b, "  live: count %d size %d\n", stats.LiveFileCount, stats.LiveFileSize)
-				fmt.Fprintf(&b, "  failover: switches %d pri-dur %s sec-dur %s\n", stats.Failover.DirSwitchCount,
-					stats.Failover.PrimaryWriteDuration.String(), stats.Failover.SecondaryWriteDuration.String())
-				var latencyProto io_prometheus_client.Metric
-				stats.Failover.FailoverWriteAndSyncLatency.Write(&latencyProto)
-				latencySampleCount := *latencyProto.Histogram.SampleCount
-				if latencySampleCount > 0 {
-					fmt.Fprintf(&b, "  latency sample count: %d\n", latencySampleCount)
-				}
-				return b.String()
+					fmt.Fprintf(&b, "stats:\n")
+					fmt.Fprintf(&b, "  obsolete: count %d size %d\n", stats.ObsoleteFileCount, stats.ObsoleteFileSize)
+					fmt.Fprintf(&b, "  live: count %d size %d\n", stats.LiveFileCount, stats.LiveFileSize)
+					fmt.Fprintf(&b, "  failover: switches %d pri-dur %s sec-dur %s\n", stats.Failover.DirSwitchCount,
+						stats.Failover.PrimaryWriteDuration.String(), stats.Failover.SecondaryWriteDuration.String())
+					var latencyProto io_prometheus_client.Metric
+					stats.Failover.FailoverWriteAndSyncLatency.Write(&latencyProto)
+					latencySampleCount := *latencyProto.Histogram.SampleCount
+					if latencySampleCount > 0 {
+						fmt.Fprintf(&b, "  latency sample count: %d\n", latencySampleCount)
+					}
+					return b.String()
+				})
 
 			case "write-record":
 				var value string

wal/failover_writer.go

@@ -476,7 +476,7 @@ type failoverWriterOpts struct {
 	// 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)
+	segmentClosed func(logicalLogWithSizesEtc)
 
 	writerCreatedForTest chan<- struct{}
 
@@ -702,13 +702,18 @@ func (ww *failoverWriter) switchToNewDir(dir dirAndFileHandle) error {
 				// 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,
+				ww.opts.segmentClosed(logicalLogWithSizesEtc{
+					num: ww.opts.wn,
+					segments: []segmentWithSizeEtc{
+						{
+							segment: segment{
+								logNameIndex: LogNameIndex(writerIndex),
+								dir:          dir.Dir,
+							},
+							approxFileSize:      initialFileSize,
+							synchronouslyClosed: false,
+						},
 					},
-					approxFileSize:      initialFileSize,
-					synchronouslyClosed: false,
 				})
 			})
 		}

wal/failover_writer_test.go

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

wal/testdata/manager_failover

@@ -290,8 +290,7 @@ ok
 list-and-stats
 ----
 logs:
-  000001: {(sec,001)}
-  000001: {(pri,002)}
+  000001: {(sec,001), (pri,002)}
 stats:
   obsolete: count 0 size 0
   live: count 2 size 18