Stream consumers: Re-purpose seen-time, add active-time (#11099)

1. "Fixed" the current code so that seen-time/idle actually refers to interaction
  attempts (as documented; breaking change)
2. Added active-time/inactive to refer to successful interaction (what
  seen-time/idle used to be)

At first, I tried to avoid changing the behavior of seen-time/idle but then realized
that, in this case, the odds are the people read the docs and implemented their
code based on the docs (which didn't match the behavior).
For the most part, that would work fine, except that issue #9996 was found.

I was working under the assumption that people relied on the docs, and for
the most part, it could have worked well enough. so instead of fixing the docs,
as I would usually do, I fixed the code to match the docs in this particular case.

Note that, in case the consumer has never read any entries, the values
for both "active-time" (XINFO FULL) and "inactive" (XINFO CONSUMERS) will
be -1, meaning here that the consumer was never active.

Note that seen/active time is only affected by XREADGROUP / X[AUTO]CLAIM, not
by XPENDING, XINFO, and other "read-only" stream CG commands (always has been,
even before this PR)

Other changes:
* Another behavioral change (arguably a bugfix) is that XREADGROUP and X[AUTO]CLAIM
  create the consumer regardless of whether it was able to perform some reading/claiming
* RDB format change to save the `active_time`, and set it to the same value of `seen_time` in old rdb files.

1 files changed, 1 insertions, 1 deletions

diff --git a/src/blocked.c b/src/blocked.c
index 069d3ba95..5d6d0c800 100644
--- a/src/blocked.c
+++ b/src/blocked.c
@@ -489,7 +489,7 @@ void serveClientsBlockedOnStreamKey(robj *o, readyList *rl) {
                 if (group) {
                     noack = receiver->bpop.xread_group_noack;
                     sds name = receiver->bpop.xread_consumer->ptr;
-                    consumer = streamLookupConsumer(group,name,SLC_DEFAULT);
+                    consumer = streamLookupConsumer(group,name);
                     if (consumer == NULL) {
                         consumer = streamCreateConsumer(group,name,rl->key,
                                                         rl->db->id,SCC_DEFAULT);