Skip to main content

WorkerSession

aion_worker::protocol::session

Trait WorkerSession 

Source 
pub trait WorkerSession: Send {
    // Required methods
    fn handshake<'life0, 'life1, 'async_trait>(
        &'life0 mut self,
        config: &'life1 WorkerConfig,
    ) -> Pin<Box<dyn Future<Output = Result<(), WorkerError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn register<'life0, 'life1, 'async_trait>(
        &'life0 mut self,
        activity_types: Vec<String>,
        available_handlers: &'life1 BTreeSet<String>,
    ) -> Pin<Box<dyn Future<Output = Result<(), WorkerError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn receive_tasks(&mut self) -> WorkerTaskStream;
    fn report_result<'life0, 'async_trait>(
        &'life0 mut self,
        workflow_id: WorkflowId,
        activity_id: ActivityId,
        run_id: Option<RunId>,
        completion_token: String,
        result: Payload,
    ) -> Pin<Box<dyn Future<Output = Result<(), WorkerError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn report_failure<'life0, 'async_trait>(
        &'life0 mut self,
        workflow_id: WorkflowId,
        activity_id: ActivityId,
        run_id: Option<RunId>,
        completion_token: String,
        failure: ActivityError,
    ) -> Pin<Box<dyn Future<Output = Result<(), WorkerError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn send_heartbeat<'life0, 'async_trait>(
        &'life0 mut self,
        workflow_id: WorkflowId,
        activity_id: ActivityId,
        progress: Option<Payload>,
    ) -> Pin<Box<dyn Future<Output = Result<(), WorkerError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;

    // Provided methods
    fn register_with_contract<'life0, 'life1, 'async_trait>(
        &'life0 mut self,
        activity_types: Vec<String>,
        activities: Vec<ActivityDescriptor>,
        available_handlers: &'life1 BTreeSet<String>,
    ) -> Pin<Box<dyn Future<Output = Result<(), WorkerError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait { ... }
    fn send_connection_heartbeat<'life0, 'async_trait>(
        &'life0 mut self,
    ) -> Pin<Box<dyn Future<Output = Result<(), WorkerError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait { ... }
    fn answer_liveness_ping<'life0, 'async_trait>(
        &'life0 mut self,
        sequence: u64,
    ) -> Pin<Box<dyn Future<Output = Result<(), WorkerError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait { ... }
    fn heartbeat_window(&self) -> Option<Duration> { ... }
}
Expand description

Transport abstraction for the AW-owned worker protocol.

The current aion-proto worker endpoint is WorkerProtocol::StreamWorker, a single bidirectional gRPC stream. These methods intentionally present the worker conversation as handshake/register/receive/report/heartbeat phases so execution machinery can be tested against fakes and never touches generated stubs directly. If AW changes the wire shape, this trait adapts in this module.

Required Methods§

Source

fn handshake<'life0, 'life1, 'async_trait>( &'life0 mut self, config: &'life1 WorkerConfig, ) -> Pin<Box<dyn Future<Output = Result<(), WorkerError>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Performs the worker handshake for the configured namespace, task queue, and identity.

Maps to transport/channel establishment for AW’s StreamWorker RPC. The wire carries the genuine namespace (correctness boundary) and task_queue (pool selector) as disjoint registration fields; it has no identity field, so identity is retained at this SDK boundary until the wire adds a corresponding shape.

Source

fn register<'life0, 'life1, 'async_trait>( &'life0 mut self, activity_types: Vec<String>, available_handlers: &'life1 BTreeSet<String>, ) -> Pin<Box<dyn Future<Output = Result<(), WorkerError>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Registers activity-type names implemented by this worker.

Maps to opening AW’s StreamWorker RPC with RegisterWorker queued as the mandatory first frame and then awaiting the server’s RegisterAck — the guaranteed first frame on the response stream. Registration succeeds only when the ack arrives; a denial fails the RPC with a gRPC error status (PermissionDenied / Unauthenticated), and an ack that does not arrive within the reconnect policy’s max_backoff is a retryable registration failure. The caller supplies available_handlers so registration can be rejected before serving if any requested name lacks a handler.

Source

fn receive_tasks(&mut self) -> WorkerTaskStream

Opens the receive side of AW’s StreamWorker RPC and yields pushed tasks.

Source

fn report_result<'life0, 'async_trait>( &'life0 mut self, workflow_id: WorkflowId, activity_id: ActivityId, run_id: Option<RunId>, completion_token: String, result: Payload, ) -> Pin<Box<dyn Future<Output = Result<(), WorkerError>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Reports successful activity output and echoes its opaque execution generation via WorkerToServer.result.

Source

fn report_failure<'life0, 'async_trait>( &'life0 mut self, workflow_id: WorkflowId, activity_id: ActivityId, run_id: Option<RunId>, completion_token: String, failure: ActivityError, ) -> Pin<Box<dyn Future<Output = Result<(), WorkerError>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Reports explicit activity failure and echoes its opaque execution generation via WorkerToServer.result.

Source

fn send_heartbeat<'life0, 'async_trait>( &'life0 mut self, workflow_id: WorkflowId, activity_id: ActivityId, progress: Option<Payload>, ) -> Pin<Box<dyn Future<Output = Result<(), WorkerError>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Sends cooperative progress via WorkerToServer.heartbeat.

Provided Methods§

Source

fn register_with_contract<'life0, 'life1, 'async_trait>( &'life0 mut self, activity_types: Vec<String>, activities: Vec<ActivityDescriptor>, available_handlers: &'life1 BTreeSet<String>, ) -> Pin<Box<dyn Future<Output = Result<(), WorkerError>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Registers activity names together with their committed wire schemas.

Session implementations predating contract handshakes remain useful as test and custom transport adapters: the default validates through their legacy registration path. Production transports override this method and carry activities on the wire.

Source

fn send_connection_heartbeat<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<(), WorkerError>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Heartbeat the worker connection while it is idle.

The gRPC transport encodes this on the existing heartbeat path with no task identifiers. Transport fakes may retain the no-op default.

Source

fn answer_liveness_ping<'life0, 'async_trait>( &'life0 mut self, sequence: u64, ) -> Pin<Box<dyn Future<Output = Result<(), WorkerError>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Answer a server WorkerSessionEvent::LivenessPing, echoing sequence verbatim (#197).

This is a RUNTIME obligation and the serve loop discharges it directly; no handler is consulted and no concurrency permit is taken, because the question is about the transport, not about the work.

The default is a no-op for transports that carry no such frame (fakes, and the liminal session, which answers its own ping/pong pair on its own connection). That is not a silent failure: a session whose wire does carry the ping and which does not answer it simply never clears its dispatch probation, which is the honest verdict for a worker the server cannot prove it reaches.

§Errors

Returns WorkerError::Transport when the answer cannot be sent.

Source

fn heartbeat_window(&self) -> Option<Duration>

Server-assigned liveness window from the RegisterAck, once registered.

The serve loop derives its AUTOMATIC liveness-heartbeat cadence from this window (see serve_activity_tasks_until): the server’s heartbeat sweeper expires any worker whose in-flight task goes longer than the window without a heartbeat, so the runtime — not each handler — must keep every in-flight activity beating. None (the default, and the value for fake/unregistered sessions) disables the automatic pump.

Dyn Compatibility§

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors§