perf(store-and-forward): enable WAL + per-connection busy_timeout on the S&F SQLite store

src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs

@@ -33,8 +33,18 @@ public class StoreAndForwardStorage
     {
         EnsureDatabaseDirectoryExists();
 
-        await using var connection = new SqliteConnection(_connectionString);
+        await using var connection = await OpenConnectionAsync();
-        await connection.OpenAsync();
+
+        // Enable WAL so the concurrent writers this store has by design (script
+        // enqueues, the retry sweep's per-target lanes, standby replication applies,
+        // central pull queries) read/write without "database is locked". WAL is
+        // persistent + file-scoped; in-memory DBs report "memory" instead — harmless,
+        // so it is not asserted here.
+        await using (var walCmd = connection.CreateCommand())
+        {
+            walCmd.CommandText = "PRAGMA journal_mode=WAL";
+            await walCmd.ExecuteNonQueryAsync();
+        }
 
         await using var command = connection.CreateCommand();
         command.CommandText = @"
@@ -129,6 +139,23 @@ public class StoreAndForwardStorage
         }
     }
 
+    /// <summary>
+    /// Opens a connection with a 5s busy_timeout. Concurrent writers exist by
+    /// design (script enqueues, the sweep's lanes, standby replication applies,
+    /// central pull queries); with connection-per-operation the pragma must be
+    /// set per open (it is per-connection, and Microsoft.Data.Sqlite pooling
+    /// makes the extra statement cheap on a pooled physical connection).
+    /// </summary>
+    private async Task<SqliteConnection> OpenConnectionAsync()
+    {
+        var connection = new SqliteConnection(_connectionString);
+        await connection.OpenAsync();
+        await using var pragma = connection.CreateCommand();
+        pragma.CommandText = "PRAGMA busy_timeout = 5000";
+        await pragma.ExecuteNonQueryAsync();
+        return connection;
+    }
+
     /// <summary>
     /// Enqueues a new message with Pending status.
     /// </summary>
@@ -136,8 +163,7 @@ public class StoreAndForwardStorage
     /// <returns>A task that represents the asynchronous operation.</returns>
     public async Task EnqueueAsync(StoreAndForwardMessage message)
     {
-        await using var connection = new SqliteConnection(_connectionString);
+        await using var connection = await OpenConnectionAsync();
-        await connection.OpenAsync();
 
         await using var cmd = connection.CreateCommand();
         cmd.CommandText = @"
@@ -189,8 +215,7 @@ public class StoreAndForwardStorage
     /// <returns>A task that represents the asynchronous operation.</returns>
     public async Task UpsertMessageAsync(StoreAndForwardMessage message)
     {
-        await using var connection = new SqliteConnection(_connectionString);
+        await using var connection = await OpenConnectionAsync();
-        await connection.OpenAsync();
 
         await using var cmd = connection.CreateCommand();
         cmd.CommandText = @"
@@ -243,8 +268,7 @@ public class StoreAndForwardStorage
     /// <returns>A task that resolves to the list of messages due for retry, ordered by creation time ascending.</returns>
     public async Task<List<StoreAndForwardMessage>> GetMessagesForRetryAsync(int limit = 0)
     {
-        await using var connection = new SqliteConnection(_connectionString);
+        await using var connection = await OpenConnectionAsync();
-        await connection.OpenAsync();
 
         await using var cmd = connection.CreateCommand();
         cmd.CommandText = @"
@@ -277,8 +301,7 @@ public class StoreAndForwardStorage
     /// <returns>A task that represents the asynchronous operation.</returns>
     public async Task UpdateMessageAsync(StoreAndForwardMessage message)
     {
-        await using var connection = new SqliteConnection(_connectionString);
+        await using var connection = await OpenConnectionAsync();
-        await connection.OpenAsync();
 
         await using var cmd = connection.CreateCommand();
         cmd.CommandText = @"
@@ -312,8 +335,7 @@ public class StoreAndForwardStorage
         StoreAndForwardMessage message,
         StoreAndForwardMessageStatus expectedStatus)
     {
-        await using var connection = new SqliteConnection(_connectionString);
+        await using var connection = await OpenConnectionAsync();
-        await connection.OpenAsync();
 
         await using var cmd = connection.CreateCommand();
         cmd.CommandText = @"
@@ -343,8 +365,7 @@ public class StoreAndForwardStorage
     /// <returns>A task that represents the asynchronous operation.</returns>
     public async Task RemoveMessageAsync(string messageId)
     {
-        await using var connection = new SqliteConnection(_connectionString);
+        await using var connection = await OpenConnectionAsync();
-        await connection.OpenAsync();
 
         await using var cmd = connection.CreateCommand();
         cmd.CommandText = "DELETE FROM sf_messages WHERE id = @id";
@@ -365,8 +386,7 @@ public class StoreAndForwardStorage
         int pageNumber = 1,
         int pageSize = 50)
     {
-        await using var connection = new SqliteConnection(_connectionString);
+        await using var connection = await OpenConnectionAsync();
-        await connection.OpenAsync();
 
         await using var transaction = (SqliteTransaction)await connection.BeginTransactionAsync();
 
@@ -411,8 +431,7 @@ public class StoreAndForwardStorage
     /// <returns>A task that resolves to <c>true</c> if the message was found and reset to Pending; <c>false</c> if not found or not in Parked status.</returns>
     public async Task<bool> RetryParkedMessageAsync(string messageId)
     {
-        await using var connection = new SqliteConnection(_connectionString);
+        await using var connection = await OpenConnectionAsync();
-        await connection.OpenAsync();
 
         await using var cmd = connection.CreateCommand();
         cmd.CommandText = @"
@@ -435,8 +454,7 @@ public class StoreAndForwardStorage
     /// <returns>A task that resolves to <c>true</c> if the message was found and deleted; <c>false</c> if not found or not in Parked status.</returns>
     public async Task<bool> DiscardParkedMessageAsync(string messageId)
     {
-        await using var connection = new SqliteConnection(_connectionString);
+        await using var connection = await OpenConnectionAsync();
-        await connection.OpenAsync();
 
         await using var cmd = connection.CreateCommand();
         cmd.CommandText = "DELETE FROM sf_messages WHERE id = @id AND status = @parked";
@@ -453,8 +471,7 @@ public class StoreAndForwardStorage
     /// <returns>A task that resolves to a dictionary mapping each category to its pending message count.</returns>
     public async Task<Dictionary<StoreAndForwardCategory, int>> GetBufferDepthByCategoryAsync()
     {
-        await using var connection = new SqliteConnection(_connectionString);
+        await using var connection = await OpenConnectionAsync();
-        await connection.OpenAsync();
 
         await using var cmd = connection.CreateCommand();
         cmd.CommandText = @"
@@ -484,8 +501,7 @@ public class StoreAndForwardStorage
     /// <returns>A task that resolves to the number of messages whose origin instance matches <paramref name="instanceName"/>.</returns>
     public async Task<int> GetMessageCountByOriginInstanceAsync(string instanceName)
     {
-        await using var connection = new SqliteConnection(_connectionString);
+        await using var connection = await OpenConnectionAsync();
-        await connection.OpenAsync();
 
         await using var cmd = connection.CreateCommand();
         cmd.CommandText = @"
@@ -504,8 +520,7 @@ public class StoreAndForwardStorage
     /// <returns>A task that resolves to the matching message, or <c>null</c> if not found.</returns>
     public async Task<StoreAndForwardMessage?> GetMessageByIdAsync(string messageId)
     {
-        await using var connection = new SqliteConnection(_connectionString);
+        await using var connection = await OpenConnectionAsync();
-        await connection.OpenAsync();
 
         await using var cmd = connection.CreateCommand();
         cmd.CommandText = @"
@@ -526,8 +541,7 @@ public class StoreAndForwardStorage
     /// <returns>A task that resolves to the number of messages currently in Parked status.</returns>
     public async Task<int> GetParkedMessageCountAsync()
     {
-        await using var conn = new SqliteConnection(_connectionString);
+        await using var conn = await OpenConnectionAsync();
-        await conn.OpenAsync();
         await using var cmd = conn.CreateCommand();
         cmd.CommandText = "SELECT COUNT(*) FROM sf_messages WHERE status = @parked";
         cmd.Parameters.AddWithValue("@parked", (int)StoreAndForwardMessageStatus.Parked);
@@ -542,8 +556,7 @@ public class StoreAndForwardStorage
     /// <returns>A task that resolves to the count of messages with the specified status.</returns>
     public async Task<int> GetMessageCountByStatusAsync(StoreAndForwardMessageStatus status)
     {
-        await using var connection = new SqliteConnection(_connectionString);
+        await using var connection = await OpenConnectionAsync();
-        await connection.OpenAsync();
 
         await using var cmd = connection.CreateCommand();
         cmd.CommandText = "SELECT COUNT(*) FROM sf_messages WHERE status = @status";

tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs

@@ -656,4 +656,31 @@ public class StoreAndForwardStorageTests : IAsyncLifetime, IDisposable
                 Directory.Delete(directory, recursive: true);
         }
     }
+
+    [Fact]
+    public async Task Initialize_EnablesWalJournalMode_OnFileDatabase()
+    {
+        // WAL lets the retry-sweep lanes, script enqueues, and standby replication
+        // applies read/write concurrently without "database is locked". journal_mode
+        // is persistent + file-scoped, so a fresh connection observes it.
+        var directory = Path.Combine(Path.GetTempPath(), "sf-wal-test-" + Guid.NewGuid().ToString("N"));
+        var path = Path.Combine(directory, "wal-test.db");
+        try
+        {
+            var storage = new StoreAndForwardStorage(
+                $"Data Source={path}", NullLogger<StoreAndForwardStorage>.Instance);
+            await storage.InitializeAsync();
+
+            await using var conn = new SqliteConnection($"Data Source={path}");
+            await conn.OpenAsync();
+            await using var cmd = conn.CreateCommand();
+            cmd.CommandText = "PRAGMA journal_mode";
+            Assert.Equal("wal", (string)(await cmd.ExecuteScalarAsync())!);
+        }
+        finally
+        {
+            if (Directory.Exists(directory))
+                Directory.Delete(directory, recursive: true);
+        }
+    }
 }