// backend\disk_guard.go package main import ( "context" "fmt" "net/http" "os" "os/exec" "path/filepath" "strings" "sync/atomic" "time" godisk "github.com/shirou/gopsutil/v3/disk" ) // ------------------------- // Low disk space guard // - pausiert Autostart // - stoppt laufende Downloads // ------------------------- const ( diskGuardInterval = 1 * time.Second ) var diskEmergency int32 // 0=false, 1=true type diskStatusResp struct { Emergency bool `json:"emergency"` PauseGB int `json:"pauseGB"` ResumeGB int `json:"resumeGB"` FreeBytes uint64 `json:"freeBytes"` FreeBytesHuman string `json:"freeBytesHuman"` RecordPath string `json:"recordPath"` } func diskStatusHandler(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet && r.Method != http.MethodHead { http.Error(w, "Nur GET/HEAD erlaubt", http.StatusMethodNotAllowed) return } s := getSettings() pauseGB, resumeGB, _, _, _ := computeDiskThresholds() free, dir, _ := minFreeBytesForDiskGuard(s) resp := diskStatusResp{ Emergency: atomic.LoadInt32(&diskEmergency) == 1, PauseGB: pauseGB, ResumeGB: resumeGB, FreeBytes: free, FreeBytesHuman: formatBytesSI(int64(free)), RecordPath: dir, } w.Header().Set("Cache-Control", "no-store") writeJSON(w, http.StatusOK, resp) } // stopJobsInternal markiert Jobs als "stopping" und cancelt sie (inkl. Preview-FFmpeg Kill). // Nutzt 2 notify-Pushes, damit die UI Phase/Progress sofort sieht. func stopJobsInternal(list []*RecordJob) { if len(list) == 0 { return } type payload struct { cmd *exec.Cmd cancel context.CancelFunc previewCancel context.CancelFunc } pl := make([]payload, 0, len(list)) ids := make([]string, 0, len(list)) nowMs := time.Now().UnixMilli() jobsMu.Lock() for _, job := range list { if job == nil { continue } if job.EndedAt != nil || isTerminalJobStatus(job.Status) || isPostworkJob(job) { continue } job.Phase = "stopping" job.Progress = 10 job.StopRequestedAtMs = nowMs job.StopAttempts++ pl = append(pl, payload{ cmd: job.previewCmd, cancel: job.cancel, previewCancel: job.previewCancel, }) ids = append(ids, job.ID) job.previewCmd = nil job.previewCancel = nil job.LiveThumbStarted = false job.PreviewDir = "" } jobsMu.Unlock() for _, id := range ids { _ = publishJobSnapshot(id) _ = publishModelJobUpsert(id) } for _, p := range pl { if p.previewCancel != nil { p.previewCancel() } if p.cmd != nil && p.cmd.Process != nil { _ = p.cmd.Process.Kill() } if p.cancel != nil { p.cancel() } } for _, id := range ids { _ = publishJobSnapshot(id) _ = publishModelJobUpsert(id) } } func stopAllStoppableJobs() int { stoppable := make([]*RecordJob, 0, 16) jobsMu.RLock() for _, j := range jobs { if j == nil { continue } if j.Status != JobRunning { continue } phase := strings.ToLower(strings.TrimSpace(j.Phase)) // ✅ Im Disk-Notfall ALLES stoppen, was noch schreibt. // Wir skippen nur Jobs, die sowieso schon im "stopping" sind. if phase == "stopping" { continue } stoppable = append(stoppable, j) } jobsMu.RUnlock() stopJobsInternal(stoppable) return len(stoppable) } func sizeOfPathBestEffort(p string) uint64 { p = strings.TrimSpace(p) if p == "" { return 0 } // relativ -> absolut versuchen if !filepath.IsAbs(p) { if abs, err := resolvePathRelativeToApp(p); err == nil && strings.TrimSpace(abs) != "" { p = abs } } fi, err := os.Stat(p) if err != nil || fi.IsDir() || fi.Size() <= 0 { return 0 } return uint64(fi.Size()) } func inFlightBytesForJob(j *RecordJob) uint64 { if j == nil { return 0 } // Dateisystem ist die harte Wahrheit. SizeBytes kann bei laufenden Downloads // verzögert oder gar nicht aktualisiert sein. statSize := sizeOfPathBestEffort(j.Output) liveSize := uint64(0) if j.SizeBytes > 0 { liveSize = uint64(j.SizeBytes) } if statSize > liveSize { return statSize } return liveSize } const giB = uint64(1024 * 1024 * 1024) // computeDiskThresholds: // // Wir brauchen: // 1. die konfigurierte Mindestreserve // 2. zusätzlich ungefähr die Größe aller aktiven/queued TS-Dateien, // weil TS -> MP4 temporär nochmal ungefähr dieselbe Größe benötigt. // // PauseNeed = configReserve + conversionReserve // ResumeNeed = PauseNeed + 3 GiB Hysterese func computeDiskThresholds() (pauseGB int, resumeGB int, relevantInFlight uint64, pauseNeed uint64, resumeNeed uint64) { s := getSettings() configPauseGB := s.LowDiskPauseBelowGB if configPauseGB <= 0 { configPauseGB = 5 } runningBytes := sumActiveRecordingBytes() postworkBytes := reservedPostworkBytes() relevantInFlight = runningBytes + postworkBytes configReserve := uint64(configPauseGB) * giB pauseNeed = configReserve + relevantInFlight resumeNeed = pauseNeed + 3*giB pauseGB = int((pauseNeed + giB - 1) / giB) resumeGB = int((resumeNeed + giB - 1) / giB) if pauseGB > 10_000 { pauseGB = 10_000 } if resumeGB > 10_000 { resumeGB = 10_000 } return } // ✅ Summe der "wachsenden" Daten (running + remuxing etc.) // Idee: Für TS->MP4 Peak brauchst du grob nochmal die Größe der aktuellen Datei als Reserve. func sumInFlightBytes() uint64 { var sum uint64 jobsMu.RLock() defer jobsMu.RUnlock() for _, j := range jobs { if j == nil { continue } if j.Status != JobRunning { continue } sum += inFlightBytesForJob(j) } return sum } func sumActiveRecordingBytes() uint64 { var sum uint64 jobsMu.RLock() defer jobsMu.RUnlock() for _, j := range jobs { if j == nil { continue } // Nur echte aktive Recorder. Postwork wird separat gezählt. if !isActiveRecordingJob(j) { continue } size := inFlightBytesForJob(j) if size == 0 { continue } sum += size } return sum } func shouldReservePostworkJob(j *RecordJob) bool { if j == nil { return false } if isTerminalJobStatus(j.Status) { return false } if !isPostworkJob(j) { return false } if strings.TrimSpace(j.PostWorkKey) == "" { return false } if j.PostWork != nil { st := strings.ToLower(strings.TrimSpace(j.PostWork.State)) if st == "queued" || st == "running" { return true } if j.PostWork.Position > 0 { return true } } state := getEffectivePostworkState(j) return state == "queued" || state == "running" } func reservedPostworkBytes() uint64 { var sum uint64 jobsMu.RLock() defer jobsMu.RUnlock() for _, j := range jobs { if !shouldReservePostworkJob(j) { continue } size := inFlightBytesForJob(j) if size == 0 { continue } sum += size } return sum } func resolveDiskGuardDirs(s RecorderSettings) []string { rawDirs := []string{s.RecordDir, s.DoneDir} seen := map[string]bool{} out := make([]string, 0, len(rawDirs)) for _, raw := range rawDirs { raw = strings.TrimSpace(raw) if raw == "" { continue } dir, _ := resolvePathRelativeToApp(raw) dir = strings.TrimSpace(dir) if dir == "" { dir = raw } if dir == "" { continue } dir = filepath.Clean(dir) key := strings.ToLower(dir) if seen[key] { continue } seen[key] = true out = append(out, dir) } return out } func minFreeBytesForDiskGuard(s RecorderSettings) (uint64, string, bool) { dirs := resolveDiskGuardDirs(s) if len(dirs) == 0 { return 0, "", false } var minFree uint64 var minDir string okAny := false for _, dir := range dirs { u, err := godisk.Usage(dir) if err != nil || u == nil { continue } if !okAny || u.Free < minFree { minFree = u.Free minDir = dir okAny = true } } return minFree, minDir, okAny } // startDiskSpaceGuard läuft im Backend und reagiert auch ohne offenen Browser. // Bei wenig freiem Platz: // - diskEmergency aktivieren (Autostart blockieren) // - laufende Jobs stoppen // // Bei Erholung (Resume-Schwelle): // - diskEmergency automatisch wieder freigeben func startDiskSpaceGuard() { checkOnce := func() { s := getSettings() free, dir, ok := minFreeBytesForDiskGuard(s) if !ok { return } pauseGB, resumeGB, relevantInFlight, pauseNeed, resumeNeed := computeDiskThresholds() wasEmergency := atomic.LoadInt32(&diskEmergency) == 1 isLowForPause := free < pauseNeed isHighEnoughForResume := free >= resumeNeed if !wasEmergency { if !isLowForPause { return } atomic.StoreInt32(&diskEmergency, 1) broadcastAutostartPaused() fmt.Printf( "🛑 [disk] Low space: free=%s (%dB) (< need=%s, %dB, pause=%dGB resume=%dGB, reservedForTS2MP4=%s, %dB) -> stop downloads + block autostart (path=%s)\n", formatBytesSI(u64ToI64(free)), free, formatBytesSI(u64ToI64(pauseNeed)), pauseNeed, pauseGB, resumeGB, formatBytesSI(u64ToI64(relevantInFlight)), relevantInFlight, dir, ) stopped := stopAllStoppableJobs() if stopped > 0 { fmt.Printf("🛑 [disk] Stop requested for %d download job(s)\n", stopped) } return } if isHighEnoughForResume { atomic.StoreInt32(&diskEmergency, 0) broadcastAutostartPaused() fmt.Printf( "✅ [disk] Space recovered: free=%s (%dB) (>= resume=%s, %dB, resume=%dGB, reservedForTS2MP4=%s, %dB) -> unblock autostart (path=%s)\n", formatBytesSI(u64ToI64(free)), free, formatBytesSI(u64ToI64(resumeNeed)), resumeNeed, resumeGB, formatBytesSI(u64ToI64(relevantInFlight)), relevantInFlight, dir, ) } } // Sofort beim Start prüfen, nicht erst nach dem ersten Tick. checkOnce() t := time.NewTicker(diskGuardInterval) defer t.Stop() for range t.C { checkOnce() } } func ensureDiskGuardAllowsStart() error { if atomic.LoadInt32(&diskEmergency) == 1 { return appErrorf("Speicherplatz-Notbremse aktiv: neuer Download wird blockiert") } s := getSettings() free, dir, ok := minFreeBytesForDiskGuard(s) if !ok { return nil } pauseGB, resumeGB, relevantInFlight, pauseNeed, _ := computeDiskThresholds() if free >= pauseNeed { return nil } atomic.StoreInt32(&diskEmergency, 1) broadcastAutostartPaused() stopped := stopAllStoppableJobs() return appErrorf( "Speicherplatz-Notbremse aktiv: frei=%s, benötigt=%s, pause=%dGB, resume=%dGB, reserve=%s, gestoppt=%d, pfad=%s", formatBytesSI(u64ToI64(free)), formatBytesSI(u64ToI64(pauseNeed)), pauseGB, resumeGB, formatBytesSI(u64ToI64(relevantInFlight)), stopped, dir, ) }