Interface IConsumerGroupPositioner
Moves this gateway pod's Redis consumer group for a topic to the stream's tail right before the gateway opens its Dapr subscription for that topic, so a new subscription only receives messages published after it opened.
public interface IConsumerGroupPositionerRemarks
Dapr's Redis pub/sub uses one consumer group per sidecar
(consumerID: {podName}, see the helm pubsub component) and the group
outlives the subscription. When the gateway reopens a topic after its last
caller left, Dapr resumes the group at its last-delivered-id; a group
it doesn't have yet it creates at the stream's start ("0", hard-coded
in components-contrib, with no per-subscription override). Both hand the new
subscriber whatever is still in the stream, and Dapr's reclaim loop
redelivers the group's unacked entries too. streamTTL doesn't bound
this between runs: it trims only when something publishes to the topic.
Observed live: a Binance order-book worker created at 15:31 was fed
11:16 WebSocket diffs and spent its first minutes on hours-old backlog.
The policy is no catch-up at all: whatever was published while this pod had no subscription for the topic is not delivered, including during the brief gap of a worker reload.
Methods
AckAbandonedPendingAsync(IReadOnlyCollection<string>, TimeSpan, CancellationToken)
Ack this pod's consumer-group entries on topics that have
been pending at least minIdle. Returns how many were acked.
Never throws; failures are logged.
Task<int> AckAbandonedPendingAsync(IReadOnlyCollection<string> topics, TimeSpan minIdle, CancellationToken cancellationToken)Parameters
topicsIReadOnlyCollection<string>minIdleTimeSpancancellationTokenCancellationToken
Returns
Remarks
The gateway's handler always answers Dapr Success right after fanning out,
so a live entry is pending for microseconds. An entry pending for seconds was
taken by a closed subscription's reader, which on a busy topic grabs one
message before any sweep can wake it (see UnblockSidecarReadersAsync(CancellationToken)).
Left alone, Dapr's reclaim loop redelivers it 60s late (observed: order-book
events 74-86s old). Acking it drops it instead, consistent with
subscriptions starting at the stream tail with no catch-up.
PositionAtTailAsync(string, CancellationToken)
Position the group at the tail and ack its pending entries. Throws when that fails, so the subscription fails rather than replaying silently.
Task PositionAtTailAsync(string topic, CancellationToken cancellationToken)Parameters
topicstringcancellationTokenCancellationToken
Returns
UnblockSidecarReadersAsync(CancellationToken)
Wake this pod's sidecar readers parked in a blocking XREADGROUP
(CLIENT UNBLOCK, timeout mode: the read returns empty, no message
is consumed). A reader left behind by a closed subscription then sees its
cancelled context and exits; a live reader just reads again. Returns how
many were woken. Never throws; failures are logged.
Task<int> UnblockSidecarReadersAsync(CancellationToken cancellationToken)Parameters
cancellationTokenCancellationToken
Returns
Remarks
daprd cancels a closed streaming subscription's component context, but the
Redis component's reader is parked in XREADGROUP ... BLOCK 0 with no
socket deadline, and go-redis doesn't interrupt it. So it stays parked until
the topic's next message, which it then takes into the consumer group's
pending list and abandons unacked. The live subscriber on the same pod gets
that message only once the reclaim loop picks it up (60s later). Reproduced on
daprd 1.18.1, 1.18.4 and 1.19.0-rc.1, with both Dapr.Messaging and a raw
gRPC call: one leftover reader per closed subscription. Observed live: 40
leftover readers on one gateway sidecar delivered order-book events 30-100s
late, starved its Redis pool (publish circuit breaker open), and stole
diffs from workers.