Rust — Artigo #43

Sistemas Distribuídos — Consenso, Mensageria e Tolerância a Falhas

Por Prof. Dr. Marcelo Fontes | Série: Dominando Rust em 1 Ano

Sistemas distribuídos são onde a complexidade do software atinge seu pico. Falhas de rede são inevitáveis. Clocks não sincronizam perfeitamente. Mensagens chegam fora de ordem ou não chegam. Nós travam no meio de uma operação. O teorema CAP garante matematicamente que você não pode ter consistência, disponibilidade e tolerância a partições ao mesmo tempo — você precisa escolher dois.

Rust é excepcionalmente adequado para sistemas distribuídos: o sistema de tipos força você a modelar explicitamente os casos de falha, o modelo de ownership torna impossível compartilhar estado mutável acidentalmente entre threads, e a performance previsível é essencial quando latência importa.

Fundamentos: o que torna sistemas distribuídos difíceis

// O problema fundamental: não sabemos se a operação falhou
// ou se a resposta simplesmente não chegou ainda

async fn operacao_remota() -> Result<String, Erro> {
    // Do ponto de vista do chamador, estas situações são INDISTINGUÍVEIS
    // após um timeout:

    // 1. O servidor recebeu, processou, mas a resposta se perdeu
    // 2. O servidor recebeu mas ainda está processando
    // 3. O servidor nunca recebeu a mensagem
    // 4. O servidor travou após receber
    todo!()
}

// Consequência: operações distribuídas precisam ser idempotentes
// Executar duas vezes deve ter o mesmo efeito que executar uma vez

// NÃO idempotente:
// INSERT INTO pedidos VALUES (...)  → duplica o pedido

// Idempotente:
// INSERT INTO pedidos VALUES (...) ON CONFLICT (id) DO NOTHING
// ou: verificar antes de inserir

// A solução: IDs de idempotência
#[derive(Debug, Clone)]
struct OperacaoIdempotente {
    id_idempotencia: uuid::Uuid,  // gerado pelo cliente, único por tentativa
    payload: Vec<u8>,
}

Message passing com canais — fundamento assíncrono

use tokio::sync::{mpsc, broadcast, oneshot};
use std::collections::HashMap;
use std::time::Duration;

// Padrão: mensagem + canal de resposta (request-reply)
#[derive(Debug)]
enum MensagemAtor {
    Incrementar { chave: String },
    Obter {
        chave: String,
        resposta: oneshot::Sender<Option<i64>>,
    },
    Resetar { chave: String },
    Snapshot {
        resposta: oneshot::Sender<HashMap<String, i64>>,
    },
}

// Ator: thread/task com estado privado, comunicação apenas por mensagens
struct AtorContadores {
    contadores: HashMap<String, i64>,
    receptor: mpsc::Receiver<MensagemAtor>,
}

impl AtorContadores {
    fn novo() -> (Self, mpsc::Sender<MensagemAtor>) {
        let (tx, rx) = mpsc::channel(1024);
        (
            AtorContadores {
                contadores: HashMap::new(),
                receptor: rx,
            },
            tx,
        )
    }

    async fn executar(mut self) {
        while let Some(msg) = self.receptor.recv().await {
            match msg {
                MensagemAtor::Incrementar { chave } => {
                    *self.contadores.entry(chave).or_insert(0) += 1;
                }

                MensagemAtor::Obter { chave, resposta } => {
                    let valor = self.contadores.get(&chave).copied();
                    // Ignora erro se o receptor já foi descartado
                    let _ = resposta.send(valor);
                }

                MensagemAtor::Resetar { chave } => {
                    self.contadores.remove(&chave);
                }

                MensagemAtor::Snapshot { resposta } => {
                    let _ = resposta.send(self.contadores.clone());
                }
            }
        }
    }
}

// Handle: interface pública para o ator
#[derive(Clone)]
struct HandleContadores {
    tx: mpsc::Sender<MensagemAtor>,
}

impl HandleContadores {
    async fn incrementar(&self, chave: &str) -> Result<(), String> {
        self.tx.send(MensagemAtor::Incrementar {
            chave: chave.to_string(),
        })
        .await
        .map_err(|e| e.to_string())
    }

    async fn obter(&self, chave: &str) -> Result<Option<i64>, String> {
        let (tx, rx) = oneshot::channel();
        self.tx.send(MensagemAtor::Obter {
            chave: chave.to_string(),
            resposta: tx,
        })
        .await
        .map_err(|e| e.to_string())?;

        rx.await.map_err(|e| e.to_string())
    }

    async fn snapshot(&self) -> Result<HashMap<String, i64>, String> {
        let (tx, rx) = oneshot::channel();
        self.tx.send(MensagemAtor::Snapshot { resposta: tx })
            .await
            .map_err(|e| e.to_string())?;
        rx.await.map_err(|e| e.to_string())
    }
}

async fn demo_ator() {
    let (ator, tx) = AtorContadores::novo();
    let handle = HandleContadores { tx };

    // Inicia o ator em background
    tokio::spawn(ator.executar());

    // Múltiplos clientes concorrentes — sem locks, sem condições de corrida
    let handles_clone: Vec<HandleContadores> = (0..10)
        .map(|_| handle.clone())
        .collect();

    let tarefas: Vec<_> = handles_clone.into_iter().map(|h| {
        tokio::spawn(async move {
            for _ in 0..100 {
                h.incrementar("visitas").await.unwrap();
            }
        })
    }).collect();

    for t in tarefas {
        t.await.unwrap();
    }

    let visitas = handle.obter("visitas").await.unwrap();
    println!("Visitas totais: {:?}", visitas); // 1000 — sem race condition

    let snap = handle.snapshot().await.unwrap();
    println!("Snapshot: {:?}", snap);
}

#[tokio::main]
async fn main() {
    demo_ator().await;
}

Circuit Breaker — padrão de resiliência

use std::sync::Arc;
use tokio::sync::Mutex;
use std::time::{Duration, Instant};

#[derive(Debug, Clone, PartialEq)]
enum EstadoCircuito {
    Fechado,         // Normal — deixa requisições passar
    Aberto,          // Falhas demais — bloqueia requisições
    MeioAberto,      // Testando recuperação — deixa uma passar
}

#[derive(Debug)]
struct CircuitBreaker {
    estado: EstadoCircuito,
    falhas_consecutivas: u32,
    limite_falhas: u32,
    ultimo_teste: Option<Instant>,
    timeout_recuperacao: Duration,
    total_requisicoes: u64,
    total_falhas: u64,
    total_rejeitadas: u64,
}

impl CircuitBreaker {
    fn novo(limite_falhas: u32, timeout_recuperacao: Duration) -> Self {
        CircuitBreaker {
            estado: EstadoCircuito::Fechado,
            falhas_consecutivas: 0,
            limite_falhas,
            ultimo_teste: None,
            timeout_recuperacao,
            total_requisicoes: 0,
            total_falhas: 0,
            total_rejeitadas: 0,
        }
    }

    fn pode_executar(&mut self) -> bool {
        match self.estado {
            EstadoCircuito::Fechado => true,

            EstadoCircuito::Aberto => {
                // Verifica se está na hora de testar recuperação
                if let Some(ultimo) = self.ultimo_teste {
                    if ultimo.elapsed() >= self.timeout_recuperacao {
                        self.estado = EstadoCircuito::MeioAberto;
                        println!("[Circuit] → Meio-Aberto: testando recuperação");
                        return true;
                    }
                }
                self.total_rejeitadas += 1;
                false
            }

            EstadoCircuito::MeioAberto => true,
        }
    }

    fn registrar_sucesso(&mut self) {
        self.total_requisicoes += 1;
        self.falhas_consecutivas = 0;

        if self.estado == EstadoCircuito::MeioAberto {
            self.estado = EstadoCircuito::Fechado;
            println!("[Circuit] → Fechado: serviço recuperado");
        }
    }

    fn registrar_falha(&mut self) {
        self.total_requisicoes += 1;
        self.total_falhas += 1;
        self.falhas_consecutivas += 1;
        self.ultimo_teste = Some(Instant::now());

        if self.falhas_consecutivas >= self.limite_falhas {
            if self.estado != EstadoCircuito::Aberto {
                self.estado = EstadoCircuito::Aberto;
                println!(
                    "[Circuit] → Aberto: {} falhas consecutivas",
                    self.falhas_consecutivas
                );
            }
        }
    }

    fn metricas(&self) -> String {
        let taxa_falha = if self.total_requisicoes > 0 {
            self.total_falhas as f64 / self.total_requisicoes as f64 * 100.0
        } else {
            0.0
        };

        format!(
            "Estado: {:?} | Req: {} | Falhas: {} ({:.1}%) | Rejeitadas: {}",
            self.estado,
            self.total_requisicoes,
            self.total_falhas,
            taxa_falha,
            self.total_rejeitadas,
        )
    }
}

// Circuit breaker thread-safe
#[derive(Clone)]
struct CircuitBreakerArc {
    interno: Arc<Mutex<CircuitBreaker>>,
}

impl CircuitBreakerArc {
    fn novo(limite: u32, timeout: Duration) -> Self {
        CircuitBreakerArc {
            interno: Arc::new(Mutex::new(CircuitBreaker::novo(limite, timeout))),
        }
    }

    async fn executar<F, Fut, T, E>(
        &self,
        operacao: F,
    ) -> Result<T, ErroCircuito<E>>
    where
        F: FnOnce() -> Fut,
        Fut: std::future::Future<Output = Result<T, E>>,
    {
        let pode = {
            let mut cb = self.interno.lock().await;
            cb.pode_executar()
        };

        if !pode {
            return Err(ErroCircuito::CircuitoAberto);
        }

        match operacao().await {
            Ok(resultado) => {
                self.interno.lock().await.registrar_sucesso();
                Ok(resultado)
            }
            Err(e) => {
                self.interno.lock().await.registrar_falha();
                Err(ErroCircuito::FalhaOperacao(e))
            }
        }
    }

    async fn metricas(&self) -> String {
        self.interno.lock().await.metricas()
    }
}

#[derive(Debug)]
enum ErroCircuito<E> {
    CircuitoAberto,
    FalhaOperacao(E),
}

// Simulação de serviço instável
async fn servico_instavel(
    tentativa: u32,
) -> Result<String, String> {
    // Falha nas primeiras 5 tentativas, depois recupera
    if tentativa < 5 {
        tokio::time::sleep(Duration::from_millis(10)).await;
        Err(format!("Timeout na tentativa {tentativa}"))
    } else {
        Ok(format!("Sucesso na tentativa {tentativa}"))
    }
}

async fn demo_circuit_breaker() {
    println!("══ Circuit Breaker ══
");

    let cb = CircuitBreakerArc::novo(3, Duration::from_millis(200));

    for i in 0..15u32 {
        let resultado = cb.executar(|| servico_instavel(i)).await;

        match &resultado {
            Ok(msg)  => println!("  [{}] ✓ {}", i, msg),
            Err(ErroCircuito::CircuitoAberto) =>
                println!("  [{}] ⊘ Circuito aberto — rejeitado", i),
            Err(ErroCircuito::FalhaOperacao(e)) =>
                println!("  [{}] ✗ Falha: {}", i, e),
        }

        // Pequena espera para simular tempo entre requisições
        tokio::time::sleep(Duration::from_millis(50)).await;
    }

    println!("
Métricas: {}", cb.metricas().await);
}

Retry com backoff exponencial

use std::time::Duration;
use tokio::time::sleep;

#[derive(Debug, Clone)]
struct ConfigRetry {
    max_tentativas: u32,
    espera_inicial: Duration,
    fator_multiplicador: f64,
    espera_maxima: Duration,
    jitter: bool,
}

impl Default for ConfigRetry {
    fn default() -> Self {
        ConfigRetry {
            max_tentativas: 5,
            espera_inicial: Duration::from_millis(100),
            fator_multiplicador: 2.0,
            espera_maxima: Duration::from_secs(30),
            jitter: true,
        }
    }
}

async fn com_retry<F, Fut, T, E>(
    config: &ConfigRetry,
    nome_operacao: &str,
    mut operacao: F,
) -> Result<T, E>
where
    F: FnMut(u32) -> Fut,
    Fut: std::future::Future<Output = Result<T, E>>,
    E: std::fmt::Display,
{
    let mut tentativa = 0;
    let mut espera = config.espera_inicial;

    loop {
        tentativa += 1;

        match operacao(tentativa).await {
            Ok(resultado) => {
                if tentativa > 1 {
                    println!(
                        "[Retry] '{}' sucesso na tentativa {}",
                        nome_operacao, tentativa
                    );
                }
                return Ok(resultado);
            }

            Err(e) if tentativa >= config.max_tentativas => {
                println!(
                    "[Retry] '{}' falhou após {} tentativas: {}",
                    nome_operacao, tentativa, e
                );
                return Err(e);
            }

            Err(e) => {
                println!(
                    "[Retry] '{}' tentativa {}/{} falhou: {}. Aguardando {:?}",
                    nome_operacao, tentativa, config.max_tentativas,
                    e, espera
                );

                sleep(espera).await;

                // Backoff exponencial com jitter
                let proxima = espera.as_secs_f64() * config.fator_multiplicador;
                espera = Duration::from_secs_f64(proxima)
                    .min(config.espera_maxima);

                if config.jitter {
                    use rand::Rng;
                    let jitter_fator = rand::thread_rng().gen_range(0.8..1.2);
                    espera = Duration::from_secs_f64(
                        espera.as_secs_f64() * jitter_fator
                    );
                }
            }
        }
    }
}

async fn demo_retry() {
    println!("══ Retry com Backoff Exponencial ══
");

    let config = ConfigRetry {
        max_tentativas: 4,
        espera_inicial: Duration::from_millis(50),
        fator_multiplicador: 2.0,
        espera_maxima: Duration::from_millis(500),
        jitter: false,
    };

    let mut tentativas_banco = 0u32;

    let resultado = com_retry(
        &config,
        "conectar_banco",
        |_n| {
            tentativas_banco += 1;
            let t = tentativas_banco;
            async move {
                if t < 3 {
                    Err(format!("Conexão recusada (tentativa {t})"))
                } else {
                    Ok("Conectado com sucesso!".to_string())
                }
            }
        },
    ).await;

    println!("Resultado: {:?}", resultado);
}

Protocolo de Gossip — propagação de estado

use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use std::time::{Duration, Instant};

// Vetor de relógio para ordenação causal de eventos
#[derive(Debug, Clone, Default)]
struct VetorRelogio {
    tempos: HashMap<String, u64>,
}

impl VetorRelogio {
    fn incrementar(&mut self, no_id: &str) {
        *self.tempos.entry(no_id.to_string()).or_insert(0) += 1;
    }

    fn merge(&mut self, outro: &VetorRelogio) {
        for (no, &tempo) in &outro.tempos {
            let entry = self.tempos.entry(no.clone()).or_insert(0);
            *entry = (*entry).max(tempo);
        }
    }

    fn aconteceu_antes(&self, outro: &VetorRelogio) -> bool {
        // self → outro (self aconteceu causalmente antes de outro)
        self.tempos.iter().all(|(no, &t_self)| {
            outro.tempos.get(no).copied().unwrap_or(0) >= t_self
        }) && self.tempos != outro.tempos
    }
}

// Estado de cada nó no cluster
#[derive(Debug, Clone)]
struct EstadoNo {
    id: String,
    endereco: String,
    versao: u64,
    dados: HashMap<String, String>,
    relogio: VetorRelogio,
    ultimo_heartbeat: Instant,
}

// Nó participante do protocolo gossip
struct NoGossip {
    id: String,
    estado: Arc<RwLock<EstadoNo>>,
    nos_conhecidos: Arc<RwLock<HashMap<String, EstadoNo>>>,
}

impl NoGossip {
    fn novo(id: &str, endereco: &str) -> Self {
        let estado = EstadoNo {
            id: id.to_string(),
            endereco: endereco.to_string(),
            versao: 0,
            dados: HashMap::new(),
            relogio: VetorRelogio::default(),
            ultimo_heartbeat: Instant::now(),
        };

        NoGossip {
            id: id.to_string(),
            estado: Arc::new(RwLock::new(estado)),
            nos_conhecidos: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    async fn definir(&self, chave: &str, valor: &str) {
        let mut estado = self.estado.write().await;
        estado.relogio.incrementar(&self.id);
        estado.versao += 1;
        estado.dados.insert(chave.to_string(), valor.to_string());
        println!(
            "[{}] Definido: {} = {} (v{})",
            self.id, chave, valor, estado.versao
        );
    }

    async fn obter(&self, chave: &str) -> Option<String> {
        self.estado.read().await.dados.get(chave).cloned()
    }

    // Simula envio de gossip para outro nó
    async fn gossip_para(&self, destino: &NoGossip) {
        let meu_estado = self.estado.read().await.clone();

        let mut nos_destino = destino.nos_conhecidos.write().await;
        let mut estado_destino = destino.estado.write().await;

        // Merge do estado recebido
        let entrada = nos_destino
            .entry(self.id.clone())
            .or_insert_with(|| meu_estado.clone());

        if meu_estado.versao > entrada.versao {
            println!(
                "[{}] ← gossip de [{}]: versão {} → {}",
                destino.id, self.id, entrada.versao, meu_estado.versao
            );

            // Merge de dados (last-write-wins baseado em versão)
            for (chave, valor) in &meu_estado.dados {
                if !estado_destino.dados.contains_key(chave) {
                    estado_destino.dados.insert(chave.clone(), valor.clone());
                    println!(
                        "[{}] Aprendeu: {} = {}",
                        destino.id, chave, valor
                    );
                }
            }

            estado_destino.relogio.merge(&meu_estado.relogio);
            *entrada = meu_estado;
        }
    }

    async fn status(&self) -> String {
        let estado = self.estado.read().await;
        let nos = self.nos_conhecidos.read().await;

        format!(
            "Nó [{}] | v{} | {} chaves | {} nós conhecidos",
            self.id,
            estado.versao,
            estado.dados.len(),
            nos.len(),
        )
    }
}

async fn demo_gossip() {
    println!("══ Protocolo Gossip ══
");

    let no1 = NoGossip::novo("A", "10.0.0.1:7000");
    let no2 = NoGossip::novo("B", "10.0.0.2:7000");
    let no3 = NoGossip::novo("C", "10.0.0.3:7000");

    // Cada nó define suas próprias chaves
    no1.definir("usuario:1", "Ana").await;
    no1.definir("usuario:2", "Bob").await;
    no2.definir("config:timeout", "30s").await;
    no3.definir("config:max_conn", "100").await;

    println!("
── Antes do gossip ──");
    println!("{}", no1.status().await);
    println!("{}", no2.status().await);
    println!("{}", no3.status().await);

    println!("
── Rodada de gossip ──");
    // A → B
    no1.gossip_para(&no2).await;
    // B → C
    no2.gossip_para(&no3).await;
    // C → A
    no3.gossip_para(&no1).await;

    println!("
── Após gossip ──");
    println!("{}", no1.status().await);
    println!("{}", no2.status().await);
    println!("{}", no3.status().await);

    // Verifica convergência
    println!("
── Verificação de convergência ──");
    for no in [&no1, &no2, &no3] {
        let val = no.obter("usuario:1").await;
        println!("[{}] usuario:1 = {:?}", no.id, val);
    }
}

Raft simplificado — eleição de líder

use std::sync::Arc;
use tokio::sync::{RwLock, mpsc};
use std::time::Duration;
use rand::Rng;

#[derive(Debug, Clone, PartialEq)]
enum PapelRaft {
    Seguidor,
    Candidato,
    Lider,
}

#[derive(Debug, Clone)]
struct EntradaLog {
    termo: u64,
    indice: u64,
    comando: String,
}

#[derive(Debug)]
struct EstadoRaft {
    id: u32,
    papel: PapelRaft,
    termo_atual: u64,
    votou_para: Option<u32>,
    log: Vec<EntradaLog>,
    votos_recebidos: u32,
    total_nos: u32,
}

impl EstadoRaft {
    fn novo(id: u32, total_nos: u32) -> Self {
        EstadoRaft {
            id,
            papel: PapelRaft::Seguidor,
            termo_atual: 0,
            votou_para: None,
            log: Vec::new(),
            votos_recebidos: 0,
            total_nos,
        }
    }

    fn maioria(&self) -> u32 {
        self.total_nos / 2 + 1
    }

    fn iniciar_eleicao(&mut self) {
        self.termo_atual += 1;
        self.papel = PapelRaft::Candidato;
        self.votou_para = Some(self.id); // vota em si mesmo
        self.votos_recebidos = 1;
        println!(
            "[Nó {}] Iniciando eleição — termo {}",
            self.id, self.termo_atual
        );
    }

    fn receber_voto(&mut self) -> bool {
        self.votos_recebidos += 1;
        println!(
            "[Nó {}] Recebeu voto ({}/{} necessários)",
            self.id, self.votos_recebidos, self.maioria()
        );

        if self.votos_recebidos >= self.maioria()
            && self.papel == PapelRaft::Candidato
        {
            self.papel = PapelRaft::Lider;
            println!(
                "[Nó {}] ★ Eleito LÍDER no termo {}!",
                self.id, self.termo_atual
            );
            return true;
        }
        false
    }

    fn receber_heartbeat(&mut self, termo_lider: u64, lider_id: u32) {
        if termo_lider >= self.termo_atual {
            if self.papel != PapelRaft::Seguidor {
                println!(
                    "[Nó {}] Revertendo para seguidor (líder: {})",
                    self.id, lider_id
                );
            }
            self.papel = PapelRaft::Seguidor;
            self.termo_atual = termo_lider;
        }
    }

    fn pode_votar(&mut self, candidato_id: u32, termo: u64) -> bool {
        if termo < self.termo_atual {
            return false;
        }
        if termo > self.termo_atual {
            self.termo_atual = termo;
            self.votou_para = None;
        }
        if self.votou_para.is_none() || self.votou_para == Some(candidato_id) {
            self.votou_para = Some(candidato_id);
            println!(
                "[Nó {}] Votando no candidato {} (termo {})",
                self.id, candidato_id, termo
            );
            return true;
        }
        false
    }

    fn adicionar_entrada(&mut self, comando: &str) -> Option<u64> {
        if self.papel != PapelRaft::Lider {
            return None;
        }
        let indice = self.log.len() as u64 + 1;
        self.log.push(EntradaLog {
            termo: self.termo_atual,
            indice,
            comando: comando.to_string(),
        });
        println!(
            "[Nó {}] Log[{}]: '{}'",
            self.id, indice, comando
        );
        Some(indice)
    }
}

async fn simulacao_raft() {
    println!("══ Simulação Raft (eleição) ══
");

    let nos: Vec<Arc<RwLock<EstadoRaft>>> = (1..=5)
        .map(|id| Arc::new(RwLock::new(EstadoRaft::novo(id, 5))))
        .collect();

    // Nó 2 inicia eleição (timeout expirou)
    {
        let mut no2 = nos[1].write().await;
        no2.iniciar_eleicao();
    }

    // Solicita votos dos demais nós
    let (termo, candidato_id) = {
        let no2 = nos[1].read().await;
        (no2.termo_atual, no2.id)
    };

    for (i, no) in nos.iter().enumerate() {
        if i == 1 { continue; } // pula a si mesmo

        let votou = {
            let mut estado = no.write().await;
            estado.pode_votar(candidato_id, termo)
        };

        if votou {
            let mut no2 = nos[1].write().await;
            let eleito = no2.receber_voto();
            if eleito { break; }
        }
    }

    println!("
── Líder envia heartbeats ──");

    // Líder envia heartbeats para os seguidores
    let (termo_lider, lider_id) = {
        let lider = nos[1].read().await;
        (lider.termo_atual, lider.id)
    };

    for (i, no) in nos.iter().enumerate() {
        if i == 1 { continue; }
        no.write().await.receber_heartbeat(termo_lider, lider_id);
    }

    println!("
── Líder adiciona entradas ao log ──");

    {
        let mut lider = nos[1].write().await;
        lider.adicionar_entrada("SET x = 42");
        lider.adicionar_entrada("SET y = 100");
        lider.adicionar_entrada("DELETE z");
    }

    println!("
── Estado final ──");
    for no in &nos {
        let estado = no.read().await;
        println!(
            "  Nó {} | {:?} | termo {} | log: {} entradas",
            estado.id,
            estado.papel,
            estado.termo_atual,
            estado.log.len(),
        );
    }
}

#[tokio::main]
async fn main() {
    demo_ator().await;
    println!("
{}
", "═".repeat(50));

    demo_circuit_breaker().await;
    println!("
{}
", "═".repeat(50));

    demo_retry().await;
    println!("
{}
", "═".repeat(50));

    demo_gossip().await;
    println!("
{}
", "═".repeat(50));

    simulacao_raft().await;
}

Padrões de mensageria com NATS

[dependencies]
async-nats = "0.34"
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"

use async_nats::Client;
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
struct EventoPedido {
    id: String,
    usuario_id: u64,
    itens: Vec<String>,
    total: f64,
}

#[derive(Debug, Serialize, Deserialize)]
struct EventoEstoque {
    produto_id: String,
    quantidade: i32,
    operacao: String,
}

async fn produtor_pedidos(cliente: &Client) -> anyhow::Result<()> {
    let pedido = EventoPedido {
        id: "PED-001".to_string(),
        usuario_id: 42,
        itens: vec!["Teclado".to_string(), "Mouse".to_string()],
        total: 435.00,
    };

    let payload = serde_json::to_vec(&pedido)?;

    cliente.publish("pedidos.criados", payload.into()).await?;
    println!("Pedido publicado: {}", pedido.id);

    // Request-reply: aguarda resposta do processador
    let resposta = cliente
        .request(
            "pedidos.validar",
            serde_json::to_vec(&pedido)?.into(),
        )
        .await?;

    let resposta_str = String::from_utf8_lossy(&resposta.payload);
    println!("Validação: {resposta_str}");

    Ok(())
}

async fn consumidor_pedidos(cliente: &Client) -> anyhow::Result<()> {
    let mut assinatura = cliente.subscribe("pedidos.criados").await?;

    println!("Aguardando pedidos...");

    while let Some(msg) = assinatura.next().await {
        let pedido: EventoPedido = serde_json::from_slice(&msg.payload)?;
        println!("Pedido recebido: {} (R${:.2})", pedido.id, pedido.total);

        // Processa e publica evento de estoque
        for item in &pedido.itens {
            let evento_estoque = EventoEstoque {
                produto_id: item.clone(),
                quantidade: -1,
                operacao: "reserva".to_string(),
            };
            let payload = serde_json::to_vec(&evento_estoque)?;
            cliente.publish("estoque.atualizacoes", payload.into()).await?;
        }
    }

    Ok(())
}

Observabilidade com tracing distribuído

[dependencies]
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["json"] }
opentelemetry = "0.22"
tracing-opentelemetry = "0.23"

use tracing::{instrument, info, warn, error, Span};
use tracing_subscriber::prelude::*;

#[instrument(
    name = "processar_requisicao",
    fields(
        requisicao_id = %id,
        usuario_id = tracing::field::Empty,
    )
)]
async fn processar_requisicao(id: &str, payload: &str) -> Result<String, String> {
    info!("Iniciando processamento");

    let resultado = validar_payload(payload).await?;

    // Adiciona campo ao span atual dinamicamente
    Span::current().record("usuario_id", resultado.usuario_id);

    let resposta = executar_logica(resultado).await?;

    info!(duracao_ms = 42, "Processamento concluído");

    Ok(resposta)
}

#[instrument(skip(payload))]
async fn validar_payload(payload: &str) -> Result<DadosValidados, String> {
    if payload.is_empty() {
        warn!("Payload vazio recebido");
        return Err("Payload não pode ser vazio".to_string());
    }

    Ok(DadosValidados {
        usuario_id: 123,
        dados: payload.to_string(),
    })
}

#[instrument]
async fn executar_logica(dados: DadosValidados) -> Result<String, String> {
    Ok(format!("Processado para usuário {}", dados.usuario_id))
}

#[derive(Debug)]
struct DadosValidados {
    usuario_id: u64,
    dados: String,
}

fn configurar_tracing() {
    tracing_subscriber::registry()
        .with(tracing_subscriber::fmt::layer().json())
        .with(tracing_subscriber::EnvFilter::from_default_env())
        .init();
}

Fontes e leituras recomendadas

"Designing Data-Intensive Applications" — Martin Kleppmann — a bíblia de sistemas distribuídos — https://dataintensive.net
"Distributed Systems" — Maarten van Steen & Andrew Tanenbaum — fundamentos teóricos
tokio documentation — runtime assíncrono — https://tokio.rs
async-nats crate — cliente NATS para Rust — https://docs.rs/async-nats
Raft paper — "In Search of an Understandable Consensus Algorithm" — https://raft.github.io/raft.pdf
openraft crate — implementação Raft em Rust — https://docs.rs/openraft
"The CAP Theorem" — Eric Brewer — https://www.infoq.com/articles/cap-twelve-years-later-how-the-rules-have-changed/
tracing crate — observabilidade — https://docs.rs/tracing

Artigo #43 de 52 | Série: Dominando Rust em 1 Ano Próximo → Artigo #44: Compiladores e Interpretadores — Construindo uma linguagem com Rust

Sistemas Distribuídos — Consenso, Mensageria e Tolerância a Falhas Já leu