diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/DependencyInjection/ReplicationServiceCollectionExtensions.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/DependencyInjection/ReplicationServiceCollectionExtensions.cs index f9ccba9..be7995c 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/DependencyInjection/ReplicationServiceCollectionExtensions.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/DependencyInjection/ReplicationServiceCollectionExtensions.cs @@ -16,7 +16,9 @@ public static class ReplicationServiceCollectionExtensions /// configuration section, registers the passive as a singleton /// (its single-stream guard must span calls), and registers the initiating /// as a hosted service — on a passive node (no - /// ) it starts and immediately idles. Idempotent. + /// ) it starts and immediately idles. The periodic + /// (oplog caps enforcement + tombstone pruning) is + /// also always registered and runs on BOTH roles — passive nodes need it too. Idempotent. /// Requires to have been called /// for the the adapters depend on. /// @@ -59,6 +61,11 @@ public static class ReplicationServiceCollectionExtensions services.AddSingleton(); services.AddHostedService(); + // Factory registration because the ctor takes the internal shared OplogStore singleton. + services.AddHostedService(sp => new MaintenanceBackgroundService( + sp.GetRequiredService(), + sp.GetRequiredService>(), + sp.GetRequiredService())); return services; } diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/MaintenanceBackgroundService.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/MaintenanceBackgroundService.cs new file mode 100644 index 0000000..f098cfb --- /dev/null +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/MaintenanceBackgroundService.cs @@ -0,0 +1,77 @@ +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.LocalDb.Replication.Internal; + +namespace ZB.MOM.WW.LocalDb.Replication; + +/// +/// Periodic oplog maintenance: every it +/// enforces the backlog/age caps (which flag needs_snapshot when exceeded) and prunes +/// expired tombstone row-versions. Runs on BOTH roles — a passive node with a long-dead peer is +/// exactly the case where the oplog would otherwise grow unbounded. A failing tick is logged and +/// skipped; maintenance never dies. +/// +public sealed class MaintenanceBackgroundService : BackgroundService +{ + private readonly OplogStore _store; + private readonly ReplicationOptions _options; + private readonly ILogger _logger; + + // Rate-limits the caps-exceeded warning to one per flag episode: EnforceCapsAsync keeps + // returning true every tick while the backlog stays over cap, which would otherwise flood + // the log at the maintenance rate. Reset when a tick comes back clean. + private bool _snapshotFlagWarned; + + internal MaintenanceBackgroundService( + OplogStore store, IOptions options, ILoggerFactory loggerFactory) + { + _store = store; + _options = options.Value; + _logger = loggerFactory.CreateLogger(); + } + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + // First tick lands one interval after startup (not immediately) to keep startup cheap. + while (!stoppingToken.IsCancellationRequested) + { + try + { + await Task.Delay(_options.MaintenanceInterval, stoppingToken); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + + try + { + var snapshotFlagged = await _store.EnforceCapsAsync(stoppingToken); + await _store.PruneTombstonesAsync(stoppingToken); + + if (snapshotFlagged && !_snapshotFlagWarned) + { + _snapshotFlagWarned = true; + _logger.LogWarning( + "Oplog backlog/age caps exceeded: pruned to the ceiling and flagged needs_snapshot — the peer must snapshot-resync. Not re-logged until a clean tick."); + } + else if (!snapshotFlagged) + { + _snapshotFlagWarned = false; + } + + _logger.LogDebug("Maintenance tick complete (snapshotFlagged={SnapshotFlagged}).", snapshotFlagged); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + catch (Exception ex) + { + // Maintenance must never die: log and try again next interval. + _logger.LogWarning(ex, "Maintenance tick failed; retrying next interval."); + } + } + } +} diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptions.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptions.cs index b1a6a8e..ed1912f 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptions.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptions.cs @@ -29,6 +29,9 @@ public sealed class ReplicationOptions /// Upper bound on the exponential reconnect backoff. public TimeSpan ReconnectBackoffMax { get; set; } = TimeSpan.FromSeconds(60); + /// How often the maintenance service enforces the oplog caps and prunes expired tombstones. + public TimeSpan MaintenanceInterval { get; set; } = TimeSpan.FromSeconds(60); + /// Maximum tolerated HLC drift where a peer's clock runs ahead of ours. public TimeSpan MaxHlcDriftAhead { get; set; } = TimeSpan.FromMinutes(5); diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptionsValidator.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptionsValidator.cs index e657f88..0851dfb 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptionsValidator.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptionsValidator.cs @@ -36,6 +36,8 @@ public sealed class ReplicationOptionsValidator : IValidateOptions ReconnectBackoffCeiling) failures.Add($"LocalDb:Replication:ReconnectBackoffMax must be at most {ReconnectBackoffCeiling}; got {options.ReconnectBackoffMax}."); + if (options.MaintenanceInterval <= TimeSpan.Zero) + failures.Add($"LocalDb:Replication:MaintenanceInterval must be greater than zero; got {options.MaintenanceInterval}."); if (options.MaxHlcDriftAhead <= TimeSpan.Zero) failures.Add($"LocalDb:Replication:MaxHlcDriftAhead must be greater than zero; got {options.MaxHlcDriftAhead}."); diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/MaintenanceTests.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/MaintenanceTests.cs new file mode 100644 index 0000000..7cdcd5d --- /dev/null +++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/MaintenanceTests.cs @@ -0,0 +1,212 @@ +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.LocalDb.Internal; +using ZB.MOM.WW.LocalDb.Replication; +using ZB.MOM.WW.LocalDb.Replication.Internal; + +namespace ZB.MOM.WW.LocalDb.Tests; + +public sealed class MaintenanceTests : IDisposable +{ + private static readonly TimeSpan WaitTimeout = TimeSpan.FromSeconds(10); + private static readonly TimeSpan Tick = TimeSpan.FromMilliseconds(50); + + private readonly List _paths = []; + private readonly List _disposables = []; + + public void Dispose() + { + foreach (var d in _disposables) + d.Dispose(); + Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools(); + foreach (var p in _paths) + { + if (File.Exists(p)) File.Delete(p); + if (File.Exists(p + "-wal")) File.Delete(p + "-wal"); + if (File.Exists(p + "-shm")) File.Delete(p + "-shm"); + } + } + + private async Task<(SqliteLocalDb Db, OplogStore Store)> NewDbAsync(ReplicationOptions options) + { + var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db"); + _paths.Add(path); + var db = new SqliteLocalDb(new LocalDbOptions { Path = path }); + _disposables.Add(db); + await db.ExecuteAsync("CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT, qty INTEGER)"); + db.RegisterReplicated("orders"); + return (db, new OplogStore(db, options)); + } + + private static MaintenanceBackgroundService NewService( + OplogStore store, ReplicationOptions options, ILoggerFactory? loggerFactory = null) => + new(store, Options.Create(options), loggerFactory ?? NullLoggerFactory.Instance); + + private static async Task WaitForAsync(Func> predicate, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + if (await predicate()) return; + await Task.Delay(20); + } + throw new TimeoutException("Condition not reached within " + timeout); + } + + [Fact] + public async Task Maintenance_EnforcesCaps_SetsNeedsSnapshot() + { + var options = new ReplicationOptions { MaxOplogRows = 5, MaintenanceInterval = Tick }; + var (db, store) = await NewDbAsync(options); + for (var i = 1; i <= 20; i++) + await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'R', @id)", new { id = i }); + + var service = NewService(store, options); + await service.StartAsync(CancellationToken.None); + try + { + await WaitForAsync(async () => + { + var snap = await db.QueryAsync("SELECT needs_snapshot FROM __localdb_peer_state WHERE id = 1", x => x.GetInt64(0)); + var depth = await db.QueryAsync("SELECT COUNT(*) FROM __localdb_oplog", x => x.GetInt64(0)); + return snap[0] == 1 && depth[0] <= 5; + }, WaitTimeout); + } + finally + { + await service.StopAsync(CancellationToken.None); + } + } + + [Fact] + public async Task Maintenance_PrunesExpiredTombstones() + { + var options = new ReplicationOptions { MaintenanceInterval = Tick }; + var (db, store) = await NewDbAsync(options); + await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'DOOMED', 1)"); + await db.ExecuteAsync("DELETE FROM orders WHERE id = 1"); + // Backdate the delete-trigger's tombstone far past any retention window. + var updated = await db.ExecuteAsync( + "UPDATE __localdb_row_version SET tombstone_utc = '2000-01-01T00:00:00.000Z' WHERE is_tombstone = 1"); + Assert.Equal(1, updated); + + var service = NewService(store, options); + await service.StartAsync(CancellationToken.None); + try + { + await WaitForAsync(async () => + { + var rows = await db.QueryAsync( + "SELECT COUNT(*) FROM __localdb_row_version WHERE is_tombstone = 1", x => x.GetInt64(0)); + return rows[0] == 0; + }, WaitTimeout); + } + finally + { + await service.StopAsync(CancellationToken.None); + } + } + + [Fact] + public async Task Maintenance_SurvivesTickFailure() + { + var options = new ReplicationOptions { MaintenanceInterval = Tick }; + var (db, store) = await NewDbAsync(options); + // Deterministic per-tick failure: every EnforceCapsAsync now hits "no such table". + await db.ExecuteAsync("DROP TABLE __localdb_oplog"); + + var logger = new ListLoggerFactory(); + var service = NewService(store, options, logger); + await service.StartAsync(CancellationToken.None); + try + { + // At least ~5 failing ticks must have elapsed without killing the loop. + await WaitForAsync( + () => Task.FromResult(logger.Count(LogLevel.Warning) >= 5), WaitTimeout); + Assert.NotNull(service.ExecuteTask); + Assert.False(service.ExecuteTask!.IsCompleted, "maintenance loop died on a tick failure"); + } + finally + { + await service.StopAsync(CancellationToken.None); + } + Assert.True(service.ExecuteTask!.IsCompletedSuccessfully); + } + + [Fact] + public void Validator_NonPositiveMaintenanceInterval_Fails() + { + var result = new ReplicationOptionsValidator().Validate( + null, new ReplicationOptions { MaintenanceInterval = TimeSpan.Zero }); + + Assert.True(result.Failed); + Assert.Contains(result.Failures, f => f.Contains("MaintenanceInterval", StringComparison.Ordinal)); + } + + [Fact] + public void AddZbLocalDbReplication_RegistersMaintenance_BothRoles() + { + foreach (var peerAddress in new string?[] { null, "http://localhost" }) + { + var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db"); + _paths.Add(path); + var dict = new Dictionary { ["LocalDb:Path"] = path }; + if (peerAddress is not null) dict["LocalDb:Replication:PeerAddress"] = peerAddress; + var config = new ConfigurationBuilder().AddInMemoryCollection(dict).Build(); + + var services = new ServiceCollection(); + services.AddLogging(); + services.AddZbLocalDb(config, db => + { + using var conn = db.CreateConnection(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = "CREATE TABLE orders (id INTEGER PRIMARY KEY)"; + cmd.ExecuteNonQuery(); + db.RegisterReplicated("orders"); + }); + services.AddZbLocalDbReplication(config); + var provider = services.BuildServiceProvider(); + _disposables.Add(provider); + + Assert.Contains(provider.GetServices(), s => s is MaintenanceBackgroundService); + } + } + + private sealed class ListLoggerFactory : ILoggerFactory + { + private readonly List _levels = []; + + public int Count(LogLevel level) + { + lock (_levels) + return _levels.Count(l => l == level); + } + + public ILogger CreateLogger(string categoryName) => new ListLogger(_levels); + public void AddProvider(ILoggerProvider provider) { } + public void Dispose() { } + + private sealed class ListLogger(List levels) : ILogger + { + public IDisposable BeginScope(TState state) where TState : notnull => NullScope.Instance; + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) + { + lock (levels) + levels.Add(logLevel); + } + + private sealed class NullScope : IDisposable + { + public static readonly NullScope Instance = new(); + public void Dispose() { } + } + } + } +}