fix(executor): bind the gRPC listener before registering with the scheduler

[andygrove]

Which issue does this PR close?

Closes #2224.

Rationale for this change

In push-based scheduling the executor tells the scheduler about itself before its own gRPC server is listening, and the scheduler dials that port back to verify connectivity. When the callback wins the race the executor dies at startup.

The ordering in executor_server::startup was:

1. tokio::spawn the tonic server. Server::serve binds inside the future, so the socket does not exist yet when spawn returns — even though the "... Grpc Server listening on ..." line is already in the log.
2. register_executor, immediately.

On the scheduler side ExecutorManager::register_executor calls test_connectivity, a single ExecutorGrpcClient::connect(...) with no retry and no backoff. If the spawned task has not reached its bind yet, that connect gets ECONNREFUSED, registration returns an error, and executor_process treats a registration error as fatal, so the executor exits instead of joining the cluster.

The opposite direction is already tolerant: the executor retries its connection to the scheduler in a loop. Only the scheduler's callback to the executor is one-shot, which is why the executor's own startup ordering has to be right.

It is timing-dependent and so shows up rarely and on loaded machines. The instance that prompted this was CI, where one of two executors lost the race and the HA chaos harness timed out waiting for it:

timed out waiting for 2 executors to register

# executor-1
INFO  Ballista v54.0.0 Rust Executor Grpc Server listening on 127.0.0.1:42635
ERROR Executor registration failed due to: ... Failed to register executor at 127.0.0.1:42635,
      could not connect: ... Os { code: 111, kind: ConnectionRefused }

# scheduler, same second
ERROR Fail to do executor registration due to: ... ConnectionRefused ...

The consequence outside of tests is the same shape: an executor that is otherwise healthy fails to join, and in a restart loop it can keep failing to join.

What changes are included in this PR?

• ballista/core/src/utils.rs — new create_grpc_server_incoming(addr, &GrpcServerConfig), which binds the listening socket eagerly and returns tonic's TcpIncoming. tonic ignores the server builder's tcp_nodelay / tcp_keepalive when serving from a pre-bound listener, so the helper applies the same values create_grpc_server sets and keeps the two in one place. The remaining settings (timeout, HTTP/2 keep-alive) still come from the builder as before, so the socket is configured exactly as it was.
• ballista/executor/src/executor_server.rs — startup binds via that helper on the current task, before spawning, and the spawned task now serves with serve_with_incoming_shutdown. Registration therefore cannot run before the port is accepting connections. This retires the standing // TODO the executor registration should happen only after the executor grpc server started. The "listening on" log line is now true when it is printed.

One incidental improvement falls out of binding eagerly: a port conflict now fails startup directly, with the bind error, rather than being discovered later through the spawned server task.

No change to the scheduler's test_connectivity. The issue also floats giving it a bounded retry; that is worth doing on its own merits for genuinely remote executors, but it is a separate resilience change and is not needed to fix this race, which is entirely local to the executor's startup ordering.

New tests in ballista-core:

• test_create_grpc_server_incoming_binds_eagerly — connects to the bound address while nothing is serving on it. This is the property the fix depends on, and it fails against a lazily-bound socket.
• test_create_grpc_server_incoming_port_in_use — binding an address twice is an error rather than a panic, so a port conflict still surfaces as a normal startup failure.

Are there any user-facing changes?

No behaviour changes and no API breakage. create_grpc_server_incoming is a new public function in ballista-core, additive alongside create_grpc_server. Note that it must be called from within a Tokio runtime, since the listener registers with the reactor; this is documented on the function.

[andygrove]

@villebro @akshaychitneni could you review?

[avantgardnerio]

Thank you for fixing this!

Nit: the body claims test_create_grpc_server_incoming_port_in_use but I don't see that in the code?

[andygrove]

Good catch, thanks. That test was dropped by a cleanup commit rather than renamed, so it really wasn't there. Restored it in d76c435, and fixed the body which was also still using the old name for the other test.