diff options
Diffstat (limited to 'src/t_stream.c')
-rw-r--r-- | src/t_stream.c | 48 |
1 files changed, 38 insertions, 10 deletions
diff --git a/src/t_stream.c b/src/t_stream.c index 1968475fa..02d9fdcaa 100644 --- a/src/t_stream.c +++ b/src/t_stream.c @@ -47,6 +47,12 @@ * setting stream_node_max_bytes to a huge number. */ #define STREAM_LISTPACK_MAX_PRE_ALLOCATE 4096 +/* Don't let listpacks grow too big, even if the user config allows it. + * doing so can lead to an overflow (trying to store more than 32bit length + * into the listpack header), or actually an assertion since lpInsert + * will return NULL. */ +#define STREAM_LISTPACK_MAX_SIZE (1<<30) + void streamFreeCG(streamCG *cg); void streamFreeNACK(streamNACK *na); size_t streamReplyWithRangeFromConsumerPEL(client *c, stream *s, streamID *start, streamID *end, size_t count, streamConsumer *consumer); @@ -415,8 +421,11 @@ void streamGetEdgeID(stream *s, int first, streamID *edge_id) * * The function returns C_OK if the item was added, this is always true * if the ID was generated by the function. However the function may return - * C_ERR if an ID was given via 'use_id', but adding it failed since the - * current top ID is greater or equal. */ + * C_ERR in several cases: + * 1. If an ID was given via 'use_id', but adding it failed since the + * current top ID is greater or equal. errno will be set to EDOM. + * 2. If a size of a single element or the sum of the elements is too big to + * be stored into the stream. errno will be set to ERANGE. */ int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_id, streamID *use_id) { /* Generate the new entry ID. */ @@ -430,7 +439,23 @@ int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_ * or return an error. Automatically generated IDs might * overflow (and wrap-around) when incrementing the sequence part. */ - if (streamCompareID(&id,&s->last_id) <= 0) return C_ERR; + if (streamCompareID(&id,&s->last_id) <= 0) { + errno = EDOM; + return C_ERR; + } + + /* Avoid overflow when trying to add an element to the stream (listpack + * can only host up to 32bit length sttrings, and also a total listpack size + * can't be bigger than 32bit length. */ + size_t totelelen = 0; + for (int64_t i = 0; i < numfields*2; i++) { + sds ele = argv[i]->ptr; + totelelen += sdslen(ele); + } + if (totelelen > STREAM_LISTPACK_MAX_SIZE) { + errno = ERANGE; + return C_ERR; + } /* Add the new entry. */ raxIterator ri; @@ -489,9 +514,10 @@ int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_ * if we need to switch to the next one. 'lp' will be set to NULL if * the current node is full. */ if (lp != NULL) { - if (server.stream_node_max_bytes && - lp_bytes >= server.stream_node_max_bytes) - { + size_t node_max_bytes = server.stream_node_max_bytes; + if (node_max_bytes == 0 || node_max_bytes > STREAM_LISTPACK_MAX_SIZE) + node_max_bytes = STREAM_LISTPACK_MAX_SIZE; + if (lp_bytes + totelelen >= node_max_bytes) { lp = NULL; } else if (server.stream_node_max_entries) { unsigned char *lp_ele = lpFirst(lp); @@ -1782,11 +1808,13 @@ void xaddCommand(client *c) { /* Append using the low level function and return the ID. */ streamID id; if (streamAppendItem(s,c->argv+field_pos,(c->argc-field_pos)/2, - &id, parsed_args.id_given ? &parsed_args.id : NULL) - == C_ERR) + &id, parsed_args.id_given ? &parsed_args.id : NULL) == C_ERR) { - addReplyError(c,"The ID specified in XADD is equal or smaller than the " - "target stream top item"); + if (errno == EDOM) + addReplyError(c,"The ID specified in XADD is equal or smaller than " + "the target stream top item"); + else + addReplyError(c,"Elements are too large to be stored"); return; } addReplyStreamID(c,&id); |