diff --git a/AGENTS.md b/AGENTS.md index cd03c11..9192ef6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -396,6 +396,8 @@ No abbreviations in names. - `IDataProvider` + its DTOs stay in `SemiPlot.Core`; every concrete provider lives in its own `SemiPlot.DataSource.*` project (`SemiPlot.DataSource.Postgres` is the only one). Core must not reference a data-source project; a further provider slots in as a sibling without touching Core. + `ILiveSource` lives in `SemiPlot.Core/Data` beside `IDataProvider`, and `AddPostgresData()` registers + `PostgresLiveSource` over the same `NpgsqlDataSource` (`docs/architecture/data-integration.md#realtime`). - Each write path is an interface of its own and `IDataProvider` stays read-only. `IPenCatalogueEditor`, `PenCatalogue` and `PenSettingChange` live in `SemiPlot.Core/Data`; `PostgresPenCatalogueEditor` lives in `SemiPlot.DataSource.Postgres`, which `AddPostgresData()` registers beside the provider over the @@ -438,12 +440,12 @@ No abbreviations in names. the configuration tables, the two roles and their grants are SemiBase's; the seeder fills the archive table and the catalogue, and creates only the day partitions its rows land in (`docs/architecture/bench.md`). -- The provider runs no cold-path reader: a failed read is mapped by `ArchiveExceptionMapper`, which - stays synchronous, pure and unit-testable, and nothing opens a second connection to enrich the - error. Each read supplies every relation its statement touches, which the detail line names: one - relation for the history and realtime reads, `semiplot_tags, trends` for the archive extent and the - archive days, all three of `semiplot_tags`, `semiplot_groups` and `semiplot_pen_groups` for the pen - catalogue, and `semiplot_markers` for every marker read and write. +- The provider and the live source run no cold-path reader: a failed read is mapped by + `ArchiveExceptionMapper`, which stays synchronous, pure and unit-testable, and nothing opens a second + connection to enrich the error. Each read supplies every relation its statement touches, which the + detail line names: one relation for the history and realtime reads, `semiplot_tags, trends` for the + archive extent and the archive days, all three of `semiplot_tags`, `semiplot_groups` and + `semiplot_pen_groups` for the pen catalogue, and `semiplot_markers` for every marker read and write. --- diff --git a/SemiPlot/SemiPlot.Core/Data/IDataProvider.cs b/SemiPlot/SemiPlot.Core/Data/IDataProvider.cs index 94106ce..49aa4cf 100644 --- a/SemiPlot/SemiPlot.Core/Data/IDataProvider.cs +++ b/SemiPlot/SemiPlot.Core/Data/IDataProvider.cs @@ -6,9 +6,6 @@ namespace SemiPlot.Core.Data; public interface IDataProvider { - // Cold per call: no samples flow until subscribed; the subscriber disposes the returned IDisposable. - IObservable> Subscribe(IReadOnlyList penIds); - Task>> QueryPensAsync(); Task>> QueryHistoryAsync( diff --git a/SemiPlot/SemiPlot.Core/Data/ILiveSource.cs b/SemiPlot/SemiPlot.Core/Data/ILiveSource.cs new file mode 100644 index 0000000..f901149 --- /dev/null +++ b/SemiPlot/SemiPlot.Core/Data/ILiveSource.cs @@ -0,0 +1,9 @@ +using SemiPlot.Core.Trends; + +namespace SemiPlot.Core.Data; + +/// The live edge of a pen set (docs/architecture/data-integration.md#realtime). +public interface ILiveSource +{ + IObservable> Subscribe(IReadOnlyList penIds); +} diff --git a/SemiPlot/SemiPlot.Core/Trends/RealtimeBatch.cs b/SemiPlot/SemiPlot.Core/Trends/RealtimeBatch.cs index ff4f3fb..cc89172 100644 --- a/SemiPlot/SemiPlot.Core/Trends/RealtimeBatch.cs +++ b/SemiPlot/SemiPlot.Core/Trends/RealtimeBatch.cs @@ -2,7 +2,7 @@ namespace SemiPlot.Core.Trends; /// /// One pen's share of a , on that pen's own timestamps only. A null is a break the -/// provider reported; a pen carries no entry, null or real, at a union timestamp it did not report. +/// live source reported; a pen carries no entry, null or real, at a union timestamp it did not report. /// public sealed record PenRealtimeValues( int PenId, diff --git a/SemiPlot/SemiPlot.DataSource.Postgres/PostgresDataProvider.cs b/SemiPlot/SemiPlot.DataSource.Postgres/PostgresDataProvider.cs index 9588c47..0b3cb7f 100644 --- a/SemiPlot/SemiPlot.DataSource.Postgres/PostgresDataProvider.cs +++ b/SemiPlot/SemiPlot.DataSource.Postgres/PostgresDataProvider.cs @@ -1,8 +1,6 @@ using System.Collections.Concurrent; using System.Diagnostics; using System.Globalization; -using System.Reactive.Concurrency; -using System.Reactive.Linq; using FluentResults; @@ -14,14 +12,13 @@ using SemiPlot.Core.Data; using SemiPlot.Core.Trends; -using SemiPlot.DataSource.Postgres.Configuration; namespace SemiPlot.DataSource.Postgres; /// /// Reads the archive over the pooled : answers the -/// configured variables, the span they cover, -/// a window of one layer and the live edge, all crossing the boundary in UTC. +/// configured variables, the span they cover and +/// a window of one layer, all crossing the boundary in UTC. /// public sealed class PostgresDataProvider : IDataProvider { @@ -42,8 +39,6 @@ public sealed class PostgresDataProvider : IDataProvider private readonly NpgsqlDataSource _dataSource; private readonly ArchiveTimeConverter _timeConverter; private readonly ArchiveExceptionMapper _exceptionMapper; - private readonly PostgresConnectionSettings _settings; - private readonly IScheduler _scheduler; private readonly TimeProvider _timeProvider; private readonly ILogger _logger; private readonly int _historyReadConcurrency; @@ -57,8 +52,6 @@ internal PostgresDataProvider( NpgsqlDataSource dataSource, ArchiveTimeConverter timeConverter, ArchiveExceptionMapper exceptionMapper, - PostgresConnectionSettings settings, - IScheduler scheduler, TimeProvider timeProvider, ILogger logger, int historyReadConcurrency) @@ -66,26 +59,11 @@ internal PostgresDataProvider( _dataSource = dataSource; _timeConverter = timeConverter; _exceptionMapper = exceptionMapper; - _settings = settings; - _scheduler = scheduler; _timeProvider = timeProvider; _logger = logger; _historyReadConcurrency = historyReadConcurrency; } - /// - /// The live edge of the requested variables. Cold: each subscription starts a poll loop of its own, on the - /// injected scheduler and at the operator's , holding - /// its own baseline; disposing it cancels the loop's query and its wait, so no further statement is issued. - /// - public IObservable> Subscribe(IReadOnlyList penIds) - { - var subscribedIds = penIds.ToArray(); - - return Observable.Create>(observer => - _scheduler.ScheduleAsync((_, cancellationToken) => PollAsync(subscribedIds, observer, cancellationToken))); - } - /// /// Every configured variable, ordered by name and then by id. /// @@ -337,45 +315,6 @@ private void LogHistoryRead( stopwatch.ElapsedMilliseconds); } - private async Task PollAsync( - int[] penIds, - IObserver> observer, - CancellationToken cancellationToken) - { - var poll = new RealtimePoll( - _dataSource, - _timeConverter, - _exceptionMapper, - penIds, - _logger); - - try - { - while (!cancellationToken.IsCancellationRequested) - { - var samples = await poll.ReadOnceAsync(cancellationToken).ConfigureAwait(false); - - // A disposal landing while the query is in flight delivers nothing further: the subscriber - // is gone, and a batch arriving behind its back is a leaked loop by another name. - if (cancellationToken.IsCancellationRequested) - { - return; - } - - if (samples.Count > 0) - { - observer.OnNext(samples); - } - - await _scheduler.Sleep(_settings.PollInterval, cancellationToken).ConfigureAwait(false); - } - } - catch (OperationCanceledException) - { - // Disposal ends the loop. - } - } - internal static NpgsqlCommand CreateCatalogueCommand(NpgsqlConnection connection) { return new NpgsqlCommand(ArchiveStatements.PenCatalog, connection) diff --git a/SemiPlot/SemiPlot.DataSource.Postgres/PostgresDataServiceCollectionExtensions.cs b/SemiPlot/SemiPlot.DataSource.Postgres/PostgresDataServiceCollectionExtensions.cs index c44a5f6..43d592f 100644 --- a/SemiPlot/SemiPlot.DataSource.Postgres/PostgresDataServiceCollectionExtensions.cs +++ b/SemiPlot/SemiPlot.DataSource.Postgres/PostgresDataServiceCollectionExtensions.cs @@ -38,11 +38,16 @@ internal static IServiceCollection AddPostgresData( provider.GetRequiredService(), provider.GetRequiredService(), provider.GetRequiredService(), - provider.GetRequiredService(), - provider.GetRequiredService(), provider.GetRequiredService(), provider.GetRequiredService>(), historyReadConcurrency)); + services.AddSingleton(provider => new PostgresLiveSource( + provider.GetRequiredService(), + provider.GetRequiredService(), + provider.GetRequiredService(), + provider.GetRequiredService(), + provider.GetRequiredService>(), + provider.GetRequiredService().PollInterval)); services.AddSingleton(provider => new PostgresPenCatalogueEditor( provider.GetRequiredService(), provider.GetRequiredService(), diff --git a/SemiPlot/SemiPlot.DataSource.Postgres/PostgresLiveSource.cs b/SemiPlot/SemiPlot.DataSource.Postgres/PostgresLiveSource.cs new file mode 100644 index 0000000..671ca62 --- /dev/null +++ b/SemiPlot/SemiPlot.DataSource.Postgres/PostgresLiveSource.cs @@ -0,0 +1,89 @@ +using System.Reactive.Concurrency; +using System.Reactive.Linq; + +using Microsoft.Extensions.Logging; + +using Npgsql; + +using SemiPlot.Core.Data; +using SemiPlot.Core.Trends; + +namespace SemiPlot.DataSource.Postgres; + +/// Serves the live edge by polling the archive over the pooled . +public sealed class PostgresLiveSource : ILiveSource +{ + private readonly NpgsqlDataSource _dataSource; + private readonly ArchiveTimeConverter _timeConverter; + private readonly ArchiveExceptionMapper _exceptionMapper; + private readonly IScheduler _scheduler; + private readonly ILogger _logger; + private readonly TimeSpan _pollInterval; + + internal PostgresLiveSource( + NpgsqlDataSource dataSource, + ArchiveTimeConverter timeConverter, + ArchiveExceptionMapper exceptionMapper, + IScheduler scheduler, + ILogger logger, + TimeSpan pollInterval) + { + _dataSource = dataSource; + _timeConverter = timeConverter; + _exceptionMapper = exceptionMapper; + _scheduler = scheduler; + _logger = logger; + _pollInterval = pollInterval; + } + + /// + /// The live edge of the requested variables. Cold: each subscription starts a poll loop of its own, on the + /// injected scheduler and at the poll interval, holding its own baseline; disposing it cancels the loop's + /// query and its wait, so no further statement is issued. + /// + public IObservable> Subscribe(IReadOnlyList penIds) + { + var subscribedIds = penIds.ToArray(); + + return Observable.Create>(observer => + _scheduler.ScheduleAsync((_, cancellationToken) => PollAsync(subscribedIds, observer, cancellationToken))); + } + + private async Task PollAsync( + int[] penIds, + IObserver> observer, + CancellationToken cancellationToken) + { + var poll = new RealtimePoll( + _dataSource, + _timeConverter, + _exceptionMapper, + penIds, + _logger); + + try + { + while (!cancellationToken.IsCancellationRequested) + { + var samples = await poll.ReadOnceAsync(cancellationToken).ConfigureAwait(false); + + // A disposal landing while the query is in flight delivers nothing further. + if (cancellationToken.IsCancellationRequested) + { + return; + } + + if (samples.Count > 0) + { + observer.OnNext(samples); + } + + await _scheduler.Sleep(_pollInterval, cancellationToken).ConfigureAwait(false); + } + } + catch (OperationCanceledException) + { + // Disposal ends the loop. + } + } +} diff --git a/SemiPlot/SemiPlot.Tests.Integration/Journeys/BreakRenderArchiveJourneyTests.cs b/SemiPlot/SemiPlot.Tests.Integration/Journeys/BreakRenderArchiveJourneyTests.cs index e6d0ea4..85d2883 100644 --- a/SemiPlot/SemiPlot.Tests.Integration/Journeys/BreakRenderArchiveJourneyTests.cs +++ b/SemiPlot/SemiPlot.Tests.Integration/Journeys/BreakRenderArchiveJourneyTests.cs @@ -57,6 +57,7 @@ public async Task TheFirstSeededBreakLeavesTheRenderedCurvesBroken() { await using var services = ArchiveProviderFactory.Build(seededArchive.Database.PlotConnectionString); var dataProvider = services.GetRequiredService(); + var liveSource = services.GetRequiredService(); var dataScheduler = services.GetRequiredService(); var catalogue = await dataProvider.QueryPensAsync(); var extent = await dataProvider.QueryArchiveExtentAsync(); @@ -68,7 +69,7 @@ public async Task TheFirstSeededBreakLeavesTheRenderedCurvesBroken() "the blank-column probe reads chroma, so a greyscale pen would be invisible to it")); using var coordinator = new TrendCoordinator( - dataProvider, catalogue.Value, dataScheduler, ImmediateScheduler.Instance); + dataProvider, liveSource, catalogue.Value, dataScheduler, ImmediateScheduler.Instance); using var chart = new TrendChartViewModel( coordinator, new NoMarkers(), diff --git a/SemiPlot/SemiPlot.Tests.Integration/Journeys/LiveEdgeArchiveJourneyTests.cs b/SemiPlot/SemiPlot.Tests.Integration/Journeys/LiveEdgeArchiveJourneyTests.cs index fa8f27d..440da1e 100644 --- a/SemiPlot/SemiPlot.Tests.Integration/Journeys/LiveEdgeArchiveJourneyTests.cs +++ b/SemiPlot/SemiPlot.Tests.Integration/Journeys/LiveEdgeArchiveJourneyTests.cs @@ -20,7 +20,7 @@ namespace SemiPlot.Tests.Integration.Journeys; // Proves a row written while the app runs reaches the chart's live edge exactly once, composing -// TrendCoordinator, TrendChartViewModel and navigation over RealtimeSubscriptionTests' provider rule. +// TrendCoordinator, TrendChartViewModel and navigation over RealtimeSubscriptionTests' live source rule. // Clones the template because this class appends; every wait is an awaited emission bounded by GateBound. [Collection(ArchiveDatabaseCollection.Name)] [Trait("Component", "UI")] @@ -169,6 +169,7 @@ private async Task OpenStandAsync() try { var dataProvider = services.GetRequiredService(); + var liveSource = services.GetRequiredService(); var dataScheduler = services.GetRequiredService(); var catalogue = await dataProvider.QueryPensAsync(); var extent = await dataProvider.QueryArchiveExtentAsync(); @@ -177,7 +178,7 @@ private async Task OpenStandAsync() extent.IsSuccess.Should().BeTrue(ArchiveReadSupport.Describe(extent)); var coordinator = new TrendCoordinator( - dataProvider, catalogue.Value, dataScheduler, ImmediateScheduler.Instance); + dataProvider, liveSource, catalogue.Value, dataScheduler, ImmediateScheduler.Instance); var messagePanel = new MessagePanelViewModel(); var chart = new TrendChartViewModel( diff --git a/SemiPlot/SemiPlot.Tests.Integration/RealtimeEmptyArchiveTests.cs b/SemiPlot/SemiPlot.Tests.Integration/RealtimeEmptyArchiveTests.cs index 57614e9..8de79cc 100644 --- a/SemiPlot/SemiPlot.Tests.Integration/RealtimeEmptyArchiveTests.cs +++ b/SemiPlot/SemiPlot.Tests.Integration/RealtimeEmptyArchiveTests.cs @@ -73,9 +73,9 @@ public async Task ASubscriptionOverAnArchiveWithNoRowLogsItsBaseline() using var armed = new ArmedGate(); using var services = ArchiveProviderFactory.Build(Database.PlotConnectionString, armed); - var provider = services.GetRequiredService(); + var liveSource = services.GetRequiredService(); - using var subscription = provider.Subscribe(SubscribedPenIds).Subscribe(_ => { }); + using var subscription = liveSource.Subscribe(SubscribedPenIds).Subscribe(_ => { }); await armed.Armed(1).WaitAsync(GateBound, TestContext.Current.CancellationToken); } diff --git a/SemiPlot/SemiPlot.Tests.Integration/RealtimePollTests.cs b/SemiPlot/SemiPlot.Tests.Integration/RealtimePollTests.cs index 94a3eab..367cbcf 100644 --- a/SemiPlot/SemiPlot.Tests.Integration/RealtimePollTests.cs +++ b/SemiPlot/SemiPlot.Tests.Integration/RealtimePollTests.cs @@ -1,7 +1,15 @@ +using System.Reactive; +using System.Reactive.Concurrency; +using System.Reactive.Linq; + using AwesomeAssertions; +using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging.Testing; +using SemiPlot.Core.Data; +using SemiPlot.Core.Trends; +using SemiPlot.DataSource.Postgres; using SemiPlot.DataSource.Postgres.Configuration; using Xunit; @@ -10,9 +18,8 @@ namespace SemiPlot.Tests.Integration; -// A failed tick and its logged run of failures, which needs no server: every tick connects to an address -// nothing answers and fails there. What a successful tick reads is a database question and is covered by -// RealtimePollReadTests. +// A failed tick and its logged run of failures. No server: every tick connects to an address nothing answers. +// What a successful tick reads is covered by RealtimePollReadTests. [Trait("Component", "Core")] [Trait("Area", "Data")] [Trait("Category", "Integration")] @@ -22,6 +29,9 @@ public sealed class RealtimePollTests private static readonly PostgresConnectionSettings _settings = UnreachableArchive.Settings(TimeSpan.FromSeconds(1)); + private static readonly PostgresConnectionSettings _fastSettings = + UnreachableArchive.Settings(TimeSpan.FromMilliseconds(10)); + [Fact] public async Task AFailedTickEmitsNoSample() { @@ -51,4 +61,92 @@ public async Task EachFailedTickLogsTheRunOfFailuresItExtends() .Select(entry => entry.GetStructuredStateValue(ArmedGate.FailureRunKey)) .Should().Equal("1", "2", "3", "4", "5"); } + + [Fact] + public async Task ASubscriptionThatOnlyFailsNeitherCompletesNorFaults() + { + // The last failure runs only if nothing terminal followed the one before it. + const int FailedTicksBeforeAssert = 6; + + using var gate = new ArmedGate(); + await using var services = UnreachableArchive.Services(_fastSettings, gate); + + var liveSource = services.GetRequiredService(); + + using var notifications = new BatchCollector>>( + liveSource.Subscribe([1]).Materialize()); + + await gate.FailuresReached(FailedTicksBeforeAssert) + .WaitAsync(ClonedArchiveTest.GateBound, TestContext.Current.CancellationToken); + + notifications.Batches.Should().BeEmpty(); + } + + [Fact] + public async Task ThePollLogsUnderTheLiveSourcesCategory() + { + using var logs = new FakeLoggerProvider(); + using var gate = new ArmedGate(logs); + await using var services = UnreachableArchive.Services(_fastSettings, gate); + + using var subscription = services.GetRequiredService().Subscribe([1]).Subscribe(_ => { }); + + // The gate counts a line before its inner logger writes it, so the second failure proves the first written. + await gate.FailuresReached(2).WaitAsync(ClonedArchiveTest.GateBound, TestContext.Current.CancellationToken); + + logs.Collector.GetSnapshot() + .Where(entry => entry.GetStructuredStateValue(ArmedGate.FailureRunKey) is not null) + .Should().NotBeEmpty() + .And.OnlyContain(entry => entry.Category == typeof(PostgresLiveSource).FullName); + } + + [Fact] + public async Task ThePollWaitsTheConfiguredIntervalAfterATick() + { + var pollInterval = TimeSpan.FromMilliseconds(1234); + var scheduler = new DueTimeRecorder(); + await using var services = UnreachableArchive.Services( + UnreachableArchive.Settings(pollInterval), + scheduler: scheduler); + + using var subscription = services.GetRequiredService().Subscribe([1]).Subscribe(_ => { }); + + var dueTime = await scheduler.FirstDueTime + .WaitAsync(ClonedArchiveTest.GateBound, TestContext.Current.CancellationToken); + + dueTime.Should().Be(pollInterval); + } + + private sealed class DueTimeRecorder : IScheduler + { + private readonly TaskCompletionSource _firstDueTime = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + public Task FirstDueTime => _firstDueTime.Task; + + public DateTimeOffset Now => DefaultScheduler.Instance.Now; + + public IDisposable Schedule(TState state, Func action) + { + return DefaultScheduler.Instance.Schedule(state, action); + } + + public IDisposable Schedule( + TState state, + TimeSpan dueTime, + Func action) + { + _firstDueTime.TrySetResult(dueTime); + + return DefaultScheduler.Instance.Schedule(state, dueTime, action); + } + + public IDisposable Schedule( + TState state, + DateTimeOffset dueTime, + Func action) + { + return DefaultScheduler.Instance.Schedule(state, dueTime, action); + } + } } diff --git a/SemiPlot/SemiPlot.Tests.Integration/RealtimeSubscriptionDisposalTests.cs b/SemiPlot/SemiPlot.Tests.Integration/RealtimeSubscriptionDisposalTests.cs index f6931ba..2b531ce 100644 --- a/SemiPlot/SemiPlot.Tests.Integration/RealtimeSubscriptionDisposalTests.cs +++ b/SemiPlot/SemiPlot.Tests.Integration/RealtimeSubscriptionDisposalTests.cs @@ -17,7 +17,7 @@ public sealed class RealtimeSubscriptionDisposalTests private const int CalibrationTicks = 3; // Silence alone proves nothing (a leaked loop against this unreachable address is also silent), so a - // second, left-running provider's own failed ticks calibrate the wait. + // second, left-running live source's own failed ticks calibrate the wait. [Fact] public async Task SubscribingAndDisposingAtOnceDeliversNothingAndStopsThePoll() { @@ -28,8 +28,8 @@ public async Task SubscribingAndDisposingAtOnceDeliversNothingAndStopsThePoll() using var droppedServices = UnreachableArchive.Services(settings, droppedGate); using var runningServices = UnreachableArchive.Services(settings, runningGate); - var dropped = droppedServices.GetRequiredService(); - var running = runningServices.GetRequiredService(); + var dropped = droppedServices.GetRequiredService(); + var running = runningServices.GetRequiredService(); using var droppedBatches = new BatchCollector>(dropped.Subscribe([1])); diff --git a/SemiPlot/SemiPlot.Tests.Integration/RealtimeSubscriptionTests.cs b/SemiPlot/SemiPlot.Tests.Integration/RealtimeSubscriptionTests.cs index 3e53921..84cfb73 100644 --- a/SemiPlot/SemiPlot.Tests.Integration/RealtimeSubscriptionTests.cs +++ b/SemiPlot/SemiPlot.Tests.Integration/RealtimeSubscriptionTests.cs @@ -1,3 +1,6 @@ +using System.Reactive; +using System.Reactive.Linq; + using AwesomeAssertions; using Microsoft.Extensions.DependencyInjection; @@ -29,7 +32,7 @@ protected override async ValueTask SeedAsync() await Writer().WriteAsync(SeededRows(), Day, NextDay); } - // The logger belongs to this provider and this one subscription, so every per-tick entry that carries a + // The logger belongs to this live source and this one subscription, so every per-tick entry that carries a // row count (written on each successful tick after the baseline) is the subscription's own. A poll the // disposal left running would write one a second. [Fact] @@ -41,9 +44,9 @@ public async Task DisposingASubscriptionStopsItsPoll() using var armed = new ArmedGate(logs); using var services = ArchiveProviderFactory.Build(Database.PlotConnectionString, armed); - var provider = services.GetRequiredService(); + var liveSource = services.GetRequiredService(); - using var subscription = new BatchCollector>(provider.Subscribe(SubscribedPenIds)); + using var subscription = new BatchCollector>(liveSource.Subscribe(SubscribedPenIds)); await armed.Armed(1).WaitAsync(GateBound, cancellationToken); @@ -82,9 +85,9 @@ public async Task TwoSubscriptionsEachKeepTheirOwnLastSeen() using var armed = new ArmedGate(); using var services = ArchiveProviderFactory.Build(Database.PlotConnectionString, armed); - var provider = services.GetRequiredService(); + var liveSource = services.GetRequiredService(); - using var first = new BatchCollector>(provider.Subscribe(SubscribedPenIds)); + using var first = new BatchCollector>(liveSource.Subscribe(SubscribedPenIds)); await armed.Armed(1).WaitAsync(GateBound, cancellationToken); @@ -94,7 +97,7 @@ await AppendAsync( Database, SubscribedPenIds[0], beforeTheSecondSubscription, 1, cancellationToken: cancellationToken); await firstSeesTheEarlyRow.WaitAsync(GateBound, cancellationToken); - using var second = new BatchCollector>(provider.Subscribe(SubscribedPenIds)); + using var second = new BatchCollector>(liveSource.Subscribe(SubscribedPenIds)); await armed.Armed(2).WaitAsync(GateBound, cancellationToken); @@ -117,6 +120,44 @@ await AppendAsync( Delivered(second).Should().Equal([ArchiveProviderFactory.TimeConverter.ToUtc(afterBoth)]); } + // The archive holds no row for the second identifier, which is what an identifier the source does not + // serve looks like to the poll. + [Fact] + public async Task AnIdentifierTheSourceDoesNotServeLeavesTheStreamRunning() + { + const int UnservedPenId = 9999; + var cancellationToken = TestContext.Current.CancellationToken; + + using var armed = new ArmedGate(); + using var services = ArchiveProviderFactory.Build(Database.PlotConnectionString, armed); + + var liveSource = services.GetRequiredService(); + + using var notifications = new BatchCollector>>( + liveSource.Subscribe([SubscribedPenIds[0], UnservedPenId]).Materialize()); + + await armed.Armed(1).WaitAsync(GateBound, cancellationToken); + + var firstDelivery = notifications.NextBatch(); + + await AppendAsync( + Database, SubscribedPenIds[0], SeededLast.AddSeconds(1), 1, cancellationToken: cancellationToken); + await firstDelivery.WaitAsync(GateBound, cancellationToken); + + var secondDelivery = notifications.NextBatch(); + + await AppendAsync( + Database, SubscribedPenIds[0], SeededLast.AddSeconds(2), 2, cancellationToken: cancellationToken); + await secondDelivery.WaitAsync(GateBound, cancellationToken); + + notifications.Batches.Should().HaveCount(2) + .And.OnlyContain(notification => notification.Kind == NotificationKind.OnNext); + notifications.Batches + .SelectMany(notification => notification.Value) + .Select(sample => sample.PenId) + .Should().Equal(SubscribedPenIds[0], SubscribedPenIds[0]); + } + private static IReadOnlyList PollEntries(FakeLoggerProvider logs) { return [.. logs.Collector.GetSnapshot().Where(entry => entry.GetStructuredStateValue("RowCount") is not null)]; diff --git a/SemiPlot/SemiPlot.Tests.Integration/UnreachableArchive.cs b/SemiPlot/SemiPlot.Tests.Integration/UnreachableArchive.cs index ab7684c..76c8649 100644 --- a/SemiPlot/SemiPlot.Tests.Integration/UnreachableArchive.cs +++ b/SemiPlot/SemiPlot.Tests.Integration/UnreachableArchive.cs @@ -1,3 +1,5 @@ +using System.Reactive.Concurrency; + using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; @@ -25,11 +27,19 @@ public static PostgresConnectionSettings Settings(TimeSpan pollInterval) public static ServiceProvider Services( PostgresConnectionSettings settings, - ILoggerProvider? loggerProvider = null) + ILoggerProvider? loggerProvider = null, + IScheduler? scheduler = null) { - return new ServiceCollection() + var services = new ServiceCollection() .AddTestLogging(loggerProvider) - .AddPostgresData(settings) - .BuildServiceProvider(); + .AddPostgresData(settings); + + if (scheduler is not null) + { + // Registered after AddPostgresData's own scheduler, so every resolution takes this one. + services.AddSingleton(scheduler); + } + + return services.BuildServiceProvider(); } } diff --git a/SemiPlot/SemiPlot.Tests.Unit/Postgres/PostgresCompositionTests.cs b/SemiPlot/SemiPlot.Tests.Unit/Postgres/PostgresCompositionTests.cs index 2fa5dd4..798ace7 100644 --- a/SemiPlot/SemiPlot.Tests.Unit/Postgres/PostgresCompositionTests.cs +++ b/SemiPlot/SemiPlot.Tests.Unit/Postgres/PostgresCompositionTests.cs @@ -51,6 +51,16 @@ public void AddPostgresDataResolvesThePostgresProvider() provider.Should().BeOfType(); } + [Fact] + public void AddPostgresDataResolvesThePostgresLiveSource() + { + using var services = BuildProvider(); + + var liveSource = services.GetRequiredService(); + + liveSource.Should().BeOfType(); + } + // The data source is a DI singleton, and ServiceProvider.Dispose() throws for an instantiated // singleton implementing only IAsyncDisposable. Resolving one and then disposing synchronously is the // case that catches an async-only wrapper. @@ -82,6 +92,7 @@ public void TheResolvedConverterCarriesTheConfiguredSourceZone() // cannot tell the two apart. [Theory] [InlineData(typeof(IDataProvider))] + [InlineData(typeof(ILiveSource))] [InlineData(typeof(IPenCatalogueEditor))] [InlineData(typeof(IMarkerStore))] [InlineData(typeof(IScheduler))] diff --git a/SemiPlot/SemiPlot.Tests.Unit/UI/Bridge/FakeDataProvider.cs b/SemiPlot/SemiPlot.Tests.Unit/UI/Bridge/FakeDataProvider.cs index d85bea0..ab3b181 100644 --- a/SemiPlot/SemiPlot.Tests.Unit/UI/Bridge/FakeDataProvider.cs +++ b/SemiPlot/SemiPlot.Tests.Unit/UI/Bridge/FakeDataProvider.cs @@ -16,7 +16,7 @@ namespace SemiPlot.Tests.Unit.UI.Bridge; internal sealed class FakeDataProvider( IScheduler scheduler, TimeSpan realtimeInterval, - IReadOnlyList? pens = null) : IDataProvider + IReadOnlyList? pens = null) : IDataProvider, ILiveSource { // The default last-column center value an unoverridden layer returns; tests assert against this. public const double DefaultCenter = 2.0; @@ -156,10 +156,12 @@ private IReadOnlyList HistoryQueries // TrendCoordinator.BuildRealtimeBatch looks like from the pipeline's side. public bool PoisonRealtimeWindow { get; set; } - // A provider that ends the live edge with OnError, the one terminal message TrendCoordinator.RealtimeBatches - // has to survive. The real provider catches everything inside its poll, so only this fake reaches it. + // A live source that ends the live edge with OnError, the terminal message TrendCoordinator.RealtimeBatches + // has to survive. The real live source catches everything inside its poll, so only this fake reaches it. public Exception? RealtimeStreamFailure { get; set; } + public bool RealtimeStreamCompletes { get; set; } + /// The next tick follows the pen's value with a break 1 ms later, as the poll's q = 32 row. public void BreakRealtimePenOnce(int penId) { @@ -214,6 +216,11 @@ private IObservable> BuildLiveEdge(IReadOnlyList penI return Observable.Throw>(failure, _scheduler); } + if (RealtimeStreamCompletes) + { + return Observable.Empty>(_scheduler); + } + var subscribed = penIds.Where(id => Pens.Any(pen => pen.PenId == id)).ToArray(); return Observable diff --git a/SemiPlot/SemiPlot.Tests.Unit/UI/Bridge/TrendCoordinatorTests.cs b/SemiPlot/SemiPlot.Tests.Unit/UI/Bridge/TrendCoordinatorTests.cs index f21a251..86b2095 100644 --- a/SemiPlot/SemiPlot.Tests.Unit/UI/Bridge/TrendCoordinatorTests.cs +++ b/SemiPlot/SemiPlot.Tests.Unit/UI/Bridge/TrendCoordinatorTests.cs @@ -67,7 +67,7 @@ public void RealtimeBatch_KeepsEachPenOnItsOwnTimestamps() } [Fact] - public void ABreakTheProviderReports_SurvivesTheBufferWindowOnItsOwnTimestamp() + public void ABreakTheLiveSourceReports_SurvivesTheBufferWindowOnItsOwnTimestamp() { var (coordinator, scheduler, provider) = CreateCoordinator(realtimeInterval: ChartTestBuilder.RealtimeInterval); using var ownedCoordinator = coordinator; @@ -95,6 +95,7 @@ public void RealtimeBatches_WithAnEmptyCatalog_EmitNothing() var scheduler = new TestScheduler(); var provider = new FakeDataProvider(scheduler, ChartTestBuilder.RealtimeInterval); using var coordinator = new TrendCoordinator( + provider, provider, [], scheduler, @@ -107,18 +108,19 @@ public void RealtimeBatches_WithAnEmptyCatalog_EmitNothing() batches.Should().BeEmpty(); } - // The provider ending the stream reaches every batch observer: with no onError Rx rethrows it out of the + // A live source that faults reaches every batch observer: with no onError Rx rethrows the fault out of the // advance below and takes the emitting thread down. [Fact] public void AFaultingRealtimeStream_ReachesTheFailureChannelInsteadOfTheScheduler() { var scheduler = new TestScheduler(); - var thrown = new InvalidOperationException("the provider ended the live edge"); + var thrown = new InvalidOperationException("the live source ended the live edge"); var provider = new FakeDataProvider(scheduler, ChartTestBuilder.RealtimeInterval) { RealtimeStreamFailure = thrown }; using var coordinator = new TrendCoordinator( + provider, provider, provider.Pens, scheduler, @@ -171,7 +173,7 @@ public void SetPens_WithAnAddedId_ReachesTheOneSubscriberOfTheLiveEdge() } [Fact] - public void SetPens_LeavesTheProviderHoldingExactlyOneLiveSubscription() + public void SetPens_LeavesTheLiveSourceHoldingExactlyOneLiveSubscription() { var (coordinator, scheduler, provider) = CreateCoordinator(realtimeInterval: ChartTestBuilder.RealtimeInterval); using var ownedCoordinator = coordinator; @@ -188,7 +190,7 @@ public void SetPens_LeavesTheProviderHoldingExactlyOneLiveSubscription() } [Fact] - public void AProviderFailureAfterASwitch_ReachesRealtimeFailuresOnce() + public void ALiveSourceFailureAfterASwitch_ReachesRealtimeFailuresOnce() { var (coordinator, scheduler, provider) = CreateCoordinator(realtimeInterval: ChartTestBuilder.RealtimeInterval); using var ownedCoordinator = coordinator; @@ -197,7 +199,7 @@ public void AProviderFailureAfterASwitch_ReachesRealtimeFailuresOnce() using var liveEdge = coordinator.RealtimeBatches.Subscribe(_ => { }); scheduler.AdvanceBy(TrendCoordinator.BatchWindow.Ticks); - var streamFailure = new InvalidOperationException("the provider ended the switched live edge"); + var streamFailure = new InvalidOperationException("the live source ended the switched live edge"); provider.RealtimeStreamFailure = streamFailure; coordinator.SetPens([1]); var advance = () => scheduler.AdvanceBy(TrendCoordinator.BatchWindow.Ticks * 3); @@ -206,12 +208,82 @@ public void AProviderFailureAfterASwitch_ReachesRealtimeFailuresOnce() failures.Should().ContainSingle().Which.Should().BeSameAs(streamFailure); } + [Fact] + public async Task TheLiveEdgeFollowsTheLiveSource_AndTheReadsGoToTheProvider() + { + const int ChartColumnCount = 100; + var scheduler = new TestScheduler(); + var dataProvider = new FakeDataProvider(scheduler, ChartTestBuilder.RealtimeInterval); + var liveSource = new FakeDataProvider( + scheduler, + ChartTestBuilder.RealtimeInterval, + [new Pen(3, "Pen 3", ["Group B"], "#0000ff")]); + using var coordinator = new TrendCoordinator( + dataProvider, + liveSource, + liveSource.Pens, + scheduler, + ImmediateScheduler.Instance); + var batches = new List(); + using var liveEdge = coordinator.RealtimeBatches.Subscribe(batches.Add); + + scheduler.AdvanceBy(TrendCoordinator.BatchWindow.Ticks); + + batches.Should().ContainSingle().Which.Pens.Select(values => values.PenId).Should().Equal(3); + + await coordinator.QueryHistoryAsync( + [FakeDataProvider.FirstPen.PenId], + FakeDataProvider.RealtimeEpoch.AddHours(-1), + FakeDataProvider.RealtimeEpoch, + AggregationLayer.Raw, + ChartColumnCount, + TestContext.Current.CancellationToken); + await coordinator.QueryArchiveExtentAsync(); + + liveSource.LiveSubscriptionsOpened.Should().Be(1); + liveSource.HistoryQueryCount.Should().Be(0); + liveSource.ExtentQueryCount.Should().Be(0); + dataProvider.LiveSubscriptionsOpened.Should().Be(0); + dataProvider.HistoryQueryCount.Should().Be(1); + dataProvider.ExtentQueryCount.Should().Be(1); + } + + // docs/architecture/data-integration.md#realtime + [Fact] + public void ALiveSourceThatCompletes_LeavesTheLiveEdgeSilentUntilTheNextSetPens() + { + var (coordinator, scheduler, provider) = CreateCoordinator(realtimeInterval: ChartTestBuilder.RealtimeInterval); + using var ownedCoordinator = coordinator; + var batches = new List(); + var failures = new List(); + var completed = false; + using var reported = coordinator.RealtimeFailures.Subscribe(failures.Add); + provider.RealtimeStreamCompletes = true; + using var liveEdge = coordinator.RealtimeBatches.Subscribe(batches.Add, () => completed = true); + + scheduler.AdvanceBy(TrendCoordinator.BatchWindow.Ticks * 2); + + batches.Should().BeEmpty(); + failures.Should().BeEmpty(); + completed.Should().BeFalse(); + provider.OpenLiveSubscriptionCount.Should().Be(0); + + provider.RealtimeStreamCompletes = false; + coordinator.SetPens([FakeDataProvider.FirstPen.PenId]); + scheduler.AdvanceBy(TrendCoordinator.BatchWindow.Ticks); + + batches.Should().ContainSingle().Which.Pens.Select(values => values.PenId).Should() + .Equal(FakeDataProvider.FirstPen.PenId); + failures.Should().BeEmpty(); + } + private static (TrendCoordinator Coordinator, TestScheduler Scheduler, FakeDataProvider Provider) CreateCoordinator(TimeSpan? realtimeInterval = null) { var scheduler = new TestScheduler(); var provider = new FakeDataProvider(scheduler, realtimeInterval ?? ChartTestBuilder.RealtimeInterval); var coordinator = new TrendCoordinator( + provider, provider, provider.Pens, scheduler, diff --git a/SemiPlot/SemiPlot.Tests.Unit/UI/Chart/ChartTestBuilder.cs b/SemiPlot/SemiPlot.Tests.Unit/UI/Chart/ChartTestBuilder.cs index d9914a9..394919c 100644 --- a/SemiPlot/SemiPlot.Tests.Unit/UI/Chart/ChartTestBuilder.cs +++ b/SemiPlot/SemiPlot.Tests.Unit/UI/Chart/ChartTestBuilder.cs @@ -57,7 +57,7 @@ public static TrendChartViewModel CreateChart( public static TrendCoordinator CreateCoordinator(TestScheduler scheduler, FakeDataProvider provider) { - return new TrendCoordinator(provider, provider.Pens, scheduler, ImmediateScheduler.Instance); + return new TrendCoordinator(provider, provider, provider.Pens, scheduler, ImmediateScheduler.Instance); } // The UI scheduler is immediate unless a test passes its own, so no default-built test realises a view. diff --git a/SemiPlot/SemiPlot.Tests.Unit/UI/Chart/TrendChartViewModelTests.cs b/SemiPlot/SemiPlot.Tests.Unit/UI/Chart/TrendChartViewModelTests.cs index a26e98f..5629c3a 100644 --- a/SemiPlot/SemiPlot.Tests.Unit/UI/Chart/TrendChartViewModelTests.cs +++ b/SemiPlot/SemiPlot.Tests.Unit/UI/Chart/TrendChartViewModelTests.cs @@ -2601,10 +2601,10 @@ public void AThrowInTheBatchProjectionIsReportedAndTheRealtimeApplyContinues() } [AvaloniaFact] - public void AProviderThatEndsTheLiveEdge_IsReportedOnce() + public void ALiveSourceThatEndsTheLiveEdge_IsReportedOnce() { var scheduler = new TestScheduler(); - var thrown = new InvalidOperationException("the provider ended the live edge"); + var thrown = new InvalidOperationException("the live source ended the live edge"); var provider = new FakeDataProvider(scheduler, RealtimeInterval) { RealtimeStreamFailure = thrown diff --git a/SemiPlot/SemiPlot.Tests.Unit/UI/Di/CompositionRootTests.cs b/SemiPlot/SemiPlot.Tests.Unit/UI/Di/CompositionRootTests.cs index 5c061fd..590993e 100644 --- a/SemiPlot/SemiPlot.Tests.Unit/UI/Di/CompositionRootTests.cs +++ b/SemiPlot/SemiPlot.Tests.Unit/UI/Di/CompositionRootTests.cs @@ -32,6 +32,14 @@ public void DefaultContainer_ResolvesThePostgresProvider() provider.GetRequiredService().Should().BeOfType(); } + [Fact] + public void Container_ResolvesTheLiveSource() + { + using var provider = BuildContainer(); + + provider.GetRequiredService().Should().BeOfType(); + } + [Fact] public void Container_ResolvesThePenCatalogueEditor() { diff --git a/SemiPlot/SemiPlot.Tests.Unit/UI/Di/TrendWindowBuildTests.cs b/SemiPlot/SemiPlot.Tests.Unit/UI/Di/TrendWindowBuildTests.cs index 6b38672..96aafd4 100644 --- a/SemiPlot/SemiPlot.Tests.Unit/UI/Di/TrendWindowBuildTests.cs +++ b/SemiPlot/SemiPlot.Tests.Unit/UI/Di/TrendWindowBuildTests.cs @@ -6,6 +6,7 @@ using AwesomeAssertions; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Reactive.Testing; using ReactiveUI.Avalonia; @@ -86,4 +87,25 @@ public async Task TheWindowReadsTheCatalogueAtOnce_OnTheRealScheduler() dataProvider.PensQueryCount.Should().Be(pensQueriesBefore + 1); } + + // One fake serves both interfaces elsewhere, so only a second instance tells a cast from a resolve. + [AvaloniaFact] + public async Task TheWindowFollowsTheContainersLiveSource() + { + var liveSource = new FakeDataProvider(new TestScheduler(), TimeSpan.FromSeconds(1)); + using var stand = NewArchiveStand(liveSource: liveSource); + + var probe = await StartupProbe.ReadAsync(stand.Data.ServiceProvider, StartupProbe.DefaultReadBound); + + using var window = TrendWindow.Build( + probe.Value, + TestLaunch.LauncherAt(AppContext.BaseDirectory), + AvaloniaScheduler.Instance, + App.CsvDialectFor(UiLanguage.En), + TimeZoneInfo.Utc); + Dispatcher.UIThread.RunJobs(); + + liveSource.LiveSubscriptionsOpened.Should().Be(1); + stand.Provider.LiveSubscriptionsOpened.Should().Be(0); + } } diff --git a/SemiPlot/SemiPlot.Tests.Unit/UI/MainWindow/MainWindowTestBuilder.cs b/SemiPlot/SemiPlot.Tests.Unit/UI/MainWindow/MainWindowTestBuilder.cs index e10d325..d151121 100644 --- a/SemiPlot/SemiPlot.Tests.Unit/UI/MainWindow/MainWindowTestBuilder.cs +++ b/SemiPlot/SemiPlot.Tests.Unit/UI/MainWindow/MainWindowTestBuilder.cs @@ -98,7 +98,8 @@ public static ArchiveStand NewArchiveStand( IReadOnlyList? pens = null, IPenCatalogueEditor? penCatalogueEditor = null, MessagePanelViewModel? panel = null, - IArchiveRowReader? rowReader = null) + IArchiveRowReader? rowReader = null, + ILiveSource? liveSource = null) { var scheduler = new TestScheduler(); var provider = new FakeDataProvider(scheduler, TimeSpan.FromSeconds(1), pens); @@ -106,6 +107,7 @@ public static ArchiveStand NewArchiveStand( var services = new ServiceCollection() .AddSingleton(scheduler) .AddSingleton(provider) + .AddSingleton(liveSource ?? provider) .AddSingleton(penCatalogueEditor ?? new FakePenCatalogueEditor()) .AddSingleton(rowReader ?? new FakeArchiveRowReader()) .AddSingleton(markerStore) diff --git a/SemiPlot/SemiPlot.Tests.Unit/UI/Minimap/MinimapStand.cs b/SemiPlot/SemiPlot.Tests.Unit/UI/Minimap/MinimapStand.cs index ae2ed7c..e826d8e 100644 --- a/SemiPlot/SemiPlot.Tests.Unit/UI/Minimap/MinimapStand.cs +++ b/SemiPlot/SemiPlot.Tests.Unit/UI/Minimap/MinimapStand.cs @@ -34,7 +34,7 @@ public MinimapStand(ILogger? logger = null) ArchiveFirstUtc = ExtentFirst, ArchiveLastUtc = ExtentLast }; - _coordinator = new TrendCoordinator(Provider, Provider.Pens, Scheduler, Scheduler); + _coordinator = new TrendCoordinator(Provider, Provider, Provider.Pens, Scheduler, Scheduler); Chart = ChartTestBuilder.CreateChart(Scheduler, _coordinator, Panel, MarkerChannel); ViewModel = new MinimapViewModel( _coordinator, Chart, Scheduler, Panel, logger ?? NullLogger.Instance); diff --git a/SemiPlot/SemiPlot.UI/Bridge/TrendCoordinator.cs b/SemiPlot/SemiPlot.UI/Bridge/TrendCoordinator.cs index 02a2928..6a8e0ff 100644 --- a/SemiPlot/SemiPlot.UI/Bridge/TrendCoordinator.cs +++ b/SemiPlot/SemiPlot.UI/Bridge/TrendCoordinator.cs @@ -14,24 +14,26 @@ public sealed class TrendCoordinator : IDisposable internal static readonly TimeSpan BatchWindow = TimeSpan.FromMilliseconds(100); private readonly IDataProvider _dataProvider; + private readonly ILiveSource _liveSource; private readonly IScheduler _dataScheduler; private readonly IScheduler _uiScheduler; private readonly BehaviorSubject> _penIds; - // Written from the data scheduler (a failed window, a provider that ends the stream) and completed from + // Written from the data scheduler (a failed window, a live source that faults) and completed from // the UI thread at disposal, so the subject serializes its callers. private readonly ISubject _realtimeFailures = Subject.Synchronize(new Subject()); - // Every pen set, pens here and each one SetPens pushes, must be dataProvider's own catalogue: the - // coordinator subscribes to these identifiers without asking the provider whether it knows them, and a - // provider silently drops the ones it does not. + // Every pen set, pens here and each one SetPens pushes, holds identifiers the live source serves + // (docs/architecture/data-integration.md#realtime). public TrendCoordinator( IDataProvider dataProvider, + ILiveSource liveSource, IReadOnlyList pens, IScheduler dataScheduler, IScheduler uiScheduler) { _dataProvider = dataProvider; + _liveSource = liveSource; _dataScheduler = dataScheduler; _uiScheduler = uiScheduler; _penIds = new BehaviorSubject>([.. pens.Select(pen => pen.PenId)]); @@ -40,14 +42,14 @@ public TrendCoordinator( } /// - /// The live edge. It never faults: a provider that ends the stream is reported through + /// The live edge. It never faults: a live source that faults is reported through /// and this stream completes instead. /// public IObservable RealtimeBatches { get; } /// /// Every realtime failure once: a throw inside the batch projection, which the pipeline skips, and a - /// provider that ends the stream. Republished on the UI scheduler, like . + /// live source that faults. Republished on the UI scheduler, like . /// public IObservable RealtimeFailures { get; } @@ -95,7 +97,7 @@ public Task>> QueryArchiveDaysAsync( private IObservable BuildRealtimeBatches() { return _penIds - .Select(_dataProvider.Subscribe) + .Select(_liveSource.Subscribe) .Switch() .Buffer(BatchWindow, _dataScheduler) .Select(TryBuildRealtimeBatch) diff --git a/SemiPlot/SemiPlot.UI/MainWindow/TrendWindow.cs b/SemiPlot/SemiPlot.UI/MainWindow/TrendWindow.cs index d692a48..251a3af 100644 --- a/SemiPlot/SemiPlot.UI/MainWindow/TrendWindow.cs +++ b/SemiPlot/SemiPlot.UI/MainWindow/TrendWindow.cs @@ -70,6 +70,7 @@ public static TrendWindow Build( { var serviceProvider = startupData.ServiceProvider; var dataProvider = serviceProvider.GetRequiredService(); + var liveSource = serviceProvider.GetRequiredService(); var dataScheduler = serviceProvider.GetRequiredService(); var messagePanel = serviceProvider.GetRequiredService(); var penCatalogueEditor = serviceProvider.GetRequiredService(); @@ -77,7 +78,8 @@ public static TrendWindow Build( var markerStore = serviceProvider.GetRequiredService(); var loggerFactory = serviceProvider.GetRequiredService(); - var coordinator = new TrendCoordinator(dataProvider, startupData.Pens, dataScheduler, uiScheduler); + var coordinator = new TrendCoordinator( + dataProvider, liveSource, startupData.Pens, dataScheduler, uiScheduler); var markerSync = new MarkerSync( markerStore, messagePanel, uiScheduler, loggerFactory.CreateLogger()); diff --git a/docs/architecture/README.md b/docs/architecture/README.md index 471f9a5..6974ccc 100644 --- a/docs/architecture/README.md +++ b/docs/architecture/README.md @@ -19,10 +19,10 @@ implementation plans see `docs/plans/`. the application's own surface keys, how the `theme` key picks a variant, the five ScottPlot surfaces, and the rule that a colour literal in AXAML is a defect. - [data-integration.md](./data-integration.md) - the contract between SemiPlot and the archive - database: responsibility zones, the `IDataProvider` surface, the statement each operation issues - and what it must keep, layer selection and the measured read cost per layer, the time boundary, gap - mapping, error semantics, the connection settings with the session's `jit = off`, the startup path - and field triage. + database: responsibility zones, the `IDataProvider` and `ILiveSource` surfaces, the statement each + operation issues and what it must keep, layer selection and the measured read cost per layer, the + time boundary, gap mapping, error semantics, the connection settings with the session's `jit = off`, + the startup path and field triage. - [scada-archive.md](./scada-archive.md) — the Simple-Scada 2 archive as it exists: tables and columns, archive layers, quality marks and gaps, write and retention behaviour, reader hazards, and what remains unverified. diff --git a/docs/architecture/charting.md b/docs/architecture/charting.md index ddfb961..f807b95 100644 --- a/docs/architecture/charting.md +++ b/docs/architecture/charting.md @@ -64,7 +64,7 @@ later read seats starts thin, as in a new window. **Gaps via NaN.** A column carrying `NaN` in `Min`/`Max` is the envelope's gap marker: `Build` breaks the path there and the next real column opens a new segment with a `MoveTo`. A `NaN` column comes from one of two routes. The history path writes one after a bucket whose `breaks` flag a `q = 32` marker or a null -`v` sets (`data-integration.md#quality-and-gaps`). The live edge writes one for a break the provider +`v` sets (`data-integration.md#quality-and-gaps`). The live edge writes one for a break the live source reported: `TrendPenState.AppendRealtime` encodes a null value as a `NaN` column and keeps `CurrentValue` at the last real value. @@ -770,9 +770,9 @@ floor is read again, while a window far from either floor reads nothing (`overvi ## Data contract (UI ↔ provider) -The viewer consumes data only through `IDataProvider` (see data-integration.md), in-process and -strongly typed — **no JSON message bridge**. `TrendCoordinator` is the Rx hub between provider and -view model: +The viewer consumes data only through `IDataProvider` and, for the live edge, `ILiveSource` (see +data-integration.md), in-process and strongly typed — **no JSON message bridge**. `TrendCoordinator` is +the Rx hub between the two and the view model: - **History:** `QueryHistoryAsync(penIds, from, to, layer, targetColumnCount)` is the single history query, returning one `PenHistoryEnvelope` per pen (ascending `Timestamps` + `Min` + `Max` + @@ -780,16 +780,17 @@ view model: `ChartHistoryRequestDebouncer`, which runs one query at a time and lets the newest window asked for run last. The minimap band is a second reader outside the debouncer: one pen at 250 columns every 2-60 s (`trend-interaction.md#archive-overview-minimap`). -- **Realtime:** `IObservable` — an ascending union timeline the live edge advances +- **Realtime:** the live edge comes through `ILiveSource` as lists of `Sample`, and the coordinator + republishes it as `IObservable` — an ascending union timeline the live edge advances from, plus one `PenRealtimeValues` per pen carrying that pen's **own** timestamps and `double?` values; buffered on the data scheduler and observed on the UI scheduler. The values are per pen rather than a column over the union because the archive is per-variable and change-based with a deadband, so one buffer window routinely spans timestamps only one pen sampled. A pen carries no entry, null or real, at a timestamp it did not report, and a consumer leaves it alone there. A null - is a break the provider reported, a `Sample` whose `Value` is null: the pen has no data from that + is a break the live source reported, a `Sample` whose `Value` is null: the pen has no data from that moment until its next sample, and the chart appends a `NaN` column there. `RealtimePoll` writes one 1 ms after a `q = 32` row (`data-integration.md#realtime`). The pen - set the stream carries follows the catalogue: `TrendCoordinator.SetPens` switches the provider + set the stream carries follows the catalogue: `TrendCoordinator.SetPens` switches the live-source subscription and `RealtimeBatches` stays one published stream (`data-integration.md#realtime`). - **Archive extent:** `QueryArchiveExtentAsync()` returns an `ArchiveExtent(FirstUtc, LastUtc)` — the full stored time span. `TrendCoordinator.QueryArchiveExtentAsync()` is a pass-through to the diff --git a/docs/architecture/data-integration.md b/docs/architecture/data-integration.md index 8a83331..a0df4c6 100644 --- a/docs/architecture/data-integration.md +++ b/docs/architecture/data-integration.md @@ -27,20 +27,17 @@ Two rules follow and are not negotiable: SemiPlot never writes to vendor objects ## The provider surface -The UI reads through `IDataProvider` (`SemiPlot.Core/Data/IDataProvider.cs`), and two write paths sit -apart from it: the pen editor writes through `IPenCatalogueEditor` (The pen catalogue editor, below), -and the chart's time markers are read and written through `IMarkerStore` (The marker store, below). The -interfaces and their DTOs live in `SemiPlot.Core`; each concrete provider is a sibling `SemiPlot.DataSource.*` -project, so Core never references a data source. `PostgresDataProvider` in -`SemiPlot.DataSource.Postgres` is the only implementation of `IDataProvider`; tests build fakes -against the interface. +The UI reads through `IDataProvider` (`SemiPlot.Core/Data/IDataProvider.cs`), the chart's live edge +follows `ILiveSource` (below), and two write paths sit apart from both: the pen editor writes through +`IPenCatalogueEditor` (The pen catalogue editor, below), and the chart's time markers are read and +written through `IMarkerStore` (The marker store, below). The interfaces and their DTOs live in +`SemiPlot.Core`; each concrete provider is a sibling `SemiPlot.DataSource.*` project, so Core never +references a data source. `PostgresDataProvider` in `SemiPlot.DataSource.Postgres` is the only +implementation of `IDataProvider`; tests build fakes against the interface. ```csharp public interface IDataProvider { - // Cold per call: no samples flow until subscribed; the subscriber disposes the returned IDisposable. - IObservable> Subscribe(IReadOnlyList penIds); - Task>> QueryPensAsync(); Task>> QueryHistoryAsync( @@ -64,7 +61,7 @@ public interface IDataProvider | Type | Shape | Notes | | --- | --- | --- | | `Pen` | `PenId`, `Name`, `Groups`, `Color`, `Unit`, `Format`, `EnabledOnStart`, `ScaleMinOnStart`, `ScaleMaxOnStart`, `LogScaleOnStart`, `LineStyle` | `PenId` is the archive's `trends.id`. Equality compares `Groups` element by element, ordinal, so two reads of one stored pen are equal; `LogScaleOnStart` takes part in it, so a pen whose flag alone changed is revised. | -| `Sample` | `PenId`, `TimestampUtc`, `Value` | Realtime element. Timestamps are UTC by the time they leave the provider. A null `Value` is a break the provider reported: the pen has no data from that moment until its next sample. | +| `Sample` | `PenId`, `TimestampUtc`, `Value` | Realtime element. Timestamps are UTC by the time they leave the live source. A null `Value` is a break the live source reported: the pen has no data from that moment until its next sample. | | `PenHistoryEnvelope` | parallel `Timestamps` / `Min` / `Max` / `Center`, strictly ascending, `NaN` marks a gap | One per pen per history query. | | `ArchiveExtent` | `FirstUtc`, `LastUtc`, `IsEmpty` | The span of the configured variables, consumed by the minimap (`TM-4`). `ArchiveExtent.Empty` is the no-span form. | | `AggregationLayer` | `Raw`, `Minute`, `Hour` | Maps one-to-one onto the archive's `l` column. | @@ -126,6 +123,20 @@ for 16 pens and 450 000 for 50, which the reader holds as one list. Measured on 30 s `statement_timeout`, so the width stayed at 15 minutes. The export reads one such slice at a time; a 24 h range is 96 slice statements plus one seed probe per pen. +The live edge reads through a port of its own, `ILiveSource` (`SemiPlot.Core/Data/ILiveSource.cs`), with +one member. `PostgresLiveSource` in `SemiPlot.DataSource.Postgres` implements it by polling the archive, and +`AddPostgresData` registers it beside the provider over the same `NpgsqlDataSource`, with the operator's +`poll_interval_ms` and the bare data `IScheduler`. Its one consumer is `TrendCoordinator`, which +`TrendWindow.Build` hands both interfaces. The contract, and what the coordinator does with a source that +breaks it, are under Realtime, below. + +```csharp +public interface ILiveSource +{ + IObservable> Subscribe(IReadOnlyList penIds); +} +``` + ## The pen catalogue editor The write path is a second interface, `IPenCatalogueEditor` (`SemiPlot.Core/Data/IPenCatalogueEditor.cs`), @@ -308,11 +319,12 @@ of the repeated autumn hour. ## Operation to statement -Every statement the provider, the editor, the row reader and the marker store issue is a constant in -`SemiPlot.DataSource.Postgres/ArchiveStatements.cs`; parameters are always bound. The bench seeder -and the test harness own SQL of their own (`bench.md`). `ArchiveStatementTextTests` pins each -provider constant below clause by clause except `PenCatalog` and `ArchiveExtent`; `ExplainPlanTests` -asserts the plan shape of each one except `PenCatalog` against a container. The editor's and the marker store's statements are in their own tables above. +Every statement the provider, the live source, the editor, the row reader and the marker store issue is +a constant in `SemiPlot.DataSource.Postgres/ArchiveStatements.cs`; parameters are always bound. The +bench seeder and the test harness own SQL of their own (`bench.md`). `ArchiveStatementTextTests` pins +each constant below clause by clause except `PenCatalog` and `ArchiveExtent`; `ExplainPlanTests` +asserts the plan shape of each one except `PenCatalog` against a container. The editor's and the +marker store's statements are in their own tables above. | Operation | Constant | What the statement must keep, and why | | --- | --- | --- | @@ -940,11 +952,49 @@ nulls either, and `ArchiveStatementTextTests` plus `BucketedRowFoldTests` pin th ## Realtime -`Subscribe` returns a cold observable: each subscription runs a poll loop of its own on the injected -data scheduler, at the operator's `poll_interval_ms`, holding a baseline of its own and carrying the -variable list in every query (`RT-1`). Disposing the subscription cancels the loop. Batching, the -union timeline and the hand-off to the UI scheduler happen above the provider, in `TrendCoordinator` -(`charting.md`). +The live edge reads through `ILiveSource.Subscribe(penIds)` (The provider surface, above). Every live +source keeps this contract: + +- **Cold per call.** No sample flows until a subscription opens. Each subscription does its own work, and + disposing it stops that work. +- **The stream never completes and never faults.** A failure inside a source is the source's to log or + to report out of band, and it emits nothing. + `RealtimePollTests.ASubscriptionThatOnlyFailsNeitherCompletesNorFaults` pins it for `PostgresLiveSource`. +- **A break is a `Sample` whose `Value` is null,** at a moment the source itself observes: the pen has no + data from that moment until its next sample. +- **Timestamps are UTC.** +- **An identifier the source does not serve emits nothing** and leaves the stream running. + `RealtimeSubscriptionTests.AnIdentifierTheSourceDoesNotServeLeavesTheStreamRunning` pins it for + `PostgresLiveSource`. + +Ordering is not part of the port: the chart appends a pen's sample only past that pen's last column +(`DA-7`) and drops any other. + +Either terminal message breaks the contract, and neither crashes the window. `TrendCoordinator` catches +an `OnError` after the `Switch` and ahead of the `Publish`: `ReportTerminalFailure` writes one +`RealtimeFailures` entry and swaps in `Observable.Empty`, which completes `RealtimeBatches` and drops the +subscription to the pen set. Nothing rebuilds it, so the live edge stays empty for the rest of the +window's life (No failure stops at the log, below). An `OnCompleted` is not caught: `Switch` idles on a +completed inner stream, so the live edge goes silent with no entry until the next `SetPens` opens a new +subscription. `TrendCoordinatorTests.ALiveSourceThatCompletes_LeavesTheLiveEdgeSilentUntilTheNextSetPens` +pins the second case. + +A break travels through the port when the source can decide it from its own input, as the archive poll +does from a `q = 32` row. The port carries no chart state, so a source cannot know a pen's last column, +and no rule covers a break ahead of a pen's first live sample. A subscription restarts on every +`SetPens` (`Switch`), so a break that each subscription wrote ahead of its first sample would land past +the pen's last live column after every pen-set change, and the chart would draw it: +`EnvelopeLine.AppendColumn` rejects a column at or before the last one only when columns exist, and +`EnvelopeLine.ReplaceColumns` keeps the appended tail past the history end. + +`PostgresLiveSource` is the one implementation, and it polls the archive. Each subscription runs a poll +loop of its own on the injected data scheduler, at the operator's `poll_interval_ms`, holding a baseline +of its own and carrying the variable list in every query (`RT-1`). Disposing the subscription cancels +the loop's query and its wait. Batching, the union timeline and the hand-off to the UI scheduler happen +above the live source, in `TrendCoordinator` (`charting.md`). The poll's log lines carry the source +context `SemiPlot.DataSource.Postgres.PostgresLiveSource`; the history reads keep +`PostgresDataProvider`'s. `RealtimePollTests` pins the source context and the wait of the configured +`poll_interval_ms` after a tick. The first tick reads the baseline and emits nothing; every later tick reads the rows past `lastSeen`, converts them to UTC and emits them. `lastSeen` is the archive's own naive clock, not @@ -963,7 +1013,7 @@ clock by passing a `TimeProvider` to `ArchiveProviderFactory.Build`, which regis The pen set can change while the viewer runs. `TrendCoordinator` holds it in a `BehaviorSubject>` seeded from its constructor's pens and subscribes through -`Select(_dataProvider.Subscribe)` and `Switch()`, ahead of the `Buffer`. `SetPens` pushes the list it +`Select(_liveSource.Subscribe)` and `Switch()`, ahead of the `Buffer`. `SetPens` pushes the list it is handed, which the chart builds fresh for each call, and the chart calls it only when the id set it shows differs from `PenIds`, the subject's current set, so the coordinator carries no change check of its own. @@ -972,10 +1022,11 @@ the window, sees no change. A switch disposes the old subscription and starts a tick reads the baseline and emits nothing. The rows written between the old poll's last tick and the new baseline, at most one poll interval, are not delivered live; the history query the chart issues on every set change covers them. Every set the chart pushes holds only pens of the provider's own -catalogue. `Dispose` completes the subject. +catalogue, and the live source drops any identifier it does not serve. `Dispose` completes the subject. + +`PostgresLiveSource` keeps the contract, and the poll adds rules of its own: -- **The sequence never completes and never faults.** A query error logs, drops that tick's rows and - leaves the observable running. +- **A query error logs, drops that tick's rows and leaves the observable running.** - **No timestamp is emitted at or before the last one already delivered** (`DA-7`). - **A row whose `v` is null is dropped, and `lastSeen` still advances past it.** The history path flags such a bucket as a gap instead ("Quality and gaps"), so a null run the poll reads draws as a held line @@ -1245,7 +1296,7 @@ exception observer under it. projection runs inside `Select`, and Rx turns a throwing projection into `OnError`, which would end the `Publish().RefCount()` stream for every subscriber. `TryBuildRealtimeBatch` catches instead, skips the window and publishes the exception on `TrendCoordinator.RealtimeFailures`. A terminal -failure out of the provider takes the same channel: a `Catch` ahead of the `Publish` turns it into +failure out of the live source takes the same channel: a `Catch` ahead of the `Publish` turns it into one `RealtimeFailures` message and completes the batch stream, so `RealtimeBatches` never faults and the one failure is not counted once per subscriber of it. `TrendChartViewModel` is the only reader of that channel, and it reports through the guarded `ResultReporting.TryReportFailure` because an escape diff --git a/docs/architecture/overview.md b/docs/architecture/overview.md index bd65db4..2988c25 100644 --- a/docs/architecture/overview.md +++ b/docs/architecture/overview.md @@ -66,8 +66,8 @@ Constraint: **$0 budget** — only free/OSS components. | StartupFailureWindow (+ view model): shown instead of | | MainWindow when the start fails | +----------------------────────────────────────--------------+ - │ IDataProvider (subscribe realtime, query history, - │ re-read the catalogue) + │ IDataProvider (query history, re-read the catalogue) + │ ILiveSource (the live edge, via TrendCoordinator) │ IPenCatalogueEditor (the pen editor's writes) │ IArchiveRowReader (the export window's row reads) │ IMarkerStore (the chart's time markers, via MarkerSync) @@ -76,6 +76,7 @@ Constraint: **$0 budget** — only free/OSS components. | SemiPlot.Core | | - IDataProvider abstraction + records (Pen, envelope, | | ArchiveExtent, …) | +| - ILiveSource + Sample, the live edge | | - IPenCatalogueEditor + records, the catalogue writes | | - IArchiveRowReader + StoredArchiveRow, the export reads | | - IMarkerStore + StoredMarker, the time markers | @@ -87,21 +88,22 @@ Constraint: **$0 budget** — only free/OSS components. +-------------------------------------------------------------+ | SemiPlot.DataSource.Postgres (the only one) | | - PostgresDataProvider over the Simple-Scada archive | -| - history, extent, catalogue and the live-edge poll | +| - history, extent, days and catalogue reads | +| - PostgresLiveSource, the live-edge poll | | - PostgresPenCatalogueEditor, the catalogue writes | | - PostgresArchiveRowReader, the export's row reads | | - PostgresMarkerStore, the time markers | +-------------------------------------------------------------+ ``` -The UI never talks to a data source directly. It reads through `IDataProvider`; the pen editor writes -through `IPenCatalogueEditor`, and `Markers/MarkerSync` reads and writes the chart's time markers through -`IMarkerStore` (see [data-integration.md](./data-integration.md)). The export window reads stored rows -through `IArchiveRowReader`. The composition root resolves the PostgreSQL -provider, which is the only one the application ships; an archive that does not answer shows the -startup failure in its own window rather than falling back to invented data. There is **no web -bridge**: the chart is a native ScottPlot control, fed in-process by `TrendCoordinator` over -`IObservable`/awaitable seams. +The UI never talks to a data source directly. It reads through `IDataProvider`, and `TrendCoordinator` +follows the live edge through `ILiveSource`; the pen editor writes through `IPenCatalogueEditor`, and +`Markers/MarkerSync` reads and writes the chart's time markers through `IMarkerStore` (see +[data-integration.md](./data-integration.md)). The export window reads stored rows through +`IArchiveRowReader`. The composition root resolves the PostgreSQL provider, which is the only one the +application ships; an archive that does not answer shows the startup failure in its own window rather +than falling back to invented data. There is **no web bridge**: the chart is a native ScottPlot control, +fed in-process by `TrendCoordinator` over `IObservable`/awaitable seams. ### One window per process @@ -122,7 +124,7 @@ sequence's `StartupData`, the process's `InstanceLauncher` (which names the conf UI scheduler, the `CsvDialect` that `App.ConfigureStarted` picks through `App.CsvDialectFor` and the archive's time zone (`PostgresConnectionSettings.SourceTimeZone`), and constructs in order: -1. the coordinator; +1. the coordinator, over the container's `IDataProvider` and `ILiveSource`; 2. the marker sync, over the container's `IMarkerStore` (`#the-live-markers`); 3. the chart, whose constructor opens the live edge, seeded from the start sequence's pens and extent, with the marker sync as its marker channel; @@ -155,9 +157,9 @@ on the UI thread. The container keeps the process services. `AddPostgresData()` registers the data scheduler, `TimeProvider.System`, the `PostgresConnectionSettings`, the `NpgsqlDataSource`, the `ArchiveTimeConverter`, -the `ArchiveExceptionMapper`, the provider, the editor, the marker store and the row reader; `AddUi()` -registers the message panel, and `StartupProbe.BuildArchiveServiceProvider` adds the logger factory. The -container holds no window part, the status bar included. The UI scheduler is not a +the `ArchiveExceptionMapper`, the provider, the live source, the editor, the marker store and the row +reader; `AddUi()` registers the message panel, and `StartupProbe.BuildArchiveServiceProvider` adds the +logger factory. The container holds no window part, the status bar included. The UI scheduler is not a container registration; `App` passes `AvaloniaScheduler.Instance` to `TrendWindow.Build`, while the data scheduler comes from the container. @@ -330,12 +332,13 @@ one method: nothing else may resolve a service through a static. ## Data flow -- **Realtime:** the provider polls the raw layer for the samples written past the last one it saw → - `TrendCoordinator` buffers them on the data scheduler into a coalesced `RealtimeBatch` - (≤ 10 Hz / 100 ms), crosses to the UI scheduler via `ObserveOn`, and exposes them as - `IObservable`; the chart view model subscribes and appends to the per-pen - plottables. The provider reports no connection state: the catalogue sync derives it from its reads, - and the status bar shows it and the message panel records it (`#a-failed-read`). +- **Realtime:** `PostgresLiveSource`, behind `ILiveSource`, polls the raw layer for the samples + written past the last one it saw → `TrendCoordinator` buffers them on the data scheduler into a + coalesced `RealtimeBatch` (≤ 10 Hz / 100 ms), crosses to the UI scheduler via `ObserveOn`, and + exposes them as `IObservable`; the chart view model subscribes and appends to the + per-pen plottables. Neither the provider nor the live source reports a connection state: the + catalogue sync derives it from its reads, and the status bar shows it and the message panel records + it (`#a-failed-read`). - **History:** the chart requests a window → `TrendCoordinator.QueryHistoryAsync` (the single history query, reached through the debouncer by the initial load and every gesture alike) → provider returns one server-reduced `PenHistoryEnvelope` per pen (ascending `Timestamps` + `Min`/`Max`/`Center`) → the view @@ -903,7 +906,8 @@ the startup failure and the fatal catch alike — so a launcher can tell one fro The application reads the real archive and nothing else: the composition root registers `AddPostgresData`, and every member of `IDataProvider` is implemented over it — the pen catalogue, -the archive extent, the archive days read, the windowed history read and the live-edge poll. The chart draws history and +the archive extent, the archive days read and the windowed history read. `PostgresLiveSource` +implements `ILiveSource` with the live-edge poll over the same archive. The chart draws history and follows the archive as it grows. The pen editor writes the catalogue through `IPenCatalogueEditor`, and every running chart follows what it writes within 5 s (`#the-live-catalogue`), except the start values, which change nothing on screen (`#what-a-read-changes`). The chart places, drags, renames and diff --git a/docs/architecture/postgres-topology.md b/docs/architecture/postgres-topology.md index f040bc4..7f44e3d 100644 --- a/docs/architecture/postgres-topology.md +++ b/docs/architecture/postgres-topology.md @@ -105,6 +105,8 @@ flowchart TB coord["TrendCoordinator"] iface{{"IDataProvider"}} pg["PostgresDataProvider
the only implementation"] + live{{"ILiveSource"}} + poll["PostgresLiveSource
the live-edge poll"] subgraph plumbing["SemiPlot.DataSource.Postgres"] direction TB @@ -118,17 +120,23 @@ flowchart TB errs["SemiPlot.Core/Data/Errors
two sealed errors, two enums"] ui --> coord --> iface + coord --> live iface --> pg + live --> poll pg --> stmts pg --> ds pg --> conv pg --> mapper + poll --> stmts + poll --> ds + poll --> conv + poll --> mapper loader --> ds mapper --> errs ``` -The composition root resolves `PostgresDataProvider`, and the pen editor's -`PostgresPenCatalogueEditor`, the chart's `PostgresMarkerStore` and the export window's +The composition root resolves `PostgresDataProvider`, `PostgresLiveSource` for the live-edge poll, and +the pen editor's `PostgresPenCatalogueEditor`, the chart's `PostgresMarkerStore` and the export window's `PostgresArchiveRowReader` over the same `NpgsqlDataSource`; there is no fallback to resolve: an archive that does not answer opens the startup-failure window rather than @@ -137,7 +145,8 @@ falling back to invented data, which `data-integration.md` states under **Startu per type, Core's and the UI-local ones, and inside the `ArchiveError` arm one per `ArchiveFault` member, twelve. Which kind gets which severity, and why, is in `data-integration.md#two-error-planes`. Every member of `IDataProvider` is implemented: the pen -catalogue, the archive extent, the archive days, the windowed history read and the live-edge poll. +catalogue, the archive extent, the archive days and the windowed history read. `PostgresLiveSource` +implements `ILiveSource` with the live-edge poll. ## The three provisioning states, and the fourth diff --git a/docs/architecture/testing-strategy.md b/docs/architecture/testing-strategy.md index 1214864..acd1b2c 100644 --- a/docs/architecture/testing-strategy.md +++ b/docs/architecture/testing-strategy.md @@ -170,10 +170,12 @@ layer: `ArchiveStatementTextTests` pins it, and `PostgresHistoryReadTests.AWindowWiderThanTheLookBackFloorSeeksBackAsFarAsItAsks` runs it. **Against a socket, with no server** — `RealtimePollTests` (a failed tick and its logged run of failures, -every tick connecting to `127.0.0.1:1`), `CatalogueReadUnreachableTests` (the catalogue read over a refused -socket answers `Unreachable` naming the archive), `RealtimeSubscriptionDisposalTests` and -`ConvergeReadinessTests` (a loopback listener). They cross the machine boundary and need no container, so they carry no collection and take no -`PostgresContainerFixture`; a missing container runtime does not fail them. `UnreachableArchive.Settings` +every tick connecting to `127.0.0.1:1`; a subscription through `ILiveSource` that only fails emits neither +`OnCompleted` nor `OnError`, logs under `PostgresLiveSource` and waits the configured poll interval after +a tick), `CatalogueReadUnreachableTests` (the catalogue read over a refused socket answers `Unreachable` +naming the archive), `RealtimeSubscriptionDisposalTests` and `ConvergeReadinessTests` (a loopback +listener). They cross the machine boundary and need no container, so they carry no collection and take +no `PostgresContainerFixture`; a missing container runtime does not fail them. `UnreachableArchive.Settings` gives them the unreachable address and an adjustable poll interval. **Against a real Avalonia** — `SemiPlot.Tests.Unit/UI/`, under `[AvaloniaFact]` and `[AvaloniaTheory]`, @@ -223,7 +225,7 @@ archive, each closing a chain whose halves are already proved apart: | Test | Chain it closes | The seam tests it joins | | --- | --- | --- | | `BreakRenderArchiveJourneyTests` | a seeded break reaches the canvas: archive → `AddPostgresData` → `TrendCoordinator` → `TrendChartViewModel` → the pixels the rasteriser leaves blank | `PostgresHistoryReadTests` counts the fold's NaN anchor; `ChartGapRenderTests` measures the blank pixels a NaN column leaves | -| `LiveEdgeArchiveJourneyTests` | three: a row appended while the application runs reaches the chart's live edge and reaches it once; rows on a variable of its own reach the chart without breaking a pen that has no sample at that timestamp; and a `q = 32` row written while the application runs reaches the chart as its value and then a `NaN` column 1 ms past it, while `CurrentValue` keeps the value | `RealtimeSubscriptionTests` asserts the first rule over the provider alone; `TrendCoordinatorTests` and `TrendChartViewModelTests` cover the per-variable batch shape above it; `RealtimePollReadTests` asserts the value-then-break pair, and `ChartRealtimeApplierTests` and `TrendPenStateTests` cover the append-never-fold rule and the 1 ms step | +| `LiveEdgeArchiveJourneyTests` | three: a row appended while the application runs reaches the chart's live edge and reaches it once; rows on a variable of its own reach the chart without breaking a pen that has no sample at that timestamp; and a `q = 32` row written while the application runs reaches the chart as its value and then a `NaN` column 1 ms past it, while `CurrentValue` keeps the value | `RealtimeSubscriptionTests` asserts the first rule over the live source alone; `TrendCoordinatorTests` and `TrendChartViewModelTests` cover the per-variable batch shape above it; `RealtimePollReadTests` asserts the value-then-break pair, and `ChartRealtimeApplierTests` and `TrendPenStateTests` cover the append-never-fold rule and the 1 ms step | The composed path adds what those seam tests cannot see: the coordinator's buffering and folding, the chart view model's applier, the navigation controller. Each can lose a sample or replay one without diff --git a/docs/architecture/trend-interaction.md b/docs/architecture/trend-interaction.md index 2334aa4..aefadce 100644 --- a/docs/architecture/trend-interaction.md +++ b/docs/architecture/trend-interaction.md @@ -58,8 +58,8 @@ operator interaction. `ui-theme.md` is the record of what it does and how far it reaches. - **Scheduler seam:** Core keeps the bare `IScheduler` (`DefaultScheduler.Instance`) for data timing; the UI scheduler (`AvaloniaScheduler.Instance`) is captured in `AfterSetup` and passed explicitly to - the coordinator — `TrendCoordinator(IDataProvider dataProvider, IReadOnlyList pens, - IScheduler dataScheduler, IScheduler uiScheduler)`, + the coordinator — `TrendCoordinator(IDataProvider dataProvider, ILiveSource liveSource, + IReadOnlyList pens, IScheduler dataScheduler, IScheduler uiScheduler)`, `Buffer` (the 100 ms `TrendCoordinator.BatchWindow`) on the data scheduler, `ObserveOn` on the UI one. No second `IScheduler` container registration. The pen catalogue is passed in because the coordinator needs the pen identifiers in its constructor and `QueryPensAsync` cannot be awaited there; the composition root hands over the start sequence's @@ -69,17 +69,19 @@ operator interaction. (see Renderer). *As-built note:* realtime is `double?` on each pen's own timestamps, not on a shared column: `PenRealtimeValues` carries no entry at a timestamp the pen did not report, and a null is a break the - provider reported. The poll writes one 1 ms after a `q = 32` row (data-integration.md#realtime); every + live source reported. The poll writes one 1 ms after a `q = 32` row (data-integration.md#realtime); every other gap the chart draws comes from the history path (charting.md, "Gaps via NaN"). - **Δ cursors:** Δy is reported only for the **active/selected pen** (pens share X but have independent Y scales, so a global Δy is meaningless). - **Bad quality → gap:** OPC bad-quality is mapped to `null` at the `IDataProvider` boundary, reusing the null = gap path; a distinct value-present-but-bad-quality flag is deferred. - *As-built note:* no OPC quality exists yet, so no bad-quality status maps to `null`. The archive marks - the last sample before a break with `q = 32`, and that mark maps to a break. On the history path the - server closes the bucket there, and `BucketedRowFold` writes a `NaN` column one tick after it - (data-integration.md#quality-and-gaps). At the live edge the poll emits the row's value and then a null - `Sample` 1 ms after it, which the chart appends as a `NaN` column (data-integration.md#realtime). + *As-built note:* no OPC quality exists yet, so no bad-quality status maps to `null`. The live edge + crosses `ILiveSource`, not `IDataProvider`, so that port is the boundary an OPC status maps at. The + archive marks the last sample before a break with `q = 32`, and that mark maps to a break. On the + history path the server closes the bucket there, and `BucketedRowFold` writes a `NaN` column one tick + after it (data-integration.md#quality-and-gaps). At the live edge the poll emits the row's value and + then a null `Sample` 1 ms after it, which the chart appends as a `NaN` column + (data-integration.md#realtime). `q = 16` takes no branch: the next bucket or sample resumes the line. - **Real-time return:** "jump to real-time" re-attaches sticky immediately; now-marker at the **right edge** (centering is at most a transient transition animation). diff --git a/docs/plans/completed/20261020-live-source-port.md b/docs/plans/completed/20261020-live-source-port.md new file mode 100644 index 0000000..afec28d --- /dev/null +++ b/docs/plans/completed/20261020-live-source-port.md @@ -0,0 +1,529 @@ +# Live-source port + +## Overview + +- The live edge reads through a port of its own, `ILiveSource` in `SemiPlot.Core/Data`, instead of + through `IDataProvider.Subscribe`. The port has one member: a cold `IObservable>` + of samples and breaks for a pen set. Its contract states that the stream never completes and never + faults. `TrendCoordinator` turns an `OnError` into one `RealtimeFailures` entry and a live edge that stays + empty for the rest of the window's life: the `Catch` swaps in `Observable.Empty`, which completes + `RealtimeBatches` and drops the subscription to the pen set, and nothing rebuilds it + (`TrendCoordinator.cs:38,42-44,97-115`). An `OnCompleted` is not caught at all: `Switch` idles on a + completed inner stream, so the live edge goes silent with no entry until the next `SetPens`. +- The archive poll becomes the port's first implementation, `PostgresLiveSource` in + `SemiPlot.DataSource.Postgres`. `RealtimePoll` and `ArchiveStatements` do not change, and neither does + the poll's cadence: one tick, then a wait of `poll_interval_ms` on the data scheduler. +- `IDataProvider` loses `Subscribe` and keeps the history, extent, days and catalogue reads. + `TrendCoordinator` takes both interfaces; `TrendWindow.Build` resolves both from the container. +- Nothing the operator sees changes. The only output that differs is the source context of the poll's + own log lines, which moves from `PostgresDataProvider` to `PostgresLiveSource` (Technical Details). +- This prepares the roadmap's `opc-live-source` slice: an `OpcUaLiveSource` in a new + `SemiPlot.DataSource.OpcUa` project implements the same port, and `PostgresLiveSource` is deleted with + the poll. + +## Context (from discovery) + +- Roadmap: docs/plans/roadmaps/20261019-opc-live-edge-roadmap.md — slice live-source-port +- Verified at `85e91eb` (the tip of `archive-state-from-catalogue`, PR #142); `master` at `54f455e` carries + the same tree (`git diff 85e91eb 54f455e` prints nothing). Line numbers below are at `85e91eb`. Paths are + under `SemiPlot/` unless they start with `docs/` or name `AGENTS.md`. +- The member that moves: `IObservable> Subscribe(IReadOnlyList penIds)` with its + "Cold per call" comment (`SemiPlot.Core/Data/IDataProvider.cs:9-10`). `IDataProvider` holds nothing else + of the live edge (`:12-28`). +- `Sample(int PenId, DateTime TimestampUtc, double? Value)`, a null `Value` is a break + (`SemiPlot.Core/Trends/Sample.cs:3-4`). +- The one production implementation: `PostgresDataProvider.Subscribe` wraps `PollAsync` in + `Observable.Create` over `_scheduler.ScheduleAsync` (`SemiPlot.DataSource.Postgres/PostgresDataProvider.cs:76-87`). + `PollAsync` builds one `RealtimePoll` per subscription, emits non-empty tick results, sleeps + `_settings.PollInterval` on `_scheduler`, and ends silently on cancellation (`:340-377`). It never calls + `OnError` or `OnCompleted`. `_settings` and `_scheduler` (`:45-46,69-70`) have no other reader + (`git grep -n -E "_settings|_scheduler"` on the file prints `:45,46,69,70,86,370`). The class summary + names `Subscribe` in a `cref` (`:21-25`), and the `Subscribe` summary names + `PostgresConnectionSettings.PollInterval` in a `cref` (`:76-80`). `System.Reactive.Concurrency`, + `System.Reactive.Linq` (`:4-5`) and `SemiPlot.DataSource.Postgres.Configuration` (`:17`) serve only the + live edge: `PostgresConnectionSettings` appears only at `:45,60,78`. +- `RealtimePoll(dataSource, timeConverter, exceptionMapper, penIds, ILogger logger)` + (`SemiPlot.DataSource.Postgres/RealtimePoll.cs:15-20`). `ReadOnceAsync` catches every failure but + `OperationCanceledException`, logs it with `ConsecutiveFailures` and returns no sample (`:43-70,198-209`). +- The registration: `AddPostgresData` registers `IDataProvider` by factory with `settings` and `IScheduler` + among its arguments (`SemiPlot.DataSource.Postgres/PostgresDataServiceCollectionExtensions.cs:37-45`), and + the other ports beside it (`:46-57`). The bare `IScheduler` is `DefaultScheduler.Instance` (`:29`). Every + sibling class registered there is public with an explicit `internal` constructor + (`PostgresDataProvider.cs:56-74`, `PostgresMarkerStore.cs:20-28`), and `ArchiveExceptionMapper` is + `internal` (`ArchiveExceptionMapper.cs:17`). +- The one consumer: `TrendCoordinator` holds `_dataProvider` (`SemiPlot.UI/Bridge/TrendCoordinator.cs:16`), + takes it first in its constructor (`:28-34`), and subscribes through + `_penIds.Select(_dataProvider.Subscribe).Switch()` (`:97-99`). A terminal failure is caught after + `Switch` and ahead of `Publish`, reported once on `RealtimeFailures`, and ends the batch stream + (`:104,110-115`). The comment at `:25-27` says "a provider silently drops" unknown ids; the summaries at + `:42-52` speak of "a provider that ends the stream". The history, extent and days pass-throughs stay on + `_dataProvider` (`:71-93`). +- `RealtimeFailures` has one reader, `TrendChartViewModel` (`SemiPlot.UI/Chart/TrendChartViewModel.cs:83`). +- The window: `TrendWindow.Build` resolves `IDataProvider` (`SemiPlot.UI/MainWindow/TrendWindow.cs:72`) and + builds the coordinator over it (`:80`); `PenCatalogueSync` takes the same provider (`:100-105`). +- `new TrendCoordinator(` sites: `TrendWindow.cs:80`; `SemiPlot.Tests.Unit/UI/Bridge/TrendCoordinatorTests.cs:97,121,214`; + `SemiPlot.Tests.Unit/UI/Chart/ChartTestBuilder.cs:60`; `SemiPlot.Tests.Unit/UI/Minimap/MinimapStand.cs:37`; + `SemiPlot.Tests.Integration/Journeys/LiveEdgeArchiveJourneyTests.cs:179-180`; + `SemiPlot.Tests.Integration/Journeys/BreakRenderArchiveJourneyTests.cs:70-71`. +- Direct callers of `IDataProvider.Subscribe` in tests: + `SemiPlot.Tests.Integration/RealtimeEmptyArchiveTests.cs:76-78`, + `SemiPlot.Tests.Integration/RealtimeSubscriptionDisposalTests.cs:31-38`, + `SemiPlot.Tests.Integration/RealtimeSubscriptionTests.cs:44-46,85-97`. Each resolves `IDataProvider` from + a container that `ArchiveProviderFactory.Build` or `UnreachableArchive.Services` composes with + `AddPostgresData` (`SemiPlot.Tests.Integration/ArchiveProviderFactory.cs:29-60`, + `SemiPlot.Tests.Integration/UnreachableArchive.cs:26-34`). The unit project calls it nowhere. +- `SemiPlot.AppHost` references `SemiPlot.UI` (`SemiPlot.AppHost/SemiPlot.AppHost.csproj:13`). +- The test fake: `FakeDataProvider : IDataProvider` (`SemiPlot.Tests.Unit/UI/Bridge/FakeDataProvider.cs:16-19`). + Its live half is `OpenLiveSubscriptionCount`, `LiveSubscriptionsOpened`, `LiveSubscriptionChanging` + (`:141-146`), `StaggerRealtimeTimestamps`, `PoisonRealtimeWindow`, `RealtimeStreamFailure` (`:153-161`), + `BreakRealtimePenOnce`, `TheBreakIn` (`:164-190`) and `Subscribe` with its helpers (`:192-246`). The tick + filters the subscribed ids by the fake's own `Pens` (`:217`), and + `TrendCoordinatorTests.SetPens_WithAnAddedId_ReachesTheOneSubscriberOfTheLiveEdge` grows `provider.Pens` + and expects the live edge to carry the new pen (`TrendCoordinatorTests.cs:165-170`). Six test files outside + the fake read its live members. +- Test containers that `TrendWindow.Build` resolves from: `MainWindowTestBuilder.NewArchiveStand` + creates its own `TestScheduler` and registers the fake as `IDataProvider` + (`SemiPlot.Tests.Unit/UI/MainWindow/MainWindowTestBuilder.cs:97-130`); `TrendWindowTests`, + `TrendWindowBuildTests`, `AppMainWindowTests`, `AppConfigurationTests` and `EmptyCatalogueStartupTests` + build through it. `StartupProbeTests` builds containers of its own (`StartupProbeTests.cs:149-151,208-213`) + and never builds a window. +- Composition tests: `PostgresCompositionTests` resolves `IDataProvider` and lists the singleton + registrations (`SemiPlot.Tests.Unit/Postgres/PostgresCompositionTests.cs:44-52,83-100`); + `CompositionRootTests` resolves each port from the container `StartupProbe` builds + (`SemiPlot.Tests.Unit/UI/Di/CompositionRootTests.cs:27-49`). +- The socket class for the poll: `RealtimePollTests`, no collection, no fixture, against + `UnreachableArchive` (`SemiPlot.Tests.Integration/RealtimePollTests.cs:13-23`). `ArmedGate` counts the + highest `ConsecutiveFailures` across every logger of a container and awaits it + (`SemiPlot.Tests.Integration/ArmedGate.cs:9-13,53-68`). +- Log category: the log template prints `{SourceContext}` (`SemiPlot.UI/Program.cs:100-102`). No test + filters the poll's lines by category: `ArmedGate` reads every logger (`ArmedGate.cs:70-73`), + `RealtimeSubscriptionTests.PollEntries` filters on `RowCount` only (`RealtimeSubscriptionTests.cs:120-123`), + and `PostgresHistoryReadTests.HistoryReadEntries` filters on the provider's category and the `Outcome` + key, which no poll line carries (`PostgresHistoryReadTests.cs:1041-1048`). +- Docs that state the old seam: `docs/architecture/data-integration.md:28-62` (the surface and its code + block), `:941-947,963-978` (Realtime), `:1244-1250` (the terminal-failure guard); + `docs/architecture/overview.md:69-73,77,88-90,97-104,333-338,905-906`; + `docs/architecture/charting.md:773-775,789-793`; `docs/architecture/postgres-topology.md:104-107,120-121,130-132,139-140`; + `docs/architecture/trend-interaction.md:61-62`; `docs/architecture/testing-strategy.md:226`; + `AGENTS.md:396-403`. +- The later slices this port must serve (roadmap `:211-251`, `:270-288`): `OpcUaLiveSource` opens one + session per process, writes breaks through `Sample`, reports OPC failures to the message panel through + `ArchiveFailureMapper` with a new error type in `SemiPlot.Core/Data/Errors`, and drives a second + status-bar indicator; `live-hold-to-edge` needs to know whether the OPC session is up. + +## Development Approach + +- **testing approach**: Regular (code first, then tests in the same task). +- Complete each task fully before moving to the next. Each task leaves `dotnet build SemiPlot.slnx` green. +- Every task includes new or updated tests for the code it changes. +- All tests pass before the next task starts. +- Update this plan file when scope changes during implementation. +- `dotnet build SemiPlot.slnx` holds at 0 errors and no compiler or analyzer warning (`TreatWarningsAsErrors`, + which also makes IDE0005 and CS1574 errors). The NuGet audit warnings `NU1900`-`NU1904` are outside the + gate: `SemiPlot/Directory.Build.props:14` keeps them warnings, and `85e91eb` already prints two `NU1902` for + `SemiPlot.AppHost` (MessagePack 2.5.302). `dotnet format SemiPlot.slnx --verify-no-changes` and `dotnet terse` pass over the touched + `.cs` files. A new `.cs` file gets one `dotnet format SemiPlot.slnx --include ` (`AGENTS.md`, Build). + +## Testing Strategy + +- Unit tests (`SemiPlot.Tests.Unit`, no container): + - `PostgresCompositionTests`: `AddPostgresData` resolves `ILiveSource` as `PostgresLiveSource` and + registers it once, as a singleton. + - `CompositionRootTests`: the container `StartupProbe` builds resolves `ILiveSource` as + `PostgresLiveSource`. + - `TrendCoordinatorTests`: with two distinct fakes, the live edge subscribes the live source and the + history and extent reads go to the provider; every existing case passes with only its construction + changed. + - `TrendWindowBuildTests`: the window's live edge opens on the container's `ILiveSource`, not on the + `IDataProvider` instance. + - Every other suite that builds a coordinator or a window passes unchanged in its assertions. +- Integration tests (`SemiPlot.Tests.Integration`): + - `RealtimePollTests` (socket, no server): a subscription through `ILiveSource` over an unreachable + archive emits no notification of any kind, `OnCompleted` and `OnError` included, through five failed + ticks. + - `RealtimeEmptyArchiveTests`, `RealtimeSubscriptionTests`, `RealtimeSubscriptionDisposalTests`: the same + assertions, against the resolved `ILiveSource`. + - `RealtimePollReadTests`, `ArchiveStatementTextTests` (unit project) and both journeys pass with no + assertion edited: the scope-guard proof that the poll's statements and samples are unchanged. + - The whole project runs in Task 4, which changes the `PostgresDataProvider` factory every collection + class resolves. +- Every new test carries the class's `Component`, `Area` and `Category` traits. AwesomeAssertions throughout. + +## Acceptance Evidence + +This slice fixes no defect; it moves one seam. The evidence is that the guard fires at `85e91eb`, that +every consumer moved, and that the poll behaves as before. + +- Guard at `85e91eb`, run as Task 1's first step: delete `IDataProvider.Subscribe` (`IDataProvider.cs:9-10`) + and run `dotnet build SemiPlot.slnx`. Hypothesis: + - the build fails with one error, CS1061 at `SemiPlot.UI/Bridge/TrendCoordinator.cs:98`, and + `SemiPlot.Tests.Unit`, `SemiPlot.Tests.Integration` and `SemiPlot.AppHost` do not compile because they + reference `SemiPlot.UI`; + - with `:98` patched in the same probe to `.Select(_ => Observable.Never>())`, the + build fails with CS0411 at `RealtimeEmptyArchiveTests.cs:78`, `RealtimeSubscriptionDisposalTests.cs:34,38` + and `RealtimeSubscriptionTests.cs:46,87,97`: without the interface member, overload resolution binds + `provider.Subscribe(ids)` to `System.ObservableExtensions.Subscribe(IObservable, Action)` and + type inference fails; + - `FakeDataProvider` and `PostgresDataProvider` keep compiling: a public `Subscribe` method needs no + interface. + Record the error list in Task 1; restore both files. +- What the compiler does not catch, and the checks that do: + - a cast from `IDataProvider` to `ILiveSource` in `TrendWindow.Build`: the new + `TrendWindowBuildTests` case registers a second fake as `ILiveSource`, so a cast subscribes the wrong + instance and the case fails; + - a forgotten registration in a test container: `GetRequiredService` throws in every + `NewWindowStand` test. +- After the change these commands pass: + +```powershell +dotnet build SemiPlot.slnx +dotnet test SemiPlot/SemiPlot.Tests.Unit/SemiPlot.Tests.Unit.csproj +dotnet test SemiPlot/SemiPlot.Tests.Integration/SemiPlot.Tests.Integration.csproj +``` + +- These print exactly what is stated: + - `git grep -n "Subscribe(IReadOnlyList" -- SemiPlot` prints `ILiveSource.cs`, `PostgresLiveSource.cs` + and `FakeDataProvider.cs` lines only; + - `git grep -n "ILiveSource" -- SemiPlot/SemiPlot.UI` prints `Bridge/TrendCoordinator.cs` and + `MainWindow/TrendWindow.cs` lines only; + - `git grep -n -E "[^f]\(ILiveSource\)|as ILiveSource" -- SemiPlot` prints nothing (the `[^f]` keeps + `typeof(ILiveSource)`, which is no cast, out of the match); + - `git diff --stat master -- SemiPlot/SemiPlot.DataSource.Postgres/RealtimePoll.cs SemiPlot/SemiPlot.DataSource.Postgres/ArchiveStatements.cs` + prints nothing (the statements and the tick are unchanged). +- Integration runs need a container runtime for the collection classes. ASSUMPTION: Docker is available on + the executing machine. The CI `linux` leg runs the integration project in any case. + +## Progress Tracking + +- Mark completed items with `[x]` immediately when done. +- Add newly discovered tasks with ➕ prefix. +- Document issues or blockers with ⚠️ prefix. +- Keep the plan in sync with the work done. + +## Solution Overview + +- **One port, one member.** `ILiveSource` carries `Subscribe(IReadOnlyList penIds)` and nothing else. + The name and the signature are the ones `IDataProvider` carries today, so every caller changes the object + it calls and not the call. The port is not `IDisposable`: the container disposes a singleton that owns a + resource whatever interface it is registered under, and the poll owns nothing past a subscription. It + carries no state channel: the archive state already comes from `PenCatalogueSync.ConnectionStates`, and + the OPC session state is the `opc-live-source` slice's to add, as a member of the port or a sibling + observable, once its consumers (the second indicator, the hold to the edge) exist. +- **The contract lives in `data-integration.md#realtime`.** The interface carries a one-line summary and a + pointer to it. The contract: cold per call, and disposal stops the work; never completes, never faults, + so a failure inside a source is the source's to log or report out of band and emits nothing; a break is + a `Sample` whose `Value` is null, for the moments the source itself observes; timestamps are UTC; an + identifier the source does not serve emits nothing and leaves the stream running. Ordering is not part of + the port: the chart appends a pen's sample only past that pen's last column (`DA-7`) and drops any other. +- **The coordinator keeps its guard.** `Catch` and `RealtimeFailures` stay. A source that faults still + reaches the panel once, though its live edge is gone for the window's life, and the three + `RealtimeStreamFailure` tests keep pinning that (`TrendCoordinatorTests.cs:119,201`, + `TrendChartViewModelTests.cs:2610`). The guard covers `OnError` only; a source that completes + leaves the live edge silent with no entry. Both are contract violations, and the documentation says what + each one does. +- **Which breaks travel through the port.** Every break a source can decide from its own input: a `q = 32` + row today, a session loss or a `Bad` status in `opc-live-source`. The port carries no chart state, so a + source cannot know a pen's last column. The roadmap's break ahead of a pen's first live sample is + therefore not settled here: a subscription restarts on every `SetPens` (`Switch`), so a per-subscription + ahead-break would land past the last live column after every pen-set change, and the chart's two filters + (`AppendAfterLastColumn` rejects at or before the last column only when columns exist, + `EnvelopeLine.cs:271-286`; `ReplaceColumns` keeps the appended tail past the history end, `:195-214`) would + draw it. `opc-live-source` designs that rule; this slice's documentation names it as open and states no + rule for it. +- **A class of its own for the poll.** `PostgresLiveSource` holds the `Observable.Create` and the loop that + `PostgresDataProvider` holds today, so `opc-live-source` deletes the poll implementation as one file plus + `RealtimePoll`, and `PostgresDataProvider` (626 lines) drops its two live-only fields. +- **One fake for both interfaces.** `FakeDataProvider` implements `IDataProvider` and `ILiveSource`. Its + live edge filters by its own `Pens`, which the catalogue tests grow at run time, so a separate live fake + would have to share that list with the provider fake. Tests that need the two apart build two + `FakeDataProvider` instances. + +## Technical Details + +- `SemiPlot.Core/Data/ILiveSource.cs`: + +```csharp +using SemiPlot.Core.Trends; + +namespace SemiPlot.Core.Data; + +/// The live edge of a pen set (docs/architecture/data-integration.md#realtime). +public interface ILiveSource +{ + IObservable> Subscribe(IReadOnlyList penIds); +} +``` + +- `SemiPlot.DataSource.Postgres/PostgresLiveSource.cs`: `public sealed class PostgresLiveSource : ILiveSource` + with an explicit `internal` constructor over `NpgsqlDataSource`, `ArchiveTimeConverter`, + `ArchiveExceptionMapper`, `IScheduler`, `ILogger` and `TimeSpan pollInterval`, in that + order (the configuration value last, as `PostgresDataProvider`'s `historyReadConcurrency`), assigning one + `_camelCase` `readonly` field each. A primary constructor takes no access modifier, + and a public one would expose the `internal` `ArchiveExceptionMapper` (CS0051); the sibling classes have + the same shape (Context). `Subscribe` and `PollAsync` move from `PostgresDataProvider.cs:76-87,340-377` + with their bodies unchanged except for the field names; the sleep is + `_scheduler.Sleep(_pollInterval, cancellationToken)`. The `Subscribe` summary says "the poll interval" + instead of the `PostgresConnectionSettings.PollInterval` `cref`, because the class takes a `TimeSpan`. The + registration is a factory, and the composition tests and the integration containers reach the class only + through `AddPostgresData`. +- `AddPostgresData` registers it beside the provider: + `services.AddSingleton(provider => new PostgresLiveSource(...))`, passing + `provider.GetRequiredService().PollInterval`. `PostgresDataProvider`'s factory + loses its `settings` and `IScheduler` arguments. +- `PostgresDataProvider` after the move: no `Subscribe`, no `PollAsync`, no `_settings` or `_scheduler` + fields or constructor parameters, no `System.Reactive.*` or `SemiPlot.DataSource.Postgres.Configuration` + usings; the class summary drops the live edge and its `cref`. +- `TrendCoordinator(IDataProvider dataProvider, ILiveSource liveSource, IReadOnlyList pens, IScheduler + dataScheduler, IScheduler uiScheduler)`; field `_liveSource` after `_dataProvider`; + `BuildRealtimeBatches` reads `_penIds.Select(_liveSource.Subscribe)`. The comment at `:25-27` says that + the live source drops the identifiers it does not serve. The summaries at `:42-52` say "a live source that + faults", because an `OnCompleted` is not reported (Overview). +- `TrendWindow.Build` resolves `var liveSource = serviceProvider.GetRequiredService();` beside + the provider and passes it to the coordinator. +- `MainWindowTestBuilder.NewArchiveStand` gains an optional `ILiveSource? liveSource = null` and registers + `liveSource ?? provider` as `ILiveSource`. Its one caller with a value is the new `TrendWindowBuildTests` + case. +- Log output: the poll's lines (`RealtimePoll.cs:116-122,178-185,204-208`) carry the source context + `SemiPlot.DataSource.Postgres.PostgresLiveSource`. Their text and level are unchanged. + +## What Goes Where + +- **Implementation Steps** (`[ ]` checkboxes): code, tests and documentation in this repository. +- **Post-Completion** (no checkboxes): the manual smoke on the demo stand and the remaining roadmap slices. + +## Implementation Steps + +### Task 1: Declare the port and serve it from the archive poll + +**Files:** +- Create: `SemiPlot/SemiPlot.Core/Data/ILiveSource.cs` +- Create: `SemiPlot/SemiPlot.DataSource.Postgres/PostgresLiveSource.cs` +- Modify: `SemiPlot/SemiPlot.DataSource.Postgres/PostgresDataServiceCollectionExtensions.cs` +- Modify: `SemiPlot/SemiPlot.Tests.Unit/Postgres/PostgresCompositionTests.cs` +- Modify: `SemiPlot/SemiPlot.Tests.Unit/UI/Di/CompositionRootTests.cs` +- Modify: `SemiPlot/SemiPlot.Tests.Integration/RealtimePollTests.cs` + +- [x] run the guard probe in Acceptance Evidence at `85e91eb` and record the error list here; restore + `IDataProvider.cs` and `TrendCoordinator.cs` + - Run on the `live-source-port` working tree at `54f455e` (same tree as `85e91eb`); both files restored + with `git checkout --`. The observed list matches the hypothesis exactly. + - `Subscribe` deleted: one error, `SemiPlot.UI/Bridge/TrendCoordinator.cs(98,26): error CS1061` + (`IDataProvider` has no `Subscribe`); nothing downstream of `SemiPlot.UI` compiled. + - `:98` patched to `Observable.Never`: six errors, all `CS0411` on + `ObservableExtensions.Subscribe(IObservable, Action)`: `RealtimeEmptyArchiveTests.cs(78,37)`, + `RealtimeSubscriptionDisposalTests.cs(34,80)`, `(38,31)`, `RealtimeSubscriptionTests.cs(46,79)`, + `(87,72)`, `(97,73)`. `FakeDataProvider` and `PostgresDataProvider` compiled. +- [x] create `ILiveSource` as in Technical Details +- [x] create `PostgresLiveSource` with the `Subscribe` and `PollAsync` bodies of + `PostgresDataProvider.cs:76-87,340-377`; `PostgresDataProvider` keeps its own copy until Task 4 + deletes it +- [x] register `ILiveSource` in `AddPostgresData` after `IDataProvider` +- [x] `PostgresCompositionTests`: add `AddPostgresDataResolvesThePostgresLiveSource` and + `[InlineData(typeof(ILiveSource))]` to `AddPostgresDataRegistersASingleton` +- [x] `CompositionRootTests`: add `Container_ResolvesTheLiveSource` (`PostgresLiveSource`) +- [x] `RealtimePollTests`: add `ASubscriptionThatOnlyFailsNeitherCompletesNorFaults`: an `ArmedGate` over + `UnreachableArchive.Services` at a 10 ms poll interval, a `BatchCollector` over + `ILiveSource.Subscribe([1]).Materialize()`, await `FailuresReached(TickCount)` within + `ClonedArchiveTest.GateBound`, then `Batches` is empty +- [x] run the unit project and + `dotnet test SemiPlot/SemiPlot.Tests.Integration/SemiPlot.Tests.Integration.csproj --filter "FullyQualifiedName~RealtimePollTests"`; + both pass before Task 2 (unit 2300/2300, `RealtimePollTests` 3/3) + +### Task 2: Point the poll's integration tests at the port + +**Files:** +- Modify: `SemiPlot/SemiPlot.Tests.Integration/RealtimeEmptyArchiveTests.cs` +- Modify: `SemiPlot/SemiPlot.Tests.Integration/RealtimeSubscriptionTests.cs` +- Modify: `SemiPlot/SemiPlot.Tests.Integration/RealtimeSubscriptionDisposalTests.cs` + +- [x] `RealtimeEmptyArchiveTests.ASubscriptionOverAnArchiveWithNoRowLogsItsBaseline` (`:70-81`) resolves + `ILiveSource` into a local named `liveSource` +- [x] `RealtimeSubscriptionTests` (`:44,85`) resolves `ILiveSource` into `liveSource` in both cases +- [x] `RealtimeSubscriptionDisposalTests` (`:31-32`) resolves `ILiveSource` for `dropped` and `running`; + the comments at `RealtimeSubscriptionDisposalTests.cs:19-20` and `RealtimeSubscriptionTests.cs:32` say + "live source" where they say "provider" +- [x] no assertion changes in the three classes +- [x] run `dotnet test SemiPlot/SemiPlot.Tests.Integration/SemiPlot.Tests.Integration.csproj --filter "FullyQualifiedName~Realtime"`; + it passes before Task 3 (17/17, 0 skipped) + +### Task 3: Move the coordinator and the window onto the port + +**Files:** +- Modify: `SemiPlot/SemiPlot.UI/Bridge/TrendCoordinator.cs` +- Modify: `SemiPlot/SemiPlot.UI/MainWindow/TrendWindow.cs` +- Modify: `SemiPlot/SemiPlot.Tests.Unit/UI/Bridge/FakeDataProvider.cs` +- Modify: `SemiPlot/SemiPlot.Tests.Unit/UI/Bridge/TrendCoordinatorTests.cs` +- Modify: `SemiPlot/SemiPlot.Tests.Unit/UI/Chart/ChartTestBuilder.cs` +- Modify: `SemiPlot/SemiPlot.Tests.Unit/UI/Minimap/MinimapStand.cs` +- Modify: `SemiPlot/SemiPlot.Tests.Unit/UI/MainWindow/MainWindowTestBuilder.cs` +- Modify: `SemiPlot/SemiPlot.Tests.Unit/UI/Di/TrendWindowBuildTests.cs` +- Modify: `SemiPlot/SemiPlot.Tests.Unit/UI/Chart/TrendChartViewModelTests.cs` +- Modify: `SemiPlot/SemiPlot.Tests.Integration/Journeys/LiveEdgeArchiveJourneyTests.cs` +- Modify: `SemiPlot/SemiPlot.Tests.Integration/Journeys/BreakRenderArchiveJourneyTests.cs` + +- [x] `TrendCoordinator` takes `ILiveSource` as in Technical Details; reword `:25-27` and `:42-52` +- [x] `TrendWindow.Build` resolves `ILiveSource` and passes it +- [x] `FakeDataProvider` implements `ILiveSource` beside `IDataProvider`; its comments at `:159-160` say + "live source" for the live edge +- [x] pass the fake as both arguments at `TrendCoordinatorTests.cs:97,121,214`, `ChartTestBuilder.cs:60` and + `MinimapStand.cs:37`; pass the container's `ILiveSource` at the two journey sites +- [x] `MainWindowTestBuilder.NewArchiveStand` takes `ILiveSource? liveSource = null` and registers + `liveSource ?? provider` +- [x] `TrendCoordinatorTests`: rename the three cases that say "provider" for the live edge (`:70,174,191`) + to say "live source", and say "live source" in the comment at `:110-111` and the exception texts at + `:116,200`; the same for `TrendChartViewModelTests.AProviderThatEndsTheLiveEdge_IsReportedOnce` + (`:2604`) and its exception text, and the comment at `LiveEdgeArchiveJourneyTests.cs:23` +- [x] `TrendCoordinatorTests`: add `TheLiveEdgeFollowsTheLiveSource_AndTheReadsGoToTheProvider`: two fakes + over one `TestScheduler`, one subscription to `RealtimeBatches`, one `QueryHistoryAsync` and one + `QueryArchiveExtentAsync`; the live fake opened one live subscription and saw no history or extent + query, the archive fake opened none and saw one of each +- [x] `TrendWindowBuildTests`: add `TheWindowFollowsTheContainersLiveSource`: `NewArchiveStand(liveSource: )`, build the window as + `TheWindowReadsTheCatalogueAtOnce_OnTheRealScheduler` does (`:70-88`); the second fake's + `LiveSubscriptionsOpened` is 1 and the stand provider's is 0 +- [x] run the unit project and + `dotnet test SemiPlot/SemiPlot.Tests.Integration/SemiPlot.Tests.Integration.csproj --filter "FullyQualifiedName~JourneyTests"`; + both pass before Task 4 (unit 2302/2302, journeys 4/4, 0 skipped) + +### Task 4: Delete `IDataProvider.Subscribe` and the provider's live half + +**Files:** +- Modify: `SemiPlot/SemiPlot.Core/Data/IDataProvider.cs` +- Modify: `SemiPlot/SemiPlot.DataSource.Postgres/PostgresDataProvider.cs` +- Modify: `SemiPlot/SemiPlot.DataSource.Postgres/PostgresDataServiceCollectionExtensions.cs` + +- [x] delete `IDataProvider.Subscribe` and its comment; `dotnet build SemiPlot.slnx` is green, because Tasks + 2 and 3 moved every consumer +- [x] delete `Subscribe`, `PollAsync`, `_settings`, `_scheduler`, their constructor parameters, the + `System.Reactive.*` and `SemiPlot.DataSource.Postgres.Configuration` usings and the live-edge `cref` + from `PostgresDataProvider`; drop the two arguments from its factory in `AddPostgresData` +- [x] run the three `git grep` checks of Acceptance Evidence + - `Subscribe(IReadOnlyList` prints `ILiveSource.cs:8`, `PostgresLiveSource.cs:44`, + `FakeDataProvider.cs:192` only; `ILiveSource` in `SemiPlot.UI` prints `Bridge/TrendCoordinator.cs:17,31` + and `MainWindow/TrendWindow.cs:73` only. + - ⚠️ the cast check prints one line, `PostgresCompositionTests.cs:95: [InlineData(typeof(ILiveSource))]` + (added in Task 1): the regex `\(ILiveSource\)` also matches `typeof(ILiveSource)`. It is no cast; + `git grep -n -E "[^f]\(ILiveSource\)|as ILiveSource" -- SemiPlot` prints nothing. + - the `git diff --stat master` over `RealtimePoll.cs` and `ArchiveStatements.cs` prints nothing. +- [x] `PostgresCompositionTests` and `CompositionRootTests` pass unchanged: they pin both registrations +- [x] run the unit project and the whole integration project + (`dotnet test SemiPlot/SemiPlot.Tests.Integration/SemiPlot.Tests.Integration.csproj`): every collection + class resolves the changed `PostgresDataProvider` factory; both pass before Task 5 + - unit: passed 2302, failed 0, skipped 0 (2302 total); integration: passed 197, failed 0, skipped 0 + (197 total), container runtime reached + +### Task 5: Verify acceptance criteria + +- [x] `dotnet build SemiPlot.slnx` at 0 errors and no warning beyond the two `NU1902` of `SemiPlot.AppHost` + - run fresh at `4940ea1`, also with `--no-incremental`: 0 errors, 2 warnings, both `NU1902` (MessagePack + 2.5.302) on `SemiPlot.AppHost` +- [x] the three Acceptance Evidence test commands pass; record Task 4's counts here + - fresh runs at `4940ea1`: build green as above; unit: passed 2302, failed 0, skipped 0 (2302 total); + integration: passed 197, failed 0, skipped 0 (197 total), container runtime reached +- [x] every `git grep` and `git diff --stat` check of Acceptance Evidence prints what it states + - `Subscribe(IReadOnlyList` prints `ILiveSource.cs:8`, `PostgresLiveSource.cs:44`, + `FakeDataProvider.cs:192` only; `ILiveSource` in `SemiPlot.UI` prints `Bridge/TrendCoordinator.cs:17,31` + and `MainWindow/TrendWindow.cs:73` only. + - ⚠️ the literal cast check `\(ILiveSource\)|as ILiveSource` prints one line, + `PostgresCompositionTests.cs:95: [InlineData(typeof(ILiveSource))]`, which is no cast (Task 4); the + narrowed `[^f]\(ILiveSource\)|as ILiveSource` prints nothing (exit 1). + - the `git diff --stat master` over `RealtimePoll.cs` and `ArchiveStatements.cs` prints nothing. +- [x] `dotnet format SemiPlot.slnx --verify-no-changes` and `dotnet terse` over the touched `.cs` files + exit 0 + - touched = `git diff --name-only master...HEAD -- '*.cs'`, 22 files; `dotnet format` exit 0, + `dotnet terse <22 files>` exit 0 + +### Task 6: [Final] Update documentation + +**Files:** +- Modify: `docs/architecture/data-integration.md` +- Modify: `docs/architecture/overview.md` +- Modify: `docs/architecture/charting.md` +- Modify: `docs/architecture/postgres-topology.md` +- Modify: `docs/architecture/trend-interaction.md` +- Modify: `docs/architecture/testing-strategy.md` +- Modify: `AGENTS.md` + +- [x] `data-integration.md`, The provider surface (`:28-62`): drop `Subscribe` from the `IDataProvider` + block; add `ILiveSource` and `PostgresLiveSource` as a fourth Core interface beside the reader and the + two write paths, with its code block +- [x] `data-integration.md`, Realtime (`:941-947`): open with the port's contract (Solution Overview), + naming what each terminal message does to the live edge (Overview) and which breaks travel through + the port, with no rule for a break ahead of a first live sample (Solution Overview); the bullet "never completes and never faults" (`:977-978`) and the poll's own points stay, + stated for `PostgresLiveSource`; `Select(_dataProvider.Subscribe)` at `:963-967` reads + `Select(_liveSource.Subscribe)`, and `:974` says "pens of the live source's catalogue"; `:1248` says + "out of the live source"; note that the poll's lines carry the `PostgresLiveSource` source context + - as built, the pen-set paragraph keeps "pens of the provider's own catalogue" and adds "the live + source drops any identifier it does not serve": `TrendCoordinator.cs:26-28` states the set is the + provider's catalogue +- [x] `overview.md`: the diagram's interface list and Core box (`:69-73,77`), the Postgres box (`:88-90`), + the paragraph at `:97-104`, the Realtime data flow (`:333-338`) and Scope status (`:905-906`) name the + port +- [x] `charting.md:773-775,789-793`: the live edge comes through `ILiveSource`; `SetPens` switches the + live-source subscription; `:67` and `trend-interaction.md:72` say "a break the live source reported" +- [x] `postgres-topology.md`: the diagram (`:104-107,120-121`) and the text at `:130-132,139-140` name + `PostgresLiveSource` for the live-edge poll +- [x] `trend-interaction.md:61-62`: the coordinator signature carries `ILiveSource liveSource` +- [x] `testing-strategy.md:226`: the `LiveEdgeArchiveJourneyTests` row says "over the live source alone" + where it says "over the provider alone"; `git grep -n "over the provider alone" -- docs` prints nothing + - ⚠️ over `docs` the grep prints two plan lines: this checkbox, which quotes the phrase, and + `docs/plans/completed/20260824-postgres-live-edge-and-demo.md:1649`, a completed plan left as + written; `git grep -n "over the provider alone" -- docs ':!docs/plans'` prints nothing (exit 1). +- [x] `AGENTS.md`, Data-source projects (`:396-403`): one sentence: `ILiveSource` lives in + `SemiPlot.Core/Data` beside `IDataProvider`, and `AddPostgresData()` registers `PostgresLiveSource` + over the same `NpgsqlDataSource` (`docs/architecture/data-integration.md#realtime`) + +## Post-Completion + +**Manual verification:** + +1. Start the demo stand: `dotnet run --project SemiPlot/SemiPlot.AppHost`. +2. With the time window at the live edge, the newest point of a drawn pen advances as the demo writer + appends, as before. +3. Run the viewer at `--logging-level Debug` and open `semiplot.log`: the realtime poll lines carry the + source context `SemiPlot.DataSource.Postgres.PostgresLiveSource`; the history-read lines still carry + `PostgresDataProvider`. + +**Remaining slices** (docs/plans/roadmaps/20261019-opc-live-edge-roadmap.md): + +- opc-demo-server: an OPC UA server in the ArchiveSeeder demo writer with the Simple-Scada address space, + port 49875. +- opc-live-source (Closes #116): `SemiPlot.DataSource.OpcUa` feeds the live edge; the poll is deleted. +- opc-settings: an OPC block after the DB block in the settings window. +- live-hold-to-edge: a grey pending gap and a render-only hold of the last value to the edge. + +**Executed by exec:** + +- branch: live-source-port + +## Verify it yourself + +This slice moves a seam and fixes no defect, so no test fails at `master` and passes on the branch: the +new tests name `ILiveSource`, which does not exist at `master`. The checks below prove the move instead. + +1. Build and run both projects at the branch tip (`5e8e59b` or later): + + ```powershell + dotnet build SemiPlot.slnx + dotnet test SemiPlot/SemiPlot.Tests.Unit/SemiPlot.Tests.Unit.csproj + dotnet test SemiPlot/SemiPlot.Tests.Integration/SemiPlot.Tests.Integration.csproj + ``` + + Expected: 0 errors and only the two `NU1902` warnings of `SemiPlot.AppHost`; unit 2303 passed, integration + 200 passed (Docker running). Task 5's 2302 and 197 were taken at `4940ea1`, before review added four tests. +2. Run the four `git grep`/`git diff` checks of Acceptance Evidence. Expected: `Subscribe(IReadOnlyList` + prints `ILiveSource.cs`, `PostgresLiveSource.cs` and `FakeDataProvider.cs` only; `ILiveSource` under + `SemiPlot.UI` prints `TrendCoordinator.cs` and `TrendWindow.cs` only; the cast check prints nothing; the + diff over `RealtimePoll.cs` and `ArchiveStatements.cs` is empty. +3. The guard: delete `IDataProvider.Subscribe` at `master` and build. Expected: CS1061 at + `TrendCoordinator.cs:98` (Task 1 records the full list). On the branch the member is already gone and + the build is green. +4. The tests that pin what the compiler cannot: + `dotnet test SemiPlot/SemiPlot.Tests.Unit/SemiPlot.Tests.Unit.csproj --filter "FullyQualifiedName~TheWindowFollowsTheContainersLiveSource|FullyQualifiedName~TheLiveEdgeFollowsTheLiveSource|FullyQualifiedName~ALiveSourceThatCompletes"` + and + `dotnet test SemiPlot/SemiPlot.Tests.Integration/SemiPlot.Tests.Integration.csproj --filter "FullyQualifiedName~RealtimePollTests"`. + Replacing `GetRequiredService()` in `TrendWindow.Build` with `(ILiveSource)dataProvider` + turns `TheWindowFollowsTheContainersLiveSource` red. +5. The one output change is manual: Post-Completion step 3 (the poll's log lines under + `SemiPlot.DataSource.Postgres.PostgresLiveSource`). `RealtimePollTests.ThePollLogsUnderTheLiveSourcesCategory` + pins the same category without a server.