From 499a37ac78cd00eadbd355b9ded195b6efc1d6f4 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 8 Jul 2026 21:01:53 -0400 Subject: [PATCH] perf(store-and-forward): enable WAL + per-connection busy_timeout on the S&F SQLite store --- .../StoreAndForwardStorage.cs | 73 +++++++++++-------- .../StoreAndForwardStorageTests.cs | 27 +++++++ 2 files changed, 70 insertions(+), 30 deletions(-) diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs index 7c94ae58..dc0bc06d 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs +++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs @@ -33,8 +33,18 @@ public class StoreAndForwardStorage { EnsureDatabaseDirectoryExists(); - await using var connection = new SqliteConnection(_connectionString); - await connection.OpenAsync(); + await using var connection = await OpenConnectionAsync(); + + // 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 } } + /// + /// 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). + /// + private async Task 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; + } + /// /// Enqueues a new message with Pending status. /// @@ -136,8 +163,7 @@ public class StoreAndForwardStorage /// A task that represents the asynchronous operation. public async Task EnqueueAsync(StoreAndForwardMessage message) { - await using var connection = new SqliteConnection(_connectionString); - await connection.OpenAsync(); + await using var connection = await OpenConnectionAsync(); await using var cmd = connection.CreateCommand(); cmd.CommandText = @" @@ -189,8 +215,7 @@ public class StoreAndForwardStorage /// A task that represents the asynchronous operation. public async Task UpsertMessageAsync(StoreAndForwardMessage message) { - await using var connection = new SqliteConnection(_connectionString); - await connection.OpenAsync(); + await using var connection = await OpenConnectionAsync(); await using var cmd = connection.CreateCommand(); cmd.CommandText = @" @@ -243,8 +268,7 @@ public class StoreAndForwardStorage /// A task that resolves to the list of messages due for retry, ordered by creation time ascending. public async Task> GetMessagesForRetryAsync(int limit = 0) { - await using var connection = new SqliteConnection(_connectionString); - await connection.OpenAsync(); + await using var connection = await OpenConnectionAsync(); await using var cmd = connection.CreateCommand(); cmd.CommandText = @" @@ -277,8 +301,7 @@ public class StoreAndForwardStorage /// A task that represents the asynchronous operation. public async Task UpdateMessageAsync(StoreAndForwardMessage message) { - await using var connection = new SqliteConnection(_connectionString); - await connection.OpenAsync(); + await using var connection = await OpenConnectionAsync(); 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 connection.OpenAsync(); + await using var connection = await OpenConnectionAsync(); await using var cmd = connection.CreateCommand(); cmd.CommandText = @" @@ -343,8 +365,7 @@ public class StoreAndForwardStorage /// A task that represents the asynchronous operation. public async Task RemoveMessageAsync(string messageId) { - await using var connection = new SqliteConnection(_connectionString); - await connection.OpenAsync(); + await using var connection = await OpenConnectionAsync(); 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 connection.OpenAsync(); + await using var connection = await OpenConnectionAsync(); await using var transaction = (SqliteTransaction)await connection.BeginTransactionAsync(); @@ -411,8 +431,7 @@ public class StoreAndForwardStorage /// A task that resolves to true if the message was found and reset to Pending; false if not found or not in Parked status. public async Task RetryParkedMessageAsync(string messageId) { - await using var connection = new SqliteConnection(_connectionString); - await connection.OpenAsync(); + await using var connection = await OpenConnectionAsync(); await using var cmd = connection.CreateCommand(); cmd.CommandText = @" @@ -435,8 +454,7 @@ public class StoreAndForwardStorage /// A task that resolves to true if the message was found and deleted; false if not found or not in Parked status. public async Task DiscardParkedMessageAsync(string messageId) { - await using var connection = new SqliteConnection(_connectionString); - await connection.OpenAsync(); + await using var connection = await OpenConnectionAsync(); 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 /// A task that resolves to a dictionary mapping each category to its pending message count. public async Task> GetBufferDepthByCategoryAsync() { - await using var connection = new SqliteConnection(_connectionString); - await connection.OpenAsync(); + await using var connection = await OpenConnectionAsync(); await using var cmd = connection.CreateCommand(); cmd.CommandText = @" @@ -484,8 +501,7 @@ public class StoreAndForwardStorage /// A task that resolves to the number of messages whose origin instance matches . public async Task GetMessageCountByOriginInstanceAsync(string instanceName) { - await using var connection = new SqliteConnection(_connectionString); - await connection.OpenAsync(); + await using var connection = await OpenConnectionAsync(); await using var cmd = connection.CreateCommand(); cmd.CommandText = @" @@ -504,8 +520,7 @@ public class StoreAndForwardStorage /// A task that resolves to the matching message, or null if not found. public async Task GetMessageByIdAsync(string messageId) { - await using var connection = new SqliteConnection(_connectionString); - await connection.OpenAsync(); + await using var connection = await OpenConnectionAsync(); await using var cmd = connection.CreateCommand(); cmd.CommandText = @" @@ -526,8 +541,7 @@ public class StoreAndForwardStorage /// A task that resolves to the number of messages currently in Parked status. public async Task GetParkedMessageCountAsync() { - await using var conn = new SqliteConnection(_connectionString); - await conn.OpenAsync(); + await using var conn = await OpenConnectionAsync(); 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 /// A task that resolves to the count of messages with the specified status. public async Task GetMessageCountByStatusAsync(StoreAndForwardMessageStatus status) { - await using var connection = new SqliteConnection(_connectionString); - await connection.OpenAsync(); + await using var connection = await OpenConnectionAsync(); await using var cmd = connection.CreateCommand(); cmd.CommandText = "SELECT COUNT(*) FROM sf_messages WHERE status = @status"; diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs index e7d0f704..acc9b82c 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs +++ b/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.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); + } + } }