platform/app: stream SAT job output live and kick blackbox on job completion

A crash mid-command (e.g. the nvbandwidth reboot) previously lost that job's entire output: streamExecOutput only buffered stdout/stderr in memory and the job's log file was written once, after the process exited. It now also streams each line straight to that file as it arrives, so whatever printed before a crash survives. Root filesystem here is a tmpfs overlay (toram boot), so the only real persistence boundary is blackbox's mirror to removable media, not the local write itself. platform.SetJobBoundaryHook lets app wire a touch of a small kick-file after each job's output is written; blackboxWorker now polls that file's mtime alongside its normal adaptive timer and syncs immediately on a kick instead of waiting out the current flush period (up to 30s).
2026-07-27 17:40:22 +03:00
parent ced2175fb0
commit 781cf5dcbf
5 changed files with 365 additions and 24 deletions
@@ -70,6 +70,19 @@ const (
 	SATEstimatedNvidiaBandwidthSec = 2700
 )

+// satJobBoundaryHook, if set, is called with a SAT job's log file name right
+// after that job's output has been written to disk — a natural point for an
+// external blackbox sync to pick up newly-completed data promptly instead of
+// waiting out its own adaptive schedule. Nil by default (no-op); set once
+// via SetJobBoundaryHook by whichever process wires up blackbox.
+var satJobBoundaryHook func(jobName string)
+
+// SetJobBoundaryHook installs the callback invoked after each SAT job
+// finishes and its log file has been written. Pass nil to clear it.
+func SetJobBoundaryHook(hook func(jobName string)) {
+	satJobBoundaryHook = hook
+}
+
 var (
 	satExecCommand  = exec.Command
 	satLookPath     = exec.LookPath
@@ -98,21 +111,39 @@ var (
 )

 // streamExecOutput runs cmd and streams each output line to logFunc (if non-nil).
+// If livePath is non-empty, each line is also appended to that file as it
+// arrives — so the command's own output already exists on disk (and is thus
+// pickable up by a concurrent blackbox sync) while it's still running,
+// instead of only appearing once the whole pack finishes writing the final
+// job file. Best-effort: a failure to open/write livePath never fails the job.
 // Returns combined stdout+stderr as a byte slice.
-func streamExecOutput(cmd *exec.Cmd, logFunc func(string)) ([]byte, error) {
+func streamExecOutput(cmd *exec.Cmd, logFunc func(string), livePath string) ([]byte, error) {
 	pr, pw := io.Pipe()
 	cmd.Stdout = pw
 	cmd.Stderr = pw

+	var liveFile *os.File
+	if livePath != "" {
+		if f, err := os.OpenFile(livePath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0644); err == nil {
+			liveFile = f
+		}
+	}
+
 	var buf bytes.Buffer
 	var wg sync.WaitGroup
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
+		if liveFile != nil {
+			defer liveFile.Close()
+		}
 		scanner := bufio.NewScanner(pr)
 		for scanner.Scan() {
 			line := scanner.Text()
 			buf.WriteString(line + "\n")
+			if liveFile != nil {
+				_, _ = liveFile.WriteString(line + "\n")
+			}
 			if logFunc != nil {
 				logFunc(line)
 			}
@@ -787,11 +818,15 @@ func (s *System) RunStorageAcceptancePack(ctx context.Context, baseDir string, e
 				break
 			}
 			name := fmt.Sprintf("%s-%02d-%s.log", prefix, cmdIndex+1, job.name)
-			out, err := runSATCommandCtx(ctx, verboseLog, job.name, job.cmd, nil, logFunc)
+			livePath := filepath.Join(runDir, name)
+			out, err := runSATCommandCtx(ctx, verboseLog, job.name, job.cmd, nil, logFunc, livePath)
 			deviceOutputs[job.name] = out
-			if writeErr := os.WriteFile(filepath.Join(runDir, name), out, 0644); writeErr != nil {
+			if writeErr := os.WriteFile(livePath, out, 0644); writeErr != nil {
 				return "", writeErr
 			}
+			if satJobBoundaryHook != nil {
+				satJobBoundaryHook(name)
+			}
 			status, rc := classifySATResult(job.name, out, err)
 			stats.Add(status)
 			key := filepath.Base(devPath) + "_" + strings.ReplaceAll(job.name, "-", "_")
@@ -985,7 +1020,7 @@ func runAcceptancePackCtx(ctx context.Context, baseDir, prefix string, jobs []sa
 				if job.collectGPU {
 					out, err = runSATCommandWithMetrics(ctx, verboseLog, job.name, cmd, job.env, job.gpuIndices, runDir, logFunc)
 				} else {
-					out, err = runSATCommandCtx(ctx, verboseLog, job.name, cmd, job.env, logFunc)
+					out, err = runSATCommandCtx(ctx, verboseLog, job.name, cmd, job.env, logFunc, filepath.Join(runDir, job.name))
 				}
 				if err == nil || attempt >= job.retries || ctx.Err() != nil {
 					break
@@ -1015,6 +1050,9 @@ func runAcceptancePackCtx(ctx context.Context, baseDir, prefix string, jobs []sa
 		if writeErr := os.WriteFile(filepath.Join(runDir, job.name), out, 0644); writeErr != nil {
 			return "", writeErr
 		}
+		if satJobBoundaryHook != nil {
+			satJobBoundaryHook(job.name)
+		}
 		if ctx.Err() != nil {
 			return "", ctx.Err()
 		}
@@ -1237,7 +1275,11 @@ func parseNvidiaGPUHealth(raw string) []nvidiaGPUHealth {
 	return gpus
 }

-func runSATCommandCtx(ctx context.Context, verboseLog, name string, cmd []string, env []string, logFunc func(string)) ([]byte, error) {
+// runSATCommandCtx runs cmd and returns its combined output. livePath is
+// variadic purely so existing callers are unaffected: pass a path (job's
+// output file) to also stream output to disk live as it runs, so a crash
+// mid-command leaves whatever had printed so far instead of nothing at all.
+func runSATCommandCtx(ctx context.Context, verboseLog, name string, cmd []string, env []string, logFunc func(string), livePath ...string) ([]byte, error) {
 	start := time.Now().UTC()
 	resolvedCmd, err := resolveSATCommand(cmd)
 	appendSATVerboseLog(verboseLog,
@@ -1268,7 +1310,11 @@ func runSATCommandCtx(ctx context.Context, verboseLog, name string, cmd []string
 	if len(env) > 0 {
 		c.Env = append(os.Environ(), env...)
 	}
-	out, err := streamExecOutput(c, logFunc)
+	var live string
+	if len(livePath) > 0 {
+		live = livePath[0]
+	}
+	out, err := streamExecOutput(c, logFunc, live)

 	rc := 0
 	if err != nil {
@@ -1437,7 +1483,7 @@ func runSATCommand(verboseLog, name string, cmd []string, logFunc func(string))
 		return []byte(err.Error() + "\n"), err
 	}

-	out, err := streamExecOutput(satExecCommand(resolvedCmd[0], resolvedCmd[1:]...), logFunc)
+	out, err := streamExecOutput(satExecCommand(resolvedCmd[0], resolvedCmd[1:]...), logFunc, "")

 	rc := 0
 	if err != nil {
@@ -1607,7 +1653,7 @@ func runSATCommandWithMetrics(ctx context.Context, verboseLog, name string, cmd
 		}
 	}()

-	out, err := runSATCommandCtx(ctx, verboseLog, name, cmd, env, logFunc)
+	out, err := runSATCommandCtx(ctx, verboseLog, name, cmd, env, logFunc, filepath.Join(runDir, name))

 	close(stopCh)
 	<-doneCh
@@ -0,0 +1,97 @@
+package platform
+
+import (
+	"context"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"testing"
+)
+
+func TestStreamExecOutputWritesLiveFileIncrementally(t *testing.T) {
+	dir := t.TempDir()
+	livePath := filepath.Join(dir, "job.log")
+
+	cmd := satExecCommand("printf", "line1\nline2\n")
+	out, err := streamExecOutput(cmd, nil, livePath)
+	if err != nil {
+		t.Fatalf("streamExecOutput error: %v", err)
+	}
+	if string(out) != "line1\nline2\n" {
+		t.Fatalf("out=%q want %q", out, "line1\nline2\n")
+	}
+	got, err := os.ReadFile(livePath)
+	if err != nil {
+		t.Fatalf("ReadFile(livePath): %v", err)
+	}
+	if string(got) != "line1\nline2\n" {
+		t.Fatalf("livePath content=%q want %q", got, "line1\nline2\n")
+	}
+}
+
+func TestStreamExecOutputSkipsLiveFileWhenPathEmpty(t *testing.T) {
+	cmd := satExecCommand("printf", "line1\n")
+	out, err := streamExecOutput(cmd, nil, "")
+	if err != nil {
+		t.Fatalf("streamExecOutput error: %v", err)
+	}
+	if string(out) != "line1\n" {
+		t.Fatalf("out=%q want %q", out, "line1\n")
+	}
+}
+
+func TestRunSATCommandCtxWritesLivePath(t *testing.T) {
+	dir := t.TempDir()
+	verboseLog := filepath.Join(dir, "verbose.log")
+	livePath := filepath.Join(dir, "job.log")
+
+	oldExecCommand := satExecCommand
+	satExecCommand = func(name string, args ...string) *exec.Cmd {
+		return exec.Command(name, args...)
+	}
+	t.Cleanup(func() { satExecCommand = oldExecCommand })
+
+	out, err := runSATCommandCtx(context.Background(), verboseLog, "job", []string{"printf", "hello\n"}, nil, nil, livePath)
+	if err != nil {
+		t.Fatalf("runSATCommandCtx error: %v", err)
+	}
+	if string(out) != "hello\n" {
+		t.Fatalf("out=%q want %q", out, "hello\n")
+	}
+	got, err := os.ReadFile(livePath)
+	if err != nil {
+		t.Fatalf("ReadFile(livePath): %v", err)
+	}
+	if string(got) != "hello\n" {
+		t.Fatalf("livePath content=%q want %q", got, "hello\n")
+	}
+}
+
+func TestRunAcceptancePackCtxInvokesJobBoundaryHook(t *testing.T) {
+	old := satJobBoundaryHook
+	t.Cleanup(func() { satJobBoundaryHook = old })
+
+	var seen []string
+	SetJobBoundaryHook(func(jobName string) {
+		seen = append(seen, jobName)
+	})
+
+	oldExecCommand := satExecCommand
+	satExecCommand = func(name string, args ...string) *exec.Cmd {
+		return exec.Command(name, args...)
+	}
+	t.Cleanup(func() { satExecCommand = oldExecCommand })
+
+	dir := t.TempDir()
+	_, err := runAcceptancePackCtx(context.Background(), dir, "test-pack", []satJob{
+		{name: "01-a.log", cmd: []string{"printf", "a\n"}},
+		{name: "02-b.log", cmd: []string{"printf", "b\n"}},
+	}, nil)
+	if err != nil {
+		t.Fatalf("runAcceptancePackCtx error: %v", err)
+	}
+
+	if len(seen) != 2 || seen[0] != "01-a.log" || seen[1] != "02-b.log" {
+		t.Fatalf("seen=%v want [01-a.log 02-b.log]", seen)
+	}
+}