From cc2befe093180d8f43792c4ee5f4821f5193204b Mon Sep 17 00:00:00 2001 From: Linrador <68631622+Linrador@users.noreply.github.com> Date: Fri, 22 May 2026 07:53:49 +0200 Subject: [PATCH] fixes --- backend/chaturbate_autostart.go | 284 ++++++++++---------------------- backend/main.go | 1 + backend/pending_autostart.go | 70 +++++++- backend/postwork.go | 28 +++- backend/recorder.go | 205 ++++++++++++++++++++--- backend/server.go | 2 +- 6 files changed, 358 insertions(+), 232 deletions(-) diff --git a/backend/chaturbate_autostart.go b/backend/chaturbate_autostart.go index 7335f8d..540f7b3 100644 --- a/backend/chaturbate_autostart.go +++ b/backend/chaturbate_autostart.go @@ -48,6 +48,20 @@ func normUser(s string) string { return strings.ToLower(strings.TrimSpace(s)) } +func chaturbateShowFromSnapshot(showByUser map[string]string, user string) string { + user = normUser(user) + if user == "" { + return "unknown" + } + + show, exists := showByUser[user] + if !exists { + return "offline" + } + + return normalizePendingShowServer(show) +} + func watchedChaturbateUserKey(m WatchedModelLite) string { key := normUser(m.ModelKey) if key != "" { @@ -275,17 +289,12 @@ func startChaturbateAutoStartWorker(store *ModelStore) { // normaler Retry für API-public const retryCooldown = 25 * time.Second - // aggressiver vermeiden für "nicht in API, aber watched" Blind-Try - const blindRetryCooldown = 2 * time.Minute - // wie lange wir warten, ob eine echte Datei entsteht const outputProbeMax = 20 * time.Second queue := make([]autoStartItem, 0, 64) queued := map[string]bool{} lastTry := map[string]time.Time{} - lastBlindTry := map[string]time.Time{} - probeCursor := 0 scanTicker := time.NewTicker(scanInterval) startTicker := time.NewTicker(startInterval) @@ -417,7 +426,6 @@ func startChaturbateAutoStartWorker(store *ModelStore) { // laufende Jobs sammeln runningByUser := map[string]*RecordJob{} - hiddenProbeUser := "" jobsMu.RLock() for _, j := range jobs { @@ -437,10 +445,6 @@ func startChaturbateAutoStartWorker(store *ModelStore) { } runningByUser[u] = j - - if j.Hidden && hiddenProbeUser == "" { - hiddenProbeUser = u - } } jobsMu.RUnlock() @@ -493,6 +497,11 @@ func startChaturbateAutoStartWorker(store *ModelStore) { nextQueued := map[string]bool{} for _, it := range queue { + // alte Blind-Probe-Einträge aus früheren Versionen verwerfen + if it.blind { + continue + } + if runningByUser[it.userKey] != nil { continue } @@ -534,22 +543,9 @@ func startChaturbateAutoStartWorker(store *ModelStore) { } } - show := normalizePendingShowServer(showByUser[it.userKey]) - - if it.blind { - // Offline-/Unknown-Blind-Probes nur für manuell hinzugefügte Models. - // Watched/Resume dürfen bei offline nicht in "Wartend" bleiben. - if it.source != "manual" { - continue - } - - if show != "offline" && show != "unknown" { - continue - } - } else { - if show != "public" { - continue - } + show := chaturbateShowFromSnapshot(showByUser, it.userKey) + if show != "public" { + continue } nextQueue = append(nextQueue, it) @@ -559,16 +555,6 @@ func startChaturbateAutoStartWorker(store *ModelStore) { queue = nextQueue queued = nextQueued - blindQueued := false - - for _, it := range queue { - if it.blind { - blindQueued = true - break - } - } - - offlineCandidates := make([]autoStartItem, 0, len(watchedOrder)) nextWatchedPending := make([]PendingAutoStartItem, 0, len(watchedOrder)) now := time.Now() @@ -585,7 +571,7 @@ func startChaturbateAutoStartWorker(store *ModelStore) { continue } - show := normalizePendingShowServer(showByUser[user]) + show := chaturbateShowFromSnapshot(showByUser, user) if show != "private" && show != "hidden" && show != "away" { continue } @@ -633,7 +619,7 @@ func startChaturbateAutoStartWorker(store *ModelStore) { u = fmt.Sprintf("https://chaturbate.com/%s/", user) } - show := normalizePendingShowServer(showByUser[user]) + show := chaturbateShowFromSnapshot(showByUser, user) img := strings.TrimSpace(imageByUser[user]) switch show { @@ -671,6 +657,7 @@ func startChaturbateAutoStartWorker(store *ModelStore) { pendingAutoStartMu.Unlock() case "offline": + // Wenn ein Resume-Model wirklich offline ist, nicht mehr in Pending anzeigen. pendingAutoStartMu.Lock() _ = removeResumePendingAutoStartItemForProvider("chaturbate", user) pendingAutoStartMu.Unlock() @@ -701,7 +688,7 @@ func startChaturbateAutoStartWorker(store *ModelStore) { continue } - show := normalizePendingShowServer(showByUser[user]) + show := chaturbateShowFromSnapshot(showByUser, user) img := strings.TrimSpace(imageByUser[user]) switch show { @@ -769,7 +756,7 @@ func startChaturbateAutoStartWorker(store *ModelStore) { continue } - show := normalizePendingShowServer(showByUser[user]) + show := chaturbateShowFromSnapshot(showByUser, user) switch show { case "public": @@ -796,57 +783,19 @@ func startChaturbateAutoStartWorker(store *ModelStore) { queued[user] = true case "private", "hidden", "away": - // manual pending bleibt einfach stehen und wartet auf public + // Manual Pending bleibt wartend, solange das Model nicht offline ist. + continue + + case "offline": + // Offline gehört nicht in Pending Autostart. + pendingAutoStartMu.Lock() + _ = removeManualPendingAutoStartItemForProvider("chaturbate", user) + pendingAutoStartMu.Unlock() continue default: - if autostartPaused { - continue - } - if runningByUser[user] != nil { - continue - } - if queued[user] { - continue - } - if t, ok := lastBlindTry[user]; ok && now.Sub(t) < blindRetryCooldown { - continue - } - - offlineCandidates = append(offlineCandidates, autoStartItem{ - userKey: user, - url: u, - blind: true, - source: "manual", - ownerUserKey: itm.OwnerUserKey, - }) - } - } - - // Nur EIN Offline-Kandidat gleichzeitig in die Queue. - // Bei pausiertem Autostart werden oben keine normalen Offline-Kandidaten erzeugt. - if !blindQueued && !hasHiddenProbeRunningForProvider("chaturbate") && len(offlineCandidates) > 0 { - n := len(offlineCandidates) - start := 0 - if probeCursor > 0 { - start = probeCursor % n - } - - for i := 0; i < n; i++ { - idx := (start + i) % n - it := offlineCandidates[idx] - - if queued[it.userKey] { - continue - } - if runningByUser[it.userKey] != nil { - continue - } - - queue = append(queue, it) - queued[it.userKey] = true - probeCursor = (idx + 1) % n - break + // unknown nicht als offline/wartend speichern. + continue } } @@ -902,13 +851,19 @@ func startChaturbateAutoStartWorker(store *ModelStore) { queue = append(queue[:startIdx], queue[startIdx+1:]...) delete(queued, it.userKey) - show := normalizePendingShowServer(showByUser[it.userKey]) + // Blind-Probes sind deaktiviert. Alte Queue-Einträge aus früheren Versionen + // werden hier nur noch verworfen. + if it.blind { + continue + } + + show := chaturbateShowFromSnapshot(showByUser, it.userKey) isPublic := show == "public" // Normale queued Starts dürfen NICHT als Blind-Probe gestartet werden. // Wenn ein resume/watched/manual Eintrag inzwischen nicht mehr public ist, // bleibt er wartend bzw. wird im nächsten Scan korrekt neu bewertet. - if !it.blind && !isPublic { + if !isPublic { if it.source == "resume" { pendingAutoStartMu.Lock() _ = saveResumePendingAutoStartItemForProvider("chaturbate", PendingAutoStartItem{ @@ -924,131 +879,64 @@ func startChaturbateAutoStartWorker(store *ModelStore) { continue } - // Nicht-public nur einzeln nacheinander prüfen. - if it.blind && hasHiddenProbeRunningForProvider("chaturbate") { - continue - } - if isJobRunningForURL(it.url) { continue } - if !it.blind { - lastTry[it.userKey] = time.Now() - - job, err := startRecordingInternal(RecordRequest{ - URL: it.url, - Cookie: lastCookieHdr, - IgnoreConcurrentLimit: it.force, - }) - if err != nil { - if verboseLogs() { - logChaturbateAutoStartError("❌ [autostart] start failed:", it.url, err) - } - continue - } + lastTry[it.userKey] = time.Now() + job, err := startRecordingInternal(RecordRequest{ + URL: it.url, + Cookie: lastCookieHdr, + IgnoreConcurrentLimit: it.force, + }) + if err != nil { if verboseLogs() { - if it.source == "resume" { - appLogln("▶️ [autostart] resumed:", it.url) - } else { - appLogln("▶️ [autostart] started:", it.url) - } + logChaturbateAutoStartError("❌ [autostart] start failed:", it.url, err) } + continue + } - if job != nil { - if it.source == "resume" { + if verboseLogs() { + if it.source == "resume" { + appLogln("▶️ [autostart] resumed:", it.url) + } else { + appLogln("▶️ [autostart] started:", it.url) + } + } + + if job != nil { + if it.source == "resume" { + pendingAutoStartMu.Lock() + _ = removeResumePendingAutoStartItemForProvider("chaturbate", it.userKey) + pendingAutoStartMu.Unlock() + + go chaturbateAbortIfNoOutput(job.ID, outputProbeMax, nil, func() { pendingAutoStartMu.Lock() - _ = removeResumePendingAutoStartItemForProvider("chaturbate", it.userKey) - pendingAutoStartMu.Unlock() - - go chaturbateAbortIfNoOutput(job.ID, outputProbeMax, nil, func() { - pendingAutoStartMu.Lock() - _ = saveResumePendingAutoStartItemForProvider("chaturbate", PendingAutoStartItem{ - ModelKey: it.userKey, - URL: it.url, - Mode: "wait_public", - CurrentShow: "unknown", - Source: "resume", - }) - pendingAutoStartMu.Unlock() + _ = saveResumePendingAutoStartItemForProvider("chaturbate", PendingAutoStartItem{ + ModelKey: it.userKey, + URL: it.url, + Mode: "wait_public", + CurrentShow: "unknown", + Source: "resume", }) - - continue - } - - if it.source == "manual" { - go chaturbateAbortIfNoOutput(job.ID, outputProbeMax, func() { - pendingAutoStartMu.Lock() - _ = removeManualPendingAutoStartItemForProvider("chaturbate", it.userKey) - pendingAutoStartMu.Unlock() - }, nil) - } else { - pendingAutoStartMu.Lock() - _ = removeWatchedPendingAutoStartItemForProvider("chaturbate", it.userKey) pendingAutoStartMu.Unlock() + }) - go chaturbateAbortIfNoOutput(job.ID, outputProbeMax, nil, nil) - } - } - } else { - if paused && !it.force { continue } - lastBlindTry[it.userKey] = time.Now() - - job, err := startRecordingInternal(RecordRequest{ - URL: it.url, - Cookie: lastCookieHdr, - Hidden: true, - IgnoreConcurrentLimit: it.force, - }) - if err != nil || job == nil { - if verboseLogs() { - if err != nil { - logChaturbateAutoStartError("❌ [autostart] blind start failed:", it.url, err) - } else { - appLogln("❌ [autostart] blind start failed:", it.url, "job is nil") - } - } - continue - } - - if verboseLogs() { - appLogln("▶️ [autostart] blind try started:", it.url) - } - if it.source == "manual" { - go chaturbateAbortIfNoOutput( - job.ID, - outputProbeMax, - func() { - // Download läuft wirklich: manuellen Wartend-Eintrag entfernen. - pendingAutoStartMu.Lock() - _ = removeManualPendingAutoStartItemForProvider("chaturbate", it.userKey) - pendingAutoStartMu.Unlock() - }, - func() { - // Blind-Probe hat keine Datei erzeugt: - // Jetzt erst als "Wartend" speichern. - owner := strings.TrimSpace(it.ownerUserKey) - if owner == "" { - return - } - - pendingAutoStartMu.Lock() - _ = saveManualPendingAutoStartItemForUser(owner, PendingAutoStartItem{ - ModelKey: it.userKey, - URL: it.url, - Mode: "wait_public", - CurrentShow: "offline", - Source: "manual", - }) - pendingAutoStartMu.Unlock() - }, - ) + go chaturbateAbortIfNoOutput(job.ID, outputProbeMax, func() { + pendingAutoStartMu.Lock() + _ = removeManualPendingAutoStartItemForProvider("chaturbate", it.userKey) + pendingAutoStartMu.Unlock() + }, nil) } else { + pendingAutoStartMu.Lock() + _ = removeWatchedPendingAutoStartItemForProvider("chaturbate", it.userKey) + pendingAutoStartMu.Unlock() + go chaturbateAbortIfNoOutput(job.ID, outputProbeMax, nil, nil) } } diff --git a/backend/main.go b/backend/main.go index 0d18891..f822340 100644 --- a/backend/main.go +++ b/backend/main.go @@ -1689,6 +1689,7 @@ func reencodeTSToMP4(ctx context.Context, tsPath, mp4Path string) error { "-vf", "scale=trunc(iw/2)*2:trunc(ih/2)*2", "-c:v", "libx264", + "-threads", "2", "-preset", "veryfast", "-crf", "20", "-pix_fmt", "yuv420p", diff --git a/backend/pending_autostart.go b/backend/pending_autostart.go index fb373d1..c1561a4 100644 --- a/backend/pending_autostart.go +++ b/backend/pending_autostart.go @@ -167,6 +167,40 @@ func normalizePendingShowServer(v string) string { } } +func sanitizePendingAutoStartItems(items []PendingAutoStartItem) []PendingAutoStartItem { + out := make([]PendingAutoStartItem, 0, len(items)) + seen := make(map[string]struct{}, len(items)) + + for _, it := range items { + it.ModelKey = strings.ToLower(strings.TrimSpace(it.ModelKey)) + it.URL = strings.TrimSpace(it.URL) + it.Mode = normalizePendingModeServer(it.Mode) + it.CurrentShow = normalizePendingShowServer(it.CurrentShow) + it.ImageURL = strings.TrimSpace(it.ImageURL) + it.Source = normalizePendingSourceServer(it.Source) + + if it.ModelKey == "" || it.URL == "" { + continue + } + + // Wichtig: + // Offline-Models gehören nicht in Pending Autostart. + if it.CurrentShow == "offline" { + continue + } + + dedupeKey := it.Source + "|" + pendingProviderFromURL(it.URL) + "|" + it.ModelKey + if _, exists := seen[dedupeKey]; exists { + continue + } + seen[dedupeKey] = struct{}{} + + out = append(out, it) + } + + return out +} + func safeUserKeyForFile(v string) string { re := regexp.MustCompile(`[^a-zA-Z0-9._-]+`) return re.ReplaceAllString(v, "_") @@ -388,12 +422,13 @@ func loadPendingAutoStartItems(userKey string) ([]PendingAutoStartItem, error) { return []PendingAutoStartItem{}, nil } - if f.Items == nil { - f.Items = []PendingAutoStartItem{} + if len(f.Items) == 0 { + return []PendingAutoStartItem{}, nil } - // noch einmal sauber normalisieren out := make([]PendingAutoStartItem, 0, len(f.Items)) + seen := make(map[string]struct{}, len(f.Items)) + for _, it := range f.Items { key := strings.ToLower(strings.TrimSpace(it.ModelKey)) u := strings.TrimSpace(it.URL) @@ -401,14 +436,33 @@ func loadPendingAutoStartItems(userKey string) ([]PendingAutoStartItem, error) { continue } + show := normalizePendingShowServer(it.CurrentShow) + + // Offline-Models gehören nicht in Pending Autostart. + // Alte persistierte Offline-Einträge werden dadurch beim Lesen ignoriert. + if show == "offline" { + continue + } + + source := normalizePendingSourceServer(it.Source) + mode := normalizePendingModeServer(it.Mode) + imageURL := strings.TrimSpace(it.ImageURL) + provider := pendingProviderFromURL(u) + + dedupeKey := source + "|" + provider + "|" + key + if _, exists := seen[dedupeKey]; exists { + continue + } + seen[dedupeKey] = struct{}{} + out = append(out, PendingAutoStartItem{ ModelKey: key, URL: u, - Mode: normalizePendingModeServer(it.Mode), + Mode: mode, NextProbeAtMs: it.NextProbeAtMs, - CurrentShow: normalizePendingShowServer(it.CurrentShow), - ImageURL: strings.TrimSpace(it.ImageURL), - Source: normalizePendingSourceServer(it.Source), + CurrentShow: show, + ImageURL: imageURL, + Source: source, }) } @@ -560,7 +614,7 @@ func savePendingAutoStartItems(userKey string, items []PendingAutoStartItem) err } payload := pendingAutoStartFile{ - Items: items, + Items: sanitizePendingAutoStartItems(items), } raw, err := json.MarshalIndent(payload, "", " ") diff --git a/backend/postwork.go b/backend/postwork.go index 24c1058..a06dc2c 100644 --- a/backend/postwork.go +++ b/backend/postwork.go @@ -390,8 +390,8 @@ func (pq *PostWorkQueue) StatusForKey(key string) PostWorkKeyStatus { } // global (oder in deinem app struct halten) -var postWorkQ = NewPostWorkQueue(512, 1) // schneller kritischer Pfad -var enrichQ = NewPostWorkQueue(512, 1) // langsame Hintergrundaufgaben +var postWorkQ = NewPostWorkQueue(1024, 1) // schneller kritischer Pfad +var enrichQ = NewPostWorkQueue(128, 1) // langsame Hintergrundaufgaben // --- Status Refresher (ehemals postwork_refresh.go) --- @@ -622,3 +622,27 @@ func startPostWorkStatusRefresher() { } }() } + +func (pq *PostWorkQueue) CancelRunning() int { + pq.mu.Lock() + + cancels := make([]context.CancelFunc, 0, len(pq.cancelByKey)) + + for key, cancel := range pq.cancelByKey { + if cancel == nil { + continue + } + + if _, running := pq.runningKeys[key]; running { + cancels = append(cancels, cancel) + } + } + + pq.mu.Unlock() + + for _, cancel := range cancels { + cancel() + } + + return len(cancels) +} diff --git a/backend/recorder.go b/backend/recorder.go index 676c682..d82859c 100644 --- a/backend/recorder.go +++ b/backend/recorder.go @@ -3,6 +3,7 @@ package main import ( "context" + "encoding/json" "errors" "fmt" "math" @@ -597,6 +598,105 @@ func buildPostworkTempMP4Path(tsPath string) (string, error) { return uniqueDestPath(tmpDir, base) } +func moveRepairSourceTSBestEffort(tsPath string, finalMP4Path string) string { + tsPath = strings.TrimSpace(tsPath) + if tsPath == "" { + return "" + } + + if err := requireNonEmptyRegularFile(tsPath, "repair source ts"); err != nil { + appLogln("⚠️ repair source ts not usable:", err) + return tsPath + } + + s := getSettings() + + doneAbs, err := resolvePathRelativeToApp(s.DoneDir) + if err != nil || strings.TrimSpace(doneAbs) == "" { + doneAbs = filepath.Dir(strings.TrimSpace(finalMP4Path)) + } + + doneAbs = strings.TrimSpace(doneAbs) + if doneAbs == "" { + return tsPath + } + + repairDir := filepath.Join(doneAbs, ".repair_sources") + if err := os.MkdirAll(repairDir, 0o755); err != nil { + appLogln("⚠️ repair source dir create failed:", err) + return tsPath + } + + dst, err := uniqueDestPath(repairDir, filepath.Base(tsPath)) + if err != nil { + appLogln("⚠️ repair source unique path failed:", err) + return tsPath + } + + if err := renameWithRetryAggressive(tsPath, dst); err != nil { + appLogln("⚠️ repair source move failed:", err) + return tsPath + } + + purgeDurationCacheForPath(tsPath) + + return dst +} + +func markVideoNeedsRepairBestEffort(videoPath string, sourceTSPath string, reason string, detail string) { + videoPath = strings.TrimSpace(videoPath) + if videoPath == "" { + return + } + + assetID := strings.TrimSpace(assetIDFromVideoPath(videoPath)) + if assetID == "" { + assetID = canonicalAssetIDFromName(filepath.Base(videoPath)) + } + if assetID == "" { + return + } + + marker := map[string]any{ + "needsRepair": true, + "reason": strings.TrimSpace(reason), + "detail": strings.TrimSpace(detail), + "videoPath": videoPath, + "sourceTSPath": strings.TrimSpace(sourceTSPath), + "createdAt": time.Now().UTC().Format(time.RFC3339Nano), + } + + b, err := json.MarshalIndent(marker, "", " ") + if err != nil { + appLogln("⚠️ repair marker marshal failed:", err) + return + } + + repairPath := "" + + if _, _, _, _, metaPath, err := assetPathsForID(assetID); err == nil { + metaPath = strings.TrimSpace(metaPath) + if metaPath != "" { + dir := filepath.Dir(metaPath) + if err := os.MkdirAll(dir, 0o755); err == nil { + repairPath = filepath.Join(dir, "repair.json") + } + } + } + + if repairPath == "" { + stem := strings.TrimSuffix(filepath.Base(videoPath), filepath.Ext(videoPath)) + repairPath = filepath.Join(filepath.Dir(videoPath), stem+".repair.json") + } + + if err := os.WriteFile(repairPath, b, 0o644); err != nil { + appLogln("⚠️ repair marker write failed:", err) + return + } + + appLogln("⚠️ video marked for deferred repair:", filepath.Base(videoPath), "marker=", repairPath) +} + func runPrimaryPostworkPipeline( ctx context.Context, job *RecordJob, @@ -611,6 +711,11 @@ func runPrimaryPostworkPipeline( cleanupTSAfterMove := "" removeTempOutOnMoveFailure := false + needsRepair := false + repairReason := "" + repairDetail := "" + repairSourceTS := "" + // 1) TS -> MP4 zuerst if strings.EqualFold(filepath.Ext(out), ".ts") { tsPath := out @@ -669,28 +774,26 @@ func runPrimaryPostworkPipeline( } if err := validateVideoDecodesNearEnd(ctx, mp4Path); err != nil { - appLogln("⚠️ remux mp4 decode validation failed, fallback to reencode:", err) + // Wichtig: + // Nicht mehr inline reencoden. Reencode ist CPU-teuer und blockiert die postWorkQ. + // Das remuxte MP4 bleibt erhalten und wird als "needs repair" markiert. + appLogln("⚠️ remux mp4 decode validation failed; keeping remux result and deferring repair:", err) - _ = removeWithRetry(mp4Path) - - if rerr := reencodeTSToMP4(ctx, tsPath, mp4Path); rerr != nil { - _ = removeWithRetry(mp4Path) - return tsPath, appErrorf("remux invalid (%v), reencode fallback failed: %w", err, rerr) - } - - if err2 := requireNonEmptyRegularFile(mp4Path, "reencode result"); err2 != nil { - _ = removeWithRetry(mp4Path) - return tsPath, err2 - } - - if err2 := validateVideoDecodesNearEnd(ctx, mp4Path); err2 != nil { - _ = removeWithRetry(mp4Path) - return tsPath, appErrorf("reencoded mp4 still invalid: %w", err2) - } + needsRepair = true + repairReason = "decode_near_end_failed" + repairDetail = err.Error() + repairSourceTS = tsPath + } + + // TS erst NACH erfolgreichem Move löschen. + // Bei Repair-Verdacht behalten wir die TS als Reparaturquelle. + if !needsRepair { + cleanupTSAfterMove = tsPath + } else { + cleanupTSAfterMove = "" + repairSourceTS = tsPath } - // TS erst NACH erfolgreichem Move löschen - cleanupTSAfterMove = tsPath removeTempOutOnMoveFailure = true out = mp4Path } @@ -729,6 +832,15 @@ func runPrimaryPostworkPipeline( return out, err } + if needsRepair { + keptSource := strings.TrimSpace(repairSourceTS) + if keptSource != "" { + keptSource = moveRepairSourceTSBestEffort(keptSource, out) + } + + markVideoNeedsRepairBestEffort(out, keptSource, repairReason, repairDetail) + } + if cleanupTSAfterMove != "" { if err := removeWithRetry(cleanupTSAfterMove); err != nil && !os.IsNotExist(err) { appLogln("⚠️ ts cleanup after successful move:", err) @@ -1134,6 +1246,10 @@ func enqueuePostworkOrFail(job *RecordJob, out string, postTarget JobStatus) { postFile := filepath.Base(out) postAssetID := assetIDFromVideoPath(out) + _, _, maxParallel := postWorkQ.Stats() + sortBucket, sortName := postWorkSortForVideoPath(out) + addedAt := time.Now() + jobsMu.Lock() job.Phase = "postwork" job.PostWorkKey = postKey @@ -1142,18 +1258,21 @@ func enqueuePostworkOrFail(job *RecordJob, out string, postTarget JobStatus) { Position: 0, Waiting: 0, Running: 0, - MaxParallel: 1, + MaxParallel: maxParallel, } jobsMu.Unlock() okQueued := postWorkQ.Enqueue(PostWorkTask{ - Key: postKey, - Path: out, - Added: time.Now(), + Key: postKey, + Path: out, + Added: addedAt, + SortBucket: sortBucket, + SortName: sortName, Run: func(ctx context.Context) error { return runQueuedPostwork(ctx, job, out, postTarget, postKey) }, }) + if okQueued { publishQueuedPostworkState(job, postKey, postFile, postAssetID) return @@ -1367,6 +1486,25 @@ func runDeferredEnrichPipeline(ctx context.Context, outPath string, sourceURL st return nil } + waitBeforeHeavyStep := func(queue string, phase string) error { + if err := ctx.Err(); err != nil { + publishDeferredPhase(fileName, assetID, queue, phase, "missing", "", nil) + return context.Canceled + } + + if err := waitUntilEnrichMayRun(ctx); err != nil { + publishDeferredPhase(fileName, assetID, queue, phase, "missing", "", nil) + return err + } + + if err := ctx.Err(); err != nil { + publishDeferredPhase(fileName, assetID, queue, phase, "missing", "", nil) + return context.Canceled + } + + return nil + } + truth := assetsTruthForVideo(outPath) // ------------------------------------------------- @@ -1374,6 +1512,10 @@ func runDeferredEnrichPipeline(ctx context.Context, outPath string, sourceURL st // ------------------------------------------------- if !truth.MetaReady { + if err := waitBeforeHeavyStep("postwork", "meta"); err != nil { + return err + } + publishDeferredPhase(fileName, assetID, "postwork", "meta", "running", "Meta", nil) okMeta, err := ensureMetaForVideoCtx(ctx, outPath, sourceURL) @@ -1393,6 +1535,10 @@ func runDeferredEnrichPipeline(ctx context.Context, outPath string, sourceURL st } if !truth.ThumbReady { + if err := waitBeforeHeavyStep("postwork", "thumb"); err != nil { + return err + } + publishDeferredPhase(fileName, assetID, "postwork", "thumb", "running", "Vorschaubild", nil) _, err := ensurePrimaryAssetsForVideoWithProgressCtx(ctx, outPath, sourceURL, nil) @@ -1412,6 +1558,10 @@ func runDeferredEnrichPipeline(ctx context.Context, outPath string, sourceURL st } if !truth.TeaserReady { + if err := waitBeforeHeavyStep("postwork", "teaser"); err != nil { + return err + } + publishDeferredPhase(fileName, assetID, "postwork", "teaser", "running", "Teaser", nil) _, err := ensureTeaserForVideoCtx(ctx, outPath, sourceURL) @@ -1437,6 +1587,10 @@ func runDeferredEnrichPipeline(ctx context.Context, outPath string, sourceURL st // ------------------------------------------------- if !truth.SpritesReady { + if err := waitBeforeHeavyStep("postwork", "sprites"); err != nil { + return err + } + publishDeferredPhase(fileName, assetID, "postwork", "sprites", "running", "Sprites", nil) _, err := ensureSpritesForVideoCtx(ctx, outPath, sourceURL) @@ -1452,6 +1606,7 @@ func runDeferredEnrichPipeline(ctx context.Context, outPath string, sourceURL st firstErr = err } } else { + // Best effort. Wenn währenddessen postWork reinkommt, beendet ctx diesen Check. if _, checkErr := runTailBlackoutCheck(ctx, outPath); checkErr != nil { appLogln("⚠️ sprite tail blackout check failed:", outPath, checkErr) } @@ -1477,6 +1632,10 @@ func runDeferredEnrichPipeline(ctx context.Context, outPath string, sourceURL st analyzeReady := hasAIResultsForAllOutputGoals(outPath, requiredGoals) if !analyzeReady { + if err := waitBeforeHeavyStep("enrich", "analyze"); err != nil { + return err + } + publishDeferredPhase(fileName, assetID, "enrich", "analyze", "running", "Analysen", nil) _, err := ensureAnalyzeAllGoalsForVideoCtx(ctx, outPath, sourceURL) diff --git a/backend/server.go b/backend/server.go index 7ece912..0d2c494 100644 --- a/backend/server.go +++ b/backend/server.go @@ -677,7 +677,7 @@ func main() { fixKeepRootFilesIntoModelSubdirs() - postWorkQ.StartWorkers(1) + postWorkQ.StartWorkers(2) enrichQ.StartWorkers(1) startPostWorkStatusRefresher() startEnrichStatusRefresher()