crosscluster/physical: persist standby poller progress

This patch sets the standby poller job's resolved time to the system time that
standby descriptors have been updated to. This allows a reader tenant user to
easily check that the poller job is running smoothly via SHOW JOB.

Epic: none

Release note: none

Author: msbutler

pkg/crosscluster/physical/BUILD.bazel

@@ -196,6 +196,7 @@ go_test(
         "//pkg/util/span",
         "//pkg/util/syncutil",
         "//pkg/util/timeutil",
+        "@com_github_cockroachdb_apd_v3//:apd",
         "@com_github_cockroachdb_datadriven//:datadriven",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_stretchr_testify//require",

pkg/crosscluster/physical/standby_read_ts_poller_job.go

@@ -7,6 +7,7 @@ package physical
 
 import (
 	"context"
+	"math"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/jobs"
@@ -15,6 +16,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/replication"
+	"github.com/cockroachdb/cockroach/pkg/sql/isql"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/errors"
@@ -145,14 +147,20 @@ func (r *standbyReadTSPollerResumer) poll(ctx context.Context, execCfg *sql.Exec
 					replicatedTime)
 			}
 			previousReplicatedTimestamp = replicatedTime
-			if err = replication.SetupOrAdvanceStandbyReaderCatalog(
+			if err := replication.SetupOrAdvanceStandbyReaderCatalog(
 				ctx,
 				tenantID,
 				replicatedTime,
 				execCfg.InternalDB,
 				execCfg.Settings,
 			); err != nil {
 				log.Warningf(ctx, "failed to advance replicated timestamp for reader tenant {%d}: %v", tenantID, err)
+			} else {
+				if err := execCfg.InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+					return r.job.ProgressStorage().Set(ctx, txn, math.NaN(), replicatedTime)
+				}); err != nil {
+					log.Warningf(ctx, "failed to set standby poller job read time %v", err)
+				}
 			}
 		}
 	}

pkg/crosscluster/physical/standby_read_ts_poller_job_test.go

@@ -12,6 +12,7 @@ import (
 	"testing"
 	"time"
 
+	apd "github.com/cockroachdb/apd/v3"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster/replicationtestutils"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
@@ -66,7 +67,10 @@ INSERT INTO a VALUES (1);
 
 	c.SrcTenantSQL.Exec(t, defaultDBQuery)
 	waitForPollerJobToStartDest(t, c, ingestionJobID)
+	pollerResolvedTime := waitForPollerTimeToAdvance(t, c.ReaderTenantSQL, apd.New(0, 0))
+
 	observeValueInReaderTenant(t, c.ReaderTenantSQL)
+	waitForPollerTimeToAdvance(t, c.ReaderTenantSQL, pollerResolvedTime)
 
 	// Failback and setup stanby reader tenant on the og source.
 	{
@@ -164,6 +168,22 @@ WHERE job_type = 'STANDBY READ TS POLLER'
 	jobutils.WaitForJobToRun(t, readerSQL, jobID)
 }
 
+func waitForPollerTimeToAdvance(
+	t *testing.T, readerSQL *sqlutils.SQLRunner, prevTime *apd.Decimal,
+) *apd.Decimal {
+	var resolvedTime apd.Decimal
+	testutils.SucceedsSoon(t, func() error {
+		readerSQL.QueryRow(t, `SELECT COALESCE(high_water_timestamp, '0')
+		FROM crdb_internal.jobs 
+		WHERE job_type = 'STANDBY READ TS POLLER'`).Scan(&resolvedTime)
+		if resolvedTime.Cmp(prevTime) <= 0 {
+			return errors.Errorf("resolved time has not advanced past %d", prevTime)
+		}
+		return nil
+	})
+	return &resolvedTime
+}
+
 func TestReaderTenantCutover(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)