diff --git a/libs/server/Storage/Functions/MainStore/RMWMethods.cs b/libs/server/Storage/Functions/MainStore/RMWMethods.cs
index 8fd2d585550..4b6e0df350e 100644
--- a/libs/server/Storage/Functions/MainStore/RMWMethods.cs
+++ b/libs/server/Storage/Functions/MainStore/RMWMethods.cs
@@ -911,14 +911,19 @@ public bool NeedCopyUpdate(ref SpanByte key, ref RawStringInput input, ref SpanB
                 case RespCommand.DELIFGREATER:
                     if (rmwInfo.RecordInfo.ETag)
                         EtagState.SetValsForRecordWithEtag(ref functionsState.etagState, ref oldValue);
+
                     long etagFromClient = input.parseState.GetLong(0);
-                    if (etagFromClient <= functionsState.etagState.etag)
+                    if (etagFromClient > functionsState.etagState.etag)
                     {
-                        EtagState.ResetState(ref functionsState.etagState);
-                        return false;
+                        rmwInfo.Action = RMWAction.ExpireAndStop;
                     }
 
-                    return true;
+                    EtagState.ResetState(ref functionsState.etagState);
+                    // We always return false because we would rather not create a new record in hybrid log if we don't need to delete the object.
+                    // Setting no Action and returning false for non-delete case will shortcircuit the InternalRMW code to not run CU, and return SUCCESS.
+                    // If we want to delete the object setting the Action to ExpireAndStop will add the tombstone in hybrid log for us.
+                    return false;
+
                 case RespCommand.SETIFGREATER:
                 case RespCommand.SETIFMATCH:
                     if (rmwInfo.RecordInfo.ETag)
@@ -1046,13 +1051,6 @@ public bool CopyUpdater(ref SpanByte key, ref RawStringInput input, ref SpanByte
 
             switch (cmd)
             {
-                case RespCommand.DELIFGREATER:
-                    // NCU has already checked for making sure the etag is greater than the existing etag by this point
-                    long etagFromClient = input.parseState.GetLong(0);
-                    rmwInfo.Action = RMWAction.ExpireAndStop;
-                    EtagState.ResetState(ref functionsState.etagState);
-                    return false;
-
                 case RespCommand.SETIFGREATER:
                 case RespCommand.SETIFMATCH:
                     // By now the comparison for etag against existing etag has already been done in NeedCopyUpdate
@@ -1074,7 +1072,7 @@ public bool CopyUpdater(ref SpanByte key, ref RawStringInput input, ref SpanByte
                     Span<byte> dest = newValue.AsSpan(EtagConstants.EtagSize);
                     src.CopyTo(dest);
 
-                    etagFromClient = input.parseState.GetLong(1);
+                    long etagFromClient = input.parseState.GetLong(1);
 
                     functionsState.etagState.etag = etagFromClient;
 
diff --git a/libs/storage/Tsavorite/cs/src/core/Allocator/BlittableAllocator.cs b/libs/storage/Tsavorite/cs/src/core/Allocator/BlittableAllocator.cs
index 2d45b275eda..54f48093764 100644
--- a/libs/storage/Tsavorite/cs/src/core/Allocator/BlittableAllocator.cs
+++ b/libs/storage/Tsavorite/cs/src/core/Allocator/BlittableAllocator.cs
@@ -84,6 +84,11 @@ public readonly (int actualSize, int allocatedSize, int keySize) GetRMWCopyDesti
             where TVariableLengthInput : IVariableLengthInput<TValue, TInput>
              => BlittableAllocatorImpl<TKey, TValue, TStoreFunctions>.GetRMWCopyDestinationRecordSize(ref key, ref input, ref value, ref recordInfo, varlenInput);
 
+        /// <inheritdoc/>
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        public (int actualSize, int allocatedSize, int keySize) GetTombstoneRecordSize(ref TKey key)
+            => BlittableAllocatorImpl<TKey, TValue, TStoreFunctions>.GetTombstoneRecordSize(ref key);
+
         /// <inheritdoc/>
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         public readonly int GetRequiredRecordSize(long physicalAddress, int availableBytes) => GetAverageRecordSize();
diff --git a/libs/storage/Tsavorite/cs/src/core/Allocator/BlittableAllocatorImpl.cs b/libs/storage/Tsavorite/cs/src/core/Allocator/BlittableAllocatorImpl.cs
index d60df042f4d..fdca149f000 100644
--- a/libs/storage/Tsavorite/cs/src/core/Allocator/BlittableAllocatorImpl.cs
+++ b/libs/storage/Tsavorite/cs/src/core/Allocator/BlittableAllocatorImpl.cs
@@ -86,6 +86,10 @@ void ReturnPage(int index)
         internal static (int actualSize, int allocatedSize, int keySize) GetRMWCopyDestinationRecordSize<TInput, TVariableLengthInput>(ref TKey key, ref TInput input, ref TValue value, ref RecordInfo recordInfo, TVariableLengthInput varlenInput)
             => (RecordSize, RecordSize, KeySize);
 
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        internal static (int actualSize, int allocatedSize, int keySize) GetTombstoneRecordSize(ref TKey key)
+            => (RecordSize, RecordSize, KeySize);
+
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         public static (int actualSize, int allocatedSize, int keySize) GetRMWInitialRecordSize<TInput, TSessionFunctionsWrapper>(ref TKey key, ref TInput input, TSessionFunctionsWrapper sessionFunctions)
             => (RecordSize, RecordSize, KeySize);
diff --git a/libs/storage/Tsavorite/cs/src/core/Allocator/GenericAllocator.cs b/libs/storage/Tsavorite/cs/src/core/Allocator/GenericAllocator.cs
index 6afdd1b8ed4..bc14667d390 100644
--- a/libs/storage/Tsavorite/cs/src/core/Allocator/GenericAllocator.cs
+++ b/libs/storage/Tsavorite/cs/src/core/Allocator/GenericAllocator.cs
@@ -79,6 +79,11 @@ public readonly (int actualSize, int allocatedSize, int keySize) GetRMWCopyDesti
             where TVariableLengthInput : IVariableLengthInput<TValue, TInput>
              => _this.GetRMWCopyDestinationRecordSize(ref key, ref input, ref value, ref recordInfo, varlenInput);
 
+        /// <inheritdoc />
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        public (int actualSize, int allocatedSize, int keySize) GetTombstoneRecordSize(ref TKey key)
+            => _this.GetTombstoneRecordSize(ref key);
+
         /// <inheritdoc/>
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         public readonly int GetRequiredRecordSize(long physicalAddress, int availableBytes) => GetAverageRecordSize();
diff --git a/libs/storage/Tsavorite/cs/src/core/Allocator/GenericAllocatorImpl.cs b/libs/storage/Tsavorite/cs/src/core/Allocator/GenericAllocatorImpl.cs
index e52fa9d9f9d..05ab2ca74fb 100644
--- a/libs/storage/Tsavorite/cs/src/core/Allocator/GenericAllocatorImpl.cs
+++ b/libs/storage/Tsavorite/cs/src/core/Allocator/GenericAllocatorImpl.cs
@@ -142,6 +142,9 @@ internal ref TValue GetValue(long physicalAddress)
         internal (int actualSize, int allocatedSize, int keySize) GetRMWCopyDestinationRecordSize<TInput, TVariableLengthInput>(ref TKey key, ref TInput input, ref TValue value, ref RecordInfo recordInfo, TVariableLengthInput varlenInput)
             => (RecordSize, RecordSize, KeySize);
 
+        internal (int actualSize, int allocatedSize, int keySize) GetTombstoneRecordSize(ref TKey key)
+            => (RecordSize, RecordSize, KeySize);
+
         internal int GetAverageRecordSize() => RecordSize;
 
         internal int GetFixedRecordSize() => RecordSize;
diff --git a/libs/storage/Tsavorite/cs/src/core/Allocator/IAllocator.cs b/libs/storage/Tsavorite/cs/src/core/Allocator/IAllocator.cs
index f7539b4b0b7..967e115ac6c 100644
--- a/libs/storage/Tsavorite/cs/src/core/Allocator/IAllocator.cs
+++ b/libs/storage/Tsavorite/cs/src/core/Allocator/IAllocator.cs
@@ -38,6 +38,9 @@ AllocatorBase<TKey, TValue, TStoreFunctions, TAllocator> GetBase<TAllocator>()
         /// <summary>Get record size required for the given <paramref name="key"/> and <paramref name="value"/></summary>
         (int actualSize, int allocatedSize, int keySize) GetRecordSize(ref TKey key, ref TValue value);
 
+        /// <summary>Get the record size for a tombstoned record</summary>
+        (int actualSize, int allocatedSize, int keySize) GetTombstoneRecordSize(ref TKey key);
+
         /// <summary>Get the size of the given <paramref name="value"/></summary>
         int GetValueLength(ref TValue value);
 
diff --git a/libs/storage/Tsavorite/cs/src/core/Allocator/SpanByteAllocator.cs b/libs/storage/Tsavorite/cs/src/core/Allocator/SpanByteAllocator.cs
index c2ba87b3795..4949003aae2 100644
--- a/libs/storage/Tsavorite/cs/src/core/Allocator/SpanByteAllocator.cs
+++ b/libs/storage/Tsavorite/cs/src/core/Allocator/SpanByteAllocator.cs
@@ -80,6 +80,9 @@ public readonly (int actualSize, int allocatedSize, int keySize) GetRMWCopyDesti
             where TVariableLengthInput : IVariableLengthInput<SpanByte, TInput>
              => _this.GetRMWCopyDestinationRecordSize(ref key, ref input, ref value, ref recordInfo, varlenInput);
 
+        /// <inheritdoc/>
+        public (int actualSize, int allocatedSize, int keySize) GetTombstoneRecordSize(ref SpanByte key) => _this.GetTombstoneRecordSize(ref key);
+
         /// <inheritdoc/>
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         public readonly int GetRequiredRecordSize(long physicalAddress, int availableBytes) => _this.GetRequiredRecordSize(physicalAddress, availableBytes);
diff --git a/libs/storage/Tsavorite/cs/src/core/Allocator/SpanByteAllocatorImpl.cs b/libs/storage/Tsavorite/cs/src/core/Allocator/SpanByteAllocatorImpl.cs
index f01495c1fdb..307f0a5bc47 100644
--- a/libs/storage/Tsavorite/cs/src/core/Allocator/SpanByteAllocatorImpl.cs
+++ b/libs/storage/Tsavorite/cs/src/core/Allocator/SpanByteAllocatorImpl.cs
@@ -131,6 +131,16 @@ public ref SpanByte GetAndInitializeValue(long physicalAddress, long endAddress)
             return (size, RoundUp(size, Constants.kRecordAlignment), keySize);
         }
 
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        internal (int actualSize, int allocatedSize, int keySize) GetTombstoneRecordSize(ref SpanByte key)
+        {
+            int keySize = key.TotalSize;
+            // Only metadata space needed since this is going to be used for tombstoning anyway.
+            int minAllocationForTombstone = sizeof(int);
+            int size = RecordInfo.GetLength() + RoundUp(keySize, Constants.kRecordAlignment) + minAllocationForTombstone;
+            return (size, RoundUp(size, Constants.kRecordAlignment), keySize);
+        }
+
         public int GetRequiredRecordSize(long physicalAddress, int availableBytes)
         {
             // We need at least [average record size]...
diff --git a/libs/storage/Tsavorite/cs/src/core/ClientSession/SessionFunctionsWrapper.cs b/libs/storage/Tsavorite/cs/src/core/ClientSession/SessionFunctionsWrapper.cs
index 9f3eb138255..49feed18b1a 100644
--- a/libs/storage/Tsavorite/cs/src/core/ClientSession/SessionFunctionsWrapper.cs
+++ b/libs/storage/Tsavorite/cs/src/core/ClientSession/SessionFunctionsWrapper.cs
@@ -101,7 +101,7 @@ public bool PostCopyUpdater(ref TKey key, ref TInput input, ref TValue oldValue,
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         public bool InPlaceUpdater(long physicalAddress, ref TKey key, ref TInput input, ref TValue value, ref TOutput output, ref RMWInfo rmwInfo, out OperationStatus status, ref RecordInfo recordInfo)
         {
-            (rmwInfo.UsedValueLength, rmwInfo.FullValueLength, _) = _clientSession.store.GetRecordLengths(physicalAddress, ref value, ref recordInfo);
+            (rmwInfo.UsedValueLength, rmwInfo.FullValueLength, rmwInfo.FullRecordLength) = _clientSession.store.GetRecordLengths(physicalAddress, ref value, ref recordInfo);
 
             if (_clientSession.functions.InPlaceUpdater(ref key, ref input, ref value, ref output, ref rmwInfo, ref recordInfo))
             {
diff --git a/libs/storage/Tsavorite/cs/src/core/Index/Interfaces/CallbackInfos.cs b/libs/storage/Tsavorite/cs/src/core/Index/Interfaces/CallbackInfos.cs
index bb5c7d7d07c..81b28f81f91 100644
--- a/libs/storage/Tsavorite/cs/src/core/Index/Interfaces/CallbackInfos.cs
+++ b/libs/storage/Tsavorite/cs/src/core/Index/Interfaces/CallbackInfos.cs
@@ -214,6 +214,8 @@ public struct RMWInfo
         /// </summary>
         public int FullValueLength { get; internal set; }
 
+        public int FullRecordLength { get; internal set; }
+
         /// <summary>
         /// If set true by CopyUpdater, the source record for the RCU will not be elided from the tag chain even if this is otherwise possible.
         /// </summary>
diff --git a/libs/storage/Tsavorite/cs/src/core/Index/Tsavorite/Implementation/Helpers.cs b/libs/storage/Tsavorite/cs/src/core/Index/Tsavorite/Implementation/Helpers.cs
index cee1505547a..05b6dd99952 100644
--- a/libs/storage/Tsavorite/cs/src/core/Index/Tsavorite/Implementation/Helpers.cs
+++ b/libs/storage/Tsavorite/cs/src/core/Index/Tsavorite/Implementation/Helpers.cs
@@ -253,5 +253,57 @@ private bool FindTagAndTryTransientSLock<TInput, TOutput, TContext, TSessionFunc
             stackCtx.SetRecordSourceToHashEntry(hlogBase);
             return true;
         }
+
+
+        // Note: We do not currently consider this reuse for mid-chain records (records past the HashBucket), because TracebackForKeyMatch would need
+        //  to return the next-higher record whose .PreviousAddress points to this one, *and* we'd need to make sure that record was not revivified out.
+        //  Also, we do not consider this in-chain reuse for records with different keys, because we don't get here if the keys don't match.
+        private void HandleRecordElision<TInput, TOutput, TContext, TSessionFunctionsWrapper>(
+            TSessionFunctionsWrapper sessionFunctions, ref OperationStackContext<TKey, TValue, TStoreFunctions, TAllocator> stackCtx, ref RecordInfo srcRecordInfo, int usedValueLength, int fullValueLength, int fullRecordLength)
+            where TSessionFunctionsWrapper : ISessionFunctionsWrapper<TKey, TValue, TInput, TOutput, TContext, TStoreFunctions, TAllocator>
+        {
+            if (!RevivificationManager.IsEnabled)
+            {
+                // We are not doing revivification, so we just want to remove the record from the tag chain so we don't potentially do an IO later for key 
+                // traceback. If we succeed, we need to SealAndInvalidate. It's fine if we don't succeed here; this is just tidying up the HashBucket. 
+                if (stackCtx.hei.TryElide())
+                    srcRecordInfo.SealAndInvalidate();
+            }
+            else if (RevivificationManager.UseFreeRecordPool)
+            {
+                // For non-FreeRecordPool revivification, we leave the record in as a normal tombstone so we can revivify it in the chain for the same key.
+                // For FreeRecord Pool we must first Seal here, even if we're using the LockTable, because the Sealed state must survive this Delete() call.
+                // We invalidate it also for checkpoint/recovery consistency (this removes Sealed bit so Scan would enumerate records that are not in any
+                // tag chain--they would be in the freelist if the freelist survived Recovery), but we restore the Valid bit if it is returned to the chain,
+                // which due to epoch protection is guaranteed to be done before the record can be written to disk and violate the "No Invalid records in
+                // tag chain" invariant.
+                srcRecordInfo.SealAndInvalidate();
+
+                bool isElided = false, isAdded = false;
+
+                Debug.Assert(stackCtx.recSrc.LogicalAddress < hlogBase.ReadOnlyAddress || srcRecordInfo.Tombstone, $"Unexpected loss of Tombstone; Record should have been XLocked or SealInvalidated. RecordInfo: {srcRecordInfo.ToString()}");
+
+                (isElided, isAdded) = TryElideAndTransferToFreeList<TInput, TOutput, TContext, TSessionFunctionsWrapper>(sessionFunctions, ref stackCtx, ref srcRecordInfo,
+                                        (usedValueLength, fullValueLength, fullRecordLength));
+
+                if (!isElided)
+                {
+                    // Leave this in the chain as a normal Tombstone; we aren't going to add a new record so we can't leave this one sealed.
+                    srcRecordInfo.UnsealAndValidate();
+                }
+                else if (!isAdded && RevivificationManager.restoreDeletedRecordsIfBinIsFull)
+                {
+                    // The record was not added to the freelist, but was elided. See if we can put it back in as a normal Tombstone. Since we just
+                    // elided it and the elision criteria is that it is the only above-BeginAddress record in the chain, and elision sets the
+                    // HashBucketEntry.word to 0, it means we do not expect any records for this key's tag to exist after the elision. Therefore,
+                    // we can re-insert the record iff the HashBucketEntry's address is <= kTempInvalidAddress.
+                    stackCtx.hei = new(stackCtx.hei.hash);
+                    FindOrCreateTag(ref stackCtx.hei, hlogBase.BeginAddress);
+
+                    if (stackCtx.hei.entry.Address <= Constants.kTempInvalidAddress && stackCtx.hei.TryCAS(stackCtx.recSrc.LogicalAddress))
+                        srcRecordInfo.UnsealAndValidate();
+                }
+            }
+        }
     }
 }
\ No newline at end of file
diff --git a/libs/storage/Tsavorite/cs/src/core/Index/Tsavorite/Implementation/InternalDelete.cs b/libs/storage/Tsavorite/cs/src/core/Index/Tsavorite/Implementation/InternalDelete.cs
index ea977dd681f..218b8c0b822 100644
--- a/libs/storage/Tsavorite/cs/src/core/Index/Tsavorite/Implementation/InternalDelete.cs
+++ b/libs/storage/Tsavorite/cs/src/core/Index/Tsavorite/Implementation/InternalDelete.cs
@@ -119,51 +119,10 @@ internal OperationStatus InternalDelete<TInput, TOutput, TContext, TSessionFunct
 
                         // Try to transfer the record from the tag chain to the free record pool iff previous address points to invalid address.
                         // Otherwise an earlier record for this key could be reachable again.
-                        // Note: We do not currently consider this reuse for mid-chain records (records past the HashBucket), because TracebackForKeyMatch would need
-                        //  to return the next-higher record whose .PreviousAddress points to this one, *and* we'd need to make sure that record was not revivified out.
-                        //  Also, we do not consider this in-chain reuse for records with different keys, because we don't get here if the keys don't match.
                         if (CanElide<TInput, TOutput, TContext, TSessionFunctionsWrapper>(sessionFunctions, ref stackCtx, ref srcRecordInfo))
                         {
-                            if (!RevivificationManager.IsEnabled)
-                            {
-                                // We are not doing revivification, so we just want to remove the record from the tag chain so we don't potentially do an IO later for key 
-                                // traceback. If we succeed, we need to SealAndInvalidate. It's fine if we don't succeed here; this is just tidying up the HashBucket. 
-                                if (stackCtx.hei.TryElide())
-                                    srcRecordInfo.SealAndInvalidate();
-                            }
-                            else if (RevivificationManager.UseFreeRecordPool)
-                            {
-                                // For non-FreeRecordPool revivification, we leave the record in as a normal tombstone so we can revivify it in the chain for the same key.
-                                // For FreeRecord Pool we must first Seal here, even if we're using the LockTable, because the Sealed state must survive this Delete() call.
-                                // We invalidate it also for checkpoint/recovery consistency (this removes Sealed bit so Scan would enumerate records that are not in any
-                                // tag chain--they would be in the freelist if the freelist survived Recovery), but we restore the Valid bit if it is returned to the chain,
-                                // which due to epoch protection is guaranteed to be done before the record can be written to disk and violate the "No Invalid records in
-                                // tag chain" invariant.
-                                srcRecordInfo.SealAndInvalidate();
-
-                                bool isElided = false, isAdded = false;
-                                Debug.Assert(srcRecordInfo.Tombstone, $"Unexpected loss of Tombstone; Record should have been XLocked or SealInvalidated. RecordInfo: {srcRecordInfo.ToString()}");
-                                (isElided, isAdded) = TryElideAndTransferToFreeList<TInput, TOutput, TContext, TSessionFunctionsWrapper>(sessionFunctions, ref stackCtx, ref srcRecordInfo,
-                                                        (deleteInfo.UsedValueLength, deleteInfo.FullValueLength, fullRecordLength));
-
-                                if (!isElided)
-                                {
-                                    // Leave this in the chain as a normal Tombstone; we aren't going to add a new record so we can't leave this one sealed.
-                                    srcRecordInfo.UnsealAndValidate();
-                                }
-                                else if (!isAdded && RevivificationManager.restoreDeletedRecordsIfBinIsFull)
-                                {
-                                    // The record was not added to the freelist, but was elided. See if we can put it back in as a normal Tombstone. Since we just
-                                    // elided it and the elision criteria is that it is the only above-BeginAddress record in the chain, and elision sets the
-                                    // HashBucketEntry.word to 0, it means we do not expect any records for this key's tag to exist after the elision. Therefore,
-                                    // we can re-insert the record iff the HashBucketEntry's address is <= kTempInvalidAddress.
-                                    stackCtx.hei = new(stackCtx.hei.hash);
-                                    FindOrCreateTag(ref stackCtx.hei, hlogBase.BeginAddress);
-
-                                    if (stackCtx.hei.entry.Address <= Constants.kTempInvalidAddress && stackCtx.hei.TryCAS(stackCtx.recSrc.LogicalAddress))
-                                        srcRecordInfo.UnsealAndValidate();
-                                }
-                            }
+                            HandleRecordElision<TInput, TOutput, TContext, TSessionFunctionsWrapper>(
+                                sessionFunctions, ref stackCtx, ref srcRecordInfo, deleteInfo.UsedValueLength, deleteInfo.FullValueLength, fullRecordLength);
                         }
 
                         status = OperationStatusUtils.AdvancedOpCode(OperationStatus.SUCCESS, StatusCode.InPlaceUpdatedRecord);
diff --git a/libs/storage/Tsavorite/cs/src/core/Index/Tsavorite/Implementation/InternalRMW.cs b/libs/storage/Tsavorite/cs/src/core/Index/Tsavorite/Implementation/InternalRMW.cs
index 0de862c3d3d..72c47819d7a 100644
--- a/libs/storage/Tsavorite/cs/src/core/Index/Tsavorite/Implementation/InternalRMW.cs
+++ b/libs/storage/Tsavorite/cs/src/core/Index/Tsavorite/Implementation/InternalRMW.cs
@@ -138,16 +138,34 @@ internal OperationStatus InternalRMW<TInput, TOutput, TContext, TSessionFunction
                     }
 
                     // rmwInfo's lengths are filled in and GetValueLengths and SetLength are called inside InPlaceUpdater.
-                    if (sessionFunctions.InPlaceUpdater(stackCtx.recSrc.PhysicalAddress, ref key, ref input, ref recordValue, ref output, ref rmwInfo, out status, ref srcRecordInfo)
-                        || (rmwInfo.Action == RMWAction.ExpireAndStop))
+                    if (sessionFunctions.InPlaceUpdater(stackCtx.recSrc.PhysicalAddress, ref key, ref input, ref recordValue, ref output, ref rmwInfo, out status, ref srcRecordInfo))
                     {
                         MarkPage(stackCtx.recSrc.LogicalAddress, sessionFunctions.Ctx);
 
+                        pendingContext.recordInfo = srcRecordInfo;
+                        pendingContext.logicalAddress = stackCtx.recSrc.LogicalAddress;
+                        // status has been set by InPlaceUpdater
+                        goto LatchRelease;
+                    }
+
+                    if (rmwInfo.Action == RMWAction.ExpireAndStop)
+                    {
+                        MarkPage(stackCtx.recSrc.LogicalAddress, sessionFunctions.Ctx);
+                        srcRecordInfo.SetDirtyAndModified();
+
                         // ExpireAndStop means to override default Delete handling (which is to go to InitialUpdater) by leaving the tombstoned record as current.
                         // Our SessionFunctionsWrapper.InPlaceUpdater implementation has already reinitialized-in-place or set Tombstone as appropriate and marked the record.
+
+                        // Try to transfer the record from the tag chain to the free record pool iff previous address points to invalid address.
+                        // Otherwise an earlier record for this key could be reachable again.
+                        if (CanElide<TInput, TOutput, TContext, TSessionFunctionsWrapper>(sessionFunctions, ref stackCtx, ref srcRecordInfo))
+                        {
+                            HandleRecordElision<TInput, TOutput, TContext, TSessionFunctionsWrapper>(
+                                sessionFunctions, ref stackCtx, ref srcRecordInfo, rmwInfo.UsedValueLength, rmwInfo.FullValueLength, rmwInfo.FullRecordLength);
+                        }
+
                         pendingContext.recordInfo = srcRecordInfo;
                         pendingContext.logicalAddress = stackCtx.recSrc.LogicalAddress;
-                        // status has been set by InPlaceUpdater
                         goto LatchRelease;
                     }
 
@@ -182,7 +200,6 @@ internal OperationStatus InternalRMW<TInput, TOutput, TContext, TSessionFunction
 
                 // No record exists - drop through to create new record.
                 Debug.Assert(!sessionFunctions.IsManualLocking || LockTable.IsLockedExclusive(ref stackCtx.hei), "A Lockable-session RMW() of an on-disk or non-existent key requires a LockTable lock");
-
             CreateNewRecord:
                 {
                     TValue tempValue = default;
@@ -362,6 +379,7 @@ private OperationStatus CreateNewRecordRMW<TInput, TOutput, TContext, TSessionFu
             where TSessionFunctionsWrapper : ISessionFunctionsWrapper<TKey, TValue, TInput, TOutput, TContext, TStoreFunctions, TAllocator>
         {
             bool forExpiration = false;
+            bool addTombstone = false;
 
         RetryNow:
 
@@ -374,6 +392,12 @@ private OperationStatus CreateNewRecordRMW<TInput, TOutput, TContext, TSessionFu
                 IsFromPending = pendingContext.type != OperationType.NONE,
             };
 
+            AllocateOptions allocOptions = new()
+            {
+                Recycle = true,
+                ElideSourceRecord = stackCtx.recSrc.HasMainLogSrc && CanElide<TInput, TOutput, TContext, TSessionFunctionsWrapper>(sessionFunctions, ref stackCtx, ref srcRecordInfo)
+            };
+
             // Perform Need*
             if (doingCU)
             {
@@ -388,7 +412,21 @@ private OperationStatus CreateNewRecordRMW<TInput, TOutput, TContext, TSessionFu
                         forExpiration = true;
                     }
                     else if (rmwInfo.Action == RMWAction.ExpireAndStop)
-                        return OperationStatus.NOTFOUND;
+                    {
+                        if (allocOptions.ElideSourceRecord)
+                        {
+                            srcRecordInfo.SetTombstone();
+                            srcRecordInfo.SetDirtyAndModified();
+                            var oldRecordLengths = GetRecordLengths(stackCtx.recSrc.PhysicalAddress, ref hlog.GetValue(stackCtx.recSrc.PhysicalAddress), ref srcRecordInfo);
+                            // Elide from hei, and try to either do in-chain tombstoning or free list transfer.
+                            HandleRecordElision<TInput, TOutput, TContext, TSessionFunctionsWrapper>(
+                                sessionFunctions, ref stackCtx, ref srcRecordInfo, oldRecordLengths.usedValueLength, oldRecordLengths.fullValueLength, oldRecordLengths.fullRecordLength);
+                            // no new record created and hash entry is empty now
+                            return OperationStatusUtils.AdvancedOpCode(OperationStatus.SUCCESS, StatusCode.Found | StatusCode.Expired);
+                        }
+                        // otherwise we shall continue down the tombstoning path
+                        addTombstone = true;
+                    }
                     else
                         return OperationStatus.SUCCESS;
                 }
@@ -402,15 +440,19 @@ private OperationStatus CreateNewRecordRMW<TInput, TOutput, TContext, TSessionFu
             }
 
             // Allocate and initialize the new record
-            var (actualSize, allocatedSize, keySize) = doingCU ?
-                stackCtx.recSrc.AllocatorBase._wrapper.GetRMWCopyDestinationRecordSize(ref key, ref input, ref value, ref srcRecordInfo, sessionFunctions) :
-                hlog.GetRMWInitialRecordSize(ref key, ref input, sessionFunctions);
-
-            AllocateOptions allocOptions = new()
+            int actualSize; int allocatedSize; int keySize;
+            if (!addTombstone)
             {
-                Recycle = true,
-                ElideSourceRecord = stackCtx.recSrc.HasMainLogSrc && CanElide<TInput, TOutput, TContext, TSessionFunctionsWrapper>(sessionFunctions, ref stackCtx, ref srcRecordInfo)
-            };
+                (actualSize, allocatedSize, keySize) = doingCU ?
+                    stackCtx.recSrc.AllocatorBase._wrapper.GetRMWCopyDestinationRecordSize(ref key, ref input, ref value, ref srcRecordInfo, sessionFunctions) :
+                    hlog.GetRMWInitialRecordSize(ref key, ref input, sessionFunctions);
+            }
+            else
+            {
+                Debug.Assert(!allocOptions.ElideSourceRecord, "Elidable records going down the deletion via RMW path from NCU should have already been handled." +
+                    "This block only handles NCU requested deletion for unelidable src records.");
+                (actualSize, allocatedSize, keySize) = stackCtx.recSrc.AllocatorBase._wrapper.GetTombstoneRecordSize(ref key);
+            }
 
             if (!TryAllocateRecord(sessionFunctions, ref pendingContext, ref stackCtx, actualSize, ref allocatedSize, keySize, allocOptions,
                     out long newLogicalAddress, out long newPhysicalAddress, out OperationStatus status))
@@ -444,7 +486,7 @@ private OperationStatus CreateNewRecordRMW<TInput, TOutput, TContext, TSessionFu
                     return OperationStatus.NOTFOUND | (forExpiration ? OperationStatus.EXPIRED : OperationStatus.NOTFOUND);
                 }
             }
-            else
+            else if (!addTombstone)
             {
                 if (srcRecordInfo.ETag)
                     newRecordInfo.SetHasETag();
@@ -473,6 +515,10 @@ private OperationStatus CreateNewRecordRMW<TInput, TOutput, TContext, TSessionFu
                 }
                 if (rmwInfo.Action == RMWAction.ExpireAndStop)
                 {
+                    Debug.Assert(!addTombstone, "Should not have gone down RCU if NCU had already requested tombstoning." +
+                        "This block should only handle expiration/tombstoning via RCU.");
+                    addTombstone = true;
+                    newRecordInfo.SetDirtyAndModified();
                     newRecordInfo.SetTombstone();
                     status = OperationStatusUtils.AdvancedOpCode(OperationStatus.SUCCESS, StatusCode.CreatedRecord | StatusCode.Expired);
                     goto DoCAS;
@@ -501,8 +547,17 @@ private OperationStatus CreateNewRecordRMW<TInput, TOutput, TContext, TSessionFu
                 else
                     return OperationStatus.SUCCESS | (forExpiration ? OperationStatus.EXPIRED : OperationStatus.SUCCESS);
             }
+            else
+            {
+                Debug.Assert(!addTombstone, "This block should only be handling tombstoning requests by NCU where the previous record was not elidable.");
+                newRecordInfo.SetDirtyAndModified();
+                newRecordInfo.SetTombstone();
+                status = OperationStatusUtils.AdvancedOpCode(OperationStatus.SUCCESS, StatusCode.CreatedRecord | StatusCode.Expired);
+            }
 
         DoCAS:
+            // The record being cas'd below is going to be the tombstone record in the case of RCU requested tombstone, and NCU tombstoning.
+            // For all other cases this is the new computed record after an RMW.
             // Insert the new record by CAS'ing either directly into the hash entry or splicing into the readcache/mainlog boundary.
             bool success = CASRecordIntoChain(ref key, ref stackCtx, newLogicalAddress, ref newRecordInfo);
             if (success)
@@ -518,8 +573,8 @@ private OperationStatus CreateNewRecordRMW<TInput, TOutput, TContext, TSessionFu
                 }
                 else
                 {
-                    // Else it was a CopyUpdater so call PCU
-                    if (!sessionFunctions.PostCopyUpdater(ref key, ref input, ref value, ref hlog.GetValue(newPhysicalAddress), ref output, ref rmwInfo, ref newRecordInfo))
+                    // Else it was a CopyUpdater so call PCU if tombstoning has not been requested by NCU or CU
+                    if (!addTombstone && !sessionFunctions.PostCopyUpdater(ref key, ref input, ref value, ref hlog.GetValue(newPhysicalAddress), ref output, ref rmwInfo, ref newRecordInfo))
                     {
                         if (rmwInfo.Action == RMWAction.ExpireAndStop)
                         {
diff --git a/libs/storage/Tsavorite/cs/test/RevivificationTests.cs b/libs/storage/Tsavorite/cs/test/RevivificationTests.cs
index 4ec87cdb829..ac965b26ca1 100644
--- a/libs/storage/Tsavorite/cs/test/RevivificationTests.cs
+++ b/libs/storage/Tsavorite/cs/test/RevivificationTests.cs
@@ -460,6 +460,12 @@ internal class RevivificationSpanByteFunctions : SpanByteFunctions<Empty>
             internal int expectedSingleFullValueLength = -1;
             internal int expectedInputLength = InitialLength;
 
+            // used to configurably change RMW behavior to test tombstoning via RMW route.
+            internal bool deleteInIpu = false;
+            internal bool deleteInNCU = false;
+            internal bool deleteInCU = false;
+            internal bool forceSkipIpu = false;
+
             // This is a queue rather than a single value because there may be calls to, for example, ConcurrentWriter with one length
             // followed by SingleWriter with another.
             internal Queue<int> expectedUsedValueLengths = new();
@@ -536,9 +542,28 @@ public override bool InitialUpdater(ref SpanByte key, ref SpanByte input, ref Sp
                 return base.InitialUpdater(ref key, ref input, ref value, ref output, ref rmwInfo, ref recordInfo);
             }
 
+            public override bool NeedCopyUpdate(ref SpanByte key, ref SpanByte input, ref SpanByte oldValue, ref SpanByteAndMemory output, ref RMWInfo rmwInfo)
+            {
+                if (deleteInNCU)
+                {
+                    rmwInfo.Action = RMWAction.ExpireAndStop;
+                    return false;
+                }
+
+                return base.NeedCopyUpdate(ref key, ref input, ref oldValue, ref output, ref rmwInfo);
+            }
+
             public override bool CopyUpdater(ref SpanByte key, ref SpanByte input, ref SpanByte oldValue, ref SpanByte newValue, ref SpanByteAndMemory output, ref RMWInfo rmwInfo, ref RecordInfo recordInfo)
             {
+                if (deleteInCU)
+                {
+                    rmwInfo.Action = RMWAction.ExpireAndStop;
+                    return false;
+                }
+
                 AssertInfoValid(ref rmwInfo);
+
+
                 ClassicAssert.AreEqual(expectedInputLength, input.Length);
 
                 var expectedUsedValueLength = expectedUsedValueLengths.Dequeue();
@@ -561,6 +586,16 @@ public override bool CopyUpdater(ref SpanByte key, ref SpanByte input, ref SpanB
             public override bool InPlaceUpdater(ref SpanByte key, ref SpanByte input, ref SpanByte value, ref SpanByteAndMemory output, ref RMWInfo rmwInfo, ref RecordInfo recordInfo)
             {
                 AssertInfoValid(ref rmwInfo);
+
+                if (forceSkipIpu)
+                    return false;
+
+                if (deleteInIpu)
+                {
+                    rmwInfo.Action = RMWAction.ExpireAndStop;
+                    return false;
+                }
+
                 ClassicAssert.AreEqual(expectedInputLength, input.Length);
                 ClassicAssert.AreEqual(expectedConcurrentDestLength, value.Length);
                 ClassicAssert.AreEqual(expectedConcurrentFullValueLength, rmwInfo.FullValueLength);
@@ -813,10 +848,57 @@ public void SpanByteNoRevivLengthTest([Values(UpdateOp.Upsert, UpdateOp.RMW)] Up
             }
         }
 
+        internal enum DeletionRoutes
+        {
+            DELETE,
+            RMW_IPU,
+            RMW_NCU,
+            RMW_CU
+        }
+
+        private Status DeleteViaRMW(ref SpanByte key, Span<byte> mockInputVec, byte fillByte)
+        {
+            var mockInput = SpanByte.FromPinnedSpan(mockInputVec);
+            mockInputVec.Fill(fillByte);
+            return bContext.RMW(ref key, ref mockInput);
+        }
+
+        private Status PerformDeletion(DeletionRoutes deletionRoute, ref SpanByte key, byte fillByte)
+        {
+            Status status;
+            switch (deletionRoute)
+            {
+                case DeletionRoutes.DELETE:
+                    return bContext.Delete(ref key);
+                case DeletionRoutes.RMW_IPU:
+                    functions.deleteInIpu = true;
+                    Span<byte> mockInputVec = stackalloc byte[InitialLength];
+                    status = DeleteViaRMW(ref key, mockInputVec, fillByte);
+                    functions.deleteInIpu = false;
+                    break;
+                case DeletionRoutes.RMW_NCU:
+                    functions.deleteInNCU = true;
+                    mockInputVec = stackalloc byte[InitialLength];
+                    status = DeleteViaRMW(ref key, mockInputVec, fillByte);
+                    functions.deleteInNCU = false;
+                    break;
+                case DeletionRoutes.RMW_CU:
+                    functions.deleteInCU = true;
+                    mockInputVec = stackalloc byte[InitialLength];
+                    status = DeleteViaRMW(ref key, mockInputVec, fillByte);
+                    functions.deleteInCU = false;
+                    break;
+                default:
+                    throw new Exception("Unhandled deletion logic");
+            }
+
+            return status;
+        }
+
         [Test]
         [Category(RevivificationCategory)]
         [Category(SmokeTestCategory)]
-        public void SpanByteSimpleTest([Values(UpdateOp.Upsert, UpdateOp.RMW)] UpdateOp updateOp)
+        public void SpanByteSimpleTest([Values(UpdateOp.Upsert, UpdateOp.RMW)] UpdateOp updateOp, [Values(DeletionRoutes.DELETE, DeletionRoutes.RMW_IPU)] DeletionRoutes deletionRoute)
         {
             Populate();
 
@@ -828,7 +910,9 @@ public void SpanByteSimpleTest([Values(UpdateOp.Upsert, UpdateOp.RMW)] UpdateOp
             var key = SpanByte.FromPinnedSpan(keyVec);
 
             functions.expectedUsedValueLengths.Enqueue(SpanByteTotalSize(InitialLength));
-            var status = bContext.Delete(ref key);
+
+            Status status = PerformDeletion(deletionRoute, ref key, fillByte);
+
             ClassicAssert.IsTrue(status.Found, status.ToString());
 
             ClassicAssert.AreEqual(tailAddress, store.Log.TailAddress);
@@ -851,6 +935,60 @@ public void SpanByteSimpleTest([Values(UpdateOp.Upsert, UpdateOp.RMW)] UpdateOp
             ClassicAssert.AreEqual(tailAddress, store.Log.TailAddress);
         }
 
+        [Test]
+        [Category(RevivificationCategory)]
+        [Category(SmokeTestCategory)]
+        public void SpanByteDeletionViaRMWRCURevivifiesOriginalRecordAfterTombstoning(
+            [Values(UpdateOp.Upsert, UpdateOp.RMW)] UpdateOp updateOp, [Values(DeletionRoutes.RMW_NCU, DeletionRoutes.RMW_CU)] DeletionRoutes deletionRoute)
+        {
+            Populate();
+
+            Span<byte> keyVec = stackalloc byte[KeyLength];
+            byte fillByte = 42;
+            keyVec.Fill(fillByte);
+            var key = SpanByte.FromPinnedSpan(keyVec);
+
+            functions.expectedUsedValueLengths.Enqueue(SpanByteTotalSize(InitialLength));
+
+            functions.forceSkipIpu = true;
+            var status = PerformDeletion(deletionRoute, ref key, fillByte);
+            functions.forceSkipIpu = false;
+
+            RevivificationTestUtils.WaitForRecords(store, want: true);
+
+            ClassicAssert.AreEqual(1, RevivificationTestUtils.GetFreeRecordCount(store));
+
+            ClassicAssert.IsTrue(status.Found, status.ToString());
+
+            var tailAddress = store.Log.TailAddress;
+
+            Span<byte> inputVec = stackalloc byte[InitialLength];
+            var input = SpanByte.FromPinnedSpan(inputVec);
+            inputVec.Fill(fillByte);
+
+            SpanByteAndMemory output = new();
+
+            functions.expectedInputLength = InitialLength;
+            functions.expectedSingleDestLength = InitialLength;
+            functions.expectedConcurrentDestLength = InitialLength;
+            functions.expectedSingleFullValueLength = functions.expectedConcurrentFullValueLength = RoundUpSpanByteFullValueLength(input);
+            functions.expectedUsedValueLengths.Enqueue(SpanByteTotalSize(InitialLength));
+
+            // brand new value so we try to use a record out of free list
+            keyVec = stackalloc byte[KeyLength];
+            fillByte = 255;
+            keyVec.Fill(fillByte);
+            key = SpanByte.FromPinnedSpan(keyVec);
+
+            _ = updateOp == UpdateOp.Upsert ? bContext.Upsert(ref key, ref input, ref input, ref output) : bContext.RMW(ref key, ref input);
+
+            // since above would use revivification free list we should see no change of tail address.
+            ClassicAssert.AreEqual(store.Log.TailAddress, tailAddress);
+            ClassicAssert.AreEqual(tailAddress, store.Log.TailAddress);
+            ClassicAssert.AreEqual(0, RevivificationTestUtils.GetFreeRecordCount(store));
+            output.Memory?.Dispose();
+        }
+
         [Test]
         [Category(RevivificationCategory)]
         [Category(SmokeTestCategory)]