github-actions[bot] commented on code in PR #67624:
URL: https://github.com/apache/doris/pull/67624#discussion_r4058941549
##########
.github/workflows/code-review-runner.yml:
##########
@@ -711,33 +715,42 @@ jobs:
- name: Run automated code review
id: review
- timeout-minutes: 120
+ timeout-minutes: ${{ fromJSON(env.REVIEW_TIMEOUT_MINUTES) }}
continue-on-error: true
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
REPO: ${{ github.repository }}
PR_NUMBER: ${{ steps.review_inputs.outputs.pr_number }}
HEAD_SHA: ${{ steps.review_inputs.outputs.head_sha }}
+ BASE_SHA: ${{ steps.review_inputs.outputs.base_sha }}
+ HELPER_REF: ${{ github.workflow_sha || github.sha }}
Review Comment:
[P1] Pin the helper to the defining reusable workflow
This file exposes `workflow_call`, but in a reusable run the `github`
context is associated with the caller. Consequently
`github.repository`/`github.workflow_sha` can point at the caller rather than
the commit containing this job: a same-repo caller on an older SHA will fetch a
revision where this new helper does not exist, and a cross-repo caller can
fetch a different same-named file. The run then fails before Codex (or executes
the wrong helper). GitHub provides [`job.workflow_repository` and
`job.workflow_sha`](https://docs.github.com/en/actions/reference/workflows-and-actions/contexts#job-context)
for the defining workflow. Keep `REPO` for the PR target, but derive a
separate helper repository/ref from those fields for this and the other
co-located helper downloads.
##########
.github/scripts/run_review_with_resume.py:
##########
@@ -0,0 +1,467 @@
+#!/usr/bin/env python3
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Resume the same review after capacity failures, never restart the
workflow."""
+
+import argparse
+import ctypes
+import json
+import os
+import shutil
+import signal
+import subprocess
+import sys
+import threading
+import time
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+RETRY_DELAYS = (30, 60, 120)
+CAPACITY_MESSAGE = "Selected model is at capacity. Please try a different
model."
+PROCESS_EXIT_GRACE_SECONDS = 5
+
+
+class ChildReaper:
+ """Own orphaned commands in the standalone Linux helper, not the whole
runner."""
+
+ def __init__(self):
+ if sys.platform != "linux":
+ raise OSError("Review process supervision requires Linux")
+ # Fail before starting Codex if the runner cannot provide safe cleanup.
+ for module, name in (
+ (os, "pidfd_open"),
+ (os, "P_PIDFD"),
+ (signal, "pidfd_send_signal"),
+ ):
+ if not hasattr(module, name):
+ raise OSError(f"Review process supervision requires {name}")
+ fd = os.pidfd_open(os.getpid())
+ try:
+ signal.pidfd_send_signal(fd, 0)
+ try:
+ os.waitid(os.P_PIDFD, fd, os.WEXITED | os.WNOHANG | os.WNOWAIT)
+ except ChildProcessError:
+ pass # Expected: this process is not its own child.
+ finally:
+ os.close(fd)
+ self.children = Path(f"/proc/self/task/{os.getpid()}/children")
+ self.children.read_text()
+ prctl = ctypes.CDLL(None, use_errno=True).prctl
+ prctl.argtypes = [ctypes.c_int] + [ctypes.c_ulong] * 4
+ prctl.restype = ctypes.c_int
+ # PR_SET_CHILD_SUBREAPER: descendants that outlive Codex are reparented
+ # to this helper, even if shell/PTY commands created new sessions.
+ if prctl(36, 1, 0, 0, 0) != 0:
+ error = ctypes.get_errno()
+ raise OSError(error, os.strerror(error))
+
+ def reap(self):
+ # Called only after Popen.wait() reaps Codex. This dedicated helper has
+ # no other concurrent subprocesses; gh/help commands run between
attempts.
+ deadline = time.monotonic() + PROCESS_EXIT_GRACE_SECONDS
+ while children := self.children.read_text().split():
+ if time.monotonic() >= deadline:
+ raise OSError("Codex descendant cleanup did not finish; not
resuming")
+ for child in children:
+ try:
+ fd = os.pidfd_open(int(child))
+ except ProcessLookupError:
+ continue
+ try:
+ # Kernel-verified parenthood plus a stable pidfd prevents
+ # signalling an unrelated process if a PID was recycled.
+ os.waitid(os.P_PIDFD, fd, os.WEXITED | os.WNOHANG |
os.WNOWAIT)
+ signal.pidfd_send_signal(fd, signal.SIGKILL)
+ os.waitid(os.P_PIDFD, fd, os.WEXITED | os.WNOHANG)
+ except (ChildProcessError, ProcessLookupError):
+ pass
+ finally:
+ os.close(fd)
+ # Killing one orphan can reparent its children to us on the next
pass.
+ time.sleep(0.01)
+
+
+def read_events(path):
+ with path.open() as handle:
+ events = [json.loads(line) for line in handle if line.strip()]
+ if any(not isinstance(event, dict) for event in events):
+ raise ValueError("Codex JSONL contains a non-object event")
+ return events
+
+
+def failure(events, status, stderr_path):
+ for event_type in ("turn.failed", "error"):
+ for event in reversed(events):
+ if event.get("type") == event_type:
+ error = event.get("error") or event
+ return error.get("message") or f"Codex exited with status
{status}"
+ lines = stderr_path.read_text(errors="replace").splitlines()
+ return next(
+ (line for line in reversed(lines) if line.strip()),
+ f"Codex exited with status {status}",
+ )
+
+
+def session_id(events):
+ ids = [
+ event.get("thread_id")
+ for event in events
+ if event.get("type") == "thread.started"
+ ]
+ if len(ids) != 1 or not isinstance(ids[0], str):
+ raise ValueError("Expected exactly one main thread.started event")
+ # Names and --last can fall back to a new thread in codex exec. A UUID uses
+ # thread/resume directly and fails if the persisted thread cannot be
loaded.
+ if str(uuid.UUID(ids[0])) != ids[0]:
+ raise ValueError("Main session ID is not a canonical UUID")
+ return ids[0]
+
+
+def require_rollout(codex_home, thread_id, cwd):
+ for path in (codex_home / "sessions").rglob(f"*{thread_id}.jsonl"):
+ with path.open() as handle:
+ meta = json.loads(handle.readline())
+ payload = meta.get("payload") or {}
+ if (
+ meta.get("type") == "session_meta"
+ and payload.get("id") == thread_id
+ and payload.get("cwd")
+ and Path(payload["cwd"]).resolve() == cwd.resolve()
+ ):
+ return
+ raise ValueError("Main session rollout is missing; refusing to start a new
review")
+
+
+def stop_process(process):
+ try:
+ # Codex handles SIGINT through its graceful turn-interrupt/shutdown
path.
+ process.send_signal(signal.SIGINT)
+ process.wait(timeout=PROCESS_EXIT_GRACE_SECONDS)
+ except subprocess.TimeoutExpired:
+ pass
+ finally:
+ process.kill()
+ process.wait()
+
+
+def run_attempt(command, events_path, stderr_path, timeout, reaper=None):
+ with events_path.open("w") as stdout, stderr_path.open("w") as stderr:
+ process = None
+ copier = None
+ try:
+ process = subprocess.Popen(
+ command,
+ stdout=stdout,
+ stderr=subprocess.PIPE,
+ text=True,
+ encoding="utf-8",
+ errors="replace",
+ start_new_session=True,
+ )
+
+ def copy_stderr():
+ for line in process.stderr:
+ stderr.write(line)
+ stderr.flush()
+ print(line, end="", file=sys.stderr, flush=True)
+
+ copier = threading.Thread(target=copy_stderr, daemon=True)
+ # Do not interrupt Thread.start() between the native thread being
+ # created and its ident being published. The copier inherits this
+ # mask; pending cancellation reaches the main thread on
restoration,
+ # inside the cleanup-protected region. Codex was spawned unmasked.
+ previous_mask = signal.pthread_sigmask(
+ signal.SIG_BLOCK, {signal.SIGINT, signal.SIGTERM}
+ )
+ try:
+ copier.start()
+ finally:
+ signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask)
+ return process.wait(timeout=timeout)
+ finally:
+ # Finish bounded cleanup before delivering cancellation, including
a
+ # second signal while an earlier cancellation is already unwinding.
+ # The stderr copier inherited a blocked mask at startup, so pending
+ # SIGINT/SIGTERM can only reach this thread after everything is
closed.
+ previous_mask = signal.pthread_sigmask(
Review Comment:
[P1] Close the cancellation window before cleanup
A SIGINT/SIGTERM can arrive while this outer `finally` is entering, before
`pthread_sigmask` completes. Because the handler raises `KeyboardInterrupt`,
either a late first signal after normal process exit or a repeated signal after
cancellation can leave the `finally` before the nested cleanup `try` exists,
skipping `stop_process`, `reaper.reap`, the copier join, and pipe close. Fault
injection reproduced both variants with the reaper uncalled and a
child/descendant still alive. The new repeated-signal tests inject after
`pidfd_open`, once signals are already blocked, so they miss this window.
Please make handlers record cancellation and enter one guaranteed cleanup path
(or otherwise eliminate the unmasked entry window), and add a regression that
delivers a signal immediately before cleanup acquires its mask.
##########
.github/scripts/run_review_with_resume.py:
##########
@@ -0,0 +1,467 @@
+#!/usr/bin/env python3
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Resume the same review after capacity failures, never restart the
workflow."""
+
+import argparse
+import ctypes
+import json
+import os
+import shutil
+import signal
+import subprocess
+import sys
+import threading
+import time
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+RETRY_DELAYS = (30, 60, 120)
+CAPACITY_MESSAGE = "Selected model is at capacity. Please try a different
model."
+PROCESS_EXIT_GRACE_SECONDS = 5
+
+
+class ChildReaper:
+ """Own orphaned commands in the standalone Linux helper, not the whole
runner."""
+
+ def __init__(self):
+ if sys.platform != "linux":
+ raise OSError("Review process supervision requires Linux")
+ # Fail before starting Codex if the runner cannot provide safe cleanup.
+ for module, name in (
+ (os, "pidfd_open"),
+ (os, "P_PIDFD"),
+ (signal, "pidfd_send_signal"),
+ ):
+ if not hasattr(module, name):
+ raise OSError(f"Review process supervision requires {name}")
+ fd = os.pidfd_open(os.getpid())
+ try:
+ signal.pidfd_send_signal(fd, 0)
+ try:
+ os.waitid(os.P_PIDFD, fd, os.WEXITED | os.WNOHANG | os.WNOWAIT)
+ except ChildProcessError:
+ pass # Expected: this process is not its own child.
+ finally:
+ os.close(fd)
+ self.children = Path(f"/proc/self/task/{os.getpid()}/children")
+ self.children.read_text()
+ prctl = ctypes.CDLL(None, use_errno=True).prctl
+ prctl.argtypes = [ctypes.c_int] + [ctypes.c_ulong] * 4
+ prctl.restype = ctypes.c_int
+ # PR_SET_CHILD_SUBREAPER: descendants that outlive Codex are reparented
+ # to this helper, even if shell/PTY commands created new sessions.
+ if prctl(36, 1, 0, 0, 0) != 0:
+ error = ctypes.get_errno()
+ raise OSError(error, os.strerror(error))
+
+ def reap(self):
+ # Called only after Popen.wait() reaps Codex. This dedicated helper has
+ # no other concurrent subprocesses; gh/help commands run between
attempts.
+ deadline = time.monotonic() + PROCESS_EXIT_GRACE_SECONDS
+ while children := self.children.read_text().split():
+ if time.monotonic() >= deadline:
+ raise OSError("Codex descendant cleanup did not finish; not
resuming")
+ for child in children:
+ try:
+ fd = os.pidfd_open(int(child))
+ except ProcessLookupError:
+ continue
+ try:
+ # Kernel-verified parenthood plus a stable pidfd prevents
+ # signalling an unrelated process if a PID was recycled.
+ os.waitid(os.P_PIDFD, fd, os.WEXITED | os.WNOHANG |
os.WNOWAIT)
+ signal.pidfd_send_signal(fd, signal.SIGKILL)
+ os.waitid(os.P_PIDFD, fd, os.WEXITED | os.WNOHANG)
+ except (ChildProcessError, ProcessLookupError):
+ pass
+ finally:
+ os.close(fd)
+ # Killing one orphan can reparent its children to us on the next
pass.
+ time.sleep(0.01)
+
+
+def read_events(path):
+ with path.open() as handle:
+ events = [json.loads(line) for line in handle if line.strip()]
+ if any(not isinstance(event, dict) for event in events):
+ raise ValueError("Codex JSONL contains a non-object event")
+ return events
+
+
+def failure(events, status, stderr_path):
+ for event_type in ("turn.failed", "error"):
+ for event in reversed(events):
+ if event.get("type") == event_type:
+ error = event.get("error") or event
+ return error.get("message") or f"Codex exited with status
{status}"
+ lines = stderr_path.read_text(errors="replace").splitlines()
+ return next(
+ (line for line in reversed(lines) if line.strip()),
+ f"Codex exited with status {status}",
+ )
+
+
+def session_id(events):
+ ids = [
+ event.get("thread_id")
+ for event in events
+ if event.get("type") == "thread.started"
+ ]
+ if len(ids) != 1 or not isinstance(ids[0], str):
+ raise ValueError("Expected exactly one main thread.started event")
+ # Names and --last can fall back to a new thread in codex exec. A UUID uses
+ # thread/resume directly and fails if the persisted thread cannot be
loaded.
+ if str(uuid.UUID(ids[0])) != ids[0]:
+ raise ValueError("Main session ID is not a canonical UUID")
+ return ids[0]
+
+
+def require_rollout(codex_home, thread_id, cwd):
+ for path in (codex_home / "sessions").rglob(f"*{thread_id}.jsonl"):
+ with path.open() as handle:
+ meta = json.loads(handle.readline())
+ payload = meta.get("payload") or {}
+ if (
+ meta.get("type") == "session_meta"
+ and payload.get("id") == thread_id
+ and payload.get("cwd")
+ and Path(payload["cwd"]).resolve() == cwd.resolve()
+ ):
+ return
+ raise ValueError("Main session rollout is missing; refusing to start a new
review")
+
+
+def stop_process(process):
+ try:
+ # Codex handles SIGINT through its graceful turn-interrupt/shutdown
path.
+ process.send_signal(signal.SIGINT)
+ process.wait(timeout=PROCESS_EXIT_GRACE_SECONDS)
+ except subprocess.TimeoutExpired:
+ pass
+ finally:
+ process.kill()
+ process.wait()
+
+
+def run_attempt(command, events_path, stderr_path, timeout, reaper=None):
+ with events_path.open("w") as stdout, stderr_path.open("w") as stderr:
+ process = None
+ copier = None
+ try:
+ process = subprocess.Popen(
+ command,
+ stdout=stdout,
+ stderr=subprocess.PIPE,
+ text=True,
+ encoding="utf-8",
+ errors="replace",
+ start_new_session=True,
+ )
+
+ def copy_stderr():
+ for line in process.stderr:
+ stderr.write(line)
+ stderr.flush()
+ print(line, end="", file=sys.stderr, flush=True)
+
+ copier = threading.Thread(target=copy_stderr, daemon=True)
+ # Do not interrupt Thread.start() between the native thread being
+ # created and its ident being published. The copier inherits this
+ # mask; pending cancellation reaches the main thread on
restoration,
+ # inside the cleanup-protected region. Codex was spawned unmasked.
+ previous_mask = signal.pthread_sigmask(
+ signal.SIG_BLOCK, {signal.SIGINT, signal.SIGTERM}
+ )
+ try:
+ copier.start()
+ finally:
+ signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask)
+ return process.wait(timeout=timeout)
+ finally:
+ # Finish bounded cleanup before delivering cancellation, including
a
+ # second signal while an earlier cancellation is already unwinding.
+ # The stderr copier inherited a blocked mask at startup, so pending
+ # SIGINT/SIGTERM can only reach this thread after everything is
closed.
+ previous_mask = signal.pthread_sigmask(
+ signal.SIG_BLOCK, {signal.SIGINT, signal.SIGTERM}
+ )
+ try:
+ try:
+ try:
+ if process is not None:
+ stop_process(process)
+ finally:
+ # Popen itself may be interrupted before returning a
handle.
+ if reaper is not None:
+ reaper.reap()
+ finally:
+ if copier is not None and copier.ident is not None:
+ copier.join(timeout=5)
+ if copier.is_alive():
+ raise OSError(
+ "Codex stderr remained open after descendant
cleanup"
+ )
+ if process is not None:
+ process.stderr.close()
+ finally:
+ signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask)
+
+
+def append_events(source, target):
+ # Interrupted writes can leave a partial JSON line. Retain raw bytes in the
+ # attempt file, but keep the aggregate parseable by jq and the trace
uploader.
+ with source.open() as src, target.open("a") as dst:
+ for line in src:
+ if not line.strip():
+ continue
+ try:
+ event = json.loads(line)
+ except json.JSONDecodeError:
+ continue
+ if isinstance(event, dict):
+ dst.write(json.dumps(event) + "\n")
+
+
+def check_resume_target(args, started_at, remaining):
+ def api(path, paginated=False):
+ command = ["gh", "api", path]
+ if paginated:
+ command += ["--paginate", "--slurp"]
+ result = subprocess.run(
+ command,
+ check=True,
+ capture_output=True,
+ text=True,
+ timeout=min(30, remaining()),
+ )
+ return json.loads(result.stdout)
+
+ pr = api(f"repos/{args.repository}/pulls/{args.pr_number}")
+ if (
+ pr["state"] != "open"
+ or pr["head"]["sha"] != args.head_sha
+ or pr["base"]["sha"] != args.base_sha
+ ):
+ raise ValueError(
+ "PR base/head or open state changed; refusing to resume stale
context"
+ )
+ pages = api(
+ f"repos/{args.repository}/pulls/{args.pr_number}/reviews",
paginated=True
+ )
+ for page in pages:
+ for review in page:
+ if (
+ review.get("user", {}).get("login") == "github-actions[bot]"
+ and review.get("commit_id") == args.head_sha
+ and (review.get("submitted_at") or "") >= started_at
Review Comment:
[P1] Fence duplicate reviews without comparing two clocks
`started_at` is taken from the runner's wall clock, whereas `submitted_at`
is assigned by GitHub. If the runner is even one second ahead across the
truncated-second boundary, a review successfully submitted by the failed
attempt compares older than `started_at`, is ignored here, and the resumed goal
can submit it again. Snapshot the already-submitted current-head bot review IDs
before attempt 1 and fail closed when a new ID appears (or attach a unique run
marker); timestamps can remain supporting evidence but should not be the sole
idempotency fence.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
