Fix: snapshot checkpoint flush corner case (#1413)

* Fix: snapshot checkpoint flush corner case

* nit

* Ensure serialization completion before nullifying. Introduce a loop to check `serializationState` and ensure that any concurrent serialization of `oldValue` is completed before setting it to `null` and returning. This prevents concurrent modification of the shallow-copied object contents.

* misc bugfixes

* update version

Author: badrishc

Version.props

@@ -1,6 +1,6 @@
 <Project>
 	<!-- VersionPrefix property for builds and packages -->
 	<PropertyGroup>
-		<VersionPrefix>1.0.86</VersionPrefix>
+		<VersionPrefix>1.0.87</VersionPrefix>
 	</PropertyGroup>
 </Project>

benchmark/Resp.benchmark/ReqGen.cs

@@ -305,7 +305,12 @@ private void RandomString()
             if (numericValue)
             {
                 for (int i = 0; i < valueBuffer.Length; i++)
-                    valueBuffer[i] = ascii_digits[valueRandomGen.Next(ascii_digits.Length)];
+                {
+                    if (i == 0)
+                        valueBuffer[i] = ascii_digits[1 + valueRandomGen.Next(ascii_digits.Length - 1)];
+                    else
+                        valueBuffer[i] = ascii_digits[valueRandomGen.Next(ascii_digits.Length)];
+                }
             }
             else
             {

benchmark/Resp.benchmark/ReqGenUtils.cs

@@ -364,7 +364,7 @@ private bool WriteKey(ref byte* curr, byte* vend, out byte[] keyData)
             else
                 key = Start + keyIndex++;
 
-            keyData = Encoding.ASCII.GetBytes(key.ToString().PadLeft(keyLen, numericValue ? '0' : 'X'));
+            keyData = Encoding.ASCII.GetBytes(key.ToString().PadLeft(keyLen, numericValue ? '1' : 'X'));
             return WriteStringBytes(ref curr, vend, keyData);
         }
 

libs/host/Configuration/Options.cs

@@ -146,7 +146,7 @@ internal sealed class Options : ICloneable
         public bool? ObjectStoreCopyReadsToTail { get; set; }
 
         [LogDirValidation(false, false)]
-        [Option('l', "logdir", Required = false, HelpText = "Storage directory for tiered records (hybrid log), if storage tiering (--storage) is enabled. Uses current directory if unspecified.")]
+        [Option('l', "logdir", Required = false, HelpText = "Storage directory for tiered records (hybrid log), if storage tiering (--storage-tier) is enabled. Uses current directory if unspecified.")]
         public string LogDir { get; set; }
 
         [CheckpointDirValidation(false, false)]

libs/server/Objects/Types/GarnetObjectBase.cs

@@ -86,6 +86,17 @@ public void CopyUpdate(ref IGarnetObject oldValue, ref IGarnetObject newValue, b
             // since the new version of the object is already created.
             if (!isInNewVersion)
             {
+                // Wait for any concurrent ongoing serialization of oldValue to complete
+                while (true)
+                {
+                    if (serializationState == (int)SerializationPhase.REST && MakeTransition(SerializationPhase.REST, SerializationPhase.SERIALIZED))
+                        break;
+
+                    if (serializationState >= (int)SerializationPhase.SERIALIZED)
+                        break;
+
+                    _ = Thread.Yield();
+                }
                 oldValue = null;
                 return;
             }

libs/server/Storage/Functions/MainStore/PrivateMethods.cs

@@ -532,6 +532,7 @@ static void TryCopyUpdateNumber(ref SpanByte oldValue, ref SpanByte newValue, re
                 // Move to tail of the log even when oldValue is alphanumeric
                 // We have already paid the cost of bringing from disk so we are treating as a regular access and bring it into memory
                 oldValue.CopyTo(ref newValue);
+                output.SpanByte.AsSpan()[0] = (byte)OperationError.INVALID_TYPE;
                 return;
             }
 

libs/server/Storage/Functions/MainStore/VarLenInputMethods.cs

@@ -144,7 +144,11 @@ public int GetRMWModifiedValueLength(ref SpanByte t, ref RawStringInput input)
                     case RespCommand.INCRBY:
                         var incrByValue = input.header.cmd == RespCommand.INCRBY ? input.arg1 : 1;
 
-                        var curr = NumUtils.ReadInt64(t.AsSpan(functionsState.etagState.etagOffsetForVarlen));
+                        if (!NumUtils.TryReadInt64(t.AsSpan(functionsState.etagState.etagOffsetForVarlen), out var curr))
+                        {
+                            // Return enough space to copy over old value
+                            return sizeof(int) + t.Length + functionsState.etagState.etagOffsetForVarlen;
+                        }
                         var next = curr + incrByValue;
 
                         var ndigits = NumUtils.CountDigits(next, out var isNegative);

libs/storage/Tsavorite/cs/src/core/Index/Checkpointing/SnapshotCheckpointSMTask.cs

@@ -46,6 +46,12 @@ public override void GlobalBeforeEnteringState(SystemState next, StateMachineDri
                     // If we are using a NullDevice then storage tier is not enabled and FlushedUntilAddress may be ReadOnlyAddress; get all records in memory.
                     store._hybridLogCheckpoint.info.snapshotStartFlushedLogicalAddress = store.hlogBase.IsNullDevice ? store.hlogBase.HeadAddress : store.hlogBase.FlushedUntilAddress;
 
+                    if (store._hybridLogCheckpoint.info.finalLogicalAddress <= store._hybridLogCheckpoint.info.snapshotStartFlushedLogicalAddress)
+                    {
+                        // Nothing to flush because the flushed region already contains everything up to finalLogicalAddress.
+                        break;
+                    }
+
                     long startPage = store.hlogBase.GetPage(store._hybridLogCheckpoint.info.snapshotStartFlushedLogicalAddress);
                     long endPage = store.hlogBase.GetPage(store._hybridLogCheckpoint.info.finalLogicalAddress);
                     if (store._hybridLogCheckpoint.info.finalLogicalAddress >

playground/ClusterStress/ReqGenForCluster.cs

@@ -199,7 +199,7 @@ private byte[] GetClusterKeyInterleaved()
 
             string keyStr = "{" + key.ToString() + "}";
             currKeyInSlot++;
-            return Encoding.ASCII.GetBytes(keyStr.PadRight(keyLen, numericValue ? '0' : 'X'));
+            return Encoding.ASCII.GetBytes(keyStr.PadRight(keyLen, numericValue ? '1' : 'X'));
         }
 
     }