[forwarder] Fix error on removal of transactions (#3287)

Fixes `ValueError: list.remove(x): x not in list` error

Fix in 2 parts:

1. Recompute next flush of transactions on flush timeout.

This is needed so that when a flush timeout occurs, the transactions
of that flush do not stay at the head of the queue (if they do stay
at the head of the queue then, when the max queue size is hit, newer
transactions are deleted first, which is not desirable).

Recomputing the next flush time might delay these transactions
slightly (up to MAX_WAIT_FOR_REPLAY seconds, i.e. 90 seconds), but it's
ok here since at this point these transactions have probably been
waiting for some time already.

2. Safely remove transactions from list

So that the forwarder doesn't crash when a removal attempt is made on
a transaction that was already removed.

Author: olivielpeau

transaction.py

@@ -134,9 +134,7 @@ def append(self,tr):
             new_trs = sorted(self._transactions,key=attrgetter('_next_flush'), reverse = True)
             for tr2 in new_trs:
                 if (self._total_size + tr_size) > self._MAX_QUEUE_SIZE:
-                    self._transactions.remove(tr2)
-                    self._total_count = self._total_count - 1
-                    self._total_size = self._total_size - tr2.get_size()
+                    self._remove(tr2)
                     log.warn("Removed transaction %s from queue" % tr2.get_id())
 
         # Done
@@ -148,6 +146,17 @@ def append(self,tr):
         log.debug("Transaction %s added" % (tr.get_id()))
         self.print_queue_stats()
 
+    def _remove(self, tr):
+        '''Safely remove transaction from list'''
+        try:
+            self._transactions.remove(tr)
+        except ValueError:
+            # Should not happen if we order the queue consistently, but we should catch the error anyway
+            log.warn("Tried to remove transaction %s from queue but it was not in the queue anymore.", tr.get_id())
+        else:
+            self._total_count -= 1
+            self._total_size -= tr.get_size()
+
     def flush(self):
 
         if self._trs_to_flush is not None:
@@ -201,6 +210,10 @@ def flush_next(self):
             # Running for too long?
             if datetime.utcnow() - self._flush_time >= self._MAX_FLUSH_DURATION:
                 log.warn('Flush %s is taking more than 10s, stopping it', self._flush_count)
+                for tr in self._trs_to_flush:
+                    # Recompute these transactions' next flush so that if we hit the max queue size
+                    # newer transactions are preserved
+                    tr.compute_next_flush(self._MAX_WAIT_FOR_REPLAY)
                 self._trs_to_flush = []
                 return self.flush_next()
 
@@ -277,9 +290,7 @@ def tr_error_reject_request(self, tr):
                  "It will not be replayed.",
                  tr.get_id(),
                  tr.get_size() / 1024)
-        self._transactions.remove(tr)
-        self._total_count -= 1
-        self._total_size -= tr.get_size()
+        self._remove(tr)
         self._transactions_flushed += 1
         self.print_queue_stats()
         self._transactions_rejected += 1
@@ -295,8 +306,6 @@ def tr_success(self, tr):
         self._running_flushes -= 1
         self._finished_flushes += 1
         log.debug("Transaction %d completed",  tr.get_id())
-        self._transactions.remove(tr)
-        self._total_count -= 1
-        self._total_size -= tr.get_size()
+        self._remove(tr)
         self._transactions_flushed += 1
         self.print_queue_stats()