[geoHeil]

Closes #826.

What

An optional WorkerCallAuthorizer on Worker, installed with Worker::with_call_authorizer. The gRPC service hands it every CoordinatorChannel and ExecuteTask call as a WorkerCall - the request headers plus the decoded TaskKey - after the first message is decoded and before task_data_entries is touched. An Err(Status) is returned to the caller as is. With no authorizer set, nothing changes.

#[async_trait]
pub trait WorkerCallAuthorizer: Send + Sync {
    async fn authorize(&self, call: WorkerCall<'_>) -> Result<(), Status>;
}

#[non_exhaustive]
pub enum WorkerCall<'a> {
    SetPlan { headers: &'a HeaderMap, task_key: &'a TaskKey },
    ExecuteTask { headers: &'a HeaderMap, task_key: &'a TaskKey },
}

Both are exported from datafusion_distributed::grpc, behind the grpc feature. WorkerCall is #[non_exhaustive] so a later RPC can be added without breaking an authorizer, which then has to decide what to do with a call it does not know.

Notes

• In-process calls (LocalWorkerContext) cross no gRPC and are not asked. That seemed right: they come from the worker's own process.
• coordinator_channel now takes the headers it already built from metadata.into_headers() before the authorizer runs, instead of building them later; execute_task splits the request with into_parts() to keep the headers.

Tests

src/protocol/grpc/call_authorizer.rs drives a Worker through its own WorkerServiceServer with an authorizer that admits a call only with an x-admitted header:

• a_refused_call_never_reaches_the_task_entries - a refused SetPlan and a refused ExecuteTask both return the authorizer's status, and the worker has no task running afterwards.
• an_admitted_call_is_served_after_the_authorizer_saw_its_task_key - an admitted pair is served, and the authorizer saw both calls with their task key.

Both fail when the two self.authorize(..) calls are replaced with a no-op.

We are running this patch against 4.0.0 now, so a release carrying it would let us go back to the published crate.

🤖 Generated with Claude Code