Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions contributors/emails/nmurray@gmail.com
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
keenvc
22 changes: 17 additions & 5 deletions cron/scheduler.py
Original file line number Diff line number Diff line change
Expand Up @@ -3520,11 +3520,23 @@ def _run_external_worker_payload(payload_path: Path, ack_path: Path) -> bool:
return False
try:
ack_path.parent.mkdir(parents=True, exist_ok=True)
fd = os.open(ack_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(fd, "w", encoding="utf-8") as ack_file:
json.dump({"pid": os.getpid(), "execution_id": execution_id}, ack_file)
ack_file.flush()
os.fsync(ack_file.fileno())
# Publish via write-to-temp + atomic rename. Writing ack_path in place
# (the old approach) let O_CREAT make the empty file visible to the
# scheduler's exists()-then-read polling loop before the JSON body was
# written, occasionally handing it a 0-byte file and a JSONDecodeError.
# os.replace() is a single atomic syscall on the same filesystem, so
# readers only ever see the file fully absent or fully written.
ack_tmp_path = ack_path.with_name(f"{ack_path.name}.tmp{os.getpid()}")
fd = os.open(ack_tmp_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
try:
with os.fdopen(fd, "w", encoding="utf-8") as ack_file:
json.dump({"pid": os.getpid(), "execution_id": execution_id}, ack_file)
ack_file.flush()
os.fsync(ack_file.fileno())
os.replace(ack_tmp_path, ack_path)
except BaseException:
ack_tmp_path.unlink(missing_ok=True)
raise
except Exception:
logger.exception(
"Cron external worker could not publish ready acknowledgement for %s",
Expand Down
37 changes: 37 additions & 0 deletions tests/cron/test_restart_safe_worker.py
Original file line number Diff line number Diff line change
Expand Up @@ -194,6 +194,43 @@ def test_external_worker_adopts_execution_and_runs_payload_once(
assert not stderr_capture.exists()


def test_external_worker_ack_is_never_observable_half_written(tmp_path, monkeypatch):
"""The gateway polls ``ack_path.exists()`` then reads it (#107184, #116164 form 1): the ack
must appear atomically with its full body, or the parent logs "unreadable acknowledgement"
and loses the worker pid for a handoff that actually succeeded."""
import cron.scheduler as scheduler

payload = tmp_path / "payload.json"
ack = tmp_path / "exec-1.ready"
payload.write_text(
json.dumps({
"job": {"id": "job-1", "execution_id": "exec-1"},
"profile_home": str(tmp_path / "profile"),
}),
encoding="utf-8",
)
monkeypatch.setattr(
"cron.executions.adopt_claimed_execution",
lambda execution_id: {"id": execution_id, "status": "running"})
monkeypatch.setattr(scheduler, "run_one_job", lambda *_a, **_k: True)

real_dump = json.dump
visible_while_writing = []

def spying_dump(obj, fp, *args, **kwargs):
# The body is being produced right now: a reader must not be able to see the ack yet.
visible_while_writing.append(ack.exists())
return real_dump(obj, fp, *args, **kwargs)

monkeypatch.setattr(scheduler.json, "dump", spying_dump)

assert scheduler._run_external_worker_payload(payload, ack) is True

assert visible_while_writing == [False]
assert json.loads(ack.read_text(encoding="utf-8"))["execution_id"] == "exec-1"
assert [p.name for p in tmp_path.iterdir() if p.name.startswith("exec-1")] == [ack.name]


def test_external_worker_refuses_to_run_without_durable_ownership(
tmp_path, monkeypatch
):
Expand Down
Loading