platform/app: fix nvbandwidth split fallback, blackbox discovery churn, and add blocking sync brackets around load steps

- gpuBandwidthSocketGroups: a single GPU whose NUMA node fails to resolve
  no longer collapses the whole per-socket nvbandwidth split into one
  fallback pass — it now folds into the last resolved group instead,
  preserving isolation for the sockets that did resolve.
- blackbox discoverMarkedTargets: skip mounting/unmounting devices that
  already have a running worker on every 2s discovery tick. This was
  observed hammering the same USB target continuously (mount+unmount
  every ~2s for the whole session) and contending with the worker's own
  sync cycle, plausibly explaining multi-minute sync cycles seen on a
  real crash bundle.
- syncFilesystem now calls syscall.Sync() directly instead of spawning
  /bin/sync per copied file; blackbox mounts removable targets with
  -o sync so writes are durable without relying on the app-level sync as
  the primary mechanism.
- New platform.SetSyncBracketHook / satJob.syncBracket: blocks (with a
  bounded timeout) on blackbox actually reaching removable media right
  before and right after a diagnostic's real load step (nvbandwidth,
  memtester, stress-ng, dcgmi diag, nccl, smartctl/nvme self-test...),
  instead of only firing a fire-and-forget kick after the job's own log
  file is written. A crash mid-load now has durable evidence the load
  started, not just whatever streamed to the RAM-backed export dir before
  blackbox's next scheduled cycle.

Found investigating a real support bundle where blackbox's last
successful sync (19:55:25) predated both the previous job finishing and
the crashing nvbandwidth job starting (19:56:57) — none of the crash
window ever reached durable media.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Mikhail Chusavitin
2026-07-28 14:22:05 +03:00
co-authored by Claude Sonnet 5
parent 6d13c17d36
commit e03267a72f
8 changed files with 438 additions and 51 deletions
+124 -6
View File
@@ -15,6 +15,7 @@ import (
"sort"
"strings"
"sync"
"syscall"
"time"
"bee/audit/internal/platform"
@@ -33,6 +34,13 @@ const (
// output reaches removable media promptly instead of waiting out
// whatever the current adaptive flush period happens to be.
blackboxKickFileName = ".blackbox-kick"
// blackboxSyncBracketTimeout bounds how long a syncBracket-marked SAT job
// (see platform.SetSyncBracketHook) blocks waiting for blackbox to catch
// up before/after the actual load step. Generous relative to the normal
// sub-second kick-to-sync latency, but bounded so a genuinely stuck or
// unplugged target can't hang the diagnostic indefinitely.
blackboxSyncBracketTimeout = 20 * time.Second
)
// blackboxKickPollInterval is how often an idle worker checks the kick file
@@ -53,6 +61,64 @@ func requestBlackboxSync(exportDir string) {
_ = os.WriteFile(path, []byte(now.Format(time.RFC3339Nano)+"\n"), 0644)
}
// blackboxSyncWaitPollInterval is how often requestBlackboxSyncAndWait
// re-reads the state file while waiting for a sync to catch up. A package
// var so tests can shrink it.
var blackboxSyncWaitPollInterval = 200 * time.Millisecond
// requestBlackboxSyncAndWait kicks every enrolled blackbox target and blocks
// until each one has completed a sync cycle that started at or after this
// call, or until timeout elapses. Use this (instead of the fire-and-forget
// requestBlackboxSync) around an event where losing the data to a crash
// before it reaches removable media would matter — e.g. bracket a risky load
// step so its own SAT job directory (and the fact that it started at all) is
// durable on the flash drive before the load runs, and durable again once it
// finishes.
//
// Returns nil once caught up. Returns an error naming what didn't catch up
// (timeout, or a target stuck "degraded") — callers should treat that as
// best-effort informational (log it) rather than fail the job over it: a
// blackbox problem should not block the diagnostic the operator actually
// asked for.
func requestBlackboxSyncAndWait(exportDir, statePath string, timeout time.Duration) error {
requestedAt := blackboxNow()
requestBlackboxSync(exportDir)
deadline := time.Now().Add(timeout)
for {
state, err := ReadBlackboxState(statePath)
if err == nil {
if ok, pending := blackboxStateCaughtUp(state, requestedAt); ok {
return nil
} else if time.Now().After(deadline) {
return fmt.Errorf("blackbox sync did not catch up within %s (pending: %s)", timeout, pending)
}
} else if time.Now().After(deadline) {
return fmt.Errorf("blackbox sync wait: could not read state after %s: %w", timeout, err)
}
time.Sleep(blackboxSyncWaitPollInterval)
}
}
// blackboxStateCaughtUp reports whether every non-degraded enrolled target
// has synced at or after requestedAt. No targets enrolled counts as caught
// up (nothing to wait for — e.g. no removable media plugged in). A target
// stuck "degraded" (mount/copy failing) is skipped rather than waited on
// forever; its enrollment ID is named in the returned pending string so
// callers can log which target is the reason a real wait timed out.
func blackboxStateCaughtUp(state BlackboxState, requestedAt time.Time) (bool, string) {
for _, t := range state.Targets {
if t.Status == "degraded" {
continue
}
syncedAt, err := time.Parse(time.RFC3339Nano, t.LastSyncAtUTC)
if err != nil || syncedAt.Before(requestedAt) {
return false, t.EnrollmentID
}
}
return true, ""
}
// blackboxKickModTime returns the kick file's mtime, or the zero Time if it
// doesn't exist yet (nothing has requested a sync since boot).
func blackboxKickModTime(exportDir string) time.Time {
@@ -270,7 +336,7 @@ func DisableBlackboxTarget(device, enrollmentID string) error {
}
func (rt *blackboxRuntime) reconcile() {
discovered, _ := rt.discoverMarkedTargets()
discovered, _ := rt.discoverMarkedTargets(rt.trackedDevices())
rt.mu.Lock()
defer rt.mu.Unlock()
@@ -311,7 +377,35 @@ func (rt *blackboxRuntime) stopAll() {
}
}
func (rt *blackboxRuntime) discoverMarkedTargets() ([]discoveredBlackboxTarget, error) {
// trackedDevices returns the already-enrolled targets, keyed by device path,
// for every worker currently running. Used by discoverMarkedTargets to skip
// re-mounting devices a worker already owns.
func (rt *blackboxRuntime) trackedDevices() map[string]discoveredBlackboxTarget {
rt.mu.Lock()
defer rt.mu.Unlock()
known := make(map[string]discoveredBlackboxTarget, len(rt.workers))
for _, worker := range rt.workers {
worker.mu.Lock()
if worker.target.Device != "" {
known[worker.target.Device] = discoveredBlackboxTarget{
marker: worker.marker,
target: worker.target,
}
}
worker.mu.Unlock()
}
return known
}
// discoverMarkedTargets probes removable media for the bee-blackbox marker.
// known holds devices a worker is already running against (see
// trackedDevices) — those are reported back as-is, without mounting, since
// mounting/unmounting the same device on every discovery tick (every
// blackboxDiscoverInterval) fights the worker's own mount for the device and
// was observed to slow its actual sync cycle by an order of magnitude on
// FUSE-backed filesystems (NTFS via ntfs-3g). Only devices with no running
// worker get probed.
func (rt *blackboxRuntime) discoverMarkedTargets(known map[string]discoveredBlackboxTarget) ([]discoveredBlackboxTarget, error) {
targets, err := rt.system.ListRemovableTargets()
if err != nil {
return nil, err
@@ -322,6 +416,11 @@ func (rt *blackboxRuntime) discoverMarkedTargets() ([]discoveredBlackboxTarget,
if target.Device == "" {
continue
}
if cached, ok := known[target.Device]; ok {
cached.target = target
out = append(out, cached)
continue
}
mountpoint, mountedByBee, err := ensureMountedTarget(target, "probe")
if err != nil {
continue
@@ -578,7 +677,11 @@ func (rt *blackboxRuntime) persistStateLocked() {
Mountpoint: worker.mountpoint,
}
if !worker.lastSyncAt.IsZero() {
targetState.LastSyncAtUTC = worker.lastSyncAt.Format(time.RFC3339)
// Nanosecond precision, not RFC3339's default seconds — a sync
// that lands in the same wall-clock second as a kick request
// must still compare as "after" it (see blackboxStateCaughtUp),
// which second-only precision can get wrong.
targetState.LastSyncAtUTC = worker.lastSyncAt.Format(time.RFC3339Nano)
}
if worker.lastDuration > 0 {
targetState.LastCycleDuration = worker.lastDuration.String()
@@ -647,7 +750,15 @@ func ensureMountedTarget(target platform.RemovableTarget, suffix string) (mountp
if err := os.MkdirAll(mountpoint, 0755); err != nil {
return "", false, err
}
if raw, err := blackboxExecCommand("mount", target.Device, mountpoint).CombinedOutput(); err != nil {
// -o sync makes every write to the target synchronous at the VFS layer
// (no page-cache write-back to lose on a hard reset) instead of relying
// solely on the explicit syscall.Sync() calls in writeFileAtomic/
// unmountTarget to flush it after the fact. Those explicit syncs stay in
// place as a fallback (e.g. for target.Mountpoint above, an
// already-mounted filesystem we don't control the options of) — with -o
// sync already doing the work, they become a fast no-op most of the time
// instead of the primary durability mechanism.
if raw, err := blackboxExecCommand("mount", "-o", "sync", target.Device, mountpoint).CombinedOutput(); err != nil {
return "", false, formatBlackboxMountTargetError(target, string(raw), err)
}
if err := ensureWritableBlackboxMountpoint(mountpoint); err != nil {
@@ -658,7 +769,7 @@ func ensureMountedTarget(target platform.RemovableTarget, suffix string) (mountp
}
func unmountTarget(mountpoint string) error {
_ = blackboxExecCommand("sync").Run()
syscall.Sync()
raw, err := blackboxExecCommand("umount", mountpoint).CombinedOutput()
if err != nil {
msg := strings.TrimSpace(string(raw))
@@ -812,8 +923,15 @@ func writeFileAtomic(path string, data []byte, perm os.FileMode) error {
return syncFilesystem(filepath.Dir(path))
}
// syncFilesystem flushes pending writes to durable storage. Uses the sync(2)
// syscall directly rather than spawning /bin/sync — writeFileAtomic calls
// this once per copied file (syncDirectoryTree can touch hundreds of files
// per cycle), and syscall.Sync() does the same flush without a fork+exec per
// call. path is unused (sync(2) always flushes system-wide; kept as a
// parameter for call-site clarity about which tree the caller cares about).
func syncFilesystem(path string) error {
return blackboxExecCommand("sync").Run()
syscall.Sync()
return nil
}
func ensureWritableBlackboxMountpoint(mountpoint string) error {