Graceful Shutdown em Python Puro: O Protocolo Que Faz Seu Processo Morrer Sem Perder Dados (Nem Corromper o Estado)

PorOlivetto

Seu Processo Morre. Seus Dados Morrem Junto?

Você tem um worker Python processando uma fila. Recebe SIGTERM. O que acontece? Se você não implementou graceful shutdown, a resposta é brutal: o processo morre no meio de uma transação, conexões ficam abertas, arquivos ficam pela metade, e aquele dado que estava sendo processado? Perdido.

Graceful shutdown não é luxo de produção. É a diferença entre um deploy tranquilo e uma madrugada apagando dados corrompidos. E a implementação em Python puro leva menos de 60 linhas.

O Que Acontece Quando o Python Morre Sem Aviso

Quando o sistema operacional manda SIGTERM pro seu processo Python (via kill, deploy, Docker stop, Kubernetes pod eviction), o comportamento padrão é:

  • O interpretador Python recebe o sinal
  • Lança SystemExit se não houver handler
  • O processo encerra imediatamente
  • Todos os recursos abertos são abandonados

Na prática, isso significa:

import time
import sqlite3

# Código sem graceful shutdown — receita pro desastre
conn = sqlite3.connect("jobs.db")
cursor = conn.cursor()

while True:
    job = buscar_proximo_job(cursor)
    if job:
        processar(job)  # Se morrer aqui, o job fica pela metade
        marcar_concluido(cursor, job)
        conn.commit()  # Ou nunca chega aqui
    time.sleep(1)

# conn.close() nunca é chamado
# Transações ficam abertas
# Locks no banco permanecem

Cada vez que esse processo é reiniciado sem graceful shutdown, você acumula jobs em estado inconsistente. Depois de 10 deploys, tem um banco de dados com dezenas de registros órfãos que ninguém sabe se foram processados ou não.

O Padrão: Graceful Shutdown em 3 Fases

Um shutdown graceful segue três etapas claras:

  1. Parar de aceitar trabalho novo — o processo não pega mais jobs da fila
  2. Terminar o trabalho atual — o job em andamento é completado (ou tem timeout)
  3. Limpar recursos — conexões fechadas, arquivos flushados, estado persistido

Isso parece óbvio. Mas a implementação precisa de cuidado porque sinais Unix são assíncronos — chegam a qualquer momento, inclusive no meio de uma operação crítica.

A Implementação: 60 Linhas Que Salvam Seus Dados

import signal
import sys
import time
import logging
import threading
from typing import Optional, Callable, List

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)


class GracefulShutdown:
    def __init__(self, timeout: float = 30.0):
        self.timeout = timeout
        self._shutdown_requested = False
        self._active_tasks: List[threading.Thread] = []
        self._lock = threading.Lock()
        self._cleanup_callbacks: List[Callable] = []

        # Registra handlers para os sinais de terminação
        signal.signal(signal.SIGTERM, self._handle_signal)
        signal.signal(signal.SIGINT, self._handle_signal)

    def _handle_signal(self, signum: int, frame):
        sig_name = signal.Signals(signum).name
        logger.info(f"Recebido {sig_name}. Iniciando shutdown graceful...")
        self._shutdown_requested = True

    @property
    def should_stop(self) -> bool:
        """Verifica se shutdown foi requisitado. Use no loop principal."""
        return self._shutdown_requested

    def on_cleanup(self, callback: Callable):
        """Registra função de limpeza para executar no shutdown."""
        self._cleanup_callbacks.append(callback)

    def track_task(self, task: threading.Thread):
        """Registra uma task ativa para aguardar no shutdown."""
        with self._lock:
            self._active_tasks.append(task)

    def untrack_task(self, task: threading.Thread):
        """Remove uma task completada da lista."""
        with self._lock:
            if task in self._active_tasks:
                self._active_tasks.remove(task)

    def wait_and_cleanup(self):
        """Aguarda tasks ativas e executa callbacks de limpeza."""
        if self._active_tasks:
            logger.info(
                f"Aguardando {len(self._active_tasks)} tasks "
                f"(timeout: {self.timeout}s)..."
            )
            deadline = time.time() + self.timeout
            for task in self._active_tasks[:]:
                remaining = max(0.1, deadline - time.time())
                task.join(timeout=remaining)
                if task.is_alive():
                    logger.warning(f"Task {task.name} não completou no timeout!")

        for callback in self._cleanup_callbacks:
            try:
                callback()
            except Exception as e:
                logger.error(f"Erro no cleanup callback: {e}")

        logger.info("Shutdown completo.")

Uso Prático: Worker de Fila com Shutdown Seguro

Agora vamos aplicar isso num cenário real — um worker que processa jobs de uma fila com SQLite:

import sqlite3
import threading


def criar_worker():
    shutdown = GracefulShutdown(timeout=15.0)
    conn = sqlite3.connect("jobs.db", check_same_thread=False)
    cursor = conn.cursor()

    # Callback de limpeza: fecha conexão com o banco
    def fechar_conexao():
        logger.info("Fechando conexão com banco de dados...")
        conn.close()

    shutdown.on_cleanup(fechar_conexao)

    job_atual = None
    job_lock = threading.Lock()

    while not shutdown.should_stop:
        # Buscar próximo job
        cursor.execute(
            "SELECT id, payload FROM jobs WHERE status = 'pendente' "
            "ORDER BY created_at LIMIT 1"
        )
        row = cursor.fetchone()

        if not row:
            time.sleep(2)  # Polling interval
            continue

        job_id, payload = row

        # Marcar como processando (transação atômica)
        cursor.execute(
            "UPDATE jobs SET status = 'processando' WHERE id = ? AND status = 'pendente'",
            (job_id,)
        )
        conn.commit()

        if cursor.rowcount == 0:
            continue  # Outro worker pegou primeiro

        # Processar o job (com tracking para shutdown)
        with job_lock:
            job_atual = job_id

        try:
            resultado = processar_payload(payload)
            cursor.execute(
                "UPDATE jobs SET status = 'concluido', resultado = ? WHERE id = ?",
                (resultado, job_id)
            )
            conn.commit()
            logger.info(f"Job {job_id} concluído com sucesso.")
        except Exception as e:
            cursor.execute(
                "UPDATE jobs SET status = 'erro', erro = ? WHERE id = ?",
                (str(e), job_id)
            )
            conn.commit()
            logger.error(f"Job {job_id} falhou: {e}")
        finally:
            with job_lock:
                job_atual = None

    # Se há job em andamento quando shutdown foi pedido,
    # a flag should_stop garante que não pega novo job
    logger.info("Worker parou de aceitar novos jobs.")
    shutdown.wait_and_cleanup()


if __name__ == "__main__":
    criar_worker()

Por Que signal.signal() Funciona (E Quando Não Funciona)

O módulo signal do Python registra handlers que são executados no thread principal. Isso tem implicações importantes:

  • Funciona: scripts single-threaded, loops de polling, servidores simples
  • Problemas: handlers de signal só rodam na thread principal. Se sua lógica está em threads secundárias, o handler ainda é chamado na main thread — o que é bom, porque a flag should_stop é visível para todas as threads.

A flag booleana é thread-safe em Python por causa do GIL (Global Interpreter Lock). Ler e escrever um booleano é uma operação atômica no CPython. Se você precisa de garantias mais fortes, use threading.Event():

class GracefulShutdownComEvent(GracefulShutdown):
    def __init__(self, timeout: float = 30.0):
        super().__init__(timeout)
        self._event = threading.Event()

    def _handle_signal(self, signum, frame):
        super()._handle_signal(signum, frame)
        self._event.set()  # Acorda threads bloqueadas no wait()

    @property
    def should_stop(self) -> bool:
        return self._event.is_set()

    def wait_for_stop(self, timeout: float = None) -> bool:
        """Bloqueia até shutdown ou timeout. Retorna True se shutdown."""
        return self._event.wait(timeout=timeout)

Timeout: Quando o Job Não Quer Morrer

O maior risco do graceful shutdown é um job que nunca termina. Um upload de 2GB, uma query que trava, um deadlock. Por isso o timeout existe — e quando ele estoura, você tem uma decisão:

  • SIGKILL (forçado): o processo morre na hora. Dados podem corromper. Mas pelo menos você sabe que o job ficou como “processando” e pode retomar.
  • Marcar como interrompido: antes do timeout, atualize o status do job para que o próximo worker saiba que precisa retomar.
def wait_and_cleanup_com_fallback(self, conn, cursor):
    """Shutdown com marcação de jobs interrompidos."""
    deadline = time.time() + self.timeout

    # Aguarda tasks ativas
    for task in self._active_tasks[:]:
        remaining = max(0.1, deadline - time.time())
        task.join(timeout=remaining)
        if task.is_alive():
            logger.warning(f"Task {task.name} excedeu timeout!")
            # Marca jobs como interrompidos para retomar depois
            cursor.execute(
                "UPDATE jobs SET status = 'pendente' "
                "WHERE status = 'processando'"
            )
            conn.commit()
            logger.info("Jobs em andamento marcados como pendentes para retry.")

    # Executa callbacks de limpeza
    for callback in self._cleanup_callbacks:
        try:
            callback()
        except Exception as e:
            logger.error(f"Erro no cleanup: {e}")

Docker, Systemd e Kubernetes: Sinais na Prática

Cada orquestrador tem seu comportamento de shutdown:

  • Docker: docker stop manda SIGTERM, espera 10s (padrão), depois SIGKILL. Configure --stop-timeout 30 no Dockerfile ou compose.
  • Kubernetes: manda SIGTERM, espera terminationGracePeriodSeconds (padrão 30s), depois SIGKILL.
  • Systemd: manda SIGTERM, espera TimeoutStopSec (padrão 90s), depois SIGKILL.

A lição: seu timeout de graceful shutdown deve ser menor que o timeout do orquestrador. Se o Docker mata em 10s, seu graceful shutdown precisa completar em 8s.

# Dockerfile: aumenta o grace period do Docker
# STOPSIGNAL SIGTERM
# (Docker já usa SIGTERM por padrão)

# docker-compose.yml:
# services:
#   worker:
#     stop_grace_period: 30s

# Kubernetes pod spec:
# terminationGracePeriodSeconds: 30

Shutdown Hook: O Padrão Completo

Para aplicações mais complexas, um shutdown hook global permite que qualquer módulo registre sua rotina de limpeza:

class ShutdownRegistry:
    """Registry global de hooks de shutdown."""
    _instance = None

    def __init__(self):
        self._hooks: List[tuple] = []  # (priority, name, callback)
        self._shutdown = GracefulShutdown(timeout=25.0)

    @classmethod
    def get(cls):
        if cls._instance is None:
            cls._instance = cls()
        return cls._instance

    def register(self, name: str, callback: Callable, priority: int = 100):
        """Registra hook. Prioridade menor = executa primeiro."""
        self._hooks.append((priority, name, callback))
        self._hooks.sort(key=lambda x: x[0])
        self._shutdown.on_cleanup(self._run_hooks_ordered)

    def _run_hooks_ordered(self):
        """Executa hooks em ordem de prioridade."""
        executados = set()
        for priority, name, callback in self._hooks:
            if name not in executados:
                try:
                    logger.info(f"Executando shutdown hook: {name}")
                    callback()
                    executados.add(name)
                except Exception as e:
                    logger.error(f"Hook '{name}' falhou: {e}")


# Uso em qualquer módulo:
# ShutdownRegistry.get().register("db", fechar_db, priority=10)
# ShutdownRegistry.get().register("cache", flushar_cache, priority=20)
# ShutdownRegistry.get().register("logs", fechar_logs, priority=90)

🧱 O Perrengue do Olivetto

Tinha um worker que processava pagamentos. Cada payment levava ~3 segundos. Quando fizemos deploy, o Docker mandou SIGTERM. O worker morreu no meio de um requests.post() pro gateway de pagamento.

Resultado: o gateway processou o pagamento, mas o worker nunca registrou a confirmação no banco. O cliente foi cobrado, o pedido ficou como “pendente”, e o suporte levou 2 dias pra entender o que aconteceu.

Depois disso, todo worker nosso tem graceful shutdown com três regras: (1) não pegar job novo depois do SIGTERM, (2) terminar o job atual com timeout de 10s, (3) se não der, marcar como pendente para retry. Zero pagamentos perdidos desde então.

Checklist: Seu Processo Está Pronto pra Morrer?

  • Sinal capturado: SIGTERM e SIGINT registrados?
  • Loop principal responde: verifica should_stop a cada iteração?
  • Jobs em andamento completam: tem lógica pra terminar o trabalho atual?
  • Timeout configurado: menor que o timeout do orquestrador?
  • Recursos liberados: conexões, arquivos, locks — tudo tem callback de cleanup?
  • Estado persistido: se morrer no meio, o próximo processo sabe de onde continuar?

E Agora?

Seu worker atual tem graceful shutdown ou morre na hora que recebe SIGTERM? Já perdeu dados por causa de deploy sem cuidado? Conta aí nos comentários — quanto mais perrengues a gente compartilha, menos madrugadas apagando incêndio.

Posts Similares

Deixe um comentário

O seu endereço de e-mail não será publicado. Campos obrigatórios são marcados com *