Processos Distribuídos

Plataforma modular de gestão e execução distribuída de tarefas, numa arquitetura cliente-servidor com suporte para dispositivos móveis.

Nós (dispositivos) registam-se num núcleo central, recebem tarefas de uma fila, executam-nas, sincronizam o estado e devolvem resultados — com identificação, validação, controlo de execução, tratamento de falhas e recuperação após interrupções.

Estado atual: Fases 1–7 (roteiro completo) — comunicação, persistência (SQLite/PostgreSQL), recuperação, backoff + long-polling, app Flutter (segundo plano via WorkManager/BGTask), segurança (JWT + refresh, admin), observabilidade (request-id, /metrics), migrações (Alembic) e executores reais acopláveis + registo de device tokens/push. Servidor e worker Python testados (51 testes); a app Flutter está escrita (compilar requer SDK).

Estrutura do repositório

processos-distribuidos/
├── server/          # Núcleo central (FastAPI)
│   ├── app/         # Código da API
│   └── tests/       # Testes do servidor
├── worker/          # Nó de referência em Python (simula o dispositivo móvel)
│   ├── worker/      # Código do worker
│   └── tests/       # Testes ponta-a-ponta
├── mobile/          # App Flutter (o nó real; mesmo contrato do worker)
├── docs/            # Arquitetura e decisões (ADRs)
└── scripts/         # Scripts de apoio

Pré-requisitos

• Python 3.11+

Arranque rápido

# 1. Criar ambiente e instalar dependências
python3 -m venv .venv
. .venv/bin/activate
pip install -r server/requirements.txt -r worker/requirements.txt

# 2. Arrancar o servidor (com 3 tarefas de demonstração)
PD_SEED_DEMO=1 python -m uvicorn app.main:app --app-dir server --port 8000

# 3. Noutro terminal: correr um nó que processa 3 tarefas
python -m worker --server http://127.0.0.1:8000 --max-tasks 3

Documentação interativa da API: http://127.0.0.1:8000/docs

Testes

. .venv/bin/activate
cd server && python -m pytest -q && cd ..
cd worker && python -m pytest -q && cd ..

Ou tudo de uma vez:

bash scripts/test_all.sh

Persistência (Fase 2)

O estado (nós, tarefas, resultados) é guardado numa base de dados, pelo que sobrevive a reinícios do servidor.

Variável	Omissão	Descrição
PD_STORE	sql	sql (persistente) ou memory
PD_DATABASE_URL	sqlite:///./pd.db	Em produção: postgresql+psycopg://utilizador:senha@host:5432/base
PD_LEASE_SECONDS	60	Prazo de uma tarefa atribuída antes de ser dada como perdida
PD_REAPER_INTERVAL	0	Intervalo (s) da varredura que repõe tarefas expiradas (0 = desligado)
PD_BACKOFF_BASE	2	Base (s) do atraso exponencial entre tentativas (0 = desligado)
PD_BACKOFF_CAP	300	Teto (s) do atraso entre tentativas
PD_LONGPOLL_MAX	25	Tempo máximo (s) que /tasks/next?wait= pode esperar
PD_TOKEN_SECRET	dev-...	Segredo para assinar os JWT (trocar em produção)
PD_ACCESS_TTL	900	Validade (s) do access token
PD_REFRESH_TTL	2592000	Validade (s) do refresh token (30 dias)
PD_ADMIN_KEY	(vazio)	Chave para criar tarefas; vazio = aberto (só dev)

Para usar PostgreSQL, instalar o driver (ver server/requirements.txt) e definir PD_DATABASE_URL. O código é o mesmo — só muda a variável.

Contrato da API

Método	Rota	Descrição
GET	/health	Estado do servidor
POST	/v1/nodes/register	Regista um nó, devolve access_token + refresh_token
POST	/v1/nodes/refresh	Troca o refresh_token por novo access_token
POST	/v1/nodes/{id}/heartbeat	Atualiza estado/capacidade (autenticado)
POST	/v1/nodes/{id}/push-token	Regista o device token (FCM/APNs) do nó
POST	/v1/tasks	Coloca uma tarefa na fila; X-Admin-Key se configurado
GET	/metrics	Contadores de pedidos/erros (observabilidade)
GET	/v1/tasks/next	Entrega a próxima tarefa ao nó, com prazo; ?wait=N para long-poll (autenticado)
GET	/v1/tasks/assigned	Tarefas a cargo do nó, para retomar (autenticado)
POST	/v1/tasks/{id}/status	Reporta progresso e renova o prazo (autenticado)
POST	/v1/tasks/{id}/fail	Reporta falha → requeue ou dead-letter (autenticado)
POST	/v1/tasks/{id}/result	Entrega resultado, idempotente (autenticado)

Autenticação: Authorization: Bearer <access_token> (JWT). Quando expira, trocar o refresh_token em /v1/nodes/refresh (os clientes fazem-no sozinhos).

Tipos de tarefa (executores)

Cada tarefa tem um type resolvido por um executor registado (worker/worker/executor.py). Incluídos:

type	payload	Resultado
echo	qualquer	devolve o payload
sum	{"numeros": [1,2,3]}	{"soma": 6, "n": 3}
wordcount	{"texto": "..."}	{"palavras": N, "caracteres": M}
sha256	{"texto": "..."}	{"sha256": "..."}
fail	{"motivo": "..."}	falha propositada (testa recuperação)

Acrescentar um tipo novo é só registar uma função com @register("tipo").

Migrações de base de dados (Alembic)

O esquema é versionado em server/alembic/. Em produção:

cd server
PD_DATABASE_URL=postgresql+psycopg://... alembic upgrade head

Gerar uma nova migração após alterar os modelos:

alembic revision --autogenerate -m "descricao"
alembic check   # confirma que migrações e modelos coincidem

Próximas fases

Ver docs/ARCHITECTURE.md para o plano completo.