fix #920 add _isRetry when retrieving from retry queue

sarracenia/flowcb/nodupe/disk.py

@@ -154,8 +154,6 @@ def after_accept(self, worklist):
 
         if self.o.fileAgeMin > 0:
             max_mtime = self.now - self.o.fileAgeMin
-        elif type(self.o.inflight) in [ int, float ] and self.o.inflight > 0:
-            max_mtime = self.now - self.o.inflight
         else:
             # FIXME: should we add some time here to allow for different clocks?
             #        100 seconds in the future? hmm...
@@ -177,7 +175,7 @@ def after_accept(self, worklist):
                     worklist.rejected.append(m)
                     continue
 
-            if self.check_message(m):
+            if '_isRetry' in m or self.check_message(m):
                 new_incoming.append(m)
             else:
                 m['_deleteOnPost'] |= set(['reject'])

sarracenia/flowcb/nodupe/redis.py

@@ -173,7 +173,7 @@ def after_accept(self, worklist):
                     worklist.rejected.append(m)
                     continue
 
-            if self._is_new(m):
+            if '_isRetry' in m or self._is_new(m):
                 new_incoming.append(m)
             else:
                 m['_deleteOnPost'] |= set(['reject'])

sarracenia/flowcb/retry.py

@@ -95,7 +95,9 @@ def gather(self, qty) -> None:
              for k in m:
                  if k in m['_deleteOnPost'] or k.startswith('new_'):
                      del m[k]
-             del m['_deleteOnPost']
+             m['_isRetry'] = True
+             m['_deleteOnPost'] = set( [ '_isRetry' ] )
+
 
         return message_list