aion-server 0.27.1

//! The gRPC arm of the delivery seam: push the task onto the worker's stream.
//!
//! This is the transport the dispatcher always used, unchanged in what it does
//! — the change is that it now *says* what it did, so the caller acts on a type
//! instead of on the absence of a sender.

use async_trait::async_trait;

use aion_proto::ProtoActivityTask;

use super::delivery_intent::SharedDeliveryIntent;
use super::registry::{WorkerHandle, WorkerMessage};
use super::task_delivery::{TaskDelivery, WorkerTaskDelivery};

/// Delivers by pushing a [`WorkerMessage::ActivityTask`] onto the worker's
/// registered stream.
///
/// # Only two outcomes are reachable here, and that is not an oversight
///
/// `mpsc::Sender::send().await` fails **only** when the receiver is gone, i.e.
/// the worker's stream is closed — which is exactly
/// [`Undeliverable::WorkerUnreachable`](super::task_delivery::Undeliverable::WorkerUnreachable).
/// A *full* channel does not error; it applies backpressure and the send waits.
/// So this transport never produces
/// [`DeliveryFailed`](super::task_delivery::Undeliverable::DeliveryFailed).
///
/// I record that because my own design note previously claimed a full channel
/// would map to `DeliveryFailed`. It would — under `try_send`. Under the
/// `send().await` this path has always used, that outcome does not exist, and
/// switching to `try_send` to manufacture it would turn a worker that is merely
/// busy into a failed delivery. `DeliveryFailed` exists for the blocking
/// transport, which really can fail while its worker is alive.
///
/// # The intent is asked once, and the backpressure wait is not re-polled
///
/// The caller's intent is checked immediately before the push. If the caller
/// loses its claim *during* the backpressure wait inside `send().await`, this
/// arm does not notice: there is no poll boundary to re-ask at, and reaching for
/// one would mean abandoning a send mid-flight.
///
/// That is safe, and by two independent mechanisms rather than one:
///
/// 1. **The result cannot be recorded.** The completion token is one per pass,
///    minted by the caller and revoked once when the pass places nothing, so a
///    task that lands after its pass withdrew carries an authorization the
///    fences no longer honour.
/// 2. **The external effect cannot be duplicated.** The task's
///    `idempotency_key` is derived from `(workflow_id, run_id, activity_id)`
///    alone — attempts and execution generations deliberately do not participate
///    (`aion-proto/src/worker.rs:143-146`) — so a redelivery by whichever pass
///    re-claimed the row carries the *same* key and is deduplicated at the
///    worker.
///
/// The token covers the recording, the idempotency key covers the effect. If
/// either ever stops holding, that is a finding about the fences or the key —
/// not a reason to start polling a send.
#[derive(Debug, Clone, Copy, Default)]
pub struct GrpcTaskDelivery;

#[async_trait]
impl WorkerTaskDelivery for GrpcTaskDelivery {
    async fn deliver(
        &self,
        worker: &WorkerHandle,
        task: &ProtoActivityTask,
        intent: &SharedDeliveryIntent,
    ) -> TaskDelivery {
        let Some(sender) = worker.sender() else {
            // A worker with no stream sender is not a gRPC-delivered worker, so
            // this transport genuinely cannot reach it. It is NOT the worker's
            // fault and NOT evidence the worker is gone — which is why the
            // dispatcher no longer routes here by default. Reaching this arm at
            // all now means a caller chose the wrong transport for the worker it
            // selected, so it reports a delivery failure and leaves the
            // registration standing rather than destroying a healthy worker.
            return TaskDelivery::failed(
                "worker is not delivered over gRPC; the gRPC transport cannot reach it",
            );
        };
        // The one wait this transport has is the backpressure wait inside
        // `send().await` on a full channel. Asking the caller's intent before
        // entering it is the same question the blocking transport re-asks
        // throughout its own wait: a pass that has already lost its claim must
        // not enqueue work it no longer owns.
        if !intent.still_wanted() {
            return TaskDelivery::failed("caller abandoned the delivery before it was enqueued");
        }
        match sender
            .send(WorkerMessage::ActivityTask(Box::new(task.clone())))
            .await
        {
            Ok(()) => TaskDelivery::Delivered,
            Err(error) => TaskDelivery::unreachable(format!("worker stream closed: {error}")),
        }
    }
}