[SPARK-57787][CONNECT][FOLLOWUP] Harden persistent local Connect server management

[cloud-fan]

What changes were proposed in this pull request?

Followup to #56907.

This change hardens persistent local Spark Connect server management in two ways. Before sending
SIGTERM, the client verifies that the PID recorded in discovery still belongs to
SparkConnectServer. If process inspection fails transiently, discovery is preserved so shutdown
can be retried. The launcher also removes inherited binding-address configuration and explicitly
binds the server to 127.0.0.1.

Why are the changes needed?

A stale discovery file can refer to a PID that the operating system has reused for an unrelated
process. Signaling that process would be unsafe. Separately, an inherited binding-address setting
can expose a server that is intended to remain local.

Does this PR introduce any user-facing change?

No. The persistent local server remains opt-in. The change only makes its shutdown and network
binding behavior safer.

How was this patch tested?

Added regression tests covering a reused PID, an inconclusive process probe, and forced loopback
binding. Ran:

build/sbt -Phive package
python/run-tests --testnames pyspark.sql.tests.connect.test_connect_local_server

All 16 tests passed.

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

Generated-by: Codex (GPT-5)

[cloud-fan]

@ericm-db @HyukjinKwon

[uros-b]

LGTM, thank you @cloud-fan

[dongjoon-hyun]

Both hardenings are real improvements that match this feature's security model (loopback-only, same-user sharing), and treating a failed/timed-out ps probe as "not our server" is the right conservative default. A few comments, mostly minor:

1. Orphaned server when the ps probe fails transiently. stop() calls self._discovery.clear() regardless of the verification outcome. If the server is actually alive but ps transiently fails (e.g. hits the 5s timeout), we skip the signal yet still delete the discovery file and the daemon pid file, so a later --stop can never find that server again and the user has to kill it manually. It may be worth distinguishing "could not verify" (keep discovery) from "verified as a different process" (clear it), or at least reflecting this case in the --stop CLI output.

2. The _await_ready timeout path still signals without verification. The cleanup path in _await_ready SIGTERMs daemon_pid() unchecked. That pid was written moments earlier by spark-daemon.sh, so reuse risk is negligible — just noting it for consistency with the new guard.

3. TOCTOU window. There is still a window between the ps check and os.kill where the pid could be recycled. Unavoidable without pidfd, and the check shrinks the window substantially — fine as is, just noting the guard is a mitigation, not an elimination.

4. IPv6 environments. The client connects to sc://localhost:<port> while the server now binds only 127.0.0.1. On hosts where localhost resolves to ::1 first, this relies on the gRPC resolver falling back to IPv4 (it generally does, but IPv6-only setups would fail). The readiness probe is unaffected since it uses an AF_INET socket. Might deserve a one-line mention somewhere.

5. Explicitly-set binding.address is silently overridden. Intentional and consistent with how binding.port is handled, but since a user-provided conf is now silently dropped, a line in the module docstring would help.

6. Test mock breadth (nit). test_server_launcher_binds_to_loopback patches os.path.isfile globally to True, which also affects _find_spark_home() internals when SPARK_HOME is unset. Narrowing the patch to the script-existence check would make the test more robust.

7. PR description (nit). "The complete PySpark Connect test environment will be run before the staged proposal is published as an OSS PR" in the testing section reads like a leftover from internal tooling — probably worth replacing with the actual test command.

None of these are blockers.

[HyukjinKwon]

cc @gaogaotiantian

[cloud-fan]

Thanks for the detailed review. I addressed the actionable points in c419b45:

1. The process check now has an explicit inconclusive result. On a failed or timed-out ps probe, --stop preserves discovery and the daemon PID file so the user can retry, and reports that it could not verify the server.
2. I left the _await_ready timeout cleanup unchanged. That PID comes directly from the daemon launched in the same startup attempt, so the reuse risk is negligible; applying the same inconclusive handling there could instead leave a startup-time daemon with no discovery record.
3. Agreed that the ps/kill TOCTOU window remains. Avoiding it would require a platform-specific mechanism such as pidfd, so this guard is intentionally a mitigation.
4. Agreed on the IPv4 limitation. I left localhost in discovery because the Connect client currently uses that literal hostname to select gRPC local credentials; changing it to 127.0.0.1 would require a broader client-channel change.
5. The module documentation now states that the server always binds IPv4 loopback and overrides any configured binding address.
6. The launcher test now sets SPARK_HOME explicitly and verifies the single expected script-existence check, avoiding interaction with _find_spark_home().
7. I updated the PR testing section with the actual commands. build/sbt -Phive package and all 16 tests in pyspark.sql.tests.connect.test_connect_local_server pass.

Thanks again!