From dcbce50ec28d70e19b13d37b828551eacfe99da3 Mon Sep 17 00:00:00 2001 From: Chinesezjc Date: Thu, 20 Aug 2026 15:29:02 +0800 Subject: [PATCH] fix(code-runtime-python): close the log-fragment OOM, CPU classification, and done-send transitive-dependency findings Addresses the bot's v16 review on the settlement-path code: - critical: _LogStream._pending now seals the fragment list past a chunk cap (like the host captureStray seal), so a newline-free single-character drip no longer accumulates one list slot per write and OOMs on its own accounting. - _clamped lowers a soft==hard result by one unit (when hard >= 2) so a dual-limit ulimit -t leaves SIGXCPU a window to fire and a definite CPU overrun is reported as a timeout, not a worker-exit. - send_done wraps its encode+write in a try and, on any throw from a rebound transitive name (_dump_scalar/os), writes a fixed pre-encoded done frame via the import-time captured os.write, so a settled exception verdict is never downgraded to worker-exit. - drainReplies clears the consumed replyQueue slot so a wide written payload is released immediately, bounding host memory to the current backlog under sustained fd-3 backpressure. Tests added for each (fragment cap drip, dual-limit CPU overrun, transitive-name rebind done frame). --- .../code-runtime-python/py/bootstrap.py | 80 ++++++++++++++++-- .../code-runtime-python/src/index.ts | 9 +- .../code-runtime-python/tests/runtime.spec.ts | 84 +++++++++++++++++++ 3 files changed, 167 insertions(+), 6 deletions(-) diff --git a/packages/code-runtime/code-runtime-python/py/bootstrap.py b/packages/code-runtime/code-runtime-python/py/bootstrap.py index 3885a11a99..a422622937 100644 --- a/packages/code-runtime/code-runtime-python/py/bootstrap.py +++ b/packages/code-runtime/code-runtime-python/py/bootstrap.py @@ -41,6 +41,22 @@ from protocol import PROTOCOL_FD, log_truncation_marker # noqa: E402 # simply takes more reads. 64 KiB matches the usual pipe capacity. _READ_CHUNK_BYTES = 65536 +# Captured primitive for the done-frame LAST-resort fallback. This bootstrap IS +# ``__main__``, so ``import __main__; __main__.os = ...`` would rebind ``os.write`` +# at call time inside ``ProtocolChannel.write_encoded``. ``os_write`` is a closure +# cell captured at import, before model code runs, so a one-line rebind cannot +# change which write the fallback uses. See ``send_done``'s try/except below. +_os_write = os.write + +# A fixed, pre-encoded done frame for the fallback. It carries no live model +# value, so it can always be written even when a transitive name (a ``_dump_*`` +# helper or ``os``) has been rebound and the normal encode/write threw. The +# message is the same fixed literal the failure reporter uses for an +# unrenderable diagnostic; the host renders the run as an exception rather than +# a worker-exit, which is the honest verdict for a settled run whose reporting +# was sabotaged. The bytes are JSON-valid and newline-terminated. +_FALLBACK_DONE_FRAME = b'{"type":"done","error":{"kind":"exception","message":""}}\n' + # Code-unit ceiling on the exception class name interpolated into the LAST-resort # failure diagnostic. A metaclass `__name__` property can return any length, and # that construction runs outside the guard that would otherwise absorb a @@ -179,6 +195,16 @@ class _LogStream(io.TextIOBase): # ``print("x", end="")`` must not concatenate quadratically. self._pending: list[str] = [] self._pending_chars = 0 + # A newline-free drip must not accumulate one list slot per ``write``: + # under a large ``maxLogBytes`` the list-of-fragments pointer array and + # the per-fragment str objects cost host memory well before the byte + # budget is reached, and a 25 M single-character drip would OOM on its + # own accounting (plus the same-size list ``_push_bounded_prefix`` then + # builds). Past this many fragments the chunks are sealed into one + # joined block (the character count is unchanged), bounding the live + # fragment count exactly as the host-side ``captureStray`` does with its + # ``MAX_PENDING_CHUNKS``. + self._PENDING_MAX_CHUNKS = 1024 def writable(self) -> bool: # noqa: D401 -- inherited contract return True @@ -292,6 +318,18 @@ class _LogStream(io.TextIOBase): else: self._pending.append(text) self._pending_chars += len(text) + # Seal the fragment list past the chunk cap: a newline-free drip + # appends one fragment per write, so a 25 M single-character flood + # would accumulate that many list slots (and str objects) long before + # the byte budget is met — the pointer array alone being ~25 M slots. + # Joining the fragments into one block keeps the SAME character count + # (`_pending_chars` is unchanged) while bounding the live fragment + # count, mirroring the host-side `captureStray` seal. The join is + # only as large as the buffered characters, which the budget already + # bounds; the fragments are otherwise un-sealable mid-newline because + # a newline never starts a multi-byte sequence. + if len(self._pending) >= self._PENDING_MAX_CHUNKS: + self._pending = ["".join(self._pending)] # A newline-free flood must hit the budget while running, not at # settlement: once the buffered tail alone can no longer fit the # ledger (chars lower-bound the serialized cost), push it through — LogBuffer @@ -640,7 +678,21 @@ def _clamped(which: int, soft: int, hard: int) -> tuple[int, int]: # invert them (a finite inherited soft below the clamped hard is fine, but a # requested hard below the inherited soft would leave soft > hard), so pin # soft under hard as the final step; the stricter hard ceiling wins. - return (min(clamped_soft, clamped_hard), clamped_hard) + result_soft = min(clamped_soft, clamped_hard) + result_hard = clamped_hard + # A soft limit EQUAL to the hard limit leaves the kernel no window to send + # SIGXCPU: it checks the hard limit in the same tick and SIGKILLs directly + # (a `ulimit -t N` sets both, and a busy loop then dies by SIGKILL, not + # SIGXCPU). The host classifies a CPU overrun ONLY on ``signal === + # 'SIGXCPU'``, so a definite budget exhaustion would be misreported as a + # `worker-exit`. Lowering the soft limit one unit below the hard (when the + # hard is at least 2, so soft stays positive) keeps the stricter-of-the-two + # containment semantics while giving SIGXCPU a window to fire — the CPU + # overrun is then reported as a timeout, not a worker-exit. For RLIMIT_AS + # this is one byte stricter, harmless. + if result_soft == result_hard and result_hard >= 2: + result_soft = result_hard - 1 + return (result_soft, result_hard) # --------------------------------------------------------------------------- @@ -913,10 +965,28 @@ async def _run(channel: ProtocolChannel) -> None: write_encoded_bound = channel.write_encoded def send_done(payload: dict[str, Any] | str) -> None: - if isinstance(payload, str): - write_encoded_bound(payload) - else: - write_encoded_bound(encode_plain_bound(payload)) + try: + if isinstance(payload, str): + write_encoded_bound(payload) + else: + write_encoded_bound(encode_plain_bound(payload)) + except BaseException: # noqa: BLE001 -- a rebind must not cost the done frame + # `encode_plain_bound`/`write_encoded_bound` are bound callables, but + # their BODIES still resolve transitive module globals at call time — + # `_encode_json_plain` reaches `_dump_scalar`/`_dump_string`/`json.dumps`, + # `write_encoded` reaches `os.write` (via the `os` module). This + # bootstrap is `__main__`, so `__main__._dump_scalar = boom` (or + # `__main__.os = ...`) makes the error-frame encode/write throw AFTER + # the `except` block, which would drop the `done` frame and downgrade a + # settled `exception` verdict to a host-side `worker-exit`. Write a fixed + # literal done frame with the captured `_os_write` (itself immune to a + # rebind) so the host still gets a verdict. The literal is JSON-valid + # and newline-terminated; the lock is the channel's, so the write is + # serialized against any concurrent writer. + with channel._write_lock: + view = memoryview(_FALLBACK_DONE_FRAME) + while view: + view = view[_os_write(channel._fd, view):] max_value_bytes = int(boot["maxValueBytes"]) done: dict[str, Any] | str diff --git a/packages/code-runtime/code-runtime-python/src/index.ts b/packages/code-runtime/code-runtime-python/src/index.ts index 855a04b0cf..be709339c3 100644 --- a/packages/code-runtime/code-runtime-python/src/index.ts +++ b/packages/code-runtime/code-runtime-python/src/index.ts @@ -1486,8 +1486,15 @@ export class PythonCodeRuntime extends CodeRuntime { // bindings awaiting fd 3's `drain` can queue many frames, and each // `shift()` re-slices the remaining array (O(n) per pop, O(n²) over // the whole drain). A head cursor keeps the cost linear; the `finally` - // below discards everything consumed once the drain ends. + // below discards everything consumed once the drain ends. The consumed + // slot is CLEARED here (not just advanced past) so a wide payload the + // pipe has already taken is released immediately: under sustained + // backpressure the drain loop can live across many `await drain` + // ticks, and leaving the slot set would pin the written value's bytes + // in `replyQueue` for the whole busy period, making host memory grow + // with cumulative processing rather than the current backlog. const payload = replyQueue[head] as ReplyMessage + replyQueue[head] = undefined as unknown as ReplyMessage head += 1 // Encode inside the loop, not up front: a queued reply the run no // longer needs is dropped by the `settled` check above without ever diff --git a/packages/code-runtime/code-runtime-python/tests/runtime.spec.ts b/packages/code-runtime/code-runtime-python/tests/runtime.spec.ts index 85376bf56b..a5425d7e2e 100644 --- a/packages/code-runtime/code-runtime-python/tests/runtime.spec.ts +++ b/packages/code-runtime/code-runtime-python/tests/runtime.spec.ts @@ -511,6 +511,37 @@ describe('PythonCodeRuntime — inherited resource limits', () => { expect(result.value).toBe(5) }, 15_000) + it('reports a CPU overrun under a dual-limit ulimit as a timeout, not a worker-exit', async () => { + // `ulimit -t N` sets BOTH the soft and hard CPU limit to N. The kernel + // checks the hard limit in the same tick and SIGKILLs a busy loop directly, + // so SIGXCPU is never delivered — and the host classifies a CPU overrun + // ONLY on `signal === 'SIGXCPU'`, so the overrun would be misreported as a + // `worker-exit` instead of a timeout. `_clamped` now lowers a clamped + // soft==hard result by one unit (when hard >= 2), so SIGXCPU fires at the + // softer limit and the run reports a timeout. This drives a busy loop past + // the inherited cap and asserts the run classifies as a timeout. + const dir = await mkdtemp(join(tmpdir(), 'dsh-rlimit-dual-')) + const wrapper = join(dir, 'python3-dual-capped') + // Both soft and hard CPU 1 s; configured cpuSeconds 30 s. + await writeFile(wrapper, '#!/bin/sh\nulimit -t 1\nexec python3 "$@"\n', { mode: 0o755 }) + const { runtime } = await setup({ pythonBin: wrapper, cpuSeconds: 30, maxWallMs: 12_000 }) + const result = await runtime.run({ + program: [ + 'import signal', + // Trap SIGXCPU; with the soft limit one unit below the hard it fires at + // 1 s and the run is classified as a CPU timeout, not a worker-exit. + 'signal.signal(signal.SIGXCPU, lambda *a: None)', + 'end = 2.5', + 'while True:', + ' pass', + 'return "unreachable"', + ].join('\n'), + bindings: [], + }) + expect(result.error?.kind).toBe('timeout') + expect(result.error?.message).toContain('SIGXCPU') + }, 15_000) + it('rechecks CPU at settlement against the effective inherited soft limit', async () => { // The settlement-time CPU recheck must compare against the EFFECTIVE soft // limit (`_clamped` may have lowered it to a stricter inherited value), not @@ -793,6 +824,31 @@ describe('PythonCodeRuntime — programs and bindings', () => { expect(result.logs.join('').length).toBeLessThan(4096) }) + it('bounds a newline-free single-character Python write drip by the fragment cap, not OOM', async () => { + // The child-side `_LogStream` buffers one fragment per `write` (so + // `print("x", end="")` does not concatenate quadratically). A newline-free + // drip of one character per call past a large `maxLogBytes` would otherwise + // accumulate one list slot (and one str object) per call — 25 M calls = + // ~25 M slots, which OOMs the host on its own accounting before the byte + // budget is reached. The stream seals the fragment list past + // `_PENDING_MAX_CHUNKS` into one joined block (character count unchanged), + // bounding the live fragment count exactly as the host-side `captureStray` + // seal does. This drives well past the cap and asserts the run still + // completes with a truncation marker rather than a MemoryError. + const { runtime } = await setup({ maxLogBytes: 4096 }) + const result = await runtime.run({ + program: [ + 'import sys', + 'for _ in range(200_000):', + ' sys.stdout.write("x")', + 'return None', + ].join('\n'), + bindings: [], + }) + expect(result.error).toBeUndefined() + expect(result.logs.at(-1)).toBe(logTruncationMarker(4096)) + }) + it('bounds a control-char-dense native residual by serialized cost, not raw length', async () => { // A newline-free NUL flood passes the cheap `length + 3` lower bound at a // raw length well under the budget, but each NUL serializes to `` (6 @@ -1427,6 +1483,34 @@ describe('PythonCodeRuntime — programs and bindings', () => { expect(result.error?.message).not.toContain('hijacked') }, 15_000) + it('still delivers a done frame when a transitive encode name is rebound', async () => { + // `send_done` binds `_encode_json_plain` and `ProtocolChannel.write_encoded` + // into locals, but those callables' BODIES still resolve transitive module + // globals at call time: `_encode_json_plain` reaches `_dump_scalar`/`_dump_string`/ + // `json.dumps`, and `write_encoded` reaches `os.write`. This bootstrap is + // `__main__`, so rebinding `__main__._dump_scalar` to a raising function makes + // the error-frame encode throw AFTER the `except` block. `send_done` catches + // that and writes a fixed literal done frame (kind `exception`) with the + // captured `os.write`, so the host still gets a verdict — the run must be an + // `exception`, never a `worker-exit`. The real message is lost (the literal + // carries a fixed `` text), which is acceptable: the verdict + // outranks the diagnostic detail. + const { runtime } = await setup({ maxWallMs: 10_000 }) + const result = await runtime.run({ + program: [ + 'import __main__', + 'def boom(*a, **k):', + ' raise RuntimeError("hijacked")', + '__main__._dump_scalar = boom', + '__main__.os = boom', + 'raise ValueError("real failure")', + ].join('\n'), + bindings: [], + }) + expect(result.error?.kind).toBe('exception') + expect(result.error?.kind).not.toBe('worker-exit') + }, 15_000) + it('bounds an over-cap exception-group nesting on the copy', async () => { // Exception groups link through `exceptions`, not the cause/context // dunders, so the cap has to count that edge too — otherwise a deeply