Need to make sure we advance sequence correctly when no items

Fixes #34
mhart · Jun 30, 2016 · 7384fcc · 7384fcc
1 parent cf059b4
commit 7384fcc
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 6 deletions.
diff --git a/actions/getRecords.js b/actions/getRecords.js
@@ -32,7 +32,7 @@ module.exports = function getRecords(store, data, cb) {
   if (!/^shardId-[\d]{12}$/.test(shardId) || !(shardIx >= 0 && shardIx < 2147483648))
     return cb(invalidShardIterator())
 
-  if (!(iteratorTime > 0 && iteratorTime <= Date.now()))
+  if (!(iteratorTime > 0 && iteratorTime <= now))
     return cb(invalidShardIterator())
 
   if (!/[a-zA-Z0-9_.-]+/.test(streamName) || !streamName.length || streamName.length > 128)
@@ -67,7 +67,7 @@ module.exports = function getRecords(store, data, cb) {
 
     cb = once(cb)
 
-    var streamDb = store.getStreamDb(streamName), cutoffTime = Date.now() - (stream.RetentionPeriodHours * 60 * 60 * 1000),
+    var streamDb = store.getStreamDb(streamName), cutoffTime = now - (stream.RetentionPeriodHours * 60 * 60 * 1000),
       keysToDelete = [], lastItem, opts
 
     opts = {
@@ -87,15 +87,15 @@ module.exports = function getRecords(store, data, cb) {
       })
       .filter(function(item) { return !item._tooOld })
       .join(function(items) {
-        var nextSeq = lastItem ? db.incrementSequence(lastItem._seqObj) : seqNo,
+        var nextSeq = db.incrementSequence(lastItem ? lastItem._seqObj : seqObj, lastItem ? null : now),
           nextShardIterator = db.createShardIterator(streamName, shardId, nextSeq),
           millisBehind = 0
 
         if (!items.length && stream.Shards[shardIx].SequenceNumberRange.EndingSequenceNumber) {
           var endSeqObj = db.parseSequence(stream.Shards[shardIx].SequenceNumberRange.EndingSequenceNumber)
           if (seqObj.seqTime >= endSeqObj.seqTime) {
             nextShardIterator = undefined
-            millisBehind = Math.max(0, Date.now() - endSeqObj.seqTime)
+            millisBehind = Math.max(0, now - endSeqObj.seqTime)
           }
         }
 

diff --git a/db/index.js b/db/index.js
@@ -199,13 +199,13 @@ function stringifySequence(obj) {
   }
 }
 
-function incrementSequence(seqObj) {
+function incrementSequence(seqObj, seqTime) {
   if (typeof seqObj == 'string') seqObj = parseSequence(seqObj)
 
   return stringifySequence({
     shardCreateTime: seqObj.shardCreateTime,
     seqIx: seqObj.seqIx,
-    seqTime: seqObj.seqTime + 1000,
+    seqTime: seqTime || (seqObj.seqTime + 1000),
     shardIx: seqObj.shardIx,
   })
 }