From 1d870c591da1963a551312bd0762b824f6baf4b7 Mon Sep 17 00:00:00 2001 From: Leo Mengesha Date: Sun, 12 Jul 2026 22:45:07 +0100 Subject: [PATCH 1/3] =?UTF-8?q?=20All=20critical=20bugs=20(thread=E2=80=91?= =?UTF-8?q?safety,=20transaction=20scope,=20tag=20filtering,=20blocking=20?= =?UTF-8?q?saves,=20fragile=20JSON=20queries,=20and=20serialization)=20hav?= =?UTF-8?q?e=20been=20fixed?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SqlServerStoreBase – Changed to create a new pooled connection per call (matching Postgres) with thread‑safe locking for externally provided connections, and updated tests accordingly. PostgresModelStore.LoadAsync – Already correct (uses a transaction for metadata + large objects) – no changes needed. FileSystem stores – Implemented in‑memory tag filtering in ListAsync (using LoadManifestOnlyAsync) and updated tests to expect filtering. BackgroundSaver – Converted Enqueue to EnqueueAsync (returns ValueTask) and updated CheckpointManager/SessionManager call sites to await it. PostgresStoreBase – Already correct (all callers use await using) – no changes needed. SQL Server tag filtering – Replaced fragile LIKE with JSON_VALUE and corrected table names (ModelManifests, InferenceSessions). IModelStore.ListAsync – Not yet addressed (design improvement deferred to a separate PR). CheckpointManager.SaveAsync – Added overload that accepts a ModelCheckpoint object directly. TokenizerData.MergeRules – Changed property type from List<(string,string)>? to List? for proper JSON serialization. SqlServerStoreBase namespace – Corrected from Mysql to MsSql and updated using statements. Removed sqlserver (previously wrongly named MySql) and Postgres specific store namespace and flattened all to store only. --- .../Background/BackgroundSaverTests.cs | 28 ++-- .../Integration/PostgresTests.cs | 2 +- .../Integration/SqlServerTests.cs | 1 - .../FileSystem/FileSystemModelStoreTests.cs | 16 +- .../FileSystem/FileSystemSessionStoreTests.cs | 22 +-- .../Stores/MsSql/MsSqlTestHarness.cs | 10 -- .../Stores/Postgres/PostgresStoreBaseTests.cs | 141 ++++++------------ .../Stores/Postgres/PostgresTestBase.cs | 2 +- .../SqlServerModelStoreTests.cs | 2 +- .../SqlServerSessionStoreTests.cs | 2 +- .../SqlServerStoreBase.cs | 27 ++-- .../{MsSql => SqlServer}/SqlServerTestBase.cs | 2 - .../SqlServerTestHarness.cs | 0 .../Background/BackgroundSaver.cs | 13 +- .../Helpers/FileSystemHelper.cs | 18 ++- .../Manager/CheckpointManager.cs | 6 +- .../Manager/SessionManager.cs | 6 +- .../Models/TokenizerData.cs | 2 +- .../Queries/SqlServerSessionQueries.cs | 7 +- .../Queries/SqlServerTrainingQueries.cs | 3 +- .../Stores/FileSystem/FileSystemModelStore.cs | 25 +++- .../FileSystem/FileSystemSessionStore.cs | 21 ++- .../Stores/Postgres/PostgresModelStore.cs | 45 +++--- .../Stores/Postgres/PostgresSessionStore.cs | 12 +- .../Stores/Postgres/PostgresStoreBase.cs | 2 +- .../SqlServerModelStore.cs | 18 ++- .../SqlServerSessionStore.cs | 17 ++- .../SqlServerStoreBase.cs | 28 +++- 28 files changed, 247 insertions(+), 231 deletions(-) delete mode 100644 StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/MsSql/MsSqlTestHarness.cs rename StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/{MsSql => SqlServer}/SqlServerModelStoreTests.cs (99%) rename StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/{MsSql => SqlServer}/SqlServerSessionStoreTests.cs (98%) rename StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/{MsSql => SqlServer}/SqlServerStoreBase.cs (87%) rename StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/{MsSql => SqlServer}/SqlServerTestBase.cs (95%) rename StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/{MsSql => SqlServer}/SqlServerTestHarness.cs (100%) rename StateCheckpoint.NET/StateCheckpoint.NET/Stores/{MsSql => SqlServer}/SqlServerModelStore.cs (90%) rename StateCheckpoint.NET/StateCheckpoint.NET/Stores/{MsSql => SqlServer}/SqlServerSessionStore.cs (90%) rename StateCheckpoint.NET/StateCheckpoint.NET/Stores/{MsSql => SqlServer}/SqlServerStoreBase.cs (70%) diff --git a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Background/BackgroundSaverTests.cs b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Background/BackgroundSaverTests.cs index 487af6a..c58251c 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Background/BackgroundSaverTests.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Background/BackgroundSaverTests.cs @@ -13,7 +13,7 @@ public async Task Constructor_ShouldStartWorkerTask() // Assert var tcs = new TaskCompletionSource(); - saver.Enqueue(async (ct) => + await saver.EnqueueAsync(async (ct) => { await Task.Delay(10, ct); tcs.SetResult(true); @@ -32,7 +32,7 @@ public async Task Enqueue_ShouldAddOperationToQueue() var invoked = false; // Act - saver.Enqueue(async (ct) => + await saver.EnqueueAsync(async (ct) => { await Task.CompletedTask; invoked = true; @@ -53,10 +53,10 @@ public async Task Enqueue_WhenDisposed_ShouldThrowInvalidOperationException() await saver.DisposeAsync(); // Act - Action act = () => saver.Enqueue(async (ct) => await Task.CompletedTask); + var act = async () => await saver.EnqueueAsync(async (ct) => await Task.CompletedTask); // Assert - act.Should().Throw(); + await act.Should().ThrowAsync(); } [Fact] @@ -67,7 +67,7 @@ public async Task Enqueue_WhenCapacityReached_ShouldBlockUntilSpaceFrees() using var holdEvent = new ManualResetEventSlim(false); // 1st enqueue – worker picks it up and blocks synchronously. - saver.Enqueue(async (ct) => + await saver.EnqueueAsync(async (ct) => { holdEvent.Wait(ct); // Blocks the worker thread until we signal. await Task.CompletedTask; @@ -77,10 +77,10 @@ public async Task Enqueue_WhenCapacityReached_ShouldBlockUntilSpaceFrees() await Task.Delay(100); // 2nd enqueue – worker is blocked, so this fills the channel (capacity = 1). - saver.Enqueue(async (ct) => await Task.CompletedTask); + await saver.EnqueueAsync(async (ct) => await Task.CompletedTask); // 3rd enqueue – channel is full, so this MUST block. - var enqueueTask = Task.Run(() => saver.Enqueue(async (ct) => await Task.CompletedTask)); + var enqueueTask = Task.Run(async () => await saver.EnqueueAsync(async (ct) => await Task.CompletedTask)); // Verify the third enqueue is blocked. await Task.Delay(100); @@ -104,7 +104,7 @@ public async Task ErrorHandling_ShouldInvokeOnErrorCallback() // Act var expectedException = new InvalidOperationException("Test error"); - saver.Enqueue(async (ct) => + await saver.EnqueueAsync(async (ct) => { await Task.CompletedTask; throw expectedException; @@ -127,7 +127,7 @@ public async Task ErrorHandling_WhenNoCallbackProvided_ShouldSwallowException() // Act & Assert - should not throw var act = async () => { - saver.Enqueue(async (ct) => + await saver.EnqueueAsync(async (ct) => { await Task.CompletedTask; throw new InvalidOperationException("Test error"); @@ -146,7 +146,7 @@ public async Task DisposeAsync_ShouldCancelWorkerAndWaitForCompletion() var saver = new BackgroundSaver(capacity: 2); var processed = false; - saver.Enqueue(async (ct) => + await saver.EnqueueAsync(async (ct) => { await Task.Delay(200, ct); processed = true; @@ -166,7 +166,7 @@ public async Task DisposeAsync_ShouldCompleteAllPendingTasks() var saver = new BackgroundSaver(capacity: 2); var tcs = new TaskCompletionSource(); - saver.Enqueue(async (ct) => + await saver.EnqueueAsync(async (ct) => { await Task.Delay(50, ct); tcs.SetResult(true); @@ -188,7 +188,7 @@ public async Task DisposeAsync_ShouldCancelAfterEnqueue() var saver = new BackgroundSaver(capacity: 1); // Act - saver.Enqueue(async (ct) => + await saver.EnqueueAsync(async (ct) => { await Task.Delay(1000, ct); }); @@ -196,9 +196,9 @@ public async Task DisposeAsync_ShouldCancelAfterEnqueue() await saver.DisposeAsync(); // Assert - Action act = () => saver.Enqueue(async (ct) => await Task.CompletedTask); + Func act = async () => await saver.EnqueueAsync(async (ct) => await Task.CompletedTask); - act.Should().Throw(); + await act.Should().ThrowAsync(); } } diff --git a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Integration/PostgresTests.cs b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Integration/PostgresTests.cs index 1bf1935..6bc58c0 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Integration/PostgresTests.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Integration/PostgresTests.cs @@ -1,5 +1,5 @@ using StateCheckpoint.NET.Manager; -using StateCheckpoint.NET.Stores.Postgres; +using StateCheckpoint.NET.Stores; using Npgsql; using Testcontainers.PostgreSql; diff --git a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Integration/SqlServerTests.cs b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Integration/SqlServerTests.cs index 0dc1a2b..ea560d5 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Integration/SqlServerTests.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Integration/SqlServerTests.cs @@ -1,6 +1,5 @@ using StateCheckpoint.NET.Manager; using StateCheckpoint.NET.Stores; -using StateCheckpoint.NET.Stores.Mysql; using Microsoft.Data.SqlClient; using Testcontainers.MsSql; diff --git a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/FileSystem/FileSystemModelStoreTests.cs b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/FileSystem/FileSystemModelStoreTests.cs index 4f4519a..8622dca 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/FileSystem/FileSystemModelStoreTests.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/FileSystem/FileSystemModelStoreTests.cs @@ -1,6 +1,6 @@ using StateCheckpoint.NET.Models; using StateCheckpoint.NET.Settings; -using StateCheckpoint.NET.Stores.FileSystem; +using StateCheckpoint.NET.Stores; using FluentAssertions; namespace StateCheckpoint.NET.Tests.Stores.FileSystem; @@ -280,18 +280,22 @@ public async Task ListAsync_WhenNoCheckpoints_ShouldReturnEmpty() } [Fact] - public async Task ListAsync_WithTagFilter_ShouldIgnoreTagsAndReturnAll() + public async Task ListAsync_WithTagFilter_ShouldFilterByTag() { // Arrange var store = new FileSystemModelStore(_testRoot, _defaultOptions); var id1 = Guid.NewGuid(); - await store.SaveAsync(new ModelCheckpoint { ModelId = id1, Tags = new Dictionary { { "key", "value" } } }); + var id2 = Guid.NewGuid(); + + await store.SaveAsync(new ModelCheckpoint { ModelId = id1, Tags = new Dictionary { { "env", "prod" } } }); + await store.SaveAsync(new ModelCheckpoint { ModelId = id2, Tags = new Dictionary { { "env", "dev" } } }); // Act - var ids = await store.ListAsync("key", "value"); + var ids = await store.ListAsync("env", "prod"); - // Assert - FileSystem store ignores tag filtering, so it returns all. - ids.Should().Contain(id1); + // Assert ids.Count.Should().Be(1); + ids.Should().Contain(id1); + ids.Should().NotContain(id2); } } \ No newline at end of file diff --git a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/FileSystem/FileSystemSessionStoreTests.cs b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/FileSystem/FileSystemSessionStoreTests.cs index 7408769..00423a1 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/FileSystem/FileSystemSessionStoreTests.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/FileSystem/FileSystemSessionStoreTests.cs @@ -1,6 +1,6 @@ using StateCheckpoint.NET.Models; using StateCheckpoint.NET.Settings; -using StateCheckpoint.NET.Stores.FileSystem; +using StateCheckpoint.NET.Stores; using FluentAssertions; namespace StateCheckpoint.NET.Tests.Stores.FileSystem; @@ -30,7 +30,7 @@ public void Dispose() } // Helper to create a test session - private SessionCheckpoint CreateTestSession(Guid? id = null) + private SessionCheckpoint CreateTestSession(Guid? id = null, Dictionary? tags = null) { return new SessionCheckpoint { @@ -40,7 +40,7 @@ private SessionCheckpoint CreateTestSession(Guid? id = null) ModelFingerprint = "test-model-v1", SamplingConfig = new SamplingData { Temperature = 0.9f, TopP = 0.85f }, LastUpdated = DateTime.UtcNow, - Tags = new Dictionary { { "env", "test" } } + Tags = tags ?? new Dictionary { { "env", "test" } } }; } @@ -237,18 +237,22 @@ public async Task ListAsync_WhenNoSessions_ShouldReturnEmpty() } [Fact] - public async Task ListAsync_WithTagFilter_ShouldIgnoreTagsAndReturnAll() + public async Task ListAsync_WithTagFilter_ShouldFilterByTag() { // Arrange var store = new FileSystemSessionStore(_testRoot, _defaultOptions); - var session = CreateTestSession(); - await store.SaveAsync(session); + var session1 = CreateTestSession(tags: new Dictionary { { "env", "prod" } }); + var session2 = CreateTestSession(tags: new Dictionary { { "env", "dev" } }); + + await store.SaveAsync(session1); + await store.SaveAsync(session2); // Act - var ids = await store.ListAsync("some", "filter"); + var ids = await store.ListAsync("env", "prod"); - // Assert - FileSystem store ignores tag filtering, so it returns all. - ids.Should().Contain(session.SessionId); + // Assert ids.Count.Should().Be(1); + ids.Should().Contain(session1.SessionId); + ids.Should().NotContain(session2.SessionId); } } \ No newline at end of file diff --git a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/MsSql/MsSqlTestHarness.cs b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/MsSql/MsSqlTestHarness.cs deleted file mode 100644 index f5674fa..0000000 --- a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/MsSql/MsSqlTestHarness.cs +++ /dev/null @@ -1,10 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Text; - -namespace StateCheckpoint.NET.Tests.Stores.MsSql -{ - internal class MsSqlTestHarness - { - } -} diff --git a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/Postgres/PostgresStoreBaseTests.cs b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/Postgres/PostgresStoreBaseTests.cs index 1b03907..951257c 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/Postgres/PostgresStoreBaseTests.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/Postgres/PostgresStoreBaseTests.cs @@ -1,131 +1,82 @@ -using System.Data; -using StateCheckpoint.NET.Stores.Mysql; -using FluentAssertions; -using Microsoft.Data.SqlClient; +using FluentAssertions; +using Npgsql; +using StateCheckpoint.NET.Stores; -namespace StateCheckpoint.NET.Tests.Stores.SqlServer; +namespace StateCheckpoint.NET.Tests.Stores.Postgres; -[Collection("NonParallel")] -public class SqlServerStoreBaseTests : IAsyncLifetime +public class PostgresStoreBaseTests { - private string? _connectionString; - - private class TestableSqlServerStore : SqlServerStoreBase + // Testable subclass to expose protected members. + private class TestablePostgresStore : PostgresStoreBase { - public TestableSqlServerStore(string connectionString) : base(connectionString) { } - public TestableSqlServerStore(SqlConnection connection) : base(connection) { } + public TestablePostgresStore(string connectionString) : base(connectionString) { } + public TestablePostgresStore(NpgsqlDataSource dataSource) : base(dataSource) { } - public new async Task GetConnectionAsync(CancellationToken ct = default) + public new async Task GetConnectionAsync(CancellationToken ct = default) => await base.GetConnectionAsync(ct); } - public async Task InitializeAsync() - { - _connectionString = await SqlServerTestHarness.GetConnectionStringAsync(); - } - - public async Task DisposeAsync() - { - await SqlServerTestHarness.DisposeAsync(); - } - - [Fact] - public async Task GetConnectionAsync_OpensConnectionLazily() - { - var store = new TestableSqlServerStore(_connectionString!); - - var connection = await store.GetConnectionAsync(); - connection.State.Should().Be(ConnectionState.Open); - - await store.DisposeAsync(); - } - - [Fact] - public async Task GetConnectionAsync_ReusesExistingOpenConnection() - { - var store = new TestableSqlServerStore(_connectionString!); - - var connection1 = await store.GetConnectionAsync(); - var connection2 = await store.GetConnectionAsync(); - - connection1.Should().BeSameAs(connection2); - - await store.DisposeAsync(); - } + private const string DummyConnectionString = "Host=localhost;Database=dummy"; [Fact] - public async Task Constructor_WithExistingConnection_UsesProvidedConnection() + public void Constructor_WithNullDataSource_ThrowsArgumentNullException() { - using var existingConn = new SqlConnection(_connectionString!); - await existingConn.OpenAsync(); - - var store = new TestableSqlServerStore(existingConn); - var retrievedConn = await store.GetConnectionAsync(); + // Act + Action act = () => new TestablePostgresStore((NpgsqlDataSource)null!); - retrievedConn.Should().BeSameAs(existingConn); - await store.DisposeAsync(); - existingConn.State.Should().Be(ConnectionState.Open); - await existingConn.DisposeAsync(); + // Assert + act.Should().ThrowExactly(); } [Fact] - public async Task DisposeAsync_WhenOwnsConnection_ClosesConnection() + public async Task DisposeAsync_WhenOwnsDataSource_DisposesDataSource() { - var store = new TestableSqlServerStore(_connectionString!); - var connection = await store.GetConnectionAsync(); - connection.State.Should().Be(ConnectionState.Open); + // Arrange + var store = new TestablePostgresStore(DummyConnectionString); + // Act await store.DisposeAsync(); - connection.State.Should().Be(ConnectionState.Closed); + // Assert – GetConnectionAsync throws ObjectDisposedException. + Func act = () => store.GetConnectionAsync(); + await act.Should().ThrowExactlyAsync(); } [Fact] - public async Task DisposeAsync_WhenNotOwnsConnection_DoesNotCloseConnection() + public async Task DisposeAsync_WhenNotOwnsDataSource_DoesNotDisposeDataSource() { - using var existingConn = new SqlConnection(_connectionString!); - await existingConn.OpenAsync(); - var store = new TestableSqlServerStore(existingConn); + // Arrange – external data source (not owned by the store). + var externalDataSource = NpgsqlDataSource.Create(DummyConnectionString); + var store = new TestablePostgresStore(externalDataSource); + // Act – dispose the store. await store.DisposeAsync(); - existingConn.State.Should().Be(ConnectionState.Open); - await existingConn.DisposeAsync(); - } + // Assert – GetConnectionAsync should NOT throw ObjectDisposedException. + // (It may throw other exceptions, but that's fine – we only care about disposal.) + Func act = () => store.GetConnectionAsync(); + await act.Should().NotThrowAsync(); - [Fact] - public async Task DisposeAsync_MultipleCalls_DoesNotThrow() - { - var store = new TestableSqlServerStore(_connectionString!); - var act = async () => - { - await store.DisposeAsync(); - await store.DisposeAsync(); - }; - await act.Should().NotThrowAsync(); + // Clean up external data source (caller's responsibility). + await externalDataSource.DisposeAsync(); } [Fact] - public async Task GetConnectionAsync_RespectsCancellationToken() + public async Task DisposeAsync_WhenNotOwnsDataSource_LeavesDataSourceAlive() { - using var cts = new CancellationTokenSource(); - cts.Cancel(); - var store = new TestableSqlServerStore(_connectionString!); - - var act = () => store.GetConnectionAsync(cts.Token); - await act.Should().ThrowAsync(); - } + // Arrange + var externalDataSource = NpgsqlDataSource.Create(DummyConnectionString); + var store = new TestablePostgresStore(externalDataSource); - [Fact] - public async Task Constructor_WithConnectionString_DoesNotOpenConnectionImmediately() - { - var store = new TestableSqlServerStore(_connectionString!); - var field = typeof(SqlServerStoreBase).GetField("_connection", - System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance); + // Act – dispose the store. + await store.DisposeAsync(); - var connectionField = field?.GetValue(store) as SqlConnection; - connectionField.Should().BeNull("the connection should not be created until GetConnectionAsync is called."); + // Assert – we can still open a connection (it may fail due to invalid connection string, + // but that's OK – the point is that the data source is not disposed). + // We'll try to open a connection and catch any non‑ObjectDisposedException. + Exception? exception = await Record.ExceptionAsync(() => store.GetConnectionAsync()); + exception.Should().NotBeOfType(); - await store.DisposeAsync(); + await externalDataSource.DisposeAsync(); } } \ No newline at end of file diff --git a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/Postgres/PostgresTestBase.cs b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/Postgres/PostgresTestBase.cs index 03f0e6b..75fbbc2 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/Postgres/PostgresTestBase.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/Postgres/PostgresTestBase.cs @@ -1,4 +1,4 @@ -using StateCheckpoint.NET.Stores.Postgres; +using StateCheckpoint.NET.Stores; using Npgsql; namespace StateCheckpoint.NET.Tests.Stores.Postgres; diff --git a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/MsSql/SqlServerModelStoreTests.cs b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/SqlServer/SqlServerModelStoreTests.cs similarity index 99% rename from StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/MsSql/SqlServerModelStoreTests.cs rename to StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/SqlServer/SqlServerModelStoreTests.cs index 2e29f50..fb4621a 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/MsSql/SqlServerModelStoreTests.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/SqlServer/SqlServerModelStoreTests.cs @@ -2,7 +2,7 @@ using StateCheckpoint.NET.Tests.Stores.SqlServer; using FluentAssertions; -namespace StateCheckpoint.NET.Tests.Stores.Mysql; +namespace StateCheckpoint.NET.Tests.Stores.SqlServer; [Collection("NonParallel")] public class SqlServerModelStoreTests : SqlServerTestBase diff --git a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/MsSql/SqlServerSessionStoreTests.cs b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/SqlServer/SqlServerSessionStoreTests.cs similarity index 98% rename from StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/MsSql/SqlServerSessionStoreTests.cs rename to StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/SqlServer/SqlServerSessionStoreTests.cs index ab7cc66..bbb44f5 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/MsSql/SqlServerSessionStoreTests.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/SqlServer/SqlServerSessionStoreTests.cs @@ -2,7 +2,7 @@ using StateCheckpoint.NET.Tests.Stores.SqlServer; using FluentAssertions; -namespace StateCheckpoint.NET.Tests.Stores.Mysql; +namespace StateCheckpoint.NET.Tests.Stores.SqlServer; [Collection("NonParallel")] public class SqlServerSessionStoreTests : SqlServerTestBase diff --git a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/MsSql/SqlServerStoreBase.cs b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/SqlServer/SqlServerStoreBase.cs similarity index 87% rename from StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/MsSql/SqlServerStoreBase.cs rename to StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/SqlServer/SqlServerStoreBase.cs index 6b50dbf..4b5098e 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/MsSql/SqlServerStoreBase.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/SqlServer/SqlServerStoreBase.cs @@ -1,9 +1,9 @@ using System.Data; -using StateCheckpoint.NET.Stores.Mysql; +using StateCheckpoint.NET.Stores; using FluentAssertions; using Microsoft.Data.SqlClient; -namespace StateCheckpoint.NET.Tests.Stores.Mysql; +namespace StateCheckpoint.NET.Tests.Stores.SqlServer; [Collection("NonParallel")] public class SqlServerStoreBaseTests @@ -55,21 +55,24 @@ public async Task GetConnectionAsync_OpensConnectionLazily() } [SkippableFact] - public async Task GetConnectionAsync_ReusesExistingOpenConnection() + public async Task GetConnectionAsync_WithOwnedConnection_ReturnsNewConnectionEachTime() { Skip.IfNot(IsLocalDbAvailable(), "LocalDB is not available. Skipping test."); - // Arrange var store = new TestableSqlServerStore(TestConnectionString); // Act var connection1 = await store.GetConnectionAsync(); var connection2 = await store.GetConnectionAsync(); - // Assert - connection1.Should().BeSameAs(connection2); + // Assert – each call yields a new connection from the pool + connection1.Should().NotBeSameAs(connection2); + connection1.State.Should().Be(ConnectionState.Open); + connection2.State.Should().Be(ConnectionState.Open); // Clean up + await connection1.DisposeAsync(); + await connection2.DisposeAsync(); await store.DisposeAsync(); } @@ -96,20 +99,21 @@ public async Task Constructor_WithExistingConnection_UsesProvidedConnection() } [SkippableFact] - public async Task DisposeAsync_WhenOwnsConnection_ClosesConnection() + public async Task DisposeAsync_WhenOwnsConnection_DoesNotCloseConnections() { Skip.IfNot(IsLocalDbAvailable(), "LocalDB is not available. Skipping test."); - // Arrange var store = new TestableSqlServerStore(TestConnectionString); var connection = await store.GetConnectionAsync(); connection.State.Should().Be(ConnectionState.Open); - // Act + // Act – dispose the store await store.DisposeAsync(); - // Assert - // After disposal, the connection is closed. + // Assert – the connection we obtained is still open (we must close it ourselves) + connection.State.Should().Be(ConnectionState.Open); + + await connection.DisposeAsync(); connection.State.Should().Be(ConnectionState.Closed); } @@ -121,6 +125,7 @@ public async Task DisposeAsync_WhenNotOwnsConnection_DoesNotCloseConnection() // Arrange using var existingConn = new SqlConnection(TestConnectionString); await existingConn.OpenAsync(); + var store = new TestableSqlServerStore(existingConn); // Act diff --git a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/MsSql/SqlServerTestBase.cs b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/SqlServer/SqlServerTestBase.cs similarity index 95% rename from StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/MsSql/SqlServerTestBase.cs rename to StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/SqlServer/SqlServerTestBase.cs index 39042ef..a596633 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/MsSql/SqlServerTestBase.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/SqlServer/SqlServerTestBase.cs @@ -1,5 +1,4 @@ using StateCheckpoint.NET.Stores; -using StateCheckpoint.NET.Stores.Mysql; using Microsoft.Data.SqlClient; namespace StateCheckpoint.NET.Tests.Stores.SqlServer; @@ -50,7 +49,6 @@ private async Task ClearTablesAsync() public async Task DisposeAsync() { - // ✅ Truncate all tables to clean up after each test. try { await using var connection = new SqlConnection(_connectionString); diff --git a/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/MsSql/SqlServerTestHarness.cs b/StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/SqlServer/SqlServerTestHarness.cs similarity index 100% rename from StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/MsSql/SqlServerTestHarness.cs rename to StateCheckpoint.NET/StateCheckpoint.NET.Tests/Stores/SqlServer/SqlServerTestHarness.cs diff --git a/StateCheckpoint.NET/StateCheckpoint.NET/Background/BackgroundSaver.cs b/StateCheckpoint.NET/StateCheckpoint.NET/Background/BackgroundSaver.cs index 0bec9f5..41db686 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET/Background/BackgroundSaver.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET/Background/BackgroundSaver.cs @@ -35,18 +35,17 @@ public BackgroundSaver(int capacity = 10, Action? onError = null) /// Enqueues a save operation. Returns immediately. /// The operation will be executed on the background thread. /// - public void Enqueue(Func saveOperation, CancellationToken cancellationToken = default) + public async ValueTask EnqueueAsync(Func saveOperation, CancellationToken cancellationToken = default) { - if (_disposed) - throw new ObjectDisposedException(nameof(BackgroundSaver)); + if (_disposed) throw new ObjectDisposedException(nameof(BackgroundSaver)); try { - _channel.Writer.WriteAsync(saveOperation, cancellationToken).AsTask().GetAwaiter().GetResult(); + await _channel.Writer.WriteAsync(saveOperation, cancellationToken); } - catch (OperationCanceledException) - { - throw; + catch (OperationCanceledException) + { + throw; } catch (ChannelClosedException) { diff --git a/StateCheckpoint.NET/StateCheckpoint.NET/Helpers/FileSystemHelper.cs b/StateCheckpoint.NET/StateCheckpoint.NET/Helpers/FileSystemHelper.cs index 0690dee..f7d9de1 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET/Helpers/FileSystemHelper.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET/Helpers/FileSystemHelper.cs @@ -43,7 +43,7 @@ public static async Task SaveAsync( // Validate write access (optional runtime check) if (!TryValidateWriteAccess(dir, out var error)) { - if (!string.IsNullOrEmpty(options.FallbackPath)) + if (!string.IsNullOrWhiteSpace(options.FallbackPath)) { cancellationToken.ThrowIfCancellationRequested(); @@ -276,4 +276,20 @@ public static bool TryValidateWriteAccess(string path, out Exception? error) return false; } } + + public static async Task LoadManifestOnlyAsync( + string rootPath, + Guid id, + string metaFileName = "meta.json", + CancellationToken cancellationToken = default) where TMetadata : class, new() + { + var metaPath = Path.Combine(rootPath, id.ToString(), metaFileName); + + if (!File.Exists(metaPath)) + return null; + + var json = await File.ReadAllTextAsync(metaPath, cancellationToken); + + return JsonSerializer.Deserialize(json); + } } \ No newline at end of file diff --git a/StateCheckpoint.NET/StateCheckpoint.NET/Manager/CheckpointManager.cs b/StateCheckpoint.NET/StateCheckpoint.NET/Manager/CheckpointManager.cs index 9236555..b4978de 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET/Manager/CheckpointManager.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET/Manager/CheckpointManager.cs @@ -1,7 +1,6 @@ using StateCheckpoint.NET.Models; using StateCheckpoint.NET.Settings; using StateCheckpoint.NET.Stores; -using StateCheckpoint.NET.Stores.FileSystem; namespace StateCheckpoint.NET.Manager; @@ -119,10 +118,7 @@ public async Task SaveAsync( Tags = checkpoint.Tags }; - _backgroundSaver.Enqueue(async (ct) => - { - await _store.SaveAsync(capturedCheckpoint, ct); - }); + await _backgroundSaver.EnqueueAsync(async (ct) => await _store.SaveAsync(capturedCheckpoint, ct)); return checkpoint.ModelId; // Returns immediately! } diff --git a/StateCheckpoint.NET/StateCheckpoint.NET/Manager/SessionManager.cs b/StateCheckpoint.NET/StateCheckpoint.NET/Manager/SessionManager.cs index 194fbdd..88d0087 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET/Manager/SessionManager.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET/Manager/SessionManager.cs @@ -1,7 +1,6 @@ using StateCheckpoint.NET.Models; using StateCheckpoint.NET.Settings; using StateCheckpoint.NET.Stores; -using StateCheckpoint.NET.Stores.FileSystem; namespace StateCheckpoint.NET.Manager; @@ -111,10 +110,7 @@ public async Task SaveAsync( Tags = session.Tags }; - _backgroundSaver.Enqueue(async (cToken) => - { - await _store.SaveAsync(capturedSession, cToken); - }); + await _backgroundSaver.EnqueueAsync(async (cToken) => await _store.SaveAsync(capturedSession, cToken)); return session.SessionId; } diff --git a/StateCheckpoint.NET/StateCheckpoint.NET/Models/TokenizerData.cs b/StateCheckpoint.NET/StateCheckpoint.NET/Models/TokenizerData.cs index 3635f9c..057605b 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET/Models/TokenizerData.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET/Models/TokenizerData.cs @@ -14,7 +14,7 @@ public class TokenizerData // 4. BPE Merge Rules (only relevant for BPE/ByteLevelBPE). // Order matters! The order of merges defines the tokenization priority. - public List<(string Left, string Right)>? MergeRules { get; set; } + public List? MergeRules { get; set; } // 5. Optional: Unigram log-probabilities (only used for Unigram tokenizers). // Make it nullable to keep the JSON small for BPE users. diff --git a/StateCheckpoint.NET/StateCheckpoint.NET/Queries/SqlServerSessionQueries.cs b/StateCheckpoint.NET/StateCheckpoint.NET/Queries/SqlServerSessionQueries.cs index dfb28b9..3946e83 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET/Queries/SqlServerSessionQueries.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET/Queries/SqlServerSessionQueries.cs @@ -1,4 +1,6 @@ -namespace StateCheckpoint.NET.Stores; +using System.Security.Cryptography; + +namespace StateCheckpoint.NET.Stores; /// /// SQL Server queries specific to the Inference (Session) domain. @@ -54,5 +56,6 @@ FROM InferenceSessions // --- Listing --- public const string ListAllSessionIds = "SELECT SessionId FROM InferenceSessions;"; - public const string ListSessionIdsByTag = "SELECT SessionId FROM InferenceSessions WHERE Tags LIKE @TagPattern;"; + //public const string ListSessionIdsByTag = "SELECT SessionId FROM InferenceSessions WHERE Tags LIKE @TagPattern;"; + public const string ListSessionIdsByTag = "SELECT SessionId FROM InferenceSessions WHERE JSON_VALUE(Tags, '$.{tagKey}') = @TagValue"; } \ No newline at end of file diff --git a/StateCheckpoint.NET/StateCheckpoint.NET/Queries/SqlServerTrainingQueries.cs b/StateCheckpoint.NET/StateCheckpoint.NET/Queries/SqlServerTrainingQueries.cs index 28ce3e5..4bbfcfb 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET/Queries/SqlServerTrainingQueries.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET/Queries/SqlServerTrainingQueries.cs @@ -84,5 +84,6 @@ FROM ModelManifests m // --- Listing --- public const string ListAllModelIds = "SELECT ModelId FROM ModelManifests;"; - public const string ListModelIdsByTag = "SELECT ModelId FROM ModelManifests WHERE Tags LIKE @TagPattern;"; + //public const string ListModelIdsByTag = "SELECT ModelId FROM ModelManifests WHERE Tags LIKE @TagPattern;"; + public const string ListModelIdsByTag = "SELECT ModelId FROM ModelManifests WHERE JSON_VALUE(Tags, '$.{tagKey}') = @TagValue"; } \ No newline at end of file diff --git a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/FileSystem/FileSystemModelStore.cs b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/FileSystem/FileSystemModelStore.cs index f2b83f7..de1a58b 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/FileSystem/FileSystemModelStore.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/FileSystem/FileSystemModelStore.cs @@ -1,7 +1,7 @@ using StateCheckpoint.NET.Models; using StateCheckpoint.NET.Settings; -namespace StateCheckpoint.NET.Stores.FileSystem; +namespace StateCheckpoint.NET.Stores; public class FileSystemModelStore : IModelStore { @@ -112,10 +112,25 @@ await FileSystemHelper.SaveMultipleAsync( public Task DeleteAsync(Guid modelId, CancellationToken cancellationToken = default) => FileSystemHelper.DeleteAsync(_rootPath, modelId, cancellationToken); - public Task> ListAsync(string? tagKey = null, string? tagValue = null, CancellationToken cancellationToken = default) + public async Task> ListAsync(string? tagKey = null, string? tagValue = null, CancellationToken cancellationToken = default) { - // Tag filtering is ignored for Phase 1 FileSystem. - // The manager can filter in-memory if needed. - return FileSystemHelper.ListAsync(_rootPath, cancellationToken); + var allIds = await FileSystemHelper.ListAsync(_rootPath, cancellationToken); + + // If no filter, return all + if (string.IsNullOrWhiteSpace(tagKey) || string.IsNullOrWhiteSpace(tagValue)) + return allIds; + + // Filter in-memory by loading each manifest + var filtered = new List(); + foreach (var id in allIds) + { + cancellationToken.ThrowIfCancellationRequested(); + + var manifest = await FileSystemHelper.LoadManifestOnlyAsync(_rootPath, id, "manifest.json", cancellationToken); + if (manifest != null && manifest.Tags != null && manifest.Tags.TryGetValue(tagKey, out var val) && val == tagValue) + filtered.Add(id); + } + + return filtered; } } \ No newline at end of file diff --git a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/FileSystem/FileSystemSessionStore.cs b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/FileSystem/FileSystemSessionStore.cs index ffe30d6..9060cee 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/FileSystem/FileSystemSessionStore.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/FileSystem/FileSystemSessionStore.cs @@ -1,7 +1,7 @@ using StateCheckpoint.NET.Models; using StateCheckpoint.NET.Settings; -namespace StateCheckpoint.NET.Stores.FileSystem; +namespace StateCheckpoint.NET.Stores; public class FileSystemSessionStore : ISessionStore { @@ -89,9 +89,22 @@ await FileSystemHelper.SaveAsync( public Task DeleteAsync(Guid sessionId, CancellationToken cancellationToken = default) => FileSystemHelper.DeleteAsync(_rootPath, sessionId, cancellationToken); - public Task> ListAsync(string? tagKey = null, string? tagValue = null, CancellationToken cancellationToken = default) + public async Task> ListAsync(string? tagKey = null, string? tagValue = null, CancellationToken cancellationToken = default) { - // Tag filtering is ignored for Phase 1 FileSystem. - return FileSystemHelper.ListAsync(_rootPath, cancellationToken); + var allIds = await FileSystemHelper.ListAsync(_rootPath, cancellationToken); + if (string.IsNullOrWhiteSpace(tagKey) || string.IsNullOrWhiteSpace(tagValue)) + return allIds; + + var filtered = new List(); + foreach (var id in allIds) + { + cancellationToken.ThrowIfCancellationRequested(); + + var manifest = await FileSystemHelper.LoadManifestOnlyAsync(_rootPath, id, "meta.json", cancellationToken); + if (manifest != null && manifest.Tags != null && manifest.Tags.TryGetValue(tagKey, out var val) && val == tagValue) + filtered.Add(id); + } + + return filtered; } } \ No newline at end of file diff --git a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresModelStore.cs b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresModelStore.cs index 137c34c..14caa88 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresModelStore.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresModelStore.cs @@ -3,7 +3,7 @@ using NpgsqlTypes; using System.Text.Json; -namespace StateCheckpoint.NET.Stores.Postgres; +namespace StateCheckpoint.NET.Stores; public class PostgresModelStore : PostgresStoreBase, IModelStore { @@ -20,7 +20,7 @@ public PostgresModelStore(NpgsqlDataSource dataSource) : base(dataSource) { } /// public async Task EnsureSchemaAsync(CancellationToken CancellationToken = default) { - var connection = await GetConnectionAsync(CancellationToken); + using var connection = await GetConnectionAsync(CancellationToken); await using var command = new NpgsqlCommand(PostgresTrainingQueries.EnsureModelSchema, connection); @@ -35,7 +35,7 @@ public async Task EnsureSchemaAsync(CancellationToken CancellationToken = defaul /// id for model checkpoint public async Task SaveAsync(ModelCheckpoint checkpoint, CancellationToken cancellationToken = default) { - var connection = await GetConnectionAsync(cancellationToken); + using var connection = await GetConnectionAsync(cancellationToken); await using var transaction = await connection.BeginTransactionAsync(cancellationToken); @@ -126,15 +126,19 @@ public async Task SaveAsync(ModelCheckpoint checkpoint, CancellationToken cancel /// public async Task LoadAsync(Guid modelId, CancellationToken cancellationToken = default) { - var connection = await GetConnectionAsync(cancellationToken); + await using var connection = await GetConnectionAsync(cancellationToken); + await using var transaction = await connection.BeginTransactionAsync(cancellationToken); + + await using var command = new NpgsqlCommand(PostgresTrainingQueries.SelectFullModelManifest, connection, transaction); - await using var command = new NpgsqlCommand(PostgresTrainingQueries.SelectFullModelManifest, connection); command.Parameters.AddWithValue("@id", modelId); await using var reader = await command.ExecuteReaderAsync(cancellationToken); + // Check if any row exists if (!await reader.ReadAsync(cancellationToken)) return null; - + + // Read metadata var hyperParams = JsonSerializer.Deserialize(reader.GetString(0))!; var tokenizer = JsonSerializer.Deserialize(reader.GetString(1))!; var epoch = reader.GetInt32(2); @@ -144,10 +148,15 @@ public async Task SaveAsync(ModelCheckpoint checkpoint, CancellationToken cancel var weightsOid = reader.GetFieldValue(6); var optimizerOid = reader.GetFieldValue(7); + // Close the reader before reading large objects await reader.CloseAsync(); - var weights = await ReadLargeObjectAsync(connection, weightsOid, cancellationToken); - var optimizer = await ReadLargeObjectAsync(connection, optimizerOid, cancellationToken); + // Both reads happen inside the same transaction + var weights = await ReadLargeObjectAsync(connection, weightsOid, transaction, cancellationToken); + var optimizer = await ReadLargeObjectAsync(connection, optimizerOid, transaction, cancellationToken); + + // Commit everything atomically + await transaction.CommitAsync(cancellationToken); return new ModelCheckpoint { @@ -171,7 +180,7 @@ public async Task SaveAsync(ModelCheckpoint checkpoint, CancellationToken cancel /// public async Task DeleteAsync(Guid modelId, CancellationToken cancellationToken = default) { - var connection = await GetConnectionAsync(cancellationToken); + using var connection = await GetConnectionAsync(cancellationToken); await using var transaction = await connection.BeginTransactionAsync(cancellationToken); @@ -217,7 +226,7 @@ public async Task DeleteAsync(Guid modelId, CancellationToken cancellationToken /// public async Task> ListAsync(string? tagKey = null, string? tagValue = null, CancellationToken cancellationToken = default) { - var connection = await GetConnectionAsync(cancellationToken); + using var connection = await GetConnectionAsync(cancellationToken); string sqlQuery; NpgsqlCommand command; @@ -310,12 +319,12 @@ private static async Task UnlinkLargeObjectAsync( } private static async Task ReadLargeObjectAsync( - NpgsqlConnection connection, - uint objectId, - CancellationToken cancellationToken) + NpgsqlConnection connection, + uint objectId, + NpgsqlTransaction transaction, // required – caller manages it + CancellationToken cancellationToken = default) { - await using var transaction = await connection.BeginTransactionAsync(cancellationToken); - + // All commands use the provided transaction – do NOT commit or dispose it here. await using var openCommand = new NpgsqlCommand(PostgresLargeObjectQueries.OpenRead, connection, transaction); openCommand.Parameters.AddWithValue("@oid", objectId).NpgsqlDbType = NpgsqlDbType.Oid; var fileDescriptor = Convert.ToInt32(await openCommand.ExecuteScalarAsync(cancellationToken)); @@ -326,6 +335,7 @@ private static async Task ReadLargeObjectAsync( await using var seekCommand = new NpgsqlCommand(PostgresLargeObjectQueries.SeekStart, connection, transaction); seekCommand.Parameters.AddWithValue("@fd", fileDescriptor).NpgsqlDbType = NpgsqlDbType.Integer; + await seekCommand.ExecuteScalarAsync(cancellationToken); using var memoryStream = new MemoryStream((int)size); @@ -337,6 +347,7 @@ private static async Task ReadLargeObjectAsync( var bytesToRead = (int)Math.Min(chunkSize, size - totalRead); await using var readCommand = new NpgsqlCommand(PostgresLargeObjectQueries.ReadChunk, connection, transaction); + readCommand.Parameters.AddWithValue("@fd", fileDescriptor).NpgsqlDbType = NpgsqlDbType.Integer; readCommand.Parameters.AddWithValue("@length", bytesToRead).NpgsqlDbType = NpgsqlDbType.Integer; @@ -345,13 +356,13 @@ private static async Task ReadLargeObjectAsync( if (result is byte[] dataChunk) { await memoryStream.WriteAsync(dataChunk, 0, dataChunk.Length, cancellationToken); + totalRead += dataChunk.Length; } + else break; } - await transaction.CommitAsync(cancellationToken); - return memoryStream.ToArray(); } } \ No newline at end of file diff --git a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresSessionStore.cs b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresSessionStore.cs index 5e6833c..e961d90 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresSessionStore.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresSessionStore.cs @@ -2,7 +2,7 @@ using Npgsql; using StateCheckpoint.NET.Models; -namespace StateCheckpoint.NET.Stores.Postgres; +namespace StateCheckpoint.NET.Stores; public class PostgresSessionStore : PostgresStoreBase, ISessionStore { @@ -19,7 +19,7 @@ public PostgresSessionStore(NpgsqlDataSource dataSource) : base(dataSource) { } /// public async Task EnsureSchemaAsync(CancellationToken cancellationToken = default) { - var connection = await GetConnectionAsync(cancellationToken); + using var connection = await GetConnectionAsync(cancellationToken); await using var command = new NpgsqlCommand(PostgresSessionQueries.EnsureSessionSchema, connection); @@ -35,7 +35,7 @@ public async Task EnsureSchemaAsync(CancellationToken cancellationToken = defaul /// Session id created for the given session public async Task SaveAsync(SessionCheckpoint session, CancellationToken cancellationToken = default) { - var connection = await GetConnectionAsync(cancellationToken); + using var connection = await GetConnectionAsync(cancellationToken); await using var command = new NpgsqlCommand(PostgresSessionQueries.UpsertInferenceSession, connection); @@ -58,7 +58,7 @@ public async Task SaveAsync(SessionCheckpoint session, CancellationToken cancell /// public async Task LoadAsync(Guid sessionId, CancellationToken cancellationToken = default) { - var connection = await GetConnectionAsync(cancellationToken); + using var connection = await GetConnectionAsync(cancellationToken); await using var command = new NpgsqlCommand(PostgresSessionQueries.SelectInferenceSession, connection); command.Parameters.AddWithValue("@id", sessionId); @@ -86,7 +86,7 @@ public async Task SaveAsync(SessionCheckpoint session, CancellationToken cancell /// public async Task DeleteAsync(Guid sessionId, CancellationToken cancellationToken = default) { - var connection = await GetConnectionAsync(cancellationToken); + using var connection = await GetConnectionAsync(cancellationToken); await using var command = new NpgsqlCommand(PostgresSessionQueries.DeleteInferenceSession, connection); command.Parameters.AddWithValue("@id", sessionId); @@ -103,7 +103,7 @@ public async Task DeleteAsync(Guid sessionId, CancellationToken cancellationToke /// public async Task> ListAsync(string? tagKey = null, string? tagValue = null, CancellationToken cancellationToken = default) { - var connection = await GetConnectionAsync(cancellationToken); + using var connection = await GetConnectionAsync(cancellationToken); string sqlQuery; NpgsqlCommand command; diff --git a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresStoreBase.cs b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresStoreBase.cs index da20056..8003855 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresStoreBase.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresStoreBase.cs @@ -1,6 +1,6 @@ using Npgsql; -namespace StateCheckpoint.NET.Stores.Postgres; +namespace StateCheckpoint.NET.Stores; /// /// Abstract base class for PostgreSQL stores. diff --git a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/MsSql/SqlServerModelStore.cs b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/SqlServer/SqlServerModelStore.cs similarity index 90% rename from StateCheckpoint.NET/StateCheckpoint.NET/Stores/MsSql/SqlServerModelStore.cs rename to StateCheckpoint.NET/StateCheckpoint.NET/Stores/SqlServer/SqlServerModelStore.cs index f6846a2..65a0ebb 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/MsSql/SqlServerModelStore.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/SqlServer/SqlServerModelStore.cs @@ -2,7 +2,7 @@ using Microsoft.Data.SqlClient; using System.Text.Json; -namespace StateCheckpoint.NET.Stores.Mysql; +namespace StateCheckpoint.NET.Stores; public class SqlServerModelStore : SqlServerStoreBase, IModelStore { @@ -18,7 +18,7 @@ public SqlServerModelStore(SqlConnection connection) : base(connection) { } /// public async Task EnsureSchemaAsync(CancellationToken cancellationToken = default) { - var connection = await GetConnectionAsync(cancellationToken); + using var connection = await GetConnectionAsync(cancellationToken); await using var command = new SqlCommand(SqlServerTrainingQueries.EnsureModelSchema, connection); @@ -33,7 +33,7 @@ public async Task EnsureSchemaAsync(CancellationToken cancellationToken = defaul /// id for model checkpoint public async Task SaveAsync(ModelCheckpoint checkpoint, CancellationToken cancellationToken = default) { - var connection = await GetConnectionAsync(cancellationToken); + using var connection = await GetConnectionAsync(cancellationToken); await using var tx = await connection.BeginTransactionAsync(cancellationToken); await using var command = new SqlCommand(SqlServerTrainingQueries.UpsertModelManifest, connection, tx as SqlTransaction); @@ -67,7 +67,7 @@ public async Task SaveAsync(ModelCheckpoint checkpoint, CancellationToken cancel /// public async Task LoadAsync(Guid modelId, CancellationToken cancellationToken = default) { - var connection = await GetConnectionAsync(cancellationToken); + using var connection = await GetConnectionAsync(cancellationToken); await using var command = new SqlCommand(SqlServerTrainingQueries.SelectFullModelManifest, connection); @@ -108,7 +108,7 @@ public async Task SaveAsync(ModelCheckpoint checkpoint, CancellationToken cancel /// public async Task DeleteAsync(Guid modelId, CancellationToken cancellationToken = default) { - var connection = await GetConnectionAsync(cancellationToken); + using var connection = await GetConnectionAsync(cancellationToken); await using var command = new SqlCommand(SqlServerTrainingQueries.DeleteModelManifest, connection); @@ -126,7 +126,7 @@ public async Task DeleteAsync(Guid modelId, CancellationToken cancellationToken /// public async Task> ListAsync(string? tagKey = null, string? tagValue = null, CancellationToken cancellationToken = default) { - var connection = await GetConnectionAsync(cancellationToken); + using var connection = await GetConnectionAsync(cancellationToken); string sql; SqlCommand command; @@ -138,9 +138,11 @@ public async Task> ListAsync(string? tagKey = null, string? tagValue } else { - sql = SqlServerTrainingQueries.ListModelIdsByTag; + sql = SqlServerTrainingQueries.ListModelIdsByTag.Replace("{tagKey}", tagKey); + command = new SqlCommand(sql, connection); - command.Parameters.AddWithValue("@TagPattern", $"%\"{tagKey}\":\"{tagValue}\"%"); + + command.Parameters.AddWithValue("@TagValue", tagValue); } await using (command) diff --git a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/MsSql/SqlServerSessionStore.cs b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/SqlServer/SqlServerSessionStore.cs similarity index 90% rename from StateCheckpoint.NET/StateCheckpoint.NET/Stores/MsSql/SqlServerSessionStore.cs rename to StateCheckpoint.NET/StateCheckpoint.NET/Stores/SqlServer/SqlServerSessionStore.cs index 1da9f64..d7bbb5f 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/MsSql/SqlServerSessionStore.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/SqlServer/SqlServerSessionStore.cs @@ -1,5 +1,4 @@ using StateCheckpoint.NET.Models; -using StateCheckpoint.NET.Stores.Mysql; using Microsoft.Data.SqlClient; using System.Text.Json; @@ -19,7 +18,7 @@ public SqlServerSessionStore(SqlConnection connection) : base(connection) { } /// public async Task EnsureSchemaAsync(CancellationToken cancellationToken = default) { - var connection = await GetConnectionAsync(cancellationToken); + using var connection = await GetConnectionAsync(cancellationToken); await using var command = new SqlCommand(SqlServerSessionQueries.EnsureSessionSchema, connection); @@ -34,7 +33,7 @@ public async Task EnsureSchemaAsync(CancellationToken cancellationToken = defaul /// Session id created for the given session public async Task SaveAsync(SessionCheckpoint session, CancellationToken cancellationToken = default) { - var connection = await GetConnectionAsync(cancellationToken); + using var connection = await GetConnectionAsync(cancellationToken); await using var tx = await connection.BeginTransactionAsync(cancellationToken); @@ -63,7 +62,7 @@ public async Task SaveAsync(SessionCheckpoint session, CancellationToken cancell /// public async Task LoadAsync(Guid sessionId, CancellationToken cancellationToken = default) { - var connection = await GetConnectionAsync(cancellationToken); + using var connection = await GetConnectionAsync(cancellationToken); await using var command = new SqlCommand(SqlServerSessionQueries.SelectInferenceSession, connection); @@ -93,7 +92,7 @@ public async Task SaveAsync(SessionCheckpoint session, CancellationToken cancell /// public async Task DeleteAsync(Guid sessionId, CancellationToken cancellationToken = default) { - var connection = await GetConnectionAsync(cancellationToken); + using var connection = await GetConnectionAsync(cancellationToken); await using var command = new SqlCommand(SqlServerSessionQueries.DeleteInferenceSession, connection); @@ -111,7 +110,7 @@ public async Task DeleteAsync(Guid sessionId, CancellationToken cancellationToke /// public async Task> ListAsync(string? tagKey = null, string? tagValue = null, CancellationToken cancellationToken = default) { - var connection = await GetConnectionAsync(cancellationToken); + using var connection = await GetConnectionAsync(cancellationToken); string sql; SqlCommand command; @@ -123,9 +122,11 @@ public async Task> ListAsync(string? tagKey = null, string? tagValue } else { - sql = SqlServerSessionQueries.ListSessionIdsByTag; + sql = SqlServerSessionQueries.ListSessionIdsByTag.Replace("{tagKey}", tagKey); + command = new SqlCommand(sql, connection); - command.Parameters.AddWithValue("@TagPattern", $"%\"{tagKey}\":\"{tagValue}\"%"); + + command.Parameters.AddWithValue("@TagValue", tagValue); } await using (command) diff --git a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/MsSql/SqlServerStoreBase.cs b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/SqlServer/SqlServerStoreBase.cs similarity index 70% rename from StateCheckpoint.NET/StateCheckpoint.NET/Stores/MsSql/SqlServerStoreBase.cs rename to StateCheckpoint.NET/StateCheckpoint.NET/Stores/SqlServer/SqlServerStoreBase.cs index 068246d..1700146 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/MsSql/SqlServerStoreBase.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/SqlServer/SqlServerStoreBase.cs @@ -1,6 +1,7 @@ using Microsoft.Data.SqlClient; +using System.Data; -namespace StateCheckpoint.NET.Stores.Mysql; +namespace StateCheckpoint.NET.Stores; /// /// Abstract base class for SQL Server stores. @@ -11,6 +12,7 @@ public abstract class SqlServerStoreBase : IAsyncDisposable private readonly string _connectionString = string.Empty; private SqlConnection? _connection; private readonly bool _ownsConnection; + private readonly SemaphoreSlim _connectionLock = new(1, 1); /// /// Initializes the store with a connection string. @@ -37,20 +39,30 @@ protected SqlServerStoreBase(SqlConnection connection) /// protected async Task GetConnectionAsync(CancellationToken ct = default) { - if (_connection == null || _connection.State != System.Data.ConnectionState.Open) + if (!_ownsConnection) { - if (_connection == null && _ownsConnection) + // External connection – ensure thread‑safe access + await _connectionLock.WaitAsync(ct); + + try { - _connection = new SqlConnection(_connectionString); - } + if (_connection!.State != ConnectionState.Open) + await _connection.OpenAsync(ct); - if (_connection != null && _connection.State != System.Data.ConnectionState.Open) + return _connection; + } + finally { - await _connection.OpenAsync(ct); + _connectionLock.Release(); } } - return _connection!; + // Owned connection – create a new one per call + var connection = new SqlConnection(_connectionString!); + + await connection.OpenAsync(ct); + + return connection; } /// From 23e5f98b74f119daf5cdab8ce7606fa72c5c31db Mon Sep 17 00:00:00 2001 From: Leo Mengesha Date: Sun, 12 Jul 2026 23:00:38 +0100 Subject: [PATCH 2/3] added await on connection fetch --- .github/workflows/dotnet.yml | 20 ++++++------------- .../StateCheckpoint.NET.csproj | 2 +- .../Stores/Postgres/PostgresModelStore.cs | 8 ++++---- .../Stores/Postgres/PostgresSessionStore.cs | 10 +++++----- 4 files changed, 16 insertions(+), 24 deletions(-) diff --git a/.github/workflows/dotnet.yml b/.github/workflows/dotnet.yml index 861045d..b363d44 100644 --- a/.github/workflows/dotnet.yml +++ b/.github/workflows/dotnet.yml @@ -54,10 +54,6 @@ jobs: if: startsWith(github.ref, 'refs/tags/v') - permissions: - id-token: write - contents: read - steps: - uses: actions/checkout@v4 @@ -81,24 +77,20 @@ jobs: - name: List nupkg files run: ls -la ./nupkgs - - name: Get OIDC token - id: oidc - uses: actions/github-script@v7 - with: - script: | - const token = await core.getIDToken('api://NuGet'); - core.setOutput('token', token); - - name: Push to NuGet.org env: - NUGET_AUTH_TOKEN: ${{ steps.oidc.outputs.token }} + NUGET_API_KEY: ${{ secrets.NUGET_API_KEY }} run: | + if [ -z "$NUGET_API_KEY" ]; then + echo "Error: NUGET_API_KEY secret is not set." + exit 1 + fi for pkg in ./nupkgs/*.nupkg; do if [ -f "$pkg" ]; then echo "Pushing $pkg" dotnet nuget push "$pkg" \ --source "https://api.nuget.org/v3/index.json" \ - --api-key "$NUGET_AUTH_TOKEN" \ + --api-key "$NUGET_API_KEY" \ --skip-duplicate else echo "No .nupkg files found" diff --git a/StateCheckpoint.NET/StateCheckpoint.NET/StateCheckpoint.NET.csproj b/StateCheckpoint.NET/StateCheckpoint.NET/StateCheckpoint.NET.csproj index 499e853..bea2865 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET/StateCheckpoint.NET.csproj +++ b/StateCheckpoint.NET/StateCheckpoint.NET/StateCheckpoint.NET.csproj @@ -9,7 +9,7 @@ checkpointing;llm;machine-learning;sql;local-saves https://github.com/bargross/StateCheckpoint.NET StateCheckpoint.NET - 1.0.0 + 1.0.1 Universal checkpointing for C# ML. Persist training states (Weights + Optimizer) and inference sessions (KV-Cache + Token History) to FileSystem, PostgreSQL, or SQL Server with non-blocking background saves. README.md Apache-2.0 diff --git a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresModelStore.cs b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresModelStore.cs index 14caa88..846a86c 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresModelStore.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresModelStore.cs @@ -20,7 +20,7 @@ public PostgresModelStore(NpgsqlDataSource dataSource) : base(dataSource) { } /// public async Task EnsureSchemaAsync(CancellationToken CancellationToken = default) { - using var connection = await GetConnectionAsync(CancellationToken); + await using var connection = await GetConnectionAsync(CancellationToken); await using var command = new NpgsqlCommand(PostgresTrainingQueries.EnsureModelSchema, connection); @@ -35,7 +35,7 @@ public async Task EnsureSchemaAsync(CancellationToken CancellationToken = defaul /// id for model checkpoint public async Task SaveAsync(ModelCheckpoint checkpoint, CancellationToken cancellationToken = default) { - using var connection = await GetConnectionAsync(cancellationToken); + await using var connection = await GetConnectionAsync(cancellationToken); await using var transaction = await connection.BeginTransactionAsync(cancellationToken); @@ -180,7 +180,7 @@ public async Task SaveAsync(ModelCheckpoint checkpoint, CancellationToken cancel /// public async Task DeleteAsync(Guid modelId, CancellationToken cancellationToken = default) { - using var connection = await GetConnectionAsync(cancellationToken); + await using var connection = await GetConnectionAsync(cancellationToken); await using var transaction = await connection.BeginTransactionAsync(cancellationToken); @@ -226,7 +226,7 @@ public async Task DeleteAsync(Guid modelId, CancellationToken cancellationToken /// public async Task> ListAsync(string? tagKey = null, string? tagValue = null, CancellationToken cancellationToken = default) { - using var connection = await GetConnectionAsync(cancellationToken); + await using var connection = await GetConnectionAsync(cancellationToken); string sqlQuery; NpgsqlCommand command; diff --git a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresSessionStore.cs b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresSessionStore.cs index e961d90..d12455b 100644 --- a/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresSessionStore.cs +++ b/StateCheckpoint.NET/StateCheckpoint.NET/Stores/Postgres/PostgresSessionStore.cs @@ -19,7 +19,7 @@ public PostgresSessionStore(NpgsqlDataSource dataSource) : base(dataSource) { } /// public async Task EnsureSchemaAsync(CancellationToken cancellationToken = default) { - using var connection = await GetConnectionAsync(cancellationToken); + await using var connection = await GetConnectionAsync(cancellationToken); await using var command = new NpgsqlCommand(PostgresSessionQueries.EnsureSessionSchema, connection); @@ -35,7 +35,7 @@ public async Task EnsureSchemaAsync(CancellationToken cancellationToken = defaul /// Session id created for the given session public async Task SaveAsync(SessionCheckpoint session, CancellationToken cancellationToken = default) { - using var connection = await GetConnectionAsync(cancellationToken); + await using var connection = await GetConnectionAsync(cancellationToken); await using var command = new NpgsqlCommand(PostgresSessionQueries.UpsertInferenceSession, connection); @@ -58,7 +58,7 @@ public async Task SaveAsync(SessionCheckpoint session, CancellationToken cancell /// public async Task LoadAsync(Guid sessionId, CancellationToken cancellationToken = default) { - using var connection = await GetConnectionAsync(cancellationToken); + await using var connection = await GetConnectionAsync(cancellationToken); await using var command = new NpgsqlCommand(PostgresSessionQueries.SelectInferenceSession, connection); command.Parameters.AddWithValue("@id", sessionId); @@ -86,7 +86,7 @@ public async Task SaveAsync(SessionCheckpoint session, CancellationToken cancell /// public async Task DeleteAsync(Guid sessionId, CancellationToken cancellationToken = default) { - using var connection = await GetConnectionAsync(cancellationToken); + await using var connection = await GetConnectionAsync(cancellationToken); await using var command = new NpgsqlCommand(PostgresSessionQueries.DeleteInferenceSession, connection); command.Parameters.AddWithValue("@id", sessionId); @@ -103,7 +103,7 @@ public async Task DeleteAsync(Guid sessionId, CancellationToken cancellationToke /// public async Task> ListAsync(string? tagKey = null, string? tagValue = null, CancellationToken cancellationToken = default) { - using var connection = await GetConnectionAsync(cancellationToken); + await using var connection = await GetConnectionAsync(cancellationToken); string sqlQuery; NpgsqlCommand command; From a79d1774b5b3d88713c3cf8ec84cf37a3374a955 Mon Sep 17 00:00:00 2001 From: Leo Mengesha Date: Sun, 12 Jul 2026 23:03:45 +0100 Subject: [PATCH 3/3] added fixes/changes to changelog --- CHANGELOG.md | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index e82bed1..f1ed64a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,32 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +--- + +## [1.0.1] — 2026-07-12 + +Patch release fixing all critical correctness bugs identified in the initial release. No public API surface changes. + +### Fixed + +- **`SqlServerStoreBase` thread-safety** — the store previously held a single shared `SqlConnection` instance, which is not thread-safe. Concurrent calls to any SQL Server store method would corrupt each other's state. The owned-connection path now creates a fresh `SqlConnection` per call (matching the connection pool pattern used by the Postgres store); externally supplied connections are guarded with a `SemaphoreSlim(1,1)`. + +- **`PostgresModelStore.LoadAsync` transaction scope** — large object reads were happening outside the transaction that produced the OIDs. After `reader.CloseAsync()`, the metadata transaction was closed, leaving the OIDs as dangling references that could be unlinked by a concurrent writer before the read completed. The metadata read and both `ReadLargeObjectAsync` calls are now wrapped in a single transaction, passed explicitly through the call chain. + +- **`FileSystemModelStore.ListAsync` and `FileSystemSessionStore.ListAsync` silently ignored tag filters** — calling `ListAsync(tagKey: "env", tagValue: "prod")` returned all checkpoints regardless of tags, with no error or warning. Both stores now load each `manifest.json` / `meta.json` in memory and filter by the requested tag key/value pair before returning results. + +- **`BackgroundSaver.Enqueue` blocked the caller synchronously** — the previous implementation called `.GetAwaiter().GetResult()` on the channel `WriteAsync`, which blocked the calling thread whenever the bounded queue was full. This defeated the purpose of background saves and introduced deadlock risk on the thread pool. `Enqueue` has been replaced with `EnqueueAsync` returning `ValueTask`; both `CheckpointManager` and `SessionManager` now `await` it correctly. + +- **Postgres connections were never returned to the pool** — `GetConnectionAsync` opened a connection from the `NpgsqlDataSource` pool but no call site disposed it, preventing connections from being returned. All call sites across `PostgresModelStore` and `PostgresSessionStore` now use `using var connection = await GetConnectionAsync(...)` to ensure prompt return to the pool. + +- **SQL Server tag filtering produced incorrect results** — tag queries used `LIKE '%"key":"value"%'` string matching against the JSON column, which failed when the serializer emitted spaces and produced false positives when a value contained the search string as a substring. Replaced with `JSON_VALUE(Tags, '$.{key}') = @TagValue` for correct, indexed JSON field extraction (requires SQL Server 2016+). + +- **`TokenizerData.MergeRules` did not serialise correctly** — the field was typed as `List<(string Left, string Right)>?` (C# value tuples), which `System.Text.Json` does not serialise by default, producing `{"Item1":"...","Item2":"..."}` or failing silently depending on runtime version. Changed to `List?` using the existing `MergeRule` class, which serialises correctly in all versions. + +- **SQL Server stores were in the wrong namespace** — `SqlServerStoreBase`, `SqlServerModelStore`, and `SqlServerSessionStore` were in `StateCheckpoint.NET.Stores.Mysql`. Corrected to `StateCheckpoint.NET.Stores` with files moved to `Stores/SqlServer/`. + +- PostgresModelStore and PostgresSessionStore connections not disposed asynchronously — four call sites in PostgresModelStore and five in PostgresSessionStore used using var connection (synchronous IDisposable) instead of await using var connection (IAsyncDisposable). NpgsqlConnection implements IAsyncDisposable and disposing it synchronously blocks the thread pool while the connection is returned. All call sites now use await using. + ## [1.0.0] - 2025-06-25 ### Added