[SPARK-58021][CONNECT] Recover abandoned local pool launches

[ericm-db]

What changes were proposed in this pull request?

This is layer 6 of the ten-PR local Connect pool stack:

#57684 -> #57685 -> #57907 -> #57686 -> #58247 -> #58367 ->
#57687 -> #58248 -> #57102 -> #57688

This layer adds recovery before a pool member has finished publishing:

• validated pending-launch records;
• cleanup of dead, timed-out, malformed, and partially published launches;
• attendant command and UID checks before acting on a reused PID;
• complete cleanup of an attendant's launch process group;
• recovery of server PIDs from malformed records and daemon PID files; and
• stale conf-file and unreferenced member-directory garbage collection.

Acquisition, forceful purge, SparkSession integration, and warmup remain in later PRs.

Why are the changes needed?

An attendant or its spawning client can die at any point between writing the launch seed, starting
the JVM, and publishing the ready server. Those partial states must stop counting toward refills,
but cleanup must not signal an unrelated process after PID reuse.

This layer builds on #58247's server-retirement machinery while keeping the separate attendant and
process-group ownership contract independently reviewable.

Does this PR introduce any user-facing change?

No. The pool is not wired into SparkSession in this layer.

How was this patch tested?

All 59 focused pool tests at this stack layer passed:

PYTHONPATH=python:python/lib/pyspark.zip:python/lib/py4j-0.10.9.9-src.zip \
SPARK_TESTING=1 \
.venv/bin/python -m unittest -v \
pyspark.sql.tests.connect.test_connect_local_server_pool

The changed files also passed Python AST parsing, Ruff checking and formatting,
git diff --check, and changed-file ASCII and 100-column checks.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Fable 5) and OpenAI Codex (GPT-5)

[dtenedor]

Reviewed at head acfbb688 (diffed against its base ed95123c03b), including the sbin/spark-daemon.sh launch chain the recovery relies on and the new tests.

Overall this is careful work — the PID-reuse guards, generation-ID checks, and re-entrancy handling are thorough and well-commented. I found one correctness gap worth addressing before this lands, plus a couple of minor notes. Security posture looks good.

Correctness: the pending-recovery group kill can SIGKILL a published/claimed (in-use) server

reap() runs _reap_pending before _reap_server/_reap_claimed, and for a dead attendant _reap_pending signals the attendant's entire process group with SIGKILL unconditionally — without checking whether the launch already produced a server/claimed record:

ifnotattendant_alive:
ifpendingisnotNoneandnotself._signal_attendant_group(
attendant_pid, signal.SIGKILL, leader_may_be_dead=True
):
...
self.abort_launch(uid)

The daemonized server JVM stays in the attendant's process group: sbin/spark-daemon.sh backgrounds it with nohup -- "$@" ... & and no setsid, and nothing in start-connect-server.sh -> spark-daemon.sh -> spark-class calls setsid either. Since the attendant is the group leader (start_new_session=True), the server's pgid equals the attendant's pid, so killpg(attendant_pid, SIGKILL) hits the server itself.

Reachable sequence (all pool ops are serialized by the pool-dir flock):

1. Attendant dies mid-publish — after os.replace(... server-*.json) but before os.remove(pending-*.json) — leaving {server, pending}. Death releases the lock between those two syscalls, so the intermediate state is observable. This is in scope per the PR description ("die at any point ... between ... and publishing the ready server").
2. A client wins the next lock and claims it. claim() inspects only server-* records + is_usable(); it does not check for a pending marker. State becomes {claimed, pending} and the client is now serving on that JVM.
3. A janitor runs reap(uid) -> _reap_pending sees the dead attendant -> killpg -> SIGKILLs the in-use server, and _reap_claimed then retires the now-dead claim. The client's Connect session dies.

The unclaimed published case is fine (discarding matches intent), but test_reap_pending_with_published_server_retires_server uses a standalone sleep process that isn't in any attendant group, so it never exercises the killpg path.

I confirmed the OS mechanism with a standalone repro (attendant as session leader; a daemonized child left in its group; attendant killed+reaped; then the PR's exact _signal_attendant_group(..., leader_may_be_dead=True)):

server pgid=57356 (== attendant_pid? True)
after kill+reap: _pid_alive(attendant)=False server_alive=True
getpgid(attendant): ProcessLookupError (leader gone, group may persist)
_signal_attendant_group -> True; server_alive_now=False
RESULT: the surviving in-group server was KILLED by the group signal.

Severity is low-likelihood (needs an unclean death in the 2-syscall publish window, then a claim racing the janitor) but high-impact (kills a live session). Two possible directions, either of which resolves it:

• In _reap_pending, when a server or claimed record already exists for the uid, skip the process-group kill and let the generation-verified _reap_server/_reap_claimed/abort_launch paths own the server's lifecycle. By publish time the launch scaffolding (bash/spark-daemon.sh) has already exited, so the group's only remaining member is the server you don't want to blanket-kill. (abort_launch would then also want to avoid synthesizing a daemon-pid retirement for a live claimed server — only withdraw pending/conf in that case.)
• Or have claim() skip servers that still have a pending-* marker, so a server isn't claimable until pending recovery has resolved.

Minor

• Malformed record collapses the grace period. In _reap_pending, if PendingState.from_data returns None for any reason (e.g. a missing/empty fingerprint) while attendant_pid and created are individually valid, created is discarded and age becomes _LAUNCH_TIMEOUT_SECONDS + 1 (immediately "expired"). You already recover attendant_pid independently; recovering created the same way would avoid removing the launch-timeout grace for an otherwise-fresh launch. Bounded by the _is_pool_attendant identity check before any kill, so only your own uid is affected.
• Daemon-pid-only server can linger. If a half-started JVM is recorded only via the spark-daemon.sh pid file (no server-* record, hence no process_start_id) and it survives the group kill, abort_launch retires it by pid only and _reap_retired returns early forever while it is alive (if process_start_id is None: return), never signalling or cleaning up. This is the intended "don't signal an unverifiable/possibly-reused pid" trade-off and the group kill normally handles it — just flagging the retention corner.

Security

No concerns. Server signals go through generation IDs (boot_id:start_tick on Linux, else ps lstart); attendant-group signals combine getpgid(pid) == pid, the re-_pid_alive check under leader_may_be_dead, the POSIX guarantee that a PGID isn't recycled while the group is non-empty, and the ps-command identity check (_is_pool_attendant) on the still-alive path. The residual TOCTOU (pid recycled into a new group leader between getpgid raising and killpg) is negligible and same-user only. ps is invoked list-form with the pid rendered as str(pid) (no shell), state files stay 0600 / the pool dir 0700, and remove_member_dir only touches hex-validated member-<uid> paths.

Tests

Good breadth (dead / timed-out / malformed / reused-pid attendants, launch-group kill, published-then-orphaned server, interrupted-cleanup re-entrancy, conf GC, member-dir GC). The one gap that maps to the finding above: there is no test where a pending marker coexists with a claimed (or real in-group published) server whose attendant is dead — placing a real in-group process behind a claimed/server record and asserting it survives reap would catch it.

[ericm-db]

Thanks for the detailed review, @dtenedor. Fixed in 3ffdbe226ff: claim() now skips servers with a pending marker, closing the {server,pending} -> {claimed,pending} race, and pending recovery preserves an independently valid created timestamp from malformed records. The regression uses an otherwise-usable server in the process group of a dead attendant; all 66 focused tests pass. I left daemon-PID-only retention unchanged as the deliberate PID-reuse safety tradeoff you noted.

[dtenedor]

I re-reviewed the current head (3ffdbe2, which incorporates my earlier feedback) against its merge base: 3 files, +542/-17. I read all three in full context, traced every new branch, and ran the cheap CI-parity checks locally (py_compile, 100-col, non-ASCII -- all clean; 66 test methods).

Verdict

High-quality, defensive work. I found no correctness or security defects that block merge. The hard part -- signalling abandoned launches without ever touching an unrelated process after PID reuse -- is handled carefully and, as far as I can tell, correctly. The notes below are one test-coverage gap worth closing and a few low-severity observations.

Correctness

The change rests on the POSIX invariant stated in _signal_attendant_group's docstring: a process group outlives its leader while any member remains, and its id cannot be recycled during that time. That holds (on Linux the leader's struct pid stays pinned while it is referenced as any task's pgrp, so the value is reused neither as a PID nor a PGID until the group empties), and the code leans on it rather than on leader liveness.

The reuse handling -- the part I scrutinized most -- is sound in each path:

• Dead recorded attendant (leader_may_be_dead=True): a reaped leader whose group is still non-empty is allowed through to killpg (via getpgid -> ESRCH); a reused pid is rejected by getpgid(pid) != pid (reused as a non-leader) or by the _pid_alive(pid) re-check (reused as a new leader).
• Timed-out live attendant (leader_may_be_dead=False): identity is confirmed by _is_pool_attendant (module + --attend + matching --uid) before signalling, so a live-but-reused pid or an unrelated process is never killed.
• Server retirement still gates every signal on the process-generation id via _signal_server / _same_server_instance.

Also verified end-to-end: reap ordering and the had_retired guard against double-processing a freshly-created retired record; abort_launch's idempotent retired-already-present recovery and its non-double-removal of pending; and claim skipping any uid with a live pending marker, which closes the {server,pending} -> {claimed,pending} race from the last round.

One design choice worth confirming (not a bug): a fully-published, still-usable server whose attendant dies in the write server-* -> remove pending-* window is retired (killed), not adopted (test_reap_pending_with_published_server_retires_server). That discards a good warm JVM in a rare race. It's the safe choice and I think it's fine -- just flagging it as intentional.

Security

Trust boundary is the OS user (per-user 0700 dir, 0600 token, forced loopback bind; SECURITY.md -> the Spark security doc). This PR stays within it:

• No signal to a wrong / out-of-boundary process (see above).
• No command injection -- ps -ww -p <int> -o command= only ever interpolates str(pid).
• File perms unchanged; no new state escapes the 0600 / 0700 regime.
• A same-user process could spoof an --attend --uid <uid> command line, but the only effect is that the pool might SIGKILL that process -- self-inflicted, inside the trust boundary. Not a vulnerability.

Test coverage

Strong and precise -- most new branches have a dedicated test. One gap I'd close:

The case-A PID-reuse guard is untested.test_reap_does_not_signal_reused_attendant_pid exercises only the timed-out live path (is_attendant is False). The symmetric, more security-relevant path -- a dead recorded attendant whose pid is reused by a live process in the check->signal window, guarded here:

spark/python/pyspark/sql/connect/local_server_pool.py

Lines 717 to 729 in 3ffdbe2

	try:
	try:
	ifos.getpgid(pid) !=pid:
	returnFalse
	ifleader_may_be_deadand_pid_alive(pid):
	# The caller observed a dead leader, but this pid now belongs to a live
	# process. It was reused between checks, so do not signal its group.
	returnFalse
	exceptProcessLookupError:
	ifnotleader_may_be_dead:
	returnFalse
	os.killpg(pid, sig)
	returnTrue

-- has no test. It's easy to add by patching _pid_alive to report dead-then-alive (or reusing the live-process fixture) and asserting the group is never signalled. The pid == os.getpgrp() self-protection guards and _attendant_group_alive's EPERM->alive branch are similarly untested, though lower priority.

Low-severity / nits

• Malformed-pending launch-group leak: in _reap_pending, when the record fails validation (pending is None) but a pid is recovered and the attendant is dead, group signalling is skipped (if pending is not None and ...), so a surviving launch group would be orphaned. Only reachable via a corrupt pending-*.json (single atomic writer), so rare -- a one-line comment noting the deliberate skip would help the next reader.
• CLI-contract coupling:_is_pool_attendant hard-codes a space-separated -m <module> ... --attend ... --uid <uid> argv shape (no --uid=<uid> form; -m must immediately precede the module). Since a later layer implements the actual --attend entrypoint, consider sharing a constant / cross-reference so the two can't silently drift.
• claim lists the directory twice (paths_of_kind("pending") then ("server")); negligible at pool sizes, noting only for completeness.

Nice work -- this layer is easy to reason about in isolation.