Redis Streams: A Log With Consumer Groups

    /* Populate the listpack with the new entry. We use the following
     * encoding:
     *
     * +-----+--------+----------+-------+-------+-/-+-------+-------+--------+
     * |flags|entry-id|num-fields|field-1|value-1|...|field-N|value-N|lp-count|
     * +-----+--------+----------+-------+-------+-/-+-------+-------+--------+
     *
     * However if the SAMEFIELD flag is set, we have just to populate
     * the entry with the values, so it becomes:
     *
     * +-----+--------+-------+-/-+-------+--------+
     * |flags|entry-id|value-1|...|value-N|lp-count|
     * +-----+--------+-------+-/-+-------+--------+
     *
     * The entry-id field is actually two separated fields: the ms
     * and seq difference compared to the master entry.
     *
     * The lp-count field is a number that states the number of listpack pieces
     * that compose the entry, so that it's possible to travel the entry
     * in reverse order: we can just start from the end of the listpack, read
     * the entry, and jump back N times to seek the "flags" field to read
     * the stream full entry. */
    size_t oldsize = lpBytes(lp);
    lp = lpAppendInteger(lp,flags);
    lp = lpAppendInteger(lp,id.ms - master_id.ms);
    lp = lpAppendInteger(lp,id.seq - master_id.seq);
    if (!(flags & STREAM_ITEM_FLAG_SAMEFIELDS))
        lp = lpAppendInteger(lp,numfields);
    for (int64_t i = 0; i < numfields; i++) {
        sds field = argv[i*2]->ptr, value = argv[i*2+1]->ptr;
        if (!(flags & STREAM_ITEM_FLAG_SAMEFIELDS))
            lp = lpAppend(lp,(unsigned char*)field,sdslen(field));
        lp = lpAppend(lp,(unsigned char*)value,sdslen(value));
    }
    /* Compute and store the lp-count field. */
    int64_t lp_count = numfields;
    lp_count += 3; /* Add the 3 fixed fields flags + ms-diff + seq-diff. */
    if (!(flags & STREAM_ITEM_FLAG_SAMEFIELDS)) {
        /* If the item is not compressed, it also has the fields other than
         * the values, and an additional num-fields field. */
        lp_count += numfields+1;
    }
    lp = lpAppendInteger(lp,lp_count);
    s->alloc_size -= oldsize;
    s->alloc_size += lpBytes(lp);

    /* Insert back into the tree in order to update the listpack pointer. */
    if (ri.data != lp)
        raxInsert(s->rax,(unsigned char*)&rax_key,sizeof(rax_key),lp,NULL);
    s->length++;
    s->entries_added++;
    s->last_id = id;
    if (s->length == 1) s->first_id = id;
    if (added_id) *added_id = id;
    return C_OK;
}

streamCG *streamCreateCG(stream *s, char *name, size_t namelen, streamID *id, long long entries_read) {
    if (s->cgroups == NULL)
        s->cgroups = raxNewWithMetadata(0, &s->alloc_size);
    if (raxFind(s->cgroups,(unsigned char*)name,namelen,NULL))
        return NULL;

    size_t usable;
    streamCG *cg = zmalloc_usable(sizeof(*cg), &usable);
    s->alloc_size += usable;
    cg->pel = raxNewWithMetadata(0, &s->alloc_size);
    cg->pel_time_head = NULL;
    cg->pel_time_tail = NULL;
    cg->pel_nack_tail = NULL;
    cg->consumers = raxNewWithMetadata(0, &s->alloc_size);
    cg->last_id.ms = 0;
    cg->last_id.seq = 0;
    streamUpdateCGroupLastId(s, cg, id);
    cg->entries_read = entries_read;
    raxInsert(s->cgroups,(unsigned char*)name,namelen,cg,NULL);
    return cg;
}

/* Free a consumer group and all its associated data. */
static void streamFreeCG(stream *s, streamCG *cg) {
    /* Free the pel, unlinking each NACK from the time list in the callback */
    streamFreeNACKCtx ctx = {s, cg};
    raxFreeWithCbAndContext(cg->pel, streamFreeNACKGeneric, &ctx);
    
    /* pel_time_head/tail/pel_nack_tail should now be NULL after unlinking all NACKs */
    serverAssert(cg->pel_time_head == NULL && cg->pel_time_tail == NULL && cg->pel_nack_tail == NULL);
    
    raxFreeWithCbAndContext(cg->consumers, streamFreeConsumerGeneric, s);
    size_t usable;
    zfree_usable(cg, &usable);
    s->alloc_size -= usable;
}

/* Destroy a consumer group and clean up all associated references. */
void streamDestroyCG(stream *s, streamCG *cg) {
    /* Remove all references from the cgroups_ref. */
    raxIterator it;
    raxStart(&it, cg->pel);
    raxSeek(&it, "^", NULL, 0);
    while (raxNext(&it)) {
        streamNACK *nack = it.data;
        streamUnlinkEntryFromCGroupRef(s, nack, it.key);
    }
    raxStop(&it);

    /* If we're destroying the group with the minimum last_id, the cached
     * minimum is no longer valid and needs to be recalculated from the
     * remaining groups. */
    if (s->min_cgroup_last_id_valid && streamCompareID(&s->min_cgroup_last_id, &cg->last_id) == 0)
        s->min_cgroup_last_id_valid = 0;

    streamFreeCG(s, cg);
}

        if (group && !noack) {
            unsigned char buf[sizeof(streamID)];
            streamEncodeID(buf,&id);

            /* Try to add a new NACK. Most of the time this will work and
             * will not require extra lookups. We'll fix the problem later
             * if we find that there is already an entry for this ID. */
            streamNACK *nack = streamCreateNACK(s, consumer, &id);
            int group_inserted =
                raxTryInsert(group->pel,buf,sizeof(buf),nack,NULL);

            /* Now we can check if the entry was already busy, and
             * in that case reassign the entry to the new consumer,
             * or update it if the consumer is the same as before. */
            if (group_inserted == 0) {
                streamFreeNACK(s,nack);
                void *result;
                int found = raxFind(group->pel,buf,sizeof(buf),&result);
                serverAssert(found);
                nack = result;
                /* Only transfer between consumers if they're different */
                if (nack->consumer != consumer) {
                    if (nack->consumer)
                        raxRemove(nack->consumer->pel,buf,sizeof(buf),NULL);
                    nack->consumer = consumer;
                    raxInsert(consumer->pel,buf,sizeof(buf),nack,NULL);
                }
                nack->delivery_count = 1;
                /* Update delivery time and reposition in time list */
                pelListUpdate(group, nack, cmd_time_snapshot);
            } else {
                /* New NACK - insert into consumer's PEL and time list */
                raxInsert(consumer->pel,buf,sizeof(buf),nack,NULL);
                nack->cgroup_ref_node = streamLinkCGroupToEntry(s, group, buf);
                pelListInsertAtTail(group, nack);
            }

            consumer->active_time = cmd_time_snapshot;

            /* Propagate as XCLAIM. */
            if (spi) {
                robj *delivery_count = createStringObjectFromLongLong(nack->delivery_count);
                streamPropagateXCLAIMCopyFree(db_id,spi->keyname,group_last_id,spi->groupname,idarg,consumername,delivery_time,delivery_count);
                decrRefCount(delivery_count);
                if (propCount) (*propCount)++;
            }
        }
        decrRefCount(idarg);
        arraylen++;
        if (count && count == arraylen) break;
    }

    if (spi && consumer) {

void xackCommand(client *c) {
    streamCG *group = NULL;
    kvobj *kv = lookupKeyRead(c->db, c->argv[1]);
    if (kv) {
        if (checkType(c, kv, OBJ_STREAM)) return; /* Type error. */
        group = streamLookupCG(kv->ptr, c->argv[2]->ptr);
    }

    /* No key or group? Nothing to ack. */
    if (kv == NULL || group == NULL) {
        addReply(c,shared.czero);
        return;
    }

    /* Start parsing the IDs, so that we abort ASAP if there is a syntax
     * error: the return value of this command cannot be an error in case
     * the client successfully acknowledged some messages, so it should be
     * executed in a "all or nothing" fashion. */
    streamID static_ids[STREAMID_STATIC_VECTOR_LEN];
    streamID *ids = static_ids;
    int id_count = c->argc-3;
    if (id_count > STREAMID_STATIC_VECTOR_LEN)
        ids = zmalloc(sizeof(streamID)*id_count);
    for (int j = 3; j < c->argc; j++) {
        if (streamParseStrictIDOrReply(c,c->argv[j],&ids[j-3],0,NULL) != C_OK) goto cleanup;
    }

    int acknowledged = 0;
    size_t old_alloc = server.memory_tracking_enabled ? kvobjAllocSize(kv) : 0;
    for (int j = 3; j < c->argc; j++) {
        unsigned char buf[sizeof(streamID)];
        streamEncodeID(buf,&ids[j-3]);

        /* Lookup the ID in the group PEL: it will have a reference to the
         * NACK structure that will have a reference to the consumer, so that
         * we are able to remove the entry from both PELs. */
        void *result;
        if (raxFind(group->pel,buf,sizeof(buf),&result)) {
            streamNACK *nack = result;
            pelListUnlink(group, nack);
            raxRemove(group->pel,buf,sizeof(buf),NULL);
            if (nack->consumer)
                raxRemove(nack->consumer->pel,buf,sizeof(buf),NULL);
            streamDestroyNACK(kv->ptr, nack, buf);
            acknowledged++;
            server.dirty++;
            keyModified(c,c->db,c->argv[1],kv,0);
        }
    }
    if (server.memory_tracking_enabled)
        updateSlotAllocSize(c->db,getKeySlot(c->argv[1]->ptr),kv,old_alloc,kvobjAllocSize(kv));
    addReplyLongLong(c,acknowledged);
cleanup:
    if (ids != static_ids) zfree(ids);
}

/* XNACK key group <SILENT|FAIL|FATAL> IDS numids id [id ...]
 *       [RETRYCOUNT count] [FORCE]
 *
 * Release pending messages back to the group's PEL without acknowledging them.

void xclaimCommand(client *c) {
    streamCG *group = NULL;
    kvobj *o = lookupKeyRead(c->db,c->argv[1]);
    long long minidle; /* Minimum idle time argument. */
    long long retrycount = -1;   /* -1 means RETRYCOUNT option not given. */
    mstime_t deliverytime = -1;  /* -1 means IDLE/TIME options not given. */
    int force = 0;
    int justid = 0;

    if (o) {
        if (checkType(c,o,OBJ_STREAM)) return; /* Type error. */
        group = streamLookupCG(o->ptr,c->argv[2]->ptr);
    }

    /* No key or group? Send an error given that the group creation
     * is mandatory. */
    if (o == NULL || group == NULL) {
        addReplyErrorFormat(c,"-NOGROUP No such key '%s' or "
                              "consumer group '%s'", (char*)c->argv[1]->ptr,
                              (char*)c->argv[2]->ptr);
        return;
    }

    if (getLongLongFromObjectOrReply(c,c->argv[4],&minidle,
        "Invalid min-idle-time argument for XCLAIM")
        != C_OK) return;
    if (minidle < 0) minidle = 0;

    /* Start parsing the IDs, so that we abort ASAP if there is a syntax
     * error: the return value of this command cannot be an error in case
     * the client successfully claimed some message, so it should be
     * executed in a "all or nothing" fashion. */
    int j;
    streamID static_ids[STREAMID_STATIC_VECTOR_LEN];
    streamID *ids = static_ids;
    int id_count = c->argc-5;
    if (id_count > STREAMID_STATIC_VECTOR_LEN)
        ids = zmalloc(sizeof(streamID)*id_count);
    for (j = 5; j < c->argc; j++) {
        if (streamParseStrictIDOrReply(NULL,c->argv[j],&ids[j-5],0,NULL) != C_OK) break;
    }
    int last_id_arg = j-1; /* Next time we iterate the IDs we now the range. */

    /* If we stopped because some IDs cannot be parsed, perhaps they
     * are trailing options. */
    mstime_t now = commandTimeSnapshot();
    streamID last_id = {0,0};
    int propagate_last_id = 0;
    for (; j < c->argc; j++) {
        int moreargs = (c->argc-1) - j; /* Number of additional arguments. */
        char *opt = c->argv[j]->ptr;
        if (!strcasecmp(opt,"FORCE")) {
            force = 1;
        } else if (!strcasecmp(opt,"JUSTID")) {
            justid = 1;
        } else if (!strcasecmp(opt,"IDLE") && moreargs) {
            j++;
            if (getLongLongFromObjectOrReply(c,c->argv[j],&deliverytime,
                "Invalid IDLE option argument for XCLAIM")
                != C_OK) goto cleanup;

int64_t streamTrim(stream *s, streamAddTrimArgs *args) {
    size_t maxlen = args->maxlen;
    streamID *id = &args->minid;
    int approx = args->approx_trim;
    int64_t limit = args->limit;
    int trim_strategy = args->trim_strategy;
    int delete_strategy = args->delete_strategy;

    if (trim_strategy == TRIM_STRATEGY_NONE)
        return 0;

    raxIterator ri;
    raxStart(&ri,s->rax);
    raxSeek(&ri,"^",NULL,0);

    int64_t deleted = 0;
    while (raxNext(&ri)) {
        if (trim_strategy == TRIM_STRATEGY_MAXLEN && s->length <= maxlen)
            break;

        unsigned char *lp = ri.data, *p = lpFirst(lp);
        int64_t entries = lpGetInteger(p);

        /* Check if we exceeded the amount of work we could do */
        if (limit && (deleted + entries) > limit)
            break;

        /* Check if we can remove the whole node */
        int remove_node = 0; /* Final decision flag for node removal */
        int node_eligible_for_remove = 0; /* Whether node meets the basic criteria for removal */
        streamID master_id = {0};
        /* Read the master ID from the radix tree key. */
        streamDecodeID(ri.key, &master_id);
        if (trim_strategy == TRIM_STRATEGY_MAXLEN) {
            node_eligible_for_remove = s->length - entries >= maxlen;
        } else {
            /* Read last ID. */
            streamID last_id = {0,0};
            lpGetEdgeStreamID(lp, 0, &master_id, &last_id);

            /* We can remove the entire node id its last ID < 'id' */
            node_eligible_for_remove = streamCompareID(&last_id, id) < 0;
        }

        if (node_eligible_for_remove && delete_strategy == DELETE_STRATEGY_KEEPREF) {
            /* With KEEPREF strategy, we can remove the whole node directly since we don't need
             * to check or clean up consumer group references. */
            remove_node = 1;
        }

        if (remove_node) {
            s->alloc_size -= lpBytes(lp);
            lpFree(lp);
            raxRemove(s->rax,ri.key,ri.key_len,NULL);
            raxSeek(&ri,">=",ri.key,ri.key_len);
            s->length -= entries;
            deleted += entries;
            continue;
        }