Merge #158411

158411: loqrecovery: use LogEngine in obvious places r=arulajmani a=pav-kv

Part of #97627

Co-authored-by: Pavel Kalinnikov <pavel@cockroachlabs.com>

Author: craig[bot]

pkg/kv/kvserver/loqrecovery/collect.go

@@ -180,21 +180,22 @@ func CollectStoresReplicaInfo(
 
 func visitStoreReplicas(
 	ctx context.Context,
-	state, raft storage.Reader,
+	stateRO kvstorage.StateRO,
+	raftRO kvstorage.RaftRO,
 	storeID roachpb.StoreID,
 	nodeID roachpb.NodeID,
 	send func(info loqrecoverypb.ReplicaInfo) error,
 ) error {
-	if err := kvstorage.IterateRangeDescriptorsFromDisk(ctx, state, func(desc roachpb.RangeDescriptor) error {
+	if err := kvstorage.IterateRangeDescriptorsFromDisk(ctx, stateRO, func(desc roachpb.RangeDescriptor) error {
 		rsl := kvstorage.MakeStateLoader(desc.RangeID)
-		rstate, err := rsl.Load(ctx, state, &desc)
+		rstate, err := rsl.Load(ctx, stateRO, &desc)
 		if err != nil {
 			return err
 		}
 		// TODO(pav-kv): the LoQ recovery flow uses only the applied index, and the
 		// HardState.Commit loaded here is unused. Consider removing. Make sure this
 		// doesn't break compatibility for ReplicaInfo unmarshalling.
-		hstate, err := rsl.LoadHardState(ctx, raft)
+		hstate, err := rsl.LoadHardState(ctx, raftRO)
 		if err != nil {
 			return err
 		}
@@ -208,7 +209,7 @@ func visitStoreReplicas(
 		// For the heuristics here, it would probably make sense to read from all
 		// LogIDs with unapplied entries.
 		rangeUpdates, err := GetDescriptorChangesFromRaftLog(
-			ctx, desc.RangeID, rstate.RaftAppliedIndex+1, math.MaxInt64, raft)
+			ctx, desc.RangeID, rstate.RaftAppliedIndex+1, math.MaxInt64, raftRO)
 		if err != nil {
 			return err
 		}
@@ -234,10 +235,10 @@ func visitStoreReplicas(
 // lo (inclusive) and hi (exclusive) and searches for changes to range
 // descriptors, as identified by presence of a commit trigger.
 func GetDescriptorChangesFromRaftLog(
-	ctx context.Context, rangeID roachpb.RangeID, lo, hi kvpb.RaftIndex, reader storage.Reader,
+	ctx context.Context, rangeID roachpb.RangeID, lo, hi kvpb.RaftIndex, raftRO kvstorage.RaftRO,
 ) ([]loqrecoverypb.DescriptorChangeInfo, error) {
 	var changes []loqrecoverypb.DescriptorChangeInfo
-	if err := raftlog.Visit(ctx, reader, rangeID, lo, hi, func(ent raftpb.Entry) error {
+	if err := raftlog.Visit(ctx, raftRO, rangeID, lo, hi, func(ent raftpb.Entry) error {
 		e, err := raftlog.NewEntry(ent)
 		if err != nil {
 			return err

pkg/kv/kvserver/loqrecovery/collect_raft_log_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverbase"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvstorage"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/loqrecovery"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/loqrecovery/loqrecoverypb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -72,9 +73,9 @@ func TestFindUpdateDescriptor(t *testing.T) {
 
 					return srk
 				},
-				func(t *testing.T, ctx context.Context, reader storage.Reader) {
+				func(t *testing.T, ctx context.Context, raftRO kvstorage.RaftRO) {
 					seq, err := loqrecovery.GetDescriptorChangesFromRaftLog(
-						ctx, testRangeID, 0, math.MaxInt64, reader)
+						ctx, testRangeID, 0, math.MaxInt64, raftRO)
 					require.NoError(t, err, "failed to read raft log data")
 
 					requireContainsDescriptor(t, loqrecoverypb.DescriptorChangeInfo{
@@ -125,9 +126,9 @@ func TestFindUpdateRaft(t *testing.T) {
 
 					return srk
 				},
-				func(t *testing.T, ctx context.Context, reader storage.Reader) {
+				func(t *testing.T, ctx context.Context, raftRO kvstorage.RaftRO) {
 					seq, err := loqrecovery.GetDescriptorChangesFromRaftLog(
-						ctx, sRD.RangeID, 0, math.MaxInt64, reader)
+						ctx, sRD.RangeID, 0, math.MaxInt64, raftRO)
 					require.NoError(t, err, "failed to read raft log data")
 					requireContainsDescriptor(t, loqrecoverypb.DescriptorChangeInfo{
 						ChangeType: loqrecoverypb.DescriptorChangeType_ReplicaChange,
@@ -142,13 +143,13 @@ func checkRaftLog(
 	ctx context.Context,
 	nodeToMonitor int,
 	action func(ctx context.Context, tc *testcluster.TestCluster) roachpb.RKey,
-	assertRaftLog func(*testing.T, context.Context, storage.Reader),
+	assertRaftLog func(*testing.T, context.Context, kvstorage.RaftRO),
 	leaseType roachpb.LeaseType,
 ) {
 	t.Helper()
 
 	makeSnapshot := make(chan storage.Engine, 2)
-	snapshots := make(chan storage.Reader, 2)
+	snapshots := make(chan kvstorage.RaftRO, 2)
 
 	raftFilter := func(args kvserverbase.ApplyFilterArgs) (int, *kvpb.Error) {
 		t.Helper()
@@ -209,7 +210,7 @@ func checkRaftLog(
 
 	// TODO(sep-raft-log): the receiver doesn't use the engine, so remove this
 	// altogether and use a `chan struct{}{}`.
-	eng := tc.GetFirstStoreFromServer(t, nodeToMonitor).TODOEngine()
+	eng := tc.GetFirstStoreFromServer(t, nodeToMonitor).LogEngine()
 	makeSnapshot <- eng
 	// After the test action is complete raft might be completely caught up with
 	// its messages, so we will write a value into the range to ensure filter

pkg/kv/kvserver/loqrecovery/record.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverpb"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvstorage"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/loqrecovery/loqrecoverypb"
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
@@ -64,7 +65,7 @@ func writeReplicaRecoveryStoreRecord(
 // recovery actions are properly reflected in server logs as needed.
 func RegisterOfflineRecoveryEvents(
 	ctx context.Context,
-	readWriter storage.ReadWriter,
+	readWriter kvstorage.RaftRW,
 	registerEvent func(context.Context, loqrecoverypb.ReplicaRecoveryRecord) (bool, error),
 ) (int, error) {
 	successCount := 0

pkg/kv/kvserver/loqrecovery/recovery_env_test.go

@@ -843,6 +843,7 @@ func (e *quorumRecoveryEnv) dumpRecoveryEvents(
 		if !ok {
 			t.Fatalf("store s%d doesn't exist, but event dump is requested for it", store)
 		}
+		// TODO(sep-raft-log): store.engine should be the log engine.
 		if _, err := RegisterOfflineRecoveryEvents(ctx, store.engine, logEvents); err != nil {
 			return "", err
 		}

pkg/kv/kvserver/loqrecovery/server.go

@@ -479,7 +479,7 @@ func (s Server) NodeStatus(
 		status.PendingPlanID = &plan.PlanID
 	}
 	err = s.stores.VisitStores(func(s *kvserver.Store) error {
-		r, ok, err := readNodeRecoveryStatusInfo(ctx, s.TODOEngine())
+		r, ok, err := readNodeRecoveryStatusInfo(ctx, s.LogEngine())
 		if err != nil {
 			return err
 		}

pkg/server/loss_of_quorum.go

@@ -62,7 +62,7 @@ func logPendingLossOfQuorumRecoveryEvents(ctx context.Context, stores *kvserver.
 		// cluster-replicated destinations.
 		eventCount, err := loqrecovery.RegisterOfflineRecoveryEvents(
 			ctx,
-			s.TODOEngine(),
+			s.LogEngine(),
 			func(ctx context.Context, record loqrecoverypb.ReplicaRecoveryRecord) (bool, error) {
 				event := record.AsStructuredLog()
 				log.StructuredEvent(ctx, severity.INFO, &event)
@@ -94,7 +94,7 @@ func maybeRunLossOfQuorumRecoveryCleanup(
 		if err := stores.VisitStores(func(s *kvserver.Store) error {
 			_, err := loqrecovery.RegisterOfflineRecoveryEvents(
 				ctx,
-				s.TODOEngine(),
+				s.LogEngine(),
 				func(ctx context.Context, record loqrecoverypb.ReplicaRecoveryRecord) (bool, error) {
 					sqlExec := func(ctx context.Context, stmt string, args ...interface{}) (int, error) {
 						return ie.ExecEx(ctx, "", nil,