Add a long-poll `WaitJob` API so clients can block until a job terminates

[LinZhihao-723]

Request

Motivation

Today a client can learn that a job has terminated only by polling GetJobState on JobOrchestrationService. SpiderClient has no wait helper, so every caller writes its own loop; the e2e driver, for example, polls every 10 ms (poll_until_terminal in tests/huntsman/e2e/src/test_driver.rs). Polling forces a trade-off:

• Polling frequently adds per-call gRPC overhead and network traffic, and every poll takes the job cache and job control block locks on the storage server.
• Polling infrequently delays the caller's notification by up to one poll interval.

We want an API that, from the caller's perspective, blocks until the job terminates and returns as soon as it does.

Prior art

This problem is well solved in production systems. The dominant pattern is a bounded long-poll: a unary RPC that the server holds until the state changes or a server-side cap elapses. When the cap elapses, the server returns a normal "not done yet" response, and the client library re-issues the call.

• Temporal: GetWorkflowExecutionHistory with wait_new_event and the CLOSE_EVENT filter. The server caps each call at 20 s, and the SDK loops.
• Google AIP-151: Operations.WaitOperation, which is best-effort and may return before the timeout with the operation still running.
• Consul blocking queries (?index=&wait=) and AWS SQS long polling (WaitTimeSeconds, max 20 s).

Server streaming (etcd and Kubernetes watches) is used for change feeds over many objects. For a single job's terminal state, a unary long-poll is simpler.

Challenges

• Long-lived calls vs. timeouts. gRPC has no default deadline, but proxies and load balancers do (Envoy route timeout 15 s, GCP LB 30 s, nginx and AWS ALB 60 s). tonic reports a server-enforced grpc-timeout expiry as CANCELLED. Unbounded calls also hold server resources for clients that are gone, and stall graceful shutdown.
• Lost wakeups. A waiter that checks the state and then subscribes can miss a transition that happens in between.
• Jobs that never terminate on their own. A job that is registered but never started would park its waiters forever.
• Dead peers. An idle waiting client never notices a storage host that died silently (host crash, network partition). This is a system-level gap tracked in gRPC connections never detect a silently dead peer (host crash, network partition, frozen process) #482, which this feature depends on.
• Connection sharing. Waits parked on the same connections as short calls are harmless with tonic's default server settings, which impose no HTTP/2 stream limit. With a per-connection stream limit (e.g., 128 in nginx or AWS ALB), however, short calls such as SubmitJob and CancelJob stall once waits fill the slots. A benchmark using Spider's ConnectionPool (8 connections, 1 KiB unary calls, loopback):

Server stream limit	Parked waits	Waits share the short-call pool	Waits use a dedicated 8-connection pool
None (tonic default)	10,000	Unaffected (p50 ~80 µs)	Unaffected (p50 ~85 µs)
128 per connection	1,000	Unaffected	Unaffected
128 per connection	1,100 or 2,000	Every short call timed out	Unaffected (p50 ~110 µs)

Possible implementation

API contract

service JobOrchestrationService {
  // ...
  rpc WaitJob(JobIdRequest) returns (JobStatusResponse);
}

message JobStatusResponse {
  JobState state = 1;
  optional string error_message = 2;
}

• The call returns as soon as the job reaches a terminal state (SUCCEEDED, FAILED, or CANCELLED), with error_message set when the job failed. Outputs are still fetched with GetJobOutputs.
• If the job doesn't terminate within 60 s (a server-side constant), the call returns OK with the current non-terminal state. This is a normal outcome, not an error.
• The call returns immediately if the job has already terminated, including jobs already evicted from the cache (read from the DB).
• A job that hasn't been started (READY) is rejected with FAILED_PRECONDITION. An unknown job is rejected with NOT_FOUND.
• The server must not return a non-terminal state without having waited, except on shutdown, since clients re-issue immediately.

Server (spider-storage)

• Give each job a tokio::sync::watch::Sender<JobStatus> stored on the job control block, outside the lock around the job's execution state, so that subscribing never contends with state transitions.
• Replace the direct job.state = ... assignments in components/spider-storage/src/cache/job.rs with a single setter that updates the state and publishes it. Publishing happens under the job's write lock, after the DB write (the existing order).
  • Publish every transition, not only terminal ones, so that a future progress stream can reuse the same sender.
  • For FAILED, publish the error message, which fail_task_instance already has.
• In the handler:
  • Subscribe first, then check the current value. A watch receiver sees the current value on subscription, so no transition is missed.
  • Then wait for a terminal state, racing a 60 s deadline and the service's cancellation_token. The deadline covers the whole handler: cache lookup, locks, and the DB fallback.
  • On deadline or shutdown, return the latest published value.
  • If the sender is dropped, fall back to the DB.
• A client disconnect drops the handler future, which also drops the receiver.
• Keep per-call logging at debug, since every waiting job re-issues about once a minute.

Client (spider-client)

• Add SpiderClient::wait_for_job(job_id), which loops on WaitJob until the job reaches a terminal state and returns its status.
  • The client sets no deadline, so the user-facing wait is unbounded. Dropping the future cancels the wait.
  • A non-terminal OK response is re-issued immediately.
  • A retryable transport error (UNAVAILABLE, or UNKNOWN caused by a transport error) is retried with the existing backoff (call_with_retry).
  • FAILED_PRECONDITION maps to a new ClientError::JobNotStarted.
• Send long-polls over a dedicated ConnectionPool, separate from the job orchestration and resource group pools. Its size is fixed at construction through a SpiderClientBuilder option, with a small default (e.g., 2). Without a server stream limit, a couple of connections carry thousands of waits, and each client already opens 16 connections. Dynamic sizing can come later.
• A user-supplied Endpoint::timeout below 60 s would cut off every wait. Either override it for the wait pool or document the constraint.

Future: opt-in progress streaming

Add rpc WatchJob(...) returns (stream JobStatusResponse), driven by the same per-job sender. Each update is a snapshot of the latest state, and a slow reader simply skips intermediate updates, so no buffering is needed. A version field would let a reconnecting client resume.

Open questions

• Retry budget while storage is unreachable. One option is to reset the RetryConfig budget after every successful response, so a short storage restart is transparent while a long outage surfaces as an error. The other is to retry forever.
• Shorter client deadlines. Should the server honor a client grpc-timeout shorter than 60 s (returning OK about 1 s before it expires) for third-party callers, or just document it?
• Default size of the long-poll pool.

Testing

• A job terminating between subscribe and check.
• A job already evicted from the cache.
• Rejecting a READY job.
• Returning early on shutdown.
• Re-issuing after the 60 s cut, using an overridable constant or paused tokio time.
• Switching the e2e driver's poll_until_terminal to wait_for_job.

Related

• gRPC connections never detect a silently dead peer (host crash, network partition, frozen process) #482: dead-peer detection (keepalive and connect timeout). An unbounded client wait is safe against a storage host that dies silently only once this lands.
• The storage server should keep HTTP/2 max_concurrent_streams unset (the tonic default), so parked waits never consume stream slots on connections that carry short calls.