fix: cancelacion forzosa inmediata de FFmpeg y tracking robusto por PID

internal/ffmpeg/ffmpeg.go

@@ -270,8 +270,6 @@ func wmOverlayExpr(pos string) string {
 }
 
 // Compress ejecuta la compresión usando FFmpeg y reporta el progreso en vivo.
-// wmImagePath es la ruta local a un logo de marca de agua (opcional); si se
-// indica, tiene prioridad sobre la marca de agua de texto.
 func Compress(
 	ctx context.Context,
 	inputPath, outputPath string,
@@ -279,6 +277,19 @@ func Compress(
 	wmImagePath string,
 	totalDuration float64,
 	onProgress func(p protocol.TaskProgress),
+) error {
+	return CompressWithCmd(ctx, inputPath, outputPath, settings, wmImagePath, totalDuration, nil, onProgress)
+}
+
+// CompressWithCmd ejecuta la compresión y permite capturar el puntero a *exec.Cmd para terminación forzosa inmediata
+func CompressWithCmd(
+	ctx context.Context,
+	inputPath, outputPath string,
+	settings protocol.VideoSettings,
+	wmImagePath string,
+	totalDuration float64,
+	onCmd func(cmd *exec.Cmd),
+	onProgress func(p protocol.TaskProgress),
 ) error {
 	settings = SanitizeSettings(settings)
 
@@ -388,6 +399,13 @@ func Compress(
 	)
 
 	cmd := exec.CommandContext(ctx, "ffmpeg", args...)
+	cmd.Cancel = func() error {
+		if cmd.Process != nil {
+			return cmd.Process.Kill()
+		}
+		return nil
+	}
+	cmd.WaitDelay = 1 * time.Second
 
 	stderr, err := cmd.StderrPipe()
 	if err != nil {
@@ -398,6 +416,22 @@ func Compress(
 		return fmt.Errorf("error iniciando ffmpeg: %w", err)
 	}
 
+	if onCmd != nil {
+		onCmd(cmd)
+	}
+
+	stopWatch := make(chan struct{})
+	defer close(stopWatch)
+	go func() {
+		select {
+		case <-ctx.Done():
+			if cmd.Process != nil {
+				_ = cmd.Process.Kill()
+			}
+		case <-stopWatch:
+		}
+	}()
+
 	scanner := bufio.NewScanner(stderr)
 	// FFmpeg usa retorno de carro '\r' para actualizar la línea de estado
 	scanner.Split(scanCRLFOrCR)

main.go

@@ -12,6 +12,7 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"os/exec"
 	"os/signal"
 	"path/filepath"
 	"runtime"
@@ -42,6 +43,7 @@ type CurrentJob struct {
 	FileName  string
 	UploadURL string
 	Cancel    context.CancelFunc
+	Cmd       *exec.Cmd
 	Done      bool
 }
 
@@ -53,7 +55,7 @@ type Worker struct {
 
 	conn       *websocket.Conn
 	connMu     sync.Mutex
-	currentJob *CurrentJob
+	activeJobs map[string]*CurrentJob
 	jobMu      sync.RWMutex
 
 	connected bool
@@ -118,6 +120,7 @@ func main() {
 		managerURL: managerURL,
 		secret:     secret,
 		workDir:    workDir,
+		activeJobs: make(map[string]*CurrentJob),
 	}
 
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
@@ -285,15 +288,15 @@ func (w *Worker) run(ctx context.Context) error {
 	// 1. Enviar Handshake (incluyendo cualquier tarea activa si reconecta tras reinicio del Manager)
 	w.jobMu.RLock()
 	var activeTasks []protocol.TaskState
-	if w.currentJob != nil {
+	for _, j := range w.activeJobs {
 		status := "converting"
-		if w.currentJob.Done {
+		if j.Done {
 			status = "completed"
 		}
 		activeTasks = append(activeTasks, protocol.TaskState{
-			TaskID:    w.currentJob.TaskID,
+			TaskID:    j.TaskID,
 			Status:    status,
-			UploadURL: w.currentJob.UploadURL,
+			UploadURL: j.UploadURL,
 		})
 	}
 	w.jobMu.RUnlock()
@@ -359,19 +362,37 @@ func (w *Worker) run(ctx context.Context) error {
 		case protocol.TypeTaskAssign:
 			var task protocol.TaskAssign
 			if err := json.Unmarshal(msg.Payload, &task); err == nil {
-				go w.executeTask(ctx, task)
+				taskCtx, cancel := context.WithCancel(ctx)
+				job := &CurrentJob{
+					TaskID:    task.TaskID,
+					FileName:  task.FileName,
+					UploadURL: task.UploadURL,
+					Cancel:    cancel,
+				}
+				w.jobMu.Lock()
+				w.activeJobs[task.TaskID] = job
+				w.jobMu.Unlock()
+
+				go w.executeTask(taskCtx, job, task)
 			}
 
 		case protocol.TypeTaskCancel:
 			var cancel protocol.TaskCancel
 			if err := json.Unmarshal(msg.Payload, &cancel); err == nil {
 				w.jobMu.Lock()
-				if w.currentJob != nil && w.currentJob.TaskID == cancel.TaskID {
-					log.Printf("⛔ Cancelando tarea %s...", cancel.TaskID)
-					if w.currentJob.Cancel != nil {
-						w.currentJob.Cancel()
+				job, exists := w.activeJobs[cancel.TaskID]
+				if exists {
+					log.Printf("⛔ [Worker] Cancelando tarea %s...", cancel.TaskID)
+					if job.Cancel != nil {
+						job.Cancel()
 					}
-					w.currentJob = nil
+					if job.Cmd != nil && job.Cmd.Process != nil {
+						log.Printf("💀 [Worker] Matando proceso FFmpeg PID %d de tarea %s...", job.Cmd.Process.Pid, cancel.TaskID)
+						_ = job.Cmd.Process.Kill()
+					}
+					delete(w.activeJobs, cancel.TaskID)
+				} else {
+					log.Printf("⚠️ [Worker] Tarea %s no encontrada en activeJobs para cancelar", cancel.TaskID)
 				}
 				w.jobMu.Unlock()
 			}
@@ -379,25 +400,16 @@ func (w *Worker) run(ctx context.Context) error {
 	}
 }
 
-func (w *Worker) executeTask(parentCtx context.Context, task protocol.TaskAssign) {
+func (w *Worker) executeTask(taskCtx context.Context, job *CurrentJob, task protocol.TaskAssign) {
 	log.Printf("📥 Nueva tarea recibida: %s (Archivo: %s)", task.TaskID, task.FileName)
 
-	taskCtx, cancel := context.WithCancel(parentCtx)
-	w.jobMu.Lock()
-	w.currentJob = &CurrentJob{
-		TaskID:    task.TaskID,
-		FileName:  task.FileName,
-		UploadURL: task.UploadURL,
-		Cancel:    cancel,
-		Done:      false,
-	}
-	w.jobMu.Unlock()
-
 	defer func() {
 		w.jobMu.Lock()
-		w.currentJob = nil
+		delete(w.activeJobs, task.TaskID)
 		w.jobMu.Unlock()
-		cancel()
+		if job.Cancel != nil {
+			job.Cancel()
+		}
 	}()
 
 	ext := filepath.Ext(task.FileName)
@@ -482,21 +494,31 @@ func (w *Worker) executeTask(parentCtx context.Context, task protocol.TaskAssign
 	log.Printf("[%s] Iniciando compresión con FFmpeg (Duración: %.1fs)...", task.TaskID, duration)
 	startTime := time.Now()
 
-	err = ffmpeg.Compress(taskCtx, inputPath, outputPath, task.Settings, wmImagePath, duration, func(p protocol.TaskProgress) {
+	err = ffmpeg.CompressWithCmd(taskCtx, inputPath, outputPath, task.Settings, wmImagePath, duration, func(cmd *exec.Cmd) {
+		w.jobMu.Lock()
+		job.Cmd = cmd
+		w.jobMu.Unlock()
+		if cmd.Process != nil {
+			log.Printf("[%s] Proceso FFmpeg iniciado con PID %d", task.TaskID, cmd.Process.Pid)
+		}
+	}, func(p protocol.TaskProgress) {
 		p.TaskID = task.TaskID
 		_ = w.sendMessage(protocol.TypeTaskProgress, p)
 	})
 
 	if err != nil {
+		if taskCtx.Err() != nil {
+			log.Printf("[%s] Compresión cancelada por solicitud del usuario/manager.", task.TaskID)
+			w.reportFailed(task.TaskID, "compresión cancelada")
+			return
+		}
 		log.Printf("[%s] Error en compresión: %v", task.TaskID, err)
 		w.reportFailed(task.TaskID, fmt.Sprintf("falló compresión: %v", err))
 		return
 	}
 
 	w.jobMu.Lock()
-	if w.currentJob != nil {
-		w.currentJob.Done = true
-	}
+	job.Done = true
 	w.jobMu.Unlock()
 
 	outStat, err := os.Stat(outputPath)