hpz937/zurg
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Rework repairs again

Browse Source
This commit is contained in:
Ben Sarmiento
2024-01-30 02:06:39 +01:00
parent b505400f60
commit 108607b4dd
4 changed files with 155 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
internal/torrent/fixer.go Normal file
View File

@@ -0,0 +1,96 @@
package torrent
import (
"io"
"os"
cmap "github.com/orcaman/concurrent-map/v2"
)
// fixers are commands that will be run on the next refresh
// they are stored in a file so that they can be run on startup
// they follow the format of:
// <id_trigger>: <command> <id>
// id_trigger = this means a specific torrent id's completion
// commands: delete | repair
func (t *TorrentManager) fixerAddCommand(trigger, command string) {
t.fixers.Set(trigger, command)
t.writeFixersToFile()
}
func (t *TorrentManager) handleFixers() {
var toDelete []string
allTorrents, _ := t.DirectoryMap.Get(INT_ALL)
infoCache, _ := t.DirectoryMap.Get(INT_INFO_CACHE)
infoCache.IterCb(func(id string, torrent *Torrent) {
if !t.fixers.Has(id) || torrent.AnyInProgress() {
return
}
command, _ := t.fixers.Pop(id)
switch command {
case "delete":
toDelete = append(toDelete, id)
case "repair":
toDelete = append(toDelete, id)
t.log.Debugf("Repairing torrent %s again now that fixer is done", t.GetKey(torrent))
repairMe, _ := allTorrents.Get(t.GetKey(torrent))
t.TriggerRepair(repairMe)
}
})
for _, id := range toDelete {
t.log.Debugf("Deleting fixer torrent id=%s", id)
t.Api.DeleteTorrent(id)
infoCache.Remove(id)
t.deleteTorrentFile(id)
}
}
func (t *TorrentManager) writeFixersToFile() {
filePath := "data/fixers.json"
file, err := os.Create(filePath)
if err != nil {
t.log.Warnf("Cannot create fixer file %s: %v", filePath, err)
return
}
defer file.Close()
fileData, err := t.fixers.MarshalJSON()
if err != nil {
t.log.Warnf("Cannot marshal fixers: %v", err)
return
}
_, err = file.Write(fileData)
if err != nil {
t.log.Warnf("Cannot write to fixer file %s: %v", filePath, err)
return
}
}
func (t *TorrentManager) readFixersFromFile() (ret cmap.ConcurrentMap[string, string]) {
ret = cmap.New[string]()
filePath := "data/fixers.json"
file, err := os.Open(filePath)
if err != nil {
if os.IsNotExist(err) {
return
}
t.log.Warnf("Cannot open fixer file %s: %v", filePath, err)
return
}
defer file.Close()
fileData, err := io.ReadAll(file)
if err != nil {
t.log.Warnf("Cannot read fixer file %s: %v", filePath, err)
return
}
err = ret.UnmarshalJSON(fileData)
if err != nil {
t.log.Warnf("Cannot unmarshal fixer file %s: %v", filePath, err)
return
}
return
}

5
internal/torrent/manager.go
View File

@@ -26,7 +26,7 @@ type TorrentManager struct {
DirectoryMap cmap.ConcurrentMap[string, cmap.ConcurrentMap[string, *Torrent]] // directory -> accessKey -> Torrent
DownloadCache cmap.ConcurrentMap[string, *realdebrid.Download]
DownloadMap cmap.ConcurrentMap[string, *realdebrid.Download]
fixers cmap.ConcurrentMap[string, *Torrent]
fixers cmap.ConcurrentMap[string, string] // trigger -> [command, id]
allAccessKeys mapset.Set[string]
latestState *LibraryState
requiredVersion string
@@ -54,7 +54,6 @@ func NewTorrentManager(cfg config.ConfigInterface, api *realdebrid.RealDebrid, w
DownloadMap: cmap.New[*realdebrid.Download](),
RefreshKillSwitch: make(chan struct{}, 1),
RepairKillSwitch: make(chan struct{}, 1),
fixers: cmap.New[*Torrent](),
allAccessKeys: mapset.NewSet[string](),
latestState: &LibraryState{},
requiredVersion: "0.9.3-hotfix.3",
@@ -63,7 +62,7 @@ func NewTorrentManager(cfg config.ConfigInterface, api *realdebrid.RealDebrid, w
repairPool: repairPool,
log: log,
}
t.fixers = t.readFixersFromFile()
t.initializeDirectories()
t.mountDownloads()
t.refreshTorrents()

27
internal/torrent/refresh.go
View File

@@ -29,16 +29,7 @@ func (t *TorrentManager) refreshTorrents() []string {
wg.Add(1)
_ = t.workerPool.Submit(func() {
defer wg.Done()
if !t.fixers.Has(instances[idx].ID) {
// not a fixer, just regular torrent
infoChan <- t.getMoreInfo(instances[idx])
return
} else if instances[idx].IsDone() {
// fixer is done, let's check if it fixed the torrent
infoChan <- t.handleFixers(instances[idx])
return
}
infoChan <- nil
})
}
@@ -54,7 +45,7 @@ func (t *TorrentManager) refreshTorrents() []string {
continue
}
accessKey := t.GetKey(info)
if !info.AnyInProgress() {
if !info.AllInProgress() {
newlyFetchedKeys.Add(accessKey)
}
@@ -95,6 +86,10 @@ func (t *TorrentManager) refreshTorrents() []string {
return false
})
if t.Config.EnableRepair() {
t.handleFixers()
}
return updatedPaths
}
@@ -134,8 +129,7 @@ func (t *TorrentManager) getMoreInfo(rdTorrent realdebrid.Torrent) *Torrent {
!torrentFromCache.AnyInProgress() &&
torrentFromCache.SelectedFiles.Count() == len(rdTorrent.Links) {
return torrentFromCache
}
} else if !exists {
torrentFromFile := t.readTorrentFromFile(rdTorrent.ID)
if torrentFromFile != nil &&
!torrentFromFile.AnyInProgress() &&
@@ -156,6 +150,7 @@ func (t *TorrentManager) getMoreInfo(rdTorrent realdebrid.Torrent) *Torrent {
t.log.Warnf("Torrent %s has broken files, will not save on info cache", rdTorrent.ID)
}
}
}
info, err := t.Api.GetTorrentInfo(rdTorrent.ID)
if err != nil {
@@ -222,6 +217,7 @@ func (t *TorrentManager) getMoreInfo(rdTorrent realdebrid.Torrent) *Torrent {
return &torrent
}
// ResetSelectedFiles will rename the file based on config
func (t *TorrentManager) ResetSelectedFiles(torrent *Torrent) {
// reset selected files
newSelectedFiles := cmap.New[*File]()
@@ -262,13 +258,12 @@ func (t *TorrentManager) mergeToMain(existing, toMerge *Torrent) Torrent {
InProgressIDs: newer.InProgressIDs.Union(older.InProgressIDs),
UnassignedLinks: newer.UnassignedLinks.Union(older.UnassignedLinks),
SelectedFiles: newer.SelectedFiles,
UnrepairableReason: newer.UnrepairableReason,
}
// unrepairable reason
if newer.UnrepairableReason != "" && older.UnrepairableReason != "" && newer.UnrepairableReason != older.UnrepairableReason {
mainTorrent.UnrepairableReason = fmt.Sprintf("%s, %s", newer.UnrepairableReason, older.UnrepairableReason)
} else if newer.UnrepairableReason != "" {
mainTorrent.UnrepairableReason = newer.UnrepairableReason
if mainTorrent.UnrepairableReason != "" && older.UnrepairableReason != "" && mainTorrent.UnrepairableReason != older.UnrepairableReason {
mainTorrent.UnrepairableReason = fmt.Sprintf("%s, %s", mainTorrent.UnrepairableReason, older.UnrepairableReason)
} else if older.UnrepairableReason != "" {
mainTorrent.UnrepairableReason = older.UnrepairableReason
}

89
internal/torrent/repair.go
View File

@@ -173,13 +173,13 @@ func (t *TorrentManager) repair(torrent *Torrent) {
t.log.Debugf("Torrent %s has %d broken out of %d files", t.GetKey(torrent), len(brokenFiles), torrent.SelectedFiles.Count())
brokenFileIDs := getFileIDs(brokenFiles)
// first solution: reinsert and see if the broken file is now working
t.log.Debugf("repair_method#1: Redownloading whole torrent %s", t.GetKey(torrent))
// first step: redownload the whole torrent
t.log.Debugf("Repairing torrent %s by redownloading", t.GetKey(torrent))
info, err := t.redownloadTorrent(torrent, "") // reinsert the torrent, passing ""
if err != nil {
t.log.Warnf("repair_method#1: Cannot repair torrent %s (error=%s)", t.GetKey(torrent), err.Error())
t.log.Warnf("Cannot repair torrent %s by redownloading (error=%s)", t.GetKey(torrent), err.Error())
} else if info != nil && info.Progress != 100 {
t.log.Infof("repair_method#1: Torrent %s is still in progress but it should work once done (torrent is temporarily hidden until download has completed)", t.GetKey(torrent))
t.log.Infof("Torrent %s is still in progress after redownloading but it should be repaired once done", t.GetKey(torrent))
return
} else if info != nil && info.IsDone() && !t.isStillBroken(info, brokenFiles) {
selectedFiles := getSelectedFiles(info)
@@ -192,29 +192,35 @@ func (t *TorrentManager) repair(torrent *Torrent) {
}
})
t.saveTorrentChangesToDisk(torrent, nil)
t.log.Infof("Successfully repaired torrent %s using repair_method#1", t.GetKey(torrent))
t.log.Infof("Successfully repaired torrent %s by redownloading", t.GetKey(torrent))
return
}
if torrent.UnrepairableReason != "" {
t.log.Debugf("Torrent %s has been marked as unfixable during repair_method#1 (%s), ending repair process early", t.GetKey(torrent), torrent.UnrepairableReason)
t.log.Debugf("Torrent %s has been marked as unfixable during redownload (%s), ending repair process early", t.GetKey(torrent), torrent.UnrepairableReason)
return
}
if len(brokenFiles) == torrent.SelectedFiles.Count() {
t.log.Warnf("Torrent %s has all broken files, marking as unplayable", t.GetKey(torrent))
// all files are broken, nothing we can do
t.log.Warnf("Torrent %s has broken cached files (cached but cannot be downloaded), you can repair it manually, marking as unfixable", t.GetKey(torrent))
t.markAsUnfixable(torrent, "broken cache")
return
}
// second solution: add only the broken files
// second step: download the broken files
if len(brokenFiles) > 0 {
t.log.Infof("repair_method#2: Redownloading %dof%d files only for torrent %s", len(brokenFiles), torrent.SelectedFiles.Count(), t.GetKey(torrent))
_, err := t.redownloadTorrent(torrent, brokenFileIDs)
t.log.Infof("Repairing by downloading only the %d broken out of %d files of torrent %s", len(brokenFiles), torrent.SelectedFiles.Count(), t.GetKey(torrent))
info, err := t.redownloadTorrent(torrent, brokenFileIDs)
if err != nil {
t.log.Warnf("repair_method#2: Cannot repair torrent %s (error=%s)", t.GetKey(torrent), err.Error())
}
t.log.Warnf("Cannot repair torrent %s by downloading broken files (error=%s) giving up", t.GetKey(torrent), err.Error())
return
}
if info != nil {
t.fixerAddCommand(info.ID, "repair")
return
}
}
t.log.Infof("Torrent %s has no broken files to repair", t.GetKey(torrent))
}
@@ -298,7 +304,7 @@ func (t *TorrentManager) redownloadTorrent(torrent *Torrent, selection string) (
oldTorrentIDs := make([]string, 0)
if selection == "" {
// only delete the old torrent if we are redownloading all files
oldTorrentIDs = torrent.DownloadedIDs.ToSlice()
oldTorrentIDs = torrent.DownloadedIDs.Union(torrent.InProgressIDs).ToSlice()
tmpSelection := ""
torrent.SelectedFiles.IterCb(func(_ string, file *File) {
tmpSelection += fmt.Sprintf("%d,", file.ID) // select all files
@@ -336,7 +342,7 @@ func (t *TorrentManager) redownloadTorrent(torrent *Torrent, selection string) (
newTorrentID := resp.ID
err = t.Api.SelectTorrentFiles(newTorrentID, selection)
if err != nil {
t.Api.DeleteTorrent(newTorrentID)
t.fixerAddCommand(newTorrentID, "delete")
return nil, fmt.Errorf("cannot start redownloading: %v", err)
}
@@ -346,7 +352,7 @@ func (t *TorrentManager) redownloadTorrent(torrent *Torrent, selection string) (
// see if the torrent is ready
info, err := t.Api.GetTorrentInfo(newTorrentID)
if err != nil {
t.Api.DeleteTorrent(newTorrentID)
t.fixerAddCommand(newTorrentID, "delete")
return nil, fmt.Errorf("cannot get info on redownloaded torrent %s (id=%s) : %v", t.GetKey(torrent), newTorrentID, err)
}
@@ -361,30 +367,29 @@ func (t *TorrentManager) redownloadTorrent(torrent *Torrent, selection string) (
}
}
if !isOkStatus {
t.Api.DeleteTorrent(newTorrentID)
t.fixerAddCommand(newTorrentID, "delete")
return nil, fmt.Errorf("the redownloaded torrent %s (id=%s) is in error state: %s", t.GetKey(torrent), newTorrentID, info.Status)
}
// check if incorrect number of links
selectionCount := len(strings.Split(selection, ","))
if info.Progress == 100 && len(info.Links) != selectionCount {
t.Api.DeleteTorrent(newTorrentID)
t.fixerAddCommand(newTorrentID, "delete")
return nil, fmt.Errorf("it did not fix the issue for %s (id=%s), only got %d files but we need %d, undoing", t.GetKey(torrent), info.ID, len(info.Links), selectionCount)
}
// looks like it's fixed
if len(oldTorrentIDs) > 0 {
// replace the old torrent
// replace the old torrent (empty selection)
infoCache, _ := t.DirectoryMap.Get(INT_INFO_CACHE)
for _, id := range oldTorrentIDs {
t.log.Debugf("Deleting torrent %s (id=%s) to replace with repaired torrent", t.GetKey(torrent), id)
torrent.DownloadedIDs.Remove(id)
t.Api.DeleteTorrent(id)
infoCache.Remove(id)
t.deleteTorrentFile(id)
}
} else {
// it's a fixer
t.fixers.Set(newTorrentID, torrent)
}
return info, nil
}
@@ -503,48 +508,6 @@ func (t *TorrentManager) isStillBroken(info *realdebrid.TorrentInfo, brokenFiles
return false
}
func (t *TorrentManager) handleFixers(fixer realdebrid.Torrent) *Torrent {
// since fixer is done, we can pop it from the fixers set
torrent, _ := t.fixers.Pop(fixer.ID)
if torrent == nil {
t.log.Warnf("repair_method#2: Fixer for %s (id=%s) is done but torrent has been deleted, deleting fixer...", fixer.Name, fixer.ID)
t.Api.DeleteTorrent(fixer.ID)
return nil
}
brokenFiles := getBrokenFiles(torrent)
info, err := t.redownloadTorrent(torrent, "") // reinsert
if err != nil {
t.log.Warnf("repair_method#2: Cannot repair torrent %s (error=%s)", t.GetKey(torrent), err.Error())
return nil
}
if info.IsDone() {
if !t.isStillBroken(info, brokenFiles) {
selectedFiles := getSelectedFiles(info)
torrent.SelectedFiles.IterCb(func(_ string, oldFile *File) {
for _, newFile := range selectedFiles {
if oldFile.Bytes == newFile.Bytes {
oldFile.Link = newFile.Link
break
}
}
})
t.saveTorrentChangesToDisk(torrent, nil)
t.log.Infof("Successfully repaired torrent %s using repair_method#2", t.GetKey(torrent))
} else {
t.log.Warnf("repair_method#2: Fixer is done but torrent %s is still broken; let's keep the fixer", t.GetKey(torrent))
t.Api.DeleteTorrent(info.ID)
return t.getMoreInfo(fixer)
}
} else {
t.log.Infof("repair_method#2: Torrent %s is still in progress but it should work once done (torrent is temporarily hidden until download has completed)", t.GetKey(torrent))
}
t.Api.DeleteTorrent(fixer.ID) // delete the fixer
return nil
}
func getSelectedFiles(info *realdebrid.TorrentInfo) []*File {
var selectedFiles []*File
// if some Links are empty, we need to repair it