Pre-decompress query updates (#302)

This dramatically improves performance by avoiding the default
implementation of BrotliStream.ReadByte() inherited from Stream, which
allocates an array per byte read.

## API

 - [ ] This is an API breaking change to the SDK

*If the API is breaking, please state below what will break*

## Requires SpacetimeDB PRs

## Testsuite

SpacetimeDB branch name: master

## Testing
*Write instructions for a test that you performed for this PR*

- [ ] Describe a test for this PR that you have completed

Author: kazimuth

src/MultiDictionary.cs

@@ -74,7 +74,7 @@ public bool Add(TKey key, TValue value)
             Debug.Assert(key != null);
             if (RawDict.TryGetValue(key, out var result))
             {
-                Debug.Assert(ValueComparer.Equals(value, result.Value), $"Added key-value pair with mismatched value to existing data, {key} {value} {result.Value}");
+                Debug.Assert(ValueComparer.Equals(value, result.Value));
                 RawDict[key] = (value, result.Multiplicity + 1);
                 return false;
             }
@@ -252,7 +252,7 @@ public void Apply(MultiDictionaryDelta<TKey, TValue> delta, List<KeyValuePair<TK
                         // However, it may remove the key-value pair entirely.
 
                         var theirDelta = their.NonValueChange;
-                        Debug.Assert(ValueComparer.Equals(my.Value, theirDelta.Value), $"mismatched value change: {my.Value} {theirDelta.Value} {their}");
+                        Debug.Assert(ValueComparer.Equals(my.Value, theirDelta.Value));
                         var newMultiplicity = (int)my.Multiplicity + theirDelta.Delta;
                         if (newMultiplicity > 0)
                         {

src/SpacetimeDBClient.cs

@@ -314,7 +314,13 @@ private static ServerMessage DecompressDecodeMessage(byte[] bytes)
                 _ => throw new InvalidOperationException("Unknown compression type"),
             };
 
-            return new ServerMessage.BSATN().Read(new BinaryReader(decompressedStream));
+            // TODO: consider pooling these.
+            // DO NOT TRY TO TAKE THIS OUT. The BrotliStream ReadByte() implementation allocates an array
+            // PER BYTE READ. You have to do it all at once to avoid that problem.
+            MemoryStream memoryStream = new MemoryStream();
+            decompressedStream.CopyTo(memoryStream);
+            memoryStream.Seek(0, SeekOrigin.Begin);
+            return new ServerMessage.BSATN().Read(new BinaryReader(memoryStream));
         }
 
         private static QueryUpdate DecompressDecodeQueryUpdate(CompressableQueryUpdate update)
@@ -338,7 +344,13 @@ private static QueryUpdate DecompressDecodeQueryUpdate(CompressableQueryUpdate u
                     throw new InvalidOperationException();
             }
 
-            return new QueryUpdate.BSATN().Read(new BinaryReader(decompressedStream));
+            // TODO: consider pooling these.
+            // DO NOT TRY TO TAKE THIS OUT. The BrotliStream ReadByte() implementation allocates an array
+            // PER BYTE READ. You have to do it all at once to avoid that problem.
+            MemoryStream memoryStream = new MemoryStream();
+            decompressedStream.CopyTo(memoryStream);
+            memoryStream.Seek(0, SeekOrigin.Begin);
+            return new QueryUpdate.BSATN().Read(new BinaryReader(memoryStream));
         }
 
         private static IEnumerable<byte[]> BsatnRowListIter(BsatnRowList list)