(2.12) Async docs

Signed-off-by: Maurice van Veen <[email protected]>

Author: MauriceVanVeen

server/filestore.go

@@ -4206,6 +4206,7 @@ func (fs *fileStore) storeRawMsg(subj string, hdr, msg []byte, seq uint64, ts, t
 }
 
 // StoreRawMsg stores a raw message with expected sequence number and timestamp.
+// The ceIndex reflects the index of the entry in the WAL, used for signaling when it's persisted.
 func (fs *fileStore) StoreRawMsg(subj string, hdr, msg []byte, seq uint64, ts, ttl int64, ceIndex uint64) error {
 	fs.mu.Lock()
 	err := fs.storeRawMsg(subj, hdr, msg, seq, ts, ttl, ceIndex)
@@ -4288,6 +4289,7 @@ func (mb *msgBlock) skipMsg(seq uint64, now time.Time, ceIndex uint64) {
 }
 
 // SkipMsg will use the next sequence number but not store anything.
+// The ceIndex reflects the index of the entry in the WAL, used for signaling when it's persisted.
 func (fs *fileStore) SkipMsg(ceIndex uint64) uint64 {
 	fs.mu.Lock()
 	defer fs.mu.Unlock()

server/raft.go

@@ -44,8 +44,8 @@ type RaftNode interface {
 	SendSnapshot(snap []byte) error
 	NeedSnapshot() bool
 	Applied(index uint64) (entries uint64, bytes uint64)
-	TrackWrite(index uint64)
-	WritePersisted(index uint64)
+	ApplyWritePending(index uint64)
+	ApplyWritePersisted(index uint64)
 	State() RaftState
 	Size() (entries, bytes uint64)
 	Progress() (index, commit, applied uint64)
@@ -1126,42 +1126,49 @@ func (n *raft) appliedLocked(index uint64) {
 	}
 }
 
-// TrackWrite signals writes need to happen for the specified index.
-// Applied can still be called, but will not move up until WritePersisted
-// is called, signaling writes were persisted.
-func (n *raft) TrackWrite(index uint64) {
+// ApplyWritePending signals writes need to happen for the specified index,
+// i.e. writes are done asynchronously and might take some time for them to hit the disk.
+// Applied can still be called after ApplyWritePending if there are any async writes.
+// But applied will not move up until ApplyWritePersisted is called, signaling writes
+// were persisted.
+// If there are async writes ApplyWritePending MUST ALWAYS be called before Applied.
+// This ensures Applied can be blocked from moving up until the writes are fully persisted.
+// If ApplyWritePending is not called, for example when all writes are persisted by default,
+// Applied will be able to freely move up.
+func (n *raft) ApplyWritePending(index uint64) {
 	n.Lock()
 	defer n.Unlock()
-	n.trackWritesLocked(index, false)
+	n.applyWritesLocked(index, false)
 }
 
-// WritePersisted signals all writes up to and including the index were persisted.
-// Applied automatically moves up if it was called in the meantime.
-func (n *raft) WritePersisted(index uint64) {
+// ApplyWritePersisted signals all writes up to and including the index were persisted,
+// this can be called either before Applied or after as long as the writes were persisted.
+// ApplyWritePending MUST be called before this for every index that requires persisting.
+// Applied will have been called in the meantime, and it will be moved up either:
+// - up to the point where we're waiting for other writes to be persisted
+// - or all writes are persisted, and it can fully move up.
+func (n *raft) ApplyWritePersisted(index uint64) {
 	n.Lock()
 	defer n.Unlock()
-	n.trackWritesLocked(index, true)
+	n.applyWritesLocked(index, true)
 }
 
-// trackWritesLocked keeps track of pending and persisted writes.
-func (n *raft) trackWritesLocked(index uint64, persisted bool) {
+// applyWritesLocked keeps track of pending and persisted writes.
+func (n *raft) applyWritesLocked(index uint64, persisted bool) {
 	if persisted {
 		// Persisted moves the floor up.
+		// This indicates all writes up to and including this floor were persisted.
 		n.persistFloor = max(n.persistFloor, index)
-	} else if n.persistFloor > 0 {
-		// Track the lowest to-be-persisted index.
-		n.persistFloor = min(n.persistFloor, index)
-	} else {
+		n.persistFloorApplied = true
+	} else if n.persistFloor == 0 {
 		// Initializes if not set.
 		n.persistFloor = index
 	}
 	// Track the highest observed pending write.
 	n.persistHigh = max(n.persistHigh, index)
-	// Track whether the floor was persisted.
-	n.persistFloorApplied = n.persistFloorApplied || persisted
 
 	if n.persistFloorApplied {
-		// Move applied up partially.
+		// If the floor is persisted, we can move applied up if possible.
 		if n.persistFloor != n.persistHigh {
 			n.appliedLocked(min(n.persistFloor, n.pendingApplied))
 			return

server/raft_test.go

@@ -2759,36 +2759,36 @@ func TestNRGPendingAppliedPermutations(t *testing.T) {
 	t.Run("ApplyOnDone", func(t *testing.T) {
 		defer reset()
 		// Call applied after.
-		n.TrackWrite(1)
-		n.WritePersisted(1)
+		n.ApplyWritePending(1)
+		n.ApplyWritePersisted(1)
 		n.Applied(10)
 		require_Equal(t, n.applied, 10)
 
 		// Call applied before.
-		n.TrackWrite(11)
+		n.ApplyWritePending(11)
 		n.Applied(20)
 		require_Equal(t, n.applied, 10)
-		n.WritePersisted(11)
+		n.ApplyWritePersisted(11)
 		require_Equal(t, n.applied, 20)
 	})
 
 	// Applied can't move up until all pending entries referencing the same index are applied.
 	t.Run("ApplyWhenNotPending", func(t *testing.T) {
 		defer reset()
-		n.TrackWrite(1)
-		n.TrackWrite(2)
+		n.ApplyWritePending(1)
+		n.ApplyWritePending(2)
 		n.Applied(3)
 		require_Equal(t, n.applied, 0)
-		n.WritePersisted(1)
+		n.ApplyWritePersisted(1)
 		require_Equal(t, n.applied, 1)
-		n.WritePersisted(2)
+		n.ApplyWritePersisted(2)
 		require_Equal(t, n.applied, 3)
-		n.TrackWrite(10)
+		n.ApplyWritePending(10)
 		n.Applied(10)
 		require_Equal(t, n.applied, 9)
 
-		// Apply is unblocked once the last pending is applied.
-		n.WritePersisted(10)
+		// Apply is unblocked once the last pending is persisted.
+		n.ApplyWritePersisted(10)
 		require_Equal(t, n.applied, 10)
 		n.Applied(11)
 		require_Equal(t, n.applied, 11)
@@ -2797,10 +2797,10 @@ func TestNRGPendingAppliedPermutations(t *testing.T) {
 	// Pending state must be reset if all is done.
 	t.Run("DonePendingResetsState", func(t *testing.T) {
 		defer reset()
-		n.TrackWrite(1)
+		n.ApplyWritePending(1)
 		n.Applied(1)
 		require_Equal(t, n.applied, 0)
-		n.WritePersisted(1)
+		n.ApplyWritePersisted(1)
 		require_Equal(t, n.applied, 1)
 		require_Equal(t, n.persistFloor, 0)
 		require_Equal(t, n.persistHigh, 0)

server/store.go

@@ -88,7 +88,7 @@ type StorageUpdateHandler func(msgs, bytes int64, seq uint64, subj string)
 type StorageRemoveMsgHandler func(seq uint64)
 
 // Used to call back into the upper layers that a write is scheduled for a given replication index.
-// Below handler will be called once this data is persisted.
+// Below StorageWritePersistedHandler will be called once this data is persisted.
 type StorageTrackWriteHandler func(index uint64)
 
 // Used to call back into the upper layers that certain writes were persisted up to

server/stream.go

@@ -914,7 +914,7 @@ func (mset *stream) setStreamAssignment(sa *streamAssignment) {
 	}
 	// Update store callbacks that call into the Raft node (if any).
 	if node != nil {
-		mset.store.RegisterStorageTrackWrites(node.TrackWrite, node.WritePersisted)
+		mset.store.RegisterStorageTrackWrites(node.ApplyWritePending, node.ApplyWritePersisted)
 	} else {
 		mset.store.RegisterStorageTrackWrites(nil, nil)
 	}
@@ -4853,6 +4853,7 @@ var (
 )
 
 // processJetStreamMsg is where we try to actually process the stream msg.
+// The ceIndex reflects the index of the entry in the WAL, used for signaling when it's persisted.
 func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte, lseq uint64, ts int64, mt *msgTrace, sourced bool, ceIndex uint64) (retErr error) {
 	if mt != nil {
 		// Only the leader/standalone will have mt!=nil. On exit, send the