Class NoopConsumerGroupPositioner

Namespace

Virtufin.Api.Services

Assembly

Virtufin.Api.dll

No-op positioner, for a gateway without Redis access configured (local dev).

public sealed class NoopConsumerGroupPositioner : IConsumerGroupPositioner

Inheritance

object

NoopConsumerGroupPositioner

Implements

IConsumerGroupPositioner

Inherited Members

object.Equals(object)

object.Equals(object, object)

object.GetHashCode()

object.GetType()

object.ReferenceEquals(object, object)

object.ToString()

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.

public Task<int> AckAbandonedPendingAsync(IReadOnlyCollection<string> topics, TimeSpan minIdle, CancellationToken cancellationToken)

Parameters

topics IReadOnlyCollection<string>

minIdle TimeSpan

cancellationToken CancellationToken

Returns

Task<int>

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.

public Task PositionAtTailAsync(string topic, CancellationToken cancellationToken)

Parameters

topic string

cancellationToken CancellationToken

Returns

Task

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.

public Task<int> UnblockSidecarReadersAsync(CancellationToken cancellationToken)

Parameters

cancellationToken CancellationToken

Returns

Task<int>

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.