Table of Contents

Class RedisConsumerGroupPositioner

Namespace
Virtufin.Api.Services
Assembly
Virtufin.Api.dll

IConsumerGroupPositioner over a direct Redis connection to the pub/sub component's Redis. Safe only because the group is this pod's alone ({podName}) and the registry calls it while this process holds no Dapr subscription for the topic.

public sealed class RedisConsumerGroupPositioner : IConsumerGroupPositioner, IAsyncDisposable
Inheritance
RedisConsumerGroupPositioner
Implements
Inherited Members

Constructors

RedisConsumerGroupPositioner(string, string, ILogger<RedisConsumerGroupPositioner>, IReadOnlySet<IPAddress>?) 

public RedisConsumerGroupPositioner(string connectionString, string consumerGroup, ILogger<RedisConsumerGroupPositioner> logger, IReadOnlySet<IPAddress>? sidecarAddresses = null)

Parameters

connectionString string
consumerGroup string
logger ILogger<RedisConsumerGroupPositioner>
sidecarAddresses IReadOnlySet<IPAddress>

Properties

ConsumerGroup

The consumer group this pod's sidecar reads with (its Dapr consumerID).

public string ConsumerGroup { get; }

Property Value

string

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.

DisposeAsync()

Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources asynchronously.

public ValueTask DisposeAsync()

Returns

ValueTask

A task that represents the asynchronous dispose operation.

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

ResolveConsumerGroup(string?, Func<string, string?>)

Resolve the consumer group from the pub/sub component's consumerID template, substituting {podName} the way the Dapr runtime does (from POD_NAME, falling back to the hostname, which Kubernetes sets to the pod name). Returns null when the template is unset: Dapr then uses the app id, a group every gateway replica shares, which must not be moved from under the other replicas.

public static string? ResolveConsumerGroup(string? consumerIdTemplate, Func<string, string?> getEnv)

Parameters

consumerIdTemplate string
getEnv Func<string, string>

Returns

string

ResolveSidecarAddresses(Func<string, string?>)

The addresses this pod's sidecar connects to Redis from: POD_IP (the sidecar shares the pod's network namespace), else every local address including loopback (local dev, where a shared localhost makes this wake other apps' readers too -- harmless, they read again).

public static IReadOnlySet<IPAddress> ResolveSidecarAddresses(Func<string, string?> getEnv)

Parameters

getEnv Func<string, string>

Returns

IReadOnlySet<IPAddress>

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.