# QueueMaster - Arquitetura Detalhada

## Visão Geral da Arquitetura

O QueueMaster implementa uma arquitetura hierárquica de supervisão baseada em processos, onde cada nível tem responsabilidades específicas e bem definidas.

## Hierarquia de Componentes

```mermaid
graph TD
    A[Master Supervisor<br/>QueueMasterCommand] --> B1[Queue Supervisor 1<br/>default]
    A --> B2[Queue Supervisor 2<br/>reports]
    A --> B3[Queue Supervisor 3<br/>saft-export]

    B1 --> C1[Worker Process 1]
    B1 --> C2[Worker Process 2]
    B1 --> C3[Worker Process N]

    B2 --> D1[Worker Process 1]
    B2 --> D2[Worker Process N]

    B3 --> E1[Worker Process 1]

    F[Redis] -.->|Registry| A
    F -.->|Metrics| B1
    F -.->|Metrics| B2
    F -.->|Metrics| B3

    G[RabbitMQ] -.->|Jobs| C1
    G -.->|Jobs| C2
    G -.->|Jobs| C3
    G -.->|Jobs| D1
    G -.->|Jobs| D2
    G -.->|Jobs| E1

    style A fill:#ff6b6b
    style B1 fill:#4ecdc4
    style B2 fill:#4ecdc4
    style B3 fill:#4ecdc4
    style C1 fill:#95e1d3
    style C2 fill:#95e1d3
    style C3 fill:#95e1d3
```

## Componentes Principais

### 1. Master Supervisor

**Arquivo**: `Modules/QueueMaster/Console/QueueMasterCommand.php`
**Classe**: `MasterSupervisor`
**Responsabilidades**:

- Processo principal que coordena toda a infraestrutura
- Cria e gerencia Queue Supervisors
- Responde a sinais do sistema (SIGTERM, SIGINT, SIGUSR2)
- Monitora saúde dos supervisors
- Limpa workers órfãos
- Persiste estado no Redis

**Ciclo de Vida**:

```
Início
  ↓
Verifica duplicatas (não permite 2 masters)
  ↓
Cria Queue Supervisors (baseado na config)
  ↓
Loop Principal (while true)
  ↓
  ├─→ Processa sinais pendentes
  ├─→ Processa comandos da fila
  ├─→ Monitora supervisors (verifica se estão vivos)
  ├─→ Limpa workers órfãos (a cada 60s)
  ├─→ Persiste estado no Redis
  ├─→ Dispara evento MasterSupervisorLooped
  └─→ Sleep 1s
  ↓
Término (via sinal)
  ↓
Termina todos os supervisors
  ↓
Remove registro do Redis
  ↓
Exit(0)
```

**Código-chave**:

```php
// QueueMasterCommand.php:62
$master->monitor();

// MasterSupervisor.php:45-48
while (true) {
    sleep(1);
    $this->loop();
}
```

### 2. Queue Supervisor

**Arquivo**: `Modules/QueueMaster/MasterSupervisor/Supervisor/QueueSupervisor.php`
**Responsabilidades**:

- Gerencia um pool de workers para uma fila específica
- Cria processos workers filhos via `ProcessPool`
- Monitora workers e recicla quando necessário
- Escala workers (up/down) baseado em configuração
- Verifica se o Master parent ainda está vivo

**Ciclo de Vida**:

```
Início (criado pelo Master)
  ↓
Cria ProcessPool
  ↓
Escala para maxProcesses configurado
  ↓
Loop Principal (while true)
  ↓
  ├─→ Verifica se parent (Master) está vivo
  ├─→ Processa sinais pendentes
  ├─→ Processa comandos da fila
  ├─→ Monitora ProcessPool
  │   ├─→ Verifica workers ativos
  │   └─→ Remove workers mortos
  ├─→ Persiste estado no Redis
  ├─→ Dispara evento SupervisorLooped
  └─→ Sleep 1s
  ↓
Término (via sinal ou parent morto)
  ↓
Termina todos os workers (gracefully)
  ↓
Aguarda até 20s workers terminarem
  ↓
Force kill se necessário
  ↓
Remove registro do Redis
  ↓
Exit(0 ou 13)
```

**Detecção de Parent Morto**:

```php
// QueueSupervisor.php:78-96
protected function ensureParentIsRunning(): void
{
    $parentId = $this->options->parentId ?? getmypid();

    if (!posix_kill($parentId, 0)) {
        // Parent morreu, terminar supervisor
        $this->terminate(13);
    }
}
```

### 3. Command Listener (qm:listen-commands)

**Arquivo**: `Modules/QueueMaster/Console/ListenCommandsCommand.php`  
**Responsabilidade**: ficar residente em cada host escutando a fila de comandos (`CommandQueue`) e executar, localmente, qualquer instrução de escala/restart direcionada para aquele hostname.

#### Por que ele é obrigatório?

1. Quando o autoscaler (ou um operador via API) decide escalar, ele persiste um `ScalingCommand` no Redis e publica um evento `qm:commands:{hostname}`.
2. **Somente o listener** consome essa fila (`CommandQueue::getPendingCommands($hostname)`) e invoca o `CommandExecutor`.
3. Sem o listener, a fila `qm:command_queue:{hostname}` cresce indefinidamente e **nenhum scale up/down/restart sai do papel**, mesmo que o master esteja ativo.

#### Ciclo de Vida do Listener

```
Início
  ↓
Descobre hostname atual (ex.: app01)
  ↓
Loop principal (sleep 1s)
  ↓
  ├─→ A cada N segundos (poll-interval): busca `getPendingCommands(hostname)`
  ├─→ Para cada comando:
  │     ├─→ Valida se pode executar (hostname + ação permitida)
  │     ├─→ Invoca CommandExecutor (systemctl/signal)
  │     └─→ Marca resultado na CommandQueue (executed/failed)
  ├─→ Respeita sinais SIGTERM/SIGINT para shutdown limpo
  └─→ Loga falhas e continua
```

#### Sintomas de Listener Offline

- Command queue aumenta (`LLEN qm:command_queue:{hostname} >> 0`).
- Autoscaling mostra comandos “pendentes” no dashboard e nenhuma alteração ocorre.
- Logs exibem “CommandQueue: Published command ...” sem pares “Marked command ... as executed”.

#### Quantidade por host

- **1 listener por host físico/container** que roda workers.  
- Rodá-lo em background via systemd/supervisor garante que autoscaling continue após deploys e reboots.

#### Relação com o master

- O master **não** executa comandos de worker. Ele apenas injeta comandos e coordena supervisores.  
- O listener é o “braço operacional” que aplica as decisões na máquina local.

### 3. Process Pool

**Arquivo**: `Modules/QueueMaster/MasterSupervisor/Supervisor/ProcessPool.php`
**Responsabilidades**:

- Gerencia coleção de workers como array de processos
- Implementa scaling (up/down)
- Marca processos para terminação
- Remove processos mortos (pruning)

**Estrutura de Dados**:

```php
public array $processes = [];              // Workers ativos
public array $terminatingProcesses = [];   // Workers sendo terminados

// Estrutura de $terminatingProcesses:
[
    [
        'process' => WorkerProcess,
        'terminatedAt' => CarbonImmutable
    ],
    ...
]
```

**Operações Principais**:

```php
// Scale Up: adiciona novos workers
public function scaleUp(int $processes): void
{
    $difference = $processes - count($this->processes);

    for ($i = 0; $i < $difference; $i++) {
        $this->start(); // Cria novo worker
    }
}

// Scale Down: remove workers excedentes
public function scaleDown(int $processes): void
{
    $difference = count($this->processes) - $processes;

    // Remove workers do array e marca para terminação
    $terminating = array_splice($this->processes, 0, $difference);

    collect($terminating)->each(fn($process) =>
        $this->markForTermination($process)
    );
}
```

**Pruning de Processos**:

```php
// ProcessPool.php:171-179
protected function pruneTerminatingProcesses(): void
{
    // Para processos que estão hangados além do timeout
    $this->stopTerminatingProcessesThatAreHanging();

    // Remove processos que já terminaram
    $this->terminatingProcesses = collect($this->terminatingProcesses)
        ->filter(fn($entry) => $entry['process']->isRunning())
        ->values()
        ->all();
}
```

### 4. Worker Process

**Arquivo**: `Modules/QueueMaster/Console/QueueMasterWorkCommand.php`
**Classe Worker**: `Modules/Common/Queue/Worker.php`
**Responsabilidades**:

- Processo PHP individual que executa jobs
- Consome jobs de uma fila RabbitMQ específica
- Aplica limites (max_time, max_jobs, memory)
- Registra métricas no Redis
- Auto-recicla após atingir limites

**Ciclo de Vida**:

```
Início (via php og queue-master:work)
  ↓
Registra worker no Redis
  ↓
Registra hooks de lifecycle:
  ├─→ afterProcess: verifica max_time
  ├─→ afterProcess: verifica max_jobs
  ├─→ afterProcess: verifica memory_limit
  └─→ afterProcess: envia heartbeat
  ↓
Daemon Loop (while $isRunning)
  ↓
  ├─→ Verifica limites (ENTRE jobs)
  ├─→ Busca próximo job da fila
  ├─→ Se tem job:
  │   ├─→ Adquire lock distribuído
  │   ├─→ Restaura contexto tenant
  │   ├─→ Executa job em sandbox
  │   ├─→ Registra métricas
  │   ├─→ Libera lock
  │   └─→ Incrementa jobsProcessed
  ├─→ Se não tem job:
  │   └─→ Sleep configurado
  └─→ Coleta garbage (gc_collect_cycles)
  ↓
Término (limite atingido ou sinal)
  ↓
Cleanup:
  ├─→ Marca worker como stopping
  ├─→ Dispara evento WorkerStopping
  └─→ Remove registro do Redis
  ↓
Exit(0)
```

**Verificação de Limites (CRÍTICO)**:

```php
// QueueMasterWorkCommand.php:159-180
protected function registerMaxTimeCheck(Worker $worker, int $maxTime): void
{
    $startedAt = microtime(true);

    // Hook AFTER job processing (não durante!)
    $worker->afterProcess(function () use ($worker, $startedAt, $maxTime) {
        $elapsed = microtime(true) - $startedAt;

        if ($elapsed >= $maxTime) {
            $worker->stop('max_time_reached');
        }
    });
}
```

**IMPORTANTE**: Limites são verificados no hook `afterProcess`, ou seja, APÓS cada job terminar. Um job em execução NUNCA é interrompido.

## Sistema de Sinais POSIX

O QueueMaster usa sinais POSIX para comunicação entre processos:

### Sinais Suportados

| Sinal | Target | Ação |
|-------|--------|------|
| `SIGTERM` | Master/Worker | Shutdown gracioso |
| `SIGINT` | Master/Worker | Shutdown gracioso (Ctrl+C) |
| `SIGUSR2` | Master | Restart todos os supervisors |
| `SIGUSR1` | Supervisor | Pause supervisor |
| `SIGCONT` | Supervisor | Resume supervisor |
| `SIGALRM` | Worker | Timeout de job (interno) |

### Fluxo de Sinais

```mermaid
sequenceDiagram
    participant User
    participant Master
    participant Supervisor
    participant Worker

    User->>Master: kill -SIGTERM <master_pid>
    Master->>Master: working = false
    Master->>Supervisor: terminate()
    Supervisor->>Supervisor: working = false
    Supervisor->>Worker: terminate() via SIGTERM
    Worker->>Worker: Aguarda job atual terminar
    Worker->>Supervisor: Exit(0)
    Supervisor->>Supervisor: Aguarda todos workers (até 20s)
    Supervisor->>Master: Exit(0)
    Master->>Master: Remove registro Redis
    Master->>User: Exit(0)
```

### Implementação de Signal Handlers

```php
// MasterSupervisor.php (via trait ListensForSignals)
pcntl_async_signals(true);

pcntl_signal(SIGTERM, function () use ($master) {
    $master->terminate();
});

pcntl_signal(SIGUSR2, function () use ($master) {
    $master->restart(); // Restart supervisors
});
```

## Integração com Redis

O Redis é usado como:

### 1. Registry de Workers

**Key Pattern**: `queuemaster:workers:{hostname}:{pid}`

**Dados Armazenados**:
```php
[
    'pid' => 12345,
    'hostname' => 'server1',
    'name' => 'default:worker-12345',
    'supervisor' => 'default',
    'queue' => 'default',
    'vhost' => '/tenant1',
    'started_at' => 1704067200,
    'memory_usage' => 12582912,
    'memory_limit' => 536870912,
    'max_time' => 300,
    'max_jobs' => 0,
    'status' => 'processing',
    'jobs_processed' => 42,
    'current_job' => 'job-uuid',
    'manager' => 'queue-master',
    'last_heartbeat' => 1704067500,
]
```

**TTL**: 90 segundos (renovado a cada heartbeat)

### 2. Tracking de Jobs

**Key Pattern**: `queuemaster:jobs:active:{job_id}`

**Dados**:
```php
[
    'job_id' => 'uuid',
    'job_class' => 'App\\Jobs\\ExampleJob',
    'hostname' => 'server1',
    'pid' => 12345,
    'queue' => 'default',
    'vhost' => '/tenant1',
    'tenant' => 'tenant1.test',
    'started_at' => 1704067200.123,
    'status' => 'running',
]
```

### 3. Shutdown History

**Key**: `queuemaster:shutdowns:history` (sorted set)

**Score**: timestamp
**Member**:
```php
json_encode([
    'pid' => 12345,
    'hostname' => 'server1',
    'shutdown_time' => 1704067500,
    'shutdown_reason' => 'max_time_reached',
    'uptime_seconds' => 300,
    'jobs_processed' => 60,
    'queue' => 'default',
])
```

### 4. Tenant Switch Monitoring

**Key**: `queuemaster:tenant-switches` (sorted set)

**Dados**:
```php
[
    'pid' => 12345,
    'hostname' => 'server1',
    'from_tenant' => 'tenant1.test',
    'to_tenant' => 'tenant2.test',
    'from_vhost' => '/tenant1',
    'to_vhost' => '/tenant1',
    'company_id' => 100,
    'success' => true,
    'timestamp' => 1704067500.456,
]
```

## Integração com RabbitMQ

### Conexão e Isolamento

Cada worker se conecta ao RabbitMQ usando:

```php
[
    'connection' => 'rabbitmq',
    'vhost' => '/tenant1',  // Isolamento por tenant
    'queue' => 'default',
]
```

### Fluxo de Consumo de Jobs

```mermaid
sequenceDiagram
    participant Worker
    participant RabbitMQ
    participant Redis
    participant Job

    Worker->>RabbitMQ: pop(queue)
    RabbitMQ-->>Worker: rawPayload (ou null)

    alt Job disponível
        Worker->>Redis: acquireLock(jobId)
        Redis-->>Worker: lockToken

        Worker->>Worker: unserialize(payload)
        Worker->>Worker: restartApplication(tenant)
        Worker->>Job: handle()
        Job-->>Worker: success

        Worker->>Redis: recordMetrics()
        Worker->>RabbitMQ: delete(jobId)
        Worker->>Redis: releaseLock(jobId, token)
    else Sem jobs
        Worker->>Worker: gc_collect_cycles()
        Worker->>Worker: sleep(3s)
    end
```

### Heartbeat RabbitMQ

Workers mantêm conexão viva via heartbeat:

```php
// Worker.php:1466-1475
private function connectionWait(int $lastHeartbeat): void
{
    if (time() - $lastHeartbeat >= $this->manager->driver()->heartbeatTime()) {
        $this->manager->driver()->wait(); // Envia heartbeat
    }
}
```

**Chamado**: Via `SIGALRM` a cada 10 segundos durante execução de job

## Fluxo Completo de Execução de Job

```mermaid
sequenceDiagram
    participant Pool as ProcessPool
    participant WP as WorkerProcess
    participant W as Worker
    participant Job as JobClass
    participant RMQ as RabbitMQ
    participant Redis

    Pool->>WP: start() - fork process
    WP->>W: daemon(queue)

    loop Daemon Loop
        W->>W: Verifica limites (max_time, jobs, memory)

        alt Limite atingido
            W->>W: stop('reason')
            W->>WP: exit(0)
        end

        W->>RMQ: pop(queue)

        alt Job encontrado
            RMQ-->>W: payload
            W->>Redis: acquireLock(jobId)

            W->>W: restartApplication(job.tenant)
            Note over W: Muda $_SERVER['HTTP_HOST']<br/>Recarrega config<br/>Troca contexto tenant

            W->>Redis: Registra job ativo
            W->>Job: start(timestamp)

            Note over W,Job: Sandbox com:<br/>- Error handler<br/>- SIGALRM timeout<br/>- Lock extension

            W->>Job: handle()
            Job-->>W: result

            W->>Job: finish(timestamp)
            W->>Redis: Registra job completo
            W->>RMQ: delete(jobId)
            W->>Redis: releaseLock()

        else Job falhou
            W->>W: handleJobFailure()

            alt Retry disponível
                W->>RMQ: push(payload, retry)
            else Max retries
                W->>RMQ: moveToDLQ()
            end
        end

        alt Sem jobs
            W->>W: sleep(3s)
        end

        W->>W: gc_collect_cycles()
    end
```

## Provisioning Plan

O `ProvisioningPlan` cria supervisors baseado na configuração:

**Arquivo**: `Modules/QueueMaster/MasterSupervisor/Master/ProvisioningPlan.php`

```php
public function deploy(string $environment, string $vhost): void
{
    // Lê config do ambiente
    $supervisorConfigs = config("queue-master.environments.{$environment}");

    foreach ($supervisorConfigs as $queue => $config) {
        // Cria SupervisorOptions
        $options = new SupervisorOptions(
            name: "{$queue}-supervisor",
            queue: $queue,
            connection: $config['connection'],
            maxProcesses: $config['maxProcesses'],
            // ... outros parâmetros
        );

        // Inicia supervisor em processo filho
        $supervisor = new QueueSupervisor($options);
        $supervisor->monitor(); // Loop infinito
    }
}
```

## Autoscaling (Balance Strategies)

### Estratégias Disponíveis

#### 1. Balance: OFF

```php
'balance' => 'off'
```

- Sem autoscaling
- Número fixo de workers (maxProcesses)
- Ideal para desenvolvimento

#### 2. Balance: SIMPLE

```php
'balance' => 'simple'
```

- Scaling básico baseado em carga de fila
- Fórmula: `workers = min(maxProcesses, max(minProcesses, ceil(queueSize / threshold)))`
- Reage a mudanças graduais

#### 3. Balance: AUTO

```php
'balance' => 'auto'
```

- Scaling dinâmico avançado
- Considera:
  - Tamanho da fila
  - Taxa de processamento
  - Latência média
  - Pressão da fila (jobs/consumer)
- Cooldown entre ajustes (evita flapping)
- Ajuste máximo por vez: `maxShift` workers

**Configuração**:

```php
'autoscaling' => [
    'enabled' => true,
    'strategy' => 'time',
    'cooldown' => 3,           // segundos
    'maxShift' => 1,           // workers por ajuste
    'scale_up_threshold' => 3.0,    // jobs/worker
    'scale_down_threshold' => 1.0,
]
```

## Cleanup de Workers Órfãos

Workers órfãos são criados quando:

- Supervisor morre inesperadamente
- Network partition entre supervisor e Redis
- Crash do host

**Detecção**:

```php
// MasterSupervisor.php:87-127
protected function cleanupOrphanedWorkers(): void
{
    // Executa a cada 60s

    $supervisorNames = $this->supervisors->map->getName();

    // Busca workers no Redis que não pertencem a nenhum supervisor ativo
    $registry->cleanupOrphans($supervisorNames);
}
```

**Processo**:

1. Master lista todos supervisors ativos
2. Busca workers registrados no Redis
3. Workers sem supervisor correspondente = órfãos
4. Remove registro do Redis (TTL expira)

## Eventos Disparados

O QueueMaster dispara diversos eventos para integração:

```php
// Master
'master.supervisor.looped'    => MasterSupervisorLooped

// Supervisor
'supervisor.looped'           => SupervisorLooped
'supervisor.parent.died'      => SupervisorParentDied

// Worker
'worker.started'              => WorkerStarting
'worker.stopping'             => WorkerStopping

// Job
'job.started'                 => JobProcessing
'job.completed'               => JobCompleted
'job.failed'                  => JobFailed
'job.finished'                => JobFinished

// Tenant
'tenant.context'              => TenantExtractingContext
'tenant.context.validated'    => TenantContextValidated
```

## Considerações de Performance

### 1. Process Forking

- Cada worker é um processo filho (fork)
- Isolamento total de memória
- Overhead: ~2-5MB por processo
- Limite prático: 100-200 workers por host

### 2. Redis Overhead

- Heartbeat a cada 30s por worker
- Registry updates: ~1KB por worker
- Escalabilidade: até 1000+ workers

### 3. RabbitMQ Connections

- Uma conexão por worker
- Heartbeat AMQP: a cada 60s
- Pool de conexões não implementado

### 4. Memory Management

- Workers auto-reciclam ao atingir limite
- GC forçado após cada job
- Memória não compartilhada entre workers

## Resumo de Arquivos-Chave

| Arquivo | Responsabilidade | LOC |
|---------|------------------|-----|
| `QueueMasterCommand.php` | Entry point, setup inicial | ~105 |
| `MasterSupervisor.php` | Coordenação de supervisors | ~210 |
| `QueueSupervisor.php` | Gerenciamento de pool | ~228 |
| `ProcessPool.php` | Scaling e lifecycle | ~229 |
| `QueueMasterWorkCommand.php` | Worker bootstrap | ~314 |
| `Worker.php` | Execução de jobs | ~1504 |
| `ProvisioningPlan.php` | Deployment de supervisors | ~150 |

## Próximos Documentos

- **[CONFIGURATION.md](CONFIGURATION.md)** - Todos os parâmetros de configuração explicados
- **[MONITORING.md](MONITORING.md)** - Como monitorar cada camada da arquitetura
- **[TROUBLESHOOTING.md](TROUBLESHOOTING.md)** - Debug de problemas em cada componente
