fix(storage): stream job output in real-time with StdoutPipe/StderrPipe

- Replace blocking outBuf/errBuf capture with StdoutPipe+StderrPipe + goroutines
- scanAndEmit runs in parallel for stdout and stderr, emitting each line
  via SSE broadcast and DB append immediately (not after cmd.Run completes)
- Custom scanner split: bytes.TrimRight strips trailing \r so snapraid
  progress lines (e.g. '2%, 57396 MB\r') are stored as clean lines
- scanner.Buffer increased to 128KB max to handle large outputs
- Use sync.WaitGroup to ensure both scanners finish before MarkJobFinished
- Fix PID capture: cmd.Process is nil before cmd.Start(), pass 0 instead
This commit is contained in:
2026-07-07 09:07:08 -04:00
parent d42a0ab412
commit 5b459c0b2e
2 changed files with 54 additions and 37 deletions
+1 -1
View File
@@ -1,5 +1,5 @@
BINARY=nasctl BINARY=nasctl
VERSION?=0.8.1 VERSION?=0.8.2
GO?=go GO?=go
LDFLAGS=-s -w -X github.com/darroyo/nasctl/internal/web.Version=$(VERSION) -X github.com/darroyo/nasctl/internal/web.Commit=$(shell git rev-parse --short HEAD 2>/dev/null || echo unknown) LDFLAGS=-s -w -X github.com/darroyo/nasctl/internal/web.Version=$(VERSION) -X github.com/darroyo/nasctl/internal/web.Commit=$(shell git rev-parse --short HEAD 2>/dev/null || echo unknown)
BUILD_FLAGS=CGO_ENABLED=0 BUILD_FLAGS=CGO_ENABLED=0
+55 -38
View File
@@ -2,14 +2,14 @@ package storage
import ( import (
"bufio" "bufio"
"bytes"
"context" "context"
"encoding/json" "encoding/json"
"fmt" "fmt"
"io"
"log" "log"
"os/exec" "os/exec"
"strings"
"sync" "sync"
"time"
"github.com/darroyo/nasctl/internal/db" "github.com/darroyo/nasctl/internal/db"
) )
@@ -197,57 +197,58 @@ func (jm *JobManager) runJob(job db.StorageJob) {
} }
cmd := exec.CommandContext(ctx, args[0], args[1:]...) cmd := exec.CommandContext(ctx, args[0], args[1:]...)
var outBuf, errBuf bytes.Buffer
cmd.Stdout = &outBuf
cmd.Stderr = &errBuf
// Mark running and get PID stdout, err := cmd.StdoutPipe()
jid := job.ID
pid := 0
if cmd.Process != nil {
pid = cmd.Process.Pid
}
_ = jm.db.MarkJobRunning(jid, pid)
jm.broadcast(jid, Event{Type: "status", Status: "running"})
err = cmd.Run()
// Collect all output
combined := outBuf.String() + errBuf.String()
scanner := bufio.NewScanner(bufio.NewReader(bytes.NewReader([]byte(combined))))
lineCount := 0
flushTicker := time.NewTicker(1 * time.Second)
defer flushTicker.Stop()
for scanner.Scan() {
line := scanner.Text()
lineCount++
jm.logLine(jid, line)
if lineCount%25 == 0 {
_ = jm.db.TrimJobOutput(jid, jm.maxLines)
}
}
exitCode := 0
if err != nil { if err != nil {
if exitErr, ok := err.(*exec.ExitError); ok { jm.failJob(job.ID, -1, fmt.Sprintf("stdout pipe: %v", err))
exitCode = exitErr.ExitCode()
} else {
jm.failJob(jid, exitCode, fmt.Sprintf("execution error: %v", err))
return return
} }
stderr, err := cmd.StderrPipe()
if err != nil {
jm.failJob(job.ID, -1, fmt.Sprintf("stderr pipe: %v", err))
return
} }
jid := job.ID
_ = jm.db.MarkJobRunning(jid, 0)
jm.broadcast(jid, Event{Type: "status", Status: "running"})
var wg sync.WaitGroup
wg.Add(2)
go jm.scanAndEmit(stdout, jid, &wg)
go jm.scanAndEmit(stderr, jid, &wg)
err = cmd.Start()
var exitCode int
if err != nil {
exitCode = -1
jm.failJob(jid, exitCode, fmt.Sprintf("start: %v", err))
wg.Wait()
return
}
_ = cmd.Wait()
wg.Wait()
_ = jm.db.TrimJobOutput(jid, jm.maxLines) _ = jm.db.TrimJobOutput(jid, jm.maxLines)
status := "success" status := "success"
errMsg := "" errMsg := ""
if exitErr, ok := err.(*exec.ExitError); ok {
exitCode = exitErr.ExitCode()
if exitCode != 0 { if exitCode != 0 {
status = "failed" status = "failed"
errMsg = fmt.Sprintf("exit code %d", exitCode) errMsg = fmt.Sprintf("exit code %d", exitCode)
} }
} else if err != nil {
status = "failed"
errMsg = fmt.Sprintf("execution error: %v", err)
exitCode = -1
}
if err := jm.db.MarkJobFinished(jid, status, exitCode, combined, errMsg); err != nil { if err := jm.db.MarkJobFinished(jid, status, exitCode, "", errMsg); err != nil {
log.Printf("mark job finished: %v", err) log.Printf("mark job finished: %v", err)
} }
@@ -264,6 +265,22 @@ func (jm *JobManager) failJob(jobID int64, exitCode int, msg string) {
jm.broadcast(jobID, Event{Type: "end", Status: "failed", ExitCode: exitCode, Message: msg}) jm.broadcast(jobID, Event{Type: "end", Status: "failed", ExitCode: exitCode, Message: msg})
} }
func (jm *JobManager) scanAndEmit(r io.Reader, jobID int64, wg *sync.WaitGroup) {
defer wg.Done()
scanner := bufio.NewScanner(r)
scanner.Buffer(make([]byte, 1024), 128*1024)
lineCount := 0
for scanner.Scan() {
line := scanner.Text()
line = strings.TrimRight(line, "\r")
lineCount++
jm.logLine(jobID, line)
if lineCount%25 == 0 {
_ = jm.db.TrimJobOutput(jobID, jm.maxLines)
}
}
}
func (jm *JobManager) Get(id int64) (db.StorageJob, error) { func (jm *JobManager) Get(id int64) (db.StorageJob, error) {
return jm.db.GetStorageJob(id) return jm.db.GetStorageJob(id)
} }