Implement (most of) the rest of redis.* properties and methods for Lu…

…a scripts (#1050)

* implement redis.pcall

* implement redis.sha1hex

* half implement redis.log ; needs a config option, probably

* add a custom error for redis.set_repl; that just cannot work in Garnet

* add errors for redis.debug and redis.breakpoint

* implementation of redis.acl_check_cmd

* sort of implement setresp

* implement redis.REDIS_VERSION and redis.REDIS_VERSION_NUM

* formatting

* prevent declarations of globals in Lua scripts

* fix tests

* add LuaLoggingMode option; redis.log(...) now gets recorded, but can be disabled (either silently, or raising and error) if desired

* fix benchmarks

* fix benchmarks

* Fix SET KEEPTTL return string (#1039)

* SETKEEPTTL (no other flags) should always return ok. It doesn't care about NX or XX

* Simplify GETSET

---------

Co-authored-by: prvyk <[email protected]>
Co-authored-by: Tal Zaccai <[email protected]>

* Support for Diskless Replication (#997)

* expose diskless replication parameters

* refactor/cleanup legacy ReplicaSyncSession

* add interface to support diskless replication session and aof tasks

* core diskless replication implementation

* expose diskless replication API

* adding test for diskless replication

* update gcs extension to clearly mark logging progress

* fix gcs dispose on diskless attach, call dispose of replicationSyncManager, add more logging

* complete first diskless replication test

* fix iterator check for null when empty store

* fix iterator for object store cluster sync

* add simple diskless sync test

* cleanup code

* replica fall behind test

* wip

* register cts at wait for sync completion

* add db version alignment test

* avoid using close lock for leader based syncing

* truncate AOF after streaming checkpoint is taken

* add tests for failover with diskless replication

* fix formatting and conversion to IPEndpoint

* fix RepCommandsTests

* dispose aofSyncTask if failed to add to AofSyncTaskStore

* overload dispose ReplicaSyncSession

* explicitly dispose gcs used for full sync at replicaSyncSession sync

* dispose gcs once on return

* code cleanup

* update tests to provide more context logging

* add more comprehensive logging of syncMetadata

* add timeout for streaming checkpoint

* add clusterTimeout for diskless repl tests

* some more logging

* cleanup and refactor code

* truncate AOF only when main-memory-replication is switched on

* adding logging for cancellation when streaming

* split checkpoint commit marker to allow for disk checkpoints

* update sync metadata log message

* add progress based timeout implementation

* deprecate main-memory-replication

* formatting

* prevent declarations of globals in Lua scripts

* fix tests

* add LuaLoggingMode option; redis.log(...) now gets recorded, but can be disabled (either silently, or raising and error) if desired

* fix benchmarks

* fix benchmarks

* bit of sandbox_env cleanup

* Allow redis.setresp(3); this allows a mix of RESP(2|3) for connections and inner sessions.
Lots of testing, comparing against Redis.
Fixes RESP3 behavior of HGETALL, SDIFF, ZSCORE (more fixes for other commands needed in separate PR).

* fix null ref when there's not underlying session

* limit size of log in LuaScriptTests

* cleanup this constant string registry stuff

* convert a couple PushBuffers to PushConstantStrings

* serializing logger access in LuaScriptTests

* address feedback

* remove more bogus usings

---------

Co-authored-by: prvyk <[email protected]>
Co-authored-by: prvyk <[email protected]>
Co-authored-by: Tal Zaccai <[email protected]>
Co-authored-by: Vasileios Zois <[email protected]>

benchmark/BDN.benchmark/Lua/LuaParams.cs

@@ -29,7 +29,7 @@ public LuaParams(LuaMemoryManagementMode mode, bool memoryLimit, TimeSpan? timeo
         /// Get the equivalent <see cref="LuaOptions"/>.
         /// </summary>
         public LuaOptions CreateOptions()
-        => new(Mode, MemoryLimit ? "2m" : "", Timeout ?? System.Threading.Timeout.InfiniteTimeSpan);
+        => new(Mode, MemoryLimit ? "2m" : "", Timeout ?? System.Threading.Timeout.InfiniteTimeSpan, LuaLoggingMode.Enable);
 
         /// <summary>
         /// String representation

benchmark/BDN.benchmark/Operations/OperationsBase.cs

@@ -56,7 +56,7 @@ public virtual void GlobalSetup()
                 QuietMode = true,
                 EnableLua = true,
                 DisablePubSub = true,
-                LuaOptions = new(LuaMemoryManagementMode.Native, "", Timeout.InfiniteTimeSpan),
+                LuaOptions = new(LuaMemoryManagementMode.Native, "", Timeout.InfiniteTimeSpan, LuaLoggingMode.Enable),
             };
 
             if (Params.useAof)

libs/common/RespReadUtils.cs

@@ -304,13 +304,13 @@ public static bool TryReadInt32Safe(ref byte* ptr, byte* end, out int value, out
         /// <param name="length">If parsing was successful, contains the extracted length from the header.</param>
         /// <param name="ptr">The starting position in the RESP string. Will be advanced if parsing is successful.</param>
         /// <param name="end">The current end of the RESP string.</param>
-        /// <param name="isArray">Whether to parse an array length header ('*...\r\n') or a string length header ('$...\r\n').</param>
+        /// <param name="expectedSigil">Expected type of RESP header, defaults to string ('$').</param>
         /// <returns>True if a length header was successfully read.</returns>
         /// <exception cref="RespParsingException">Thrown if the length header is negative.</exception>
         /// <exception cref="RespParsingException">Thrown if unexpected token is read.</exception>
         /// <exception cref="RespParsingException">Thrown if integer overflow occurs.</exception>
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
-        public static bool TryReadUnsignedLengthHeader(out int length, ref byte* ptr, byte* end, bool isArray = false)
+        public static bool TryReadUnsignedLengthHeader(out int length, ref byte* ptr, byte* end, char expectedSigil = '$')
         {
             length = -1;
             if (ptr + 3 > end)
@@ -324,7 +324,7 @@ public static bool TryReadUnsignedLengthHeader(out int length, ref byte* ptr, by
                 RespParsingException.ThrowInvalidStringLength(length);
             }
 
-            if (!TryReadSignedLengthHeader(out length, ref ptr, end, isArray))
+            if (!TryReadSignedLengthHeader(out length, ref ptr, end, expectedSigil))
                 return false;
 
             return true;
@@ -342,12 +342,12 @@ public static bool TryReadUnsignedLengthHeader(out int length, ref byte* ptr, by
         /// <param name="length">If parsing was successful, contains the extracted length from the header.</param>
         /// <param name="ptr">The starting position in the RESP string. Will be advanced if parsing is successful.</param>
         /// <param name="end">The current end of the RESP string.</param>
-        /// <param name="isArray">Whether to parse an array length header ('*...\r\n') or a string length header ('$...\r\n').</param>
+        /// <param name="expectedSigil">Expected type of RESP header, defaults to string ('$').</param>
         /// <returns>True if a length header was successfully read.</returns>
         /// <exception cref="RespParsingException">Thrown if unexpected token is read.</exception>
         /// <exception cref="RespParsingException">Thrown if integer overflow occurs.</exception>
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
-        public static bool TryReadSignedLengthHeader(out int length, ref byte* ptr, byte* end, bool isArray = false)
+        public static bool TryReadSignedLengthHeader(out int length, ref byte* ptr, byte* end, char expectedSigil = '$')
         {
             length = -1;
             if (ptr + 3 > end)
@@ -357,7 +357,7 @@ public static bool TryReadSignedLengthHeader(out int length, ref byte* ptr, byte
             var negative = *readHead == '-';
 
             // String length headers must start with a '$', array headers with '*'
-            if (*ptr != (isArray ? '*' : '$'))
+            if (*ptr != expectedSigil)
             {
                 RespParsingException.ThrowUnexpectedToken(*ptr);
             }
@@ -486,7 +486,7 @@ public static bool TryReadInt64(out long number, ref byte* ptr, byte* end, out b
         /// <exception cref="RespParsingException">Thrown if unexpected token is read.</exception>
         /// <exception cref="RespParsingException">Thrown if integer overflow occurs.</exception>
         public static bool TryReadUnsignedArrayLength(out int length, ref byte* ptr, byte* end)
-            => TryReadUnsignedLengthHeader(out length, ref ptr, end, isArray: true);
+            => TryReadUnsignedLengthHeader(out length, ref ptr, end, expectedSigil: '*');
 
         /// <summary>
         /// Tries to read a RESP array length header from the given ASCII-encoded RESP string
@@ -498,11 +498,55 @@ public static bool TryReadUnsignedArrayLength(out int length, ref byte* ptr, byt
         /// <param name="ptr">The starting position in the RESP string. Will be advanced if parsing is successful.</param>
         /// <param name="end">The current end of the RESP string.</param>
         /// <returns>True if a length header was successfully read.</returns>
-        /// <exception cref="RespParsingException">Thrown if the length header is negative.</exception>
         /// <exception cref="RespParsingException">Thrown if unexpected token is read.</exception>
         /// <exception cref="RespParsingException">Thrown if integer overflow occurs.</exception>
         public static bool TryReadSignedArrayLength(out int length, ref byte* ptr, byte* end)
-            => TryReadSignedLengthHeader(out length, ref ptr, end, isArray: true);
+            => TryReadSignedLengthHeader(out length, ref ptr, end, expectedSigil: '*');
+
+        /// <summary>
+        /// Tries to read a RESP3 map length header from the given ASCII-encoded RESP string
+        /// and, if successful, moves the given ptr to the end of the length header.
+        /// <para />
+        /// NOTE: It will not throw an exception if length header is negative.
+        /// </summary>
+        /// <param name="length">If parsing was successful, contains the extracted length from the header.</param>
+        /// <param name="ptr">The starting position in the RESP string. Will be advanced if parsing is successful.</param>
+        /// <param name="end">The current end of the RESP string.</param>
+        /// <returns>True if a length header was successfully read.</returns>
+        /// <exception cref="RespParsingException">Thrown if unexpected token is read.</exception>
+        /// <exception cref="RespParsingException">Thrown if integer overflow occurs.</exception>
+        public static bool TryReadSignedMapLength(out int length, ref byte* ptr, byte* end)
+            => TryReadSignedLengthHeader(out length, ref ptr, end, expectedSigil: '%');
+
+        /// <summary>
+        /// Tries to read a RESP3 set length header from the given ASCII-encoded RESP string
+        /// and, if successful, moves the given ptr to the end of the length header.
+        /// <para />
+        /// NOTE: It will not throw an exception if length header is negative.
+        /// </summary>
+        /// <param name="length">If parsing was successful, contains the extracted length from the header.</param>
+        /// <param name="ptr">The starting position in the RESP string. Will be advanced if parsing is successful.</param>
+        /// <param name="end">The current end of the RESP string.</param>
+        /// <returns>True if a length header was successfully read.</returns>
+        /// <exception cref="RespParsingException">Thrown if unexpected token is read.</exception>
+        /// <exception cref="RespParsingException">Thrown if integer overflow occurs.</exception>
+        public static bool TryReadSignedSetLength(out int length, ref byte* ptr, byte* end)
+            => TryReadSignedLengthHeader(out length, ref ptr, end, expectedSigil: '~');
+
+        /// <summary>
+        /// Tries to read a RESP3 verbatim string length header from the given ASCII-encoded RESP string
+        /// and, if successful, moves the given ptr to the end of the length header.
+        /// <para />
+        /// NOTE: It will not throw an exception if length header is negative.
+        /// </summary>
+        /// <param name="length">If parsing was successful, contains the extracted length from the header.</param>
+        /// <param name="ptr">The starting position in the RESP string. Will be advanced if parsing is successful.</param>
+        /// <param name="end">The current end of the RESP string.</param>
+        /// <returns>True if a length header was successfully read.</returns>
+        /// <exception cref="RespParsingException">Thrown if unexpected token is read.</exception>
+        /// <exception cref="RespParsingException">Thrown if integer overflow occurs.</exception>
+        public static bool TryReadVerbatimStringLength(out int length, ref byte* ptr, byte* end)
+            => TryReadSignedLengthHeader(out length, ref ptr, end, expectedSigil: '=');
 
         /// <summary>
         /// Reads a signed 32-bit integer with length header

libs/common/RespWriteUtils.cs

@@ -79,6 +79,21 @@ public static bool TryWriteArrayLength(int len, ref byte* curr, byte* end)
             return true;
         }
 
+        /// <summary>
+        /// Writes an set length
+        /// </summary>
+        public static bool TryWriteSetLength(int len, ref byte* curr, byte* end)
+        {
+            var numDigits = NumUtils.CountDigits(len);
+            var totalLen = 1 + numDigits + 2;
+            if (totalLen > (int)(end - curr))
+                return false;
+            *curr++ = (byte)'~';
+            NumUtils.WriteInt32(len, numDigits, ref curr);
+            WriteNewline(ref curr);
+            return true;
+        }
+
         public static bool TryWriteArrayLength(int len, ref byte* curr, byte* end, out int numDigits, out int totalLen)
         {
             numDigits = NumUtils.CountDigits(len);
@@ -115,7 +130,7 @@ public static bool TryWriteArrayItem(long integer, ref byte* curr, byte* end)
         }
 
         /// <summary>
-        /// Writes a null
+        /// Writes a RESP2 null ($-1\r\n)
         /// </summary>
         public static bool TryWriteNull(ref byte* curr, byte* end)
         {
@@ -127,6 +142,19 @@ public static bool TryWriteNull(ref byte* curr, byte* end)
             return true;
         }
 
+        /// <summary>
+        /// Writes a RESP3 null (_\r\n)
+        /// </summary>
+        public static bool TryWriteResp3Null(ref byte* curr, byte* end)
+        {
+            if (3 > (int)(end - curr))
+                return false;
+
+            *curr++ = (byte)'_';
+            WriteBytes<ushort>(ref curr, "\r\n"u8);
+            return true;
+        }
+
         /// <summary>
         /// Writes a null array
         /// </summary>
@@ -674,6 +702,42 @@ public static bool TryWriteEmptyArray(ref byte* curr, byte* end)
             return true;
         }
 
+        /// <summary>
+        /// Write empty set
+        /// </summary>
+        public static bool TryWriteEmptySet(ref byte* curr, byte* end)
+        {
+            if (4 > (int)(end - curr))
+                return false;
+
+            WriteBytes<uint>(ref curr, "~0\r\n"u8);
+            return true;
+        }
+
+        /// <summary>
+        /// Write RESP3 true
+        /// </summary>
+        public static bool TryWriteTrue(ref byte* curr, byte* end)
+        {
+            if (4 > (int)(end - curr))
+                return false;
+
+            WriteBytes<uint>(ref curr, "#t\r\n"u8);
+            return true;
+        }
+
+        /// <summary>
+        /// Write RESP3 false
+        /// </summary>
+        public static bool TryWriteFalse(ref byte* curr, byte* end)
+        {
+            if (4 > (int)(end - curr))
+                return false;
+
+            WriteBytes<uint>(ref curr, "#f\r\n"u8);
+            return true;
+        }
+
         /// <summary>
         /// Write an array with len number of null elements
         /// </summary>

libs/host/Configuration/Options.cs

@@ -556,11 +556,14 @@ internal sealed class Options
         [Option("lua-script-timeout", Default = null, Required = false, HelpText = "Timeout for a Lua instance while running a script, specified in positive milliseconds (0 = disabled)")]
         public int LuaScriptTimeoutMs { get; set; }
 
+        [OptionValidation]
+        [Option("lua-logging-mode", Required = false, HelpText = "Behavior of redis.log(...) when called from Lua scripts.  Defaults to Enable.")]
+        public LuaLoggingMode LuaLoggingMode { get; set; }
+
         [FilePathValidation(false, true, false)]
         [Option("unixsocket", Required = false, HelpText = "Unix socket address path to bind server to")]
         public string UnixSocketPath { get; set; }
 
-
         [IntRangeValidation(0, 777, isRequired: false)]
         [SupportedOSValidation(isRequired: false, nameof(OSPlatform.Linux), nameof(OSPlatform.OSX), nameof(OSPlatform.FreeBSD))]
         [Option("unixsocketperm", Required = false, HelpText = "Unix socket permissions in octal (Unix platforms only)")]
@@ -808,7 +811,7 @@ public GarnetServerOptions GetServerOptions(ILogger logger = null)
                 LoadModuleCS = LoadModuleCS,
                 FailOnRecoveryError = FailOnRecoveryError.GetValueOrDefault(),
                 SkipRDBRestoreChecksumValidation = SkipRDBRestoreChecksumValidation.GetValueOrDefault(),
-                LuaOptions = EnableLua.GetValueOrDefault() ? new LuaOptions(LuaMemoryManagementMode, LuaScriptMemoryLimit, LuaScriptTimeoutMs == 0 ? Timeout.InfiniteTimeSpan : TimeSpan.FromMilliseconds(LuaScriptTimeoutMs), logger) : null,
+                LuaOptions = EnableLua.GetValueOrDefault() ? new LuaOptions(LuaMemoryManagementMode, LuaScriptMemoryLimit, LuaScriptTimeoutMs == 0 ? Timeout.InfiniteTimeSpan : TimeSpan.FromMilliseconds(LuaScriptTimeoutMs), LuaLoggingMode, logger) : null,
                 UnixSocketPath = UnixSocketPath,
                 UnixSocketPermission = unixSocketPermissions
             };

libs/host/GarnetServer.cs

@@ -30,6 +30,11 @@ namespace Garnet
     /// </summary>
     public class GarnetServer : IDisposable
     {
+        /// <summary>
+        /// Resp protocol version
+        /// </summary>
+        internal const string RedisProtocolVersion = "7.2.5";
+
         static readonly string version = GetVersion();
         static string GetVersion()
         {
@@ -60,11 +65,6 @@ static string GetVersion()
         /// </summary>
         protected StoreWrapper storeWrapper;
 
-        /// <summary>
-        /// Resp protocol version
-        /// </summary>
-        readonly string redisProtocolVersion = "7.2.5";
-
         /// <summary>
         /// Metrics API
         /// </summary>
@@ -254,7 +254,7 @@ private void InitializeServer()
             // Create Garnet TCP server if none was provided.
             this.server ??= new GarnetServerTcp(opts.EndPoint, 0, opts.TlsOptions, opts.NetworkSendThrottleMax, opts.NetworkConnectionLimit, opts.UnixSocketPath, opts.UnixSocketPermission, logger);
 
-            storeWrapper = new StoreWrapper(version, redisProtocolVersion, server, store, objectStore, objectStoreSizeTracker,
+            storeWrapper = new StoreWrapper(version, RedisProtocolVersion, server, store, objectStore, objectStoreSizeTracker,
                     customCommandManager, appendOnlyFile, opts, subscribeBroker, clusterFactory: clusterFactory, loggerFactory: loggerFactory);
 
             // Create session provider for Garnet

libs/host/defaults.conf

@@ -377,6 +377,9 @@
 	/* Impose no timeout on Lua scripts by default */
 	"LuaScriptTimeoutMs": 0,
 
+	/* Allow redis.log(...) to write to the Garnet logs */
+	"LuaLoggingMode": "Enable",
+
 	/* Unix socket address path to bind the server to */
 	"UnixSocketPath": null,
 

libs/resources/RespCommandsInfo.json

@@ -624,7 +624,7 @@
         "Command": "CLUSTER_BANLIST",
         "Name": "CLUSTER|BANLIST",
         "IsInternal": true,
-        "Arity": 1,
+        "Arity": 2,
         "Flags": "Admin, NoMulti, NoScript",
         "AclCategories": "Admin, Dangerous, Slow, Garnet"
       },
@@ -742,7 +742,7 @@
       {
         "Command": "CLUSTER_HELP",
         "Name": "CLUSTER|HELP",
-        "Arity": 1,
+        "Arity": 2,
         "Flags": "Admin, NoMulti, NoScript",
         "AclCategories": "Admin, Slow, Garnet"
       },
@@ -790,7 +790,7 @@
         "Command": "CLUSTER_MTASKS",
         "Name": "CLUSTER|MTASKS",
         "IsInternal": true,
-        "Arity": 1,
+        "Arity": 2,
         "Flags": "Admin, NoMulti, NoScript",
         "AclCategories": "Admin, Dangerous, Slow, Garnet"
       },
@@ -805,7 +805,7 @@
         "Command": "CLUSTER_MYPARENTID",
         "Name": "CLUSTER|MYPARENTID",
         "IsInternal": true,
-        "Arity": 1,
+        "Arity": 2,
         "Flags": "Admin, NoMulti, NoScript",
         "AclCategories": "Admin, Garnet"
       },
@@ -916,7 +916,7 @@
         "Command": "CLUSTER_SLOTSTATE",
         "Name": "CLUSTER|SLOTSTATE",
         "IsInternal": true,
-        "Arity": 1,
+        "Arity": 2,
         "Flags": "Admin, NoMulti, NoScript",
         "AclCategories": "Admin, Dangerous, Slow, Garnet"
       },
@@ -2579,6 +2579,7 @@
       {
         "Command": "LATENCY_HELP",
         "Name": "LATENCY|HELP",
+        "Arity": 2,
         "Flags": "Admin, Loading, NoScript, Stale",
         "AclCategories": "Slow, Garnet"
       },

libs/server/Custom/CustomRespCommands.cs

@@ -25,7 +25,7 @@ private bool TryTransactionProc(byte id, CustomTransactionProcedure proc, int st
 
             latencyMetrics?.Start(LatencyMetricsType.TX_PROC_LAT);
 
-            var procInput = new CustomProcedureInput(ref parseState, startIdx: startIdx);
+            var procInput = new CustomProcedureInput(ref parseState, startIdx: startIdx, respVersion: respProtocolVersion);
             if (txnManager.RunTransactionProc(id, ref procInput, proc, ref output))
             {
                 // Write output to wire
@@ -62,7 +62,7 @@ private void TryCustomProcedure(CustomProcedure proc, int startIdx = 0)
 
             var output = new MemoryResult<byte>(null, 0);
 
-            var procInput = new CustomProcedureInput(ref parseState, startIdx: startIdx);
+            var procInput = new CustomProcedureInput(ref parseState, startIdx: startIdx, respVersion: respProtocolVersion);
             if (proc.Execute(basicGarnetApi, ref procInput, ref output))
             {
                 if (output.MemoryOwner != null)