road-runner-common 0.22.0

//! Generic follower for cex-policy's governance stream (`cex_policy.events`,
//! externally-tagged `DomainEvent`s). Every governed action (FeeChange,
//! AuthzChange, AuthRiskRuleset, ...) reacts to the same `QuorumReached`
//! shape; implement [`GovernanceHandler`] once per action and this consumer
//! handles the reconnect loop, event parsing, and commit-after-apply.

use std::sync::Arc;

use async_trait::async_trait;
use rdkafka::config::ClientConfig;
use rdkafka::consumer::{CommitMode, Consumer, StreamConsumer};
use rdkafka::message::Message;

/// Applies (or releases) a single governed action once cex-policy resolves its
/// proposal. Implement this for each governed action a service owns; the
/// `action` the handler cares about is whatever it decides to apply in
/// [`Self::apply_approved_proposal`] — a service can either wire up one
/// handler per action (multiple consumers) or dispatch internally on the
/// proposal's resolved `action` field, matching cex-metadata's FeeChange today.
#[async_trait]
pub trait GovernanceHandler: Send + Sync {
    /// A `QuorumReached { status: Approved }` fired for `proposal_id`. Fetch the
    /// proposal (payload included) and apply the mutation it governs. Must be
    /// idempotent — Kafka delivery is at-least-once.
    ///
    /// The return value decides whether the offset advances, so it is a request,
    /// not a report:
    ///
    /// - `Ok(())` — done, **or** permanently undone. Return this for anything a
    ///   second delivery would resolve identically: a proposal for another action,
    ///   a payload missing a field it will never grow, work that needs an operator
    ///   rather than a retry. The event is consumed.
    /// - `Err(_)` — ask me again. The offset is held and the consumer reconnects,
    ///   so this message is redelivered in ~10s. Reserve it for transient causes
    ///   (cex-policy unreachable, the database down), because a handler that keeps
    ///   returning `Err` stalls every later governance decision behind it.
    async fn apply_approved_proposal(&self, proposal_id: &str) -> anyhow::Result<()>;

    /// A `QuorumReached { status: Rejected | Expired }` fired for `proposal_id`.
    /// Default no-op — only override if the proposer side tracks a
    /// `pending_proposal_id` that needs releasing so a new proposal can be
    /// submitted (cex-metadata's FeeChange does this; most handlers don't).
    async fn release_rejected_proposal(&self, _proposal_id: &str) -> anyhow::Result<()> {
        Ok(())
    }
}

pub struct GovernanceConsumerConfig {
    pub brokers: String,
    /// e.g. `cex_policy.events`.
    pub topic: String,
    /// Consumer group id. Keep this stable across deploys/rewrites — changing
    /// it replays the whole topic from `auto.offset.reset`.
    pub group_id: String,
}

/// Runs forever, reconnecting on any consumer error. Intended to be
/// `tokio::spawn`ed; hold the returned `JoinHandle` if you want to await/abort it.
pub fn spawn_governance_consumer(
    cfg: GovernanceConsumerConfig,
    handler: Arc<dyn GovernanceHandler>,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        loop {
            match build_consumer(&cfg) {
                Ok(consumer) => {
                    #[cfg(feature = "observability")]
                    tracing::info!(topic = %cfg.topic, group = %cfg.group_id, "governance consumer connected");
                    run(&consumer, &handler).await;
                }
                Err(e) => {
                    #[cfg(feature = "observability")]
                    tracing::warn!(error = %e, "governance consumer connect failed; retrying");
                    let _ = e;
                }
            }
            tokio::time::sleep(std::time::Duration::from_secs(10)).await;
        }
    })
}

async fn run(consumer: &StreamConsumer, handler: &Arc<dyn GovernanceHandler>) {
    loop {
        match consumer.recv().await {
            Ok(message) => {
                let applied = match message.payload() {
                    Some(payload) => handle(payload, handler).await,
                    // Nothing to apply, so nothing to retry.
                    None => true,
                };
                if !applied {
                    // Deliberately leave the offset where it is and drop the
                    // connection, so this message is redelivered.
                    //
                    // The alternative — commit and move on — is what this loop used
                    // to do, and it silently discarded governance decisions the
                    // handler had said it could not apply. Every handler in the
                    // platform was already written against the contract restored
                    // here: `Ok` for "done, or permanently undone", `Err` for
                    // "ask me again". The vault follower's comments state it
                    // outright, and it returns `Ok(())` precisely for the cases
                    // that will never succeed on a retry (a payload with no
                    // subject, a claim that needs an operator) so they cannot wedge
                    // the stream.
                    //
                    // The cost is that a handler which keeps failing stalls the
                    // partition rather than skipping past. That is the right way
                    // round for this stream: a stall stops proposals from applying,
                    // which somebody notices, while a skip loses an approved
                    // decision in a way nothing reports. Reconnect is on a 10s
                    // cycle, so a cex-policy outage resolves itself.
                    break;
                }
                if let Err(e) = consumer.commit_message(&message, CommitMode::Async) {
                    #[cfg(feature = "observability")]
                    tracing::warn!(error = %e, "governance consumer commit failed");
                    let _ = e;
                }
            }
            Err(e) => {
                #[cfg(feature = "observability")]
                tracing::warn!(error = %e, "governance consumer recv error; reconnecting");
                let _ = e;
                break;
            }
        }
    }
}

/// cex-policy's `DomainEvent` is externally tagged:
/// `{ "QuorumReached": { proposal_id, status, occurred_at } }`.
///
/// Returns whether the offset may advance. A malformed event, an event of another
/// kind, and a handler that returned `Ok` are all "advance": none of them will read
/// differently on a second delivery. Only a handler asking to be retried holds the
/// offset back.
async fn handle(payload: &[u8], handler: &Arc<dyn GovernanceHandler>) -> bool {
    let event: serde_json::Value = match serde_json::from_slice(payload) {
        Ok(event) => event,
        Err(e) => {
            #[cfg(feature = "observability")]
            tracing::warn!(error = %e, "unparseable policy event");
            let _ = e;
            return true;
        }
    };
    let Some(quorum) = event.get("QuorumReached").and_then(|v| v.as_object()) else {
        return true;
    };
    let Some(proposal_id) = quorum.get("proposal_id").and_then(|v| v.as_str()) else {
        return true;
    };
    let status = quorum.get("status").and_then(|v| v.as_str());

    let result = match status {
        Some("Approved") => handler.apply_approved_proposal(proposal_id).await,
        Some("Rejected") | Some("Expired") => {
            handler.release_rejected_proposal(proposal_id).await
        }
        _ => Ok(()),
    };
    match result {
        Ok(()) => true,
        Err(e) => {
            // Error, not warn: an approved decision that has not been applied is a
            // governance outage, and it stays one until this stops appearing.
            #[cfg(feature = "observability")]
            tracing::error!(
                proposal_id, ?status, error = %e,
                "applying QuorumReached failed; holding the offset and retrying"
            );
            let _ = (e, status);
            false
        }
    }
}

fn build_consumer(cfg: &GovernanceConsumerConfig) -> anyhow::Result<StreamConsumer> {
    let settings = {
        let mut s = crate::kafka::KafkaSettings::from_env();
        s.brokers = cfg.brokers.clone();
        s
    };
    let mut client = ClientConfig::new();
    for (k, v) in settings.security_settings() {
        client.set(k, v);
    }
    let consumer: StreamConsumer = client
        .set("group.id", &cfg.group_id)
        .set("enable.auto.commit", "false")
        .set("auto.offset.reset", "earliest")
        .set("isolation.level", "read_committed")
        .create()?;
    consumer.subscribe(&[cfg.topic.as_str()])?;
    Ok(consumer)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};

    /// The offset contract is the whole safety property of this loop, and it was
    /// wrong once in the direction that loses data: the commit used to happen
    /// regardless of what the handler returned, so a handler saying "I could not
    /// apply this" had its approved governance decision discarded. Every handler in
    /// the platform is written against the contract these tests pin.
    #[derive(Default)]
    struct FakeHandler {
        applied: AtomicUsize,
        released: AtomicUsize,
        fail_apply: bool,
    }

    #[async_trait]
    impl GovernanceHandler for FakeHandler {
        async fn apply_approved_proposal(&self, _proposal_id: &str) -> anyhow::Result<()> {
            self.applied.fetch_add(1, Ordering::SeqCst);
            if self.fail_apply {
                anyhow::bail!("cex-policy unreachable");
            }
            Ok(())
        }

        async fn release_rejected_proposal(&self, _proposal_id: &str) -> anyhow::Result<()> {
            self.released.fetch_add(1, Ordering::SeqCst);
            Ok(())
        }
    }

    fn quorum_event(status: &str) -> Vec<u8> {
        serde_json::to_vec(&serde_json::json!({
            "QuorumReached": {
                "proposal_id": "11111111-1111-4111-8111-111111111111",
                "status": status,
                "occurred_at": "2026-08-08T00:00:00Z"
            }
        }))
        .unwrap()
    }

    #[tokio::test]
    async fn an_applied_proposal_advances_the_offset() {
        let handler: Arc<dyn GovernanceHandler> = Arc::new(FakeHandler::default());
        assert!(handle(&quorum_event("Approved"), &handler).await);
    }

    #[tokio::test]
    async fn a_handler_asking_to_retry_holds_the_offset() {
        let handler: Arc<dyn GovernanceHandler> =
            Arc::new(FakeHandler { fail_apply: true, ..Default::default() });
        assert!(
            !handle(&quorum_event("Approved"), &handler).await,
            "an approved decision the handler could not apply must be redelivered, not discarded"
        );
    }

    #[tokio::test]
    async fn rejected_and_expired_go_to_the_release_path() {
        for status in ["Rejected", "Expired"] {
            let fake = Arc::new(FakeHandler::default());
            let handler: Arc<dyn GovernanceHandler> = fake.clone();
            assert!(handle(&quorum_event(status), &handler).await);
            assert_eq!(fake.released.load(Ordering::SeqCst), 1, "{status}");
            assert_eq!(fake.applied.load(Ordering::SeqCst), 0, "{status}");
        }
    }

    /// None of these will read differently on a second delivery, so holding the
    /// offset for them would wedge the stream on a message nothing can consume.
    #[tokio::test]
    async fn unconsumable_events_advance_rather_than_wedge_the_stream() {
        let fake = Arc::new(FakeHandler::default());
        let handler: Arc<dyn GovernanceHandler> = fake.clone();

        assert!(handle(b"not json at all", &handler).await);
        assert!(handle(br#"{"BootstrapSealed":{"occurred_at":"2026-08-08T00:00:00Z"}}"#, &handler).await);
        assert!(handle(br#"{"QuorumReached":{"status":"Approved"}}"#, &handler).await);
        assert!(handle(&quorum_event("Pending"), &handler).await);

        assert_eq!(fake.applied.load(Ordering::SeqCst), 0);
        assert_eq!(fake.released.load(Ordering::SeqCst), 0);
    }
}