Merge #143354 #143990 #144007 #144113 #144165

143354: server: enable TestServerControllerLoginLogout for secondary tenants r=dhartunian,cthumuluru-crdb a=shubhamdhama

Fixes: #110002
Epic: CRDB-38970
Release note: none

143990: distsql: cancel remote flows on quiesce r=yuzefovich a=yuzefovich

We recently saw a test timeout where the cluster shutdown was blocked on the outbox goroutine attempting to set up a connection to one of the nodes. We blocked there indefinitely since the remote flows previously didn't respect quiesce signal (we're using the background context there because the remote flows outlive the SetupFlow RPC handler context). This is now fixed.

The bug has been present since forever, and given that we haven't hit this before, it seems like an extreme edge case, so I decided to omit the release note.

Fixes: #143875.

Release note: None

144007: logical: have tombstone updater accept datums r=jeffswenson a=jeffswenson

Previously, the tombstone deleter accepted a cdcevent.Row. Now, it accepts datums derived from the cdcevent.Row. This allows the tombstone updater to be used by the crud sql writer which internally expects datums.

This also removes an extra `isLwwLoser(err)` check that caused the tombstone updater to silently drop errors. This bug was caught by testing in #143988.

Release Note: none
Epic: CRDB-48647

144113: util/json: remove `AllPathsWithDepth` r=normanchenn a=normanchenn

This commit reverts a65115d. `AllPathsWithDepth` was initially created to iterate through the children of a json array or object, but due to performance reasons (jsonEncoded would decode the entire object on each call), another approach was taken to use `json.ObjectIter` and `json.FetchValIdx`, thus eliminating the need for `AllPathsWithDepth`.

Epic: None
Release note: None

144165: workflows: run update_releases on release-25.2 r=iskettaneh a=iskettaneh

Epic: none

Release note: None

Co-authored-by: Shubham Dhama <[email protected]>
Co-authored-by: Yahor Yuzefovich <[email protected]>
Co-authored-by: Jeff Swenson <[email protected]>
Co-authored-by: Norman Chen <[email protected]>
Co-authored-by: Ibrahim Kettaneh <[email protected]>

Author: 6 people

.github/workflows/update_releases.yaml

@@ -31,6 +31,7 @@ jobs:
           - "release-24.1"
           - "release-24.3"
           - "release-25.1"
+          - "release-25.2"
     name: Update pkg/testutils/release/cockroach_releases.yaml on ${{ matrix.branch }}
     runs-on: ubuntu-latest
     steps:

pkg/ccl/serverccl/server_controller_test.go

@@ -563,16 +563,18 @@ func TestServerControllerLoginLogout(t *testing.T) {
 
 	ctx := context.Background()
 
-	srv := serverutils.StartServerOnly(t, base.TestServerArgs{
-		DefaultTestTenant: base.TestIsForStuffThatShouldWorkWithSecondaryTenantsButDoesntYet(110002),
-	})
+	srv := serverutils.StartServerOnly(t, base.TestServerArgs{})
 	defer srv.Stopper().Stop(ctx)
 	s := srv.ApplicationLayer()
+	isExternal := srv.DeploymentMode().IsExternal()
 
-	client, err := s.GetAuthenticatedHTTPClient(false, serverutils.SingleTenantSession)
+	sessionType := serverutils.SingleTenantSession
+	client, err := s.GetAuthenticatedHTTPClient(false, sessionType)
 	require.NoError(t, err)
 
-	resp, err := client.Post(s.AdminURL().WithPath("/logout").String(), "", nil)
+	// Using `Get` here instead of `Post` because in external process mode, the
+	// server returns a `StatusMethodNotAllowed` error when using `Post`.
+	resp, err := client.Get(s.AdminURL().WithPath("/logout").String())
 	require.NoError(t, err)
 	defer resp.Body.Close()
 
@@ -582,18 +584,33 @@ func TestServerControllerLoginLogout(t *testing.T) {
 	for i, c := range resp.Cookies() {
 		cookieNames[i] = c.Name
 		cookieValues[i] = c.Value
-		require.True(t, c.Secure)
+		// Secure isn't set in external-mode but it is set in other modes.
+		// See https://github.com/cockroachdb/cockroach/pull/143354#pullrequestreview-2751413632.
+		if !isExternal {
+			require.True(t, c.Secure)
+		}
 		if c.Name == "session" {
 			require.True(t, c.HttpOnly)
 		}
 	}
-	require.ElementsMatch(t, []string{"session", "tenant"}, cookieNames)
-	require.ElementsMatch(t, []string{"", ""}, cookieValues)
-
+	expectedCookies := []string{"session"}
+	expectedValues := []string{""}
+	if !isExternal {
+		expectedCookies = append(expectedCookies, "tenant")
+		expectedValues = append(expectedValues, "")
+	}
+	require.ElementsMatch(t, expectedCookies, cookieNames)
+	require.ElementsMatch(t, expectedValues, cookieValues)
+
+	// This part of the test doesn't run in external-process mode since we can't
+	// set a session cookie with name `tenant`—that's only supported by HTTP
+	// servers in the system tenant, which also handle routing for secondary
+	// tenants.
+	if isExternal {
+		return
+	}
 	// Need a new server because the HTTP Client is memoized.
-	srv2 := serverutils.StartServerOnly(t, base.TestServerArgs{
-		DefaultTestTenant: base.TestIsForStuffThatShouldWorkWithSecondaryTenantsButDoesntYet(110002),
-	})
+	srv2 := serverutils.StartServerOnly(t, base.TestServerArgs{})
 	defer srv2.Stopper().Stop(ctx)
 	s2 := srv2.ApplicationLayer()
 
@@ -623,7 +640,7 @@ func TestServerControllerLoginLogout(t *testing.T) {
 	require.NoError(t, err)
 	cookieJar.SetCookies(s2.AdminURL().URL, []*http.Cookie{
 		{
-			Name:  "multitenant-session",
+			Name:  "tenant",
 			Value: "abc-123",
 		},
 	})

pkg/crosscluster/logical/lww_row_processor.go

@@ -711,7 +711,7 @@ func (lww *lwwQuerier) DeleteRow(
 		// NOTE: at this point we don't know if we are updating a tombstone or if
 		// we are losing LWW. As long as it is a LWW loss or a tombstone update,
 		// updateTombstone will return okay.
-		return lww.tombstoneUpdaters[row.TableID].updateTombstone(ctx, txn, row)
+		return lww.tombstoneUpdaters[row.TableID].updateTombstoneAny(ctx, txn, row.MvccTimestamp, datums)
 	}
 	return batchStats{}, nil
 }

pkg/crosscluster/logical/tombstone_updater.go

@@ -8,7 +8,6 @@ package logical
 import (
 	"context"
 
-	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcevent"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
@@ -19,6 +18,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/row"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 )
 
 // tombstoneUpdater is a helper for updating the mvcc origin timestamp assigned
@@ -40,12 +40,8 @@ type tombstoneUpdater struct {
 		// deleter is a row.Deleter that uses the leased descriptor. Callers should
 		// use getDeleter to ensure the lease is valid for the current transaction.
 		deleter row.Deleter
-		// columns are the name of the columns that are expected in the cdc row.
-		columns []string
 	}
 
-	// scratch is a scratch buffer for the tombstone updater. This is reused
-	// across calls to addToBatch to avoid allocations.
 	scratch []tree.Datum
 }
 
@@ -56,7 +52,6 @@ func (c *tombstoneUpdater) ReleaseLeases(ctx context.Context) {
 		c.leased.descriptor.Release(ctx)
 		c.leased.descriptor = nil
 		c.leased.deleter = row.Deleter{}
-		c.leased.columns = c.leased.columns[:0]
 	}
 }
 
@@ -78,19 +73,31 @@ func newTombstoneUpdater(
 	}
 }
 
+// updateTombstoneAny is an `updateTombstone` wrapper that accepts the []any
+// datum slice from the original sql writer's datum builder.
+func (tu *tombstoneUpdater) updateTombstoneAny(
+	ctx context.Context, txn isql.Txn, mvccTimestamp hlc.Timestamp, datums []any,
+) (batchStats, error) {
+	tu.scratch = tu.scratch[:0]
+	for _, datum := range datums {
+		tu.scratch = append(tu.scratch, datum.(tree.Datum))
+	}
+	return tu.updateTombstone(ctx, txn, mvccTimestamp, tu.scratch)
+}
+
 // updateTombstone attempts to update the tombstone for the given row. This is
 // expected to always succeed. The delete will only return zero rows if the
 // operation loses LWW or the row does not exist. So if the cput fails on a
 // condition, it should also fail on LWW, which is treated as a success.
 func (tu *tombstoneUpdater) updateTombstone(
-	ctx context.Context, txn isql.Txn, afterRow cdcevent.Row,
+	ctx context.Context, txn isql.Txn, mvccTimestamp hlc.Timestamp, afterRow []tree.Datum,
 ) (batchStats, error) {
 	err := func() error {
 		if txn != nil {
 			// If updateTombstone is called in a transaction, create and run a batch
 			// in the transaction.
 			batch := txn.KV().NewBatch()
-			if err := tu.addToBatch(ctx, txn.KV(), batch, afterRow); err != nil {
+			if err := tu.addToBatch(ctx, txn.KV(), batch, mvccTimestamp, afterRow); err != nil {
 				return err
 			}
 			return txn.KV().Run(ctx, batch)
@@ -99,13 +106,13 @@ func (tu *tombstoneUpdater) updateTombstone(
 		// 1pc transaction.
 		return tu.db.Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
 			batch := txn.NewBatch()
-			if err := tu.addToBatch(ctx, txn, batch, afterRow); err != nil {
+			if err := tu.addToBatch(ctx, txn, batch, mvccTimestamp, afterRow); err != nil {
 				return err
 			}
 			return txn.CommitInBatch(ctx, batch)
 		})
 	}()
-	if err != nil && isLwwLoser(err) {
+	if err != nil {
 		if isLwwLoser(err) {
 			return batchStats{kvWriteTooOld: 1}, nil
 		}
@@ -115,44 +122,28 @@ func (tu *tombstoneUpdater) updateTombstone(
 }
 
 func (tu *tombstoneUpdater) addToBatch(
-	ctx context.Context, txn *kv.Txn, batch *kv.Batch, afterRow cdcevent.Row,
+	ctx context.Context,
+	txn *kv.Txn,
+	batch *kv.Batch,
+	mvccTimestamp hlc.Timestamp,
+	afterRow []tree.Datum,
 ) error {
 	deleter, err := tu.getDeleter(ctx, txn)
 	if err != nil {
 		return err
 	}
 
-	tu.scratch = tu.scratch[:0]
-
-	// Note that the columns in the cdcevent row are decoded using a descriptor
-	// for the source column, whereas the row.Deleter column list is initialized
-	// using columns from the destination table. This is funky, but it works
-	// because we validate LDR schemas are compatible.
-
-	datums, err := afterRow.DatumsNamed(tu.leased.columns)
-	if err != nil {
-		return err
-	}
-	if err := datums.Datum(func(d tree.Datum, col cdcevent.ResultColumn) error {
-		tu.scratch = append(tu.scratch, d)
-		return nil
-	}); err != nil {
-		return err
-	}
-
-	// the index helpers are never really used since we are always updating a
-	// tombstone.
 	var ph row.PartialIndexUpdateHelper
 	var vh row.VectorIndexUpdateHelper
 
 	return deleter.DeleteRow(
 		ctx,
 		batch,
-		tu.scratch,
+		afterRow,
 		ph,
 		vh,
 		&row.OriginTimestampCPutHelper{
-			OriginTimestamp:    afterRow.MvccTimestamp,
+			OriginTimestamp:    mvccTimestamp,
 			PreviousWasDeleted: true,
 		},
 		false, /* mustValidateOldPKValues */
@@ -176,10 +167,6 @@ func (tu *tombstoneUpdater) getDeleter(ctx context.Context, txn *kv.Txn) (row.De
 			return row.Deleter{}, err
 		}
 
-		for _, col := range cols {
-			tu.leased.columns = append(tu.leased.columns, col.GetName())
-		}
-
 		tu.leased.deleter = row.MakeDeleter(tu.codec, tu.leased.descriptor.Underlying().(catalog.TableDescriptor), nil /* lockedIndexes */, cols, tu.sd, &tu.settings.SV, nil /* metrics */)
 	}
 	if err := txn.UpdateDeadline(ctx, tu.leased.descriptor.Expiration(ctx)); err != nil {

pkg/sql/distsql/server.go

@@ -624,6 +624,10 @@ func (ds *ServerImpl) SetupFlow(
 	// Note: the passed context will be canceled when this RPC completes, so we
 	// can't associate it with the flow since it outlives the RPC.
 	ctx = ds.AnnotateCtx(context.Background())
+	// Ensure that the flow respects the node being shut down. Note that since
+	// the flow outlives the RPC, we cannot defer the cancel function, so we
+	// simply ignore it.
+	ctx, _ = ds.Stopper.WithCancelOnQuiesce(ctx)
 	if err := func() error {
 		// Reserve some memory for this remote flow which is a poor man's
 		// admission control based on the RAM usage.

pkg/util/json/encoded.go

@@ -797,12 +797,12 @@ func (j *jsonEncoded) numInvertedIndexEntries() (int, error) {
 	return decoded.numInvertedIndexEntries()
 }
 
-func (j *jsonEncoded) allPathsWithDepth(depth int) ([]JSON, error) {
+func (j *jsonEncoded) allPaths() ([]JSON, error) {
 	decoded, err := j.decode()
 	if err != nil {
 		return nil, err
 	}
-	return decoded.allPathsWithDepth(depth)
+	return decoded.allPaths()
 }
 
 // HasContainerLeaf implements the JSON interface.

pkg/util/json/json.go

@@ -138,12 +138,10 @@ type JSON interface {
 	// produced if this JSON gets included in an inverted index.
 	numInvertedIndexEntries() (int, error)
 
-	// allPathsWithDepth returns a slice of new JSON documents, each a path
-	// through the receiver. The depth parameter specifies the maximum depth of
-	// the paths to return. If the depth is negative, all paths of any depth are
-	// returned. If the depth is 0, the receiver itself is returned. Note that
-	// leaves include the empty object and array in addition to scalars.
-	allPathsWithDepth(depth int) ([]JSON, error)
+	// allPaths returns a slice of new JSON documents, each a path to a leaf
+	// through the receiver. Note that leaves include the empty object and array
+	// in addition to scalars.
+	allPaths() ([]JSON, error)
 
 	// FetchValKey implements the `->` operator for strings, returning nil if the
 	// key is not found.
@@ -1795,51 +1793,36 @@ func (j jsonObject) numInvertedIndexEntries() (int, error) {
 // through the input. Note that leaves include the empty object and array
 // in addition to scalars.
 func AllPaths(j JSON) ([]JSON, error) {
-	return j.allPathsWithDepth(-1)
+	return j.allPaths()
 }
 
-// AllPathsWithDepth returns a slice of new JSON documents, each a path
-// through the receiver. The depth parameter specifies the maximum depth of
-// the paths to return. If the depth is negative, all paths of any depth are
-// returned. If the depth is 0, the receiver itself is returned. Note that
-// leaves include the empty object and array in addition to scalars.
-func AllPathsWithDepth(j JSON, depth int) ([]JSON, error) {
-	return j.allPathsWithDepth(depth)
-}
-
-func (j jsonNull) allPathsWithDepth(depth int) ([]JSON, error) {
+func (j jsonNull) allPaths() ([]JSON, error) {
 	return []JSON{j}, nil
 }
 
-func (j jsonTrue) allPathsWithDepth(depth int) ([]JSON, error) {
+func (j jsonTrue) allPaths() ([]JSON, error) {
 	return []JSON{j}, nil
 }
 
-func (j jsonFalse) allPathsWithDepth(depth int) ([]JSON, error) {
+func (j jsonFalse) allPaths() ([]JSON, error) {
 	return []JSON{j}, nil
 }
 
-func (j jsonString) allPathsWithDepth(depth int) ([]JSON, error) {
+func (j jsonString) allPaths() ([]JSON, error) {
 	return []JSON{j}, nil
 }
 
-func (j jsonNumber) allPathsWithDepth(depth int) ([]JSON, error) {
+func (j jsonNumber) allPaths() ([]JSON, error) {
 	return []JSON{j}, nil
 }
 
-func (j jsonArray) allPathsWithDepth(depth int) ([]JSON, error) {
-	if len(j) == 0 || depth == 0 {
+func (j jsonArray) allPaths() ([]JSON, error) {
+	if len(j) == 0 {
 		return []JSON{j}, nil
 	}
 	ret := make([]JSON, 0, len(j))
 	for i := range j {
-		var paths []JSON
-		var err error
-		if depth > 0 {
-			paths, err = j[i].allPathsWithDepth(depth - 1)
-		} else {
-			paths, err = j[i].allPathsWithDepth(depth)
-		}
+		paths, err := j[i].allPaths()
 		if err != nil {
 			return nil, err
 		}
@@ -1850,19 +1833,13 @@ func (j jsonArray) allPathsWithDepth(depth int) ([]JSON, error) {
 	return ret, nil
 }
 
-func (j jsonObject) allPathsWithDepth(depth int) ([]JSON, error) {
-	if len(j) == 0 || depth == 0 {
+func (j jsonObject) allPaths() ([]JSON, error) {
+	if len(j) == 0 {
 		return []JSON{j}, nil
 	}
 	ret := make([]JSON, 0, len(j))
 	for i := range j {
-		var paths []JSON
-		var err error
-		if depth > 0 {
-			paths, err = j[i].v.allPathsWithDepth(depth - 1)
-		} else {
-			paths, err = j[i].v.allPathsWithDepth(depth)
-		}
+		paths, err := j[i].v.allPaths()
 		if err != nil {
 			return nil, err
 		}