fix(streaming): release ADO.NET messages during queue handoff

[ReubenBond]

Fixes #10458.

ADO.NET stream rows dequeued by a pulling agent remain invisible until the visibility timeout if that agent relinquishes queue ownership before confirming them. A successor can therefore register a cold stream from a later row and miss the stream's first item within the test timeout. Sorting dequeue results in #10536 fixed within-poll ordering, but it could not recover rows still hidden by the previous owner.

This change tracks each receiver's unconfirmed dequeue receipts and releases those rows during shutdown so the successor can redeliver them immediately. Release operations match both message ID and dequeue receipt, preventing a stale receiver from changing a row already claimed by a new owner. The existing confirmation query contract carries the release marker, preserving startup compatibility with older database schemas.

The pulling agent now also prevents a later queue read from overtaking in-flight cold-stream registration and keeps the first cached batch pinned until subscriber cursors attach. This closes the related race where producer registration and subscriber notification cross, causing the cursor to begin at item 2 even though item 1 was successfully enqueued. Pins are disposed when subscriptions attach, streams are removed, or cache pressure requires progress.

The follow-up repair restores the concrete RelationalOrleansQueries receiver constructor expected by ActivatorUtilities, while retaining the injectable query abstraction used by deterministic lifecycle tests.

Cross-provider regressions cover immediate release and stale-receipt safety for SQL Server, PostgreSQL, and MySQL. Streaming timeout failures report the stream, provider, expected item count, and observed item count.

Microsoft Reviewers: Open in CodeFlow

[Copilot]

Pull request overview

This PR addresses a flaky ADO.NET streaming scenario where dequeued rows can remain hidden until the visibility timeout when a pulling agent relinquishes queue ownership before confirming. It adds explicit “release” support (keyed by message id + dequeue receipt) so a successor agent can immediately redeliver those rows, and strengthens regressions/diagnostics across SQL Server, PostgreSQL, and MySQL.

Changes:

• Add receiver-side tracking of unconfirmed dequeues and release those rows during receiver shutdown using the existing confirmation stored query contract.
• Extend SQL Server/PostgreSQL/MySQL confirmation procedures to interpret negative dequeue receipts as “release for immediate redelivery” (while preventing stale releases via receipt matching).
• Add cross-provider regression tests and improve streaming timeout diagnostics with observed-vs-expected counts.

Show a summary per file

File	Description
test/TestInfrastructure/TestExtensions/Diagnostics/StreamingDiagnosticObserver.cs	Adds richer cancellation context when waiting for item delivery counts.
test/Extensions/Orleans.AdoNet.Tests/Streaming/RelationalOrleansQueriesTests.cs	Adds query-level regression for releasing by dequeue receipt.
test/Extensions/Orleans.AdoNet.Tests/Streaming/AdoNetQueueAdapterReceiverTests.cs	Adds receiver shutdown regression ensuring unconfirmed messages are released.
src/AdoNet/Shared/Storage/RelationalOrleansQueries.cs	Adds ReleaseStreamMessagesAsync and shared confirmation formatting for release vs confirm.
src/AdoNet/Orleans.Streaming.AdoNet/SQLServer-Streaming.sql	Adds release path to ConfirmStreamMessages when dequeue receipt is negative.
src/AdoNet/Orleans.Streaming.AdoNet/PostgreSQL-Streaming.sql	Adds release path to confirmation function when dequeue receipt is negative.
src/AdoNet/Orleans.Streaming.AdoNet/MySQL-Streaming.sql	Supports negative receipts (SIGNED), matches via ABS, and adds release update path.
src/AdoNet/Orleans.Streaming.AdoNet/AdoNetQueueAdapterReceiver.cs	Tracks pending receipts and releases unconfirmed rows on shutdown.

Review details

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

Suppressed comments (2)

src/AdoNet/Orleans.Streaming.AdoNet/AdoNetQueueAdapterReceiver.cs:162

• The confirmation-failure log is tagged with ClusterId, but the log message template and other receiver logs use the service id. This makes cross-silo/provider troubleshooting harder because it reports the wrong identifier.

            catch (Exception ex)
            {
                LogConfirmationFailed(ex, clusterOptions.ClusterId, providerId, queueId, items);
                throw;

test/TestInfrastructure/TestExtensions/Diagnostics/StreamingDiagnosticObserver.cs:214

• Same as above: actualCount++ is not thread-safe for a DiagnosticListener-backed observable and can misreport the observed count under concurrent event delivery. Use an atomic increment.

                .OfType<StreamingEvents.ItemDelivered>()
                .Where(e => MatchesSubscription(e.StreamId, e.SubscriptionId, e.StreamProvider, streamId, subscriptionId, streamProvider))
                .Do(_ => actualCount++)
                .Take(expectedCount)

• Files reviewed: 8/8 changed files
• Comments generated: 1
• Review effort level: Lite

[Copilot]

Review details

Suppressed comments (2)

src/AdoNet/Orleans.Streaming.AdoNet/AdoNetQueueAdapterReceiver.cs:52

• Shutdown waits a captured copy of _outstandingTask, but then re-checks the _outstandingTask field and can return early even when the task has completed (eg, if the async continuation which clears the field hasn't run yet). That race would skip releasing pending messages during queue handoff, which is the scenario this PR is addressing.

        RemoveExpiredPendingMessages();
        if (_outstandingTask is not null || _pendingMessages.Count == 0)
        {
            return;
        }

src/AdoNet/Orleans.Streaming.AdoNet/AdoNetQueueAdapterReceiver.cs:160

• This catch block logs clusterOptions.ClusterId as the ServiceId argument in LogConfirmationFailed, which will produce misleading identifiers in logs when confirmations fail.

            catch (Exception ex)
            {

• Files reviewed: 8/8 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Copilot]

Review details

• Files reviewed: 8/8 changed files
• Comments generated: 0 new
• Review effort level: Lite

[ReubenBond]

CI appears to be exposing a PR-related regression rather than an infrastructure-only flake. In the failed run, every ADO.NET provider matrix failed across PostgreSQL, MySQL/MariaDB, and SQL Server on both net8.0 and net10.0. The failures are widespread streaming timeouts/zero deliveries; several report that the pulling-agent manager started its assigned queue but never initialized it (StartedQueues=[AdoNet:1], InitializedQueues=[]).

Examples include the stream filtering, batching, queue-adapter, client-stream, and general streaming suites. The two macOS BVT/Functional failures look unrelated, but the consistent six-job ADO.NET failure pattern aligns with the queue handoff changes and needs investigation before retrying the full run.

[ReubenBond]

Pushed follow-up repair 2cbafa408.

Root regression from run 31819110261 was the receiver constructor change: ActivatorUtilities could not find the concrete RelationalOrleansQueries signature, so every provider failed queue initialization. The repair restores that factory-compatible constructor and adds a provider-independent type-initialization regression.

Stress testing then reproduced the original delivery flake on MySQL/net10: all 10 rows were enqueued, but the consumer received #2 through #10 and timed out at 9/10. The pulling agent now prevents a later queue read from overtaking in-flight cold-stream registration and retains the first cache pin until subscriber cursors attach, with removal/cache-pressure cleanup.

Validation on the pushed head:

• PersistentStreamPullingAgentTests focused registration/shutdown cases: 3/3 on net8.0 and 3/3 on net10.0.
• ADO.NET receiver lifecycle/factory cases: 2/2 on net8.0 and 2/2 on net10.0.
• Exact issue tests (AdoNet_16 and AdoNet_17) final spot check: 2/2 in each PostgreSQL, MySQL, and SQL Server matrix on net8.0 and net10.0.
• Repetition before the final spot check: 5/5 iterations per provider on net8.0; PostgreSQL and SQL Server 5/5 on net10.0; MySQL net10.0 10/10 plus 5/5 and 3/3 after successive hardening steps.
• Full streaming suites: PostgreSQL 64 passed/2 skipped and SQL Server 64 passed/2 skipped on both frameworks; MySQL 65 passed/2 skipped on both frameworks, including a final net10.0 full-suite pass.

Remaining uncertainty: local runs cannot reproduce GitHub-hosted runner scheduling exactly, so the newly triggered six provider CI matrices remain the authoritative merge gate.

[Copilot]

Review details

Suppressed comments (2)

Previously missed (2) — in code that hasn't changed since the last review.

src/AdoNet/Orleans.Streaming.AdoNet/AdoNetQueueAdapterReceiver.cs:86

• Shutdown currently returns early if waiting for in-flight operations throws (for example due to timeout). That skips the new release step entirely, which can reintroduce the original handoff problem (messages remain invisible until visibility timeout). Consider logging the wait failure but still proceeding to best-effort release of any tracked pending receipts.

            catch (Exception ex)
            {
                LogShutdownFault(ex, clusterOptions.ServiceId, providerId, queueId);
                return;
            }

src/AdoNet/Shared/Storage/RelationalOrleansQueries.cs:556

• FormatStreamConfirmations can emit negative dequeue receipts when release=true, but the comment implies the format is always positive ("1:2|..."). Clarifying this helps future maintainers understand the protocol contract.

        // Builds a provider-neutral receipt list in the form "1:2|3:4|5:6".

• Files reviewed: 11/11 changed files
• Comments generated: 1
• Review effort level: Lite

[Copilot]

Review details

Suppressed comments (1)

Previously missed (1) — in code that hasn't changed since the last review.

src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs:522

• The global cold-stream registration gate returns early before cache maintenance runs. When any RegistrationTask is in-flight, ReadFromQueue exits before calling queueCache.TryPurgeFromCache()/MessagesDeliveredAsync and before applying backpressure handling, which can delay confirmations/deletes for already-delivered batches and keep the cache artificially “stuck” until registration completes.

Consider moving this gate to just before rcvr.GetQueueMessagesAsync(...) so it pauses new dequeues without skipping purge/confirm and pressure handling for already-cached items.

            // Pause all queue reads so a cold stream's first batch stays pinned until registration completes.
            if (pubSubCache.Values.Any(static stream => stream.RegistrationTask is { IsCompleted: false }))
            {
                return false;
            }

• Files reviewed: 11/11 changed files
• Comments generated: 0 new
• Review effort level: Lite

[ReubenBond]

The fresh CI run on head

[ReubenBond]

The fresh CI run on head 2cbafa408 exposes a remaining PR-related Redis streaming regression. In attempt 1, the same four RedisStreamingResumeTests failed on both net8.0 and net10.0: ResumeAfterInactivityNotInCache, ResumeAfterDeactivation, ResumeAfterDeactivationActiveStream, and ResumeAfterInactivity. The failures are delivery timeouts in the resume paths.

A targeted rerun reproduced both Redis matrix failures in attempt 2, while the unrelated SQL reminder failure cleared. Since the failures are deterministic across both frameworks and align with this PR's pulling-agent/cache registration changes, they need to be fixed before merging rather than retried again.

[ReubenBond]

Pushed follow-up repair \c8b9a67f2\ for the deterministic Redis resume failures. The registration read gate exposed two stale-token paths: cold-stream subscription setup and post-deactivation delivery realignment could request a token which had already left the local queue cache. Both now resume from the receiver's first available batch instead of reporting a false cache miss or dropping that batch. I also removed the stream-level registration pin bookkeeping which could outlive registration and interfere with inactivity cleanup; the existing registration-scoped pin remains held through all initial subscriber handshakes. The five \RedisStreamingResumeTests\ now pass on both
et8.0\ and
et10.0, and the focused pulling-agent read tests pass.

[Copilot]

Review details

Suppressed comments (2)

Previously missed (2) — in code that hasn't changed since the last review.

src/AdoNet/Orleans.Streaming.AdoNet/AdoNetQueueAdapterReceiver.cs:86

• If waiting for in-flight operations exceeds the shutdown timeout, Shutdown logs and returns without attempting to release already-tracked pending messages. That can leave dequeued rows invisible until the visibility timeout, undermining the intent of releasing on handoff. Consider logging the timeout but continuing with a best-effort release of currently tracked pending messages.

            catch (Exception ex)
            {
                LogShutdownFault(ex, clusterOptions.ServiceId, providerId, queueId);
                return;
            }

src/AdoNet/Orleans.Streaming.AdoNet/MySQL-Streaming.sql:435

• The MySQL confirmation procedure switches to release mode if any item has a negative dequeue receipt, then updates every row in _Batch. If _Items contains a mix of positive (confirm) and negative (release) receipts, this would incorrectly "release" rows that were meant to be confirmed. Other providers only update rows whose receipt is negative and matches. Consider restricting the UPDATE to rows with negative receipts and matching dequeue counters.

IF EXISTS (SELECT 1 FROM _ItemsTable WHERE Dequeued < 0) THEN
    /* negative dequeue receipts release messages for immediate redelivery */
    UPDATE OrleansStreamMessage AS M
    INNER JOIN _Batch AS B
        ON M.ServiceId = B.ServiceId

• Files reviewed: 11/11 changed files
• Comments generated: 0 new
• Review effort level: Lite