gvsafronov/futriix
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

lpGetInteger returns int64_t, avoid overflow (#10068)

Browse Source 
Fix #9410

Crucial for the ms and sequence deltas, but I changed all
calls, just in case (e.g. "flags")

Before this commit:
`ms_delta` and `seq_delta` could have overflown, causing `currid` to be wrong,
which in turn would cause `streamTrim` to trim the entire rax node (see new test)
This commit is contained in:
guybe7 2022-01-07 14:31:05 +01:00 committed by GitHub
parent e88f6acb94
commit 7cd6a64d2f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 16 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
src/t_stream.c
View File

@ -793,13 +793,13 @@ int64_t streamTrim(stream *s, streamAddTrimArgs *args) {
* update it after (and if) we actually remove the entry */ * update it after (and if) we actually remove the entry */
unsigned char *pcopy = p; unsigned char *pcopy = p;
int flags = lpGetInteger(p); int64_t flags = lpGetInteger(p);
p = lpNext(lp, p); /* Skip flags. */ p = lpNext(lp, p); /* Skip flags. */
int to_skip; int64_t to_skip;
int ms_delta = lpGetInteger(p); int64_t ms_delta = lpGetInteger(p);
p = lpNext(lp, p); /* Skip ID ms delta */ p = lpNext(lp, p); /* Skip ID ms delta */
int seq_delta = lpGetInteger(p); int64_t seq_delta = lpGetInteger(p);
p = lpNext(lp, p); /* Skip ID seq delta */ p = lpNext(lp, p); /* Skip ID seq delta */
streamID currid = {0}; /* For MINID */ streamID currid = {0}; /* For MINID */
@ -1135,7 +1135,7 @@ int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
* the first entry emitted for this listpack, then we already * the first entry emitted for this listpack, then we already
* emitted the current entry, and have to go back to the previous * emitted the current entry, and have to go back to the previous
* one. */ * one. */
int lp_count = lpGetInteger(si->lp_ele); int64_t lp_count = lpGetInteger(si->lp_ele);
while(lp_count--) si->lp_ele = lpPrev(si->lp,si->lp_ele); while(lp_count--) si->lp_ele = lpPrev(si->lp,si->lp_ele);
/* Seek lp-count of prev entry. */ /* Seek lp-count of prev entry. */
si->lp_ele = lpPrev(si->lp,si->lp_ele); si->lp_ele = lpPrev(si->lp,si->lp_ele);
@ -1165,7 +1165,7 @@ int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
/* Get the flags entry. */ /* Get the flags entry. */
si->lp_flags = si->lp_ele; si->lp_flags = si->lp_ele;
int flags = lpGetInteger(si->lp_ele); int64_t flags = lpGetInteger(si->lp_ele);
si->lp_ele = lpNext(si->lp,si->lp_ele); /* Seek ID. */ si->lp_ele = lpNext(si->lp,si->lp_ele); /* Seek ID. */
/* Get the ID: it is encoded as difference between the master /* Get the ID: it is encoded as difference between the master
@ -1274,7 +1274,7 @@ void streamIteratorRemoveEntry(streamIterator *si, streamID *current) {
* deleted entries in the listpack header. * deleted entries in the listpack header.
* *
* We start flagging: */ * We start flagging: */
int flags = lpGetInteger(si->lp_flags); int64_t flags = lpGetInteger(si->lp_flags);
flags |= STREAM_ITEM_FLAG_DELETED; flags |= STREAM_ITEM_FLAG_DELETED;
lp = lpReplaceInteger(lp,&si->lp_flags,flags); lp = lpReplaceInteger(lp,&si->lp_flags,flags);

9
tests/unit/type/stream.tcl
View File

@ -210,6 +210,15 @@ start_server {
assert_equal [r XRANGE mystream - +] {{3-0 {f v}} {4-0 {f v}} {5-0 {f v}}} assert_equal [r XRANGE mystream - +] {{3-0 {f v}} {4-0 {f v}} {5-0 {f v}}}
} }
test {XTRIM with MINID option, big delta from master record} {
r DEL mystream
r XADD mystream 1-0 f v
r XADD mystream 1641544570597-0 f v
r XADD mystream 1641544570597-1 f v
r XTRIM mystream MINID 1641544570597-0
assert_equal [r XRANGE mystream - +] {{1641544570597-0 {f v}} {1641544570597-1 {f v}}}
}
proc insert_into_stream_key {key {count 10000}} { proc insert_into_stream_key {key {count 10000}} {
r multi r multi
for {set j 0} {$j < $count} {incr j} { for {set j 0} {$j < $count} {incr j} {