From 677871aa8974689c2cb8f65c0eac810b485b5e6a Mon Sep 17 00:00:00 2001
From: Dan Helfman <witten@torsion.org>
Date: Mon, 19 Jan 2026 16:26:22 -0800
Subject: [PATCH 1/8] Initial stab at fixing spot check hang by actually
 draining and consuming buffers after a process exits (#1242).

---
 borgmatic/execute.py | 76 +++++++++++++++++++++++++++++---------------
 1 file changed, 50 insertions(+), 26 deletions(-)

diff --git a/borgmatic/execute.py b/borgmatic/execute.py
index 34fdebbe..777a95fc 100644
--- a/borgmatic/execute.py
+++ b/borgmatic/execute.py
@@ -234,7 +234,7 @@ def log_outputs(  # noqa: PLR0912
     still_running = True
 
     # Log output for each process until they all exit.
-    while True:  # noqa: PLR1702
+    while True:
         if output_buffers:
             (ready_buffers, _, _) = select.select(output_buffers, [], [])
 
@@ -254,34 +254,33 @@ def log_outputs(  # noqa: PLR0912
                             # Add the process's output to output_buffers to ensure it'll get read.
                             output_buffers.append(other_process.stdout)
 
-                while True:
-                    line = ready_buffer.readline().rstrip().decode()
-                    if not line or not ready_process:
-                        break
+                line = ready_buffer.readline().rstrip().decode()
+                if not line or not ready_process:
+                    continue
 
-                    command = (
-                        ready_process.args.split(' ')
-                        if isinstance(ready_process.args, str)
-                        else ready_process.args
-                    )
+                command = (
+                    ready_process.args.split(' ')
+                    if isinstance(ready_process.args, str)
+                    else ready_process.args
+                )
 
-                    # Keep the last few lines of output in case the process errors and we need the
-                    # output for the exception below.
-                    log_record = handle_log_record(
-                        parse_log_line(
-                            line=line,
-                            log_level=output_log_level,
-                            elevate_stderr=(
-                                ready_buffer == ready_process.stderr and not capture_stderr
-                            ),
-                            borg_local_path=borg_local_path,
-                            command=command,
+                # Keep the last few lines of output in case the process errors and we need the
+                # output for the exception below.
+                log_record = handle_log_record(
+                    parse_log_line(
+                        line=line,
+                        log_level=output_log_level,
+                        elevate_stderr=(
+                            ready_buffer == ready_process.stderr and not capture_stderr
                         ),
-                        last_lines=process_last_lines[ready_process],
-                    )
+                        borg_local_path=borg_local_path,
+                        command=command,
+                    ),
+                    last_lines=process_last_lines[ready_process],
+                )
 
-                    if log_record.levelno is None and ready_process == process_to_capture:
-                        yield log_record.getMessage()
+                if log_record.levelno is None and ready_process == process_to_capture:
+                    yield log_record.getMessage()
 
         if not still_running:
             break
@@ -293,7 +292,6 @@ def log_outputs(  # noqa: PLR0912
 
             if exit_code is None:
                 still_running = True
-                command = process.args.split(' ') if isinstance(process.args, str) else process.args
                 continue
 
             command = process.args.split(' ') if isinstance(process.args, str) else process.args
@@ -347,6 +345,32 @@ def log_outputs(  # noqa: PLR0912
                 still_running = False
                 break
 
+        if still_running is True:
+            continue
+
+        for output_buffer in output_buffers:
+            process = process_for_output_buffer.get(output_buffer)
+            last_lines = process_last_lines[process]
+
+            for line in output_buffer.readlines():
+                log_record = handle_log_record(
+                    parse_log_line(
+                        line=line.rstrip().decode(),
+                        log_level=output_log_level,
+                        elevate_stderr=(
+                            output_buffer == process.stderr and not capture_stderr
+                        ),
+                        borg_local_path=borg_local_path,
+                        command=command,
+                    ),
+                    last_lines=last_lines,
+                )
+
+                if log_record.levelno is None and process == process_to_capture:
+                    yield log_record.getMessage()
+
+        return
+
 
 SECRET_COMMAND_FLAG_NAMES = {'--password'}
 

From 87c5863218a572cd3d85900bf66d19fe66ca81b3 Mon Sep 17 00:00:00 2001
From: Dan Helfman <witten@torsion.org>
Date: Mon, 19 Jan 2026 22:22:15 -0800
Subject: [PATCH 2/8] Some cleanup and also fix delayed logs (#1242).

---
 borgmatic/execute.py | 62 ++++++++++++++++++++++----------------------
 1 file changed, 31 insertions(+), 31 deletions(-)

diff --git a/borgmatic/execute.py b/borgmatic/execute.py
index 777a95fc..00beffc6 100644
--- a/borgmatic/execute.py
+++ b/borgmatic/execute.py
@@ -231,10 +231,9 @@ def log_outputs(  # noqa: PLR0912
     }
     output_buffers = list(process_for_output_buffer.keys())
     process_to_capture = processes[-1]
-    still_running = True
 
     # Log output for each process until they all exit.
-    while True:
+    while any(process.poll() is None for process in processes):
         if output_buffers:
             (ready_buffers, _, _) = select.select(output_buffers, [], [])
 
@@ -280,18 +279,18 @@ def log_outputs(  # noqa: PLR0912
                 )
 
                 if log_record.levelno is None and ready_process == process_to_capture:
+                    print('***', log_record.getMessage())
                     yield log_record.getMessage()
-
-        if not still_running:
-            break
-
-        still_running = False
+                else:
+                    # Yield an empty string just to keep the generator executing. Without this,
+                    # logged lines might not get processed by the caller until the next yielded
+                    # message.
+                    yield ''
 
         for process in processes:
             exit_code = process.poll() if output_buffers else process.wait()
 
             if exit_code is None:
-                still_running = True
                 continue
 
             command = process.args.split(' ') if isinstance(process.args, str) else process.args
@@ -323,7 +322,10 @@ def log_outputs(  # noqa: PLR0912
                         )
 
                         if log_record.levelno is None and process == process_to_capture:
+                            print('**e', log_record.getMessage())
                             yield log_record.getMessage()
+                        else:
+                            yield
 
                 if len(last_lines) == ERROR_OUTPUT_MAX_LINE_COUNT:
                     last_lines.insert(0, '...')
@@ -342,34 +344,32 @@ def log_outputs(  # noqa: PLR0912
                         '\n'.join(last_lines),
                     )
 
-                still_running = False
                 break
 
-        if still_running is True:
-            continue
+    # Now that all processes have exited, drain and consume any last output.
+    for output_buffer in output_buffers:
+        process = process_for_output_buffer.get(output_buffer)
+        last_lines = process_last_lines[process]
 
-        for output_buffer in output_buffers:
-            process = process_for_output_buffer.get(output_buffer)
-            last_lines = process_last_lines[process]
-
-            for line in output_buffer.readlines():
-                log_record = handle_log_record(
-                    parse_log_line(
-                        line=line.rstrip().decode(),
-                        log_level=output_log_level,
-                        elevate_stderr=(
-                            output_buffer == process.stderr and not capture_stderr
-                        ),
-                        borg_local_path=borg_local_path,
-                        command=command,
+        for line in output_buffer.readlines():
+            log_record = handle_log_record(
+                parse_log_line(
+                    line=line.rstrip().decode(),
+                    log_level=output_log_level,
+                    elevate_stderr=(
+                        output_buffer == process.stderr and not capture_stderr
                     ),
-                    last_lines=last_lines,
-                )
+                    borg_local_path=borg_local_path,
+                    command=command,
+                ),
+                last_lines=last_lines,
+            )
 
-                if log_record.levelno is None and process == process_to_capture:
-                    yield log_record.getMessage()
-
-        return
+            if log_record.levelno is None and process == process_to_capture:
+                yield log_record.getMessage()
+                print('**d', log_record.getMessage())
+            else:
+                yield ''
 
 
 SECRET_COMMAND_FLAG_NAMES = {'--password'}

From d5cd4efecdfd54dd67839ad9acf30091cb422174 Mon Sep 17 00:00:00 2001
From: Dan Helfman <witten@torsion.org>
Date: Thu, 22 Jan 2026 10:27:47 -0800
Subject: [PATCH 3/8] Fix spot check hang (#1242).

---
 borgmatic/execute.py              | 173 ++++++++++++++++++------------
 tests/integration/test_execute.py |  45 +++-----
 2 files changed, 117 insertions(+), 101 deletions(-)

diff --git a/borgmatic/execute.py b/borgmatic/execute.py
index 00beffc6..2446ba64 100644
--- a/borgmatic/execute.py
+++ b/borgmatic/execute.py
@@ -3,6 +3,7 @@ import contextlib
 import enum
 import json
 import logging
+import os
 import select
 import subprocess
 import textwrap
@@ -196,6 +197,50 @@ def handle_log_record(log_record, last_lines):
     return log_record
 
 
+READ_CHUNK_SIZE = 4096
+
+
+def read_lines(buffer, process, line_separator='\n'):
+    '''
+    Given a Python buffer (like stdout) ready for reading, its process, and a line separator,
+    repeatedly yield a tuple of (decoded) lines from the buffer until the process has exited.
+
+    It is assumed that this function's generator is used in conjunction with an external select()
+    call to know when to read more lines. Otherwise, the generator will busywait if it's called in a
+    tight loop.
+    '''
+    data = ''
+
+    while True:
+        #chunk = buffer.read(READ_CHUNK_SIZE).decode()
+        chunk = os.read(buffer.fileno(), READ_CHUNK_SIZE).decode()
+
+        if not chunk:  # EOF
+            # The process is still running, so we keep running too.
+            if process.poll() is None:
+                continue
+
+            break
+
+        data += chunk
+        lines = []
+
+        while True:
+            separator_position = data.find(line_separator)
+
+            if separator_position == -1:
+                break
+
+            lines.append(data[:separator_position].rstrip())
+            data = data[separator_position + 1 :]
+
+        yield tuple(lines)
+
+    # Yield any leftover data from the end of the buffer.
+    if data:
+        yield tuple(data.rstrip().splitlines())
+
+
 def log_outputs(  # noqa: PLR0912
     processes,
     exclude_stdouts,
@@ -231,6 +276,12 @@ def log_outputs(  # noqa: PLR0912
     }
     output_buffers = list(process_for_output_buffer.keys())
     process_to_capture = processes[-1]
+    reader_for_output_buffer = {
+        buffer: read_lines(buffer, process)
+        for process in processes
+        if process.stdout or process.stderr
+        for buffer in output_buffers_for_process(process, exclude_stdouts)
+    }
 
     # Log output for each process until they all exit.
     while any(process.poll() is None for process in processes):
@@ -252,40 +303,44 @@ def log_outputs(  # noqa: PLR0912
                         ):
                             # Add the process's output to output_buffers to ensure it'll get read.
                             output_buffers.append(other_process.stdout)
+                            process_for_output_buffer[other_process.stdout] = other_process
+                            reader_for_output_buffer[other_process.stdout] = read_lines(
+                                other_process.stdout, other_process
+                            )
 
-                line = ready_buffer.readline().rstrip().decode()
-                if not line or not ready_process:
+                try:
+                    lines = next(reader_for_output_buffer[ready_buffer])
+                except StopIteration:
                     continue
 
-                command = (
-                    ready_process.args.split(' ')
-                    if isinstance(ready_process.args, str)
-                    else ready_process.args
-                )
+                for line in lines:
+                    if not line or not ready_process:
+                        continue
 
-                # Keep the last few lines of output in case the process errors and we need the
-                # output for the exception below.
-                log_record = handle_log_record(
-                    parse_log_line(
-                        line=line,
-                        log_level=output_log_level,
-                        elevate_stderr=(
-                            ready_buffer == ready_process.stderr and not capture_stderr
+                    command = (
+                        ready_process.args.split(' ')
+                        if isinstance(ready_process.args, str)
+                        else ready_process.args
+                    )
+
+                    # Keep the last few lines of output in case the process errors and we need the
+                    # output for the exception below.
+                    log_record = handle_log_record(
+                        parse_log_line(
+                            line=line,
+                            log_level=output_log_level,
+                            elevate_stderr=(
+                                ready_buffer == ready_process.stderr and not capture_stderr
+                            ),
+                            borg_local_path=borg_local_path,
+                            command=command,
                         ),
-                        borg_local_path=borg_local_path,
-                        command=command,
-                    ),
-                    last_lines=process_last_lines[ready_process],
-                )
+                        last_lines=process_last_lines[ready_process],
+                    )
 
-                if log_record.levelno is None and ready_process == process_to_capture:
-                    print('***', log_record.getMessage())
-                    yield log_record.getMessage()
-                else:
-                    # Yield an empty string just to keep the generator executing. Without this,
-                    # logged lines might not get processed by the caller until the next yielded
-                    # message.
-                    yield ''
+                    if log_record.levelno is None and ready_process == process_to_capture:
+                        print('***', log_record.getMessage())
+                        yield log_record.getMessage()
 
         for process in processes:
             exit_code = process.poll() if output_buffers else process.wait()
@@ -301,32 +356,6 @@ def log_outputs(  # noqa: PLR0912
 
                 # If an error occurs, include its output in the raised exception so that we don't
                 # inadvertently hide error output.
-                for output_buffer in output_buffers_for_process(process, exclude_stdouts):
-                    # Collect any straggling output lines that came in since we last gathered output.
-                    while output_buffer:  # pragma: no cover
-                        line = output_buffer.readline().rstrip().decode()
-                        if not line:
-                            break
-
-                        log_record = handle_log_record(
-                            parse_log_line(
-                                line=line,
-                                log_level=output_log_level,
-                                elevate_stderr=(
-                                    output_buffer == process.stderr and not capture_stderr
-                                ),
-                                borg_local_path=borg_local_path,
-                                command=command,
-                            ),
-                            last_lines=last_lines,
-                        )
-
-                        if log_record.levelno is None and process == process_to_capture:
-                            print('**e', log_record.getMessage())
-                            yield log_record.getMessage()
-                        else:
-                            yield
-
                 if len(last_lines) == ERROR_OUTPUT_MAX_LINE_COUNT:
                     last_lines.insert(0, '...')
 
@@ -349,27 +378,29 @@ def log_outputs(  # noqa: PLR0912
     # Now that all processes have exited, drain and consume any last output.
     for output_buffer in output_buffers:
         process = process_for_output_buffer.get(output_buffer)
+
+        if not process:
+            continue
+
+        command = process.args.split(' ') if isinstance(process.args, str) else process.args
         last_lines = process_last_lines[process]
 
-        for line in output_buffer.readlines():
-            log_record = handle_log_record(
-                parse_log_line(
-                    line=line.rstrip().decode(),
-                    log_level=output_log_level,
-                    elevate_stderr=(
-                        output_buffer == process.stderr and not capture_stderr
+        for lines in reader_for_output_buffer[output_buffer]:
+            for line in lines:
+                log_record = handle_log_record(
+                    parse_log_line(
+                        line=line.rstrip(),
+                        log_level=output_log_level,
+                        elevate_stderr=(output_buffer == process.stderr and not capture_stderr),
+                        borg_local_path=borg_local_path,
+                        command=command,
                     ),
-                    borg_local_path=borg_local_path,
-                    command=command,
-                ),
-                last_lines=last_lines,
-            )
+                    last_lines=last_lines,
+                )
 
-            if log_record.levelno is None and process == process_to_capture:
-                yield log_record.getMessage()
-                print('**d', log_record.getMessage())
-            else:
-                yield ''
+                if log_record.levelno is None and process == process_to_capture:
+                    print('***', log_record.getMessage())
+                    yield log_record.getMessage()
 
 
 SECRET_COMMAND_FLAG_NAMES = {'--password'}
diff --git a/tests/integration/test_execute.py b/tests/integration/test_execute.py
index 2d58bd76..c38493be 100644
--- a/tests/integration/test_execute.py
+++ b/tests/integration/test_execute.py
@@ -269,7 +269,7 @@ def test_log_outputs_kills_other_processes_and_raises_when_one_errors():
         other_process,
         (),
     ).and_return((other_process.stdout,))
-    flexmock(other_process).should_receive('kill').once()
+    flexmock(other_process).should_call('kill').once()
 
     with pytest.raises(subprocess.CalledProcessError) as error:
         tuple(
@@ -291,12 +291,6 @@ def test_log_outputs_kills_other_processes_and_returns_when_one_exits_with_warni
     flexmock(module).should_receive('command_for_process').and_return('grep')
 
     process = subprocess.Popen(['grep'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
-    flexmock(module).should_receive('interpret_exit_code').with_args(
-        ['grep'],
-        None,
-        'borg',
-        None,
-    ).and_return(module.Exit_status.SUCCESS)
     flexmock(module).should_receive('interpret_exit_code').with_args(
         ['grep'],
         2,
@@ -313,7 +307,7 @@ def test_log_outputs_kills_other_processes_and_returns_when_one_exits_with_warni
         None,
         'borg',
         None,
-    ).and_return(module.Exit_status.SUCCESS)
+    ).and_return(module.Exit_status.STILL_RUNNING)
     flexmock(module).should_receive('output_buffers_for_process').with_args(process, ()).and_return(
         (process.stdout,),
     )
@@ -321,7 +315,7 @@ def test_log_outputs_kills_other_processes_and_returns_when_one_exits_with_warni
         other_process,
         (),
     ).and_return((other_process.stdout,))
-    flexmock(other_process).should_receive('kill').once()
+    flexmock(other_process).should_call('kill').once()
 
     assert (
         tuple(
@@ -372,18 +366,15 @@ def test_log_outputs_vents_other_processes_when_one_exits():
     ).and_return((other_process.stdout,))
     flexmock(process.stdout).should_call('readline').at_least().once()
 
-    assert (
-        tuple(
-            module.log_outputs(
-                (process, other_process),
-                exclude_stdouts=(process.stdout,),
-                output_log_level=logging.INFO,
-                borg_local_path='borg',
-                borg_exit_codes=None,
-            )
+    assert tuple(
+        module.log_outputs(
+            (process, other_process),
+            exclude_stdouts=(process.stdout,),
+            output_log_level=logging.INFO,
+            borg_local_path='borg',
+            borg_exit_codes=None,
         )
-        == ()
-    )
+    ) == ('',)
 
 
 def test_log_outputs_does_not_error_when_one_process_exits():
@@ -433,12 +424,6 @@ def test_log_outputs_truncates_long_error_output():
     flexmock(module).should_receive('command_for_process').and_return('grep')
 
     process = subprocess.Popen(['grep'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
-    flexmock(module).should_receive('interpret_exit_code').with_args(
-        ['grep'],
-        None,
-        'borg',
-        None,
-    ).and_return(module.Exit_status.SUCCESS)
     flexmock(module).should_receive('interpret_exit_code').with_args(
         ['grep'],
         2,
@@ -458,8 +443,8 @@ def test_log_outputs_truncates_long_error_output():
             )
         )
 
-    assert error.value.returncode == 2
-    assert error.value.output.startswith('...')
+        assert error.value.returncode == 2
+        assert error.value.output.startswith('...')
 
 
 def test_log_outputs_with_no_output_logs_nothing():
@@ -487,8 +472,8 @@ def test_log_outputs_with_unfinished_process_re_polls():
     flexmock(module.logger).should_receive('log').never()
     flexmock(module).should_receive('interpret_exit_code').and_return(module.Exit_status.SUCCESS)
 
-    process = subprocess.Popen(['true'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
-    flexmock(process).should_receive('poll').and_return(None).and_return(0).times(3)
+    process = subprocess.Popen(['sleep', '0.001'], stdout=subprocess.PIPE)
+    flexmock(process).should_call('poll').at_least().times(3)
     flexmock(module).should_receive('output_buffers_for_process').and_return((process.stdout,))
 
     assert (

From bd051beceda34c61466719f6187ffb149b079465 Mon Sep 17 00:00:00 2001
From: Dan Helfman <witten@torsion.org>
Date: Thu, 22 Jan 2026 14:02:48 -0800
Subject: [PATCH 4/8] Structural refactor just to get code out of log_outputs()
 and into separate utility functions (#1242).

---
 borgmatic/execute.py | 310 ++++++++++++++++++++++++-------------------
 1 file changed, 176 insertions(+), 134 deletions(-)

diff --git a/borgmatic/execute.py b/borgmatic/execute.py
index 2446ba64..acda5dce 100644
--- a/borgmatic/execute.py
+++ b/borgmatic/execute.py
@@ -42,7 +42,9 @@ def interpret_exit_code(command, exit_code, borg_local_path=None, borg_exit_code
     if exit_code == 0:
         return Exit_status.SUCCESS
 
-    if borg_local_path and command[0] == borg_local_path:
+    parsed_command = command.args.split(' ', 1) if isinstance(command, str) else command
+
+    if borg_local_path and parsed_command[0] == borg_local_path:
         # First try looking for the exit code in the borg_exit_codes configuration.
         for entry in borg_exit_codes or ():
             if entry.get('code') == exit_code:
@@ -164,7 +166,9 @@ def parse_log_line(line, log_level, elevate_stderr, borg_local_path, command):
     came from stderr and the string "warning:" appears at the start of the log line. In that case,
     just elevate the log level to a WARN.
     '''
-    if borg_local_path and command[0] == borg_local_path:
+    parsed_command = command.split(' ', 1) if isinstance(command, str) else command
+
+    if borg_local_path and parsed_command[0] == borg_local_path:
         log_record = borg_json_log_line_to_record(line, log_level)
 
         if log_record:
@@ -178,18 +182,20 @@ def parse_log_line(line, log_level, elevate_stderr, borg_local_path, command):
     return log_line_to_record(line, log_level)
 
 
-def handle_log_record(log_record, last_lines):
+def handle_log_record(log_record, last_lines=None):
     '''
     Given a log record to be logged and a rolling list of last lines, append the record's message to
-    the last lines. Then (if the log level is not None), log the record.
+    the last lines (if given). Then (if the log level is not None), log the record.
 
     Return the log record.
     '''
     log_message = log_record.getMessage()
-    last_lines.append(log_message)
 
-    if len(last_lines) > ERROR_OUTPUT_MAX_LINE_COUNT:
-        last_lines.pop(0)
+    if last_lines is not None:
+        last_lines.append(log_message)
+
+        if len(last_lines) > ERROR_OUTPUT_MAX_LINE_COUNT:
+            last_lines.pop(0)
 
     if log_record.levelno is not None:
         logger.handle(log_record)
@@ -212,7 +218,6 @@ def read_lines(buffer, process, line_separator='\n'):
     data = ''
 
     while True:
-        #chunk = buffer.read(READ_CHUNK_SIZE).decode()
         chunk = os.read(buffer.fileno(), READ_CHUNK_SIZE).decode()
 
         if not chunk:  # EOF
@@ -225,6 +230,8 @@ def read_lines(buffer, process, line_separator='\n'):
         data += chunk
         lines = []
 
+        # Split the data into lines, holding back anything leftover that might
+        # be a partial line.
         while True:
             separator_position = data.find(line_separator)
 
@@ -238,10 +245,157 @@ def read_lines(buffer, process, line_separator='\n'):
 
     # Yield any leftover data from the end of the buffer.
     if data:
-        yield tuple(data.rstrip().splitlines())
+        yield (data.rstrip(),)
 
 
-def log_outputs(  # noqa: PLR0912
+Buffer_reader = collections.namedtuple(
+    'Buffer_reader',
+    ('lines', 'process'),
+)
+
+
+def log_buffer_lines(
+    buffer_readers, processes, output_log_level, borg_local_path, capture_stderr=False
+):
+    '''
+    Given a dict from buffer object to Buffer_reader, a sequence of subprocess.Popen() instances for
+    the processes corresponding to those buffers, a requested output log level for stdout, Borg's
+    local path, and whether to capture stderr, read and log any ready output lines from the buffers.
+    Additionally, if the log level is None for any log record, then yield those log messages for
+    capture.
+
+    This function just does one "turn of the crank" on logging buffer output. It is intended to be
+    called repeatedly to continue to process buffers.
+    '''
+    if not buffer_readers:
+        return
+
+    (ready_buffers, _, _) = select.select(buffer_readers.keys(), [], [])
+
+    for ready_buffer in ready_buffers:
+        reader = buffer_readers[ready_buffer]
+
+        # The "ready" process has exited, but it might be a pipe destination with other
+        # processes (pipe sources) waiting to be read from. So as a measure to prevent
+        # hangs, vent all processes when one exits.
+        if reader.process and reader.process.poll() is not None:
+            for other_process in processes:
+                if (
+                    other_process.poll() is None
+                    and other_process.stdout
+                    and other_process.stdout not in buffer_readers
+                ):
+                    # Add the process's output to buffer_readers to ensure it'll get read.
+                    buffer_readers[other_process.stdout] = Buffer_reader(
+                        read_lines(other_process.stdout, other_process), other_process
+                    )
+
+        try:
+            lines = next(reader.lines)
+        except StopIteration:
+            continue
+
+        for line in lines:
+            if not line or not reader.process:
+                continue
+
+            # Keep the last few lines of output in case the process errors and we need the
+            # output for the exception below.
+            log_record = handle_log_record(
+                parse_log_line(
+                    line=line,
+                    log_level=output_log_level,
+                    elevate_stderr=(ready_buffer == reader.process.stderr and not capture_stderr),
+                    borg_local_path=borg_local_path,
+                    command=reader.process.args,
+                ),
+            )
+
+            # By convention, assume that we're capturing only the last process in the sequence.
+            if log_record.levelno is None and reader.process == processes[-1]:
+                print('***', log_record.getMessage())
+                yield log_record.getMessage()
+
+
+def raise_for_process_errors(buffer_readers, process_last_lines, borg_local_path, borg_exit_codes):
+    '''
+    Given a dict from buffer object to Buffer_reader, a map from subprocess.Popen() instance to a
+    sequence of last lines for that process, 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 Exit_status.STILL_RUNNING.
+    '''
+    for process in process_last_lines.keys():
+        exit_code = process.poll() if buffer_readers else process.wait()
+
+        if exit_code is None:
+            continue
+
+        exit_status = interpret_exit_code(process.args, exit_code, borg_local_path, borg_exit_codes)
+
+        if exit_status not in {Exit_status.ERROR, Exit_status.WARNING}:
+            continue
+
+        last_lines = process_last_lines[process]
+
+        # If an error occurs, include its output in the raised exception so that we don't
+        # inadvertently hide error output.
+        if len(last_lines) == ERROR_OUTPUT_MAX_LINE_COUNT:
+            last_lines.insert(0, '...')
+
+        # Something has gone wrong. So vent each process' output buffer to prevent it from
+        # hanging. And then kill the process.
+        for other_process in process_last_lines.keys():
+            if other_process.poll() is None:
+                other_process.stdout.read(0)
+                other_process.kill()
+
+        if exit_status == Exit_status.ERROR:
+            raise subprocess.CalledProcessError(
+                exit_code,
+                command_for_process(process),
+                '\n'.join(last_lines),
+            )
+
+        return exit_status
+
+    return Exit_status.STILL_RUNNING
+
+
+def log_remaining_buffer_lines(
+    buffer_readers, process_to_capture, output_log_level, borg_local_path, capture_stderr=False
+):
+    '''
+    Given a dict from buffer object to Buffer_reader, a subprocess.Popen() instance of a process to
+    capture, a requested output log level for stdout, Borg's local path, and whether to capture
+    stderr, drain and log any remaining output lines from the buffers until they're empty.
+    Additionally, if the log level is None for any log record, then yield those log messages for
+    capture.
+    '''
+    for output_buffer, reader in buffer_readers.items():
+        if not reader.process:
+            continue
+
+        for lines in reader.lines:
+            for line in lines:
+                log_record = handle_log_record(
+                    parse_log_line(
+                        line=line.rstrip(),
+                        log_level=output_log_level,
+                        elevate_stderr=(
+                            output_buffer == reader.process.stderr and not capture_stderr
+                        ),
+                        borg_local_path=borg_local_path,
+                        command=reader.process.args,
+                    ),
+                )
+
+                if log_record.levelno is None and reader.process == process_to_capture:
+                    print('***', log_record.getMessage())
+                    yield log_record.getMessage()
+
+
+def log_outputs(
     processes,
     exclude_stdouts,
     output_log_level,
@@ -268,16 +422,8 @@ def log_outputs(  # noqa: PLR0912
     '''
     # Map from output buffer to sequence of last lines.
     process_last_lines = collections.defaultdict(list)
-    process_for_output_buffer = {
-        buffer: process
-        for process in processes
-        if process.stdout or process.stderr
-        for buffer in output_buffers_for_process(process, exclude_stdouts)
-    }
-    output_buffers = list(process_for_output_buffer.keys())
-    process_to_capture = processes[-1]
-    reader_for_output_buffer = {
-        buffer: read_lines(buffer, process)
+    buffer_readers = {
+        buffer: Buffer_reader(read_lines(buffer, process), process)
         for process in processes
         if process.stdout or process.stderr
         for buffer in output_buffers_for_process(process, exclude_stdouts)
@@ -285,122 +431,18 @@ def log_outputs(  # noqa: PLR0912
 
     # Log output for each process until they all exit.
     while any(process.poll() is None for process in processes):
-        if output_buffers:
-            (ready_buffers, _, _) = select.select(output_buffers, [], [])
+        yield from log_buffer_lines(
+            buffer_readers, processes, output_log_level, borg_local_path, capture_stderr
+        )
 
-            for ready_buffer in ready_buffers:
-                ready_process = process_for_output_buffer.get(ready_buffer)
+        if raise_for_process_errors(
+            buffer_readers, process_last_lines, borg_local_path, borg_exit_codes
+        ) == Exit_status.WARNING:
+            break
 
-                # The "ready" process has exited, but it might be a pipe destination with other
-                # processes (pipe sources) waiting to be read from. So as a measure to prevent
-                # hangs, vent all processes when one exits.
-                if ready_process and ready_process.poll() is not None:
-                    for other_process in processes:
-                        if (
-                            other_process.poll() is None
-                            and other_process.stdout
-                            and other_process.stdout not in output_buffers
-                        ):
-                            # Add the process's output to output_buffers to ensure it'll get read.
-                            output_buffers.append(other_process.stdout)
-                            process_for_output_buffer[other_process.stdout] = other_process
-                            reader_for_output_buffer[other_process.stdout] = read_lines(
-                                other_process.stdout, other_process
-                            )
-
-                try:
-                    lines = next(reader_for_output_buffer[ready_buffer])
-                except StopIteration:
-                    continue
-
-                for line in lines:
-                    if not line or not ready_process:
-                        continue
-
-                    command = (
-                        ready_process.args.split(' ')
-                        if isinstance(ready_process.args, str)
-                        else ready_process.args
-                    )
-
-                    # Keep the last few lines of output in case the process errors and we need the
-                    # output for the exception below.
-                    log_record = handle_log_record(
-                        parse_log_line(
-                            line=line,
-                            log_level=output_log_level,
-                            elevate_stderr=(
-                                ready_buffer == ready_process.stderr and not capture_stderr
-                            ),
-                            borg_local_path=borg_local_path,
-                            command=command,
-                        ),
-                        last_lines=process_last_lines[ready_process],
-                    )
-
-                    if log_record.levelno is None and ready_process == process_to_capture:
-                        print('***', log_record.getMessage())
-                        yield log_record.getMessage()
-
-        for process in processes:
-            exit_code = process.poll() if output_buffers else process.wait()
-
-            if exit_code is None:
-                continue
-
-            command = process.args.split(' ') if isinstance(process.args, str) else process.args
-            exit_status = interpret_exit_code(command, exit_code, borg_local_path, borg_exit_codes)
-
-            if exit_status in {Exit_status.ERROR, Exit_status.WARNING}:
-                last_lines = process_last_lines[process]
-
-                # If an error occurs, include its output in the raised exception so that we don't
-                # inadvertently hide error output.
-                if len(last_lines) == ERROR_OUTPUT_MAX_LINE_COUNT:
-                    last_lines.insert(0, '...')
-
-                # Something has gone wrong. So vent each process' output buffer to prevent it from
-                # hanging. And then kill the process.
-                for other_process in processes:
-                    if other_process.poll() is None:
-                        other_process.stdout.read(0)
-                        other_process.kill()
-
-                if exit_status == Exit_status.ERROR:
-                    raise subprocess.CalledProcessError(
-                        exit_code,
-                        command_for_process(process),
-                        '\n'.join(last_lines),
-                    )
-
-                break
-
-    # Now that all processes have exited, drain and consume any last output.
-    for output_buffer in output_buffers:
-        process = process_for_output_buffer.get(output_buffer)
-
-        if not process:
-            continue
-
-        command = process.args.split(' ') if isinstance(process.args, str) else process.args
-        last_lines = process_last_lines[process]
-
-        for lines in reader_for_output_buffer[output_buffer]:
-            for line in lines:
-                log_record = handle_log_record(
-                    parse_log_line(
-                        line=line.rstrip(),
-                        log_level=output_log_level,
-                        elevate_stderr=(output_buffer == process.stderr and not capture_stderr),
-                        borg_local_path=borg_local_path,
-                        command=command,
-                    ),
-                    last_lines=last_lines,
-                )
-
-                if log_record.levelno is None and process == process_to_capture:
-                    print('***', log_record.getMessage())
-                    yield log_record.getMessage()
+    # Now that all processes have exited, drain and consume any last output. By convention, the last
+    # process is the process to capture.
+    yield from log_remaining_buffer_lines(buffer_readers, processes[-1], output_log_level, borg_local_path, capture_stderr)
 
 
 SECRET_COMMAND_FLAG_NAMES = {'--password'}

From 765eba531530a79b4456455c03088a440759d821 Mon Sep 17 00:00:00 2001
From: Dan Helfman <witten@torsion.org>
Date: Thu, 22 Jan 2026 17:29:30 -0800
Subject: [PATCH 5/8] Get existing unit/integration tests passing (#1242).

---
 NEWS                              |  1 +
 borgmatic/execute.py              | 34 +++++++++++++++++++------------
 tests/integration/test_execute.py | 33 ++++++++++++++++++++----------
 3 files changed, 44 insertions(+), 24 deletions(-)

diff --git a/NEWS b/NEWS
index 7c882f8e..d1bbbbb3 100644
--- a/NEWS
+++ b/NEWS
@@ -1,6 +1,7 @@
 2.1.1.dev0
  * #1241: For the "recreate" action, actually pass the "--dry-run" flag through to Borg instead of
    just skipping the Borg call.
+ * #1242: Fix a regression in which the "spot" check hung while collecting archive contents.
 
 2.1.0
  * TL;DR: Many logging, memory, and performance improvements. Mind those breaking changes!
diff --git a/borgmatic/execute.py b/borgmatic/execute.py
index acda5dce..359aa765 100644
--- a/borgmatic/execute.py
+++ b/borgmatic/execute.py
@@ -255,12 +255,12 @@ Buffer_reader = collections.namedtuple(
 
 
 def log_buffer_lines(
-    buffer_readers, processes, output_log_level, borg_local_path, capture_stderr=False
+    buffer_readers, process_last_lines, output_log_level, borg_local_path, capture_stderr=False
 ):
     '''
-    Given a dict from buffer object to Buffer_reader, a sequence of subprocess.Popen() instances for
-    the processes corresponding to those buffers, a requested output log level for stdout, Borg's
-    local path, and whether to capture stderr, read and log any ready output lines from the buffers.
+    Given a dict from buffer object to Buffer_reader, a map from subprocess.Popen() instance to a
+    sequence of last lines for that process, a requested output log level for stdout, Borg's local
+    path, and whether to capture stderr, read and log any ready output lines from the buffers.
     Additionally, if the log level is None for any log record, then yield those log messages for
     capture.
 
@@ -279,7 +279,7 @@ def log_buffer_lines(
         # processes (pipe sources) waiting to be read from. So as a measure to prevent
         # hangs, vent all processes when one exits.
         if reader.process and reader.process.poll() is not None:
-            for other_process in processes:
+            for other_process in process_last_lines.keys():
                 if (
                     other_process.poll() is None
                     and other_process.stdout
@@ -309,10 +309,11 @@ def log_buffer_lines(
                     borg_local_path=borg_local_path,
                     command=reader.process.args,
                 ),
+                last_lines=process_last_lines[reader.process],
             )
 
             # By convention, assume that we're capturing only the last process in the sequence.
-            if log_record.levelno is None and reader.process == processes[-1]:
+            if log_record.levelno is None and reader.process == tuple(process_last_lines.keys())[-1]:
                 print('***', log_record.getMessage())
                 yield log_record.getMessage()
 
@@ -357,7 +358,7 @@ def raise_for_process_errors(buffer_readers, process_last_lines, borg_local_path
                 '\n'.join(last_lines),
             )
 
-        return exit_status
+        return Exit_status.WARNING
 
     return Exit_status.STILL_RUNNING
 
@@ -421,7 +422,9 @@ def log_outputs(
     captured, in which case it won't be logged.
     '''
     # Map from output buffer to sequence of last lines.
-    process_last_lines = collections.defaultdict(list)
+    process_last_lines = {process: [] for process in processes}
+
+    # Map from buffer to Buffer_reader instance.
     buffer_readers = {
         buffer: Buffer_reader(read_lines(buffer, process), process)
         for process in processes
@@ -432,17 +435,22 @@ def log_outputs(
     # Log output for each process until they all exit.
     while any(process.poll() is None for process in processes):
         yield from log_buffer_lines(
-            buffer_readers, processes, output_log_level, borg_local_path, capture_stderr
+            buffer_readers, process_last_lines, output_log_level, borg_local_path, capture_stderr
         )
 
-        if raise_for_process_errors(
-            buffer_readers, process_last_lines, borg_local_path, borg_exit_codes
-        ) == Exit_status.WARNING:
+        if (
+            raise_for_process_errors(
+                buffer_readers, process_last_lines, borg_local_path, borg_exit_codes
+            )
+            == Exit_status.WARNING
+        ):
             break
 
     # Now that all processes have exited, drain and consume any last output. By convention, the last
     # process is the process to capture.
-    yield from log_remaining_buffer_lines(buffer_readers, processes[-1], output_log_level, borg_local_path, capture_stderr)
+    yield from log_remaining_buffer_lines(
+        buffer_readers, processes[-1], output_log_level, borg_local_path, capture_stderr
+    )
 
 
 SECRET_COMMAND_FLAG_NAMES = {'--password'}
diff --git a/tests/integration/test_execute.py b/tests/integration/test_execute.py
index c38493be..1bf4faad 100644
--- a/tests/integration/test_execute.py
+++ b/tests/integration/test_execute.py
@@ -364,17 +364,28 @@ def test_log_outputs_vents_other_processes_when_one_exits():
         other_process,
         (process.stdout,),
     ).and_return((other_process.stdout,))
-    flexmock(process.stdout).should_call('readline').at_least().once()
+    flexmock(module.os).should_call('read').with_args(
+        process.stderr.fileno(), int
+    ).at_least().once()
+    flexmock(module.os).should_call('read').with_args(
+        process.stdout.fileno(), int
+    ).at_least().once()
+    flexmock(module.os).should_call('read').with_args(
+        other_process.stdout.fileno(), int
+    ).at_least().once()
 
-    assert tuple(
-        module.log_outputs(
-            (process, other_process),
-            exclude_stdouts=(process.stdout,),
-            output_log_level=logging.INFO,
-            borg_local_path='borg',
-            borg_exit_codes=None,
+    assert (
+        tuple(
+            module.log_outputs(
+                (process, other_process),
+                exclude_stdouts=(process.stdout,),
+                output_log_level=logging.INFO,
+                borg_local_path='borg',
+                borg_exit_codes=None,
+            )
         )
-    ) == ('',)
+        == ()
+    )
 
 
 def test_log_outputs_does_not_error_when_one_process_exits():
@@ -443,8 +454,8 @@ def test_log_outputs_truncates_long_error_output():
             )
         )
 
-        assert error.value.returncode == 2
-        assert error.value.output.startswith('...')
+    assert error.value.returncode == 2
+    assert error.value.output.startswith('...')
 
 
 def test_log_outputs_with_no_output_logs_nothing():

From 97f7c65f6caf09ecdcdab55939b6d7164338ec69 Mon Sep 17 00:00:00 2001
From: Dan Helfman <witten@torsion.org>
Date: Thu, 22 Jan 2026 23:01:34 -0800
Subject: [PATCH 6/8] More refactoring and test fixes (#1242).

---
 borgmatic/execute.py | 84 +++++++++++++++++++++++++-------------------
 1 file changed, 47 insertions(+), 37 deletions(-)

diff --git a/borgmatic/execute.py b/borgmatic/execute.py
index 359aa765..baeb92cc 100644
--- a/borgmatic/execute.py
+++ b/borgmatic/execute.py
@@ -254,17 +254,22 @@ Buffer_reader = collections.namedtuple(
 )
 
 
+Process_metadata = collections.namedtuple(
+    'Process_metadata',
+    ('last_lines', 'capture'),
+)
+
+
 def log_buffer_lines(
-    buffer_readers, process_last_lines, output_log_level, borg_local_path, capture_stderr=False
+    buffer_readers, process_metadatas, output_log_level, borg_local_path, capture_stderr=False
 ):
     '''
-    Given a dict from buffer object to Buffer_reader, a map from subprocess.Popen() instance to a
-    sequence of last lines for that process, a requested output log level for stdout, Borg's local
-    path, and whether to capture stderr, read and log any ready output lines from the buffers.
-    Additionally, if the log level is None for any log record, then yield those log messages for
-    capture.
+    Given a dict from buffer object to Buffer_reader, a dict from subprocess.Popen() instance to
+    Process_metadata instance, a requested output log level for stdout, Borg's local path, and
+    whether to capture stderr, read and log any ready output lines from the buffers.  Additionally,
+    if the log level is None for any log record, then yield those log messages for capture.
 
-    This function just does one "turn of the crank" on logging buffer output. It is intended to be
+    This function just does one "turn of the crank" of logging buffer output. It is intended to be
     called repeatedly to continue to process buffers.
     '''
     if not buffer_readers:
@@ -279,7 +284,7 @@ def log_buffer_lines(
         # processes (pipe sources) waiting to be read from. So as a measure to prevent
         # hangs, vent all processes when one exits.
         if reader.process and reader.process.poll() is not None:
-            for other_process in process_last_lines.keys():
+            for other_process in process_metadatas.keys():
                 if (
                     other_process.poll() is None
                     and other_process.stdout
@@ -309,24 +314,23 @@ def log_buffer_lines(
                     borg_local_path=borg_local_path,
                     command=reader.process.args,
                 ),
-                last_lines=process_last_lines[reader.process],
+                last_lines=process_metadatas[reader.process].last_lines,
             )
 
-            # By convention, assume that we're capturing only the last process in the sequence.
-            if log_record.levelno is None and reader.process == tuple(process_last_lines.keys())[-1]:
+            if log_record.levelno is None and process_metadatas[reader.process].capture:
                 print('***', log_record.getMessage())
                 yield log_record.getMessage()
 
 
-def raise_for_process_errors(buffer_readers, process_last_lines, borg_local_path, borg_exit_codes):
+def raise_for_process_errors(buffer_readers, process_metadatas, borg_local_path, borg_exit_codes):
     '''
-    Given a dict from buffer object to Buffer_reader, a map from subprocess.Popen() instance to a
-    sequence of last lines for that process, 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 Exit_status.STILL_RUNNING.
+    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 Exit_status.STILL_RUNNING.
     '''
-    for process in process_last_lines.keys():
+    for process in process_metadatas.keys():
         exit_code = process.poll() if buffer_readers else process.wait()
 
         if exit_code is None:
@@ -337,16 +341,16 @@ def raise_for_process_errors(buffer_readers, process_last_lines, borg_local_path
         if exit_status not in {Exit_status.ERROR, Exit_status.WARNING}:
             continue
 
-        last_lines = process_last_lines[process]
+        last_lines = process_metadatas[process].last_lines
 
         # If an error occurs, include its output in the raised exception so that we don't
         # inadvertently hide error output.
-        if len(last_lines) == ERROR_OUTPUT_MAX_LINE_COUNT:
+        if len(last_lines) >= ERROR_OUTPUT_MAX_LINE_COUNT:
             last_lines.insert(0, '...')
 
         # Something has gone wrong. So vent each process' output buffer to prevent it from
         # hanging. And then kill the process.
-        for other_process in process_last_lines.keys():
+        for other_process in process_metadatas.keys():
             if other_process.poll() is None:
                 other_process.stdout.read(0)
                 other_process.kill()
@@ -364,14 +368,14 @@ def raise_for_process_errors(buffer_readers, process_last_lines, borg_local_path
 
 
 def log_remaining_buffer_lines(
-    buffer_readers, process_to_capture, output_log_level, borg_local_path, capture_stderr=False
+    buffer_readers, process_metadatas, output_log_level, borg_local_path, capture_stderr=False
 ):
     '''
-    Given a dict from buffer object to Buffer_reader, a subprocess.Popen() instance of a process to
-    capture, a requested output log level for stdout, Borg's local path, and whether to capture
-    stderr, drain and log any remaining output lines from the buffers until they're empty.
-    Additionally, if the log level is None for any log record, then yield those log messages for
-    capture.
+    Given a dict from buffer object to Buffer_reader, a dict from subprocess.Popen() instance to
+    Process_metadata instance, a requested output log level for stdout, Borg's local path, and
+    whether to capture stderr, drain and log any remaining output lines from the buffers until
+    they're empty. Additionally, if the log level is None for any log record, then yield those log
+    messages for capture.
     '''
     for output_buffer, reader in buffer_readers.items():
         if not reader.process:
@@ -391,7 +395,7 @@ def log_remaining_buffer_lines(
                     ),
                 )
 
-                if log_record.levelno is None and reader.process == process_to_capture:
+                if log_record.levelno is None and process_metadatas[reader.process].capture:
                     print('***', log_record.getMessage())
                     yield log_record.getMessage()
 
@@ -421,8 +425,12 @@ def log_outputs(
     buffers. Also note that stdout for a process can be None if output is intentionally not
     captured, in which case it won't be logged.
     '''
-    # Map from output buffer to sequence of last lines.
-    process_last_lines = {process: [] for process in processes}
+    # Map from output buffer to Process_metadata instance. By convention, the last process is the
+    # process to capture.
+    process_metadatas = {
+        process: Process_metadata(last_lines=[], capture=bool(process == processes[-1]))
+        for process in processes
+    }
 
     # Map from buffer to Buffer_reader instance.
     buffer_readers = {
@@ -432,24 +440,26 @@ def log_outputs(
         for buffer in output_buffers_for_process(process, exclude_stdouts)
     }
 
-    # Log output for each process until they all exit.
-    while any(process.poll() is None for process in processes):
+    # Log output lines for each process until they all exit.
+    while True:
         yield from log_buffer_lines(
-            buffer_readers, process_last_lines, output_log_level, borg_local_path, capture_stderr
+            buffer_readers, process_metadatas, output_log_level, borg_local_path, capture_stderr
         )
 
         if (
             raise_for_process_errors(
-                buffer_readers, process_last_lines, borg_local_path, borg_exit_codes
+                buffer_readers, process_metadatas, borg_local_path, borg_exit_codes
             )
             == Exit_status.WARNING
         ):
             break
 
-    # Now that all processes have exited, drain and consume any last output. By convention, the last
-    # process is the process to capture.
+        if all(process.poll() is not None for process in processes):
+            break
+
+    # Now that all processes have exited, drain and consume any last output.
     yield from log_remaining_buffer_lines(
-        buffer_readers, processes[-1], output_log_level, borg_local_path, capture_stderr
+        buffer_readers, process_metadatas, output_log_level, borg_local_path, capture_stderr
     )
 
 

From 248fa1db64b5cb0622a1dac2c5f932abadc0f4f3 Mon Sep 17 00:00:00 2001
From: Dan Helfman <witten@torsion.org>
Date: Fri, 23 Jan 2026 22:50:37 -0800
Subject: [PATCH 7/8] Add automated tests for new code (#1242).

---
 borgmatic/execute.py              |  45 +-
 tests/integration/test_execute.py |  36 ++
 tests/unit/test_execute.py        | 871 ++++++++++++++++++++++++++++++
 3 files changed, 930 insertions(+), 22 deletions(-)

diff --git a/borgmatic/execute.py b/borgmatic/execute.py
index baeb92cc..ae6309ee 100644
--- a/borgmatic/execute.py
+++ b/borgmatic/execute.py
@@ -42,7 +42,7 @@ def interpret_exit_code(command, exit_code, borg_local_path=None, borg_exit_code
     if exit_code == 0:
         return Exit_status.SUCCESS
 
-    parsed_command = command.args.split(' ', 1) if isinstance(command, str) else command
+    parsed_command = command.split(' ', 1) if isinstance(command, str) else command
 
     if borg_local_path and parsed_command[0] == borg_local_path:
         # First try looking for the exit code in the borg_exit_codes configuration.
@@ -284,7 +284,7 @@ def log_buffer_lines(
         # processes (pipe sources) waiting to be read from. So as a measure to prevent
         # hangs, vent all processes when one exits.
         if reader.process and reader.process.poll() is not None:
-            for other_process in process_metadatas.keys():
+            for other_process in process_metadatas:
                 if (
                     other_process.poll() is None
                     and other_process.stdout
@@ -318,7 +318,6 @@ def log_buffer_lines(
             )
 
             if log_record.levelno is None and process_metadatas[reader.process].capture:
-                print('***', log_record.getMessage())
                 yield log_record.getMessage()
 
 
@@ -328,9 +327,11 @@ def raise_for_process_errors(buffer_readers, process_metadatas, borg_local_path,
     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 Exit_status.STILL_RUNNING.
+    Exit_status.WARNING. Otherwise, return None.
     '''
-    for process in process_metadatas.keys():
+    result_status = None
+
+    for process in process_metadatas:
         exit_code = process.poll() if buffer_readers else process.wait()
 
         if exit_code is None:
@@ -341,6 +342,17 @@ def raise_for_process_errors(buffer_readers, process_metadatas, borg_local_path,
         if exit_status not in {Exit_status.ERROR, Exit_status.WARNING}:
             continue
 
+        # Something has gone wrong. So vent each process' output buffer to prevent it from
+        # hanging. And then kill the process.
+        for other_process in process_metadatas:
+            if other_process.poll() is None:
+                other_process.stdout.read(0)
+                other_process.kill()
+
+        if exit_status == Exit_status.WARNING:
+            result_status = Exit_status.WARNING
+            continue
+
         last_lines = process_metadatas[process].last_lines
 
         # If an error occurs, include its output in the raised exception so that we don't
@@ -348,23 +360,13 @@ def raise_for_process_errors(buffer_readers, process_metadatas, borg_local_path,
         if len(last_lines) >= ERROR_OUTPUT_MAX_LINE_COUNT:
             last_lines.insert(0, '...')
 
-        # Something has gone wrong. So vent each process' output buffer to prevent it from
-        # hanging. And then kill the process.
-        for other_process in process_metadatas.keys():
-            if other_process.poll() is None:
-                other_process.stdout.read(0)
-                other_process.kill()
+        raise subprocess.CalledProcessError(
+            exit_code,
+            command_for_process(process),
+            '\n'.join(last_lines),
+        )
 
-        if exit_status == Exit_status.ERROR:
-            raise subprocess.CalledProcessError(
-                exit_code,
-                command_for_process(process),
-                '\n'.join(last_lines),
-            )
-
-        return Exit_status.WARNING
-
-    return Exit_status.STILL_RUNNING
+    return result_status
 
 
 def log_remaining_buffer_lines(
@@ -396,7 +398,6 @@ def log_remaining_buffer_lines(
                 )
 
                 if log_record.levelno is None and process_metadatas[reader.process].capture:
-                    print('***', log_record.getMessage())
                     yield log_record.getMessage()
 
 
diff --git a/tests/integration/test_execute.py b/tests/integration/test_execute.py
index 1bf4faad..dba1ba61 100644
--- a/tests/integration/test_execute.py
+++ b/tests/integration/test_execute.py
@@ -8,6 +8,42 @@ from flexmock import flexmock
 from borgmatic import execute as module
 
 
+def test_read_lines_yields_single_line():
+    process = subprocess.Popen(['echo', 'hi'], stdout=subprocess.PIPE)
+
+    assert tuple(module.read_lines(process.stdout, process)) == (('hi',),)
+
+
+def test_read_lines_yields_single_line_longer_than_chunk_size():
+    process = subprocess.Popen(
+        ['echo', 'this line is longer than the chunk size'], stdout=subprocess.PIPE
+    )
+
+    assert tuple(flexmock(module, READ_CHUNK_SIZE=16).read_lines(process.stdout, process)) == (
+        (),
+        (),
+        ('this line is longer than the chunk size',),
+    )
+
+
+def test_read_lines_yields_multiple_lines():
+    process = subprocess.Popen(['echo', 'hi\nthere'], stdout=subprocess.PIPE)
+
+    assert tuple(module.read_lines(process.stdout, process)) == (('hi', 'there'),)
+
+
+def test_read_lines_yields_multiple_lines_plus_partial_line():
+    process = subprocess.Popen(['echo', '-n', 'hi\nthere\npartial'], stdout=subprocess.PIPE)
+
+    assert tuple(module.read_lines(process.stdout, process)) == (('hi', 'there'), ('partial',))
+
+
+def test_read_lines_yields_nothing():
+    process = subprocess.Popen(['echo', '-n'], stdout=subprocess.PIPE)
+
+    assert tuple(module.read_lines(process.stdout, process)) == ()
+
+
 def test_log_outputs_logs_each_line_separately():
     hi_record = flexmock(
         msg='hi',
diff --git a/tests/unit/test_execute.py b/tests/unit/test_execute.py
index 2b1b2b31..8eb727c9 100644
--- a/tests/unit/test_execute.py
+++ b/tests/unit/test_execute.py
@@ -19,12 +19,18 @@ from borgmatic import execute as module
         (['borg1'], 1, 'borg1', None, module.Exit_status.WARNING),
         (['grep'], 100, None, None, module.Exit_status.ERROR),
         (['grep'], 100, 'borg', None, module.Exit_status.ERROR),
+        ('grep', 2, None, None, module.Exit_status.ERROR),
+        ('borg', 2, 'borg', None, module.Exit_status.ERROR),
         (['borg'], 100, 'borg', None, module.Exit_status.WARNING),
         (['borg1'], 100, 'borg1', None, module.Exit_status.WARNING),
+        ('borg', 100, 'borg', None, module.Exit_status.WARNING),
+        ('borg1', 100, 'borg1', None, module.Exit_status.WARNING),
         (['grep'], 0, None, None, module.Exit_status.SUCCESS),
         (['grep'], 0, 'borg', None, module.Exit_status.SUCCESS),
         (['borg'], 0, 'borg', None, module.Exit_status.SUCCESS),
         (['borg1'], 0, 'borg1', None, module.Exit_status.SUCCESS),
+        ('grep', 0, None, None, module.Exit_status.SUCCESS),
+        ('grep', 0, 'borg', None, module.Exit_status.SUCCESS),
         # -9 exit code occurs when child process get SIGKILLed.
         (['grep'], -9, None, None, module.Exit_status.ERROR),
         (['grep'], -9, 'borg', None, module.Exit_status.ERROR),
@@ -174,6 +180,23 @@ def test_parse_log_line_with_borg_command_parses_borg_log_line():
     )
 
 
+def test_parse_log_line_with_borg_command_parses_borg_log_line_with_string_command():
+    record = flexmock()
+    flexmock(module).should_receive('borg_json_log_line_to_record').and_return(record).once()
+    flexmock(module).should_receive('log_line_to_record').never()
+
+    assert (
+        module.parse_log_line(
+            'All done',
+            module.logging.INFO,
+            elevate_stderr=False,
+            borg_local_path='borg',
+            command='borg do-stuff',
+        )
+        == record
+    )
+
+
 def test_parse_log_line_without_borg_command_parses_plain_log_line():
     record = flexmock()
     flexmock(module).should_receive('borg_json_log_line_to_record').never()
@@ -191,6 +214,23 @@ def test_parse_log_line_without_borg_command_parses_plain_log_line():
     )
 
 
+def test_parse_log_line_without_borg_command_parses_plain_log_line_with_string_command():
+    record = flexmock()
+    flexmock(module).should_receive('borg_json_log_line_to_record').never()
+    flexmock(module).should_receive('log_line_to_record').and_return(record).once()
+
+    assert (
+        module.parse_log_line(
+            'All done',
+            module.logging.INFO,
+            elevate_stderr=False,
+            borg_local_path='borg',
+            command='totally-not-borg do-stuff',
+        )
+        == record
+    )
+
+
 def test_parse_log_line_with_elevate_stderr_makes_error_record():
     record = flexmock()
     flexmock(module).should_receive('borg_json_log_line_to_record').never()
@@ -262,6 +302,837 @@ def test_handle_log_record_over_max_line_count_trims_and_appends():
     assert last_lines == [*original_last_lines[1:], 'line']
 
 
+def test_log_buffer_lines_without_buffer_readers_bails():
+    flexmock(module.select).should_receive('select').never()
+
+    assert (
+        tuple(
+            module.log_buffer_lines(
+                buffer_readers={},
+                process_metadatas={},
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+            )
+        )
+        == ()
+    )
+
+
+def test_log_buffer_lines_without_ready_buffers_bails():
+    buffer_readers = {flexmock(): flexmock()}
+    flexmock(module.select).should_receive('select').with_args(
+        buffer_readers.keys(), [], []
+    ).and_return([], [], []).once()
+
+    assert (
+        tuple(
+            module.log_buffer_lines(
+                buffer_readers=buffer_readers,
+                process_metadatas={},
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+            )
+        )
+        == ()
+    )
+
+
+def test_log_buffer_lines_with_ready_buffer_and_running_process_handles_each_log_line():
+    process = flexmock(poll=lambda: None, stderr=flexmock(), args=flexmock())
+    buffer_readers = {
+        flexmock(): module.Buffer_reader(lines=iter((('hi', 'there'),)), process=process)
+    }
+    process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)}
+    flexmock(module.select).should_receive('select').with_args(
+        buffer_readers.keys(), [], []
+    ).and_return(list(buffer_readers.keys()), [], [])
+    flexmock(module).should_receive('parse_log_line').and_return(flexmock())
+    flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).twice()
+
+    assert (
+        tuple(
+            module.log_buffer_lines(
+                buffer_readers=buffer_readers,
+                process_metadatas=process_metadatas,
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+            )
+        )
+        == ()
+    )
+
+
+def test_log_buffer_lines_with_ready_buffer_and_capture_process_yields_each_line():
+    process = flexmock(poll=lambda: None, stderr=flexmock(), args=flexmock())
+    buffer_readers = {
+        flexmock(): module.Buffer_reader(lines=iter((('hi', 'there'),)), process=process)
+    }
+    process_metadatas = {process: module.Process_metadata(last_lines=[], capture=True)}
+    flexmock(module.select).should_receive('select').with_args(
+        buffer_readers.keys(), [], []
+    ).and_return(list(buffer_readers.keys()), [], [])
+    flexmock(module).should_receive('parse_log_line').and_return(flexmock())
+    flexmock(module).should_receive('handle_log_record').and_return(
+        flexmock(levelno=None, getMessage=lambda: 'message')
+    ).twice()
+
+    assert tuple(
+        module.log_buffer_lines(
+            buffer_readers=buffer_readers,
+            process_metadatas=process_metadatas,
+            output_log_level=flexmock(),
+            borg_local_path=flexmock(),
+        )
+    ) == ('message', 'message')
+
+
+def test_log_buffer_lines_with_ready_buffer_and_log_level_and_capture_process_does_not_yield_each_line():
+    process = flexmock(poll=lambda: None, stderr=flexmock(), args=flexmock())
+    buffer_readers = {
+        flexmock(): module.Buffer_reader(lines=iter((('hi', 'there'),)), process=process)
+    }
+    process_metadatas = {process: module.Process_metadata(last_lines=[], capture=True)}
+    flexmock(module.select).should_receive('select').with_args(
+        buffer_readers.keys(), [], []
+    ).and_return(list(buffer_readers.keys()), [], [])
+    flexmock(module).should_receive('parse_log_line').and_return(flexmock())
+    flexmock(module).should_receive('handle_log_record').and_return(
+        flexmock(levelno=10, getMessage=lambda: 'message')
+    ).twice()
+
+    assert (
+        tuple(
+            module.log_buffer_lines(
+                buffer_readers=buffer_readers,
+                process_metadatas=process_metadatas,
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+            )
+        )
+        == ()
+    )
+
+
+def test_log_buffer_lines_with_ready_buffer_and_finished_process_vents_other_processes():
+    process_stdout = flexmock()
+    process = flexmock(poll=lambda: 0, stdout=process_stdout, stderr=flexmock(), args=flexmock())
+    other_process = flexmock(
+        poll=lambda: None, stdout=flexmock(), stderr=flexmock(), args=flexmock()
+    )
+    buffer_readers = {process_stdout: module.Buffer_reader(lines=iter((('hi',),)), process=process)}
+    process_metadatas = {
+        process: module.Process_metadata(last_lines=[], capture=False),
+        other_process: module.Process_metadata(last_lines=[], capture=False),
+    }
+    flexmock(module.select).should_receive('select').with_args(
+        buffer_readers.keys(), [], []
+    ).and_return(list(buffer_readers.keys()), [], [])
+    flexmock(module).should_receive('read_lines').and_return(iter((('there',),))).once()
+    flexmock(module).should_receive('parse_log_line').and_return(flexmock())
+    flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).once()
+
+    assert (
+        tuple(
+            module.log_buffer_lines(
+                buffer_readers=buffer_readers,
+                process_metadatas=process_metadatas,
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+            )
+        )
+        == ()
+    )
+
+    # log_buffer_lines() vents other processes by adding them to buffer_readers, with the idea that
+    # subsequent calls will then read from them.
+    assert len(buffer_readers) == 2
+
+    # Assert that the process' buffer has been consumed, indicating that it hasn't been accidentally
+    # replaced.
+    assert tuple(buffer_readers[process_stdout].lines) == ()
+
+
+def test_log_buffer_lines_with_ready_buffer_and_finished_process_does_not_vent_other_finished_processes():
+    process_stdout = flexmock()
+    process = flexmock(poll=lambda: 0, stdout=process_stdout, stderr=flexmock(), args=flexmock())
+    other_process = flexmock(poll=lambda: 0, stdout=flexmock(), stderr=flexmock(), args=flexmock())
+    buffer_readers = {process_stdout: module.Buffer_reader(lines=iter((('hi',),)), process=process)}
+    process_metadatas = {
+        process: module.Process_metadata(last_lines=[], capture=False),
+        other_process: module.Process_metadata(last_lines=[], capture=False),
+    }
+    flexmock(module.select).should_receive('select').with_args(
+        buffer_readers.keys(), [], []
+    ).and_return(list(buffer_readers.keys()), [], [])
+    flexmock(module).should_receive('read_lines').never()
+    flexmock(module).should_receive('parse_log_line').and_return(flexmock())
+    flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).once()
+
+    assert (
+        tuple(
+            module.log_buffer_lines(
+                buffer_readers=buffer_readers,
+                process_metadatas=process_metadatas,
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+            )
+        )
+        == ()
+    )
+
+    assert len(buffer_readers) == 1
+    assert tuple(buffer_readers[process_stdout].lines) == ()
+
+
+def test_log_buffer_lines_with_ready_eof_buffer_and_running_process_skips_it():
+    process = flexmock(poll=lambda: None, stderr=flexmock(), args=flexmock())
+    buffer_readers = {flexmock(): module.Buffer_reader(lines=iter(()), process=process)}
+    process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)}
+    flexmock(module.select).should_receive('select').with_args(
+        buffer_readers.keys(), [], []
+    ).and_return(list(buffer_readers.keys()), [], [])
+    flexmock(module).should_receive('parse_log_line').never()
+    flexmock(module).should_receive('handle_log_record').never()
+
+    assert (
+        tuple(
+            module.log_buffer_lines(
+                buffer_readers=buffer_readers,
+                process_metadatas=process_metadatas,
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+            )
+        )
+        == ()
+    )
+
+
+def test_log_buffer_lines_with_ready_buffer_with_empty_line_skips_it():
+    process = flexmock(poll=lambda: None, stderr=flexmock(), args=flexmock())
+    buffer_readers = {flexmock(): module.Buffer_reader(lines=iter((('',),)), process=process)}
+    process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)}
+    flexmock(module.select).should_receive('select').with_args(
+        buffer_readers.keys(), [], []
+    ).and_return(list(buffer_readers.keys()), [], [])
+    flexmock(module).should_receive('parse_log_line').never()
+    flexmock(module).should_receive('handle_log_record').never()
+
+    assert (
+        tuple(
+            module.log_buffer_lines(
+                buffer_readers=buffer_readers,
+                process_metadatas=process_metadatas,
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+            )
+        )
+        == ()
+    )
+
+
+def test_log_buffer_lines_with_multiple_ready_buffers_and_running_processes_handles_log_lines_from_each():
+    process = flexmock(poll=lambda: None, stderr=flexmock(), args=flexmock())
+    other_process = flexmock(poll=lambda: None, stderr=flexmock(), args=flexmock())
+    buffer_readers = {
+        flexmock(): module.Buffer_reader(lines=iter((('hi', 'there'),)), process=process),
+        flexmock(): module.Buffer_reader(lines=iter((('foo', 'bar'),)), process=other_process),
+    }
+    process_metadatas = {
+        process: module.Process_metadata(last_lines=[], capture=False),
+        other_process: module.Process_metadata(last_lines=[], capture=False),
+    }
+    flexmock(module.select).should_receive('select').with_args(
+        buffer_readers.keys(), [], []
+    ).and_return(list(buffer_readers.keys()), [], [])
+    flexmock(module).should_receive('parse_log_line').and_return(flexmock())
+    flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).times(4)
+
+    assert (
+        tuple(
+            module.log_buffer_lines(
+                buffer_readers=buffer_readers,
+                process_metadatas=process_metadatas,
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+            )
+        )
+        == ()
+    )
+
+
+def test_log_buffer_lines_with_multiple_ready_buffers_from_same_running_process_handles_all_log_lines():
+    process = flexmock(poll=lambda: None, stderr=flexmock(), args=flexmock())
+    buffer_readers = {
+        flexmock(): module.Buffer_reader(lines=iter((('hi', 'there'),)), process=process),
+        flexmock(): module.Buffer_reader(lines=iter((('foo', 'bar'),)), process=process),
+    }
+    process_metadatas = {
+        process: module.Process_metadata(last_lines=[], capture=False),
+    }
+    flexmock(module.select).should_receive('select').with_args(
+        buffer_readers.keys(), [], []
+    ).and_return(list(buffer_readers.keys()), [], [])
+    flexmock(module).should_receive('parse_log_line').and_return(flexmock())
+    flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).times(4)
+
+    assert (
+        tuple(
+            module.log_buffer_lines(
+                buffer_readers=buffer_readers,
+                process_metadatas=process_metadatas,
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+            )
+        )
+        == ()
+    )
+
+
+def test_log_buffer_lines_with_ready_stderr_buffer_and_running_process_elevates_stderr():
+    process_stderr = flexmock()
+    process = flexmock(poll=lambda: None, stderr=process_stderr, args=flexmock())
+    buffer_readers = {
+        process_stderr: module.Buffer_reader(lines=iter((('hi', 'there'),)), process=process)
+    }
+    process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)}
+    flexmock(module.select).should_receive('select').with_args(
+        buffer_readers.keys(), [], []
+    ).and_return(list(buffer_readers.keys()), [], [])
+    flexmock(module).should_receive('parse_log_line').with_args(
+        line=str, log_level=object, elevate_stderr=True, borg_local_path=object, command=object
+    ).and_return(flexmock()).twice()
+    flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).twice()
+
+    assert (
+        tuple(
+            module.log_buffer_lines(
+                buffer_readers=buffer_readers,
+                process_metadatas=process_metadatas,
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+            )
+        )
+        == ()
+    )
+
+
+def test_log_buffer_lines_with_ready_stdout_buffer_and_running_process_does_not_elevate_stderr():
+    process_stdout = flexmock()
+    process = flexmock(poll=lambda: None, stdout=process_stdout, stderr=flexmock(), args=flexmock())
+    buffer_readers = {
+        process_stdout: module.Buffer_reader(lines=iter((('hi', 'there'),)), process=process)
+    }
+    process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)}
+    flexmock(module.select).should_receive('select').with_args(
+        buffer_readers.keys(), [], []
+    ).and_return(list(buffer_readers.keys()), [], [])
+    flexmock(module).should_receive('parse_log_line').with_args(
+        line=str, log_level=object, elevate_stderr=False, borg_local_path=object, command=object
+    ).and_return(flexmock()).twice()
+    flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).twice()
+
+    assert (
+        tuple(
+            module.log_buffer_lines(
+                buffer_readers=buffer_readers,
+                process_metadatas=process_metadatas,
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+            )
+        )
+        == ()
+    )
+
+
+def test_log_buffer_lines_with_ready_stderr_buffer_and_capture_stderr_does_not_elevate_stderr():
+    process_stderr = flexmock()
+    process = flexmock(poll=lambda: None, stderr=process_stderr, args=flexmock())
+    buffer_readers = {
+        process_stderr: module.Buffer_reader(lines=iter((('hi', 'there'),)), process=process)
+    }
+    process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)}
+    flexmock(module.select).should_receive('select').with_args(
+        buffer_readers.keys(), [], []
+    ).and_return(list(buffer_readers.keys()), [], [])
+    flexmock(module).should_receive('parse_log_line').with_args(
+        line=str, log_level=object, elevate_stderr=False, borg_local_path=object, command=object
+    ).and_return(flexmock()).twice()
+    flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).twice()
+
+    assert (
+        tuple(
+            module.log_buffer_lines(
+                buffer_readers=buffer_readers,
+                process_metadatas=process_metadatas,
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+                capture_stderr=True,
+            )
+        )
+        == ()
+    )
+
+
+def test_raise_for_process_errors_with_no_processes_bails():
+    process = flexmock()
+    process.should_receive('poll').never()
+    buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)}
+
+    assert (
+        module.raise_for_process_errors(
+            buffer_readers,
+            process_metadatas={},
+            borg_local_path=flexmock(),
+            borg_exit_codes=flexmock(),
+        )
+        is None
+    )
+
+
+def test_raise_for_process_errors_with_running_process_bails():
+    process = flexmock(poll=lambda: None)
+    buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)}
+    process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)}
+
+    assert (
+        module.raise_for_process_errors(
+            buffer_readers,
+            process_metadatas,
+            borg_local_path=flexmock(),
+            borg_exit_codes=flexmock(),
+        )
+        is None
+    )
+
+
+def test_raise_for_process_errors_with_running_process_and_no_buffer_readers_waits_and_bails():
+    process = flexmock()
+    process.should_receive('poll').never()
+    process.should_receive('wait').and_return(None).once()
+    process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)}
+
+    assert (
+        module.raise_for_process_errors(
+            buffer_readers={},
+            process_metadatas=process_metadatas,
+            borg_local_path=flexmock(),
+            borg_exit_codes=flexmock(),
+        )
+        is None
+    )
+
+
+def test_raise_for_process_errors_with_successful_process_bails():
+    process = flexmock(poll=lambda: 0, args=flexmock())
+    buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)}
+    process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)}
+    flexmock(module).should_receive('interpret_exit_code').and_return(module.Exit_status.SUCCESS)
+
+    assert (
+        module.raise_for_process_errors(
+            buffer_readers,
+            process_metadatas,
+            borg_local_path=flexmock(),
+            borg_exit_codes=flexmock(),
+        )
+        is None
+    )
+
+
+def test_raise_for_process_errors_with_warning_process_returns_warning_status():
+    process = flexmock(poll=lambda: 1, args=flexmock())
+    buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)}
+    process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)}
+    flexmock(module).should_receive('interpret_exit_code').and_return(module.Exit_status.WARNING)
+
+    assert (
+        module.raise_for_process_errors(
+            buffer_readers,
+            process_metadatas,
+            borg_local_path=flexmock(),
+            borg_exit_codes=flexmock(),
+        )
+        == module.Exit_status.WARNING
+    )
+
+
+def test_raise_for_process_errors_with_error_process_raises():
+    process = flexmock(poll=lambda: 3, args=flexmock())
+    buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)}
+    process_metadatas = {
+        process: module.Process_metadata(last_lines=['hi', 'there'], capture=False)
+    }
+    flexmock(module).should_receive('interpret_exit_code').and_return(module.Exit_status.ERROR)
+    command = flexmock()
+    flexmock(module).should_receive('command_for_process').and_return(command)
+
+    with pytest.raises(module.subprocess.CalledProcessError) as error:
+        module.raise_for_process_errors(
+            buffer_readers,
+            process_metadatas,
+            borg_local_path=flexmock(),
+            borg_exit_codes=flexmock(),
+        )
+
+    assert error.value.returncode == 3
+    assert error.value.cmd == command
+    assert error.value.output == 'hi\nthere'
+
+
+def test_raise_for_process_errors_with_success_process_and_warning_process_returns_warning_status():
+    process = flexmock(poll=lambda: 0, args=flexmock())
+    other_process = flexmock(poll=lambda: 1, args=flexmock())
+    buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)}
+    process_metadatas = {
+        process: module.Process_metadata(last_lines=['hi', 'there'], capture=False),
+        other_process: module.Process_metadata(last_lines=['and', 'stuff'], capture=False),
+    }
+    flexmock(module).should_receive('interpret_exit_code').with_args(
+        object, 0, object, object
+    ).and_return(module.Exit_status.SUCCESS)
+    flexmock(module).should_receive('interpret_exit_code').with_args(
+        object, 1, object, object
+    ).and_return(module.Exit_status.WARNING)
+    flexmock(module).should_receive('command_for_process').never()
+
+    assert (
+        module.raise_for_process_errors(
+            buffer_readers,
+            process_metadatas,
+            borg_local_path=flexmock(),
+            borg_exit_codes=flexmock(),
+        )
+        == module.Exit_status.WARNING
+    )
+
+
+def test_raise_for_process_errors_with_warning_process_and_error_process_raises():
+    process = flexmock(poll=lambda: 1, args=flexmock())
+    other_process = flexmock(poll=lambda: 3, args=flexmock())
+    buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)}
+    process_metadatas = {
+        process: module.Process_metadata(last_lines=['hi', 'there'], capture=False),
+        other_process: module.Process_metadata(last_lines=['and', 'stuff'], capture=False),
+    }
+    flexmock(module).should_receive('interpret_exit_code').with_args(
+        object, 1, object, object
+    ).and_return(module.Exit_status.WARNING)
+    flexmock(module).should_receive('interpret_exit_code').with_args(
+        object, 3, object, object
+    ).and_return(module.Exit_status.ERROR)
+    command = flexmock()
+    flexmock(module).should_receive('command_for_process').and_return(command)
+
+    with pytest.raises(module.subprocess.CalledProcessError) as error:
+        module.raise_for_process_errors(
+            buffer_readers,
+            process_metadatas,
+            borg_local_path=flexmock(),
+            borg_exit_codes=flexmock(),
+        )
+
+    assert error.value.returncode == 3
+    assert error.value.cmd == command
+    assert error.value.output == 'and\nstuff'
+
+
+def test_raise_for_process_errors_with_warning_process_and_running_process_kills_and_returns_warning_status():
+    process = flexmock(poll=lambda: 1, args=flexmock())
+    other_process = flexmock(
+        poll=lambda: None, stdout=flexmock(read=lambda size: None), args=flexmock()
+    )
+    other_process.should_receive('kill').once()
+    buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)}
+    process_metadatas = {
+        process: module.Process_metadata(last_lines=['hi', 'there'], capture=False),
+        other_process: module.Process_metadata(last_lines=['and', 'stuff'], capture=False),
+    }
+    flexmock(module).should_receive('interpret_exit_code').with_args(
+        object, 1, object, object
+    ).and_return(module.Exit_status.WARNING)
+    command = flexmock()
+    flexmock(module).should_receive('command_for_process').and_return(command)
+
+    assert (
+        module.raise_for_process_errors(
+            buffer_readers,
+            process_metadatas,
+            borg_local_path=flexmock(),
+            borg_exit_codes=flexmock(),
+        )
+        == module.Exit_status.WARNING
+    )
+
+
+def test_raise_for_process_errors_with_error_process_and_running_process_kills_and_raises():
+    process = flexmock(poll=lambda: 3, args=flexmock())
+    other_process = flexmock(
+        poll=lambda: None, stdout=flexmock(read=lambda size: None), args=flexmock()
+    )
+    other_process.should_receive('kill').once()
+    buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)}
+    process_metadatas = {
+        process: module.Process_metadata(last_lines=['hi', 'there'], capture=False),
+        other_process: module.Process_metadata(last_lines=['and', 'stuff'], capture=False),
+    }
+    flexmock(module).should_receive('interpret_exit_code').with_args(
+        object, 3, object, object
+    ).and_return(module.Exit_status.ERROR)
+    command = flexmock()
+    flexmock(module).should_receive('command_for_process').and_return(command)
+
+    with pytest.raises(module.subprocess.CalledProcessError) as error:
+        module.raise_for_process_errors(
+            buffer_readers,
+            process_metadatas,
+            borg_local_path=flexmock(),
+            borg_exit_codes=flexmock(),
+        )
+
+    assert error.value.returncode == 3
+    assert error.value.cmd == command
+    assert error.value.output == 'hi\nthere'
+
+
+def test_raise_for_process_errors_with_warning_process_and_long_output_raises_with_truncated_output():
+    process = flexmock(poll=lambda: 3, args=flexmock())
+    buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)}
+    process_metadatas = {
+        process: module.Process_metadata(last_lines=['hi', 'there'], capture=False)
+    }
+    flexmock(module).should_receive('interpret_exit_code').and_return(module.Exit_status.ERROR)
+    command = flexmock()
+    flexmock(module).should_receive('command_for_process').and_return(command)
+
+    with pytest.raises(module.subprocess.CalledProcessError) as error:
+        flexmock(module, ERROR_OUTPUT_MAX_LINE_COUNT=2).raise_for_process_errors(
+            buffer_readers,
+            process_metadatas,
+            borg_local_path=flexmock(),
+            borg_exit_codes=flexmock(),
+        )
+
+    assert error.value.returncode == 3
+    assert error.value.cmd == command
+    assert error.value.output == '...\nhi\nthere'
+
+
+def test_log_remaining_buffer_lines_without_buffer_readers_bails():
+    process_metadatas = {flexmock(): module.Process_metadata(last_lines=[], capture=False)}
+    flexmock(module).should_receive('parse_log_line').never()
+
+    assert (
+        tuple(
+            module.log_remaining_buffer_lines(
+                buffer_readers={},
+                process_metadatas=process_metadatas,
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+            )
+        )
+        == ()
+    )
+
+
+def test_log_remaining_buffer_lines_without_reader_process_bails():
+    buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=None)}
+    process_metadatas = {flexmock(): module.Process_metadata(last_lines=[], capture=False)}
+    flexmock(module).should_receive('parse_log_line').never()
+
+    assert (
+        tuple(
+            module.log_remaining_buffer_lines(
+                buffer_readers,
+                process_metadatas,
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+            )
+        )
+        == ()
+    )
+
+
+def test_log_remaining_buffer_lines_logs_each_line():
+    process = flexmock(stderr=flexmock(), args=flexmock())
+    buffer_readers = {
+        flexmock(): module.Buffer_reader(
+            lines=(('hi', 'there'), ('and', 'stuff')),
+            process=process,
+        )
+    }
+    process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)}
+    flexmock(module).should_receive('parse_log_line').with_args(
+        line=str, log_level=object, elevate_stderr=False, borg_local_path=object, command=object
+    ).and_return(flexmock()).times(4)
+    flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).times(4)
+
+    assert (
+        tuple(
+            module.log_remaining_buffer_lines(
+                buffer_readers,
+                process_metadatas,
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+            )
+        )
+        == ()
+    )
+
+
+def test_log_remaining_buffer_lines_with_multiple_buffers_logs_lines_from_each():
+    process = flexmock(stderr=flexmock(), args=flexmock())
+    buffer_readers = {
+        flexmock(): module.Buffer_reader(
+            lines=(('hi', 'there'),),
+            process=process,
+        ),
+        flexmock(): module.Buffer_reader(
+            lines=(('and', 'stuff'),),
+            process=process,
+        ),
+    }
+    process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)}
+    flexmock(module).should_receive('parse_log_line').with_args(
+        line=str, log_level=object, elevate_stderr=False, borg_local_path=object, command=object
+    ).and_return(flexmock()).times(4)
+    flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).times(4)
+
+    assert (
+        tuple(
+            module.log_remaining_buffer_lines(
+                buffer_readers,
+                process_metadatas,
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+            )
+        )
+        == ()
+    )
+
+
+def test_log_remaining_buffer_lines_with_stderr_buffer_elevates_stderr():
+    stderr = flexmock()
+    process = flexmock(stderr=stderr, args=flexmock())
+    buffer_readers = {
+        stderr: module.Buffer_reader(
+            lines=(('hi', 'there'),),
+            process=process,
+        ),
+        flexmock(): module.Buffer_reader(
+            lines=(('and', 'stuff'),),
+            process=process,
+        ),
+    }
+    process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)}
+    flexmock(module).should_receive('parse_log_line').with_args(
+        line=str, log_level=object, elevate_stderr=True, borg_local_path=object, command=object
+    ).and_return(flexmock()).twice()
+    flexmock(module).should_receive('parse_log_line').with_args(
+        line=str, log_level=object, elevate_stderr=False, borg_local_path=object, command=object
+    ).and_return(flexmock()).twice()
+    flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).times(4)
+
+    assert (
+        tuple(
+            module.log_remaining_buffer_lines(
+                buffer_readers,
+                process_metadatas,
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+            )
+        )
+        == ()
+    )
+
+
+def test_log_remaining_buffer_lines_with_stderr_buffer_and_capture_stderr_does_not_elevate_stderr():
+    stderr = flexmock()
+    process = flexmock(stderr=stderr, args=flexmock())
+    buffer_readers = {
+        stderr: module.Buffer_reader(
+            lines=(('hi', 'there'),),
+            process=process,
+        ),
+        flexmock(): module.Buffer_reader(
+            lines=(('and', 'stuff'),),
+            process=process,
+        ),
+    }
+    process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)}
+    flexmock(module).should_receive('parse_log_line').with_args(
+        line=str, log_level=object, elevate_stderr=False, borg_local_path=object, command=object
+    ).and_return(flexmock()).times(4)
+    flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).times(4)
+
+    assert (
+        tuple(
+            module.log_remaining_buffer_lines(
+                buffer_readers,
+                process_metadatas,
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+                capture_stderr=True,
+            )
+        )
+        == ()
+    )
+
+
+def test_log_remaining_buffer_lines_with_capture_process_yields_each_line():
+    process = flexmock(stderr=flexmock(), args=flexmock())
+    buffer_readers = {
+        flexmock(): module.Buffer_reader(
+            lines=(('hi', 'there'),),
+            process=process,
+        )
+    }
+    process_metadatas = {process: module.Process_metadata(last_lines=[], capture=True)}
+    flexmock(module).should_receive('parse_log_line').with_args(
+        line=str, log_level=object, elevate_stderr=False, borg_local_path=object, command=object
+    ).and_return(flexmock()).twice()
+    flexmock(module).should_receive('handle_log_record').and_return(
+        flexmock(levelno=None, getMessage=lambda: 'message')
+    ).twice()
+
+    assert tuple(
+        module.log_remaining_buffer_lines(
+            buffer_readers,
+            process_metadatas,
+            output_log_level=flexmock(),
+            borg_local_path=flexmock(),
+        )
+    ) == ('message', 'message')
+
+
+def test_log_remaining_buffer_lines_with_log_level_and_capture_process_does_not_yield_each_line():
+    process = flexmock(stderr=flexmock(), args=flexmock())
+    buffer_readers = {
+        flexmock(): module.Buffer_reader(
+            lines=(('hi', 'there'),),
+            process=process,
+        )
+    }
+    process_metadatas = {process: module.Process_metadata(last_lines=[], capture=True)}
+    flexmock(module).should_receive('parse_log_line').with_args(
+        line=str, log_level=object, elevate_stderr=False, borg_local_path=object, command=object
+    ).and_return(flexmock()).twice()
+    flexmock(module).should_receive('handle_log_record').and_return(
+        flexmock(levelno=10, getMessage=lambda: 'message')
+    ).twice()
+
+    assert (
+        tuple(
+            module.log_remaining_buffer_lines(
+                buffer_readers,
+                process_metadatas,
+                output_log_level=flexmock(),
+                borg_local_path=flexmock(),
+            )
+        )
+        == ()
+    )
+
+
 def test_mask_command_secrets_masks_password_flag_value():
     assert module.mask_command_secrets(('cooldb', '--username', 'bob', '--password', 'pass')) == (
         'cooldb',

From 104fe35e39b67e21e9a598dda416dd32e31013c3 Mon Sep 17 00:00:00 2001
From: Dan Helfman <witten@torsion.org>
Date: Fri, 23 Jan 2026 23:04:15 -0800
Subject: [PATCH 8/8] Add another test to get some additional coverage that's
 timing dependent (#1242).

---
 tests/integration/test_execute.py | 13 +++++++++++++
 1 file changed, 13 insertions(+)

diff --git a/tests/integration/test_execute.py b/tests/integration/test_execute.py
index dba1ba61..4617fe5d 100644
--- a/tests/integration/test_execute.py
+++ b/tests/integration/test_execute.py
@@ -38,6 +38,19 @@ def test_read_lines_yields_multiple_lines_plus_partial_line():
     assert tuple(module.read_lines(process.stdout, process)) == (('hi', 'there'), ('partial',))
 
 
+def test_read_lines_with_longer_running_process_yields_many_lines():
+    process = subprocess.Popen(
+        [
+            sys.executable,
+            '-c',
+            "import random, string; print('\\n'.join(random.choice(string.ascii_letters) for _ in range(1000)))",
+        ],
+        stdout=subprocess.PIPE,
+    )
+
+    assert tuple(module.read_lines(process.stdout, process))
+
+
 def test_read_lines_yields_nothing():
     process = subprocess.Popen(['echo', '-n'], stdout=subprocess.PIPE)