lpGetInteger returns int64_t, avoid overflow (#10068)

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)

1 files changed, 7 insertions, 7 deletions

diff --git a/src/t_stream.c b/src/t_stream.c
index c49889abe..8c7218e28 100644
--- a/src/t_stream.c
+++ b/src/t_stream.c
@@ -793,13 +793,13 @@ int64_t streamTrim(stream *s, streamAddTrimArgs *args) {
              * update it after (and if) we actually remove the entry */
             unsigned char *pcopy = p;
 
-            int flags = lpGetInteger(p);
+            int64_t flags = lpGetInteger(p);
             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 */
-            int seq_delta = lpGetInteger(p);
+            int64_t seq_delta = lpGetInteger(p);
             p = lpNext(lp, p); /* Skip ID seq delta */
 
             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
              * emitted the current entry, and have to go back to the previous
              * 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);
             /* Seek lp-count of prev entry. */
             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. */
             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. */
 
             /* 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.
      *
      * We start flagging: */
-    int flags = lpGetInteger(si->lp_flags);
+    int64_t flags = lpGetInteger(si->lp_flags);
     flags |= STREAM_ITEM_FLAG_DELETED;
     lp = lpReplaceInteger(lp,&si->lp_flags,flags);