From 3aec98ada9cea8f1acfb553abfb909aa8e8af883 Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Thu, 22 Jun 2023 13:03:52 +0200 Subject: [PATCH] remote logging/progress: use callback to send queued records, fixes #7662 --- src/borg/remote.py | 1 + src/borg/repository.py | 16 ++++++++++++++++ 2 files changed, 17 insertions(+) diff --git a/src/borg/remote.py b/src/borg/remote.py index 398c02456..0bc977865 100644 --- a/src/borg/remote.py +++ b/src/borg/remote.py @@ -358,6 +358,7 @@ def open( storage_quota=self.storage_quota, exclusive=exclusive, make_parent_dirs=make_parent_dirs, + send_log_cb=self.send_queued_log, ) self.repository.__enter__() # clean exit handled by serve() method return self.repository.id diff --git a/src/borg/repository.py b/src/borg/repository.py index 82b7bccdf..c4b0f810d 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -178,10 +178,16 @@ def __init__( append_only=False, storage_quota=None, make_parent_dirs=False, + send_log_cb=None, ): self.path = os.path.abspath(path) self._location = Location("file://%s" % self.path) self.version = None + # long-running repository methods which emit log or progress output are responsible for calling + # the ._send_log method periodically to get log and progress output transferred to the borg client + # in a timely manner, in case we have a RemoteRepository. + # for local repositories ._send_log can be called also (it will just do nothing in that case). + self._send_log = send_log_cb or (lambda: None) self.io = None # type: LoggedIO self.lock = None self.index = None @@ -785,6 +791,7 @@ def complete_xfer(intermediate=True): logger.warning("Segment %d not found, but listed in compaction data", segment) del self.compact[segment] pi.show() + self._send_log() continue segment_size = self.io.segment_size(segment) freeable_ratio = 1.0 * freeable_space / segment_size @@ -798,6 +805,7 @@ def complete_xfer(intermediate=True): freeable_space, ) pi.show() + self._send_log() continue segments.setdefault(segment, 0) logger.debug( @@ -905,7 +913,9 @@ def complete_xfer(intermediate=True): assert segments[segment] == 0, "Corrupted segment reference count - corrupted index or hints" unused.append(segment) pi.show() + self._send_log() pi.finish() + self._send_log() complete_xfer(intermediate=False) self.io.clear_empty_dirs() quota_use_after = self.storage_quota_use @@ -924,6 +934,7 @@ def replay_segments(self, index_transaction_id, segments_transaction_id): ) for i, (segment, filename) in enumerate(self.io.segment_iterator()): pi.show(i) + self._send_log() if index_transaction_id is not None and segment <= index_transaction_id: continue if segment > segments_transaction_id: @@ -931,6 +942,7 @@ def replay_segments(self, index_transaction_id, segments_transaction_id): objects = self.io.iter_objects(segment) self._update_index(segment, objects) pi.finish() + self._send_log() self.write_index() finally: self.exclusive = remember_exclusive @@ -1061,6 +1073,7 @@ def report_error(msg): segment = -1 # avoid uninitialized variable if there are no segment files at all for i, (segment, filename) in enumerate(self.io.segment_iterator()): pi.show(i) + self._send_log() if segment <= last_segment_checked: continue if segment > transaction_id: @@ -1087,6 +1100,7 @@ def report_error(msg): self.save_config(self.path, self.config) pi.finish() + self._send_log() # self.index, self.segments, self.compact now reflect the state of the segment files up to . # We might need to add a commit tag if no committed segment is found. if repair and segments_transaction_id is None: @@ -1110,12 +1124,14 @@ def report_error(msg): current_value = current_index.get(key, not_found) if current_value != value: logger.warning(line_format, bin_to_hex(key), value, current_value) + self._send_log() for key, current_value in current_index.iteritems(): if key in self.index: continue value = self.index.get(key, not_found) if current_value != value: logger.warning(line_format, bin_to_hex(key), value, current_value) + self._send_log() if repair: self.write_index() self.rollback()