@@ -18,6 +18,7 @@ internal unsafe struct ReplayRecord
1818 {
1919 public byte * entryPtr ;
2020 public int payloadLength ;
21+ public long logAddressSequenceNumber ;
2122 }
2223
2324 internal sealed class ReplicaReplayTask (
@@ -62,13 +63,22 @@ internal async Task FullPageBasedBackgroundReplayAsync()
6263 {
6364 await replayBatchContext . LeaderFollowerBarrier . WaitReadyWorkAsync ( cancellationToken : cts . Token ) . ConfigureAwait ( false ) ;
6465 }
66+ catch ( TaskCanceledException ) when ( cts . Token . IsCancellationRequested )
67+ {
68+ // Suppress the exception if the task was cancelled because of store wrapper disposal
69+ }
6570 catch ( Exception ex )
6671 {
6772 logger ? . LogError ( ex , "{method} failed at WaitAsync" , nameof ( FullPageBasedBackgroundReplayAsync ) ) ;
6873 await cts . CancelAsync ( ) . ConfigureAwait ( false ) ;
6974 break ;
7075 }
7176
77+ // Guard: if cancellation happened during WaitReadyWorkAsync, exit cleanly
78+ // without falling through to the processing block (which would issue a spurious SignalCompleted)
79+ if ( cts . Token . IsCancellationRequested )
80+ break ;
81+
7282 try
7383 {
7484 unsafe
@@ -80,9 +90,6 @@ internal async Task FullPageBasedBackgroundReplayAsync()
8090 var isProtected = replayBatchContext . IsProtected ;
8191 var ptr = record ;
8292
83- var maxSequenceNumber = 0L ;
84-
85- // logger?.LogError("[{sublogIdx},{replayIdx}] = {currentAddress} -> {nextAddress}", sublogIdx, replayIdx, currentAddress, nextAddress);
8693 while ( ptr < record + recordLength )
8794 {
8895 cts . Token . ThrowIfCancellationRequested ( ) ;
@@ -91,25 +98,21 @@ internal async Task FullPageBasedBackgroundReplayAsync()
9198 if ( payloadLength > 0 )
9299 {
93100 var entryPtr = ptr + entryLength ;
94- if ( replicationManager . AofProcessor . CanReplay ( entryPtr , replayTaskIdx , out var sequenceNumber ) )
101+ var logAddressSequenceNumber = currentAddress + ( ptr - record ) ;
102+ if ( replicationManager . AofProcessor . CanReplay ( entryPtr , replayTaskIdx , logAddressSequenceNumber , out _ ) )
95103 {
96- replicationManager . AofProcessor . ProcessAofRecordInternal ( virtualSublogIdx , entryPtr , payloadLength , true , out var isCheckpointStart ) ;
104+ replicationManager . AofProcessor . ProcessAofRecordInternal ( virtualSublogIdx , entryPtr , payloadLength , true , out var isCheckpointStart , logAddressSequenceNumber ) ;
97105 // Encountered checkpoint start marker, log the ReplicationCheckpointStartOffset so we know the correct AOF truncation
98106 // point when we take a checkpoint at the checkpoint end marker
99107 if ( isCheckpointStart )
100108 {
101- // logger?.LogError("[{sublogIdx}] CheckpointStart {address}", sublogIdx, clusterProvider.replicationManager.GetSublogReplicationOffset(sublogIdx));
102109 replicationManager . ReplicationCheckpointStartOffset [ physicalSublogIdx ] = replicationManager . GetSublogReplicationOffset ( physicalSublogIdx ) ;
103110 }
104111 }
105- maxSequenceNumber = Math . Max ( sequenceNumber , maxSequenceNumber ) ;
106112 entryLength += TsavoriteLog . UnsafeAlign ( payloadLength ) ;
107113 }
108114 else if ( payloadLength < 0 )
109115 {
110- if ( ! clusterProvider . serverOptions . EnableFastCommit )
111- throw new GarnetException ( "Received FastCommit request at replica AOF processor, but FastCommit is not enabled" , clientResponse : false ) ;
112-
113116 // Only a single thread should commit metadata
114117 if ( replayTaskIdx == 0 )
115118 {
@@ -122,10 +125,16 @@ internal async Task FullPageBasedBackgroundReplayAsync()
122125 ptr += entryLength ;
123126 }
124127
125- // Update max sequence number for this virtual sublog which is mapped
126- appendOnlyFile . readConsistencyManager . UpdateVirtualSublogMaxSequenceNumber ( virtualSublogIdx , maxSequenceNumber ) ;
128+ // Advance frontier to nextAddress (past all entries in this page).
129+ // This ensures the read consistency protocol (which waits for frontier > sessionSeq)
130+ // can proceed once all writes in the page are complete.
131+ appendOnlyFile . readConsistencyManager . UpdateVirtualSublogMaxSequenceNumber ( virtualSublogIdx , nextAddress ) ;
127132 }
128133 }
134+ catch ( TaskCanceledException ) when ( cts . Token . IsCancellationRequested )
135+ {
136+ // Suppress the exception if the task was cancelled because of store wrapper disposal
137+ }
129138 catch ( Exception ex )
130139 {
131140 logger ? . LogError ( ex , "{method} failed at replaying" , nameof ( FullPageBasedBackgroundReplayAsync ) ) ;
@@ -156,7 +165,7 @@ internal async Task ChannelBasedBackgroundReplayAsync()
156165 {
157166 unsafe
158167 {
159- replicationManager . AofProcessor . ProcessAofRecordInternal ( virtualSublogIdx , record . entryPtr , record . payloadLength , true , out var isCheckpointStart ) ;
168+ replicationManager . AofProcessor . ProcessAofRecordInternal ( virtualSublogIdx , record . entryPtr , record . payloadLength , true , out var isCheckpointStart , record . logAddressSequenceNumber ) ;
160169
161170 // Encountered checkpoint start marker, log the ReplicationCheckpointStartOffset so we know the correct AOF truncation
162171 // point when we take a checkpoint at the checkpoint end marker
0 commit comments