changefeedccl: fix race advancing frontier in schemafeed

In the schema feed, when in `updateTableHistory`, we check that the
current frontier is less than the current time. However, since we
release the mutex protecting frontier while validating table
descriptors, it's possible for another routine to advance the frontier
before the first routine tries to advance it. For example, another
routine may call pauseOrResumePolling and pause polling at the same time
it advances the frontier. As a consequence, it's possible for the first
routine to assert fail due to the current frontier being greater than
the current time when it tries to advance it.

This change fixes this race by checking that the frontier is greater
than the current time (endTS) again before trying to advance the
frontier. If the frontier is less than or equal to the current time, the
frontier does not need to be advanced and it returns. It's worth keeping
the original check in, since it avoids the need to validate descriptors,
and releasing the mutex also prevents the routine from holding it while
validating.

Epic: none
Fixes: #148963

Release note (bug fix): Fixes a race condition when advancing a
changefeed aggregator's frontier. When hit, the race condition could
result in an internal error that would shut down the kvfeed and cause
the changefeed to retry.

Author: rharding6373

pkg/ccl/changefeedccl/schemafeed/schema_feed.go

@@ -633,6 +633,16 @@ func (tf *schemaFeed) ingestDescriptors(
 	return tf.adjustTimestamps(startTS, endTS, validateErr)
 }
 
+// frontierAdvanceCheckEnabled controls whether the changefeed will
+// attempt to advance the frontier depending on the relation between the
+// last recorded frontier and the current time.
+var frontierAdvanceCheckEnabled = settings.RegisterBoolSetting(
+	settings.ApplicationLevel,
+	"changefeed.frontier_advance_check.enabled",
+	"if true, attempts to advance the frontier only if the last recorded frontier is less than the current time",
+	true,
+)
+
 // adjustTimestamps adjusts the frontier or error timestamp appropriately.
 func (tf *schemaFeed) adjustTimestamps(startTS, endTS hlc.Timestamp, validateErr error) error {
 	tf.mu.Lock()
@@ -644,10 +654,16 @@ func (tf *schemaFeed) adjustTimestamps(startTS, endTS hlc.Timestamp, validateErr
 		}
 		return validateErr
 	}
-
-	if frontier := tf.mu.ts.frontier; frontier.Less(startTS) {
+	frontier := tf.mu.ts.frontier
+	if frontier.Less(startTS) {
 		return errors.Errorf(`gap between %s and %s`, frontier, startTS)
 	}
+	// If the current frontier is greater than the endTS,
+	// then we do not need to advance the frontier. In fact,
+	// trying to advance the frontier could result in an error.
+	if endTS.LessEq(frontier) && frontierAdvanceCheckEnabled.Get(&tf.settings.SV) {
+		return nil
+	}
 	return tf.mu.ts.advanceFrontier(endTS)
 }