tests/robustness: enhance compact failpoint

If the cluster serves requests slowly, the database has few revision
number and then Compact won't trigger BatchCommit. Add a loop to check
the last revision is big enough to trigger panic.

Signed-off-by: Wei Fu <fuweid89@gmail.com>

tests/robustness/failpoints.go

@@ -103,6 +103,18 @@ func validateFailpoint(clus *e2e.EtcdProcessCluster, failpoint Failpoint) error
 	return nil
 }
 
+func addMinimalRevForCompactFailpoint(failpoint Failpoint, batchNum int) Failpoint {
+	switch failpoint.Name() {
+	case "compactAfterCommitBatch", "compactBeforeCommitBatch":
+		if batchNum == 0 {
+			batchNum = 1000 // align with ../../server/storage/mvcc/kvstore.go
+		}
+		return goPanicFailpoint{failpoint.Name(), triggerCompact{minimalRev: int64(batchNum)}, AnyMember}
+	default:
+		return failpoint
+	}
+}
+
 func injectFailpoints(ctx context.Context, t *testing.T, lg *zap.Logger, clus *e2e.EtcdProcessCluster, failpoint Failpoint) {
 	ctx, cancel := context.WithTimeout(ctx, triggerTimeout)
 	defer cancel()
@@ -338,7 +350,9 @@ func (t triggerDefrag) Available(e2e.EtcdProcessClusterConfig, e2e.EtcdProcess)
 	return true
 }
 
-type triggerCompact struct{}
+type triggerCompact struct {
+	minimalRev int64 // zero means no requirement.
+}
 
 func (t triggerCompact) Trigger(_ *testing.T, ctx context.Context, member e2e.EtcdProcess, _ *e2e.EtcdProcessCluster) error {
 	cc, err := clientv3.New(clientv3.Config{
@@ -351,11 +365,23 @@ func (t triggerCompact) Trigger(_ *testing.T, ctx context.Context, member e2e.Et
 		return fmt.Errorf("failed creating client: %w", err)
 	}
 	defer cc.Close()
-	resp, err := cc.Get(ctx, "/")
-	if err != nil {
-		return err
+
+	var rev int64
+	for {
+		resp, err := cc.Get(ctx, "/")
+		if err != nil {
+			return err
+		}
+
+		rev = resp.Header.Revision
+		if t.minimalRev > 0 && rev <= t.minimalRev {
+			time.Sleep(50 * time.Millisecond)
+			continue
+		}
+		break
 	}
-	_, err = cc.Compact(ctx, resp.Header.Revision)
+
+	_, err = cc.Compact(ctx, rev)
 	if err != nil && !strings.Contains(err.Error(), "error reading from server: EOF") {
 		return err
 	}

tests/robustness/linearizability_test.go

@@ -157,6 +157,8 @@ func testRobustness(ctx context.Context, t *testing.T, lg *zap.Logger, s testSce
 		}
 	}
 
+	s.failpoint = addMinimalRevForCompactFailpoint(s.failpoint, report.Cluster.Cfg.CompactionBatchLimit)
+
 	// t.Failed() returns false during panicking. We need to forcibly
 	// save data on panicking.
 	// Refer to: https://github.com/golang/go/issues/49929