kvserver: use StateEngine durability for log truncations

Loosely coupled log truncations happen when the applied index is durable
on the state machine engine. This commit makes all subscriptions to
durability depend on the StateEngine, for that matter.

Epic: none
Release note: none

Author: pav-kv

pkg/kv/kvserver/client_merge_test.go

@@ -756,9 +756,9 @@ func mergeCheckingTimestampCaches(
 			// the result to apply on the majority quorum.
 			testutils.SucceedsSoon(t, func() error {
 				for _, r := range lhsRepls[1:] {
-					// Loosely-coupled truncation requires an engine flush to advance
-					// guaranteed durability.
-					require.NoError(t, r.Store().TODOEngine().Flush())
+					// Loosely-coupled truncation requires the state engine flush to
+					// advance guaranteed durability.
+					require.NoError(t, r.Store().StateEngine().Flush())
 					if firstIndex := r.GetCompactedIndex() + 1; firstIndex < truncIndex {
 						return errors.Errorf("truncate not applied, %d < %d", firstIndex, truncIndex)
 					}

pkg/kv/kvserver/client_raft_log_queue_test.go

@@ -114,7 +114,7 @@ func TestRaftLogQueue(t *testing.T) {
 				tc.GetFirstStoreFromServer(t, i).MustForceRaftLogScanAndProcess()
 			}
 			// Flush the engine to advance durability, which triggers truncation.
-			require.NoError(t, raftLeaderRepl.Store().TODOEngine().Flush())
+			require.NoError(t, raftLeaderRepl.Store().StateEngine().Flush())
 			// Ensure that compacted index has increased indicating that the log
 			// truncation has occurred.
 			afterTruncationIndex = raftLeaderRepl.GetCompactedIndex()

pkg/kv/kvserver/client_raft_test.go

@@ -840,7 +840,7 @@ func TestSnapshotAfterTruncation(t *testing.T) {
 func waitForTruncationForTesting(t *testing.T, r *kvserver.Replica, compacted kvpb.RaftIndex) {
 	testutils.SucceedsSoon(t, func() error {
 		// Flush the engine to advance durability, which triggers truncation.
-		require.NoError(t, r.Store().TODOEngine().Flush())
+		require.NoError(t, r.Store().StateEngine().Flush())
 		if index := r.GetCompactedIndex(); index != compacted {
 			return errors.Errorf("expected compacted index == %d, got %d", compacted, index)
 		}

pkg/kv/kvserver/raft_log_queue_test.go

@@ -587,7 +587,7 @@ func TestProactiveRaftLogTruncate(t *testing.T) {
 			testutils.SucceedsSoon(t, func() error {
 				if looselyCoupled {
 					// Flush the engine to advance durability, which triggers truncation.
-					require.NoError(t, store.TODOEngine().Flush())
+					require.NoError(t, store.StateEngine().Flush())
 				}
 				if newCompIndex := r.GetCompactedIndex(); newCompIndex <= oldCompIndex {
 					return errors.Errorf("log was not correctly truncated, old compacted index:%d, current:%d",
@@ -884,8 +884,9 @@ func waitForTruncationForTesting(
 ) {
 	testutils.SucceedsSoon(t, func() error {
 		if looselyCoupled {
-			// Flush the engine to advance durability, which triggers truncation.
-			require.NoError(t, r.store.TODOEngine().Flush())
+			// Flush the state machine engine to advance durability, which triggers
+			// truncation.
+			require.NoError(t, r.store.StateEngine().Flush())
 		}
 		// First index should have changed.
 		if firstIndex := r.GetCompactedIndex() + 1; firstIndex != newFirstIndex {

pkg/kv/kvserver/replica_application_state_machine_test.go

@@ -280,7 +280,7 @@ func TestReplicaStateMachineRaftLogTruncationLooselyCoupled(t *testing.T) {
 		looselyCoupledTruncationEnabled.Override(ctx, &st.SV, true)
 		// Remove the flush completed callback since we don't want a
 		// non-deterministic flush to cause the test to fail.
-		tc.store.TODOEngine().RegisterFlushCompletedCallback(func() {})
+		tc.store.StateEngine().RegisterFlushCompletedCallback(func() {})
 		r := tc.repl
 
 		{
@@ -375,7 +375,7 @@ func TestReplicaStateMachineRaftLogTruncationLooselyCoupled(t *testing.T) {
 			require.True(t, trunc.isDeltaTrusted)
 			return raftLogSize, truncatedIndex
 		}()
-		require.NoError(t, tc.store.TODOEngine().Flush())
+		require.NoError(t, tc.store.StateEngine().Flush())
 		// Asynchronous call to advance durability.
 		tc.store.raftTruncator.durabilityAdvancedCallback()
 		expectedSize := raftLogSize - 1

pkg/kv/kvserver/store.go

@@ -2236,7 +2236,7 @@ func (s *Store) Start(ctx context.Context, stopper *stop.Stopper) error {
 	{
 		truncator := s.raftTruncator
 		// When state machine has persisted new RaftAppliedIndex, fire callback.
-		s.TODOEngine().RegisterFlushCompletedCallback(func() {
+		s.StateEngine().RegisterFlushCompletedCallback(func() {
 			truncator.durabilityAdvancedCallback()
 		})
 	}