Bound direct Git blob response reads

docs/research/git-blob-response-deadlines.md

@@ -0,0 +1,106 @@
+# Deadlines for direct Git blob responses
+
+The shared `git_blobs` reader now gives each complete request/response 90 seconds.
+Previously its synchronous header, payload and terminating-newline reads could
+block forever; the process-exit timeout was reached only after those reads.
+The bound includes writing/flushing the request and the initial unsupported-option
+decision. It ends before yielding the blob, so time spent by the consumer is
+excluded. This is a tooling repair, not mathematical verification or a new κ.
+
+One short-lived watchdog kills the direct Git process on expiry. The main thread
+cancels and joins the watchdog, checks its expiry flag, and raises
+`subprocess.TimeoutExpired` before any expired response can be yielded. It then
+reaps the child through the existing cleanup. Normal malformed replies retain
+their existing errors. The first-request/no-response/exit-129 fallback gate,
+exact object IDs and payload framing, one-blob memory use, source pins and
+selection policy are unchanged. There is no source cache or dependency change.
+
+The reader serves package validation, catalogue reads, package scaffolding,
+publication preflight and committed exports. This repair only covers their
+`git_blobs` batch request/response boundary. It does not give those entire
+commands a deadline. The separate `git()` helper (including legacy CR/LF-path
+resolution), the older `git_source_rows`, and descendant processes retaining
+pipe handles remain outside this bound. A direct-child kill cannot close a
+descendant's inherited stdout. Partial-clone/promisor helpers and full process-tree
+containment therefore need separate treatment. The 90-second bound is total
+response time, not an inactivity timer; an unusually slow valid response can time
+out even if it makes progress.
+
+## Evidence and reproduction
+
+Base and original reader: canonical main
+`b58ad9ef4d4e861b0eaa5dba41742bdc072dca51`, priority revision 7;
+`scripts/research_package.py` Git blob
+`ad2f94e3d38c3167feb66b07cbe6c52192284d39`.
+[Canonical claim](https://github.com/CrocSwap/integer-mult-bounds/discussions/4681#discussioncomment-19130292).
+
+The new tests substitute controlled real Python subprocesses at the Git launch
+boundary; they do not claim an observed production Git hang. On the unchanged
+reader, the no-header, partial-payload and missing-terminator controls all wait
+for the fixture's independent three-second safety kill, then raise framing
+errors instead of the required timeout. The original run fails all three
+controls; the delayed exact binary response and suspended-consumer control passes.
+After the repair all three bounded stalls raise `TimeoutExpired`, start no
+fallback, consume no later object specification, and reap the direct child.
+Files written after the fixture emits its chosen prefix establish that the
+intended response boundary was reached rather than merely timing out at startup.
+
+Further controls cover blocked request writes, a partial payload in the legacy
+line protocol, delayed binary responses, and pausing the consumer longer than
+the test deadline. A real Git process streams 128 small blobs with exact bytes
+and no live watchdog thread at any yield or after completion. Existing tests
+retain unsupported-option fallback, binary/empty/Unicode/CR-LF inputs, SHA256
+object IDs, invalid/missing/non-blob requests, bounded retry, early close,
+replacement-object rejection and exact path selection.
+
+Run the focused tests normally and with Python assertions disabled:
+
+```sh
+python3.12 -B -W error::ResourceWarning -m unittest discover -s tests -p 'test_research_git_*.py' -v
+python3.12 -O -B -W error::ResourceWarning -m unittest discover -s tests -p 'test_research_git_*.py' -v
+```
+
+All 37 focused tests pass normally (8.970 seconds) and under `-O` (8.983 seconds)
+on macOS 27 arm64, Python 3.12.2 and Git 2.44.0, with resource warnings treated
+as errors. The finite fixtures shorten the response deadline to 0.5 seconds (0.3 for the
+delayed positive control); production keeps 90 seconds. Their three-second
+independent safety kill bounds regression runs against the old code. The tests
+require Git and the Python standard library. This is not a native Windows or
+hosted CI claim. No full selected
+replay, broad historical suite, Lean or Docker run is included.
+
+An independent Codex agent reviewed the cancellation/yield ordering and protocol
+scope. Its readiness-evidence recommendation is included in the tests. This is
+AI-assisted review, not independent human acceptance.
+
+## Open next questions
+
+- Can the same boundary remain bounded when a descendant retains stdout? Start
+  with a controlled child that spawns a pipe-holding descendant, then design
+  process-tree containment without weakening one-blob streaming or early-close
+  cleanup. POSIX groups and native Windows lifetime ownership need separate
+  evidence; this patch does not supply it.
+- Does composing this deadline with the pending Windows closed-pipe repair
+  preserve `TimeoutExpired` at all native Windows cleanup boundaries? Apply
+  both changes and run the real-process and closed-stdin controls on Windows.
+  Native Windows access is required; no additional closed-pipe fix is claimed here.
+
+## Source lineage and overlap
+
+[PR #4686](https://github.com/CrocSwap/integer-mult-bounds/pull/4686), exact head
+`de93ebd044d0b00a8647a9f47f86939a6b18b9aa`, repairs Windows closed-stdin errors
+and explicitly leaves synchronous read deadlines separate. This patch is based
+on main and does not copy that repair. Both touch adjacent lines of
+`_git_blob_batch`; integration should retain its Windows error classification
+and this deadline's cancel/join boundary.
+
+Initial overlap covered 498 open PRs, 3,641 Discussion records and 5,860 local
+notes. Only the older compatibility work and its explicit remaining deadline
+limitation matched this question. Closed PR diffs and unpublished work were not
+searched; these counts are search coverage, not proof of novelty.
+
+Credit Nick Petalas for Git-version compatibility, Arham Tahir for pin batching,
+Woklerx for streamed provenance and cleanup checks, Mihir Srivastava for the
+Windows closed-pipe analysis, and Douglas Colkitt for the package contracts.
+Their frozen evidence is unchanged. Prepared for Brian (JoyboyBrian) with
+substantial OpenAI Codex assistance; Apache-2.0.

scripts/research_package.py

@@ -4,7 +4,7 @@
 Douglas Colkitt, with OpenAI Codex assistance. Apache-2.0.
 """
 import argparse
-from contextlib import closing
+from contextlib import closing, contextmanager
 from itertools import chain
 import hashlib
 import importlib.metadata
@@ -17,8 +17,10 @@
 import subprocess
 import sys
 import tempfile
+import threading
 
 ROOT = Path(__file__).resolve().parents[1]
+GIT_BLOB_RESPONSE_TIMEOUT = 90
 
 
 def require(ok, message):
@@ -139,6 +141,33 @@ class _UnsupportedNulBatch(Exception):
     """Git rejected the initial NUL-mode request before yielding any source."""
 
 
+@contextmanager
+def _git_blob_response_deadline(process):
+    """Bound one direct-child response, excluding time spent by its consumer.
+
+    Descendants retaining a pipe and separate git() calls need their own
+    containment/deadlines; killing the direct child cannot close their handles.
+    """
+    expired = threading.Event()
+
+    def terminate():
+        expired.set()
+        try:
+            process.kill()
+        except ProcessLookupError:
+            pass
+
+    timer = threading.Timer(GIT_BLOB_RESPONSE_TIMEOUT, terminate)
+    timer.start()
+    try:
+        yield
+    finally:
+        timer.cancel()
+        timer.join()
+        if expired.is_set():
+            raise subprocess.TimeoutExpired("git cat-file blob response", GIT_BLOB_RESPONSE_TIMEOUT)
+
+
 def _git_blob_batch(root, specs, nul):
     argv = ["git", "cat-file", "--batch"] + (["-z"] if nul else [])
     # Usage diagnostics from the unsupported-option probe are expected. Avoid
@@ -153,23 +182,24 @@ def _git_blob_batch(root, specs, nul):
                     # Resolve one literal argument, never split a path into requests.
                     request = git(root, "rev-parse", "--verify", "--end-of-options", spec).decode("ascii").strip()
                     require(re.fullmatch(r"[0-9a-f]{40,64}", request), "invalid Git object ID")
-                try:
-                    process.stdin.write(request.encode("utf-8") + (b"\0" if nul else b"\n"))
-                    process.stdin.flush()
-                except BrokenPipeError:
-                    header = b""
-                else:
-                    header = process.stdout.readline()
-                if nul and index == 0 and not header and process.wait(timeout=90) == 129:
-                    raise _UnsupportedNulBatch(spec)
-                fields = header.rstrip(b"\n").split(b" ")
-                require(len(fields) == 3 and re.fullmatch(rb"[0-9a-f]{40,64}", fields[0])
-                        and fields[1] == b"blob" and fields[2].isdigit(),
-                        "missing or non-blob Git source: " + spec)
-                size = int(fields[2])
-                pinned = process.stdout.read(size)
-                require(len(pinned) == size and process.stdout.read(1) == b"\n",
-                        "incomplete Git blob response: " + spec)
+                with _git_blob_response_deadline(process):
+                    try:
+                        process.stdin.write(request.encode("utf-8") + (b"\0" if nul else b"\n"))
+                        process.stdin.flush()
+                    except BrokenPipeError:
+                        header = b""
+                    else:
+                        header = process.stdout.readline()
+                    if nul and index == 0 and not header and process.wait(timeout=90) == 129:
+                        raise _UnsupportedNulBatch(spec)
+                    fields = header.rstrip(b"\n").split(b" ")
+                    require(len(fields) == 3 and re.fullmatch(rb"[0-9a-f]{40,64}", fields[0])
+                            and fields[1] == b"blob" and fields[2].isdigit(),
+                            "missing or non-blob Git source: " + spec)
+                    size = int(fields[2])
+                    pinned = process.stdout.read(size)
+                    require(len(pinned) == size and process.stdout.read(1) == b"\n",
+                            "incomplete Git blob response: " + spec)
                 yield fields[0].decode("ascii"), pinned
             process.stdin.close()
             require(process.wait(timeout=90) == 0, "Git source reader failed")

tests/test_research_git_deadline.py

@@ -0,0 +1,134 @@
+"""Real-process deadline boundaries for streaming Git source blobs."""
+from contextlib import closing, contextmanager
+import importlib.util
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[1]
+SPEC = importlib.util.spec_from_file_location("git_deadline_package", ROOT / "scripts/research_package.py")
+PACKAGE = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(PACKAGE)
+OID = b"a" * 40
+
+
+class GitBlobDeadlineTests(unittest.TestCase):
+    @contextmanager
+    def child(self, body, timeout=0.5):
+        real_popen = subprocess.Popen
+        children, calls, guards = [], [], []
+
+        def start(argv, **kwargs):
+            calls.append(argv)
+            code = "from pathlib import Path\nREADY = Path(" + repr(str(ready)) + ")\n" + body
+            process = real_popen([sys.executable, "-c", code], **kwargs)
+            children.append(process)
+            # Bound this regression even when run against the unfixed reader.
+            guard = threading.Timer(3, lambda: process.kill() if process.poll() is None else None)
+            guard.start()
+            guards.append(guard)
+            return process
+
+        with tempfile.TemporaryDirectory() as directory:
+            ready = Path(directory) / "response-ready"
+            try:
+                with (patch.object(PACKAGE.subprocess, "Popen", side_effect=start),
+                      patch.object(PACKAGE, "GIT_BLOB_RESPONSE_TIMEOUT", timeout, create=True)):
+                    yield Path(directory), children, calls
+            finally:
+                for guard in guards:
+                    guard.cancel()
+                    guard.join()
+                for process in children:
+                    if process.poll() is None:
+                        process.kill()
+                    process.wait(timeout=3)
+                    for stream in (process.stdin, process.stdout):
+                        if stream is not None and not stream.closed:
+                            stream.close()
+
+    def test_header_payload_and_terminator_stalls_timeout_without_fallback(self):
+        header = OID + b" blob 4\n"
+        for boundary, prefix in (("header", b""), ("payload", header + b"ab"),
+                                 ("terminator", header + b"abcd")):
+            body = ("import sys,time\n"
+                    "while sys.stdin.buffer.read(1) not in (bytes([0]), b''): pass\n"
+                    f"sys.stdout.buffer.write({prefix!r}); sys.stdout.buffer.flush()\n"
+                    "READY.write_text('response boundary reached')\n"
+                    "time.sleep(60)\n")
+            with self.subTest(boundary=boundary), self.child(body) as (root, children, calls):
+                consumed = []
+                def specs():
+                    consumed.append("first")
+                    yield "HEAD:first"
+                    consumed.append("later")
+                    yield "HEAD:later"
+                started = time.monotonic()
+                with self.assertRaises(subprocess.TimeoutExpired):
+                    list(PACKAGE.git_blobs(root, specs()))
+                self.assertLess(time.monotonic() - started, 2)
+                self.assertEqual(consumed, ["first"])
+                self.assertEqual(len(calls), 1)
+                self.assertIsNotNone(children[0].poll())
+                self.assertEqual((root / "response-ready").read_text(), "response boundary reached")
+
+    def test_blocked_request_write_times_out_and_reaps_child(self):
+        with self.child("import time; time.sleep(60)") as (root, children, calls):
+            with self.assertRaises(subprocess.TimeoutExpired):
+                list(PACKAGE.git_blobs(root, ["a" * 2_000_000]))
+            self.assertEqual(len(calls), 1)
+            self.assertIsNotNone(children[0].poll())
+
+    def test_legacy_batch_payload_stall_also_times_out(self):
+        body = ("import sys,time\n"
+                "sys.stdin.buffer.readline()\n"
+                f"sys.stdout.buffer.write({OID + b' blob 4' + bytes([10]) + b'ab'!r}); sys.stdout.buffer.flush()\n"
+                "READY.write_text('response boundary reached')\n"
+                "time.sleep(60)\n")
+        with self.child(body) as (root, children, calls):
+            with self.assertRaises(subprocess.TimeoutExpired):
+                list(PACKAGE._git_blob_batch(root, ["HEAD:first"], nul=False))
+            self.assertEqual(len(calls), 1)
+            self.assertIsNotNone(children[0].poll())
+            self.assertEqual((root / "response-ready").read_text(), "response boundary reached")
+
+    def test_delayed_binary_replies_and_suspended_consumer_remain_exact(self):
+        payload = b"\0\xffa\n\rZ"
+        reply = OID + b" blob 6\n" + payload + b"\n"
+        body = ("import sys,time\n"
+                "for _ in range(2):\n"
+                " while sys.stdin.buffer.read(1) not in (bytes([0]), b''): pass\n"
+                " time.sleep(.01)\n"
+                f" sys.stdout.buffer.write({reply!r}); sys.stdout.buffer.flush()\n")
+        with self.child(body, timeout=0.3) as (root, children, calls):
+            with closing(PACKAGE.git_blobs(root, ["HEAD:first", "HEAD:second"])) as rows:
+                self.assertEqual(next(rows), (OID.decode(), payload))
+                time.sleep(0.45)
+                self.assertIsNone(children[0].poll())
+                self.assertEqual(next(rows), (OID.decode(), payload))
+                with self.assertRaises(StopIteration):
+                    next(rows)
+            self.assertEqual(len(calls), 1)
+            self.assertEqual(children[0].poll(), 0)
+
+    def test_many_real_git_blobs_leave_no_watchdog_threads(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            PACKAGE.git(root, "init", "-q")
+            (root / "data").write_bytes(b"binary\0\xff\r\n")
+            oid = PACKAGE.git(root, "hash-object", "-w", "--", "data").decode().strip()
+            before = set(threading.enumerate())
+            with closing(PACKAGE.git_blobs(root, [oid] * 128)) as rows:
+                for item in rows:
+                    self.assertEqual(item, (oid, b"binary\0\xff\r\n"))
+                    self.assertEqual(set(threading.enumerate()), before)
+            self.assertEqual(set(threading.enumerate()), before)
+
+
+if __name__ == "__main__":
+    unittest.main()