From 0732cb7f6c260d058e5a32487bf0b181613d3bea Mon Sep 17 00:00:00 2001 From: zehongye-gaga <277637559+zehongye-gaga@users.noreply.github.com> Date: Wed, 11 Nov 2026 12:37:30 +0800 Subject: [PATCH] Retry interrupted OCR dispatch requests with bounded backoff --- .github/workflows/codex-code-review.yml | 71 ++++++-- README.md | 4 +- ocr/ci_client.py | 71 ++++++-- ocr/test_ci_client.py | 12 +- ocr/test_transport.py | 212 ++++++++++++++++++++++++ 5 files changed, 337 insertions(+), 33 deletions(-) create mode 100644 ocr/test_transport.py diff --git a/.github/workflows/codex-code-review.yml b/.github/workflows/codex-code-review.yml index d3718a5..ee21680 100644 --- a/.github/workflows/codex-code-review.yml +++ b/.github/workflows/codex-code-review.yml @@ -25,10 +25,13 @@ jobs: # SPDX-License-Identifier: Apache-2.0 """Trusted workflow client; never check out or execute pull request code.""" import hashlib + import http.client import json import os + import random import re import signal + import ssl import sys import time import urllib.error @@ -39,6 +42,8 @@ jobs: WAIT_STATES = {'accepted', 'dispatching', 'queued', 'running', 'retrying'} FAIL_STATES = {'failed', 'cancelled', 'superseded'} TIMEOUT = 90 + MAX_ATTEMPTS = 4 + MAX_RESPONSE_BYTES = 65536 class ReviewError(Exception): @@ -117,30 +122,68 @@ jobs: raise ReviewError('OCR acceptance exceeded 90 seconds; re-run to reuse the task.') + def log_request(path, payload, attempt, started, status, error=None, retry_delay=None): + record = {'event': 'ocr_gateway_request', 'operation': path, + 'request_id': payload.get('request_id'), 'attempt': attempt, + 'elapsed_ms': round((time.monotonic() - started) * 1000), + 'http_status': status, 'error': error, 'retry_delay_seconds': retry_delay} + print(json.dumps(record, sort_keys=True), flush=True) + + def post(path, payload, token, deadline): data = json.dumps(payload).encode() - for attempt in range(4): - remaining = deadline - time.monotonic() + for attempt in range(MAX_ATTEMPTS): + started = time.monotonic() + remaining = deadline - started if remaining <= 0: raise ReviewError('Timed out waiting for OCR. Re-run this workflow to reuse the task.') + status = None req = urllib.request.Request(BASE_URL + path, data=data, headers={ 'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json'}) try: with urllib.request.build_opener(NoRedirect).open(req, timeout=min(15, remaining)) as response: - raw = response.read(65537) - if len(raw) > 65536: + status = response.status + length = response.headers.get('Content-Length') + if length is not None: + length = length.strip() + if length is not None and (not length.isascii() or not length.isdecimal()): + raise ReviewError('OCR returned an invalid response length.') + expected = int(length) if length is not None else None + if expected is not None and expected > MAX_RESPONSE_BYTES: + raise ReviewError('OCR response exceeds the allowed size.') + raw = response.read(MAX_RESPONSE_BYTES + 1) + if len(raw) > MAX_RESPONSE_BYTES: raise ReviewError('OCR response exceeds the allowed size.') - try: - return json.loads(raw) - except (ValueError, UnicodeError): - raise ReviewError('OCR returned an invalid JSON response.') from None + # A bounded read may return a short body without raising IncompleteRead. + if expected is not None and len(raw) < expected: + raise http.client.IncompleteRead(raw, expected - len(raw)) + reply = json.loads(raw) except urllib.error.HTTPError as error: - if error.code not in [408, 429, 500, 502, 503, 504]: - raise ReviewError(f'OCR gateway HTTP {error.code}; check configuration and credentials.') from None - except (urllib.error.URLError, TimeoutError): - pass - if attempt < 3: - time.sleep(min(2 ** attempt, max(0, deadline - time.monotonic()))) + status = error.code + error.close() + if status not in [408, 429, 500, 502, 503, 504]: + log_request(path, payload, attempt + 1, started, status, 'HTTPError') + raise ReviewError(f'OCR gateway HTTP {status}; check configuration and credentials.') from None + failure = 'HTTPError' + except (urllib.error.URLError, ConnectionError, TimeoutError, + ssl.SSLEOFError, http.client.IncompleteRead, http.client.BadStatusLine, + json.JSONDecodeError, UnicodeError) as error: + failure = type(error).__name__ + except ReviewError: + log_request(path, payload, attempt + 1, started, status, 'ReviewError') + raise + else: + log_request(path, payload, attempt + 1, started, status) + return reply + remaining = max(0, deadline - time.monotonic()) + delay = None + if attempt + 1 < MAX_ATTEMPTS and remaining > 0: + backoff = 2 ** attempt + delay = min(random.uniform(backoff / 2, backoff), remaining) + log_request(path, payload, attempt + 1, started, status, failure, delay) + if delay is None: + break + time.sleep(delay) raise ReviewError('OCR gateway remains unavailable after bounded retries; re-run this workflow.') diff --git a/README.md b/README.md index 8e01507..8b241da 100644 --- a/README.md +++ b/README.md @@ -11,7 +11,9 @@ The OCR dispatch entry point for selected public repositories in the organizatio 5. Exclude this public repository from the previous OCR Ruleset that references the private `github-policy` workflow. Leave other protection rules unchanged. 6. After completing the configuration, set the new ruleset to **Active** and verify it with a new PR or commit. -The workflow exits once the backend confirms that the task has been durably accepted. Submission and polling take no more than 90 seconds in total, excluding runner queue and startup time. The job has a hard timeout of 3 minutes. n8n forwards the request. The backend owns Agent routing, default fallback, dispatch and recovery after durable acceptance. The final `ocr-review` status is reported separately. It is currently advisory and is not configured as required. To require a successful review before merging, add a separate required status check. +The workflow exits once the backend confirms that the task has been durably accepted. Submission and network recovery take no more than 90 seconds in total, excluding runner queue and startup time. The job has a hard timeout of 3 minutes. n8n forwards the request. The backend owns Agent routing, default fallback, dispatch and recovery after durable acceptance. The final `ocr-review` status is reported separately. It is currently advisory and is not configured as required. To require a successful review before merging, add a separate required status check. + +Network failures, incomplete responses and transient HTTP errors retry the same submission up to four times, with exponential backoff and jitter. All attempts reuse the PR snapshot request ID and explicit workflow-rerun key. Each request logs its attempt number, elapsed milliseconds, HTTP status, error type and next retry delay. Credentials and response bodies are excluded from these logs. A v2 durable acceptance response ends the workflow immediately, including when status publication is still pending. Legacy acceptance responses can query the status endpoint within the same 90-second deadline. ## Security and limitations diff --git a/ocr/ci_client.py b/ocr/ci_client.py index fe73337..735144e 100644 --- a/ocr/ci_client.py +++ b/ocr/ci_client.py @@ -1,10 +1,13 @@ # SPDX-License-Identifier: Apache-2.0 """Trusted workflow client; never check out or execute pull request code.""" import hashlib +import http.client import json import os +import random import re import signal +import ssl import sys import time import urllib.error @@ -15,6 +18,8 @@ WAIT_STATES = {'accepted', 'dispatching', 'queued', 'running', 'retrying'} FAIL_STATES = {'failed', 'cancelled', 'superseded'} TIMEOUT = 90 +MAX_ATTEMPTS = 4 +MAX_RESPONSE_BYTES = 65536 class ReviewError(Exception): @@ -93,30 +98,68 @@ def timeout_handler(signum, frame): raise ReviewError('OCR acceptance exceeded 90 seconds; re-run to reuse the task.') +def log_request(path, payload, attempt, started, status, error=None, retry_delay=None): + record = {'event': 'ocr_gateway_request', 'operation': path, + 'request_id': payload.get('request_id'), 'attempt': attempt, + 'elapsed_ms': round((time.monotonic() - started) * 1000), + 'http_status': status, 'error': error, 'retry_delay_seconds': retry_delay} + print(json.dumps(record, sort_keys=True), flush=True) + + def post(path, payload, token, deadline): data = json.dumps(payload).encode() - for attempt in range(4): - remaining = deadline - time.monotonic() + for attempt in range(MAX_ATTEMPTS): + started = time.monotonic() + remaining = deadline - started if remaining <= 0: raise ReviewError('Timed out waiting for OCR. Re-run this workflow to reuse the task.') + status = None req = urllib.request.Request(BASE_URL + path, data=data, headers={ 'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json'}) try: with urllib.request.build_opener(NoRedirect).open(req, timeout=min(15, remaining)) as response: - raw = response.read(65537) - if len(raw) > 65536: + status = response.status + length = response.headers.get('Content-Length') + if length is not None: + length = length.strip() + if length is not None and (not length.isascii() or not length.isdecimal()): + raise ReviewError('OCR returned an invalid response length.') + expected = int(length) if length is not None else None + if expected is not None and expected > MAX_RESPONSE_BYTES: + raise ReviewError('OCR response exceeds the allowed size.') + raw = response.read(MAX_RESPONSE_BYTES + 1) + if len(raw) > MAX_RESPONSE_BYTES: raise ReviewError('OCR response exceeds the allowed size.') - try: - return json.loads(raw) - except (ValueError, UnicodeError): - raise ReviewError('OCR returned an invalid JSON response.') from None + # A bounded read may return a short body without raising IncompleteRead. + if expected is not None and len(raw) < expected: + raise http.client.IncompleteRead(raw, expected - len(raw)) + reply = json.loads(raw) except urllib.error.HTTPError as error: - if error.code not in [408, 429, 500, 502, 503, 504]: - raise ReviewError(f'OCR gateway HTTP {error.code}; check configuration and credentials.') from None - except (urllib.error.URLError, TimeoutError): - pass - if attempt < 3: - time.sleep(min(2 ** attempt, max(0, deadline - time.monotonic()))) + status = error.code + error.close() + if status not in [408, 429, 500, 502, 503, 504]: + log_request(path, payload, attempt + 1, started, status, 'HTTPError') + raise ReviewError(f'OCR gateway HTTP {status}; check configuration and credentials.') from None + failure = 'HTTPError' + except (urllib.error.URLError, ConnectionError, TimeoutError, + ssl.SSLEOFError, http.client.IncompleteRead, http.client.BadStatusLine, + json.JSONDecodeError, UnicodeError) as error: + failure = type(error).__name__ + except ReviewError: + log_request(path, payload, attempt + 1, started, status, 'ReviewError') + raise + else: + log_request(path, payload, attempt + 1, started, status) + return reply + remaining = max(0, deadline - time.monotonic()) + delay = None + if attempt + 1 < MAX_ATTEMPTS and remaining > 0: + backoff = 2 ** attempt + delay = min(random.uniform(backoff / 2, backoff), remaining) + log_request(path, payload, attempt + 1, started, status, failure, delay) + if delay is None: + break + time.sleep(delay) raise ReviewError('OCR gateway remains unavailable after bounded retries; re-run this workflow.') diff --git a/ocr/test_ci_client.py b/ocr/test_ci_client.py index ed5992d..a453173 100644 --- a/ocr/test_ci_client.py +++ b/ocr/test_ci_client.py @@ -109,23 +109,27 @@ def test_complete_submit_and_poll_flow_without_route(self): self.assertIn('separate ocr-review', (Path(directory) / 'summary.md').read_text()) def test_gateway_401_is_not_retried(self): - error = urllib.error.HTTPError('https://example.invalid', 401, 'No', {}, None) + error = urllib.error.HTTPError('https://example.invalid', 401, 'No', {}, io.BytesIO()) with patch.object(client.urllib.request, 'build_opener') as opener: opener.return_value.open.side_effect = error with self.assertRaises(client.ReviewError): client.post('ocr-ci-submit', {}, 'test', client.time.monotonic() + 60) self.assertEqual(opener.return_value.open.call_count, 1) def test_gateway_503_has_bounded_retries(self): - error = urllib.error.HTTPError('https://example.invalid', 503, 'Unavailable', {}, None) + error = urllib.error.HTTPError('https://example.invalid', 503, 'Unavailable', {}, io.BytesIO()) with patch.object(client.urllib.request, 'build_opener') as opener, patch.object(client.time, 'sleep'): opener.return_value.open.side_effect = error with self.assertRaises(client.ReviewError): client.post('ocr-ci-submit', {}, 'test', client.time.monotonic() + 60) self.assertEqual(opener.return_value.open.call_count, 4) def test_bad_json_is_not_success(self): - with patch.object(client.urllib.request, 'build_opener') as opener: - opener.return_value.open.return_value.__enter__.return_value.read.return_value = b'proxy error' + with patch.object(client.urllib.request, 'build_opener') as opener, patch.object(client.time, 'sleep'): + response = opener.return_value.open.return_value.__enter__.return_value + response.status = 200 + response.headers = {} + response.read.return_value = b'proxy error' with self.assertRaises(client.ReviewError): client.post('ocr-ci-status', {}, 'test', client.time.monotonic() + 60) + self.assertEqual(opener.return_value.open.call_count, client.MAX_ATTEMPTS) if __name__ == '__main__': unittest.main() diff --git a/ocr/test_transport.py b/ocr/test_transport.py new file mode 100644 index 0000000..5398b09 --- /dev/null +++ b/ocr/test_transport.py @@ -0,0 +1,212 @@ +# SPDX-License-Identifier: Apache-2.0 +import contextlib +import http.client +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import io +import json +import os +from pathlib import Path +import socket +import ssl +import tempfile +import threading +import time +import unittest +from unittest.mock import patch + +import ci_client as client + + +IDENTITY = {'repository': 'lexmount/example', 'pull_number': 12, + 'head_sha': 'a' * 40, 'base_sha': 'b' * 40, 'base_ref': 'main'} +RECEIPT = {'schema': client.SCHEMA, 'request_id': client.request_id(IDENTITY), **IDENTITY} +ACK = {**RECEIPT, 'state': 'accepted', 'contract_version': 2, 'durable': True, + 'acceptance': 'accepted', 'scheduler': 'backend', 'task_id': RECEIPT['request_id']} + + +class TransportTests(unittest.TestCase): + @contextlib.contextmanager + def gateway(self, responses): + requests = [] + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_POST(self): + raw = self.rfile.read(int(self.headers['Content-Length'])) + requests.append({'path': self.path, 'body': raw}) + behavior = responses[min(len(requests) - 1, len(responses) - 1)] + if behavior == 'disconnect': + self.connection.shutdown(socket.SHUT_RDWR) + self.connection.close() + return + if behavior == 'chunked_disconnect': + self.send_response(200) + self.send_header('Transfer-Encoding', 'chunked') + self.end_headers() + self.wfile.write(b'40\r\n{"durable":') + self.wfile.flush() + self.close_connection = True + return + body = json.dumps(ACK).encode() + status = 200 + length = str(len(body)) + if behavior == 'truncated': + body = body[:20] + elif behavior == 'short_valid_json': + length = str(len(body) + 20) + elif behavior == 'invalid_json': + body = b'{"durable":' + length = str(len(body)) + elif behavior == 'invalid_utf8': + body = b'\xff' + length = '1' + elif behavior == 'oversized': + body = b' ' * (client.MAX_RESPONSE_BYTES + 1) + length = str(len(body)) + elif behavior == 'oversized_without_length': + body = b' ' * (client.MAX_RESPONSE_BYTES + 1) + length = None + elif behavior == 'invalid_length': + length = 'invalid' + elif isinstance(behavior, int): + status = behavior + self.send_response(status) + if length is not None: + self.send_header('Content-Length', length) + self.end_headers() + try: + self.wfile.write(body) + self.wfile.flush() + except ConnectionError: + pass + self.close_connection = True + + server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) + worker = threading.Thread(target=server.serve_forever, kwargs={'poll_interval': 0.01}, daemon=True) + worker.start() + try: + with patch.object(client, 'BASE_URL', f'http://127.0.0.1:{server.server_port}/'): + yield requests + finally: + server.shutdown() + server.server_close() + worker.join(timeout=2) + + def send(self): + return client.post('ocr-ci-submit', RECEIPT, 'secret-not-for-logs', time.monotonic() + 30) + + def records(self, output): + return [json.loads(line) for line in output.getvalue().splitlines() + if line.startswith('{')] + + def test_lost_ack_and_incomplete_bodies_retry_the_same_submission(self): + cases = {'disconnect': 'RemoteDisconnected', 'truncated': 'IncompleteRead', + 'short_valid_json': 'IncompleteRead', + 'chunked_disconnect': 'IncompleteRead', 'invalid_json': 'JSONDecodeError', + 'invalid_utf8': 'UnicodeDecodeError'} + for behavior, error in cases.items(): + with self.subTest(behavior=behavior), self.gateway([behavior, 'ok']) as requests: + output = io.StringIO() + with contextlib.redirect_stdout(output), patch.object(client.time, 'sleep') as sleep: + self.assertEqual(self.send(), ACK) + self.assertEqual(len(requests), 2) + self.assertEqual(requests[0], requests[1]) + self.assertEqual(requests[0]['path'], '/ocr-ci-submit') + records = self.records(output) + self.assertEqual([r['attempt'] for r in records], [1, 2]) + self.assertEqual(records[0]['error'], error) + self.assertIsNone(records[1]['error']) + self.assertTrue(all(r['elapsed_ms'] >= 0 for r in records)) + self.assertTrue(all(r['request_id'] == RECEIPT['request_id'] for r in records)) + self.assertEqual(records[1]['http_status'], 200) + self.assertEqual(sleep.call_count, 1) + self.assertNotIn('secret-not-for-logs', output.getvalue()) + self.assertNotIn('durable', output.getvalue()) + + def test_repeated_disconnects_have_four_attempts_and_bounded_backoff(self): + with self.gateway(['disconnect']) as requests, contextlib.redirect_stdout(io.StringIO()) as output: + with patch.object(client.time, 'sleep') as sleep, patch.object(client.random, 'uniform', side_effect=lambda low, high: high): + with self.assertRaises(client.ReviewError): + self.send() + self.assertEqual(len(requests), 4) + self.assertEqual([call.args[0] for call in sleep.call_args_list], [1, 2, 4]) + self.assertIsNone(self.records(output)[-1]['retry_delay_seconds']) + + def test_transient_http_statuses_retry_and_permanent_statuses_stop(self): + for status in [408, 429, 500, 502, 503, 504, 401, 403, 404, 409, 302]: + retryable = status in [408, 429, 500, 502, 503, 504] + with self.subTest(status=status), self.gateway([status, 'ok']) as requests: + with contextlib.redirect_stdout(io.StringIO()) as output, patch.object(client.time, 'sleep'): + if retryable: + self.assertEqual(self.send(), ACK) + else: + with self.assertRaises(client.ReviewError): + self.send() + self.assertEqual(len(requests), 2 if retryable else 1) + self.assertEqual(self.records(output)[0]['http_status'], status) + self.assertEqual(self.records(output)[0]['error'], 'HTTPError') + + def test_invalid_length_and_oversized_responses_stop_without_retry(self): + for behavior in ['invalid_length', 'oversized', 'oversized_without_length']: + with self.subTest(behavior=behavior), self.gateway([behavior]) as requests: + with contextlib.redirect_stdout(io.StringIO()) as output, patch.object(client.time, 'sleep') as sleep: + with self.assertRaises(client.ReviewError): + self.send() + self.assertEqual(len(requests), 1) + sleep.assert_not_called() + self.assertEqual(self.records(output)[0]['error'], 'ReviewError') + + def test_socket_and_tls_disconnects_are_retried(self): + for error in [ConnectionResetError('connection reset'), BrokenPipeError('broken pipe'), + TimeoutError('timed out'), ssl.SSLEOFError('TLS EOF'), + http.client.IncompleteRead(b'partial'), http.client.BadStatusLine('HTTP/')]: + with self.subTest(error=type(error).__name__), patch.object(client.urllib.request, 'build_opener') as opener: + response = opener.return_value.open.return_value.__enter__.return_value + response.status = 200 + response.headers = {} + response.read.side_effect = [error, json.dumps(ACK).encode()] + with contextlib.redirect_stdout(io.StringIO()) as output, patch.object(client.time, 'sleep'): + self.assertEqual(self.send(), ACK) + self.assertEqual(opener.return_value.open.call_count, 2) + self.assertEqual(self.records(output)[0]['error'], type(error).__name__) + + def test_retry_sleep_and_next_attempt_obey_the_deadline(self): + with self.gateway(['disconnect']) as requests, contextlib.redirect_stdout(io.StringIO()): + with patch.object(client.time, 'monotonic', side_effect=[10, 10.5, 10.5, 11]), patch.object(client.time, 'sleep') as sleep: + with self.assertRaises(client.ReviewError): + client.post('ocr-ci-submit', RECEIPT, 'secret', 11) + self.assertEqual(len(requests), 1) + self.assertEqual(sleep.call_args.args[0], 0.5) + + def test_hard_timeout_is_logged_and_never_retried(self): + with patch.object(client.urllib.request, 'build_opener') as opener: + opener.return_value.open.side_effect = client.ReviewError('OCR acceptance exceeded 90 seconds') + with contextlib.redirect_stdout(io.StringIO()) as output, patch.object(client.time, 'sleep') as sleep: + with self.assertRaises(client.ReviewError): + self.send() + sleep.assert_not_called() + self.assertEqual(opener.return_value.open.call_count, 1) + self.assertEqual(self.records(output)[0]['error'], 'ReviewError') + + def test_main_exits_after_durable_ack_without_a_status_query(self): + event = {'pull_request': {'number': IDENTITY['pull_number'], 'state': 'open', + 'head': {'sha': IDENTITY['head_sha']}, 'base': {'sha': IDENTITY['base_sha'], + 'ref': IDENTITY['base_ref'], 'repo': {'full_name': IDENTITY['repository']}}}} + with tempfile.TemporaryDirectory() as directory, self.gateway(['disconnect', 'ok']) as requests: + event_file = Path(directory) / 'event.json' + event_file.write_text(json.dumps(event)) + env = {'GITHUB_EVENT_NAME': 'pull_request_target', 'GITHUB_REPOSITORY': IDENTITY['repository'], + 'GITHUB_EVENT_PATH': str(event_file), 'OCR_CI_TOKEN': 'secret-not-for-logs', + 'GITHUB_RUN_ID': '123', 'GITHUB_RUN_ATTEMPT': '2'} + with patch.dict(os.environ, env, clear=True), contextlib.redirect_stdout(io.StringIO()), patch.object(client.time, 'sleep'): + self.assertEqual(client.main(), 0) + self.assertEqual([r['path'] for r in requests], ['/ocr-ci-submit', '/ocr-ci-submit']) + payloads = [json.loads(r['body']) for r in requests] + self.assertEqual(payloads[0], payloads[1]) + self.assertEqual(payloads[0]['retry_key'], '123:2') + + +if __name__ == '__main__': + unittest.main()