Fix #1103 - add explicit operator support for ulong on RedisValue (#1104)

* add failing test for #1103

* add explicit ulong handling into RedisValue

* Fix (i.e. document and fix incorrect test assertions) the change re "-"/0; add ulong support to RedisResult

* add extra tests for +/.

docs/ReleaseNotes.md

 # Release Notes
 
+## (unreleased)
+
+- add `ulong` support to `RedisValue` and `RedisResult`
+- fix: remove odd equality: `"-" != 0` (we do, however, still allow `"-0"`, as that is at least semantically valid, and is logically `== 0`)
+
 ## 2.0.593
 
 - performance: unify spin-wait usage on sync/async paths to one competitor

src/StackExchange.Redis/Format.cs

 using System;
 using System.Buffers;
+using System.Buffers.Text;
 using System.Globalization;
 using System.Net;
 using System.Runtime.InteropServices;
@@ -52,6 +53,8 @@ internal static EndPoint TryParseEndPoint(string host, string port)
 
         internal static string ToString(long value) => value.ToString(NumberFormatInfo.InvariantInfo);
 
+        internal static string ToString(ulong value) => value.ToString(NumberFormatInfo.InvariantInfo);
+
         internal static string ToString(double value)
         {
             if (double.IsInfinity(value))
@@ -149,6 +152,41 @@ internal static bool TryParseDouble(string s, out double value)
             return double.TryParse(s, NumberStyles.Any, NumberFormatInfo.InvariantInfo, out value);
         }
 
+        internal static bool TryParseUInt64(string s, out ulong value)
+                    => ulong.TryParse(s, NumberStyles.Integer, NumberFormatInfo.InvariantInfo, out value);
+
+        internal static bool TryParseUInt64(ReadOnlySpan<byte> s, out ulong value)
+            => Utf8Parser.TryParse(s, out value, out int bytes, standardFormat: 'D') & bytes == s.Length;
+
+        internal static bool TryParseInt64(ReadOnlySpan<byte> s, out long value)
+            => Utf8Parser.TryParse(s, out value, out int bytes, standardFormat: 'D') & bytes == s.Length;
+
+        internal static bool CouldBeInteger(string s)
+        {
+            if (string.IsNullOrEmpty(s) || s.Length > PhysicalConnection.MaxInt64TextLen) return false;
+            bool isSigned = s[0] == '-';
+            for (int i = isSigned ? 1 : 0; i < s.Length; i++)
+            {
+                char c = s[i];
+                if (c < '0' | c > '9') return false;
+            }
+            return true;
+        }
+        internal static bool CouldBeInteger(ReadOnlySpan<byte> s)
+        {
+            if (s.IsEmpty | s.Length > PhysicalConnection.MaxInt64TextLen) return false;
+            bool isSigned = s[0] == '-';
+            for (int i = isSigned ? 1 : 0; i < s.Length; i++)
+            {
+                byte c = s[i];
+                if (c < (byte)'0' | c > (byte)'9') return false;
+            }
+            return true;
+        }
+
+        internal static bool TryParseInt64(string s, out long value)
+            => long.TryParse(s, NumberStyles.Integer, NumberFormatInfo.InvariantInfo, out value);
+
         internal static bool TryParseDouble(ReadOnlySpan<byte> s, out double value)
         {
             if (s.IsEmpty)
@@ -172,21 +210,13 @@ internal static bool TryParseDouble(ReadOnlySpan<byte> s, out double value)
                 value = double.NegativeInfinity;
                 return true;
             }
-            var ss = DecodeUtf8(s);
-            return double.TryParse(ss, NumberStyles.Any, NumberFormatInfo.InvariantInfo, out value);
-        }
-        internal static unsafe string DecodeUtf8(ReadOnlySpan<byte> span)
-        {
-            if (span.IsEmpty) return "";
-            fixed(byte* ptr = &MemoryMarshal.GetReference(span))
-            {
-                return Encoding.UTF8.GetString(ptr, span.Length);
-            }
+            return Utf8Parser.TryParse(s, out value, out int bytes) & bytes == s.Length;
         }
+
         private static bool CaseInsensitiveASCIIEqual(string xLowerCase, ReadOnlySpan<byte> y)
         {
             if (y.Length != xLowerCase.Length) return false;
-            for(int i = 0; i < y.Length; i++)
+            for (int i = 0; i < y.Length; i++)
             {
                 if (char.ToLower((char)y[i]) != xLowerCase[i]) return false;
             }
@@ -275,7 +305,8 @@ internal static string GetString(ReadOnlySequence<byte> buffer)
         }
         internal static unsafe string GetString(ReadOnlySpan<byte> span)
         {
-            fixed (byte* ptr = &MemoryMarshal.GetReference(span))
+            if (span.IsEmpty) return "";
+            fixed (byte* ptr = span)
             {
                 return Encoding.UTF8.GetString(ptr, span.Length);
             }

src/StackExchange.Redis/PhysicalConnection.cs

@@ -667,7 +667,10 @@ internal static void WriteBulkString(in RedisValue value, PipeWriter output)
                     WriteUnifiedBlob(output, (byte[])null);
                     break;
                 case RedisValue.StorageType.Int64:
-                    WriteUnifiedInt64(output, (long)value);
+                    WriteUnifiedInt64(output, value.OverlappedValueInt64);
+                    break;
+                case RedisValue.StorageType.UInt64:
+                    WriteUnifiedUInt64(output, value.OverlappedValueUInt64);
                     break;
                 case RedisValue.StorageType.Double: // use string
                 case RedisValue.StorageType.String:
@@ -752,6 +755,7 @@ internal static void WriteCrlf(PipeWriter writer)
             writer.Advance(2);
         }
 
+
         internal static int WriteRaw(Span<byte> span, long value, bool withLengthPrefix = false, int offset = 0)
         {
             if (value >= 0 && value <= 9)
@@ -1108,7 +1112,7 @@ unsafe static internal void WriteRaw(PipeWriter writer, string value, int expect
                 {
                     // encode directly in one hit
                     var span = writer.GetSpan(expectedLength);
-                    fixed (byte* bPtr = &MemoryMarshal.GetReference(span))
+                    fixed (byte* bPtr = span)
                     {
                         totalBytes = Encoding.UTF8.GetBytes(cPtr, value.Length, bPtr, expectedLength);
                     }
@@ -1128,7 +1132,7 @@ unsafe static internal void WriteRaw(PipeWriter writer, string value, int expect
 
                         int charsUsed, bytesUsed;
                         bool completed;
-                        fixed (byte* bPtr = &MemoryMarshal.GetReference(span))
+                        fixed (byte* bPtr = span)
                         {
                             encoder.Convert(cPtr + charOffset, charsRemaining, bPtr, span.Length, final, out charsUsed, out bytesUsed, out completed);
                         }
@@ -1188,6 +1192,26 @@ private static void WriteUnifiedInt64(PipeWriter writer, long value)
             var bytes = WriteRaw(span, value, withLengthPrefix: true, offset: 1);
             writer.Advance(bytes);
         }
+
+        private static void WriteUnifiedUInt64(PipeWriter writer, ulong value)
+        {
+            // note from specification: A client sends to the Redis server a RESP Array consisting of just Bulk Strings.
+            // (i.e. we can't just send ":123\r\n", we need to send "$3\r\n123\r\n"
+
+            // ${asc-len}\r\n           = 3 + MaxInt32TextLen
+            // {asc}\r\n                = MaxInt64TextLen + 2
+            var span = writer.GetSpan(5 + MaxInt32TextLen + MaxInt64TextLen);
+
+            Span<byte> valueSpan = stackalloc byte[MaxInt64TextLen];
+            if (!Utf8Formatter.TryFormat(value, valueSpan, out var len))
+                throw new InvalidOperationException("TryFormat failed");
+            span[0] = (byte)'$';
+            int offset = WriteRaw(span, len, withLengthPrefix: false, offset: 1);
+            valueSpan.Slice(0, len).CopyTo(span.Slice(offset));
+            offset += len;
+            offset = WriteCrlf(span, offset);
+            writer.Advance(offset);
+        }
         internal static void WriteInteger(PipeWriter writer, long value)
         {
             //note: client should never write integer; only server does this

src/StackExchange.Redis/RawResult.cs

@@ -319,11 +319,7 @@ internal unsafe string GetString()
 
             if (Payload.IsSingleSegment)
             {
-                var span = Payload.First.Span;
-                fixed (byte* ptr = &MemoryMarshal.GetReference(span))
-                {
-                    return Encoding.UTF8.GetString(ptr, span.Length);
-                }
+                return Format.GetString(Payload.First.Span);
             }
             var decoder = Encoding.UTF8.GetDecoder();
             int charCount = 0;
@@ -332,7 +328,7 @@ internal unsafe string GetString()
                 var span = segment.Span;
                 if (span.IsEmpty) continue;
 
-                fixed(byte* bPtr = &MemoryMarshal.GetReference(span))
+                fixed(byte* bPtr = span)
                 {
                     charCount += decoder.GetCharCount(bPtr, span.Length, false);
                 }
@@ -349,7 +345,7 @@ internal unsafe string GetString()
                     var span = segment.Span;
                     if (span.IsEmpty) continue;
 
-                    fixed (byte* bPtr = &MemoryMarshal.GetReference(span))
+                    fixed (byte* bPtr = span)
                     {
                         var written = decoder.GetChars(bPtr, span.Length, cPtr, charCount, false);
                         cPtr += written;
@@ -383,11 +379,11 @@ internal bool TryGetInt64(out long value)
                 return false;
             }
 
-            if (Payload.IsSingleSegment) return RedisValue.TryParseInt64(Payload.First.Span, out value);
+            if (Payload.IsSingleSegment) return Format.TryParseInt64(Payload.First.Span, out value);
 
             Span<byte> span = stackalloc byte[(int)Payload.Length]; // we already checked the length was <= MaxInt64TextLen
             Payload.CopyTo(span);
-            return RedisValue.TryParseInt64(span, out value);
+            return Format.TryParseInt64(span, out value);
         }
     }
 }

src/StackExchange.Redis/RedisDatabase.cs

@@ -3541,8 +3541,8 @@ protected override void WriteImpl(PhysicalConnection physical)
                     }
                     else
                     {   // recognises well-known types
-                        var val = RedisValue.TryParse(arg);
-                        if (val.IsNull && arg != null) throw new InvalidCastException($"Unable to parse value: '{arg}'");
+                        var val = RedisValue.TryParse(arg, out var valid);
+                        if (!valid) throw new InvalidCastException($"Unable to parse value: '{arg}'");
                         physical.WriteBulkString(val);
                     }
                 }

src/StackExchange.Redis/RedisResult.cs

@@ -112,6 +112,12 @@ internal static RedisResult TryCreate(PhysicalConnection connection, in RawResul
         /// <param name="result">The result to convert to a <see cref="long"/>.</param>
         public static explicit operator long(RedisResult result) => result.AsInt64();
         /// <summary>
+        /// Interprets the result as an <see cref="ulong"/>.
+        /// </summary>
+        /// <param name="result">The result to convert to a <see cref="ulong"/>.</param>
+        [CLSCompliant(false)]
+        public static explicit operator ulong(RedisResult result) => result.AsUInt64();
+        /// <summary>
         /// Interprets the result as an <see cref="int"/>.
         /// </summary>
         /// <param name="result">The result to convert to a <see cref="int"/>.</param>
@@ -142,6 +148,12 @@ internal static RedisResult TryCreate(PhysicalConnection connection, in RawResul
         /// <param name="result">The result to convert to a <see cref="T:Nullable{long}"/>.</param>
         public static explicit operator long? (RedisResult result) => result.AsNullableInt64();
         /// <summary>
+        /// Interprets the result as a <see cref="T:Nullable{ulong}"/>.
+        /// </summary>
+        /// <param name="result">The result to convert to a <see cref="T:Nullable{ulong}"/>.</param>
+        [CLSCompliant(false)]
+        public static explicit operator ulong? (RedisResult result) => result.AsNullableUInt64();
+        /// <summary>
         /// Interprets the result as a <see cref="T:Nullable{int}"/>.
         /// </summary>
         /// <param name="result">The result to convert to a <see cref="T:Nullable{int}"/>.</param>
@@ -172,6 +184,12 @@ internal static RedisResult TryCreate(PhysicalConnection connection, in RawResul
         /// <param name="result">The result to convert to a <see cref="T:long[]"/>.</param>
         public static explicit operator long[] (RedisResult result) => result.AsInt64Array();
         /// <summary>
+        /// Interprets the result as a <see cref="T:ulong[]"/>.
+        /// </summary>
+        /// <param name="result">The result to convert to a <see cref="T:ulong[]"/>.</param>
+        [CLSCompliant(false)]
+        public static explicit operator ulong[] (RedisResult result) => result.AsUInt64Array();
+        /// <summary>
         /// Interprets the result as a <see cref="T:int[]"/>.
         /// </summary>
         /// <param name="result">The result to convert to a <see cref="T:int[]"/>.</param>
@@ -206,11 +224,14 @@ internal static RedisResult TryCreate(PhysicalConnection connection, in RawResul
         internal abstract int AsInt32();
         internal abstract int[] AsInt32Array();
         internal abstract long AsInt64();
+        internal abstract ulong AsUInt64();
         internal abstract long[] AsInt64Array();
+        internal abstract ulong[] AsUInt64Array();
         internal abstract bool? AsNullableBoolean();
         internal abstract double? AsNullableDouble();
         internal abstract int? AsNullableInt32();
         internal abstract long? AsNullableInt64();
+        internal abstract ulong? AsNullableUInt64();
         internal abstract RedisKey AsRedisKey();
         internal abstract RedisKey[] AsRedisKeyArray();
         internal abstract RedisResult[] AsRedisResultArray();
@@ -279,12 +300,22 @@ internal override long AsInt64()
                 if (IsSingleton) return _value[0].AsInt64();
                 throw new InvalidCastException();
             }
+            internal override ulong AsUInt64()
+            {
+                if (IsSingleton) return _value[0].AsUInt64();
+                throw new InvalidCastException();
+            }
 
             internal override long[] AsInt64Array()
                 => IsNull ? null
                 : IsEmpty ? Array.Empty<long>()
                 : Array.ConvertAll(_value, x => x.AsInt64());
 
+            internal override ulong[] AsUInt64Array()
+                => IsNull ? null
+                : IsEmpty ? Array.Empty<ulong>()
+                : Array.ConvertAll(_value, x => x.AsUInt64());
+
             internal override bool? AsNullableBoolean()
             {
                 if (IsSingleton) return _value[0].AsNullableBoolean();
@@ -308,6 +339,11 @@ internal override long[] AsInt64Array()
                 if (IsSingleton) return _value[0].AsNullableInt64();
                 throw new InvalidCastException();
             }
+            internal override ulong? AsNullableUInt64()
+            {
+                if (IsSingleton) return _value[0].AsNullableUInt64();
+                throw new InvalidCastException();
+            }
 
             internal override RedisKey AsRedisKey()
             {
@@ -378,11 +414,14 @@ public ErrorRedisResult(string value)
             internal override int AsInt32() => throw new RedisServerException(value);
             internal override int[] AsInt32Array() => throw new RedisServerException(value);
             internal override long AsInt64() => throw new RedisServerException(value);
+            internal override ulong AsUInt64() => throw new RedisServerException(value);
             internal override long[] AsInt64Array() => throw new RedisServerException(value);
+            internal override ulong[] AsUInt64Array() => throw new RedisServerException(value);
             internal override bool? AsNullableBoolean() => throw new RedisServerException(value);
             internal override double? AsNullableDouble() => throw new RedisServerException(value);
             internal override int? AsNullableInt32() => throw new RedisServerException(value);
             internal override long? AsNullableInt64() => throw new RedisServerException(value);
+            internal override ulong? AsNullableUInt64() => throw new RedisServerException(value);
             internal override RedisKey AsRedisKey() => throw new RedisServerException(value);
             internal override RedisKey[] AsRedisKeyArray() => throw new RedisServerException(value);
             internal override RedisResult[] AsRedisResultArray() => throw new RedisServerException(value);
@@ -415,11 +454,14 @@ public SingleRedisResult(RedisValue value, ResultType? resultType)
             internal override int AsInt32() => (int)_value;
             internal override int[] AsInt32Array() => new[] { AsInt32() };
             internal override long AsInt64() => (long)_value;
+            internal override ulong AsUInt64() => (ulong)_value;
             internal override long[] AsInt64Array() => new[] { AsInt64() };
+            internal override ulong[] AsUInt64Array() => new[] { AsUInt64() };
             internal override bool? AsNullableBoolean() => (bool?)_value;
             internal override double? AsNullableDouble() => (double?)_value;
             internal override int? AsNullableInt32() => (int?)_value;
             internal override long? AsNullableInt64() => (long?)_value;
+            internal override ulong? AsNullableUInt64() => (ulong?)_value;
             internal override RedisKey AsRedisKey() => (byte[])_value;
             internal override RedisKey[] AsRedisKeyArray() => new[] { AsRedisKey() };
             internal override RedisResult[] AsRedisResultArray() => throw new InvalidCastException();

src/StackExchange.Redis/RedisValue.cs

 using System;
 using System.Buffers;
+using System.Buffers.Text;
 using System.IO;
 using System.Linq;
 using System.Reflection;
+using System.Runtime.CompilerServices;
 using System.Runtime.InteropServices;
 using System.Text;
 
@@ -17,29 +19,31 @@ namespace StackExchange.Redis
 
         private readonly object _objectOrSentinel;
         private readonly ReadOnlyMemory<byte> _memory;
-        private readonly long _overlappedValue64;
+        private readonly long _overlappedBits64;
 
         // internal bool IsNullOrDefaultValue {  get { return (valueBlob == null && valueInt64 == 0L) || ((object)valueBlob == (object)NullSentinel); } }
         private RedisValue(long overlappedValue64, ReadOnlyMemory<byte> memory, object objectOrSentinel)
         {
-            _overlappedValue64 = overlappedValue64;
+            _overlappedBits64 = overlappedValue64;
             _memory = memory;
             _objectOrSentinel = objectOrSentinel;
         }
 
-        internal RedisValue(object obj, long val)
+        internal RedisValue(object obj, long overlappedBits)
         {   // this creates a bodged RedisValue which should **never**
             // be seen directly; the contents are ... unexpected
-            _overlappedValue64 = val;
+            _overlappedBits64 = overlappedBits;
             _objectOrSentinel = obj;
             _memory = default;
         }
+
 #pragma warning disable RCS1085 // Use auto-implemented property.
         internal object DirectObject => _objectOrSentinel;
-        internal long DirectInt64 => _overlappedValue64;
+        internal long DirectOverlappedBits64 => _overlappedBits64;
 #pragma warning restore RCS1085 // Use auto-implemented property.
 
-        private readonly static object Sentinel_Integer = new object();
+        private readonly static object Sentinel_SignedInteger = new object();
+        private readonly static object Sentinel_UnsignedInteger = new object();
         private readonly static object Sentinel_Raw = new object();
         private readonly static object Sentinel_Double = new object();
 
@@ -50,12 +54,16 @@ public object Box()
         {
             var obj = _objectOrSentinel;
             if (obj is null || obj is string || obj is byte[]) return obj;
-            if (obj == Sentinel_Integer)
+            if (obj == Sentinel_SignedInteger)
             {
-                var l = _overlappedValue64;
+                var l = OverlappedValueInt64;
                 if (l >= -1 && l <= 20) return s_CommonInt32[((int)l) + 1];
                 return l;
             }
+            if (obj == Sentinel_UnsignedInteger)
+            {
+                return OverlappedValueUInt64;
+            }
             if (obj == Sentinel_Double)
             {
                 var d = OverlappedValueDouble;
@@ -74,18 +82,9 @@ public object Box()
         /// <param name="value">The value to unbox.</param>
         public static RedisValue Unbox(object value)
         {
-            if (value == null) return RedisValue.Null;
-            if (value is string s) return s;
-            if (value is byte[] b) return b;
-            if (value is int i) return i;
-            if (value is long l) return l;
-            if (value is double d) return d;
-            if (value is float f) return f;
-            if (value is bool bo) return bo;
-            if (value is Memory<byte> mem) return mem;
-            if (value is ReadOnlyMemory<byte> rom) return rom;
-            if (value is RedisValue val) return val;
-            throw new ArgumentException(nameof(value));
+            var val = TryParse(value, out var valid);
+            if (!valid) throw new ArgumentException(nameof(value));
+            return val;
         }
 
         /// <summary>
@@ -104,9 +103,9 @@ public static RedisValue Unbox(object value)
         public static RedisValue Null { get; } = new RedisValue(0, default, null);
 
         /// <summary>
-        /// Indicates whether the value is a primitive integer
+        /// Indicates whether the value is a primitive integer (signed or unsigned)
         /// </summary>
-        public bool IsInteger => _objectOrSentinel == Sentinel_Integer;
+        public bool IsInteger => _objectOrSentinel == Sentinel_SignedInteger || _objectOrSentinel == Sentinel_UnsignedInteger;
 
         /// <summary>
         /// Indicates whether the value should be considered a null value
@@ -140,7 +139,23 @@ public bool IsNullOrEmpty
         /// <param name="y">The second <see cref="RedisValue"/> to compare.</param>
         public static bool operator !=(RedisValue x, RedisValue y) => !(x == y);
 
-        private double OverlappedValueDouble => BitConverter.Int64BitsToDouble(_overlappedValue64);
+        private double OverlappedValueDouble
+        {
+            [MethodImpl(MethodImplOptions.AggressiveInlining)]
+            get => BitConverter.Int64BitsToDouble(_overlappedBits64);
+        }
+
+        internal long OverlappedValueInt64
+        {
+            [MethodImpl(MethodImplOptions.AggressiveInlining)]
+            get => _overlappedBits64;
+        }
+
+        internal ulong OverlappedValueUInt64
+        {
+            [MethodImpl(MethodImplOptions.AggressiveInlining)]
+            get => unchecked((ulong)_overlappedBits64);
+        }
 
         /// <summary>
         /// Indicates whether two RedisValue values are equivalent
@@ -160,10 +175,11 @@ public bool IsNullOrEmpty
             {
                 switch (xType)
                 {
-                    case StorageType.Double:
+                    case StorageType.Double: // make sure we use double equality rules
                         return x.OverlappedValueDouble == y.OverlappedValueDouble;
                     case StorageType.Int64:
-                        return x._overlappedValue64 == y._overlappedValue64;
+                    case StorageType.UInt64: // as long as xType == yType, only need to check the bits
+                        return x._overlappedBits64 == y._overlappedBits64;
                     case StorageType.String:
                         return (string)x._objectOrSentinel == (string)y._objectOrSentinel;
                     case StorageType.Raw:
@@ -175,12 +191,14 @@ public bool IsNullOrEmpty
             // it can't be equal
             switch (xType)
             {
+                case StorageType.UInt64:
                 case StorageType.Int64:
                 case StorageType.Double:
                     return false;
             }
             switch (yType)
             {
+                case StorageType.UInt64:
                 case StorageType.Int64:
                 case StorageType.Double:
                     return false;
@@ -198,9 +216,8 @@ public override bool Equals(object obj)
         {
             if (obj == null) return IsNull;
             if (obj is RedisValue typed) return Equals(typed);
-            var other = TryParse(obj);
-            if (other.IsNull) return false; // parse fail
-            return this == other;
+            var other = TryParse(obj, out var valid);
+            return valid && this == other; // can't be equal if parse fail
         }
 
         /// <summary>
@@ -223,7 +240,8 @@ private static int GetHashCode(RedisValue x)
                 case StorageType.Double:
                     return x.OverlappedValueDouble.GetHashCode();
                 case StorageType.Int64:
-                    return x._overlappedValue64.GetHashCode();
+                case StorageType.UInt64:
+                    return x._overlappedBits64.GetHashCode();
                 case StorageType.Raw:
                     return ((string)x).GetHashCode(); // to match equality
                 case StorageType.String:
@@ -286,64 +304,7 @@ internal static unsafe int GetHashCode(ReadOnlyMemory<byte> memory)
                 return acc;
             }
         }
-        internal static bool TryParseInt64(ReadOnlySpan<byte> value, out long result)
-        {
-            result = 0;
-            if (value.IsEmpty) return false;
-            checked
-            {
-                int max = value.Length;
-                if (value[0] == '-')
-                {
-                    for (int i = 1; i < max; i++)
-                    {
-                        var b = value[i];
-                        if (b < '0' || b > '9') return false;
-                        result = (result * 10) - (b - '0');
-                    }
-                    return true;
-                }
-                else
-                {
-                    for (int i = 0; i < max; i++)
-                    {
-                        var b = value[i];
-                        if (b < '0' || b > '9') return false;
-                        result = (result * 10) + (b - '0');
-                    }
-                    return true;
-                }
-            }
-        }
-        internal static bool TryParseInt64(string value, out long result)
-        {
-            result = 0;
-            if (string.IsNullOrEmpty(value)) return false;
-            checked
-            {
-                int max = value.Length;
-                if (value[0] == '-')
-                {
-                    for (int i = 1; i < max; i++)
-                    {
-                        var b = value[i];
-                        if (b < '0' || b > '9') return false;
-                        result = (result * 10) - (b - '0');
-                    }
-                    return true;
-                }
-                else
-                {
-                    for (int i = 0; i < max; i++)
-                    {
-                        var b = value[i];
-                        if (b < '0' || b > '9') return false;
-                        result = (result * 10) + (b - '0');
-                    }
-                    return true;
-                }
-            }
-        }
+        
 
         internal void AssertNotNull()
         {
@@ -352,7 +313,7 @@ internal void AssertNotNull()
 
         internal enum StorageType
         {
-            Null, Int64, Double, Raw, String,
+            Null, Int64, UInt64, Double, Raw, String,
         }
 
         internal StorageType Type
@@ -361,11 +322,12 @@ internal StorageType Type
             {
                 var objectOrSentinel = _objectOrSentinel;
                 if (objectOrSentinel == null) return StorageType.Null;
-                if (objectOrSentinel == Sentinel_Integer) return StorageType.Int64;
+                if (objectOrSentinel == Sentinel_SignedInteger) return StorageType.Int64;
                 if (objectOrSentinel == Sentinel_Double) return StorageType.Double;
                 if (objectOrSentinel == Sentinel_Raw) return StorageType.Raw;
                 if (objectOrSentinel is string) return StorageType.String;
                 if (objectOrSentinel is byte[]) return StorageType.Raw; // doubled-up, but retaining the array
+                if (objectOrSentinel == Sentinel_UnsignedInteger) return StorageType.UInt64;
                 throw new InvalidOperationException("Unknown type");
             }
         }
@@ -375,7 +337,7 @@ internal StorageType Type
         /// </summary>
         public long Length()
         {
-            switch(Type)
+            switch (Type)
             {
                 case StorageType.Null: return 0;
                 case StorageType.Raw: return _memory.Length;
@@ -408,7 +370,9 @@ private static int CompareTo(RedisValue x, RedisValue y)
                         case StorageType.Double:
                             return x.OverlappedValueDouble.CompareTo(y.OverlappedValueDouble);
                         case StorageType.Int64:
-                            return x._overlappedValue64.CompareTo(y._overlappedValue64);
+                            return x.OverlappedValueInt64.CompareTo(y.OverlappedValueInt64);
+                        case StorageType.UInt64:
+                            return x.OverlappedValueUInt64.CompareTo(y.OverlappedValueUInt64);
                         case StorageType.String:
                             return string.CompareOrdinal((string)x._objectOrSentinel, (string)y._objectOrSentinel);
                         case StorageType.Raw:
@@ -417,12 +381,18 @@ private static int CompareTo(RedisValue x, RedisValue y)
                 }
 
                 switch (xType)
-                { // numbers can be compared between Int64/Double
+                { // numbers can be still be compared between types
                     case StorageType.Double:
-                        if (yType == StorageType.Int64) return x.OverlappedValueDouble.CompareTo((double)y._overlappedValue64);
+                        if (yType == StorageType.Int64) return x.OverlappedValueDouble.CompareTo((double)y.OverlappedValueInt64);
+                        if (yType == StorageType.UInt64) return x.OverlappedValueDouble.CompareTo((double)y.OverlappedValueUInt64);
                         break;
                     case StorageType.Int64:
-                        if (yType == StorageType.Double) return ((double)x._overlappedValue64).CompareTo(y.OverlappedValueDouble);
+                        if (yType == StorageType.Double) return ((double)x.OverlappedValueInt64).CompareTo(y.OverlappedValueDouble);
+                        if (yType == StorageType.UInt64) return 1; // we only use unsigned if > int64, so: y is bigger
+                        break;
+                    case StorageType.UInt64:
+                        if (yType == StorageType.Double) return ((double)x.OverlappedValueUInt64).CompareTo(y.OverlappedValueDouble);
+                        if (yType == StorageType.Int64) return -1; // we only use unsigned if > int64, so: x is bigger
                         break;
                 }
 
@@ -441,28 +411,33 @@ int IComparable.CompareTo(object obj)
         {
             if (obj == null) return CompareTo(Null);
 
-            var val = TryParse(obj);
-            if (val.IsNull) return -1; // parse fail
+            var val = TryParse(obj, out var valid);
+            if (!valid) return -1; // parse fail
 
             return CompareTo(val);
         }
 
-        internal static RedisValue TryParse(object obj)
+        internal static RedisValue TryParse(object obj, out bool valid)
         {
+            valid = true;
             switch (obj)
             {
                 case null: return Null;
-                case RedisValue v: return v;
                 case string v: return v;
                 case int v: return v;
+                case uint v: return v;
                 case double v: return v;
                 case byte[] v: return v;
                 case bool v: return v;
                 case long v: return v;
+                case ulong v: return v;
                 case float v: return v;
                 case ReadOnlyMemory<byte> v: return v;
                 case Memory<byte> v: return v;
-                default: return Null;
+                case RedisValue v: return v;
+                default:
+                    valid = false;
+                    return Null;
             }
         }
 
@@ -470,7 +445,7 @@ internal static RedisValue TryParse(object obj)
         /// Creates a new <see cref="RedisValue"/> from an <see cref="int"/>.
         /// </summary>
         /// <param name="value">The <see cref="int"/> to convert to a <see cref="RedisValue"/>.</param>
-        public static implicit operator RedisValue(int value) => new RedisValue(value, default, Sentinel_Integer);
+        public static implicit operator RedisValue(int value) => new RedisValue(value, default, Sentinel_SignedInteger);
 
         /// <summary>
         /// Creates a new <see cref="RedisValue"/> from an <see cref="T:Nullable{int}"/>.
@@ -482,7 +457,7 @@ internal static RedisValue TryParse(object obj)
         /// Creates a new <see cref="RedisValue"/> from an <see cref="long"/>.
         /// </summary>
         /// <param name="value">The <see cref="long"/> to convert to a <see cref="RedisValue"/>.</param>
-        public static implicit operator RedisValue(long value) => new RedisValue(value, default, Sentinel_Integer);
+        public static implicit operator RedisValue(long value) => new RedisValue(value, default, Sentinel_SignedInteger);
 
         /// <summary>
         /// Creates a new <see cref="RedisValue"/> from an <see cref="T:Nullable{long}"/>.
@@ -490,6 +465,40 @@ internal static RedisValue TryParse(object obj)
         /// <param name="value">The <see cref="T:Nullable{long}"/> to convert to a <see cref="RedisValue"/>.</param>
         public static implicit operator RedisValue(long? value) => value == null ? Null : (RedisValue)value.GetValueOrDefault();
 
+        /// <summary>
+        /// Creates a new <see cref="RedisValue"/> from an <see cref="ulong"/>.
+        /// </summary>
+        /// <param name="value">The <see cref="ulong"/> to convert to a <see cref="RedisValue"/>.</param>
+        [CLSCompliant(false)]
+        public static implicit operator RedisValue(ulong value)
+        {
+            const ulong MSB = (1UL) << 63;
+            return (value & MSB) == 0
+                ? new RedisValue((long)value, default, Sentinel_SignedInteger) // prefer signed whenever we can
+                : new RedisValue(unchecked((long)value), default, Sentinel_UnsignedInteger); // with unsigned as the fallback
+        }
+
+        /// <summary>
+        /// Creates a new <see cref="RedisValue"/> from an <see cref="T:Nullable{ulong}"/>.
+        /// </summary>
+        /// <param name="value">The <see cref="T:Nullable{ulong}"/> to convert to a <see cref="RedisValue"/>.</param>
+        [CLSCompliant(false)]
+        public static implicit operator RedisValue(ulong? value) => value == null ? Null : (RedisValue)value.GetValueOrDefault();
+
+        /// <summary>
+        /// Creates a new <see cref="RedisValue"/> from an <see cref="uint"/>.
+        /// </summary>
+        /// <param name="value">The <see cref="uint"/> to convert to a <see cref="RedisValue"/>.</param>
+        [CLSCompliant(false)]
+        public static implicit operator RedisValue(uint value) => new RedisValue(value, default, Sentinel_SignedInteger); // 32-bits always fits as signed
+
+        /// <summary>
+        /// Creates a new <see cref="RedisValue"/> from an <see cref="T:Nullable{uint}"/>.
+        /// </summary>
+        /// <param name="value">The <see cref="T:Nullable{uint}"/> to convert to a <see cref="RedisValue"/>.</param>
+        [CLSCompliant(false)]
+        public static implicit operator RedisValue(uint? value) => value == null ? Null : (RedisValue)value.GetValueOrDefault();
+
         /// <summary>
         /// Creates a new <see cref="RedisValue"/> from an <see cref="double"/>.
         /// </summary>
@@ -499,7 +508,8 @@ internal static RedisValue TryParse(object obj)
             try
             {
                 var i64 = (long)value;
-                if (value == i64) return new RedisValue(i64, default, Sentinel_Integer);
+                // note: double doesn't offer integer accuracy at 64 bits, so we know it can't be unsigned (only use that for 64-bit)
+                if (value == i64) return new RedisValue(i64, default, Sentinel_SignedInteger);
             }
             catch { }
             return new RedisValue(BitConverter.DoubleToInt64Bits(value), default, Sentinel_Double);
@@ -552,7 +562,7 @@ internal static RedisValue TryParse(object obj)
         /// Creates a new <see cref="RedisValue"/> from an <see cref="bool"/>.
         /// </summary>
         /// <param name="value">The <see cref="bool"/> to convert to a <see cref="RedisValue"/>.</param>
-        public static implicit operator RedisValue(bool value) => new RedisValue(value ? 1 : 0, default, Sentinel_Integer);
+        public static implicit operator RedisValue(bool value) => new RedisValue(value ? 1 : 0, default, Sentinel_SignedInteger);
 
         /// <summary>
         /// Creates a new <see cref="RedisValue"/> from an <see cref="T:Nullable{bool}"/>.
@@ -593,11 +603,53 @@ internal static RedisValue TryParse(object obj)
                 case StorageType.Null:
                     return 0; // in redis, an arithmetic zero is kinda the same thing as not-exists (think "incr")
                 case StorageType.Int64:
-                    return value._overlappedValue64;
+                    return value.OverlappedValueInt64;
+                case StorageType.UInt64:
+                    return checked((long)value.OverlappedValueUInt64); // this will throw since unsigned is always 64-bit
             }
             throw new InvalidCastException($"Unable to cast from {value.Type} to long: '{value}'");
         }
 
+        /// <summary>
+        /// Converts a <see cref="RedisValue"/> to a <see cref="uint"/>.
+        /// </summary>
+        /// <param name="value">The <see cref="RedisValue"/> to convert.</param>
+        [CLSCompliant(false)]
+        public static explicit operator uint(RedisValue value)
+        {
+            value = value.Simplify();
+            switch (value.Type)
+            {
+                case StorageType.Null:
+                    return 0; // in redis, an arithmetic zero is kinda the same thing as not-exists (think "incr")
+                case StorageType.Int64:
+                    return checked((uint)value.OverlappedValueInt64);
+                case StorageType.UInt64:
+                    return checked((uint)value.OverlappedValueUInt64);
+            }
+            throw new InvalidCastException($"Unable to cast from {value.Type} to uint: '{value}'");
+        }
+
+        /// <summary>
+        /// Converts a <see cref="RedisValue"/> to a <see cref="long"/>.
+        /// </summary>
+        /// <param name="value">The <see cref="RedisValue"/> to convert.</param>
+        [CLSCompliant(false)]
+        public static explicit operator ulong(RedisValue value)
+        {
+            value = value.Simplify();
+            switch (value.Type)
+            {
+                case StorageType.Null:
+                    return 0; // in redis, an arithmetic zero is kinda the same thing as not-exists (think "incr")
+                case StorageType.Int64:
+                    return checked((ulong)value.OverlappedValueInt64); // throw if negative
+                case StorageType.UInt64:
+                    return value.OverlappedValueUInt64;
+            }
+            throw new InvalidCastException($"Unable to cast from {value.Type} to ulong: '{value}'");
+        }
+
         /// <summary>
         /// Converts a <see cref="RedisValue"/> to a <see cref="double"/>.
         /// </summary>
@@ -610,17 +662,61 @@ internal static RedisValue TryParse(object obj)
                 case StorageType.Null:
                     return 0; // in redis, an arithmetic zero is kinda the same thing as not-exists (think "incr")
                 case StorageType.Int64:
-                    return value._overlappedValue64;
+                    return value.OverlappedValueInt64;
+                case StorageType.UInt64:
+                    return value.OverlappedValueUInt64;
                 case StorageType.Double:
                     return value.OverlappedValueDouble;
             }
             throw new InvalidCastException($"Unable to cast from {value.Type} to double: '{value}'");
         }
 
+        /// <summary>
+        /// Converts a <see cref="RedisValue"/> to a <see cref="decimal"/>.
+        /// </summary>
+        /// <param name="value">The <see cref="RedisValue"/> to convert.</param>
+        public static explicit operator decimal(RedisValue value)
+        {
+            value = value.Simplify();
+            switch (value.Type)
+            {
+                case StorageType.Null:
+                    return 0; // in redis, an arithmetic zero is kinda the same thing as not-exists (think "incr")
+                case StorageType.Int64:
+                    return value.OverlappedValueInt64;
+                case StorageType.UInt64:
+                    return value.OverlappedValueUInt64;
+                case StorageType.Double:
+                    return (decimal)value.OverlappedValueDouble;
+            }
+            throw new InvalidCastException($"Unable to cast from {value.Type} to decimal: '{value}'");
+        }
+
+        /// <summary>
+        /// Converts a <see cref="RedisValue"/> to a <see cref="float"/>.
+        /// </summary>
+        /// <param name="value">The <see cref="RedisValue"/> to convert.</param>
+        public static explicit operator float(RedisValue value)
+        {
+            value = value.Simplify();
+            switch (value.Type)
+            {
+                case StorageType.Null:
+                    return 0; // in redis, an arithmetic zero is kinda the same thing as not-exists (think "incr")
+                case StorageType.Int64:
+                    return value.OverlappedValueInt64;
+                case StorageType.UInt64:
+                    return value.OverlappedValueUInt64;
+                case StorageType.Double:
+                    return (float)value.OverlappedValueDouble;
+            }
+            throw new InvalidCastException($"Unable to cast from {value.Type} to double: '{value}'");
+        }
+
         private static bool TryParseDouble(ReadOnlySpan<byte> blob, out double value)
         {
             // simple integer?
-            if (TryParseInt64(blob, out var i64))
+            if (Format.CouldBeInteger(blob) && Format.TryParseInt64(blob, out var i64))
             {
                 value = i64;
                 return true;
@@ -636,6 +732,20 @@ private static bool TryParseDouble(ReadOnlySpan<byte> blob, out double value)
         public static explicit operator double? (RedisValue value)
             => value.IsNull ? (double?)null : (double)value;
 
+        /// <summary>
+        /// Converts the <see cref="RedisValue"/> to a <see cref="T:Nullable{float}"/>.
+        /// </summary>
+        /// <param name="value">The <see cref="RedisValue"/> to convert.</param>
+        public static explicit operator float? (RedisValue value)
+            => value.IsNull ? (float?)null : (float)value;
+
+        /// <summary>
+        /// Converts the <see cref="RedisValue"/> to a <see cref="T:Nullable{decimal}"/>.
+        /// </summary>
+        /// <param name="value">The <see cref="RedisValue"/> to convert.</param>
+        public static explicit operator decimal? (RedisValue value)
+            => value.IsNull ? (decimal?)null : (decimal)value;
+
         /// <summary>
         /// Converts the <see cref="RedisValue"/> to a <see cref="T:Nullable{long}"/>.
         /// </summary>
@@ -643,6 +753,14 @@ private static bool TryParseDouble(ReadOnlySpan<byte> blob, out double value)
         public static explicit operator long? (RedisValue value)
             => value.IsNull ? (long?)null : (long)value;
 
+        /// <summary>
+        /// Converts the <see cref="RedisValue"/> to a <see cref="T:Nullable{ulong}"/>.
+        /// </summary>
+        /// <param name="value">The <see cref="RedisValue"/> to convert.</param>
+        [CLSCompliant(false)]
+        public static explicit operator ulong? (RedisValue value)
+            => value.IsNull ? (ulong?)null : (ulong)value;
+
         /// <summary>
         /// Converts the <see cref="RedisValue"/> to a <see cref="T:Nullable{int}"/>.
         /// </summary>
@@ -650,6 +768,14 @@ private static bool TryParseDouble(ReadOnlySpan<byte> blob, out double value)
         public static explicit operator int? (RedisValue value)
             => value.IsNull ? (int?)null : (int)value;
 
+        /// <summary>
+        /// Converts the <see cref="RedisValue"/> to a <see cref="T:Nullable{uint}"/>.
+        /// </summary>
+        /// <param name="value">The <see cref="RedisValue"/> to convert.</param>
+        [CLSCompliant(false)]
+        public static explicit operator uint? (RedisValue value)
+            => value.IsNull ? (uint?)null : (uint)value;
+
         /// <summary>
         /// Converts the <see cref="RedisValue"/> to a <see cref="T:Nullable{bool}"/>.
         /// </summary>
@@ -667,7 +793,8 @@ private static bool TryParseDouble(ReadOnlySpan<byte> blob, out double value)
             {
                 case StorageType.Null: return null;
                 case StorageType.Double: return Format.ToString(value.OverlappedValueDouble);
-                case StorageType.Int64: return Format.ToString(value._overlappedValue64);
+                case StorageType.Int64: return Format.ToString(value.OverlappedValueInt64);
+                case StorageType.UInt64: return Format.ToString(value.OverlappedValueUInt64);
                 case StorageType.String: return (string)value._objectOrSentinel;
                 case StorageType.Raw:
                     var span = value._memory.Span;
@@ -675,7 +802,7 @@ private static bool TryParseDouble(ReadOnlySpan<byte> blob, out double value)
                     if (span.Length == 2 && span[0] == (byte)'O' && span[1] == (byte)'K') return "OK"; // frequent special-case
                     try
                     {
-                        return Format.DecodeUtf8(span);
+                        return Format.GetString(span);
                     }
                     catch
                     {
@@ -731,11 +858,19 @@ private static string ToHex(ReadOnlySpan<byte> src)
 
                     return value._memory.ToArray();
                 case StorageType.Int64:
-                    Span<byte> span = stackalloc byte[PhysicalConnection.MaxInt64TextLen];
-                    int len = PhysicalConnection.WriteRaw(span, value._overlappedValue64, false, 0);
+                    Span<byte> span = stackalloc byte[PhysicalConnection.MaxInt64TextLen + 2];
+                    int len = PhysicalConnection.WriteRaw(span, value.OverlappedValueInt64, false, 0);
                     arr = new byte[len - 2]; // don't need the CRLF
                     span.Slice(0, arr.Length).CopyTo(arr);
                     return arr;
+                case StorageType.UInt64:
+                    // we know it is a huge value - just jump straight to Utf8Formatter
+                    span = stackalloc byte[PhysicalConnection.MaxInt64TextLen];
+                    if (!Utf8Formatter.TryFormat(value.OverlappedValueUInt64, span, out len))
+                        throw new InvalidOperationException("TryFormat failed");
+                    arr = new byte[len];
+                    span.Slice(0, len).CopyTo(arr);
+                    return arr;
             }
             // fallback: stringify and encode
             return Encoding.UTF8.GetBytes((string)value);
@@ -751,8 +886,8 @@ private static string ToHex(ReadOnlySpan<byte> src)
         TypeCode IConvertible.GetTypeCode() => TypeCode.Object;
 
         bool IConvertible.ToBoolean(IFormatProvider provider) => (bool)this;
-        byte IConvertible.ToByte(IFormatProvider provider) => (byte)this;
-        char IConvertible.ToChar(IFormatProvider provider) => (char)this;
+        byte IConvertible.ToByte(IFormatProvider provider) => (byte)(uint)this;
+        char IConvertible.ToChar(IFormatProvider provider) => (char)(uint)this;
         DateTime IConvertible.ToDateTime(IFormatProvider provider) => DateTime.Parse((string)this, provider);
         decimal IConvertible.ToDecimal(IFormatProvider provider) => (decimal)this;
         double IConvertible.ToDouble(IFormatProvider provider) => (double)this;
@@ -772,8 +907,8 @@ object IConvertible.ToType(Type conversionType, IFormatProvider provider)
             switch (System.Type.GetTypeCode(conversionType))
             {
                 case TypeCode.Boolean: return (bool)this;
-                case TypeCode.Byte: return (byte)this;
-                case TypeCode.Char: return (char)this;
+                case TypeCode.Byte: return checked((byte)(uint)this);
+                case TypeCode.Char: return checked((char)(uint)this);
                 case TypeCode.DateTime: return DateTime.Parse((string)this, provider);
                 case TypeCode.Decimal: return (decimal)this;
                 case TypeCode.Double: return (double)this;
@@ -783,16 +918,16 @@ object IConvertible.ToType(Type conversionType, IFormatProvider provider)
                 case TypeCode.SByte: return (sbyte)this;
                 case TypeCode.Single: return (float)this;
                 case TypeCode.String: return (string)this;
-                case TypeCode.UInt16: return (ushort)this;
+                case TypeCode.UInt16: return checked((ushort)(uint)this);
                 case TypeCode.UInt32: return (uint)this;
-                case TypeCode.UInt64: return (long)this;
+                case TypeCode.UInt64: return (ulong)this;
                 case TypeCode.Object: return this;
                 default:
                     throw new NotSupportedException();
             }
         }
 
-        ushort IConvertible.ToUInt16(IFormatProvider provider) => (ushort)this;
+        ushort IConvertible.ToUInt16(IFormatProvider provider) => checked((ushort)(uint)this);
         uint IConvertible.ToUInt32(IFormatProvider provider) => (uint)this;
         ulong IConvertible.ToUInt64(IFormatProvider provider) => (ulong)this;
 
@@ -804,18 +939,28 @@ object IConvertible.ToType(Type conversionType, IFormatProvider provider)
         /// a hash key or similar - we don't want to break it; RedisConnection uses
         /// the storage type, not the "does it look like a long?" - for this reason
         /// </summary>
-        private RedisValue Simplify()
+        internal RedisValue Simplify()
         {
+            long i64;
+            ulong u64;
             switch (Type)
             {
                 case StorageType.String:
                     string s = (string)_objectOrSentinel;
-                    if (TryParseInt64(s, out var i64)) return i64;
+                    if (Format.CouldBeInteger(s))
+                    {
+                        if (Format.TryParseInt64(s, out i64)) return i64;
+                        if (Format.TryParseUInt64(s, out u64)) return u64;
+                    }
                     if (Format.TryParseDouble(s, out var f64)) return f64;
                     break;
                 case StorageType.Raw:
                     var b = _memory.Span;
-                    if (TryParseInt64(b, out i64)) return i64;
+                    if (Format.CouldBeInteger(b))
+                    {
+                        if (Format.TryParseInt64(b, out i64)) return i64;
+                        if (Format.TryParseUInt64(b, out u64)) return u64;
+                    }
                     if (TryParseDouble(b, out f64)) return f64;
                     break;
                 case StorageType.Double:
@@ -828,7 +973,7 @@ private RedisValue Simplify()
         }
 
         /// <summary>
-        /// <para>Convert to a long if possible, returning true.</para>
+        /// <para>Convert to a signed long if possible, returning true.</para>
         /// <para>Returns false otherwise.</para>
         /// </summary>
         /// <param name="val">The <see cref="long"/> value, if conversion was possible.</param>
@@ -837,16 +982,20 @@ public bool TryParse(out long val)
             switch (Type)
             {
                 case StorageType.Int64:
-                    val = _overlappedValue64;
+                    val = OverlappedValueInt64;
                     return true;
+                case StorageType.UInt64:
+                    // we only use unsigned for oversize, so no: it doesn't fit
+                    val = default;
+                    return false;
                 case StorageType.String:
-                    return TryParseInt64((string)_objectOrSentinel, out val);
+                    return Format.TryParseInt64((string)_objectOrSentinel, out val);
                 case StorageType.Raw:
-                    return TryParseInt64(_memory.Span, out val);
+                    return Format.TryParseInt64(_memory.Span, out val);
                 case StorageType.Double:
                     var d = OverlappedValueDouble;
                     try { val = (long)d; }
-                    catch { val = default;  return false; }
+                    catch { val = default; return false; }
                     return val == d;
                 case StorageType.Null:
                     // in redis-land 0 approx. equal null; so roll with it
@@ -884,7 +1033,10 @@ public bool TryParse(out double val)
             switch (Type)
             {
                 case StorageType.Int64:
-                    val = _overlappedValue64;
+                    val = OverlappedValueInt64;
+                    return true;
+                case StorageType.UInt64:
+                    val = OverlappedValueUInt64;
                     return true;
                 case StorageType.Double:
                     val = OverlappedValueDouble;
@@ -911,7 +1063,7 @@ public static RedisValue CreateFrom(MemoryStream stream)
         {
             if (stream == null) return Null;
             if (stream.Length == 0) return Array.Empty<byte>();
-            if(stream.TryGetBuffer(out var segment) || ReflectionTryGetBuffer(stream, out segment))
+            if (stream.TryGetBuffer(out var segment) || ReflectionTryGetBuffer(stream, out segment))
             {
                 return new Memory<byte>(segment.Array, segment.Offset, segment.Count);
             }
@@ -957,7 +1109,7 @@ public bool StartsWith(RedisValue value)
             var thisType = Type;
             if (thisType == value.Type) // same? can often optimize
             {
-                switch(thisType)
+                switch (thisType)
                 {
                     case StorageType.String:
                         var sThis = ((string)_objectOrSentinel);
@@ -986,7 +1138,7 @@ public bool StartsWith(RedisValue value)
 
         private ReadOnlyMemory<byte> AsMemory(out byte[] leased)
         {
-            switch(Type)
+            switch (Type)
             {
                 case StorageType.Raw:
                     leased = null;
@@ -994,7 +1146,7 @@ private ReadOnlyMemory<byte> AsMemory(out byte[] leased)
                 case StorageType.String:
                     string s = (string)_objectOrSentinel;
                     HaveString:
-                    if(s.Length == 0)
+                    if (s.Length == 0)
                     {
                         leased = null;
                         return default;
@@ -1007,7 +1159,13 @@ private ReadOnlyMemory<byte> AsMemory(out byte[] leased)
                     goto HaveString;
                 case StorageType.Int64:
                     leased = ArrayPool<byte>.Shared.Rent(PhysicalConnection.MaxInt64TextLen + 2); // reused code has CRLF terminator
-                    len = PhysicalConnection.WriteRaw(leased, _overlappedValue64) - 2; // drop the CRLF
+                    len = PhysicalConnection.WriteRaw(leased, OverlappedValueInt64) - 2; // drop the CRLF
+                    return new ReadOnlyMemory<byte>(leased, 0, len);
+                case StorageType.UInt64:
+                    leased = ArrayPool<byte>.Shared.Rent(PhysicalConnection.MaxInt64TextLen); // reused code has CRLF terminator
+                    // value is huge, jump direct to Utf8Formatter
+                    if (!Utf8Formatter.TryFormat(OverlappedValueUInt64, leased, out len))
+                        throw new InvalidOperationException("TryFormat failed");
                     return new ReadOnlyMemory<byte>(leased, 0, len);
             }
             leased = null;

tests/StackExchange.Redis.Tests/Issues/Issue1103.cs

+using System.Globalization;
+using Xunit;
+using Xunit.Abstractions;
+using static StackExchange.Redis.RedisValue;
+
+namespace StackExchange.Redis.Tests.Issues
+{
+    public class Issue1103 : TestBase
+    {
+        public Issue1103(ITestOutputHelper output) : base(output) { }
+
+        [Theory]
+        [InlineData(142205255210238005UL, (int)StorageType.Int64)]
+        [InlineData(ulong.MaxValue, (int)StorageType.UInt64)]
+        [InlineData(ulong.MinValue, (int)StorageType.Int64)]
+        [InlineData(0x8000000000000000UL, (int)StorageType.UInt64)]
+        [InlineData(0x8000000000000001UL, (int)StorageType.UInt64)]
+        [InlineData(0x7FFFFFFFFFFFFFFFUL, (int)StorageType.Int64)]
+        public void LargeUInt64StoredCorrectly(ulong value, int storageType)
+        {
+            RedisKey key = Me();
+            using (var muxer = Create())
+            {
+                var db = muxer.GetDatabase();
+                RedisValue typed = value;
+
+                // only need UInt64 for 64-bits
+                Assert.Equal((StorageType)storageType, typed.Type);
+                db.StringSet(key, typed);
+                var fromRedis = db.StringGet(key);
+
+                Log($"{fromRedis.Type}: {fromRedis}");
+                Assert.Equal(StorageType.Raw, fromRedis.Type);
+                Assert.Equal(value, (ulong)fromRedis);
+                Assert.Equal(value.ToString(CultureInfo.InvariantCulture), fromRedis.ToString());
+
+                var simplified = fromRedis.Simplify();
+                Log($"{simplified.Type}: {simplified}");
+                Assert.Equal((StorageType)storageType, typed.Type);
+                Assert.Equal(value, (ulong)simplified);
+                Assert.Equal(value.ToString(CultureInfo.InvariantCulture), fromRedis.ToString());
+            }
+        }
+
+        [Fact]
+        public void UnusualRedisValueOddities() // things we found while doing this
+        {
+            RedisValue x = 0, y = "0";
+            Assert.Equal(x, y);
+            Assert.Equal(y, x);
+
+            y = "-0";
+            Assert.Equal(x, y);
+            Assert.Equal(y, x);
+
+            y = "-"; // this is the oddness; this used to return true
+            Assert.NotEqual(x, y);
+            Assert.NotEqual(y, x);
+
+            y = "+";
+            Assert.NotEqual(x, y);
+            Assert.NotEqual(y, x);
+
+            y = ".";
+            Assert.NotEqual(x, y);
+            Assert.NotEqual(y, x);
+        }
+    }
+}

tests/StackExchange.Redis.Tests/Streams.cs

@@ -1063,10 +1063,9 @@ public void StreamPendingNoMessagesOrConsumers()
         public void StreamPositionDefaultValueIsBeginning()
         {
             RedisValue position = StreamPosition.Beginning;
-
-            Assert.Equal(StreamConstants.ReadMinValue, StreamPosition.Resolve(position, RedisCommand.XREAD));
-            Assert.Equal(StreamConstants.ReadMinValue, StreamPosition.Resolve(position, RedisCommand.XREADGROUP));
-            Assert.Equal(StreamConstants.ReadMinValue, StreamPosition.Resolve(position, RedisCommand.XGROUP));
+            Assert.Equal(StreamConstants.AllMessages, StreamPosition.Resolve(position, RedisCommand.XREAD));
+            Assert.Equal(StreamConstants.AllMessages, StreamPosition.Resolve(position, RedisCommand.XREADGROUP));
+            Assert.Equal(StreamConstants.AllMessages, StreamPosition.Resolve(position, RedisCommand.XGROUP));
         }
 
         [Fact]
@@ -1074,7 +1073,7 @@ public void StreamPositionValidateBeginning()
         {
             var position = StreamPosition.Beginning;
 
-            Assert.Equal(StreamConstants.ReadMinValue, StreamPosition.Resolve(position, RedisCommand.XREAD));
+            Assert.Equal(StreamConstants.AllMessages, StreamPosition.Resolve(position, RedisCommand.XREAD));
         }
 
         [Fact]

toys/StackExchange.Redis.Server/TypedRedisValue.cs

@@ -88,7 +88,7 @@ public ReadOnlySpan<TypedRedisValue> Span
                 if (Type != ResultType.MultiBulk) return default;
                 var arr = (TypedRedisValue[])_value.DirectObject;
                 if (arr == null) return default;
-                var length = (int)_value.DirectInt64;
+                var length = (int)_value.DirectOverlappedBits64;
                 return new ReadOnlySpan<TypedRedisValue>(arr, 0, length);
             }
         }
@@ -99,7 +99,7 @@ public ArraySegment<TypedRedisValue> Segment
                 if (Type != ResultType.MultiBulk) return default;
                 var arr = (TypedRedisValue[])_value.DirectObject;
                 if (arr == null) return default;
-                var length = (int)_value.DirectInt64;
+                var length = (int)_value.DirectOverlappedBits64;
                 return new ArraySegment<TypedRedisValue>(arr, 0, length);
             }
         }
@@ -163,7 +163,7 @@ internal void Recycle(int limit = -1)
         {
             if (_value.DirectObject is TypedRedisValue[] arr)
             {
-                if (limit < 0) limit = (int)_value.DirectInt64;
+                if (limit < 0) limit = (int)_value.DirectOverlappedBits64;
                 for (int i = 0; i < limit; i++)
                 {
                     arr[i].Recycle();