feat(data): raise the archive state from the catalogue

AGENTS.md

@@ -188,8 +188,8 @@ Frame cost during a drag is measured, not asserted: `docs/architecture/testing-s
 ### Container tests
 
 Every test of the `ArchiveDatabaseCollection` in `SemiPlot.Tests.Integration` needs a container runtime and
-nothing else. The socket classes (`RealtimePollTests`, `RealtimeSubscriptionDisposalTests`,
-`ConvergeReadinessTests`) carry no collection, take no fixture and run without a runtime.
+nothing else. The socket classes (`RealtimePollTests`, `CatalogueReadUnreachableTests`,
+`RealtimeSubscriptionDisposalTests`, `ConvergeReadinessTests`) carry no collection, take no fixture and run without a runtime.
 `SemiPlot/bench/Dockerfile` copies `/semibase` out of
 `ghcr.io/semiteq/semibase:latest`, which the fixture pulls ahead of the build, and runs
 `semibase bench` from `/docker-entrypoint-initdb.d/` before the published port opens. The container
@@ -358,10 +358,12 @@ No abbreviations in names.
   `Pens and groups`, `docs/architecture/overview.md#the-pen-and-group-editor`), so a control
   added to the row fails the suite rather than quietly shipping a second editor.
 - A stored pen change reaches the running chart through `Bridge/PenCatalogueSync` and no other route: it
-  re-reads `IDataProvider.QueryPensAsync` every 5 s and on `ReadNow()` after each write the editor lands,
-  and `MainWindow/PenCatalogueApplier` hands each read to the chart, which compares it with the pens it
-  shows, then rebuilds the sidebar. The chart is never rebuilt whole
-  (`docs/architecture/overview.md#the-live-catalogue`).
+  re-reads `IDataProvider.QueryPensAsync` once when the window opens, every 5 s and on `ReadNow()` after
+  each write the editor lands, and `MainWindow/PenCatalogueApplier` hands each read to the chart, which
+  compares it with the pens it shows, then rebuilds the sidebar. The chart is never rebuilt whole
+  (`docs/architecture/overview.md#the-live-catalogue`). The same reads are the source of the archive
+  connection state the status bar shows, `PenCatalogueSync.ConnectionStates`
+  (`docs/architecture/data-integration.md#the-connection-state-the-catalogue-read-reports`).
 - A stored time marker reaches the running chart through `Markers/MarkerSync` and no other route: it
   re-reads `IMarkerStore.ReadAsync` every 5 s and on `ReadNow()` after each marker write, and
   `Chart/ChartMarkers` reconciles each read into `Core/Trends/MarkerSet`, which keeps a marker
@@ -379,7 +381,11 @@ No abbreviations in names.
   at the call site. The one message with no error behind it is
   `AppStatusBarViewModel.ConnectionRestored`, which builds its own `Info` view because the mapper maps
   errors. A `catch`, or an Rx `onError`, that only logs is a defect
-  (`docs/architecture/data-integration.md#no-failure-stops-at-the-log`). Code-behind reaches the
+  (`docs/architecture/data-integration.md#no-failure-stops-at-the-log`). A failure for lack of a
+  connection (`ArchiveFault.Unreachable`) in the catalogue loop, the marker loop or the live-edge poll is
+  logged by that reader and reaches the panel through the status bar's one `ConnectionLost` entry, which
+  the catalogue reads raise; the three cases that reach only the log are listed in that same section.
+  Code-behind reaches the
   panel through its view model (`TrendChartViewModel.ReportFailure`, `MainWindowViewModel.ReportFailure`,
   `PenEditorViewModel.ReportFailure`, `StartupFailureViewModel.ReportFailure`,
   `AxisScalePanelViewModel.ReportFailure`, `TimeWindowPanelViewModel.ReportFailure`,

SemiPlot/SemiPlot.Core/Data/ArchiveConnectionState.cs

@@ -3,9 +3,8 @@
 namespace SemiPlot.Core.Data;
 
 /// <summary>
-/// What a provider reports about its own connection to the archive. <see cref="Fault"/> is null while the
-/// archive answers and names the fault while it does not; never filter with <c>DistinctUntilChanged</c>,
-/// every subscription's first tick reports <see cref="Connected"/>.
+/// What a reader of the archive reports about its connection to it. <see cref="Fault"/> is null while the
+/// archive answers and names the fault while it does not.
 /// </summary>
 public sealed record ArchiveConnectionState(ArchiveError? Fault)
 {

SemiPlot/SemiPlot.Core/Data/Errors/ArchiveError.cs

@@ -35,7 +35,7 @@ private static string Describe(ArchiveFault kind, string host, int port, string
 				$"The {archive} does not carry the columns SemiPlot reads (SQLSTATE 42703): {detail}",
 			ArchiveFault.QueryTimedOut => $"The server ended a statement against the {archive} (SQLSTATE 57014).",
 			ArchiveFault.ConnectionLost =>
-				$"The live edge of the {archive} stopped answering after {detail} consecutive failed reads.",
+				$"The {archive} stopped answering after {detail} consecutive failed reads.",
 			ArchiveFault.ValueRejected =>
 				$"The {archive} refused a value written to pen '{detail}' (SQLSTATE 23514).",
 			ArchiveFault.NameTaken => $"The {archive} already holds a group named '{detail}' (SQLSTATE 23505).",

SemiPlot/SemiPlot.Core/Data/Errors/ArchiveFault.cs

@@ -6,7 +6,10 @@ namespace SemiPlot.Core.Data.Errors;
 /// </summary>
 public enum ArchiveFault
 {
-	/// <summary>No connection: a refused or reset socket, a client bound firing, a host that does not answer.</summary>
+	/// <summary>
+	/// No connection: a refused or reset socket, a client bound firing, a host that does not answer, or a server
+	/// that refuses or ends the session (SQLSTATE class 08, 57P01-57P03, 53300).
+	/// </summary>
 	Unreachable,
 
 	/// <summary>The server answered and refused the credentials or a grant (28P01, 28000, 42501). Detail is the username.</summary>
@@ -24,7 +27,9 @@ public enum ArchiveFault
 	/// <summary>The server ended a statement (57014): a statement_timeout or an administrator's cancel.</summary>
 	QueryTimedOut,
 
-	/// <summary>A run of consecutive poll ticks failed. Detail is the number of failures that raised it.</summary>
+	/// <summary>
+	/// A run of consecutive reads failed for lack of a connection. Detail is the number of failures that raised it.
+	/// </summary>
 	ConnectionLost,
 
 	/// <summary>A constraint refused a written value (23514). Detail is the pen's name.</summary>

SemiPlot/SemiPlot.Core/Data/IDataProvider.cs

@@ -9,11 +9,6 @@ public interface IDataProvider
 	// Cold per call: no samples flow until subscribed; the subscriber disposes the returned IDisposable.
 	IObservable<IReadOnlyList<Sample>> Subscribe(IReadOnlyList<int> penIds);
 
-	// Hot, shared and never terminating: a consumer subscribes with an onNext handler alone. Every
-	// subscription's first successful tick reports Connected, the only point it is known armed, and a
-	// run of failed ticks reports the fault here rather than through OnError.
-	IObservable<ArchiveConnectionState> ConnectionFaults { get; }
-
 	Task<Result<IReadOnlyList<Pen>>> QueryPensAsync();
 
 	Task<Result<IReadOnlyList<PenHistoryEnvelope>>> QueryHistoryAsync(

SemiPlot/SemiPlot.DataSource.Postgres/ArchiveExceptionMapper.cs

@@ -86,8 +86,23 @@ private ArchiveError Classify(Exception exception, string? relation)
 		return Fault(IsConnectionFailure(exception) ? ArchiveFault.Unreachable : ArchiveFault.ReadFailed);
 	}
 
+	// docs/architecture/data-integration.md#the-connection-state-the-catalogue-read-reports
+	private static bool EndsOrRefusesSession(string sqlState)
+	{
+		return sqlState.StartsWith("08", StringComparison.Ordinal)
+			|| sqlState is PostgresErrorCodes.AdminShutdown
+				or PostgresErrorCodes.CrashShutdown
+				or PostgresErrorCodes.CannotConnectNow
+				or PostgresErrorCodes.TooManyConnections;
+	}
+
 	private ArchiveError MapSqlState(PostgresException postgres, string? relation)
 	{
+		if (EndsOrRefusesSession(postgres.SqlState))
+		{
+			return Fault(ArchiveFault.Unreachable);
+		}
+
 		return postgres.SqlState switch
 		{
 			PostgresErrorCodes.InvalidCatalogName => Fault(ArchiveFault.DatabaseMissing),

SemiPlot/SemiPlot.DataSource.Postgres/Configuration/PostgresConnectionSettings.cs

@@ -23,7 +23,7 @@ public sealed record PostgresConnectionSettings(
 	public const int ConnectTimeoutSeconds = 15;
 
 	/// <summary>
-	/// Client backstop for a server that stops answering; the poll overrides it per tick.
+	/// Client backstop for a server that stops answering; the poll's tick and the catalogue read override it.
 	/// </summary>
 	public const int CommandTimeoutSeconds = 300;
 

SemiPlot/SemiPlot.DataSource.Postgres/PostgresDataProvider.cs

@@ -3,7 +3,6 @@
 using System.Globalization;
 using System.Reactive.Concurrency;
 using System.Reactive.Linq;
-using System.Reactive.Subjects;
 
 using FluentResults;
 
@@ -24,14 +23,17 @@ namespace SemiPlot.DataSource.Postgres;
 /// 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.
 /// </summary>
-public sealed class PostgresDataProvider : IDataProvider, IDisposable
+public sealed class PostgresDataProvider : IDataProvider
 {
 	/// <summary>
 	/// The most pen groups one history read runs concurrently
 	/// (docs/architecture/data-integration.md#operation-to-statement).
 	/// </summary>
 	internal const int HistoryReadConcurrency = 4;
 
+	// docs/architecture/data-integration.md#configuration
+	private const int CatalogueCommandTimeoutSeconds = 5;
+
 	private const string UncommissionedPenColor = "#808080";
 	private const string HistoryReadRequestTemplate =
 		"History read at {Layer} from {FromUtc:O} to {ToUtc:O}, {TargetColumnCount} columns for {PenCount} pens "
@@ -46,19 +48,11 @@ public sealed class PostgresDataProvider : IDataProvider, IDisposable
 	private readonly ILogger<PostgresDataProvider> _logger;
 	private readonly int _historyReadConcurrency;
 
-	private readonly Subject<ArchiveConnectionState> _connectionFaultsSource = new();
-
-	// PollAsync's OnNext and Dispose's OnCompleted run on different threads; Synchronize() serializes them,
-	// as System.Reactive's contract for a multi-writer Subject requires.
-	private readonly ISubject<ArchiveConnectionState> _connectionFaults;
-
 	// docs/architecture/data-integration.md#the-catalogue-while-the-viewer-runs
 	// Written from whichever pool thread a catalogue read resumes on.
 	private readonly ConcurrentDictionary<(int PenId, int Column, string? StoredValue), byte> _warnedNormalisations =
 		new();
 
-	private int _disposed;
-
 	internal PostgresDataProvider(
 		NpgsqlDataSource dataSource,
 		ArchiveTimeConverter timeConverter,
@@ -77,15 +71,8 @@ internal PostgresDataProvider(
 		_timeProvider = timeProvider;
 		_logger = logger;
 		_historyReadConcurrency = historyReadConcurrency;
-		_connectionFaults = Subject.Synchronize(_connectionFaultsSource);
 	}
 
-	/// <summary>
-	/// The connection state of the live edge. Hot, shared by every subscription and never terminating, as
-	/// <see cref="IDataProvider.ConnectionFaults"/> states.
-	/// </summary>
-	public IObservable<ArchiveConnectionState> ConnectionFaults => _connectionFaults;
-
 	/// <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
@@ -107,7 +94,7 @@ public async Task<Result<IReadOnlyList<Pen>>> QueryPensAsync()
 		try
 		{
 			await using var connection = await _dataSource.OpenConnectionAsync().ConfigureAwait(false);
-			await using var command = new NpgsqlCommand(ArchiveStatements.PenCatalog, connection);
+			await using var command = CreateCatalogueCommand(connection);
 			await using var reader = await command.ExecuteReaderAsync().ConfigureAwait(false);
 
 			var pens = new List<Pen>();
@@ -243,22 +230,6 @@ public async Task<Result<IReadOnlyList<DateOnly>>> QueryArchiveDaysAsync(
 		}
 	}
 
-	/// <summary>
-	/// Completes <see cref="ConnectionFaults"/>; the container disposes the singleton provider.
-	/// </summary>
-	public void Dispose()
-	{
-		if (Interlocked.Exchange(ref _disposed, 1) != 0)
-		{
-			return;
-		}
-
-		// The raw subject is never disposed: PollAsync keeps writing OnNext after Dispose returns (its
-		// loop is not cancelled here), and OnNext on a disposed Subject<T> throws where OnNext after
-		// OnCompleted is a silent no-op. Subject<T> holds no unmanaged resource, so completing it is enough.
-		_connectionFaults.OnCompleted();
-	}
-
 	// docs/architecture/data-integration.md#operation-to-statement
 	private async Task<IReadOnlyList<PenHistoryEnvelope>> ReadHistoryAsync(
 		int[] penIds,
@@ -375,15 +346,14 @@ private async Task PollAsync(
 			_dataSource,
 			_timeConverter,
 			_exceptionMapper,
-			_settings,
 			penIds,
 			_logger);
 
 		try
 		{
 			while (!cancellationToken.IsCancellationRequested)
 			{
-				var tick = await poll.ReadOnceAsync(cancellationToken).ConfigureAwait(false);
+				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.
@@ -392,14 +362,9 @@ private async Task PollAsync(
 					return;
 				}
 
-				if (tick.Samples.Count > 0)
+				if (samples.Count > 0)
 				{
-					observer.OnNext(tick.Samples);
-				}
-
-				if (tick.StateChange is { } state)
-				{
-					_connectionFaults.OnNext(state);
+					observer.OnNext(samples);
 				}
 
 				await _scheduler.Sleep(_settings.PollInterval, cancellationToken).ConfigureAwait(false);
@@ -411,6 +376,14 @@ private async Task PollAsync(
 		}
 	}
 
+	internal static NpgsqlCommand CreateCatalogueCommand(NpgsqlConnection connection)
+	{
+		return new NpgsqlCommand(ArchiveStatements.PenCatalog, connection)
+		{
+			CommandTimeout = CatalogueCommandTimeoutSeconds
+		};
+	}
+
 	// Bound on the archive's own wall clock: a UTC round trip is not injective across DST.
 	internal static void BindLocalBucketedWindow(
 		NpgsqlCommand command,