db: improve compaction test timeout observability

 - set up a 1 minute timeout for any datadriven compactions;
 - print db logs on test failures;
 - upgrade datadriven to include a recent change to periodically log
   when a single command takes a long time.

Informs #5277

Author: RaduBerinde

compaction_test.go

@@ -905,24 +905,34 @@ func runCompactionTest(
 	rng := rand.New(rand.NewPCG(0, seed))
 	t.Logf("seed: %d", seed)
 
+	var dbLog base.InMemLogger
+	defer func() {
+		if t.Failed() {
+			t.Logf("db log:\n%s\n", dbLog.String())
+		}
+	}()
 	var compactionLog bytes.Buffer
-	compactionLogEventListener := &EventListener{
-		CompactionEnd: func(info CompactionInfo) {
-			// Ensure determinism.
-			info.JobID = 1
-			info.Duration = time.Second
-			info.TotalDuration = time.Second
-			fmt.Fprintln(&compactionLog, info.String())
+	eventListener := TeeEventListener(
+		MakeLoggingEventListener(&dbLog),
+		EventListener{
+			CompactionEnd: func(info CompactionInfo) {
+				// Ensure determinism.
+				info.JobID = 1
+				info.Duration = time.Second
+				info.TotalDuration = time.Second
+				fmt.Fprintln(&compactionLog, info.String())
+			},
 		},
-	}
+	)
+
 	concurrencyLow, concurrencyHigh := 1, 1
 	mkOpts := func() *Options {
 		randVersion := FormatMajorVersion(int(minVersion) + rng.IntN(int(maxVersion)-int(minVersion)+1))
 		opts := &Options{
 			FS:                          vfs.NewMem(),
 			DebugCheck:                  DebugCheckLevels,
 			DisableAutomaticCompactions: true,
-			EventListener:               compactionLogEventListener,
+			EventListener:               &eventListener,
 			FormatMajorVersion:          randVersion,
 			Logger:                      testutils.Logger{T: t},
 			Comparer:                    cmp,
@@ -941,6 +951,7 @@ func runCompactionTest(
 			require.NoError(t, closeAllSnapshots(d))
 			require.NoError(t, d.Close())
 		}
+		dbLog.Reset()
 		compactionLog.Reset()
 		concurrencyLow, concurrencyHigh = 1, 1
 		var err error
@@ -1009,7 +1020,7 @@ func runCompactionTest(
 			return ""
 
 		case "compact":
-			if err := runCompactCmd(td, d); err != nil {
+			if err := runCompactCmd(t, td, d); err != nil {
 				return err.Error()
 			}
 			s := describeLSM(d, verbose)
@@ -1029,6 +1040,8 @@ func runCompactionTest(
 				require.NoError(t, d.Close())
 			}
 
+			dbLog.Reset()
+			compactionLog.Reset()
 			var err error
 			if d, err = runDBDefineCmd(td, mkOpts()); err != nil {
 				return err.Error()
@@ -1212,7 +1225,7 @@ func runCompactionTest(
 			var s string
 			ch := make(chan error, 1)
 			go func() {
-				if err := runCompactCmd(td, d); err != nil {
+				if err := runCompactCmd(t, td, d); err != nil {
 					ch <- err
 					close(ch)
 					return
@@ -1276,12 +1289,10 @@ func runCompactionTest(
 			td.ScanArgs(t, "num-blocked", &numBlocked)
 			var cancelFunc atomic.Pointer[context.CancelFunc]
 			go func() {
-				compactFunc, cf, err := runCompactCmdAsync(td, d, true)
-				if err == nil {
-					cancelFunc.Store(&cf)
-					err = compactFunc()
-				}
-				if err != nil {
+				ctx, cf := context.WithCancel(context.Background())
+				compactFunc := runCompactCmdFn(ctx, t, td, d)
+				cancelFunc.Store(&cf)
+				if err := compactFunc(); err != nil {
 					ch <- err
 					close(ch)
 					return
@@ -1769,7 +1780,7 @@ func TestCompactionDeleteOnlyHints(t *testing.T) {
 				return buf.String()
 
 			case "compact":
-				if err := runCompactCmd(td, d); err != nil {
+				if err := runCompactCmd(t, td, d); err != nil {
 					return err.Error()
 				}
 				d.mu.Lock()

data_test.go

@@ -760,53 +760,50 @@ func runBuildSSTCmd(
 	return *metadata, nil
 }
 
-func runCompactCmdAsync(
-	td *datadriven.TestData, d *DB, cancellable bool,
-) (compactFunc func() error, cancelFunc context.CancelFunc, err error) {
+// runCompactCmdFn reurns a function that runs the compaction in the given
+// command. The context can be cancellable.
+func runCompactCmdFn(
+	ctx context.Context, t *testing.T, td *datadriven.TestData, d *DB,
+) (compactFunc func() error) {
 	if len(td.CmdArgs) == 0 {
-		return nil, nil, errors.Errorf("%s expects at least one argument", td.Cmd)
+		td.Fatalf(t, "%s expects at least one argument", td.Cmd)
 	}
 	parts := strings.Split(td.CmdArgs[0].Key, "-")
 	if len(parts) != 2 {
-		return nil, nil, errors.Errorf("expected <begin>-<end>: %s", td.Input)
+		td.Fatalf(t, "expected <begin>-<end>: %s", td.Input)
 	}
 	parallelize := td.HasArg("parallel")
-	ctx := context.Background()
-	if cancellable {
-		ctx, cancelFunc = context.WithCancel(ctx)
-	}
 	if len(td.CmdArgs) >= 2 && strings.HasPrefix(td.CmdArgs[1].Key, "L") {
 		levelString := td.CmdArgs[1].String()
 		iStart := base.MakeInternalKey([]byte(parts[0]), base.SeqNumMax, InternalKeyKindMax)
 		iEnd := base.MakeInternalKey([]byte(parts[1]), 0, 0)
 		if levelString[0] != 'L' {
-			if cancelFunc != nil {
-				cancelFunc()
-			}
-			return nil, nil, errors.Errorf("expected L<n>: %s", levelString)
+			td.Fatalf(t, "expected L<n>: %s", levelString)
 		}
 		level, err := strconv.Atoi(levelString[1:])
 		if err != nil {
-			if cancelFunc != nil {
-				cancelFunc()
-			}
-			return nil, nil, err
+			td.Fatalf(t, "cannot parse level number: %v", err)
 		}
 		return func() error {
 			return d.manualCompact(ctx, iStart.UserKey, iEnd.UserKey, level, parallelize)
-		}, cancelFunc, nil
+		}
 	}
 	return func() error {
 		return d.Compact(ctx, []byte(parts[0]), []byte(parts[1]), parallelize)
-	}, cancelFunc, nil
+	}
 }
 
-func runCompactCmd(td *datadriven.TestData, d *DB) error {
-	compactFunc, _, err := runCompactCmdAsync(td, d, false)
-	if err != nil {
-		return err
-	}
-	return compactFunc()
+func runCompactCmd(t *testing.T, td *datadriven.TestData, d *DB) error {
+	ctx := context.Background()
+	// Set up a 1-minute deadline to detect stuck compactions.
+	ctx, cancelFn := context.WithDeadline(ctx, time.Now().Add(time.Minute))
+	defer cancelFn()
+	compactFunc := runCompactCmdFn(ctx, t, td, d)
+	err := compactFunc()
+	if ctx.Err() != nil {
+		td.Fatalf(t, "compaction took too long")
+	}
+	return err
 }
 
 // runDBDefineCmd prepares a database state, returning the opened

disk_usage_test.go

@@ -103,7 +103,7 @@ func TestEstimateDiskUsageDataDriven(t *testing.T) {
 			}
 			return ""
 		case "compact":
-			if err := runCompactCmd(td, d); err != nil {
+			if err := runCompactCmd(t, td, d); err != nil {
 				return err.Error()
 			}
 			return runLSMCmd(td, d)

excise_test.go

@@ -756,15 +756,15 @@ func TestConcurrentExcise(t *testing.T) {
 			tdClone = *td
 			tdClone.CmdArgs = otherArgs
 			if !async {
-				err := runCompactCmd(td, d)
+				err := runCompactCmd(t, td, d)
 				if err != nil {
 					return err.Error()
 				}
 			} else {
 				wg.Add(1)
 				go func() {
 					defer wg.Done()
-					_ = runCompactCmd(&tdClone, d)
+					_ = runCompactCmd(t, &tdClone, d)
 				}()
 				<-bc.startBlock
 				return "spun off in separate goroutine"

ingest_test.go

@@ -988,7 +988,7 @@ func testIngestSharedImpl(
 			return "ok"
 
 		case "compact":
-			err := runCompactCmd(td, d)
+			err := runCompactCmd(t, td, d)
 			if err != nil {
 				return err.Error()
 			}

iterator_histories_test.go

@@ -185,7 +185,7 @@ func TestIterHistories(t *testing.T) {
 				}
 				return fmt.Sprintf("wrote %d keys to batch %q\n", count, name)
 			case "compact":
-				if err := runCompactCmd(td, d); err != nil {
+				if err := runCompactCmd(t, td, d); err != nil {
 					return err.Error()
 				}
 				return runLSMCmd(td, d)

metrics_test.go

@@ -329,7 +329,7 @@ func TestMetrics(t *testing.T) {
 			return ""
 
 		case "compact":
-			if err := runCompactCmd(td, d); err != nil {
+			if err := runCompactCmd(t, td, d); err != nil {
 				return err.Error()
 			}
 

range_del_test.go

@@ -68,7 +68,7 @@ func TestRangeDel(t *testing.T) {
 			return runWaitForTableStatsCmd(td, d)
 
 		case "compact":
-			if err := runCompactCmd(td, d); err != nil {
+			if err := runCompactCmd(t, td, d); err != nil {
 				return err.Error()
 			}
 			d.mu.Lock()

scan_internal_test.go

@@ -126,7 +126,7 @@ func TestScanStatistics(t *testing.T) {
 			}
 			return fmt.Sprintf("wrote %d keys to batch %q\n", count, name)
 		case "compact":
-			if err := runCompactCmd(td, d); err != nil {
+			if err := runCompactCmd(t, td, d); err != nil {
 				return err.Error()
 			}
 			return runLSMCmd(td, d)
@@ -491,7 +491,7 @@ func TestScanInternal(t *testing.T) {
 			}
 			return fmt.Sprintf("wrote %d keys to batch %q\n", count, name)
 		case "compact":
-			if err := runCompactCmd(td, d); err != nil {
+			if err := runCompactCmd(t, td, d); err != nil {
 				return err.Error()
 			}
 			return runLSMCmd(td, d)

table_stats_test.go

@@ -175,7 +175,7 @@ func TestTableStats(t *testing.T) {
 			return s
 
 		case "compact":
-			if err := runCompactCmd(td, d); err != nil {
+			if err := runCompactCmd(t, td, d); err != nil {
 				return err.Error()
 			}
 			d.mu.Lock()