test: subscription drop/resubscribe + health; fix relational end-of-stream measure (#308, #548)#551
Merged
Conversation
Contributor
PR Summary by QodoAdd subscription drop/resubscribe tests with health-check assertions
AI Description
Diagram
High-Level Assessment
Files changed (5)
|
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 82f9fee08f
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| // 6. Events produced after recovery must be processed. | ||
| var countBeforeRecovery = fixture.Handler.Count; | ||
| await GenerateAndHandleCommands(BatchSize); | ||
| var resumed = await WaitUntil(() => fixture.Handler.Count >= countBeforeRecovery + BatchSize, DropTimeout, cancellationToken); |
There was a problem hiding this comment.
In the paused-container scenario, the initial five events may not have been flushed to the checkpoint before the drop (the default checkpoint batch/delay is much larger than this batch), and resubscribe reports healthy before any backlog replay completes. If that happens, replaying the same initial events can advance Handler.Count by BatchSize and satisfy this wait before the events generated on the previous line are ever consumed, so the test can pass while recovery is still not processing post-recovery writes. Please wait on distinct event identities or force/await the checkpoint before dropping.
Useful? React with 👍 / 👎.
Contributor
Author
There was a problem hiding this comment.
Fixed in fafd6c7. Before pausing, the test now waits until the initial batch is committed to the checkpoint (polling CheckpointStore.GetLastCheckpoint against GetLastPosition), so on resubscribe there is nothing to replay and the post-recovery count can only advance for genuinely new events.
Contributor
Author
There was a problem hiding this comment.
Update: the checkpoint-await approach in fafd6c7 timed out for the KurrentDB $all subscription (GetLastPosition reads the $all head, which includes system events the subscription skips, so its checkpoint never reaches that position). Replaced in 1f4457c with your primary suggestion instead — the test now waits until the specific post-recovery events are handled, by record identity, so replayed initial events can't satisfy the assertion. Verified locally against Postgres.
Contributor
Code Review by Qodo
1. Dropped health uses Unhealthy
|
![qodo-free-for-open-source-projects[bot]](https://avatars.githubusercontent.com/in/1068000?s=60&v=4){kind=small}
Comment on lines
+47
to
+58
| protected internal ValueTask StartSubscription() | ||
| => Subscription.Subscribe( | ||
| id => { | ||
| Health.ReportHealthy(id); | ||
| Log.LogInformation("{Subscription} subscribed", id); | ||
| }, | ||
| (id, reason, ex) => { | ||
| Health.ReportUnhealthy(id, ex); | ||
| Log.LogWarning(ex, "{Subscription} dropped {Reason}", id, reason); | ||
| }, | ||
| CancellationToken.None | ||
| ); |
Contributor
There was a problem hiding this comment.
2. Health check data race 🐞 Bug ☼ Reliability
SubscriptionHealthCheck uses a mutable Dictionary that is iterated in CheckHealthAsync while
ReportHealthy/ReportUnhealthy can mutate it from subscription callbacks; concurrent access can throw
InvalidOperationException ("Collection was modified") or produce inconsistent results. The new drop
tests poll health repeatedly during drop/resubscribe, increasing the likelihood of this crash, and
the same health check is registered as a singleton in production wiring as well.
Agent Prompt
### Issue description
`SubscriptionHealthCheck` stores subscription health in a plain `Dictionary<string, HealthReport>`. `CheckHealthAsync` enumerates that dictionary while `ReportHealthy/ReportUnhealthy` mutate it, which can throw at runtime under normal concurrent usage.
This PR newly wires subscription callbacks to `SubscriptionHealthCheck` in `SubscriptionFixtureBase` and the new `SubscriptionDropBase` polls health while the subscription is transitioning, making the race much more likely (and flaky).
### Issue Context
- Subscription callbacks run on background threads.
- Health checks can be called concurrently (e.g., by ASP.NET health endpoint / test polling).
### Fix Focus Areas
- src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionHealth.cs[14-43]
- src/Core/test/Eventuous.Tests.Subscriptions.Base/Fixtures/SubscriptionFixtureBase.cs[47-58]
- src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionDropBase.cs[49-63]
### Suggested fix
- Protect `_healthReports` with a lock for both mutation and enumeration **or** switch to `ConcurrentDictionary` and enumerate a snapshot (e.g., `foreach (var report in _healthReports.ToArray())`).
- Ensure `CheckHealthAsync` cannot throw due to concurrent updates.
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Contributor
Author
There was a problem hiding this comment.
Fixed in fafd6c7. SubscriptionHealthCheck now stores reports in a ConcurrentDictionary, so CheckHealthAsync enumeration no longer races the ReportHealthy/ReportUnhealthy callbacks.
Comment on lines
+44
to
+72
| // 2. Drop the connection by pausing the container. | ||
| WriteLine("Pausing the container to drop the connection"); | ||
| await fixture.Container.PauseAsync(cancellationToken); | ||
|
|
||
| // 3. The subscription must detect the drop and report unhealthy. | ||
| var dropped = await WaitUntil(() => fixture.IsDropped, DropTimeout, cancellationToken); | ||
| await Assert.That(dropped).IsTrue(); | ||
| var unhealthy = await WaitUntil(async () => await GetHealthStatus(cancellationToken) == HealthStatus.Unhealthy, DropTimeout, cancellationToken); | ||
| await Assert.That(unhealthy).IsTrue(); | ||
| WriteLine("Subscription dropped and reported unhealthy"); | ||
|
|
||
| // 4. Restore the connection. | ||
| WriteLine("Unpausing the container to restore the connection"); | ||
| await fixture.Container.UnpauseAsync(cancellationToken); | ||
|
|
||
| // 5. The subscription must resubscribe and report healthy again. | ||
| var recovered = await WaitUntil(() => !fixture.IsDropped, DropTimeout, cancellationToken); | ||
| await Assert.That(recovered).IsTrue(); | ||
| var healthy = await WaitUntil(async () => await GetHealthStatus(cancellationToken) == HealthStatus.Healthy, DropTimeout, cancellationToken); | ||
| await Assert.That(healthy).IsTrue(); | ||
| WriteLine("Subscription resubscribed and reported healthy"); | ||
|
|
||
| // 6. Events produced after recovery must be processed. | ||
| var countBeforeRecovery = fixture.Handler.Count; | ||
| await GenerateAndHandleCommands(BatchSize); | ||
| var resumed = await WaitUntil(() => fixture.Handler.Count >= countBeforeRecovery + BatchSize, DropTimeout, cancellationToken); | ||
|
|
||
| await fixture.StopSubscription(); | ||
|
|
Contributor
There was a problem hiding this comment.
3. Paused container not restored 🐞 Bug ☼ Reliability
SubscriptionDropBase pauses the infrastructure container and stops the subscription only on the success path; if an assertion fails or the test is cancelled between pause and unpause, the container can remain paused and the subscription can remain running, contaminating subsequent tests. This is amplified here because the new drop tests construct fixtures with autoStart=false, so fixture disposal won’t automatically stop the subscription.
Agent Prompt
### Issue description
`ShouldResubscribeAfterConnectionDrop` performs destructive operations (pause container) but does not guarantee cleanup (unpause + unsubscribe) if the test fails mid-flight. This can leave the container paused and/or a subscription resubscribe loop running, causing cascading failures and hangs in later tests.
### Issue Context
The drop tests pass `autoStart=false` into the fixture constructors, and `SubscriptionFixtureBase.DisposeAsync` only calls `StopSubscription()` when `_autoStart` is true, so relying on fixture disposal is insufficient.
### Fix Focus Areas
- src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionDropBase.cs[36-75]
- src/Core/test/Eventuous.Tests.Subscriptions.Base/Fixtures/SubscriptionFixtureBase.cs[95-103]
### Suggested fix
- Wrap the pause/unpause + subscription start/stop flow in `try/finally`.
- In `finally`:
- Attempt to `UnpauseAsync` if the container was paused (use `CancellationToken.None` so cleanup still runs even when the test token is cancelled).
- Call `fixture.StopSubscription()` if it was started.
- Optionally harden `SubscriptionFixtureBase.DisposeAsync` to stop the subscription if it is running even when `_autoStart` is false (defensive cleanup).
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Contributor
Author
There was a problem hiding this comment.
Fixed in fafd6c7. The pause/subscription flow is wrapped in try/finally; the finally unpauses the container (with CancellationToken.None) and stops the subscription if either is still active, so a failed assertion or cancellation cannot leak a paused container or a resubscribing subscription into later tests.
alexeyzimarev
changed the title
Test Results 46 files + 24 46 suites +24 12m 49s ⏱️ - 3m 6s Results for commit 1f4457c. ± Comparison against base commit e676950. This pull request removes 5 and adds 14 tests. Note that renamed tests count towards both.♻️ This comment has been updated with latest results. |
Add an infra-agnostic `SubscriptionDropBase` that simulates a transport drop by pausing the infrastructure container, then asserts the subscription is dropped and the health check reports Unhealthy, unpauses, and asserts resubscribe, Healthy again, and continued processing of new events. Wire the subscription's subscribed/dropped callbacks in the shared fixture to a `SubscriptionHealthCheck` (mirroring `SubscriptionHostedService`) and expose `IsDropped` so tests can observe drop state. Covered for the container-backed stores: SQL Server, PostgreSQL, and KurrentDB (ESDB). SQLite is excluded as its fixture isn't container-backed. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
`SqlSubscriptionBase.GetSubscriptionEndOfStream` had three defects that made the gap/lag diagnostics log "Failed to get end of stream" repeatedly: - Swapped switch arms: an All subscription queried MAX(stream_position) and a Stream subscription queried MAX(global_position). - `reader.GetInt64(0)` threw `InvalidCastException` when the provider returned the position column as Int32. - No DBNull guard, so MAX(...) over an empty table threw. Query the correct column per kind, guard against DBNull, and use `Convert.ToInt64` so either integer width is accepted. Also log the actual exception instead of swallowing it. Add `SubscriptionMeasureBase` plus SQL Server and PostgreSQL subclasses that assert the measure reports position 0 on an empty store and the global end position after events are appended. Verified to fail against the unfixed code. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Address review findings on #551: - SubscriptionHealthCheck stored reports in a plain Dictionary that CheckHealthAsync enumerated while subscription callbacks mutated it from background threads — a real "Collection was modified" race on the production singleton. Switch to ConcurrentDictionary (Qodo). - SubscriptionDropBase now waits until the initial batch is committed to the checkpoint before pausing, so replayed events can't spuriously satisfy the post-recovery assertion (Codex). - Wrap the pause/subscription flow in try/finally so a failed assertion or cancellation can't leave the container paused or the subscription resubscribing into later tests — these fixtures use autoStart=false, so disposal won't stop the subscription (Qodo). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
alexeyzimarev
force-pushed
the
test/subscription-drop-resubscribe-308
branch
from
43fbcae to
fafd6c7
Compare
![github-code-quality[bot]](https://avatars.githubusercontent.com/u/9919?s=60&v=4){kind=small}
Comment on lines
+106
to
+108
| } catch (Exception ex) { | ||
| WriteLine("Cleanup: failed to unpause the container: {0}", ex.Message); | ||
| } |
Comment on lines
+114
to
+116
| } catch (Exception ex) { | ||
| WriteLine("Cleanup: failed to stop the subscription: {0}", ex.Message); | ||
| } |
…sition The checkpoint-await added in the previous commit timed out for the KurrentDB $all drop test: GetLastPosition reads the $all head, which includes system events the AllStreamSubscription skips, so its committed checkpoint never reaches that position. Relational stores passed only because their GetLastPosition reads the user-event table. Replace it with the provider-agnostic form of the same guard: capture the specific events produced after recovery and wait until the handler has handled those exact events (by record identity). A replay of the initial batch on resubscribe can no longer satisfy the assertion. TestEventHandler now exposes handled messages via a concurrent queue. Verified: Postgres_ShouldResubscribeAfterConnectionDrop passes locally. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Adds subscription test coverage requested in #308 and fixes the relational end-of-stream measure bug reported in #548.
#308 — drop/resubscribe + health
Adds an infra-agnostic base test,
SubscriptionDropBase, that verifies a subscription drops and resubscribes when the underlying infrastructure connection is lost and restored, and that the health check transitions accordingly. Follows the existingHandlerFailureBaseconvention (one base + a thin per-store subclass).Flow: produce/consume events (assert Healthy) →
docker pausethe container to drop the connection → assert subscription dropped and Unhealthy →docker unpause→ assert resubscribe and Healthy → produce more events and assert they are processed.SqlServer_ShouldResubscribeAfterConnectionDropPostgres_ShouldResubscribeAfterConnectionDropEsdb_ShouldResubscribeAfterConnectionDrop([Retry(3)])SQLite is excluded — its fixture isn't container-backed, so there's no transport to drop.
SubscriptionFixtureBasenow wires the subscribed/dropped callbacks to aSubscriptionHealthCheck(mirroringSubscriptionHostedService) and exposesIsDropped. Backward-compatible — existing handler-failure tests still pass.Note: the health check only exposes Healthy/Unhealthy (no
Degraded), so the test assertsUnhealthyon drop andHealthyon recovery.#548 —
GetSubscriptionEndOfStreamfixSqlSubscriptionBase.GetSubscriptionEndOfStreamhad three defects that made the gap/lag diagnostics log"Failed to get end of stream"repeatedly:Allsubscription queriedMAX(stream_position)and aStreamsubscription queriedMAX(global_position).GetInt64(0)threwInvalidCastExceptionwhen the provider returned the position column asInt32.DBNullguard —MAX(...)over an empty table threw.Fixed by querying the correct column per kind, guarding against
DBNull, and usingConvert.ToInt64(accepts either integer width). The catch block now logs the actual exception instead of swallowing it.Added
SubscriptionMeasureBase+ SQL Server/PostgreSQL subclasses asserting the measure returns position0on an empty store and the global end position after events are appended.Verification
Ran locally (net10.0) against real containers — all passing:
HandlerFailureResubscribe(SQL Server + PostgreSQL) ✅ — no regression🤖 Generated with Claude Code