feat(chart): cancel the read a gesture left behind

SemiPlot/SemiPlot.Core/Data/IDataProvider.cs

@@ -21,7 +21,8 @@ Task<Result<IReadOnlyList<PenHistoryEnvelope>>> QueryHistoryAsync(
 		DateTime fromUtc,
 		DateTime toUtc,
 		AggregationLayer layer,
-		int targetColumnCount);
+		int targetColumnCount,
+		CancellationToken cancellationToken = default);
 
 	Task<Result<ArchiveExtent>> QueryArchiveExtentAsync();
 }

SemiPlot/SemiPlot.DataSource.Postgres/PostgresDataProvider.cs

@@ -121,7 +121,8 @@ public async Task<Result<IReadOnlyList<PenHistoryEnvelope>>> QueryHistoryAsync(
 		DateTime fromUtc,
 		DateTime toUtc,
 		AggregationLayer layer,
-		int targetColumnCount)
+		int targetColumnCount,
+		CancellationToken cancellationToken = default)
 	{
 		ArgumentNullException.ThrowIfNull(penIds);
 
@@ -142,22 +143,30 @@ public async Task<Result<IReadOnlyList<PenHistoryEnvelope>>> QueryHistoryAsync(
 
 		try
 		{
-			await using var connection = await _dataSource.OpenConnectionAsync().ConfigureAwait(false);
+			await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 
 			var fromLocal = _timeConverter.ToArchiveLocal(fromUtc);
 			var toLocal = _timeConverter.ToArchiveLocal(toUtc);
 
 			if (layer == AggregationLayer.Raw)
 			{
-				var buckets = await ReadBucketedWindowAsync(connection, ids, fromLocal, toLocal, targetColumnCount)
+				var buckets = await ReadBucketedWindowAsync(
+						connection,
+						ids,
+						fromLocal,
+						toLocal,
+						targetColumnCount,
+						cancellationToken)
 					.ConfigureAwait(false);
 
 				return Result.Ok(BucketedRowFold.Fold(buckets, _timeConverter));
 			}
 
-			var rows = await ReadWindowAsync(connection, ids, fromLocal, toLocal, layer).ConfigureAwait(false);
+			var rows = await ReadWindowAsync(connection, ids, fromLocal, toLocal, layer, cancellationToken)
+				.ConfigureAwait(false);
 
-			rows = await FillFreshTailAsync(connection, ids, fromLocal, toLocal, layer, rows).ConfigureAwait(false);
+			rows = await FillFreshTailAsync(connection, ids, fromLocal, toLocal, layer, rows, cancellationToken)
+				.ConfigureAwait(false);
 
 			return Result.Ok(HistoryRowFold.Fold(rows, _timeConverter, targetColumnCount));
 		}
@@ -337,17 +346,18 @@ internal static TimeSpan BucketFor(DateTime fromLocal, DateTime toLocal, int tar
 		int[] penIds,
 		DateTime fromLocal,
 		DateTime toLocal,
-		AggregationLayer layer)
+		AggregationLayer layer,
+		CancellationToken cancellationToken)
 	{
 		await using var command = new NpgsqlCommand(ArchiveStatements.SparseHistoryWindow, connection);
 
 		BindLocalWindow(command, penIds, fromLocal, toLocal, layer);
 
-		await using var reader = await command.ExecuteReaderAsync().ConfigureAwait(false);
+		await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 
 		var rows = new List<HistoryRowFold.Row>();
 
-		while (await reader.ReadAsync().ConfigureAwait(false))
+		while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 		{
 			rows.Add(ReadHistoryRow(reader));
 		}
@@ -360,17 +370,18 @@ internal static TimeSpan BucketFor(DateTime fromLocal, DateTime toLocal, int tar
 		int[] penIds,
 		DateTime fromLocal,
 		DateTime toLocal,
-		int targetColumnCount)
+		int targetColumnCount,
+		CancellationToken cancellationToken)
 	{
 		await using var command = new NpgsqlCommand(ArchiveStatements.BucketedRawWindow, connection);
 
 		BindLocalBucketedWindow(command, penIds, fromLocal, toLocal, targetColumnCount);
 
-		await using var reader = await command.ExecuteReaderAsync().ConfigureAwait(false);
+		await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 
 		var rows = new List<BucketedRowFold.Row>();
 
-		while (await reader.ReadAsync().ConfigureAwait(false))
+		while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 		{
 			rows.Add(ReadBucketedRow(reader));
 		}
@@ -384,7 +395,8 @@ internal static TimeSpan BucketFor(DateTime fromLocal, DateTime toLocal, int tar
 		DateTime fromLocal,
 		DateTime toLocal,
 		AggregationLayer layer,
-		IReadOnlyList<HistoryRowFold.Row> coarseRows)
+		IReadOnlyList<HistoryRowFold.Row> coarseRows,
+		CancellationToken cancellationToken)
 	{
 		var seams = FreshTail.Seams(coarseRows, penIds, fromLocal);
 
@@ -393,7 +405,13 @@ internal static TimeSpan BucketFor(DateTime fromLocal, DateTime toLocal, int tar
 			return coarseRows;
 		}
 
-		var tailRows = await ReadWindowAsync(connection, penIds, tailStart, toLocal, AggregationLayer.Raw)
+		var tailRows = await ReadWindowAsync(
+				connection,
+				penIds,
+				tailStart,
+				toLocal,
+				AggregationLayer.Raw,
+				cancellationToken)
 			.ConfigureAwait(false);
 
 		return FreshTail.Merge(coarseRows, tailRows, seams, tailStart);

SemiPlot/SemiPlot.Tests.Integration/PostgresHistoryReadTests.cs

@@ -129,6 +129,14 @@ public sealed class PostgresHistoryReadTests(
 	// The archive's own code for the layer the coarse rows below belong to.
 	private const short MinuteLayer = (short)AggregationLayer.Minute;
 
+	private const string LockTrendsCommand = "LOCK TABLE public.trends IN ACCESS EXCLUSIVE MODE;";
+
+	private const string LockWaitOnTrendsQuery =
+		"SELECT count(*) FROM pg_locks WHERE NOT granted AND relation = 'public.trends'::regclass;";
+
+	// Far above the round trip a cancel request takes, so only a read the token never reached runs into it.
+	private static readonly TimeSpan _cancelledReadDeadline = TimeSpan.FromSeconds(20);
+
 	[Fact]
 	public async Task AWindowInsideTheFirstRunReadsTheSeedersOwnRows()
 	{
@@ -206,6 +214,50 @@ public async Task AWindowBeforeTheArchiveStartsIsASuccessfulEmptyList()
 		result.Value.Should().BeEmpty();
 	}
 
+	// The lock holds the statement on the server, so the cancel lands mid-statement and the server answers
+	// 57014. Raw reaches the bucketed window, Minute the sparse one.
+	[Theory]
+	[InlineData(AggregationLayer.Raw)]
+	[InlineData(AggregationLayer.Minute)]
+	public async Task AReadCancelledMidStatementThrowsInsteadOfFailing(AggregationLayer layer)
+	{
+		var testCancellation = TestContext.Current.CancellationToken;
+		await using var database = await postgresContainerFixture.CloneProvisionedAsync(testCancellation);
+		await using var locker = new NpgsqlConnection(database.AdminConnectionString);
+		await locker.OpenAsync(testCancellation);
+		await using var transaction = await locker.BeginTransactionAsync(testCancellation);
+		await using var lockTrends = new NpgsqlCommand(LockTrendsCommand, locker, transaction);
+		await lockTrends.ExecuteNonQueryAsync(testCancellation);
+
+		var window = QuietWindow();
+		using var cancellation = new CancellationTokenSource();
+		await using var services = ArchiveProviderFactory.Build(database.PlotConnectionString);
+
+		var read = services.GetRequiredService<IDataProvider>().QueryHistoryAsync(
+			_seededPenIds.Value,
+			_timeConverter.ToUtc(window.From),
+			_timeConverter.ToUtc(window.To),
+			layer,
+			ColumnTargetFor(window),
+			cancellation.Token);
+
+		try
+		{
+			(await AwaitALockWaitOnTrendsAsync(locker, transaction)).Should().BeTrue(
+				"the read has to be waiting on the server when the token is cancelled");
+
+			await cancellation.CancelAsync();
+
+			var settled = () => read.WaitAsync(_cancelledReadDeadline, testCancellation);
+
+			await settled.Should().ThrowAsync<OperationCanceledException>();
+		}
+		finally
+		{
+			await transaction.RollbackAsync(testCancellation);
+		}
+	}
+
 	// A break writes no rows at all, and the fold turns the q = 32 row bounding it into one NaN anchor a
 	// tick later. Counting the anchors rather than finding one is the whole instrument.
 	[Fact]
@@ -568,6 +620,26 @@ public void TheExpectedSeedLookBackIsTheStatementsOwn()
 		SeedLookBackFor(wideWindow).Should().Be(wideWindow.To - wideWindow.From);
 	}
 
+	private static async Task<bool> AwaitALockWaitOnTrendsAsync(
+		NpgsqlConnection locker,
+		NpgsqlTransaction transaction)
+	{
+		await using var command = new NpgsqlCommand(LockWaitOnTrendsQuery, locker, transaction);
+		var deadline = DateTime.UtcNow + _cancelledReadDeadline;
+
+		while (DateTime.UtcNow < deadline)
+		{
+			if (await command.ExecuteScalarAsync(TestContext.Current.CancellationToken) is > 0L)
+			{
+				return true;
+			}
+
+			await Task.Delay(TimeSpan.FromMilliseconds(20), TestContext.Current.CancellationToken);
+		}
+
+		return false;
+	}
+
 	// The archive's first two minutes: inside the first archiving run, before any break.
 	private static LocalWindow QuietWindow()
 	{

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

@@ -67,8 +67,9 @@ internal sealed class FakeDataProvider(
 	// (e.g. an initial Raw load vs. a superseding coarser-layer gesture re-query).
 	public Dictionary<AggregationLayer, double> LayerCenterOverrides { get; } = [];
 
-	// Plain, for the reason stated at PensGate: TrendChartViewModelTests asserts inline after SetResult.
-	public TaskCompletionSource<Result<IReadOnlyList<PenHistoryEnvelope>>> HistoryGate { get; } = new();
+	// Plain, for the reason stated at PensGate: TrendChartViewModelTests asserts inline after SetResult. A held
+	// read waits on the gate in place when it starts, so replacing the gate holds the next read apart from it.
+	public TaskCompletionSource<Result<IReadOnlyList<PenHistoryEnvelope>>> HistoryGate { get; set; } = new();
 
 	// Pen identifiers the history read answers with no envelope at all, the shape a real provider returns
 	// for a pen holding no row in the window. A requested pen missing from the result is not an error.
@@ -86,6 +87,8 @@ internal sealed class FakeDataProvider(
 
 	public int? LastQueriedTargetColumnCount { get; private set; }
 
+	public CancellationToken LastQueriedCancellationToken { get; private set; }
+
 	public int PensQueryCount { get; private set; }
 
 	public int ExtentQueryCount { get; private set; }
@@ -112,7 +115,7 @@ internal sealed class FakeDataProvider(
 	// 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.Start
+	// 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.
 	public Exception? RealtimeStreamFailure { get; set; }
 
@@ -205,14 +208,17 @@ public Task<Result<IReadOnlyList<PenHistoryEnvelope>>> QueryHistoryAsync(
 		DateTime fromUtc,
 		DateTime toUtc,
 		AggregationLayer layer,
-		int targetColumnCount)
+		int targetColumnCount,
+		CancellationToken cancellationToken = default)
 	{
-		HistoryQueryCount++;
+		// The count moves last: a test that waits for it on another thread then reads the query it counts.
 		LastQueriedPenIds = penIds;
 		LastQueriedLayer = layer;
 		LastQueriedFromUtc = fromUtc;
 		LastQueriedToUtc = toUtc;
 		LastQueriedTargetColumnCount = targetColumnCount;
+		LastQueriedCancellationToken = cancellationToken;
+		HistoryQueryCount++;
 
 		if (FailHistory)
 		{
@@ -221,7 +227,7 @@ public Task<Result<IReadOnlyList<PenHistoryEnvelope>>> QueryHistoryAsync(
 
 		if (GatedLayer == layer)
 		{
-			return HistoryGate.Task;
+			return HistoryGate.Task.WaitAsync(cancellationToken);
 		}
 
 		var center = LayerCenterOverrides.TryGetValue(layer, out var overridden)

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

@@ -72,16 +72,24 @@ public void RealtimeBatch_KeepsEachPenOnItsOwnTimestamps()
 	}
 
 	[Fact]
-	public async Task QueryHistoryAsync_PassesThroughToProviderWithTargetColumnCount()
+	public async Task QueryHistoryAsync_PassesThroughToProviderWithTargetColumnCountAndToken()
 	{
 		var (coordinator, _, provider) = CreateCoordinator();
+		using var cancellation = new CancellationTokenSource();
 
-		var result = await coordinator.QueryHistoryAsync([1, 2], _from, _to, AggregationLayer.Minute, 256);
+		var result = await coordinator.QueryHistoryAsync(
+			[1, 2],
+			_from,
+			_to,
+			AggregationLayer.Minute,
+			256,
+			cancellation.Token);
 
 		result.IsSuccess.Should().BeTrue();
 		result.Value.Select(pen => pen.PenId).Should().Equal(1, 2);
 		provider.LastQueriedLayer.Should().Be(AggregationLayer.Minute);
 		provider.LastQueriedTargetColumnCount.Should().Be(256);
+		provider.LastQueriedCancellationToken.Should().Be(cancellation.Token);
 	}
 
 	[Fact]