Repository: microsoft/garnet
File: libs/storage/Tsavorite/cs/src/core/Index/Tsavorite/Implementation/InternalRMW.cs (C#)

Code before:
                Version = sessionFunctions.Ctx.version,
                SessionID = sessionFunctions.Ctx.sessionID,
                Address = doingCU && !stackCtx.recSrc.HasReadCacheSrc ? stackCtx.recSrc.LogicalAddress : Constants.kInvalidAddress,
                KeyHash = stackCtx.hei.hash,
                IsFromPending = pendingContext.type != OperationType.NONE,
            };

            // Perform Need*
            if (doingCU)
            {
                rmwInfo.SetRecordInfo(ref srcRecordInfo);
                if (!sessionFunctions.NeedCopyUpdate(ref key, ref input, ref value, ref output, ref rmwInfo))
                {
                    if (rmwInfo.Action == RMWAction.CancelOperation)
                        return OperationStatus.CANCELED;
                    else if (rmwInfo.Action == RMWAction.ExpireAndResume)
                    {
                        doingCU = false;
                        forExpiration = true;
                    }
                    else if (rmwInfo.Action == RMWAction.ExpireAndStop)
                    {
                        MarkPage(stackCtx.recSrc.LogicalAddress, sessionFunctions.Ctx);

                        // Handle potential record revivification for records in in-memory region. Handle readonly region and potentially Mutable region undergoing CPR. 
                        if (stackCtx.recSrc.HasMainLogSrc)
                        {
                            // HK TODO: Ask @TedHart if record should be marked dirty and modified here?
                            srcRecordInfo.SetTombstone();
                            // 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))
                            {
                                (rmwInfo.UsedValueLength, rmwInfo.FullValueLength, rmwInfo.FullRecordLength) = GetRecordLengths(stackCtx.recSrc.PhysicalAddress, ref value, ref srcRecordInfo);
                                HandleRecordElision<TInput, TOutput, TContext, TSessionFunctionsWrapper>(
                                    sessionFunctions, ref stackCtx, ref srcRecordInfo, rmwInfo.UsedValueLength, rmwInfo.FullValueLength, rmwInfo.FullRecordLength);
                            }
                        }

                        return OperationStatus.NOTFOUND;
                    }
                    else
                        return OperationStatus.SUCCESS;
                }
            }

            if (!doingCU)
            {
                rmwInfo.ClearRecordInfo();   // There is no existing record
                if (!sessionFunctions.NeedInitialUpdate(ref key, ref input, ref output, ref rmwInfo))
                    return rmwInfo.Action == RMWAction.CancelOperation ? OperationStatus.CANCELED : OperationStatus.NOTFOUND;

Review comment (bug):
If NCU wants to expire and stop at this point, we are not adding a tombstone. The caller will see the record as deleted from the interface pov. This does not incur the cost of a new record either.

Not sure if there is a potential edge case I am missing where this would cause an issue.

Code after:
                Version = sessionFunctions.Ctx.version,
                SessionID = sessionFunctions.Ctx.sessionID,
                Address = doingCU && !stackCtx.recSrc.HasReadCacheSrc ? stackCtx.recSrc.LogicalAddress : Constants.kInvalidAddress,
                KeyHash = stackCtx.hei.hash,
                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)
            {
                rmwInfo.SetRecordInfo(ref srcRecordInfo);
                if (!sessionFunctions.NeedCopyUpdate(ref key, ref input, ref value, ref output, ref rmwInfo))
                {
                    if (rmwInfo.Action == RMWAction.CancelOperation)
                        return OperationStatus.CANCELED;
                    else if (rmwInfo.Action == RMWAction.ExpireAndResume)
                    {
                        doingCU = false;
                        forExpiration = true;
                    }
                    else if (rmwInfo.Action == RMWAction.ExpireAndStop)
                    {
                        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;
                }
            }

            if (!doingCU)
            {
                rmwInfo.ClearRecordInfo();   // There is no existing record
                if (!sessionFunctions.NeedInitialUpdate(ref key, ref input, ref output, ref rmwInfo))