[JonathanSerafini]

Description

This change adds an initial implementation for an optional reading mode for the redis stream
subscriber whereby it will attempt to read messages from the PEL first and then continue on
reading new messages on the stream.

At present this mode is an optional and disable so as to not alter the existing behaviour.

On it's own, this change doesn't "do" much. However, by utilizing the resume_from method
it now becomes possible to periodically have the consumer switch from reading new messages
to reading from the PEL.

In a subsequent PR, i could add the resumable option to the subscriber settings and potentially
also add a background task to monitor the pending list to set resume_from to consume pending.

Type of change

Please delete options that are not relevant.

• Documentation (typos, code examples, or any documentation updates)
• Bug fix (a non-breaking change that resolves an issue)
• New feature (a non-breaking change that adds functionality)
• Breaking change (a fix or feature that would disrupt existing functionality)
• This change requires a documentation update

Checklist

• My code adheres to the style guidelines of this project (just lint shows no errors)
• I have conducted a self-review of my own code
• I have made the necessary changes to the documentation
• My changes do not generate any new warnings
• I have added tests to validate the effectiveness of my fix or the functionality of my new feature
• Both new and existing unit tests pass successfully on my local environment by running just test-coverage
• I have ensured that static analysis tests are passing by running just static-analysis
• I have included code examples to illustrate the modifications

[JonathanSerafini]

@Lancetnik I was wondering if you might have an opinion on this implementation for reading of pending redis messages

[Lancetnik]

@JonathanSerafini sorry, I am nor sure, that I got your second idea correctly. Can you explain more detail, please? Probably, with pseudocode. But, I like your approach

[JonathanSerafini]

@Lancetnik Sure ...
primary goal right now is to add support for reading the pending entries without needing big changes or forcing a particular workflow. the way i'm currently thinking it could be used would be as follows ...

subscriber = broker.subscriber(stream=queue)

@subscriber
async def handler(msg): ...

async def pending_monitor(subscriber: StreamSubscriber, interval: float, idle: int) -> None:
  while True:
      # optionally get first xpending message
      # optionally filter xpending by idle
      # or optionally walk the list of xpending_range and delete those where delivery_attempts too high
      if has_pending:
          subscriber.read_pending = True
      else:
          subscriber.read_pending = False
      await asyncio.sleep(interval)
      
subscriber.add_task(pending_monitor, subscriber, 15, 30 * 1000)

above is pseudo-code ... it doesn't quite work with the declarative approach out of the box
though i have managed to get something like this working by wrapping the RedisBroker.start and adding the tasks after a super().start.

would be cleaner if it was either builtin support or subscribers had some sort of support for registering tasks as part of the decorator.

[Lancetnik]

would be cleaner if it was either builtin support or subscribers had some sort of support for registering tasks as part of the decorator.

Not "tasks" exactly, but we made smth close here - https://faststream.ag2.ai/latest/howto/nats/in-progress/

[JonathanSerafini]

Not "tasks" exactly, but we made smth close here - https://faststream.ag2.ai/latest/howto/nats/in-progress/

By the looks of it, it does seem like a background task would be warranted given that it's more "subscriber lifecycle" than it is "message lifecycle" as it is for NATs.

[Lancetnik]

@JonathanSerafini hi! Sorry for waiting so long for this feature
Can you check #2607 please? Looks like it implements the thing you want

[JonathanSerafini]

@Lancetnik It's up to you ... once this was merged i was going to have background tasks running to set the resume_from property on the subscriber to do something similar so that you can have a single subscriber that can fetch both from standard stream and pel rather than having to duplicate it ... additionally, there's an extra background task that would come after that to monitor for dead consumers ( without that autoscaling leaves references of previous consumers lying aroudn )

[Lancetnik]

@JonathanSerafini do you see any difference between "background task" and "second subscriber in parallel"? For me they looks the same from event loop view - just an endless task, fetching data from stream time-to-time

[JonathanSerafini]

@Lancetnik ultimately it's a design decision that depends on how to envision users will typically use the subscriber.

if you look at a Kafka stream, for instance, you have one subscriber that processes messages sequentially. Should a message fail to be ACKed, then the subscriber would retry the message before moving on to the next message. All of this occurs in the one subscriber you've defined.

in Redis, this is slightly different since the pending message won't ever be retried by default since it's going to the PEL.
So when we add support to retry "pending" messages, how do we want the end user to configure this ?

So should we:
a) Assume that a typical user rarely wants to retry messages, in which case it might make sense to ask them to define a distinct "retry" subscriber such as in 2607.
b) Assume that a typical user usually wants to retry messages, in which case it may be more convenient to consume retryable messages from the same subscriber.

For the background tasks this is what i was thinking of :

    async def monitor_pending(self) -> None:
        """Monitor the current worker for pending messages."""
        interval = self.stream_sub.pending_interval or 60

        while True:
            if not self._client:
                await asyncio.sleep(interval)
                continue

            response = await self.client.xpending_range(
                name=self.stream_sub.name,
                groupname=self.stream_sub.group,
                consumername=self.stream_sub.consumer,
                min="-",
                max="+",
                count=1,
                idle=self.stream_sub.pending_min_idle,
            )

            if response:
                self.resume_from("0-0")

            await asyncio.sleep(interval)

    async def monitor_claim(self) -> None:
        """Monitor the the stream for idle messages and reclaim them."""
        interval = self.stream_sub.claim_interval or 60

        while True:
            if not self._client:
                await asyncio.sleep(interval)
                continue

            await self.client.xautoclaim(
                name=self.stream_sub.name,
                groupname=self.stream_sub.group,
                consumername=self.stream_sub.consumer,
                min_idle_time=self.stream_sub.claim_min_idle,
                count=self.stream_sub.claim_max_count,
                justid=True
            )

            await asyncio.sleep(interval)

    async def monitor_consumers(self) -> None:
        """Monitor the the stream for idle consumers and delete them."""

[Lancetnik]

@JonathanSerafini fortunately #2607 closes both cases

1. Assume that a typical user rarely wants to retry messages, in which case it might make sense to ask them to define a distinct "retry" subscriber

@broker.subscriber("regular")
async def handler(): ...

@broker.subscriber("PEL")
async def retry_handler(): ...

2. Assume that a typical user usually wants to retry messages, in which case it may be more convenient to consume retryable messages from the same subscriber handler

# decorate the same handler to create 2 subscribers with same processing logic
@broker.subscriber("regular")
@broker.subscriber("PEL")  
async def handler(): ...

[JonathanSerafini]

alrighty that works, i'll just combine those in my wrapper and call it a day.
on my end, I always want retry support and it would otherwise be too easy for a developer to forget to add it.

i am curious to see how xautoclaim works for messages from the same consumer, will need to test it.

next pr will just fix the fact the redis subscriber implementation blocks the usage of add_task for the background monitor tasks i need. unless you'd rather not change that in which case lmk ( this is what i mean ).