fix(sql): scope stream subscription end-of-stream measure to the subscribed stream (#586) - #587
Conversation
…cribed stream (#586) For SubscriptionKind.Stream, the gap measure ran MAX(stream_position) over the whole messages table, so any longer stream inflated eventuous.subscription.gap.count for a fully caught-up subscription, and StartFrom=Latest could seed a stream subscription from another stream's position. Replace the constant GetEndOfStream/GetEndOfAll SQL properties on SqlSubscriptionBase with a PrepareEndOfStreamCommand(TConnection) hook, mirroring PrepareCommand, implemented per concrete subscription. Stream subscriptions filter by stream name through the streams table rather than the id resolved in BeforeSubscribe, so the measure is also valid when the metrics observer polls it before the subscription has connected. Tests: SubscriptionMeasureBase.ShouldMeasureEndOfSubscribedStream appends two streams of different lengths, subscribes to the shorter one, and asserts the measure reports its tail and the gap is zero after catch-up; wired for SQL Server, PostgreSQL and SQLite. SQLite also gets the all-stream measure test from #551. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
PR Summary by QodoScope SQL end-of-stream measures to subscribed streams
AI Description
Diagram
High-Level Assessment
Files changed (16)
|
Code Review by Qodo
1. Gap metric remains untested
|
![qodo-free-for-open-source-projects[bot]](https://avatars.githubusercontent.com/in/1068000?s=60&v=4){kind=small}
| var checkpoint = await fixture.CheckpointStore.GetLastCheckpoint(fixture.SubscriptionId, cancellationToken); | ||
| var caughtUp = await measure(cancellationToken); | ||
| await Assert.That(caughtUp.Position - checkpoint.Position!.Value).IsEqualTo(0ul); |
There was a problem hiding this comment.
1. Gap metric remains untested 📎 Requirement gap ☼ Reliability
The new relational regression tests manually subtract the measured end position from the checkpoint instead of observing eventuous.subscription.gap.count. They can pass even if metric registration, collection, tagging, or the metric-specific gap calculation is broken.
Agent Prompt
## Issue description
The relational regression coverage does not collect or assert the actual `eventuous.subscription.gap.count` metric; it only reproduces the expected subtraction directly.
## Issue Context
PR Compliance ID 6 requires SQL Server, PostgreSQL, and SQLite coverage proving the emitted caught-up gap metric is zero. Use the repository's established metric observation/exporter pattern and retain the different-length stream setup.
## Fix Focus Areas
- src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionMeasureBase.cs[70-73]
- src/Sqlite/test/Eventuous.Tests.Sqlite/Subscriptions/SubscriptionMeasureTests.cs[88-91]
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
There was a problem hiding this comment.
Addressed in 892789d. Added SubscriptionGapMetricsTestsBase.ShouldReportZeroGapWhenCaughtUp, which observes the exported eventuous.subscription.gap.count gauge through AddEventuousSubscriptions -> SubscriptionMetrics (the DI-registered measure, the checkpoint-commit listener, the gap arithmetic and the tags) instead of computing the gap from the delegate. It produces a longer unrelated stream next to the subscribed one, waits for the subscription to catch up, then polls the gauge until it reads 0. Wired for PostgreSQL, SQL Server, KurrentDB and SQLite (which gets a container-less MetricsFixture). Against the unfixed library the gauge reads 100 and the test fails; with the fix it converges on zero. The direct measure tests stay as the unit-level check of the PrepareEndOfStreamCommand contract.
Test Results 45 files ± 0 45 suites ±0 14m 15s ⏱️ +23s Results for commit c5615b7. ± Comparison against base commit a978e32. This pull request removes 9 and adds 18 tests. Note that renamed tests count towards both.♻️ This comment has been updated with latest results. |
…riptions (#586) The measure tests call the subscription's GetSubscriptionEndOfStream delegate directly and compute the gap themselves, so a regression in meter registration, the DI-registered measure, checkpoint-commit tracking, tagging or the gap arithmetic in SubscriptionMetrics would go unnoticed. Add SubscriptionGapMetricsTestsBase, which produces a longer unrelated stream next to the subscribed one, waits for the subscription to catch up, then polls the exported eventuous.subscription.gap.count gauge until it reads zero (checkpoints commit on a batch or delay) and asserts its tags. Wired for PostgreSQL, SQL Server, KurrentDB and SQLite; SQLite gets a container-less MetricsFixture on SqliteStoreFixtureBase and the existing gap-count test for parity. TestHandler is made public so the SQLite fixture can register it. Against the unfixed library the gauge reads 100 (other stream tail 199 minus checkpoint 99) and the test fails; with the fix it converges on zero. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
… stuck in queue Run 32986025539 for 892789d sat in 'queued' with no jobs for over three hours and the API refuses to cancel or re-run it; the CodeQL run for the same head ended in startup_failure and cannot be retried either. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
![github-code-quality[bot]](https://avatars.githubusercontent.com/u/9919?s=60&v=4){kind=small}
| .AddEventHandler<TestHandler>() | ||
| ); | ||
|
|
||
| services.AddOpenTelemetry().WithMetrics(builder => builder.AddEventuousSubscriptions().AddReader(new BaseExportingMetricReader(Exporter))); |
There was a problem hiding this comment.
Not applying this one — the reader is disposed. The analyzer is missing the ownership transfer through AddReader.
The chain, verified against the decompiled OpenTelemetry 1.15.3 assembly rather than assumed:
SqliteStoreFixtureBase.DisposeAsync()callsawait Provider.DisposeAsync(), which disposes the singletonMeterProvider.MeterProviderSdk.Dispose(bool)callsReader?.Shutdown(5000); Reader?.Dispose();(andcompositeMetricReader?.Dispose()).BaseExportingMetricReader.Dispose(bool)nullsIPullMetricExporter.Collect, then callsexporter.Dispose().
So both the reader and the exporter are released when the fixture tears down.
The genuinely redundant call is the existing Exporter.Dispose() in DisposeAsync, which runs after base.DisposeAsync() has already disposed the exporter through the reader. It is harmless — BaseExporter<T>.Dispose(bool) is an empty virtual and TestExporter doesn't override it, so that call is a no-op.
Holding the reader in a field would therefore duplicate a dispose the SDK already performs, and it would diverge from src/Diagnostics/test/Eventuous.Tests.OpenTelemetry/Fixtures/MetricsSubscriptionFixtureBase.cs, which this file mirrors: it carries the identical AddReader(new BaseExportingMetricReader(Exporter)) and Exporter.Dispose() pair, is untouched by this PR, and is the base for the KurrentDB, PostgreSQL and SQL Server metrics fixtures.
1 participant
Closes #586
Problem
For relational stream subscriptions (SQL Server, PostgreSQL, SQLite), the end-of-stream measure behind the subscription gap metric was a constant MAX(stream_position) over the whole messages table, with no stream filter. The checkpoint it is compared against is a position within the subscribed stream, so:
SubscriptionKind.All was unaffected.
Fix
Deviation from the issue suggestion: the stream filter uses the stream name from options rather than the stream id resolved in BeforeSubscribe. The measure is registered in DI and polled by SubscriptionMetrics from an observable gauge, which can happen before the subscription has ever connected; filtering by name makes the measure correct at any time instead of returning EndOfStream.Invalid until connect. A stream that does not exist yet reads as empty (position 0), consistent with the empty-store semantics established in #551.
Side effect: SQL Server and PostgreSQL now build the measure SQL from the resolved Schema (the same one PrepareCommand uses) instead of options.Schema, so it follows connection-options schema overrides like polling does.
Breaking for direct subclasses of SqlSubscriptionBase / the provider bases: GetEndOfStream and GetEndOfAll are removed in favour of PrepareEndOfStreamCommand. No references in eventuous-docs or eventuous-plugin.
Tests
Measure level (unit check of the PrepareEndOfStreamCommand contract):
Metric level (added after review, so a regression in the metrics pipeline itself is caught):
Verification
🤖 Generated with Claude Code