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, 3 insertions, 6 deletions

diff --git a/src/stream.h b/src/stream.h
index 2d4997919..bfc165440 100644
--- a/src/stream.h
+++ b/src/stream.h
@@ -74,7 +74,8 @@ typedef struct streamCG {
 
 /* A specific consumer in a consumer group.  */
 typedef struct streamConsumer {
-    mstime_t seen_time;         /* Last time this consumer was active. */
+    mstime_t seen_time;         /* Last time this consumer tried to perform an action (attempted reading/claiming). */
+    mstime_t active_time;       /* Last time this consumer was active (successful reading/claiming). */
     sds name;                   /* Consumer name. This is how the consumer
                                    will be identified in the consumer group
                                    protocol. Case sensitive. */
@@ -105,10 +106,6 @@ typedef struct streamPropInfo {
 /* Prototypes of exported APIs. */
 struct client;
 
-/* Flags for streamLookupConsumer */
-#define SLC_DEFAULT      0
-#define SLC_NO_REFRESH   (1<<0) /* Do not update consumer's seen-time */
-
 /* Flags for streamCreateConsumer */
 #define SCC_DEFAULT       0
 #define SCC_NO_NOTIFY     (1<<0) /* Do not notify key space if consumer created */
@@ -126,7 +123,7 @@ void streamIteratorGetField(streamIterator *si, unsigned char **fieldptr, unsign
 void streamIteratorRemoveEntry(streamIterator *si, streamID *current);
 void streamIteratorStop(streamIterator *si);
 streamCG *streamLookupCG(stream *s, sds groupname);
-streamConsumer *streamLookupConsumer(streamCG *cg, sds name, int flags);
+streamConsumer *streamLookupConsumer(streamCG *cg, sds name);
 streamConsumer *streamCreateConsumer(streamCG *cg, sds name, robj *key, int dbid, int flags);
 streamCG *streamCreateCG(stream *s, char *name, size_t namelen, streamID *id, long long entries_read);
 streamNACK *streamCreateNACK(streamConsumer *consumer);