fix(git): Report a provider stream breaking mid-response as a coded error

src/scm/git/relay.py

@@ -112,8 +112,9 @@ class GitRelay:
     one, the relay uses a plain ``HTTPAdapter``, which checks nothing.
 
     The relay replaces a requests or urllib3 exception from the upstream request with a coded error, so that it
-    cannot carry the token. ``map_error`` gets any other exception, such as the adapter's own for a refused host,
-    and returns either a coded error, which the relay raises in its place without the original, or None to let the
+    cannot carry the token, and so is one from reading the response body, whose frames hold the bytes read so far.
+    ``map_error`` gets any other exception from the request, such as the adapter's own for a refused host, and
+    returns either a coded error, which the relay raises in its place without the original, or None to let the
     original reach the caller unchanged. An error that reaches the caller this way does carry the token: an adapter
     refusing the host as it connects raises through urllib3's frames, which hold the Authorization header, so report
     such an error without its frames' local variables. GitServer answers a coded error like its other refusals. For
@@ -279,7 +280,7 @@ class _UpstreamStream:
     def __init__(self, session: requests.Session, upstream: requests.Response, chunk_size: int) -> None:
         self.session = session
         self.upstream = upstream
-        self.chunks = upstream.raw.stream(chunk_size, decode_content=False)
+        self.chunks = _read_body(upstream, chunk_size)
 
     def __iter__(self) -> "_UpstreamStream":
         return self
@@ -301,6 +302,20 @@ def close(self) -> None:
         self.session.close()
 
 
+def _read_body(upstream: requests.Response, chunk_size: int) -> Iterator[bytes]:
+    """Yield the provider's response body, raising the coded error ``forward`` would for the same failure."""
+    try:
+        yield from upstream.raw.stream(chunk_size, decode_content=False)
+    # "from None" drops urllib3's exception and the http.client one it chains, whose frames hold the partial chunk.
+    except urllib3.exceptions.ReadTimeoutError:
+        raise ResourceGatewayTimeout() from None
+    # A broken connection or a body cut off early. An SSLError here is the TLS connection breaking, not a certificate.
+    except (urllib3.exceptions.ProtocolError, urllib3.exceptions.IncompleteRead, urllib3.exceptions.SSLError):
+        raise ResourceServiceUnavailable() from None
+    except urllib3.exceptions.HTTPError:
+        raise ResourceBadGateway() from None
+
+
 def _read_until_eof(body: Readable, chunk_size: int) -> Iterator[bytes]:
     while chunk := _read(body, chunk_size):
         yield chunk

src/scm/git/server.py

@@ -80,6 +80,10 @@ class GitServer:
     ``client_closed_request`` when the client goes away first. A response the provider sends with a status other
     than 2xx is passed on to the client but counted as failed too, as ``provider_error``. The bytes sent each way
     are counted then too, as ``request_size`` and ``response_size``.
+
+    A coded error from the provider's stream breaking, such as a ``ResourceGatewayTimeout``, comes too late for a
+    refusal, so it goes to ``emit_error`` with the request's tags and is then raised, making the WSGI server drop
+    the connection rather than end the response as if it were whole. The caller need not report it again.
     """
 
     def __init__(
@@ -193,10 +197,13 @@ def finish(failure: str | None, bytes_sent: int) -> None:
             if bytes_sent:
                 self.record_count("sentry.scm.git.response_size", bytes_sent, tags)
 
+        def report(error: SCMCodedError) -> None:
+            self.emit_error(error, context)
+
         return StreamResponse(
             status_code=response.status_code,
             headers=response.headers,
-            content=_MeteredStream(response.content, finish),
+            content=_MeteredStream(response.content, finish, report),
         )
 
 
@@ -228,9 +235,15 @@ def read(self, size: int = -1, /) -> bytes:
 class _MeteredStream:
     """The relay's response stream, reporting once how it ended and how many bytes it passed on."""
 
-    def __init__(self, content: ResponseBody, finish: Callable[[str | None, int], None]) -> None:
+    def __init__(
+        self,
+        content: ResponseBody,
+        finish: Callable[[str | None, int], None],
+        report: Callable[[SCMCodedError], None],
+    ) -> None:
         self.content = content
         self.finish = finish
+        self.report = report
         self.bytes_sent = 0
         self.finished = False
 
@@ -243,6 +256,11 @@ def __next__(self) -> bytes:
         except StopIteration:
             self._end(None)
             raise
+        except SCMCodedError as e:
+            # Counted like any other break; the report says which it was.
+            self._end("stream_interrupted")
+            self.report(e)
+            raise
         except Exception:
             self._end("stream_interrupted")
             raise

tests/unit/error_reports.py

@@ -11,6 +11,16 @@ def assert_not_in_frames(error: BaseException, value: str) -> None:
         tb = tb.tb_next
 
 
+def assert_not_reported(error: BaseException, value: bytes) -> None:
+    """Fail if a report of ``error`` or what it chains, followed as Sentry's SDK does, would show ``value``."""
+    shown = repr(value)[2:-1]
+    reported: BaseException | None = error
+    while reported is not None:
+        assert shown not in str(reported) and shown not in repr(reported), f"{type(reported).__name__} holds it"
+        assert_not_in_frames(reported, shown)
+        reported = reported.__cause__ if reported.__suppress_context__ else reported.__context__
+
+
 def reported_local(error: BaseException, function: str, name: str) -> str:
     """How an error report shows the local variable ``name`` of ``function``'s frame in the error's traceback."""
     tb = error.__traceback__

tests/unit/test_git_relay.py

@@ -1,6 +1,8 @@
+import contextlib
 import gzip
 import io
 import socket
+import ssl
 import threading
 import time
 from collections.abc import Callable, Iterator
@@ -26,7 +28,7 @@
 from scm.git.relay import GitRelay, _UpstreamStream, authorize_request
 from scm.secret import Secret
 from scm.types import GitRemote
-from tests.unit.error_reports import assert_not_in_frames, reported_local
+from tests.unit.error_reports import assert_not_in_frames, assert_not_reported, reported_local
 from tests.unit.git_builders import make_grant
 from tests.unit.git_upstream import (
     ClosingAdapter,
@@ -538,6 +540,48 @@ def handler(h: UpstreamHandler) -> None:
         finally:
             release.set()
 
+    def test_provider_cutting_off_the_body(self, relay, upstream):
+        def handler(h: UpstreamHandler) -> None:
+            h.start_chunked(200)
+            h.write_chunk(b"first")
+            # A chunk that promises more than it sends, so http.client raises IncompleteRead.
+            h.wfile.write(b"%x\r\n" % (2 * len(PARTIAL_CHUNK)) + PARTIAL_CHUNK)
+            h.wfile.flush()
+            # A clean TLS close, so the relay sees the chunk end early rather than TLS breaking. It then fails
+            # waiting for the relay's close_notify.
+            with contextlib.suppress(ssl.SSLError):
+                h.connection.unwrap()  # type: ignore[attr-defined]
+            h.close_connection = True
+
+        upstream.handler = handler
+        content = fetch_refs(relay, upstream).content
+        assert next(content) == b"first"
+        with pytest.raises(ResourceServiceUnavailable) as exc_info:
+            next(content)
+        assert isinstance(exc_info.value.__context__, urllib3.exceptions.ProtocolError)
+        assert_body_not_in(exc_info.value)
+
+    def test_provider_going_silent_mid_body(self, upstream):
+        release = threading.Event()
+
+        def handler(h: UpstreamHandler) -> None:
+            h.start_chunked(200)
+            h.write_chunk(b"first")
+            h.wfile.write(b"%x\r\n" % (2 * len(PARTIAL_CHUNK)) + PARTIAL_CHUNK)
+            h.wfile.flush()
+            release.wait(WAIT)
+
+        upstream.handler = handler
+        try:
+            content = fetch_refs(GitRelay(ca_bundle=ca_bundle(), timeout=(1, 0.2)), upstream).content
+            assert next(content) == b"first"
+            with pytest.raises(ResourceGatewayTimeout) as exc_info:
+                next(content)
+            assert isinstance(exc_info.value.__context__, urllib3.exceptions.ReadTimeoutError)
+            assert_body_not_in(exc_info.value)
+        finally:
+            release.set()
+
     @pytest.mark.parametrize(
         "make_error",
         [
@@ -590,6 +634,16 @@ def assert_token_not_in(error: Exception) -> None:
     assert basic_auth("x-access-token", TOKEN.reveal()) not in repr(error)
 
 
+# Part of a pack, which must not reach an error report.
+PARTIAL_CHUNK = b"PACK-partial-repository-contents" * 64
+
+
+def assert_body_not_in(error: Exception) -> None:
+    assert error.__cause__ is None
+    assert error.__suppress_context__
+    assert_not_reported(error, PARTIAL_CHUNK[:32])
+
+
 def upstream_stream(*chunks: bytes | Exception) -> tuple[_UpstreamStream, MagicMock, MagicMock]:
     """An _UpstreamStream over a response whose body yields ``chunks``, raising any that is an exception."""
 
@@ -620,11 +674,43 @@ def test_closes_the_response_and_session_when_the_provider_stream_breaks(self):
         stream, session, response = upstream_stream(b"a", urllib3.exceptions.ProtocolError("connection broken"))
         assert next(stream) == b"a"
         response.close.assert_not_called()
-        with pytest.raises(urllib3.exceptions.ProtocolError):
+        with pytest.raises(ResourceServiceUnavailable):
             next(stream)
         response.close.assert_called_once()
         session.close.assert_called_once()
 
+    @pytest.mark.parametrize(
+        ("error", "expected"),
+        [
+            (
+                urllib3.exceptions.ReadTimeoutError(urllib3.HTTPSConnectionPool("127.0.0.1"), "", "Read timed out."),
+                ResourceGatewayTimeout,
+            ),
+            (urllib3.exceptions.ProtocolError("Connection broken"), ResourceServiceUnavailable),
+            (urllib3.exceptions.IncompleteRead(10, 5), ResourceServiceUnavailable),
+            # Not a certificate the relay does not trust: the connection is already up.
+            (urllib3.exceptions.SSLError("bad record mac"), ResourceServiceUnavailable),
+            (urllib3.exceptions.HTTPError("something else"), ResourceBadGateway),
+        ],
+        ids=lambda value: value.__name__ if isinstance(value, type) else type(value).__name__,
+    )
+    def test_provider_stream_breaking_is_a_coded_error(self, error, expected):
+        stream, _, _ = upstream_stream(b"a", error)
+        next(stream)
+        with pytest.raises(expected) as exc_info:
+            next(stream)
+        assert exc_info.value.__cause__ is None
+        assert exc_info.value.__suppress_context__
+
+    def test_other_errors_reach_the_caller_unchanged(self):
+        error = RuntimeError("not a urllib3 error")
+        stream, session, _ = upstream_stream(b"a", error)
+        next(stream)
+        with pytest.raises(RuntimeError) as exc_info:
+            next(stream)
+        assert exc_info.value is error
+        session.close.assert_called_once()
+
     def test_close_before_reading(self):
         # A response GitServer replaces, such as a 401, is closed without being read.
         stream, session, response = upstream_stream(b"a")

tests/unit/test_git_server.py

@@ -1,4 +1,6 @@
+import contextlib
 import io
+import ssl
 import time
 from collections.abc import Generator
 from types import SimpleNamespace
@@ -12,6 +14,7 @@
     GitGrantKeyError,
     GitRemoteRefused,
     RateLimitExceeded,
+    ResourceGatewayTimeout,
     ResourceServiceUnavailable,
 )
 from scm.git.grant import sign_grant
@@ -21,13 +24,22 @@
 from scm.rpc.types import StreamResponse
 from scm.secret import Secret
 from scm.types import GitRemote, Provider, Repository, RepositoryId
-from tests.unit.error_reports import assert_not_in_frames, reported_local
+from tests.unit.error_reports import assert_not_in_frames, assert_not_reported, reported_local
 from tests.unit.git_builders import make_grant, make_key_pair, make_provider, make_repository
-from tests.unit.git_upstream import ClosingAdapter, HostRefused, RefusingAdapter, basic_auth, running_upstream
+from tests.unit.git_upstream import (
+    ClosingAdapter,
+    HostRefused,
+    RefusingAdapter,
+    UpstreamHandler,
+    basic_auth,
+    running_upstream,
+)
 
 KEY = make_key_pair()
 TOKEN = Secret("ghs_secrettoken")
 REFERRER = make_grant().referrer
+# A constant, not a test's local, since the test's own frame is in the traceback.
+PARTIAL_PACK = b"PACK-partial-repository-contents" * 64
 
 
 def signed_grant(**overrides) -> str:
@@ -348,6 +360,53 @@ def broken_stream():
             )
         ]
         assert recorder.bytes == {("sentry.scm.git.response_size", "fetch"): 11}
+        # Not coded, so the caller reports it.
+        assert recorder.errors == []
+
+    def test_provider_stream_breaking_with_a_coded_error_is_reported_and_raised(self):
+        error = ResourceGatewayTimeout()
+
+        def broken_stream():
+            yield b"half a pack"
+            raise error
+
+        recorder = Recorder()
+        content = request(make_server(recorder, relay_returning(broken_stream())), signed_grant()).content
+        assert next(content) == b"half a pack"
+        assert recorder.errors == []
+        # Raised, so the WSGI server drops the connection.
+        with pytest.raises(ResourceGatewayTimeout) as exc_info:
+            next(content)
+        assert exc_info.value is error
+        assert recorder.errors == [
+            (
+                error,
+                {
+                    "repository_id": "5",
+                    "organization_id": "1",
+                    "referrer": REFERRER,
+                    "operation": "fetch",
+                    "provider": "github",
+                    "remote_host": "github.com",
+                },
+            )
+        ]
+        assert recorder.counts == [
+            (
+                "sentry.scm.git.request.failed",
+                {
+                    "referrer": REFERRER,
+                    "operation": "fetch",
+                    "provider": "github",
+                    "status": "200",
+                    "code": "stream_interrupted",
+                },
+            )
+        ]
+        # Reported once, however the caller goes on to close the stream.
+        content.close()  # type: ignore[attr-defined]
+        assert len(recorder.errors) == 1
+        assert len(recorder.counts) == 1
 
     def test_client_leaving_early_is_a_failure_and_closes_the_provider_stream(self):
         closed = []
@@ -586,6 +645,34 @@ def test_relays_end_to_end(self):
         assert received.headers["Authorization"] == basic_auth("x-access-token", TOKEN.reveal())
         assert GRANT_HEADER not in received.headers
 
+    def test_provider_cutting_off_the_body_end_to_end(self):
+        def handler(h: UpstreamHandler) -> None:
+            h.start_chunked(200)
+            h.write_chunk(b"first")
+            h.wfile.write(b"%x\r\n" % (2 * len(PARTIAL_PACK)) + PARTIAL_PACK)
+            h.wfile.flush()
+            # See test_git_relay's test_provider_cutting_off_the_body.
+            with contextlib.suppress(ssl.SSLError):
+                h.connection.unwrap()  # type: ignore[attr-defined]
+            h.close_connection = True
+
+        with running_upstream() as upstream:
+            upstream.handler = handler
+            recorder = Recorder(
+                repository=make_repository(provider_name="github_enterprise", web_base_url=upstream.url)
+            )
+            content = request(make_server(recorder, GitRelay(ca_bundle=upstream.ca_bundle)), signed_grant()).content
+            assert next(content) == b"first"
+            with pytest.raises(ResourceServiceUnavailable) as exc_info:
+                next(content)
+        assert exc_info.value.__cause__ is None
+        assert exc_info.value.__suppress_context__
+        assert_not_reported(exc_info.value, PARTIAL_PACK[:32])
+        [(error, context)] = recorder.errors
+        assert error is exc_info.value
+        assert set(context) == {"repository_id", "organization_id", "referrer", "operation", "provider", "remote_host"}
+        assert [tags["code"] for _, tags in recorder.counts] == ["stream_interrupted"]
+
 
 class TestKeptOutOfErrorReports:
     """What GitServer passes to emit_error or raises is reported to Sentry, with every frame's local variables."""