fix(chart): draw a live-edge break as a gap

SemiPlot/SemiPlot.Core/Trends/RealtimeBatch.cs

@@ -1,13 +1,13 @@
 namespace SemiPlot.Core.Trends;
 
 /// <summary>
-/// One pen's share of a <see cref="RealtimeBatch"/>, on that pen's own timestamps and nothing else;
-/// <c>double</c>, not <c>double?</c>, because a null here would draw a break the archive never recorded.
+/// 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.
 /// </summary>
 public sealed record PenRealtimeValues(
 	int PenId,
 	IReadOnlyList<DateTime> TimestampsUtc,
-	IReadOnlyList<double> Values);
+	IReadOnlyList<double?> Values);
 
 /// <summary>
 /// One buffer window of the live edge. <paramref name="Timestamps"/> is the ascending union of every pen's own

SemiPlot/SemiPlot.Core/Trends/Sample.cs

@@ -1,3 +1,4 @@
 namespace SemiPlot.Core.Trends;
 
-public sealed record Sample(int PenId, DateTime TimestampUtc, double Value);
+/// <summary>A null <see cref="Value"/> is a break for the pen from this moment until its next sample.</summary>
+public sealed record Sample(int PenId, DateTime TimestampUtc, double? Value);

SemiPlot/SemiPlot.DataSource.Postgres/RealtimePoll.cs

@@ -31,14 +31,16 @@ internal sealed class RealtimePoll(
 	int[] penIds,
 	ILogger logger)
 {
-	// q = 32 marks the last sample before a break; read for the debug line only
-	// (docs/architecture/scada-archive.md, Quality and gaps).
+	// docs/architecture/scada-archive.md#quality-and-gaps
 	private const int LastBeforeBreakQuality = 32;
 
 	// Must stay far below the connection string's backstop: three stalled ticks must raise the fault
 	// inside half a minute.
 	private const int TickCommandTimeoutSeconds = 10;
 
+	// docs/architecture/data-integration.md#realtime
+	private static readonly TimeSpan _breakAnchorStep = TimeSpan.FromMilliseconds(1);
+
 	private static readonly IReadOnlyList<Sample> _noSamples = [];
 
 	private int _consecutiveFailures;
@@ -150,11 +152,14 @@ private async Task<IReadOnlyList<Sample>> ReadSamplesAsync(
 		var newest = lastSeen;
 		var rowCount = 0;
 		var droppedCount = 0;
-		var breakMarkCount = 0;
+		var breakCount = 0;
 
 		while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 		{
+			var penId = reader.GetInt32(0);
 			var archiveLocal = reader.GetDateTime(1);
+			var timestampUtc = timeConverter.ToUtc(archiveLocal);
+			var isBreakMark = reader.GetInt32(3) == LastBeforeBreakQuality;
 
 			rowCount++;
 
@@ -163,33 +168,33 @@ private async Task<IReadOnlyList<Sample>> ReadSamplesAsync(
 				newest = archiveLocal;
 			}
 
-			if (reader.GetInt32(3) == LastBeforeBreakQuality)
-			{
-				breakMarkCount++;
-			}
-
-			// Sample.Value is non-nullable; GetDouble would throw and count as a connection failure, so the
-			// row is dropped and lastSeen still advances.
+			// docs/architecture/data-integration.md#realtime
 			if (reader.IsDBNull(2))
 			{
 				droppedCount++;
-
-				continue;
+			}
+			else
+			{
+				samples.Add(new Sample(penId, timestampUtc, reader.GetDouble(2)));
 			}
 
-			samples.Add(new Sample(reader.GetInt32(0), timeConverter.ToUtc(archiveLocal), reader.GetDouble(2)));
+			if (isBreakMark)
+			{
+				samples.Add(new Sample(penId, timestampUtc + _breakAnchorStep, null));
+				breakCount++;
+			}
 		}
 
 		Advance(newest);
 
 		logger.LogDebug(
-			"The realtime poll read {RowCount} rows past {LastSeen:O}: {SampleCount} samples, {DroppedCount} "
-			+ "dropped for a null value, {BreakMarkCount} marking the last sample before a break.",
+			"The realtime poll read {RowCount} rows past {LastSeen:O}: {ValueCount} value samples, {DroppedCount} "
+			+ "rows dropped for a null value, {BreakCount} breaks.",
 			rowCount,
 			lastSeen,
-			samples.Count,
+			rowCount - droppedCount,
 			droppedCount,
-			breakMarkCount);
+			breakCount);
 
 		return samples;
 	}

SemiPlot/SemiPlot.Tests.Integration/Journeys/LiveEdgeArchiveJourneyTests.cs

@@ -86,6 +86,44 @@ public async Task ARowWrittenAfterStartupReachesTheChartOnceAndMovesItsLiveEdge(
 		chart.Pens.Should().AllSatisfy(pen => pen.CurrentValue.Should().Be(ValueFor(pen.Pen.PenId, tick: 2)));
 	}
 
+	[AvaloniaFact]
+	public async Task ARowMarkingTheLastSampleBeforeABreakReachesTheChartAsItsValueAndThenABreak()
+	{
+		await using var stand = await OpenStandAsync();
+		var pen = stand.Catalogue[0];
+		var seededLastLocal = ArchiveProviderFactory.TimeConverter.ToArchiveLocal(stand.Extent.LastUtc);
+		var appendLocal = seededLastLocal.AddSeconds(1);
+		var appendUtc = ArchiveProviderFactory.TimeConverter.ToUtc(appendLocal);
+		var value = ValueFor(pen.PenId, tick: 1);
+		var delivery = stand.Batches.NextBatch();
+
+		await AppendAsync(appendLocal, [pen], tick: 1, ArchiveRow.LastBeforeBreakQuality);
+		await delivery.WaitAsync(GateBound, TestContext.Current.CancellationToken);
+
+		// The batch is what the poll delivered, which no history read can race.
+		var batch = stand.Batches.Batches.Should().ContainSingle().Subject;
+		var breakUtc = appendUtc.AddMilliseconds(1);
+
+		batch.Timestamps.Should().Equal([appendUtc, breakUtc]);
+		var penValues = batch.Pens.Should().ContainSingle().Subject;
+		penValues.PenId.Should().Be(pen.PenId);
+		penValues.Values.Should().Equal([value, null]);
+		stand.Chart.Navigation.To.Should().Be(breakUtc, "the break is the batch's last timestamp");
+
+		var state = stand.Chart.Pens.Should().ContainSingle(candidate => candidate.Pen.PenId == pen.PenId).Subject;
+		var columns = state.Line.SnapshotColumns();
+
+		// A landed history read may add a NaN ahead of the break:
+		// docs/architecture/charting.md#per-pen-plottable-envelopeline
+		var lastRealColumn = columns.Last(column => !double.IsNaN(column.Center));
+
+		lastRealColumn.X.Should().Be(LocalTimeAxis.ToAxis(appendUtc));
+		lastRealColumn.Center.Should().Be(value);
+		columns[^1].X.Should().Be(LocalTimeAxis.ToAxis(breakUtc));
+		double.IsNaN(columns[^1].Center).Should().BeTrue("the row marks the last sample before a break");
+		state.CurrentValue.Should().Be(value);
+	}
+
 	// The archive is per-variable and change-based with a deadband, so one poll tick spans as many distinct
 	// timestamps as there are variables; a pen with no sample at a timestamp is left alone rather than
 	// appended as null, which would draw a break the archive never recorded.
@@ -120,7 +158,7 @@ public async Task RowsOnAVariableOfTheirOwnReachTheChartWithoutBreakingAnyPen()
 		});
 	}
 
-	// Only the raw layer appends a realtime sample as a point of its own; a coarse layer folds it
+	// Only the raw layer appends a realtime value as a point of its own; a coarse layer folds a value
 	// into the last column instead, and the assertions read the appended point.
 	private async Task<LiveEdgeStand> OpenStandAsync()
 	{
@@ -192,15 +230,19 @@ private async Task<LiveEdgeStand> OpenStandAsync()
 	// One row per variable, written through the appending path ArchiveWriter takes for a follow run: the
 	// archive already carries the template's rows, so the seeded refusal has to be stood down. The day
 	// partition the rows fall into already exists, and CreateStatement's IF NOT EXISTS passes through it.
-	private async Task AppendAsync(DateTime archiveLocal, IReadOnlyList<Pen> pens, int tick)
+	private async Task AppendAsync(
+		DateTime archiveLocal,
+		IReadOnlyList<Pen> pens,
+		int tick,
+		int quality = ArchiveRow.OrdinaryQuality)
 	{
 		var rows = pens
 			.Select(pen => new ArchiveRow(
 				pen.PenId,
 				ArchiveRow.RawLayer,
 				archiveLocal,
 				ValueFor(pen.PenId, tick),
-				ArchiveRow.OrdinaryQuality))
+				quality))
 			.ToArray();
 
 		var written = await Writer()

SemiPlot/SemiPlot.Tests.Integration/RealtimePollReadTests.cs

@@ -4,6 +4,7 @@
 
 using SemiPlot.Core.Data;
 using SemiPlot.Core.Data.Errors;
+using SemiPlot.Core.Trends;
 using SemiPlot.Tools.ArchiveSeeder;
 
 using Xunit;
@@ -144,23 +145,20 @@ await AppendAsync(
 
 		await poll.ReadOnceAsync(cancellationToken);
 
-		var rowCounts = logs.Collector.GetSnapshot()
-			.Select(entry => entry.GetStructuredStateValue("RowCount"))
-			.OfType<string>();
-
-		rowCounts.Should().Equal("1", "0");
+		PollCounts(logs).Should().Equal(("1", "0", "1", "0"), ("0", "0", "0", "0"));
 	}
 
-	// The archive's break mark carries a real value (docs/architecture/scada-archive.md, Quality and gaps).
 	[Fact]
-	public async Task ARowMarkingTheLastSampleBeforeABreakEmitsAnOrdinarySample()
+	public async Task ARowMarkingTheLastSampleBeforeABreakEmitsItsValueAndThenABreak()
 	{
 		var cancellationToken = TestContext.Current.CancellationToken;
 		var appended = SeededLast.AddSeconds(1);
+		var appendedUtc = ArchiveProviderFactory.TimeConverter.ToUtc(appended);
 
 		using var services = ArchiveProviderFactory.Build(Database.PlotConnectionString);
+		using var logs = new FakeLoggerProvider();
 
-		var poll = NewPoll(services);
+		var poll = NewPoll(services, logs.CreateLogger("RealtimePoll"));
 
 		await poll.ReadOnceAsync(cancellationToken);
 		await AppendAsync(
@@ -171,10 +169,46 @@ await AppendAsync(
 			ArchiveRow.LastBeforeBreakQuality,
 			cancellationToken: cancellationToken);
 
-		var sample = (await poll.ReadOnceAsync(cancellationToken)).Samples.Should().ContainSingle().Which;
+		var tick = await poll.ReadOnceAsync(cancellationToken);
 
-		sample.Value.Should().Be(11.25);
-		sample.TimestampUtc.Should().Be(ArchiveProviderFactory.TimeConverter.ToUtc(appended));
+		tick.Samples.Should().Equal(
+			new Sample(SubscribedPenIds[0], appendedUtc, 11.25),
+			new Sample(SubscribedPenIds[0], appendedUtc.AddMilliseconds(1), null));
+		PollCounts(logs).Should().Equal(("1", "1", "0", "1"));
+	}
+
+	// docs/architecture/data-integration.md#realtime
+	[Fact]
+	public async Task ARowMarkingTheLastSampleBeforeABreakWithANullValueEmitsOnlyTheBreak()
+	{
+		var cancellationToken = TestContext.Current.CancellationToken;
+		var appended = SeededLast.AddSeconds(1);
+
+		using var services = ArchiveProviderFactory.Build(Database.PlotConnectionString);
+		using var logs = new FakeLoggerProvider();
+
+		var poll = NewPoll(services, logs.CreateLogger("RealtimePoll"));
+
+		await poll.ReadOnceAsync(cancellationToken);
+		await AppendAsync(
+			Database,
+			SubscribedPenIds[0],
+			appended,
+			null,
+			ArchiveRow.LastBeforeBreakQuality,
+			cancellationToken: cancellationToken);
+
+		var tick = await poll.ReadOnceAsync(cancellationToken);
+
+		tick.Samples.Should().Equal(new Sample(
+			SubscribedPenIds[0],
+			ArchiveProviderFactory.TimeConverter.ToUtc(appended).AddMilliseconds(1),
+			null));
+		tick.StateChange.Should().BeNull();
+
+		await poll.ReadOnceAsync(cancellationToken);
+
+		PollCounts(logs).Should().Equal(("1", "0", "1", "1"), ("0", "0", "0", "0"));
 	}
 
 	// Renamed rather than dropped: the container is shared. The tick after recovery must report nothing.
@@ -261,6 +295,19 @@ private Task RestoreTrendsAsync(CancellationToken cancellationToken)
 			cancellationToken);
 	}
 
+	// The baseline entry carries no RowCount.
+	private static IEnumerable<(string? Rows, string? Values, string? Dropped, string? Breaks)> PollCounts(
+		FakeLoggerProvider logs)
+	{
+		return logs.Collector.GetSnapshot()
+			.Where(entry => entry.GetStructuredStateValue("RowCount") is not null)
+			.Select(entry => (
+				entry.GetStructuredStateValue("RowCount"),
+				entry.GetStructuredStateValue("ValueCount"),
+				entry.GetStructuredStateValue("DroppedCount"),
+				entry.GetStructuredStateValue("BreakCount")));
+	}
+
 	// The unsubscribed variable's newest row sits a minute after the subscribed ones, so a baseline read
 	// over the whole archive rather than over the requested identifiers answers a different instant and fails.
 	private static IReadOnlyList<ArchiveRow> SeededRowsWithAnUnsubscribedVariable()

SemiPlot/SemiPlot.Tests.Unit/UI/Bridge/FakeDataProvider.cs

@@ -4,6 +4,8 @@
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
 
+using AwesomeAssertions;
+
 using FluentResults;
 
 using SemiPlot.Core.Data;
@@ -36,6 +38,8 @@ internal sealed class FakeDataProvider(
 	private readonly Lock _historyQueriesGate = new();
 	private readonly List<HistoryQuery> _historyQueries = [];
 
+	private int? _penToBreakOnNextTick;
+
 	public DateTime ArchiveFirstUtc { get; set; } = new(2025, 12, 25, 0, 0, 0, DateTimeKind.Utc);
 
 	public DateTime ArchiveLastUtc { get; set; } = new(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
@@ -181,6 +185,35 @@ public void ReportConnectionState(ArchiveConnectionState state)
 		_connectionFaults.OnNext(state);
 	}
 
+	/// <summary>The next tick follows the pen's value with a break 1 ms later, as the poll's q = 32 row.</summary>
+	public void BreakRealtimePenOnce(int penId)
+	{
+		_penToBreakOnNextTick = penId;
+	}
+
+	/// <summary>The pen's one break across the batches, and the value its tick carried 1 ms ahead of it.</summary>
+	public static (double Value, DateTime ValueUtc, DateTime BreakUtc) TheBreakIn(
+		IEnumerable<RealtimeBatch> batches,
+		int penId)
+	{
+		var samples = batches
+			.SelectMany(batch => batch.Pens)
+			.Where(values => values.PenId == penId)
+			.SelectMany(values => values.TimestampsUtc.Zip(values.Values))
+			.ToList();
+		var breakIndex = samples.FindIndex(sample => sample.Second is null);
+
+		samples.Count(sample => sample.Second is null).Should().Be(1);
+		breakIndex.Should().BePositive();
+
+		var (valueUtc, value) = samples[breakIndex - 1];
+		var breakUtc = samples[breakIndex].First;
+
+		breakUtc.Should().Be(valueUtc.AddMilliseconds(1));
+
+		return (value.GetValueOrDefault(), valueUtc, breakUtc);
+	}
+
 	public IObservable<IReadOnlyList<Sample>> Subscribe(IReadOnlyList<int> penIds)
 	{
 		return Observable
@@ -211,14 +244,30 @@ private IObservable<IReadOnlyList<Sample>> BuildLiveEdge(IReadOnlyList<int> penI
 		return Observable
 			.Interval(_realtimeInterval, _scheduler)
 			.Select(tick => PoisonRealtimeWindow
-				? (IReadOnlyList<Sample>)new PoisonedSampleList()
-				: [.. subscribed
-				.Select((id, index) => new Sample(
-					id,
-					RealtimeEpoch
-					+ TimeSpan.FromTicks(_realtimeInterval.Ticks * (tick + 1))
-					+ (StaggerRealtimeTimestamps ? TimeSpan.FromMilliseconds(index) : TimeSpan.Zero),
-					id + tick))]);
+				? new PoisonedSampleList()
+				: TickSamples(subscribed, tick));
+	}
+
+	private IReadOnlyList<Sample> TickSamples(IReadOnlyList<int> subscribed, long tick)
+	{
+		var samples = new List<Sample>(subscribed.Count + 1);
+
+		for (var index = 0; index < subscribed.Count; index++)
+		{
+			var id = subscribed[index];
+			var timestampUtc = RealtimeEpoch
+				+ TimeSpan.FromTicks(_realtimeInterval.Ticks * (tick + 1))
+				+ (StaggerRealtimeTimestamps ? TimeSpan.FromMilliseconds(index) : TimeSpan.Zero);
+			samples.Add(new Sample(id, timestampUtc, id + tick));
+
+			if (id == _penToBreakOnNextTick)
+			{
+				samples.Add(new Sample(id, timestampUtc.AddMilliseconds(1), null));
+				_penToBreakOnNextTick = null;
+			}
+		}
+
+		return samples;
 	}
 
 	public Task<Result<IReadOnlyList<Pen>>> QueryPensAsync()