Lotes Atómicos & Resurrección de Fallos

Para tareas masivas de alto impacto (importaciones de nómina, exportaciones masivas, procesamientos analíticos), Mañuti implementa el bus de procesamiento por lotes atómicos (Illuminate\Bus\Batch) combinado con un protocolo de resiliencia forense para recuperar trabajos fallidos sin pérdida de datos.

1. Arquitectura de Estado en job_batches (MariaDB)

Mientras que los cientos o miles de micro-trabajos viajan por la memoria de Redis, el estado consolidado del lote reside en MariaDB para resistir caídas de energía o reinicios de servicio:

CREATE TABLE `job_batches` (
    `id` VARCHAR(255) PRIMARY KEY,
    `name` VARCHAR(255) NOT NULL,
    `total_jobs` INT NOT NULL,
    `pending_jobs` INT NOT NULL,
    `failed_jobs` INT NOT NULL,
    `failed_job_ids` LONGTEXT NOT NULL,
    `options` MEDIUMTEXT NULL,
    `cancelled_at` INT NULL,
    `created_at` INT NOT NULL,
    `finished_at` INT NULL
);

Métricas de Seguimiento en Tiempo Real

• total_jobs: Volumen inicial del lote.
• pending_jobs: Contador decreciente. Redis ejecuta un decremento atómico en SQL cada vez que un worker finaliza una unidad de trabajo.
• failed_jobs: Contador de excepciones no controladas ocurridas dentro del lote.

2. Implementación de Trabajos Batchable

Los trabajos individuales deben utilizar el trait Illuminate\Bus\Batchable:

// app/Jobs/ImportEmployeeJob.php
namespace App\Jobs;

use App\Models\Usuario;
use Illuminate\Bus\Batchable;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Support\Facades\Hash;

class ImportEmployeeJob implements ShouldQueue
{
    use Batchable, Dispatchable, InteractsWithQueue, Queueable;

    public function __construct(public array $employeeData) {}

    public function handle(): void
    {
        // 1. Verificación de Cancelación: Si el administrador canceló el lote, abortar
        if ($this->batch()->cancelled()) {
            return;
        }

        // 2. Transacción Atómica
        Usuario::create([
            'nombres'          => $this->employeeData['nombre'],
            'apellido_paterno' => $this->employeeData['apellido'],
            'correo'           => $this->employeeData['correo'],
            'usuario'          => $this->employeeData['usuario'],
            'password'         => Hash::make($this->employeeData['password_temporal']),
            'cat_status'       => 1, // Activo
        ]);
    }
}

Despacho y Orquestación en el Controlador

use Illuminate\Support\Facades\Bus;
use App\Jobs\ImportEmployeeJob;

$jobs = [];
foreach ($csvRows as $row) {
    $jobs[] = new ImportEmployeeJob($row);
}

$batch = Bus::batch($jobs)
    ->name('Importación Nómina ' . now()->format('Y-m-d H:i'))
    ->allowFailures()       // El lote continúa aunque una fila específica falle
    ->onQueue('imports')     // Se despacha a una cola dedicada en Redis
    ->dispatch();

return response()->json([
    'batch_id' => $batch->id,
    'total'    => $batch->totalJobs,
]);

3. Repositorio Forense de Excepciones (failed_jobs)

Cuando un trabajo agota sus intentos configurados ($tries = 3), Horizon lo extrae de la RAM de Redis y lo deposita en la tabla failed_jobs:

4. Protocolo de Resurrección Manual (Disaster Recovery)

A diferencia de los reintentos automáticos inmediatos en memoria, los trabajos alojados en failed_jobs permanecen inactivos esperando intervención del equipo de SRE/DevOps.

flowchart TD
    FALLO["[Fallo Definitivo]"]
    DB["Trabajo almacenado en MariaDB 'failed_jobs'"]
    
    NET_RESTORE["[Servicio restablecido]"]
    HOTFIX["1. Despliegue de hotfix (git pull)<br/>2. php artisan horizon:terminate"]
    
    RETRY["[php artisan queue:retry {uuid} / all]"]
    
    STEPS["1. Lee payload original de MariaDB<br/>2. Elimina registro de 'failed_jobs'<br/>3. Reinyecta el Job a la cola viva en Redis<br/>4. Horizon reinicia contador de intentos a 0"]

    FALLO --> DB
    
    DB -->|"Error de Red / Timeout"| NET_RESTORE
    DB -->|"Error de Lógica / Bug en Código"| HOTFIX
    
    NET_RESTORE --> RETRY
    HOTFIX --> RETRY
    
    RETRY --> STEPS

Comandos de Operación

• Reintentar un trabajo específico por UUID:

  php artisan queue:retry 00139847-e6eb-4423-a07c-8e8e6383aa30
  

• Reintentar todos los trabajos fallidos:

  php artisan queue:retry all
  

• Eliminar un trabajo corrupto irrecuperable:

  php artisan queue:forget 00139847-e6eb-4423-a07c-8e8e6383aa30
  

• Depuración programada de trabajos fallidos antiguos (Scheduler):

  // routes/console.php
  Schedule::command('queue:prune-failed --hours=168')->daily(); // Purgar fallos > 7 días
  

Enlaces Relacionados

• Colas Híbridas & Orquestación Horizon
• Worker Asíncrono Python & Graphviz
• Disaster Recovery & Depuración Atómica