aion_server/worker/
grpc_task_delivery.rs

1//! The gRPC arm of the delivery seam: push the task onto the worker's stream.
2//!
3//! This is the transport the dispatcher always used, unchanged in what it does
4//! — the change is that it now *says* what it did, so the caller acts on a type
5//! instead of on the absence of a sender.
6
7use async_trait::async_trait;
8
9use aion_proto::ProtoActivityTask;
10
11use super::delivery_intent::SharedDeliveryIntent;
12use super::registry::{WorkerHandle, WorkerMessage};
13use super::task_delivery::{DeliveryAccepted, TaskDelivery, WorkerTaskDelivery};
14
15/// Delivers by pushing a [`WorkerMessage::ActivityTask`] onto the worker's
16/// registered stream.
17///
18/// # Only two outcomes are reachable here, and that is not an oversight
19///
20/// `mpsc::Sender::send().await` fails **only** when the receiver is gone, i.e.
21/// the worker's stream is closed — which is exactly
22/// [`Undeliverable::WorkerUnreachable`](super::task_delivery::Undeliverable::WorkerUnreachable).
23/// A *full* channel does not error; it applies backpressure and the send waits.
24/// So this transport never produces
25/// [`DeliveryFailed`](super::task_delivery::Undeliverable::DeliveryFailed).
26///
27/// I record that because my own design note previously claimed a full channel
28/// would map to `DeliveryFailed`. It would — under `try_send`. Under the
29/// `send().await` this path has always used, that outcome does not exist, and
30/// switching to `try_send` to manufacture it would turn a worker that is merely
31/// busy into a failed delivery. `DeliveryFailed` exists for the blocking
32/// transport, which really can fail while its worker is alive.
33///
34/// # The intent is asked once, and the backpressure wait is not re-polled
35///
36/// The caller's intent is checked immediately before the push. If the caller
37/// loses its claim *during* the backpressure wait inside `send().await`, this
38/// arm does not notice: there is no poll boundary to re-ask at, and reaching for
39/// one would mean abandoning a send mid-flight.
40///
41/// That is safe, and by two independent mechanisms rather than one:
42///
43/// 1. **The result cannot be recorded.** The completion token is one per pass,
44///    minted by the caller and revoked once when the pass places nothing, so a
45///    task that lands after its pass withdrew carries an authorization the
46///    fences no longer honour.
47/// 2. **The external effect cannot be duplicated.** The task's
48///    `idempotency_key` is derived from `(workflow_id, run_id, activity_id)`
49///    alone — attempts and execution generations deliberately do not participate
50///    (`aion-proto/src/worker.rs:143-146`) — so a redelivery by whichever pass
51///    re-claimed the row carries the *same* key and is deduplicated at the
52///    worker.
53///
54/// The token covers the recording, the idempotency key covers the effect. If
55/// either ever stops holding, that is a finding about the fences or the key —
56/// not a reason to start polling a send.
57#[derive(Debug, Clone, Copy, Default)]
58pub struct GrpcTaskDelivery;
59
60#[async_trait]
61impl WorkerTaskDelivery for GrpcTaskDelivery {
62    async fn deliver(
63        &self,
64        worker: &WorkerHandle,
65        task: &ProtoActivityTask,
66        intent: &SharedDeliveryIntent,
67        accepted: &dyn DeliveryAccepted,
68    ) -> TaskDelivery {
69        let Some(sender) = worker.sender() else {
70            // A worker with no stream sender is not a gRPC-delivered worker, so
71            // this transport genuinely cannot reach it. It is NOT the worker's
72            // fault and NOT evidence the worker is gone — which is why the
73            // dispatcher no longer routes here by default. Reaching this arm at
74            // all now means a caller chose the wrong transport for the worker it
75            // selected, so it reports a delivery failure and leaves the
76            // registration standing rather than destroying a healthy worker.
77            return TaskDelivery::failed(
78                "worker is not delivered over gRPC; the gRPC transport cannot reach it",
79            );
80        };
81        // The one wait this transport has is the backpressure wait inside
82        // `send().await` on a full channel. Asking the caller's intent before
83        // entering it is the same question the blocking transport re-asks
84        // throughout its own wait: a pass that has already lost its claim must
85        // not enqueue work it no longer owns.
86        if !intent.still_wanted() {
87            return TaskDelivery::failed("caller abandoned the delivery before it was enqueued");
88        }
89        match sender
90            .send(WorkerMessage::ActivityTask(Box::new(task.clone())))
91            .await
92        {
93            Ok(()) => {
94                // The stream took the frame: the worker holds the attempt from
95                // here, and the lease is recorded before anything else is
96                // said about it (WA-010 R3).
97                accepted.accepted().await;
98                TaskDelivery::Delivered
99            }
100            Err(error) => TaskDelivery::unreachable(format!("worker stream closed: {error}")),
101        }
102    }
103}
aion_server/worker/grpc_task_delivery.rs

aion_server/worker/
grpc_task_delivery.rs
