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);
+ }
+ }
}