[BUG] Streaming subscribe returns None after a reconnect, making test_streaming_subscribe_receives_published_message flaky

[CasperGN]

Expected Behavior

tests/integration/test_pubsub.py::test_streaming_subscribe_receives_published_message passes on every run, and Subscription.next_message() returns the next message even when the stream reconnects while it waits.

Actual Behavior

The test fails intermittently. It failed in 6 of the 25 most recent run-tests runs before #1229, on main pushes, merge-queue runs and unrelated PRs (for example runs 35790024069 on main, 35790699043 and 35791305700 in the merge queue). Latest occurrence: #1229, validate (3.14), runtime 1.18.4. Re-running the job on the same commit passed.

WARNING  dapr.clients.grpc.subscription:subscription.py:97 Subscription stream error (StatusCode.UNKNOWN): EOF — reconnecting
WARNING  dapr.clients.grpc.subscription:subscription.py:124 Can't send message on inactive stream: 'NoneType' object has no attribute 'id'
E           AttributeError: 'NoneType' object has no attribute 'data'
tests/integration/test_pubsub.py:113: AttributeError

Sidecar log for the same test:

Subscribing to pubsub 'pubsub' topic 'TOPIC_STREAM' ConnectionID 1
Unsubscribed from pubsub 'pubsub' topic 'TOPIC_STREAM'
error retrieving pending Redis messages: context canceled
Subscribing to pubsub 'pubsub' topic 'TOPIC_STREAM' ConnectionID 2
encountered a non-retriable error while publishing a subscribed message to topic TOPIC_STREAM, err: stream closed
Unsubscribed from pubsub 'pubsub' topic 'TOPIC_STREAM'

The first stream ends with EOF right after it connects. The SDK reconnects, and the runtime is about to deliver the message on connection 2. But next_message() has already returned None to the caller, so the test fails and its finally: subscription.close() closes connection 2 in the middle of that delivery.

Likely cause (SDK side)

1. next_message() returns None after a reconnect instead of reading from the new stream. In dapr/clients/grpc/subscription.py#L89-L102, the UNAVAILABLE / UNKNOWN / INTERNAL branch calls reconnect_stream() and falls through, so the method returns None. The docstring says None means "no message received within the timeout", but there is no timeout. A caller that does sub.next_message().data(), or iterates with for msg in sub, gets None or crashes on a transient reconnect. The async client does the same in dapr/aio/clients/grpc/subscription.py#L69-L84. It also does not treat INTERNAL as retryable, unlike the sync client.
2. Suspected, not yet confirmed: acks can be lost across a reconnect. The request generator of the old stream (subscription.py#L52-L56) blocks on self._send_queue.get() and checks _is_stream_active() only after get() returns. After start() makes the subscription active again, the old generator can take the next ack off the shared queue and yield it into the dead stream. That message is then never acked and is redelivered.

Why the first stream gets EOF right after it connects is a separate question. It may be on the runtime side.

Related: project coverage changes between runs of the same code

The same timing shows up in unit-test coverage. codecov/project uses threshold: 0%, so it fails a PR whenever these lines happen not to run. On #1229, which does not touch client.py, coverage dropped by 7 lines. They are the error and reconnect branches of subscribe_with_handler's stream_messages loop, dapr/clients/grpc/client.py#L641-L654. They ran on main at 03eebe1 and did not run on the PR head 2a11bb8.

test_subscribe_topic_with_handler (tests/clients/test_dapr_grpc_client.py#L411-L463) calls close_fn() and returns without waiting for the streaming thread. close_fn does not join it either. Whether the thread reaches the StreamInactiveError / reconnect branches before the test run ends depends on timing. A deterministic fix would make close_fn join the thread with a timeout, or have the test wait on it. A test that drives each branch on purpose would also help.

Steps to Reproduce the Problem

Run the integration suite repeatedly against runtime 1.18.4 with Redis pub/sub, e.g. uv run pytest tests/integration/test_pubsub.py -k streaming --count 50 (with pytest-repeat), or re-run validate in CI. The failure is timing-dependent.

Suggested fix

• After a successful reconnect, next_message() should read again from the new stream instead of returning None, with a bound on retries so a sidecar that keeps failing still surfaces an error. Same for the async client, including INTERNAL.
• Give each stream its own send queue, or stop the old generator on close(), so acks cannot go to a dead stream.
• Add unit tests for both clients against the fake sidecar: a stream that fails with UNKNOWN once, then delivers a message. next_message() returns that message, and the ack reaches the new stream.
• Keep the integration test as is, since it is what caught this.
• Make the subscribe_with_handler unit test deterministic, so that codecov/project stops changing between runs of the same code.

Release Note

RELEASE NOTE: FIX Streaming subscriptions return the next message after a transient reconnect instead of None.