aion-core 0.29.0

//! Durable attribution of one activity attempt to the worker that leased it.
//!
//! [`WorkerAttribution`] is what an [`Event::ActivityLeased`](crate::Event::ActivityLeased)
//! carries. Every field is a NAME an operator can act on after the recording
//! server is gone: the identity string the worker registered under, the task
//! queue it served, the node it declared, the deployment and instance it
//! announced, and the transport it was reached over. None of them is the
//! server-process `WorkerId` — that is a per-process registry counter, minted
//! at registration and forgotten at shutdown, so the same number names a
//! different worker after every restart and nothing at all in an exported
//! history. Attribution that dies with the process would be a lie in a durable
//! log, so the counter is deliberately absent from this type.

use serde::{Deserialize, Serialize};

use crate::WorkerTransport;

/// Durable names for the worker that leased one activity attempt.
///
/// Carried by [`Event::ActivityLeased`](crate::Event::ActivityLeased). Every
/// field is copied from the worker's registration frame at the moment the
/// selected worker accepted the push, so the record describes the worker AS IT
/// REGISTERED — a later re-registration under other names records a different
/// attribution on its own lease, never rewrites this one.
///
/// There is no `WorkerId` here and there must never be one: see the module
/// documentation for why a registry counter cannot attribute anything durably.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
pub struct WorkerAttribution {
    /// The identity string the worker registered under — the operator-facing
    /// name of the process (a hostname, a pod name, an SDK's default), exactly
    /// as the registration frame carried it.
    pub identity: String,
    /// Task queue the worker served this attempt from.
    pub task_queue: String,
    /// Node the worker declared at registration, when it declared one.
    pub node: Option<String>,
    /// Deployment name the worker announced at registration, when it did.
    pub deployment: Option<String>,
    /// Operator/launcher-assigned instance identifier announced at
    /// registration, when the worker carried one.
    pub instance_id: Option<String>,
    /// Transport the worker was reached over — the same [`WorkerTransport`]
    /// the cluster stream and the liveness log name, flattened so the record
    /// reads `"transport": "Grpc"` beside the other names.
    #[serde(flatten)]
    #[ts(flatten)]
    pub transport: WorkerTransport,
}

#[cfg(test)]
mod tests {
    use super::WorkerAttribution;
    use crate::WorkerTransport;

    fn attribution() -> WorkerAttribution {
        WorkerAttribution {
            identity: String::from("worker-a@host-1"),
            task_queue: String::from("billing"),
            node: Some(String::from("n1")),
            deployment: Some(String::from("billing-workers")),
            instance_id: Some(String::from("i-42")),
            transport: WorkerTransport::Grpc,
        }
    }

    #[test]
    fn attribution_round_trips_through_json() -> Result<(), serde_json::Error> {
        let value = attribution();
        let encoded = serde_json::to_string(&value)?;
        let decoded: WorkerAttribution = serde_json::from_str(&encoded)?;
        assert_eq!(decoded, value);
        Ok(())
    }

    #[test]
    fn transport_is_one_flat_key_spelled_like_the_cluster_stream() -> Result<(), serde_json::Error>
    {
        for transport in [WorkerTransport::Grpc, WorkerTransport::Liminal] {
            let encoded = serde_json::to_value(WorkerAttribution {
                transport,
                ..attribution()
            })?;
            // The flattened tag is the ONLY transport key, and its value is the
            // same word `WorkerTransport` puts on a cluster event.
            assert_eq!(
                encoded["transport"],
                serde_json::to_value(transport)?["transport"]
            );
            assert!(encoded.get("Grpc").is_none() && encoded.get("Liminal").is_none());
        }
        Ok(())
    }

    #[test]
    fn absent_optional_names_encode_as_null_and_decode_back() -> Result<(), serde_json::Error> {
        let value = WorkerAttribution {
            node: None,
            deployment: None,
            instance_id: None,
            transport: WorkerTransport::Liminal,
            ..attribution()
        };
        let encoded = serde_json::to_value(&value)?;
        assert!(encoded["node"].is_null());
        assert!(encoded["deployment"].is_null());
        assert!(encoded["instance_id"].is_null());
        assert_eq!(encoded["transport"], "Liminal");
        let decoded: WorkerAttribution = serde_json::from_value(encoded)?;
        assert_eq!(decoded, value);
        Ok(())
    }
}