Merge #156003

156003: physical: fix blocked span reconciliation jobs after stream of system r=msbutler a=kev-cao

If PCR streams from a system tenant, then the span config reconciliation job on the destination tenant after cutover will repeatedly fail. This is due to the fact that the producer job on the source system tenant is also replicated to the destination tenant, along with its corresponding PTS. As the PTS targets specifically the system tenant, the span reconciliation job on the destination tenant is unable to validate the span configuration as a non-system tenant is attempting to apply a configuration on another tenant.

Fixes: #155444

Release note: Span config reconciliation jobs no longer fail on the destination after cutover from a PCR stream of a system tenant.

Co-authored-by: Kevin Cao <39608887+kev-cao@users.noreply.github.com>

Author: craig[bot]

pkg/crosscluster/physical/stream_ingestion_job_test.go

@@ -12,6 +12,7 @@ import (
 	"net"
 	"net/url"
 	"testing"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
 	_ "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl"
@@ -742,3 +743,69 @@ func TestPhysicalReplicationGatewayRoute(t *testing.T) {
 	progress := jobutils.GetJobProgress(t, systemDB, jobspb.JobID(jobID))
 	require.Empty(t, progress.Details.(*jobspb.Progress_StreamIngest).StreamIngest.PartitionConnUris)
 }
+
+func TestPhysicalReplicationCancelsProducerOnCutoverFromSystem(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	// This test verifies that the span reconciliation job on a promoted standby
+	// tenant that was streaming from a system tenant is able to make progress.
+	// See https://github.com/cockroachdb/cockroach/issues/155444 for more
+	// details.
+	ctx := context.Background()
+	args := replicationtestutils.DefaultTenantStreamingClustersArgs
+	args.MultitenantSingleClusterNumNodes = 1
+	args.SrcTenantID = roachpb.SystemTenantID
+	args.SrcTenantName = "system"
+
+	c, cleanup := replicationtestutils.CreateMultiTenantStreamingCluster(ctx, t, args)
+	defer cleanup()
+
+	producerJobID, consumerJobID := c.StartStreamReplication(ctx)
+	jobutils.WaitForJobToRun(c.T, c.SrcSysSQL, jobspb.JobID(producerJobID))
+	jobutils.WaitForJobToRun(c.T, c.DestSysSQL, jobspb.JobID(consumerJobID))
+
+	replicationtestutils.WaitUntilStartTimeReached(t, c.DestSysSQL, jobspb.JobID(consumerJobID))
+	srcTime := c.SrcCluster.Server(0).Clock().Now()
+	c.Cutover(ctx, producerJobID, consumerJobID, srcTime.GoTime(), false /* async */)
+	destCleanup := c.StartDestTenant(ctx, nil /* withTestingKnobs */, 0 /* server */)
+	defer destCleanup()
+
+	// We speed up the span config reconciliation job manager so that it will
+	// quickly detect the missing reconciliation job on the tenant (the previous
+	// one was canceled as it was replicated) and spin up a new one.
+	c.DestSysSQL.Exec(t, "ALTER TENANT ALL SET CLUSTER SETTING spanconfig.reconciliation_job.check_interval = '10ms'")
+
+	registry := c.DestTenantServer.JobRegistry().(*jobs.Registry)
+	// We nudge the adoption queue to pick up the replicated stream producer job
+	// so that it will be dropped.
+	registry.TestingNudgeAdoptionQueue()
+
+	now := c.DestSysServer.Clock().Now()
+	testutils.SucceedsWithin(t, func() error {
+		var spanConfigJobID jobspb.JobID
+		rows := c.DestTenantSQL.Query(
+			t, `SELECT id FROM system.jobs WHERE job_type = 'AUTO SPAN CONFIG RECONCILIATION' AND status = 'running'`,
+		)
+		defer rows.Close()
+		if !rows.Next() {
+			return errors.New("no running span config reconciliation job found")
+		}
+		if err := rows.Scan(&spanConfigJobID); err != nil {
+			return err
+		}
+		spanConfigJob, err := registry.LoadJob(ctx, spanConfigJobID)
+		require.NoError(t, err)
+		spanConfigProg := spanConfigJob.Progress().
+			Details.(*jobspb.Progress_AutoSpanConfigReconciliation).
+			AutoSpanConfigReconciliation
+
+		if spanConfigProg.Checkpoint.Less(now) {
+			return errors.Newf(
+				"waiting for span config reconciliation job to checkpoint past %v, at %v",
+				now, spanConfigProg.Checkpoint,
+			)
+		}
+		return nil
+	}, 2*time.Minute)
+}

pkg/crosscluster/producer/producer_job.go

@@ -133,6 +133,19 @@ func (p *producerJobResumer) Resume(ctx context.Context, execCtx interface{}) er
 	jobExec := execCtx.(sql.JobExecContext)
 	execCfg := jobExec.ExecCfg()
 
+	// If the source tenant of a PCR stream is the system tenant, then the
+	// producer job will also be copied to the destination tenant, along with the
+	// associated PTS. After cutover, this blocks span config reconciliation since
+	// a tenant cannot lay a PTS on spans that another tenant owns. The producer
+	// job of course is not needed on the destination tenant, so we check for a
+	// mismatched cluster ID here and fast-exit the job.
+	if p.job.Payload().CreationClusterID != execCfg.NodeInfo.LogicalClusterID() {
+		return jobs.MarkAsPermanentJobError(errors.Newf(
+			"replication stream %d belongs to cluster %s, cannot resume on cluster %s",
+			p.job.ID(), p.job.Payload().CreationClusterID, execCfg.NodeInfo.LogicalClusterID(),
+		))
+	}
+
 	// Fire the timer immediately to start an initial progress check
 	p.timer.Reset(0)
 	for {

pkg/crosscluster/replicationtestutils/testutils.go

@@ -131,11 +131,12 @@ type TenantStreamingClusters struct {
 	SrcURL          url.URL
 	SrcCleanup      func()
 
-	DestCluster    *testcluster.TestCluster
-	DestSysServer  serverutils.ApplicationLayerInterface
-	DestSysSQL     *sqlutils.SQLRunner
-	DestTenantConn *gosql.DB
-	DestTenantSQL  *sqlutils.SQLRunner
+	DestCluster      *testcluster.TestCluster
+	DestSysServer    serverutils.ApplicationLayerInterface
+	DestSysSQL       *sqlutils.SQLRunner
+	DestTenantConn   *gosql.DB
+	DestTenantSQL    *sqlutils.SQLRunner
+	DestTenantServer serverutils.ApplicationLayerInterface
 
 	ReaderTenantSQL *sqlutils.SQLRunner
 
@@ -186,19 +187,18 @@ func (c *TenantStreamingClusters) init(ctx context.Context) {
 func (c *TenantStreamingClusters) StartDestTenant(
 	ctx context.Context, withTestingKnobs *base.TestingKnobs, server int,
 ) func() {
+	var err error
+	var testKnobs base.TestingKnobs
 	if withTestingKnobs != nil {
-		var err error
-		_, c.DestTenantConn, err = c.DestCluster.Server(server).TenantController().StartSharedProcessTenant(ctx, base.TestSharedProcessTenantArgs{
-			TenantID:    c.Args.DestTenantID,
-			TenantName:  c.Args.DestTenantName,
-			Knobs:       *withTestingKnobs,
-			UseDatabase: "defaultdb",
-		})
-		require.NoError(c.T, err)
-	} else {
-		c.DestSysSQL.Exec(c.T, `ALTER TENANT $1 START SERVICE SHARED`, c.Args.DestTenantName)
-		c.DestTenantConn = c.DestCluster.Server(server).SystemLayer().SQLConn(c.T, serverutils.DBName("cluster:"+string(c.Args.DestTenantName)+"/defaultdb"))
+		testKnobs = *withTestingKnobs
 	}
+	c.DestTenantServer, c.DestTenantConn, err = c.DestCluster.Server(server).TenantController().StartSharedProcessTenant(ctx, base.TestSharedProcessTenantArgs{
+		TenantID:    c.Args.DestTenantID,
+		TenantName:  c.Args.DestTenantName,
+		Knobs:       testKnobs,
+		UseDatabase: "defaultdb",
+	})
+	require.NoError(c.T, err)
 
 	c.DestTenantSQL = sqlutils.MakeSQLRunner(c.DestTenantConn)
 	testutils.SucceedsSoon(c.T, func() error {