kv: integrate raft async storage writes

pkg/kv/kvserver/apply/doc_test.go

@@ -45,7 +45,7 @@ Setting up a batch of seven log entries:
 	}
 
 	fmt.Println("\nAckCommittedEntriesBeforeApplication:")
-	if err := t.AckCommittedEntriesBeforeApplication(ctx, 10 /* maxIndex */); err != nil {
+	if err := t.AckCommittedEntriesBeforeApplication(ctx); err != nil {
 		panic(err)
 	}
 	fmt.Print(`

pkg/kv/kvserver/apply/task.go

@@ -197,13 +197,7 @@ func (t *Task) assertDecoded() {
 //     it is applied. Because of this, the client can be informed of the success of
 //     a write at this point, but we cannot release that write's latches until the
 //     write has applied. See ProposalData.signalProposalResult/finishApplication.
-//
-//  4. Note that when catching up a follower that is behind, the (etcd/raft)
-//     leader will emit an MsgApp with a commit index that encompasses the entries
-//     in the MsgApp, and Ready() will expose these as present in both the Entries
-//     and CommittedEntries slices (i.e. append and apply). We don't ack these
-//     early - the caller will pass the "old" last index in.
-func (t *Task) AckCommittedEntriesBeforeApplication(ctx context.Context, maxIndex uint64) error {
+func (t *Task) AckCommittedEntriesBeforeApplication(ctx context.Context) error {
 	t.assertDecoded()
 	if !t.anyLocal {
 		return nil // fast-path
@@ -218,11 +212,8 @@ func (t *Task) AckCommittedEntriesBeforeApplication(ctx context.Context, maxInde
 	defer iter.Close()
 
 	// Collect a batch of trivial commands from the applier. Stop at the first
-	// non-trivial command or at the first command with an index above maxIndex.
+	// non-trivial command.
 	batchIter := takeWhileCmdIter(iter, func(cmd Command) bool {
-		if cmd.Index() > maxIndex {
-			return false
-		}
 		return cmd.IsTrivial()
 	})
 

pkg/kv/kvserver/apply/task_test.go

@@ -317,7 +317,7 @@ func TestAckCommittedEntriesBeforeApplication(t *testing.T) {
 	appT := apply.MakeTask(sm, dec)
 	defer appT.Close()
 	require.NoError(t, appT.Decode(ctx, ents))
-	require.NoError(t, appT.AckCommittedEntriesBeforeApplication(ctx, 10 /* maxIndex */))
+	require.NoError(t, appT.AckCommittedEntriesBeforeApplication(ctx))
 
 	// Assert that the state machine was not updated.
 	require.Equal(t, testStateMachine{}, *sm)
@@ -338,40 +338,6 @@ func TestAckCommittedEntriesBeforeApplication(t *testing.T) {
 		require.Equal(t, exp, cmd.acked)
 		require.False(t, cmd.finished)
 	}
-
-	// Try again with a lower maximum log index.
-	appT.Close()
-	ents = makeEntries(5)
-
-	dec = newTestDecoder()
-	dec.nonLocal[2] = true
-	dec.shouldReject[3] = true
-
-	appT = apply.MakeTask(sm, dec)
-	require.NoError(t, appT.Decode(ctx, ents))
-	require.NoError(t, appT.AckCommittedEntriesBeforeApplication(ctx, 4 /* maxIndex */))
-
-	// Assert that the state machine was not updated.
-	require.Equal(t, testStateMachine{}, *sm)
-
-	// Assert that some commands were acknowledged early and that none were finished.
-	for _, cmd := range dec.cmds {
-		var exp bool
-		switch cmd.index {
-		case 1, 4:
-			exp = true // local and successful
-		case 2:
-			exp = false // remote
-		case 3:
-			exp = false // local and rejected
-		case 5:
-			exp = false // index too high
-		default:
-			t.Fatalf("unexpected index %d", cmd.index)
-		}
-		require.Equal(t, exp, cmd.acked)
-		require.False(t, cmd.finished)
-	}
 }
 
 func TestApplyCommittedEntriesWithErr(t *testing.T) {

pkg/kv/kvserver/helpers_test.go

@@ -356,7 +356,7 @@ func (r *Replica) GetRaftLogSize() (int64, bool) {
 func (r *Replica) GetCachedLastTerm() uint64 {
 	r.mu.RLock()
 	defer r.mu.RUnlock()
-	return r.mu.lastTerm
+	return r.mu.lastTermNotDurable
 }
 
 func (r *Replica) IsRaftGroupInitialized() bool {

pkg/kv/kvserver/logstore/BUILD.bazel

@@ -8,6 +8,7 @@ go_library(
         "sideload.go",
         "sideload_disk.go",
         "stateloader.go",
+        "sync_waiter.go",
     ],
     importpath = "github.com/cockroachdb/cockroach/pkg/kv/kvserver/logstore",
     visibility = ["//visibility:public"],
@@ -22,15 +23,18 @@ go_library(
         "//pkg/storage",
         "//pkg/storage/enginepb",
         "//pkg/storage/fs",
+        "//pkg/util/buildutil",
         "//pkg/util/envutil",
         "//pkg/util/hlc",
         "//pkg/util/iterutil",
         "//pkg/util/log",
         "//pkg/util/metric",
         "//pkg/util/protoutil",
+        "//pkg/util/stop",
         "//pkg/util/timeutil",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_errors//oserror",
+        "@com_github_cockroachdb_pebble//record",
         "@com_github_cockroachdb_redact//:redact",
         "@io_etcd_go_raft_v3//:raft",
         "@io_etcd_go_raft_v3//raftpb",
@@ -43,6 +47,7 @@ go_test(
     srcs = [
         "logstore_bench_test.go",
         "sideload_test.go",
+        "sync_waiter_test.go",
     ],
     args = ["-test.timeout=295s"],
     embed = [":logstore"],
@@ -60,6 +65,7 @@ go_test(
         "//pkg/util/log",
         "//pkg/util/metric",
         "//pkg/util/protoutil",
+        "//pkg/util/stop",
         "//pkg/util/tracing",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_errors//oserror",