feat(qwp): stop resending the full symbol dictionary on every message

src/net-questdb-client-tests/HttpTests.cs

@@ -28,6 +28,8 @@
 using dummy_http_server;
 using NUnit.Framework;
 using QuestDB;
+using QuestDB.Enums;
+using QuestDB.Senders;
 using QuestDB.Utils;
 
 namespace net_questdb_client_tests;
@@ -39,6 +41,52 @@ public class HttpTests
     private const int HttpPort = 29473;
     private const int HttpsPort = 29474;
 
+    [Test]
+    public void ConstructorFailureAfterClientCreation_DisposesClient()
+    {
+        var options = new SenderOptions
+        {
+            protocol         = ProtocolType.http,
+            protocol_version = (ProtocolVersion)int.MaxValue,
+        };
+        DisposalTrackingHandler? trackingHandler = null;
+
+        Assert.Throws<NotImplementedException>(() =>
+            _ = new HttpSender(options, _ =>
+            {
+                trackingHandler = new DisposalTrackingHandler();
+                return new HttpClient(trackingHandler, disposeHandler: true);
+            }));
+
+        Assert.That(trackingHandler, Is.Not.Null);
+        Assert.That(trackingHandler!.Disposed, Is.True,
+            "a constructor that fails after creating its HttpClient must release that client");
+    }
+
+    [Test]
+    public void BasicAuth_EncodesNonAsciiCredentialsAsUtf8()
+    {
+        var options = new SenderOptions
+        {
+            protocol         = ProtocolType.http,
+            protocol_version = ProtocolVersion.V1,
+            username         = "用户",
+            password         = "пароль",
+        };
+        HttpClient? client = null;
+
+        using var sender = new HttpSender(options, handler =>
+        {
+            client = new HttpClient(handler, disposeHandler: false);
+            return client;
+        });
+
+        Assert.That(client, Is.Not.Null);
+        Assert.That(client!.DefaultRequestHeaders.Authorization?.Scheme, Is.EqualTo("Basic"));
+        Assert.That(client.DefaultRequestHeaders.Authorization?.Parameter,
+            Is.EqualTo(Convert.ToBase64String(Encoding.UTF8.GetBytes("用户:пароль"))));
+    }
+
     [Test]
     public async Task BasicArrayDouble()
     {
@@ -2002,4 +2050,23 @@ await sender.Table("metrics")
 
         await server.StopAsync();
     }
-}
+
+    private sealed class DisposalTrackingHandler : HttpMessageHandler
+    {
+        public bool Disposed { get; private set; }
+
+        protected override Task<HttpResponseMessage> SendAsync(
+            HttpRequestMessage request,
+            CancellationToken cancellationToken)
+            => throw new NotSupportedException();
+
+        protected override void Dispose(bool disposing)
+        {
+            if (disposing)
+            {
+                Disposed = true;
+            }
+            base.Dispose(disposing);
+        }
+    }
+}

src/net-questdb-client-tests/Pooling/FacadeCallbackTests.cs

@@ -124,7 +124,12 @@ public async Task FacadeDrainerListener_ReceivesDrainerEvents()
             var orphanSlot = Path.Combine(root, "crashed-sibling");
             using (var ring = QwpSegmentRing.Open(orphanSlot, segmentCapacity: 4096))
             {
-                Assert.That(ring.TryAppend(new byte[] { 1 }), Is.True);
+                // Recovery validates orphan contents as QWP so the dictionary sidefile can be
+                // reconstructed safely; seed a legal zero-table commit frame, not opaque bytes.
+                var frame = QwpEncoder.Encode(
+                    Array.Empty<QwpTableBuffer>(),
+                    new QwpSymbolDictionary());
+                Assert.That(ring.TryAppend(frame), Is.True);
             }
 
             await using var server = await StartAckingServerAsync();

src/net-questdb-client-tests/QuestDbWebSocketIntegrationTests.cs

@@ -396,6 +396,57 @@ public async Task Reconnect_WsCloseMidSession_NoSfDir_ReplaysFromRamRing()
         await VerifyTableRowCountAsync("test_ws_close_ram", expected: 30, maxAttempts: 150);
     }
 
+    [Test]
+    public async Task Reconnect_WsCloseMidSession_SymbolDictionaryCatchUpReplays()
+    {
+        await DropTableAsync("test_ws_close_symbols");
+        var realEndpoint = _questDb!.GetWebSocketEndpoint();
+        using var proxy = new TcpProxy(realEndpoint);
+        await proxy.StartAsync();
+
+        using (var sender = Sender.New(
+                   $"ws::addr={proxy.LocalEndpoint};auto_flush=off;"
+                   + "reconnect_max_duration_millis=60000;"
+                   + "reconnect_initial_backoff_millis=50;reconnect_max_backoff_millis=500;"))
+        {
+            var qwp = (IQwpWebSocketSender)sender;
+
+            for (var i = 0; i < 10; i++)
+            {
+                sender.Table("test_ws_close_symbols")
+                    .Symbol("sym", i % 2 == 0 ? "alpha" : "beta")
+                    .Column("v", (long)i).At(DateTime.UtcNow);
+            }
+            await sender.SendAsync();
+            await qwp.PingAsync();
+
+            proxy.KillAllConnections();
+
+            // Post-kill rows reuse pre-kill symbol ids (restored by dictionary catch-up on the new
+            // connection) and introduce a new value that rides the post-reconnect delta.
+            for (var i = 10; i < 30; i++)
+            {
+                sender.Table("test_ws_close_symbols")
+                    .Symbol("sym", i % 3 == 0 ? "gamma" : i % 2 == 0 ? "alpha" : "beta")
+                    .Column("v", (long)i).At(DateTime.UtcNow);
+            }
+            await sender.SendAsync();
+
+            for (var i = 0; i < 200; i++)
+            {
+                try { await qwp.PingAsync(); break; }
+                catch { await Task.Delay(100); }
+            }
+        }
+
+        await VerifyTableRowCountAsync("test_ws_close_symbols", expected: 30, maxAttempts: 150);
+        var rows = await QueryAllRowsAsync(
+            "select sym, count() from test_ws_close_symbols order by sym");
+        Assert.That(
+            rows.Select(r => (r[0].GetString(), r[1].GetInt64())),
+            Is.EqualTo(new[] { ("alpha", 12L), ("beta", 12L), ("gamma", 6L) }));
+    }
+
     [Test]
     public async Task SchemaCacheReuse_TwoFlushesSameTable_BothLand()
     {

src/net-questdb-client-tests/Qwp/QwpFrameTestUtils.cs

@@ -0,0 +1,50 @@
+/*******************************************************************************
+ *     ___                  _   ____  ____
+ *    / _ \ _   _  ___  ___| |_|  _ \| __ )
+ *   | | | | | | |/ _ \/ __| __| | | |  _ \
+ *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
+ *    \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ *  Copyright (c) 2014-2019 Appsicle
+ *  Copyright (c) 2019-2026 QuestDB
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ ******************************************************************************/
+
+using QuestDB.Qwp;
+
+namespace net_questdb_client_tests.Qwp;
+
+internal static class QwpFrameTestUtils
+{
+    public static (int Start, string[] Entries) ReadSymbolDelta(byte[] frame)
+    {
+        var p = QwpConstants.HeaderSize;
+        var start = checked((int)QwpVarint.Read(frame.AsSpan(p), out var read));
+        p += read;
+        var count = checked((int)QwpVarint.Read(frame.AsSpan(p), out read));
+        p += read;
+
+        var entries = new string[count];
+        for (var i = 0; i < count; i++)
+        {
+            var len = checked((int)QwpVarint.Read(frame.AsSpan(p), out read));
+            p += read;
+            entries[i] = QwpStrictUtf8.Encoding.GetString(frame, p, len);
+            p += len;
+        }
+
+        return (start, entries);
+    }
+}

src/net-questdb-client-tests/Qwp/QwpResponseTests.cs

@@ -46,6 +46,8 @@ public void StatusCodes_HaveStableByteValues()
         Assert.That((byte)QwpStatusCode.WriteError,     Is.EqualTo((byte)0x09));
         Assert.That((byte)QwpStatusCode.Cancelled,      Is.EqualTo((byte)0x0A));
         Assert.That((byte)QwpStatusCode.LimitExceeded,  Is.EqualTo((byte)0x0B));
+        Assert.That((byte)QwpStatusCode.NotWritable,    Is.EqualTo((byte)0x0C));
+        Assert.That((byte)QwpStatusCode.DictionaryGap,  Is.EqualTo((byte)0x0D));
     }
 
     [Test]

src/net-questdb-client-tests/Qwp/QwpSymbolDictionaryTests.cs

@@ -23,7 +23,9 @@
  ******************************************************************************/
 
 using NUnit.Framework;
+using QuestDB.Enums;
 using QuestDB.Qwp;
+using QuestDB.Utils;
 
 namespace net_questdb_client_tests.Qwp;
 
@@ -50,6 +52,34 @@ public void Add_RepeatedValue_ReturnsSameId()
         Assert.That(d.Count, Is.EqualTo(2));
     }
 
+    [Test]
+    public void Add_RefusesNewValuePastProtocolCapWithoutMutatingDictionary()
+    {
+        var d = new QwpSymbolDictionary(QwpConstants.MaxSymbolDictionarySize);
+        for (var i = 0; i < QwpConstants.MaxSymbolDictionarySize; i++)
+        {
+            d.Add("s" + i);
+        }
+        Assert.That(d.Count, Is.EqualTo(QwpConstants.MaxSymbolDictionarySize));
+
+        var ex = Assert.Throws<IngressError>(() => d.Add("one-too-many"));
+        Assert.That(ex!.code, Is.EqualTo(ErrorCode.InvalidApiCall));
+        Assert.That(ex.Message, Does.Contain(QwpConstants.MaxSymbolDictionarySize.ToString()));
+        Assert.That(ex.Message, Does.Contain("close this sender"));
+
+        Assert.That(d.Count, Is.EqualTo(QwpConstants.MaxSymbolDictionarySize));
+        Assert.That(d.Add("s42"), Is.EqualTo(42), "existing values remain usable at the cap");
+        Assert.Throws<IngressError>(() => d.Add("one-too-many".AsSpan()));
+        Assert.That(d.Count, Is.EqualTo(QwpConstants.MaxSymbolDictionarySize),
+            "a refused value must not mutate the dictionary");
+    }
+
+    [Test]
+    public void MaxSymbolDictionarySize_IsPinnedAtOrBelowServerProtocolLimit()
+    {
+        Assert.That(QwpConstants.MaxSymbolDictionarySize, Is.EqualTo(1_000_000));
+    }
+
     [Test]
     public void Add_NullString_MatchesEmptySpanAndDoesNotThrow()
     {
@@ -151,6 +181,28 @@ public void RollbackTo_BelowCommittedWatermark_Throws()
         Assert.Throws<ArgumentOutOfRangeException>(() => d.RollbackTo(0));
     }
 
+    [Test]
+    public void AddRecovered_DuplicateAcrossRollback_NeverLeavesDanglingReverseMapping()
+    {
+        var d = new QwpSymbolDictionary();
+        Assert.That(d.AddRecovered("x"), Is.EqualTo(0), "recovery must not de-duplicate");
+        Assert.That(d.AddRecovered("x"), Is.EqualTo(1));
+        Assert.That(d.AddRecovered("y"), Is.EqualTo(2));
+
+        d.RollbackTo(2);
+        var idAboveDuplicate = d.Add("x");
+        Assert.That(idAboveDuplicate, Is.InRange(0, 1), "a live duplicate id must resolve, never the dropped one");
+        Assert.That(d.GetSymbol(idAboveDuplicate), Is.EqualTo("x"));
+
+        d.RollbackTo(1);
+        Assert.That(d.Count, Is.EqualTo(1));
+        Assert.That(d.GetSymbol(0), Is.EqualTo("x"), "the surviving duplicate keeps its value");
+        var idAfterRollback = d.Add("x");
+        Assert.That(idAfterRollback, Is.LessThan(d.Count),
+            "rolling back across a recovered duplicate must not leave the reverse lookup on a dead id");
+        Assert.That(d.GetSymbol(idAfterRollback), Is.EqualTo("x"));
+    }
+
     [Test]
     public void Reset_ClearsEverything()
     {