Merge #150168 #150835 #151058

150168: kvstorage: revise replica deletion plan r=tbg a=pav-kv

This PR modifies the plan for `CreateUninitializedReplica`, `DestroyReplica` and `applySnapshot`. The `LogID` concept is no longer needed, and the snapshot transitions can retain the old raft log up to `RaftAppliedIndex` since the snapshot is always created at a higher index.

Epic: CRDB-49111

150835: kvserver: add mmaStoreRebalancer r=tbg,sumeerbhola a=wenyihu6

Epic: CRDB-25222
Release note: none 

---

**kvserver: add MMARangeLoad**

Previously, we lacked a utility method for mma to inspect range load and convert
it to a mmaprototype.RangeLoad. This commit adds MMARangeLoad(), which builds a
mmaprototype.RangeLoad from a replica’s LoadStats. It is currently unused;
future commits will use it to populate the range message field for MMA input.

---

**kvserver: add mmaRangeMessageNeeded**

Previously, there was no struct under Replica to track state changes and signal
mma when an up-to-date range message with full information was needed. This
commit introduces such a struct - mmaRangeMessageNeeded. It is marked as needed
when any of the following occur: replica initialization, span config changes,
leaseholder changes, or range descriptor updates.

---

**kvserver: add more methods to mmaRangeMessageNeeded**

Previously, mmaRangeMessageNeeded was introduced to track when a replica state
change required sending updated information. This commit adds methods to
mmaRangeMessageNeeded to help mma decide whether a new range message with full
information is needed. Since processing full updates can be expensive, mma sends
them only when necessary. In addition to tracking state changes, the struct now
also tracks the content of the last full info range message sent to determine if
full information is necessary.

---

**kvserver: add store.MakeStoreLeaseholderMsg**

Previously, mma lacked a method to retrieve range messages for all leaseholder
replicas on a store. This commit adds such a method to Store, which iterates
over all replicas and constructs range message for leaseholder replicas. Note
that it is currently left unused. mma will use this as input for computing
rebalancing decisions in future commits.

---

**kvserver: add mmaStoreRebalancer**

This commit introduces mma_store_rebalancer.go, which defines the main struct
mmaStoreRebalancer and its basic methods for computing and applying rebalancing
changes. It is currently unplumbed; future commits will wire it into each store
and start a background goroutine to perform rebalancing periodically.

---

**kvserver: plumb mmaStoreRebalancer**

Previously, mmaStoreRebalancer was unplumbed. This commit initializes it per
Store. It also constructs and plumbs a shared MMAAllocator per node, which is
shared by all stores' mmaStoreRebalancer on the same node.

---

**kvserver: use RegisterCallback for mma**

Previously, we extended RegisterCallback to include origTimestampNanos in the
callback arguments. This commit puts it to use for MMA by registering a callback
in NewServer that notifies the MMA allocator of stores and updated store load
messages. This callback is triggered when the store descriptor is gossiped via
StoreGossip, either periodically or due to significant store capacity changes.

151058: backup: fix external storage error propagation r=dt a=dt

See #151050.
Fixes #151050.

Release note (bug fix): fix a bug that could cause some error returned by attempts to upload backup data to external storage providers to be undetected, potentially causing incomplete backups.
Epic: none.

Co-authored-by: Pavel Kalinnikov <[email protected]>
Co-authored-by: wenyihu6 <[email protected]>
Co-authored-by: David Taylor <[email protected]>

Author: 4 people

pkg/backup/backup_processor.go

@@ -292,6 +292,14 @@ type spanAndTime struct {
 	finishesSpec bool
 }
 
+type errInjectingStorage struct {
+	cloud.ExternalStorage
+}
+
+func (e errInjectingStorage) Writer(_ context.Context, _ string) (io.WriteCloser, error) {
+	return nil, errors.New("injected error")
+}
+
 func runBackupProcessor(
 	ctx context.Context,
 	flowCtx *execinfra.FlowCtx,
@@ -393,12 +401,19 @@ func runBackupProcessor(
 		Settings:  &flowCtx.Cfg.Settings.SV,
 		ElideMode: spec.ElidePrefix,
 	}
+
 	storage, err := flowCtx.Cfg.ExternalStorage(ctx, dest, cloud.WithClientName("backup"))
 	if err != nil {
 		return err
 	}
 	defer logClose(ctx, storage, "external storage")
 
+	if backupKnobs, ok := flowCtx.TestingKnobs().BackupRestoreTestingKnobs.(*sql.BackupRestoreTestingKnobs); ok {
+		if fn := backupKnobs.InjectErrorsInBackupRowDataStorage; fn != nil && fn() {
+			storage = errInjectingStorage{storage}
+		}
+	}
+
 	// Start start a group of goroutines which each pull spans off of `todo` and
 	// send export requests. Any spans that encounter lock conflict errors during
 	// Export are put back on the todo queue for later processing.
@@ -683,7 +698,7 @@ func runBackupProcessor(
 							var writeErr error
 							resumeSpan.span.Key, writeErr = sink.Write(ctx, ret)
 							if writeErr != nil {
-								return err
+								return writeErr
 							}
 						}
 						// Emit the stats for the processed ExportRequest.

pkg/backup/backup_test.go

@@ -6322,6 +6322,36 @@ func TestPublicIndexTableSpans(t *testing.T) {
 	}
 }
 
+// TestBackupStorageErrorPropagates ensures that errors from writing to storage
+// propagate correctly during a backup operation.
+func TestBackupStorageErrorPropagates(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	const numAccounts = 1000
+
+	var fail atomic.Bool
+
+	params := base.TestClusterArgs{}
+	knobs := base.TestingKnobs{
+		DistSQL: &execinfra.TestingKnobs{BackupRestoreTestingKnobs: &sql.BackupRestoreTestingKnobs{
+			InjectErrorsInBackupRowDataStorage: func() bool { return fail.Load() },
+		}},
+	}
+	params.ServerArgs.Knobs = knobs
+
+	tc, _, _, cleanupFn := backupRestoreTestSetupWithParams(t, singleNode, numAccounts, InitManualReplication, params)
+	defer cleanupFn()
+	db := tc.ServerConn(0)
+	runner := sqlutils.MakeSQLRunner(db)
+
+	runner.Exec(t, "BACKUP DATABASE data INTO 'nodelocal://1/success'")
+	runner.Exec(t, "RESTORE DATABASE data FROM LATEST IN 'nodelocal://1/success' WITH new_db_name = 'restored'")
+	runner.CheckQueryResults(t, "SELECT count(*) FROM restored.bank", [][]string{{"1000"}})
+	fail.Store(true)
+	runner.ExpectErr(t, "injected", "BACKUP DATABASE data INTO 'nodelocal://1/failure'")
+}
+
 // TestRestoreJobErrorPropagates ensures that errors from creating the job
 // record propagate correctly.
 func TestRestoreErrorPropagates(t *testing.T) {

pkg/kv/kvserver/BUILD.bazel

@@ -20,6 +20,8 @@ go_library(
         "merge_queue.go",
         "metric_rules.go",
         "metrics.go",
+        "mma_replica_store.go",
+        "mma_store_rebalancer.go",
         "mvcc_gc_queue.go",
         "queue.go",
         "queue_helpers_testutil.go",
@@ -130,6 +132,7 @@ go_library(
         "//pkg/kv/kvserver/allocator",
         "//pkg/kv/kvserver/allocator/allocatorimpl",
         "//pkg/kv/kvserver/allocator/load",
+        "//pkg/kv/kvserver/allocator/mmaprototype",
         "//pkg/kv/kvserver/allocator/plan",
         "//pkg/kv/kvserver/allocator/storepool",
         "//pkg/kv/kvserver/apply",
@@ -170,6 +173,7 @@ go_library(
         "//pkg/kv/kvserver/raftentry",
         "//pkg/kv/kvserver/raftlog",
         "//pkg/kv/kvserver/rafttrace",
+        "//pkg/kv/kvserver/raftutil",
         "//pkg/kv/kvserver/rangefeed",
         "//pkg/kv/kvserver/rditer",
         "//pkg/kv/kvserver/readsummary",

pkg/kv/kvserver/allocator/mmaprototype/cluster_state.go

@@ -1201,15 +1201,15 @@ func (cs *clusterState) processStoreLeaseholderMsgInternal(
 		rs, ok := cs.ranges[rangeMsg.RangeID]
 		if !ok {
 			// This is the first time we've seen this range.
-			if !rangeMsg.Populated {
+			if !rangeMsg.MaybeSpanConfIsPopulated {
 				panic(errors.AssertionFailedf("rangeMsg for new range r%v is not populated", rangeMsg.RangeID))
 			}
 			rs = newRangeState(msg.StoreID)
 			cs.ranges[rangeMsg.RangeID] = rs
 		} else if rs.localRangeOwner != msg.StoreID {
 			rs.localRangeOwner = msg.StoreID
 		}
-		if !rangeMsg.Populated {
+		if !rangeMsg.MaybeSpanConfIsPopulated {
 			// When there are no pending changes, confirm that the membership state
 			// is consistent. If not, fall through and make it consistent. We have
 			// seen an example where AdjustPendingChangesDisposition lied about
@@ -1243,7 +1243,7 @@ func (cs *clusterState) processStoreLeaseholderMsgInternal(
 		// Set the range state and store state to match the range message state
 		// initially. The pending changes which are not enacted in the range
 		// message are handled and added back below.
-		if rangeMsg.Populated {
+		if rangeMsg.MaybeSpanConfIsPopulated {
 			rs.load = rangeMsg.RangeLoad
 		}
 		for _, replica := range rs.replicas {
@@ -1342,8 +1342,8 @@ func (cs *clusterState) processStoreLeaseholderMsgInternal(
 				}
 			}
 		}
-		if rangeMsg.Populated {
-			normSpanConfig, err := makeNormalizedSpanConfig(&rangeMsg.Conf, cs.constraintMatcher.interner)
+		if rangeMsg.MaybeSpanConfIsPopulated {
+			normSpanConfig, err := makeNormalizedSpanConfig(&rangeMsg.MaybeSpanConf, cs.constraintMatcher.interner)
 			if err != nil {
 				// TODO(kvoli): Should we log as a warning here, or return further back out?
 				panic(err)

pkg/kv/kvserver/allocator/mmaprototype/cluster_state_test.go

@@ -106,7 +106,7 @@ func parseStoreLeaseholderMsg(t *testing.T, in string) StoreLeaseholderMsg {
 	tryAppendRangeMsg := func() {
 		if rMsg.RangeID != 0 {
 			if notPopulatedOverride {
-				rMsg.Populated = false
+				rMsg.MaybeSpanConfIsPopulated = false
 			}
 			msg.Ranges = append(msg.Ranges, rMsg)
 			rMsg = RangeMsg{RangeID: 0}
@@ -124,17 +124,17 @@ func parseStoreLeaseholderMsg(t *testing.T, in string) StoreLeaseholderMsg {
 					rMsg.RangeID = roachpb.RangeID(parseInt(t, parts[1]))
 				case "load":
 					rMsg.RangeLoad.Load = parseLoadVector(t, parts[1])
-					rMsg.Populated = true
+					rMsg.MaybeSpanConfIsPopulated = true
 				case "raft-cpu":
 					rMsg.RangeLoad.RaftCPU = LoadValue(parseInt(t, parts[1]))
-					rMsg.Populated = true
+					rMsg.MaybeSpanConfIsPopulated = true
 				case "not-populated":
 					notPopulatedOverride = true
 				}
 			}
 		} else if strings.HasPrefix(line, "config=") {
-			rMsg.Conf = spanconfigtestutils.ParseZoneConfig(t, strings.TrimPrefix(line, "config=")).AsSpanConfig()
-			rMsg.Populated = true
+			rMsg.MaybeSpanConf = spanconfigtestutils.ParseZoneConfig(t, strings.TrimPrefix(line, "config=")).AsSpanConfig()
+			rMsg.MaybeSpanConfIsPopulated = true
 		} else {
 			var repl StoreIDAndReplicaState
 			fields := strings.Fields(line)
@@ -158,7 +158,7 @@ func parseStoreLeaseholderMsg(t *testing.T, in string) StoreLeaseholderMsg {
 				}
 			}
 			rMsg.Replicas = append(rMsg.Replicas, repl)
-			rMsg.Populated = true
+			rMsg.MaybeSpanConfIsPopulated = true
 		}
 	}
 	tryAppendRangeMsg()

pkg/kv/kvserver/allocator/mmaprototype/messages.go

@@ -43,20 +43,20 @@ type StoreLeaseholderMsg struct {
 
 // RangeMsg is generated by the leaseholder store (and part of
 // StoreLeaseholderMsg). If there is any change for that range, the full
-// information for that range is provided (and Populated is set to true). This
-// is also the case for a new leaseholder since it does not know whether
-// something has changed since the last leaseholder informed the allocator. A
-// tiny change to the RangeLoad (decided by the caller) will not cause the
-// fields to be populated.
+// information for that range is provided (and MaybeSpanConfIsPopulated is set
+// to true). This is also the case for a new leaseholder since it does not know
+// whether something has changed since the last leaseholder informed the
+// allocator. A tiny change to the RangeLoad (decided by the caller) will not
+// cause the fields to be populated.
 //
 // To ensure that the allocator does not lose synchronization with the current
 // set of replicas, due to spurious changes (we had one undiagnosed example
 // where the allocator was spuriously told that a lease was transferred away),
 // the Replicas field is always populated).
 type RangeMsg struct {
 	roachpb.RangeID
-	Replicas  []StoreIDAndReplicaState
-	Populated bool
-	Conf      roachpb.SpanConfig
-	RangeLoad RangeLoad
+	Replicas                 []StoreIDAndReplicaState
+	MaybeSpanConfIsPopulated bool
+	MaybeSpanConf            roachpb.SpanConfig
+	RangeLoad                RangeLoad
 }

pkg/kv/kvserver/asim/mmaintegration/mma_integration.go

@@ -77,11 +77,11 @@ func MakeStoreLeaseholderMsgFromState(
 		rl.RaftCPU = mmaprototype.LoadValue(load.RaftCPUNanosPerSecond)
 
 		rangeMessages = append(rangeMessages, mmaprototype.RangeMsg{
-			RangeID:   roachpb.RangeID(replica.Range()),
-			Populated: true,
-			Replicas:  replicas,
-			Conf:      *rng.SpanConfig(),
-			RangeLoad: rl,
+			RangeID:                  roachpb.RangeID(replica.Range()),
+			MaybeSpanConfIsPopulated: true,
+			Replicas:                 replicas,
+			MaybeSpanConf:            *rng.SpanConfig(),
+			RangeLoad:                rl,
 		})
 	}
 

pkg/kv/kvserver/kvstorage/destroy.go

@@ -95,16 +95,17 @@ func ClearRangeData(
 // writes.
 //
 //  1. Log storage write (durable):
-//     1.1. Write WAG node with the state machine mutation (2).
+//     1.1. WAG: apply to RaftAppliedIndex.
+//     1.2. WAG: apply mutation (2).
 //  2. State machine mutation:
 //     2.1. Clear RangeID-local un-/replicated state.
 //     2.2. (optional) Clear replicated MVCC span.
-//     2.3. Write RangeTombstone with next ReplicaID/LogID.
+//     2.3. Write RangeTombstone with next ReplicaID.
 //  3. Log engine GC (after state machine mutation 2 is durably applied):
-//     3.1. Remove previous LogID.
+//     3.1. Remove raft state.
 //
-// TODO(sep-raft-log): support the status quo in which 1+2+3 is written
-// atomically, and 1.1 is not written.
+// TODO(sep-raft-log): support the status quo in which 2+3 is written
+// atomically, and 1 is not written.
 const DestroyReplicaTODO = 0
 
 // DestroyReplica destroys all or a part of the Replica's state, installing a
@@ -130,7 +131,7 @@ func DestroyReplica(
 	if diskReplicaID.ReplicaID >= nextReplicaID {
 		return errors.AssertionFailedf("replica r%d/%d must not survive its own tombstone", rangeID, diskReplicaID)
 	}
-	_ = DestroyReplicaTODO // 2.1 + 3.1 + 2.2
+	_ = DestroyReplicaTODO // 2.1 + 2.2 + 3.1
 	if err := ClearRangeData(ctx, rangeID, reader, writer, opts); err != nil {
 		return err
 	}

pkg/kv/kvserver/kvstorage/replica_state.go

@@ -111,7 +111,7 @@ func (r LoadedReplicaState) check(storeID roachpb.StoreID) error {
 //  1. Log storage write (durable):
 //     1.1. Write WAG node with the state machine mutation (2).
 //  2. State machine mutation:
-//     2.1. Write RaftReplicaID with the new ReplicaID/LogID.
+//     2.1. Write the new RaftReplicaID.
 //
 // TODO(sep-raft-log): support the status quo in which only 2.1 is written.
 const CreateUninitReplicaTODO = 0