Fix a race condition in which borgmatic sometimes swallows Borg error output without logging it (#1256).

NEWS

@@ -1,4 +1,6 @@
 2.1.6.dev0
+ * #1256: Fix a race condition in which borgmatic sometimes swallows Borg error output without
+   logging it.
  * #1300: Fix the "source_directories_must_exist" option to support source directories relative to a
    "working_directory".
  * #1301: Expand the "patterns_from" and "exclude_from" options to support paths containing tildes

borgmatic/execute.py

@@ -372,13 +372,21 @@ def log_buffer_lines(
                 yield log_record.getMessage()
 
 
-def raise_for_process_errors(buffer_readers, process_metadatas, borg_local_path, borg_exit_codes):
+def raise_for_process_errors(
+    buffer_readers,
+    process_metadatas,
+    borg_local_path,
+    borg_exit_codes,
+    output_log_level=None,
+    capture_stderr=False,
+):
     '''
     Given a dict from buffer object to Buffer_reader, a dict from subprocess.Popen() instance to
-    Process_metadata instance, Borg's local path, a sequence of exit code configuration dicts, check
-    the given processes for error or warning exit codes. If found, vent or kill any running
-    processes. In the case of an error exit code, raise. In the case of warning, return
-    Exit_status.WARNING. Otherwise, return None.
+    Process_metadata instance, a requested output log level for stdout, Borg's local path, a
+    sequence of exit code configuration dicts, and whether to capture stderr, check the given
+    processes for error or warning exit codes. If found, vent or kill any running processes and
+    drain any remaining buffer lines. In the case of an error exit code, raise.  In the case of
+    warning, return Exit_status.WARNING. Otherwise, return None.
     '''
     result_status = None
 
@@ -404,6 +412,13 @@ def raise_for_process_errors(buffer_readers, process_metadatas, borg_local_path,
             result_status = Exit_status.WARNING
             continue
 
+        # Drain and log remaining buffer lines, so no output gets lost. But swallow captured output,
+        # since we're raising below instead of yielding captured lines.
+        tuple(
+            log_remaining_buffer_lines(
+                buffer_readers, process_metadatas, output_log_level, borg_local_path, capture_stderr
+            )
+        )
         last_lines = process_metadatas[process].last_lines
 
         # If an error occurs, include its output in the raised exception so that we don't
@@ -429,6 +444,9 @@ def log_remaining_buffer_lines(
     whether to capture stderr, drain and log any remaining output lines from the buffers until
     they're empty. Additionally, for any log records with a log level the same as the output log
     level, yield those log messages for capture.
+
+    The main difference between this function and log_buffer_lines() is that this one doesn't just
+    do one "turn of the crank" of logging buffer output; it completely drains any remaining output.
     '''
     for output_buffer, reader in buffer_readers.items():
         if not reader.process:
@@ -446,6 +464,7 @@ def log_remaining_buffer_lines(
                         borg_local_path=borg_local_path,
                         command=reader.process.args,
                     ),
+                    last_lines=process_metadatas[reader.process].last_lines,
                 )
 
                 if (
@@ -494,7 +513,7 @@ def log_outputs(
         for buffer in output_buffers_for_process(process, exclude_stdouts)
     }
 
-    # Log output lines for each process until they all exit.
+    # Log output lines for each process until they all exit or one errors.
     while True:
         yield from log_buffer_lines(
             buffer_readers, process_metadatas, output_log_level, borg_local_path, capture_stderr
@@ -502,7 +521,12 @@ def log_outputs(
 
         if (
             raise_for_process_errors(
-                buffer_readers, process_metadatas, borg_local_path, borg_exit_codes
+                buffer_readers,
+                process_metadatas,
+                borg_local_path,
+                borg_exit_codes,
+                output_log_level,
+                capture_stderr,
             )
             == Exit_status.WARNING
         ):

tests/integration/test_execute.py

@@ -557,6 +557,43 @@ def test_log_outputs_with_unfinished_process_re_polls():
     )
 
 
+def test_log_outputs_includes_error_output_when_output_spans_multiple_chunks():
+    flexmock(module.logger).should_receive('log')
+    flexmock(module).should_receive('interpret_exit_code').and_return(module.Exit_status.ERROR)
+    flexmock(module).should_receive('command_for_process').and_return('python')
+
+    process = subprocess.Popen(
+        [
+            sys.executable,
+            '-c',
+            (
+                'import os, sys; '
+                f'os.write(sys.stdout.fileno(), b"x" * {module.READ_CHUNK_SIZE + 10}); '
+                'os.write(sys.stdout.fileno(), b"\\nERROR: critical failure"); '
+                'os.close(1); '
+                'os._exit(2)'
+            ),
+        ],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+    )
+    flexmock(module).should_receive('output_buffers_for_process').and_return((process.stdout,))
+
+    with pytest.raises(subprocess.CalledProcessError) as error:
+        tuple(
+            module.log_outputs(
+                (process,),
+                exclude_stdouts=(),
+                output_log_level=logging.INFO,
+                borg_local_path='borg',
+                borg_exit_codes=None,
+            )
+        )
+
+    assert error.value.output
+    assert 'ERROR: critical failure' in error.value.output
+
+
 def test_read_lines_uses_system_locale_when_decoding_output():
     flexmock(module.locale).should_receive('getpreferredencoding').and_return('ISO-8859-1')
     process = subprocess.Popen(['echo', b'\xc4pple'], stdout=subprocess.PIPE)

tests/unit/test_execute.py

@@ -860,6 +860,7 @@ def test_raise_for_process_errors_with_error_process_raises():
         process: module.Process_metadata(last_lines=['hi', 'there'], capture=False)
     }
     flexmock(module).should_receive('interpret_exit_code').and_return(module.Exit_status.ERROR)
+    flexmock(module).should_receive('log_remaining_buffer_lines').and_return(())
     command = flexmock()
     flexmock(module).should_receive('command_for_process').and_return(command)
 
@@ -890,6 +891,7 @@ def test_raise_for_process_errors_with_success_process_and_warning_process_retur
     flexmock(module).should_receive('interpret_exit_code').with_args(
         object, 1, object, object
     ).and_return(module.Exit_status.WARNING)
+    flexmock(module).should_receive('log_remaining_buffer_lines').and_return(())
     flexmock(module).should_receive('command_for_process').never()
 
     assert (
@@ -917,6 +919,7 @@ def test_raise_for_process_errors_with_warning_process_and_error_process_raises(
     flexmock(module).should_receive('interpret_exit_code').with_args(
         object, 3, object, object
     ).and_return(module.Exit_status.ERROR)
+    flexmock(module).should_receive('log_remaining_buffer_lines').and_return(())
     command = flexmock()
     flexmock(module).should_receive('command_for_process').and_return(command)
 
@@ -947,6 +950,7 @@ def test_raise_for_process_errors_with_warning_process_and_running_process_kills
     flexmock(module).should_receive('interpret_exit_code').with_args(
         object, 1, object, object
     ).and_return(module.Exit_status.WARNING)
+    flexmock(module).should_receive('log_remaining_buffer_lines').and_return(())
     command = flexmock()
     flexmock(module).should_receive('command_for_process').and_return(command)
 
@@ -975,6 +979,7 @@ def test_raise_for_process_errors_with_error_process_and_running_process_kills_a
     flexmock(module).should_receive('interpret_exit_code').with_args(
         object, 3, object, object
     ).and_return(module.Exit_status.ERROR)
+    flexmock(module).should_receive('log_remaining_buffer_lines').and_return(())
     command = flexmock()
     flexmock(module).should_receive('command_for_process').and_return(command)
 
@@ -998,6 +1003,7 @@ def test_raise_for_process_errors_with_warning_process_and_long_output_raises_wi
         process: module.Process_metadata(last_lines=['hi', 'there'], capture=False)
     }
     flexmock(module).should_receive('interpret_exit_code').and_return(module.Exit_status.ERROR)
+    flexmock(module).should_receive('log_remaining_buffer_lines').and_return(())
     command = flexmock()
     flexmock(module).should_receive('command_for_process').and_return(command)