Sqlite IPC Rework (#502)

* Use TEXT instead of VARCHAR

* Force ascending aliased rowid values

* Dispose rented connection and command early

* Serialize to bytes early

* Notify subscribers AFTER reading

* Update test

* Rework SqliteIPC

* Update

* Use Queue instead of Stack

* Cleanup ReaderLoop

* Seal SqliteIPC

* Add WriterLoop

* Publish jobs early when in-process

* ReaderLoop is for other processes

* Cleanup

* Cleanup

* Update messages

* Add EnsureWrite

Author: erri120

src/NexusMods.CLI/Types/IpcHandlers/NxmIpcProtocolHandler.cs

@@ -13,7 +13,7 @@ public class NxmIpcProtocolHandler : IIpcProtocolHandler
     /// <inheritdoc/>
     public string Protocol => "nxm";
 
-    private IMessageProducer<NXMUrlMessage> _messages;
+    private readonly IMessageProducer<NXMUrlMessage> _messages;
 
     /// <summary>
     /// constructor
@@ -27,6 +27,7 @@ public NxmIpcProtocolHandler(IMessageProducer<NXMUrlMessage> messages)
     public async Task Handle(string url, CancellationToken cancel)
     {
         await _messages.Write(new NXMUrlMessage { Value = NXMUrl.Parse(url) }, cancel);
+        _messages.EnsureWrite(cancel);
     }
 }
 

src/NexusMods.DataModel/Interprocess/IMessageProducer.cs

@@ -4,12 +4,19 @@ namespace NexusMods.DataModel.Interprocess;
 /// A message producer for sending messages to other processes.
 /// </summary>
 /// <typeparam name="T">Message type to send, each message type gets its own queue</typeparam>
-public interface IMessageProducer<T> where T : IMessage
+public interface IMessageProducer<in T> where T : IMessage
 {
     /// <summary>
     /// Sends a message to the queue.
     /// </summary>
     /// <param name="message">The message to write.</param>
     /// <param name="token">Can be used to cancel this operation.</param>
     public ValueTask Write(T message, CancellationToken token);
+
+    /// <summary>
+    /// Ensures all messages in-flight have been saved to the database.
+    /// This is done using an <see cref="EventWaitHandle"/> and this call
+    /// will block the entire thread.
+    /// </summary>
+    public void EnsureWrite(CancellationToken cancellationToken);
 }

src/NexusMods.DataModel/Interprocess/InterprocessProducer.cs

@@ -1,3 +1,6 @@
+using System.Diagnostics;
+using System.Reactive.Linq;
+
 namespace NexusMods.DataModel.Interprocess;
 
 /// <summary>
@@ -48,4 +51,21 @@ private void WriteInner(T message)
         var used = message.Write(buffer);
         _sqliteIpc.Send(_queueName, buffer[..used]);
     }
+
+    /// <inheritdoc/>
+    public void EnsureWrite(CancellationToken cancellationToken)
+    {
+        var timeout = SqliteIPC.WriterLoopInterval + SqliteIPC.SqliteDefaultTimeout;
+        Thread.Sleep(timeout);
+
+        const int maxIterations = 10;
+        var i = 0;
+
+        while (i < maxIterations && !cancellationToken.IsCancellationRequested)
+        {
+            var signaled = _sqliteIpc.WriterLoopFinished.WaitOne(timeout);
+            if (signaled) break;
+            i += 1;
+        }
+    }
 }

src/NexusMods.DataModel/Interprocess/SemaphoreSlimWaiter.cs

@@ -0,0 +1,32 @@
+namespace NexusMods.DataModel.Interprocess;
+
+internal readonly struct SemaphoreSlimWaiter : IDisposable
+{
+    private readonly SemaphoreSlim _semaphoreSlim;
+
+    public bool HasEntered { get; }
+
+    internal SemaphoreSlimWaiter(SemaphoreSlim semaphoreSlim, bool entered)
+    {
+        _semaphoreSlim = semaphoreSlim;
+        HasEntered = entered;
+    }
+
+    public void Dispose()
+    {
+        if (!HasEntered) return;
+        _semaphoreSlim.Release();
+    }
+}
+
+internal static class SemaphoreExtensions
+{
+    public static SemaphoreSlimWaiter CustomWait(
+        this SemaphoreSlim semaphoreSlim,
+        TimeSpan timeout,
+        CancellationToken cancellationToken = default)
+    {
+        var entered = semaphoreSlim.Wait(timeout, cancellationToken);
+        return new SemaphoreSlimWaiter(semaphoreSlim, entered);
+    }
+}