refactor(data): read the live edge through a port

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.
 
 ---
 

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<IReadOnlyList<Sample>> Subscribe(IReadOnlyList<int> penIds);
-
 	Task<Result<IReadOnlyList<Pen>>> QueryPensAsync();
 
 	Task<Result<IReadOnlyList<PenHistoryEnvelope>>> QueryHistoryAsync(

SemiPlot/SemiPlot.Core/Data/ILiveSource.cs

@@ -0,0 +1,9 @@
+using SemiPlot.Core.Trends;
+
+namespace SemiPlot.Core.Data;
+
+/// <summary>The live edge of a pen set (docs/architecture/data-integration.md#realtime).</summary>
+public interface ILiveSource
+{
+	IObservable<IReadOnlyList<Sample>> Subscribe(IReadOnlyList<int> penIds);
+}

SemiPlot/SemiPlot.Core/Trends/RealtimeBatch.cs

@@ -2,7 +2,7 @@ namespace SemiPlot.Core.Trends;
 
 /// <summary>
 /// One pen's share of a <see cref="RealtimeBatch"/>, 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.
 /// </summary>
 public sealed record PenRealtimeValues(
 	int PenId,

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;
 
 /// <summary>
 /// Reads the archive over the pooled <see cref="NpgsqlDataSource"/>: <see cref="QueryPensAsync"/> answers the
-/// configured variables, <see cref="QueryArchiveExtentAsync"/> the span they cover, <see cref="QueryHistoryAsync"/>
-/// a window of one layer and <see cref="Subscribe"/> the live edge, all crossing the boundary in UTC.
+/// configured variables, <see cref="QueryArchiveExtentAsync"/> the span they cover and
+/// <see cref="QueryHistoryAsync"/> a window of one layer, all crossing the boundary in UTC.
 /// </summary>
 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<PostgresDataProvider> _logger;
 	private readonly int _historyReadConcurrency;
@@ -57,35 +52,18 @@ internal PostgresDataProvider(
 		NpgsqlDataSource dataSource,
 		ArchiveTimeConverter timeConverter,
 		ArchiveExceptionMapper exceptionMapper,
-		PostgresConnectionSettings settings,
-		IScheduler scheduler,
 		TimeProvider timeProvider,
 		ILogger<PostgresDataProvider> logger,
 		int historyReadConcurrency)
 	{
 		_dataSource = dataSource;
 		_timeConverter = timeConverter;
 		_exceptionMapper = exceptionMapper;
-		_settings = settings;
-		_scheduler = scheduler;
 		_timeProvider = timeProvider;
 		_logger = logger;
 		_historyReadConcurrency = historyReadConcurrency;
 	}
 
-	/// <summary>
-	/// 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 <see cref="PostgresConnectionSettings.PollInterval"/>, holding
-	/// its own baseline; disposing it cancels the loop's query and its wait, so no further statement is issued.
-	/// </summary>
-	public IObservable<IReadOnlyList<Sample>> Subscribe(IReadOnlyList<int> penIds)
-	{
-		var subscribedIds = penIds.ToArray();
-
-		return Observable.Create<IReadOnlyList<Sample>>(observer =>
-			_scheduler.ScheduleAsync((_, cancellationToken) => PollAsync(subscribedIds, observer, cancellationToken)));
-	}
-
 	/// <summary>
 	/// Every configured variable, ordered by name and then by id.
 	/// </summary>
@@ -337,45 +315,6 @@ private void LogHistoryRead(
 			stopwatch.ElapsedMilliseconds);
 	}
 
-	private async Task PollAsync(
-		int[] penIds,
-		IObserver<IReadOnlyList<Sample>> 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)

SemiPlot/SemiPlot.DataSource.Postgres/PostgresDataServiceCollectionExtensions.cs

@@ -38,11 +38,16 @@ internal static IServiceCollection AddPostgresData(
 			provider.GetRequiredService<NpgsqlDataSource>(),
 			provider.GetRequiredService<ArchiveTimeConverter>(),
 			provider.GetRequiredService<ArchiveExceptionMapper>(),
-			provider.GetRequiredService<PostgresConnectionSettings>(),
-			provider.GetRequiredService<IScheduler>(),
 			provider.GetRequiredService<TimeProvider>(),
 			provider.GetRequiredService<ILogger<PostgresDataProvider>>(),
 			historyReadConcurrency));
+		services.AddSingleton<ILiveSource>(provider => new PostgresLiveSource(
+			provider.GetRequiredService<NpgsqlDataSource>(),
+			provider.GetRequiredService<ArchiveTimeConverter>(),
+			provider.GetRequiredService<ArchiveExceptionMapper>(),
+			provider.GetRequiredService<IScheduler>(),
+			provider.GetRequiredService<ILogger<PostgresLiveSource>>(),
+			provider.GetRequiredService<PostgresConnectionSettings>().PollInterval));
 		services.AddSingleton<IPenCatalogueEditor>(provider => new PostgresPenCatalogueEditor(
 			provider.GetRequiredService<NpgsqlDataSource>(),
 			provider.GetRequiredService<ArchiveExceptionMapper>(),

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;
+
+/// <summary>Serves the live edge by polling the archive over the pooled <see cref="NpgsqlDataSource"/>.</summary>
+public sealed class PostgresLiveSource : ILiveSource
+{
+	private readonly NpgsqlDataSource _dataSource;
+	private readonly ArchiveTimeConverter _timeConverter;
+	private readonly ArchiveExceptionMapper _exceptionMapper;
+	private readonly IScheduler _scheduler;
+	private readonly ILogger<PostgresLiveSource> _logger;
+	private readonly TimeSpan _pollInterval;
+
+	internal PostgresLiveSource(
+		NpgsqlDataSource dataSource,
+		ArchiveTimeConverter timeConverter,
+		ArchiveExceptionMapper exceptionMapper,
+		IScheduler scheduler,
+		ILogger<PostgresLiveSource> logger,
+		TimeSpan pollInterval)
+	{
+		_dataSource = dataSource;
+		_timeConverter = timeConverter;
+		_exceptionMapper = exceptionMapper;
+		_scheduler = scheduler;
+		_logger = logger;
+		_pollInterval = pollInterval;
+	}
+
+	/// <summary>
+	/// 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.
+	/// </summary>
+	public IObservable<IReadOnlyList<Sample>> Subscribe(IReadOnlyList<int> penIds)
+	{
+		var subscribedIds = penIds.ToArray();
+
+		return Observable.Create<IReadOnlyList<Sample>>(observer =>
+			_scheduler.ScheduleAsync((_, cancellationToken) => PollAsync(subscribedIds, observer, cancellationToken)));
+	}
+
+	private async Task PollAsync(
+		int[] penIds,
+		IObserver<IReadOnlyList<Sample>> 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.
+		}
+	}
+}

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<IDataProvider>();
+		var liveSource = services.GetRequiredService<ILiveSource>();
 		var dataScheduler = services.GetRequiredService<IScheduler>();
 		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(),

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<LiveEdgeStand> OpenStandAsync()
 		try
 		{
 			var dataProvider = services.GetRequiredService<IDataProvider>();
+			var liveSource = services.GetRequiredService<ILiveSource>();
 			var dataScheduler = services.GetRequiredService<IScheduler>();
 			var catalogue = await dataProvider.QueryPensAsync();
 			var extent = await dataProvider.QueryArchiveExtentAsync();
@@ -177,7 +178,7 @@ private async Task<LiveEdgeStand> 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(

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<IDataProvider>();
+		var liveSource = services.GetRequiredService<ILiveSource>();
 
-		using var subscription = provider.Subscribe(SubscribedPenIds).Subscribe(_ => { });
+		using var subscription = liveSource.Subscribe(SubscribedPenIds).Subscribe(_ => { });
 
 		await armed.Armed(1).WaitAsync(GateBound, TestContext.Current.CancellationToken);
 	}