From 12b8839f33801378910c82c93a7e2859f33bace8 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 17 Jul 2026 21:11:07 -0400 Subject: [PATCH] =?UTF-8?q?fix(localdb):=20Task=204=20review=20fixes=20?= =?UTF-8?q?=E2=80=94=20synchronous=20allow-list,=20dictionary=20params,=20?= =?UTF-8?q?async=20conn=20setup,=20dispose=20guard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/ZB.MOM.WW.LocalDb/ILocalDb.cs | 15 +++- .../Internal/SqliteLocalDb.cs | 79 ++++++++++++++----- .../SqliteLocalDbTests.cs | 74 +++++++++++++++++ 3 files changed, 148 insertions(+), 20 deletions(-) diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ILocalDb.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ILocalDb.cs index f8a9466..d864328 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ILocalDb.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ILocalDb.cs @@ -6,13 +6,24 @@ namespace ZB.MOM.WW.LocalDb; /// Consumer SQL API over the embedded SQLite local database. Every connection this /// hands out has the zb_hlc_next() UDF registered, so capture triggers on /// registered tables can stamp writes fail-closed. +/// SQL references parameters with the @name marker; supported parameter shapes are +/// an anonymous object / POCO (public instance properties bind as @PropertyName) or an +/// IDictionary<string, object?> (each entry binds as @Key). Null values bind as SQL NULL. /// public interface ILocalDb { - /// Executes a non-query statement; returns the affected row count. + /// + /// Executes a non-query statement; returns the affected row count. + /// may be an anonymous object / POCO (properties → @name) + /// or an IDictionary<string, object?> (entries → @key). + /// Task ExecuteAsync(string sql, object? parameters = null, CancellationToken ct = default); - /// Executes a query and projects each row through . + /// + /// Executes a query and projects each row through . + /// may be an anonymous object / POCO (properties → @name) + /// or an IDictionary<string, object?> (entries → @key). + /// Task> QueryAsync(string sql, Func map, object? parameters = null, CancellationToken ct = default); /// Begins a transaction pinned to a single UDF-registered connection. diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs index fc5084b..6efdfd7 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs @@ -12,27 +12,40 @@ namespace ZB.MOM.WW.LocalDb.Internal; /// internal sealed class SqliteLocalDb : ILocalDb, IDisposable { + // Allow-list because the value is interpolated into a PRAGMA statement (pragmas cannot be parameterized). + private static readonly string[] AllowedSynchronous = ["OFF", "NORMAL", "FULL", "EXTRA"]; + private static readonly ConcurrentDictionary PropertyCache = new(); - private readonly LocalDbOptions _options; + private readonly string _synchronous; + private readonly int _busyTimeoutMs; private readonly string _connectionString; + // Held open for the object lifetime: it anchors the WAL journal and is the connection + // Dispose flushes the HLC high-water mark through, so the flush cannot fail on reopen. private readonly SqliteConnection _master; private readonly HybridLogicalClock _clock; private readonly string _nodeId; + private bool _disposed; internal HybridLogicalClock Clock => _clock; internal string NodeId => _nodeId; public SqliteLocalDb(LocalDbOptions options) { - _options = options; + if (string.IsNullOrWhiteSpace(options.Path)) + throw new ArgumentException("LocalDbOptions.Path is required.", nameof(options)); + + _synchronous = Array.Find(AllowedSynchronous, + s => s.Equals(options.Synchronous, StringComparison.OrdinalIgnoreCase)) + ?? throw new ArgumentException( + $"LocalDbOptions.Synchronous must be one of {string.Join("/", AllowedSynchronous)}; got '{options.Synchronous}'.", + nameof(options)); + _busyTimeoutMs = options.BusyTimeoutMs; _connectionString = new SqliteConnectionStringBuilder { DataSource = options.Path }.ToString(); _master = new SqliteConnection(_connectionString); _master.Open(); - ExecuteRaw(_master, - $"PRAGMA journal_mode=WAL; PRAGMA synchronous={_options.Synchronous}; " + - $"PRAGMA busy_timeout={_options.BusyTimeoutMs}; PRAGMA foreign_keys=ON;"); + ExecuteRaw(_master, "PRAGMA journal_mode=WAL; " + PerConnectionPragmas()); LocalDbSchema.EnsureCreated(_master); @@ -46,21 +59,43 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable (_nodeId, _) = LocalDbSchema.ReadMeta(_master); } + // journal_mode is a durable DB property already set on the file; synchronous is per-connection. + private string PerConnectionPragmas() => + $"PRAGMA synchronous={_synchronous}; PRAGMA busy_timeout={_busyTimeoutMs}; PRAGMA foreign_keys=ON;"; + public SqliteConnection CreateConnection() { var conn = new SqliteConnection(_connectionString); conn.Open(); - // journal_mode is a durable DB property already set on the file; synchronous is per-connection. - ExecuteRaw(conn, - $"PRAGMA synchronous={_options.Synchronous}; " + - $"PRAGMA busy_timeout={_options.BusyTimeoutMs}; PRAGMA foreign_keys=ON;"); + ExecuteRaw(conn, PerConnectionPragmas()); conn.CreateFunction("zb_hlc_next", () => _clock.Next()); return conn; } + internal async Task CreateConnectionAsync(CancellationToken ct) + { + var conn = new SqliteConnection(_connectionString); + try + { + await conn.OpenAsync(ct); + await using (var cmd = conn.CreateCommand()) + { + cmd.CommandText = PerConnectionPragmas(); + await cmd.ExecuteNonQueryAsync(ct); + } + conn.CreateFunction("zb_hlc_next", () => _clock.Next()); + return conn; + } + catch + { + await conn.DisposeAsync(); + throw; + } + } + public async Task ExecuteAsync(string sql, object? parameters = null, CancellationToken ct = default) { - await using var conn = CreateConnection(); + await using var conn = await CreateConnectionAsync(ct); await using var cmd = conn.CreateCommand(); cmd.CommandText = sql; Bind(cmd, parameters); @@ -69,7 +104,7 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable public async Task> QueryAsync(string sql, Func map, object? parameters = null, CancellationToken ct = default) { - await using var conn = CreateConnection(); + await using var conn = await CreateConnectionAsync(ct); await using var cmd = conn.CreateCommand(); cmd.CommandText = sql; Bind(cmd, parameters); @@ -78,7 +113,7 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable public async Task BeginTransactionAsync(CancellationToken ct = default) { - var conn = CreateConnection(); + var conn = await CreateConnectionAsync(ct); try { var tx = (SqliteTransaction)await conn.BeginTransactionAsync(ct); @@ -93,6 +128,9 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable public void Dispose() { + if (_disposed) + return; + _disposed = true; LocalDbSchema.FlushHlcHighWater(_master, _clock.Current); _master.Dispose(); } @@ -108,13 +146,18 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable private static void Bind(SqliteCommand cmd, object? parameters) { - if (parameters is null) - return; - - foreach (var prop in PropertiesOf(parameters.GetType())) + switch (parameters) { - var value = prop.GetValue(parameters); - cmd.Parameters.AddWithValue("@" + prop.Name, value ?? DBNull.Value); + case null: + return; + case IDictionary dict: + foreach (var (key, value) in dict) + cmd.Parameters.AddWithValue("@" + key, value ?? DBNull.Value); + return; + default: + foreach (var prop in PropertiesOf(parameters.GetType())) + cmd.Parameters.AddWithValue("@" + prop.Name, prop.GetValue(parameters) ?? DBNull.Value); + return; } } diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SqliteLocalDbTests.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SqliteLocalDbTests.cs index fbec825..572ac1d 100644 --- a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SqliteLocalDbTests.cs +++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SqliteLocalDbTests.cs @@ -82,6 +82,80 @@ public sealed class SqliteLocalDbTests : IDisposable Assert.Equal(new[] { 1L }, ids); } + [Fact] + public async Task Transaction_DisposeWithoutCommit_RollsBack() + { + using var db = Create(); + await db.ExecuteAsync("CREATE TABLE t (id INTEGER PRIMARY KEY)"); + + await using (var tx = await db.BeginTransactionAsync()) + { + await tx.ExecuteAsync("INSERT INTO t (id) VALUES (@id)", new { id = 1 }); + } + + var count = await db.QueryAsync("SELECT COUNT(*) FROM t", r => r.GetInt64(0)); + Assert.Equal(0L, count[0]); + } + + [Fact] + public async Task Bind_GuidByteArrayAndNull_RoundTrip() + { + using var db = Create(); + await db.ExecuteAsync("CREATE TABLE t (id INTEGER PRIMARY KEY, g TEXT, blob BLOB, missing TEXT)"); + + var guid = Guid.NewGuid(); + var bytes = new byte[] { 1, 2, 3 }; + await db.ExecuteAsync( + "INSERT INTO t (id, g, blob, missing) VALUES (@id, @g, @blob, @missing)", + new { id = 1, g = guid, blob = bytes, missing = (string?)null }); + + var rows = await db.QueryAsync( + "SELECT g, blob, missing FROM t WHERE id = 1", + r => (Guid: r.GetGuid(0), Blob: (byte[])r.GetValue(1), MissingIsNull: r.IsDBNull(2))); + + Assert.Equal(guid, rows[0].Guid); + Assert.Equal(bytes, rows[0].Blob); + Assert.True(rows[0].MissingIsNull); + } + + [Fact] + public async Task Bind_DictionaryParameters_BindByKey() + { + using var db = Create(); + await db.ExecuteAsync("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)"); + + var parameters = new Dictionary { ["id"] = 7, ["name"] = null }; + await db.ExecuteAsync("INSERT INTO t (id, name) VALUES (@id, @name)", parameters); + + var rows = await db.QueryAsync( + "SELECT id, name FROM t", + r => (Id: r.GetInt64(0), NameIsNull: r.IsDBNull(1))); + + Assert.Equal(7L, rows[0].Id); + Assert.True(rows[0].NameIsNull); + } + + [Fact] + public void Ctor_EmptyPath_Throws() + { + Assert.Throws(() => new SqliteLocalDb(new LocalDbOptions { Path = " " })); + } + + [Fact] + public void Ctor_InvalidSynchronous_Throws() + { + Assert.Throws(() => + new SqliteLocalDb(new LocalDbOptions { Path = _path, Synchronous = "NORMAL; PRAGMA foreign_keys=OFF" })); + } + + [Fact] + public void Dispose_IsIdempotent() + { + var db = Create(); + db.Dispose(); + db.Dispose(); + } + [Fact] public async Task HlcUdf_RegisteredOnConnection() {