diff --git a/minode/connection.py b/minode/connection.py index 3dd5f9a..7603965 100644 --- a/minode/connection.py +++ b/minode/connection.py @@ -472,6 +472,7 @@ class Connection(ConnectionBase): logging.debug('%s:%s -> %s', self.host_print, self.port, obj) self.vectors_requested.pop(obj.vector, None) self.vectors_to_get.discard(obj.vector) + shared.objects.check(obj.vector) if obj.is_valid(): shared.objects[obj.vector] = obj if ( diff --git a/minode/manager.py b/minode/manager.py index b8caae5..f9ebf31 100644 --- a/minode/manager.py +++ b/minode/manager.py @@ -56,6 +56,8 @@ class Manager(threading.Thread): outgoing_connections = 0 for c in shared.connections.copy(): if not c.is_alive() or c.status == 'disconnected': + shared.objects.check( + *(c.vectors_to_get | c.vectors_requested.keys())) with shared.connections_lock: shared.connections.remove(c) else: diff --git a/minode/sql.py b/minode/sql.py index 4790224..991615e 100644 --- a/minode/sql.py +++ b/minode/sql.py @@ -57,6 +57,11 @@ class Inventory(): expires, obj_type, version, stream, data, offset, tag=tag, vector=vector) + def check(self, *vectors): + with self._lock: + for vector in vectors: + self._pending.discard(vector) + def cleanup(self): if len(self._pending) > 100: logging.warning( @@ -233,7 +238,6 @@ class Inventory(): return with self._lock: self._last[vector] = obj - self._pending.discard(vector) def __bool__(self): if self._last: