Feature request: Dead Letter support

[kekonn]

Is there a reason the Redis queue implementation does not have support to retrieve dead letters?

It would be very convenient for us. If you are open to an implementation, I would be willing to give it a go.

[niemyjski]

The queue has dead letter support and automatically moves items to a dead letter list which is controlled by DeadLetterTimeToLive which defaults to 1 day. Were you thinking of something else?

[kekonn]

Maybe I should have worded my question better. I meant support for retrieving dead letters and retrying them. Currently retrieving dead letters on a Redis queue throws a NotImplementedException.

[niemyjski]

That's a gap currently, you could use a redis cli and view keys in the :dead list and move them back to working and that "should" work. How do you se this working? How were you thinking of implementing it?

[kekonn]

Now I'm not really experienced with redis and manually operating queues and such. I usually just use libraries. But I asked AI to give me a hand and this is what it came up with:

protected override async Task<IEnumerable<T>> GetDeadletterItemsImplAsync(CancellationToken cancellationToken)
    {
        cancellationToken.ThrowIfCancellationRequested();

        RedisValue[] ids = await _resiliencePolicy.ExecuteAsync<RedisValue[]>(
            async _ => await Database.ListRangeAsync(_deadListName),
            cancellationToken: cancellationToken);

        if (ids == null || ids.Length == 0)
            return Enumerable.Empty<T>();

        var payloadTasks = ids.Select(async idValue =>
        {
            cancellationToken.ThrowIfCancellationRequested();

            if (idValue.IsNullOrEmpty)
                return (HasPayload: false, Id: (string)null, Value: default(T));

            string id = idValue.ToString();

            try
            {
                var envelope = await _resiliencePolicy.ExecuteAsync<CacheValue<RedisPayloadEnvelope<T>>>(
                    async _ => await _cache.GetAsync<RedisPayloadEnvelope<T>>(String.Concat(_listPrefix, ":", id)),
                    cancellationToken: cancellationToken);

                if (envelope.IsNull || envelope.Value?.Value == null)
                {
                    _logger.LogDebug("Deadletter payload missing or expired for id: {QueueEntryId}", id);

                    // Best-effort cleanup: remove phantom id from deadletter list.
                    try
                    {
                        await _resiliencePolicy.ExecuteAsync(
                            async _ => await Database.ListRemoveAsync(_deadListName, id),
                            cancellationToken: cancellationToken);
                    }
                    catch (Exception ex)
                    {
                        _logger.LogDebug(ex, "Error removing missing/expired deadletter id {QueueEntryId} from list",
                            id);
                    }

                    return (HasPayload: false, Id: id, Value: default(T));
                }

                return (HasPayload: true, Id: id, Value: envelope.Value.Value);
            }
            catch (Exception ex)
            {
                // If we error reading/deserializing, treat it like missing (don’t return it).
                _logger.LogDebug(ex, "Error getting deadletter payload for id: {QueueEntryId}", id);
                return (HasPayload: false, Id: id, Value: default(T));
            }
        }).ToArray();

        var results = await Task.WhenAll(payloadTasks);
        return results.Where(r => r.HasPayload).Select(r => r.Value).ToList();
    }

I don't feel qualified to judge if this is up to the standards of the rest of your queue implementations, but functionally it seems to work for my use case.

[niemyjski]

I definitely feel like this could be better and act more like a real queue we hand off to and then you could have a deadletter version of which you could wire up to and dequeue from that / redrive.

@ejsmith any thoughts?