redis queue implementation of schedulePending queries future time range about taskq HOT 5 CLOSED

Thanks for the replies 😀

pending, err := q.redis.XPendingExt(ctx, &redis.XPendingExtArgs{
		Stream: q.stream,
		Group:  q.streamGroup,
		Idle: q.opt.ReservationTimeout,
		Start:  "-",
		End:    "+",  // <--- I'm wondering if this should be time.Now().Add(-1 * q.opt.ReservationTimeout)?
		Count:  batchSize,
	}).Result()

I imagined a scenario like below, correct me if I've misunderstood anything:

• consumer1 makes xreadgroup call: get 10 msgs
• consumer2 makes xreadgroup call: get 10 other msgs
• consumer3 makes xreadgroup call: get 10 other msgs

So far the 30 msgs are all treated as pending. when consumer1 has successfully processed say, the first msg, it makes xdel call to delete that message (although the message is still pending, even deleted). This is the behavior of taskq that it doesn't xack the msg but simply xdel it <-- this means no other group will be able to consume it again, which I find it's ok for use cases where a msg only needs to be processed once by consumers from one group.

Now say, one of the consumers grabs the lock and runs schedulePending, and if it makes xpending for the period - to + (with default count 100): it can possibly get "old enough" msgs from the dead consumers (this is good: msgs got stuck with a dead consumer can be now rescheduled and reprocessed), but also it can get "very recent" messages that are already given to consumers who are actually still alive (which is not good: it grabs work from that alive consumer which could have processed that msg).

Hence I wonder if the intention of schedulePending is only to schedule "old enough" messages, e.g., from - to now - 5min? Since the parameter is named ReservationTimeout and defaults to 5min, it looks like we would give that "consumer" (not sure alive or dead or just "too slow") at least 5 min to process it, and only reschedule it if the msg is at least 5min old.

from taskq.

AFAIR it is supposed to release stuck messages, i.e. messages that were not processed in expected time so we assume that the consumer is dead and re-queue the messages.

from taskq.

func (q *Queue) schedulePending(ctx context.Context) (int, error) {
	tm := time.Now().Add(q.opt.ReservationTimeout)
	start := strconv.FormatInt(unixMs(tm), 10)

	pending, err := q.redis.XPendingExt(ctx, &redis.XPendingExtArgs{
		Stream: q.stream,
		Group:  q.streamGroup,
		Start:  start,
		End:    "+",
		Count:  batchSize,
	}).Result()

I don't know if I got this correctly, @zaiyiguo 's idea is that the param start is in the future, so xpending now+5min will never be able to get the message not processed.

from taskq.

Yes, it looks like you both are right. It probably should be

	pending, err := q.redis.XPendingExt(ctx, &redis.XPendingExtArgs{
		Stream: q.stream,
		Group:  q.streamGroup,
		Idle: q.opt.ReservationTimeout,
		Start:  "-",
		End:    "+",
		Count:  batchSize,
	}).Result()

Is it meant to requeue any pending messages reserved by any consumers in the same group (including other consumers in the group, not necessarily the "current consumer")? Because the xpending command doesn't specify the consumer's name.

The current consumer is presumably dead / restarting. I think we can't rely on it to come back.

from taskq.

That sounds correct. In SQS this is called visibility timeout and SQS does such cleanup automatically.

<--- I'm wondering if this should be time.Now().Add(-1 * q.opt.ReservationTimeout)?

I honestly don't know what works the best with Redis, but you have the right idea what to do :)

from taskq.