Compare commits
3
Commits
a5c3990d12
...
7fbfd9c105
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7fbfd9c105 | ||
|
|
0a33b9d653 | ||
|
|
fc0121228e |
No files matched your search
@@ -47,7 +47,7 @@ func grabAll(
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
|
||||
f.manager.grab(ctx, dl, candidate, nil)
|
||||
f.manager.grab(ctx, dl, candidate, nil, false)
|
||||
}()
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,261 @@
|
||||
package download
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// A transfer that fails on one copy of an album is not a failed
|
||||
// download while another acceptable copy exists. On Soulseek the usual
|
||||
// failure is one peer being offline, with several others offering the
|
||||
// same folder.
|
||||
|
||||
var errPeerOffline = errors.New("peer went offline")
|
||||
|
||||
func TestManagerFallsBackToTheNextCandidate(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
f := newManagerFixture(t)
|
||||
|
||||
// The failing source ranks first on priority, so the fallback is
|
||||
// what reaches the one that works.
|
||||
bad := fakeWithAlbum(1, "offline-peer", ".flac")
|
||||
bad.GrabErr = errPeerOffline
|
||||
good := fakeWithAlbum(2, "online-peer", ".flac")
|
||||
|
||||
f.manager.installProvider(Config{ID: 1, Priority: 90}, bad)
|
||||
f.manager.installProvider(Config{ID: 2, Priority: 10}, good)
|
||||
|
||||
dl := fourTrackDownload()
|
||||
|
||||
if _, err := f.manager.Start(context.Background(), dl); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
|
||||
waitForDownloadState(t, f.store, dl.ID, StateComplete)
|
||||
|
||||
if bad.GrabCalls != 1 || good.GrabCalls != 1 {
|
||||
t.Errorf(
|
||||
"grabs: failing=%d working=%d, want 1 and 1",
|
||||
bad.GrabCalls, good.GrabCalls,
|
||||
)
|
||||
}
|
||||
|
||||
// The abandoned attempt's staging goes with it; only a request that
|
||||
// fails outright keeps its staging for inspection.
|
||||
waitFor(t, func() bool {
|
||||
entries, err := os.ReadDir(f.staging.Root())
|
||||
|
||||
return err == nil && len(entries) == 0
|
||||
}, "the failed attempt's staging was never released")
|
||||
}
|
||||
|
||||
// Falling back must not lower the bar. A second choice outside the
|
||||
// user's guardrails is not a choice auto-pick may make, first or second.
|
||||
func TestManagerFallbackRespectsTheGuardrails(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
f := newManagerFixture(t)
|
||||
f.manager.SetPreferences(AutoDownloadPrefs{MaxSizeMB: 50})
|
||||
|
||||
bad := fakeWithAlbum(1, "offline-peer", ".flac")
|
||||
bad.GrabErr = errPeerOffline
|
||||
bad.Candidates[0].TotalSize = 40 << 20
|
||||
|
||||
huge := fakeWithAlbum(2, "oversized", ".flac")
|
||||
huge.Candidates[0].TotalSize = 900 << 20
|
||||
|
||||
f.manager.installProvider(Config{ID: 1, Priority: 90}, bad)
|
||||
f.manager.installProvider(Config{ID: 2, Priority: 10}, huge)
|
||||
|
||||
dl := fourTrackDownload()
|
||||
|
||||
if _, err := f.manager.Start(context.Background(), dl); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
|
||||
waitForDownloadState(t, f.store, dl.ID, StateFailed)
|
||||
|
||||
if huge.GrabCalls != 0 {
|
||||
t.Errorf("fell back to a candidate over the size ceiling")
|
||||
}
|
||||
}
|
||||
|
||||
// A copy the user picked by hand is the copy they asked for. Quietly
|
||||
// substituting another is a decision they did not make.
|
||||
func TestManagerPickDoesNotFallBack(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
f := newManagerFixture(t)
|
||||
|
||||
bad := fakeWithAlbum(1, "offline-peer", ".flac")
|
||||
bad.GrabErr = errPeerOffline
|
||||
good := fakeWithAlbum(2, "online-peer", ".flac")
|
||||
|
||||
f.manager.installProvider(Config{ID: 1, Priority: 90}, bad)
|
||||
f.manager.installProvider(Config{ID: 2, Priority: 10}, good)
|
||||
|
||||
// A ceiling below both copies parks the result set for the user.
|
||||
f.manager.SetPreferences(AutoDownloadPrefs{MaxSizeMB: 1})
|
||||
|
||||
bad.Candidates[0].TotalSize = 30 << 20
|
||||
good.Candidates[0].TotalSize = 30 << 20
|
||||
|
||||
dl := fourTrackDownload()
|
||||
|
||||
if _, err := f.manager.Start(context.Background(), dl); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
|
||||
if err := f.manager.Pick(
|
||||
context.Background(), dl.ID, "offline-peer-cand",
|
||||
); err != nil {
|
||||
t.Fatalf("Pick: %v", err)
|
||||
}
|
||||
|
||||
waitForDownloadState(t, f.store, dl.ID, StateFailed)
|
||||
|
||||
if good.GrabCalls != 0 {
|
||||
t.Errorf("a hand-picked grab fell back to another candidate")
|
||||
}
|
||||
}
|
||||
|
||||
// Fallback is for surviving an offline peer or two, not for walking a
|
||||
// forty-peer list for six hours.
|
||||
func TestManagerFallbackIsBounded(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
f := newManagerFixture(t)
|
||||
|
||||
var providers []*FakeProvider
|
||||
|
||||
for i := int64(1); i <= maxGrabAttempts+2; i++ {
|
||||
p := fakeWithAlbum(i, "peer-"+itoa(int(i)), ".flac")
|
||||
p.GrabErr = errPeerOffline
|
||||
|
||||
f.manager.installProvider(Config{ID: i, Priority: 50}, p)
|
||||
providers = append(providers, p)
|
||||
}
|
||||
|
||||
dl := fourTrackDownload()
|
||||
|
||||
if _, err := f.manager.Start(context.Background(), dl); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
|
||||
waitForDownloadState(t, f.store, dl.ID, StateFailed)
|
||||
|
||||
grabs := 0
|
||||
for _, p := range providers {
|
||||
grabs += p.GrabCalls
|
||||
}
|
||||
|
||||
if grabs != maxGrabAttempts {
|
||||
t.Errorf("grabs = %d, want %d", grabs, maxGrabAttempts)
|
||||
}
|
||||
}
|
||||
|
||||
// The veto judges the best candidate *inside* the guardrails, so the
|
||||
// grab has to take that one — not the overall best, which may be the
|
||||
// very copy the user said not to take unattended.
|
||||
func TestManagerAutoPickTakesTheBestEligibleCandidate(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
f := newManagerFixture(t)
|
||||
f.manager.SetPreferences(AutoDownloadPrefs{MaxSizeMB: 50})
|
||||
|
||||
huge := fakeWithAlbum(1, "oversized", ".flac")
|
||||
huge.Candidates[0].TotalSize = 900 << 20
|
||||
|
||||
fits := fakeWithAlbum(2, "fits", ".flac")
|
||||
fits.Candidates[0].TotalSize = 40 << 20
|
||||
|
||||
f.manager.installProvider(Config{ID: 1, Priority: 90}, huge)
|
||||
f.manager.installProvider(Config{ID: 2, Priority: 10}, fits)
|
||||
|
||||
dl := fourTrackDownload()
|
||||
|
||||
ranked, err := f.manager.Start(context.Background(), dl)
|
||||
if err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
|
||||
if ranked[0].ID != "oversized-cand" {
|
||||
t.Fatalf("fixture: best overall is %s, want the oversized copy", ranked[0].ID)
|
||||
}
|
||||
|
||||
waitForDownloadState(t, f.store, dl.ID, StateComplete)
|
||||
|
||||
if huge.GrabCalls != 0 || fits.GrabCalls != 1 {
|
||||
t.Errorf(
|
||||
"grabs: oversized=%d fits=%d, want 0 and 1",
|
||||
huge.GrabCalls, fits.GrabCalls,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// On Soulseek a failure is the peer's, so every folder that peer offered
|
||||
// goes with it. Elsewhere a failure is the release's, and one indexer's
|
||||
// other releases are still worth trying.
|
||||
func TestRuledOutBy(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
failed := []Candidate{
|
||||
{ID: "slskd:alice:Album", Kind: KindSlskd, ProviderID: 1, Origin: "alice"},
|
||||
{ID: "tracker-1", Kind: KindProwlarr, ProviderID: 2, Origin: "indexer"},
|
||||
}
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
c Candidate
|
||||
want bool
|
||||
}{
|
||||
{
|
||||
name: "the same candidate",
|
||||
c: Candidate{ID: "tracker-1", Kind: KindProwlarr, ProviderID: 2, Origin: "indexer"},
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "another folder from a failed peer",
|
||||
c: Candidate{
|
||||
ID: "slskd:alice:Album (2)",
|
||||
Kind: KindSlskd,
|
||||
ProviderID: 1,
|
||||
Origin: "alice",
|
||||
},
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "another peer",
|
||||
c: Candidate{ID: "slskd:bob:Album", Kind: KindSlskd, ProviderID: 1, Origin: "bob"},
|
||||
want: false,
|
||||
},
|
||||
{
|
||||
name: "another release from the same indexer",
|
||||
c: Candidate{ID: "tracker-2", Kind: KindProwlarr, ProviderID: 2, Origin: "indexer"},
|
||||
want: false,
|
||||
},
|
||||
{
|
||||
name: "a peer of the same name on a different daemon",
|
||||
c: Candidate{
|
||||
ID: "slskd:alice:Album",
|
||||
Kind: KindSlskd,
|
||||
ProviderID: 3,
|
||||
Origin: "alice",
|
||||
},
|
||||
want: false,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
if got := ruledOutBy(tc.c, failed); got != tc.want {
|
||||
t.Errorf("ruledOutBy = %v, want %v", got, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
+206
-59
@@ -577,12 +577,12 @@ func (m *Manager) Start(
|
||||
))
|
||||
}
|
||||
|
||||
if m.AutoPickable(dl, ranked) {
|
||||
if pick, ok := autoPick(dl, ranked, m.preferences()); ok {
|
||||
if job != nil {
|
||||
job.Logf(jobs.LevelInfo, "Auto-selected best candidate")
|
||||
}
|
||||
|
||||
go m.grab(context.WithoutCancel(ctx), dl, ranked[0], job)
|
||||
go m.grab(context.WithoutCancel(ctx), dl, pick, job, true)
|
||||
|
||||
return ranked, nil
|
||||
}
|
||||
@@ -622,6 +622,13 @@ func (m *Manager) Attempt(
|
||||
return false, veto, nil
|
||||
}
|
||||
|
||||
pick, ok := autoPick(dl, ranked, m.preferences())
|
||||
if !ok {
|
||||
// Unreachable while autoPick and AutoPickVeto agree; kept so a
|
||||
// future divergence refuses rather than grabbing blind.
|
||||
return false, "no candidate clears the auto-download bar", nil
|
||||
}
|
||||
|
||||
if err := m.store.CreateDownload(ctx, dl); err != nil {
|
||||
return false, "", err
|
||||
}
|
||||
@@ -643,7 +650,7 @@ func (m *Manager) Attempt(
|
||||
))
|
||||
}
|
||||
|
||||
go m.grab(context.WithoutCancel(ctx), dl, ranked[0], job)
|
||||
go m.grab(context.WithoutCancel(ctx), dl, pick, job, true)
|
||||
|
||||
return true, "", nil
|
||||
}
|
||||
@@ -678,7 +685,7 @@ func (m *Manager) Pick(
|
||||
|
||||
job := m.startJob(dl)
|
||||
|
||||
go m.grab(context.WithoutCancel(ctx), dl, *chosen, job)
|
||||
go m.grab(context.WithoutCancel(ctx), dl, *chosen, job, false)
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -702,13 +709,20 @@ func (m *Manager) Cancel(ctx context.Context, downloadID string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// grab drives one candidate all the way to the library. It runs on its
|
||||
// grab drives one request all the way to the library. It runs on its
|
||||
// own goroutine and owns the job from here on.
|
||||
//
|
||||
// When fallback is set and a candidate's transfer fails, the next
|
||||
// candidate that auto-pick would itself have accepted is tried in its
|
||||
// place (see nextCandidate). It is set for the two unattended routes
|
||||
// and not for a candidate the user picked by hand: they chose that copy,
|
||||
// and quietly substituting another is a decision they did not make.
|
||||
func (m *Manager) grab(
|
||||
ctx context.Context,
|
||||
dl Download,
|
||||
c Candidate,
|
||||
job *jobs.Handle,
|
||||
fallback bool,
|
||||
) {
|
||||
ctx, cancel := context.WithTimeout(ctx, grabTimeout)
|
||||
defer cancel()
|
||||
@@ -723,6 +737,82 @@ func (m *Manager) grab(
|
||||
m.actMu.Unlock()
|
||||
}()
|
||||
|
||||
var failed []Candidate
|
||||
|
||||
for {
|
||||
out := m.attemptGrab(ctx, dl, c, job)
|
||||
if out.err == nil {
|
||||
m.finishGrab(ctx, dl, out.item, out.imported, job)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
failed = append(failed, c)
|
||||
|
||||
next, ok := m.nextCandidate(ctx, dl, failed, out, fallback)
|
||||
if !ok {
|
||||
m.failDownload(ctx, job, dl.ID, out.err)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
m.logger.Info(
|
||||
"download candidate failed; trying the next",
|
||||
"download", dl.ID,
|
||||
"failed", c.ID,
|
||||
"next", next.ID,
|
||||
"error", out.err,
|
||||
)
|
||||
|
||||
if job != nil {
|
||||
job.Logf(jobs.LevelWarn, fmt.Sprintf(
|
||||
"%s failed (%v); trying %s instead",
|
||||
describeCandidate(c), out.err, describeCandidate(next),
|
||||
))
|
||||
}
|
||||
|
||||
// The failed attempt's staging holds at most a partial folder
|
||||
// nobody is going to import, and the next attempt reserves its
|
||||
// own. Only the final failure keeps its staging for inspection.
|
||||
if out.item.StagingDir != "" {
|
||||
if err := m.staging.Release(out.item.StagingDir); err != nil {
|
||||
m.logger.Warn("could not release staging dir", "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
c = next
|
||||
}
|
||||
}
|
||||
|
||||
// maxGrabAttempts bounds how many candidates one request will try. A
|
||||
// popular album can have dozens of peers; the point of falling back is
|
||||
// to survive the ordinary one or two that are offline, not to walk the
|
||||
// whole list for six hours.
|
||||
const maxGrabAttempts = 3
|
||||
|
||||
// grabOutcome is how one candidate's attempt ended.
|
||||
type grabOutcome struct {
|
||||
item DownloadItem
|
||||
imported ImportResult
|
||||
err error
|
||||
|
||||
// retryable reports whether another candidate might succeed where
|
||||
// this one failed: the transfer failed, or delivered too little of
|
||||
// the album. Anything else — no staging space, no library root, a
|
||||
// tag write failing — would fail the next candidate identically.
|
||||
retryable bool
|
||||
}
|
||||
|
||||
// attemptGrab takes one candidate through transfer and import. It
|
||||
// records the item's own failure, but not the download's: whether the
|
||||
// download has failed is the caller's decision, since another candidate
|
||||
// may yet succeed.
|
||||
func (m *Manager) attemptGrab(
|
||||
ctx context.Context,
|
||||
dl Download,
|
||||
c Candidate,
|
||||
job *jobs.Handle,
|
||||
) grabOutcome {
|
||||
// Who will move the bytes is decided before any slot is taken, so
|
||||
// the transfer waits in its own provider's queue rather than in a
|
||||
// global one. A delegate takes no slot at all: the transfer is
|
||||
@@ -731,9 +821,7 @@ func (m *Manager) grab(
|
||||
// work against our budget.
|
||||
plan, err := m.planTransfer(dl, c)
|
||||
if err != nil {
|
||||
m.failDownload(ctx, job, dl.ID, err)
|
||||
|
||||
return
|
||||
return grabOutcome{err: err}
|
||||
}
|
||||
|
||||
if !plan.delegated() {
|
||||
@@ -743,9 +831,7 @@ func (m *Manager) grab(
|
||||
case provSem <- struct{}{}:
|
||||
defer func() { <-provSem }()
|
||||
case <-ctx.Done():
|
||||
m.failDownload(ctx, job, dl.ID, ctx.Err())
|
||||
|
||||
return
|
||||
return grabOutcome{err: ctx.Err()}
|
||||
}
|
||||
|
||||
globalSem := m.globalSem()
|
||||
@@ -754,9 +840,7 @@ func (m *Manager) grab(
|
||||
case globalSem <- struct{}{}:
|
||||
defer func() { <-globalSem }()
|
||||
case <-ctx.Done():
|
||||
m.failDownload(ctx, job, dl.ID, ctx.Err())
|
||||
|
||||
return
|
||||
return grabOutcome{err: ctx.Err()}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -771,24 +855,30 @@ func (m *Manager) grab(
|
||||
|
||||
dir, err := m.staging.Reserve(item.ID)
|
||||
if err != nil {
|
||||
m.failDownload(ctx, job, dl.ID, err)
|
||||
|
||||
return
|
||||
return grabOutcome{err: err}
|
||||
}
|
||||
|
||||
item.StagingDir = dir
|
||||
|
||||
if err := m.store.CreateItem(ctx, item); err != nil {
|
||||
m.failDownload(ctx, job, dl.ID, err)
|
||||
return grabOutcome{item: item, err: err}
|
||||
}
|
||||
|
||||
return
|
||||
fail := func(err error, retryable bool) grabOutcome {
|
||||
if serr := m.store.SetItemState(
|
||||
ctx, item.ID, StateFailed, err.Error(),
|
||||
); serr != nil {
|
||||
m.logger.Warn("could not record item failure", "error", serr)
|
||||
}
|
||||
|
||||
return grabOutcome{item: item, err: err, retryable: retryable}
|
||||
}
|
||||
|
||||
result, err := m.transfer(ctx, dl, item, plan, job)
|
||||
if err != nil {
|
||||
m.failItem(ctx, job, item, dl.ID, err)
|
||||
|
||||
return
|
||||
// A delegate's failure is the external manager's verdict on the
|
||||
// whole request, not on one copy of it.
|
||||
return fail(err, !plan.delegated())
|
||||
}
|
||||
|
||||
m.setStates(ctx, dl.ID, item.ID, StateImporting)
|
||||
@@ -798,42 +888,116 @@ func (m *Manager) grab(
|
||||
job.SetStages(importStages(2))
|
||||
}
|
||||
|
||||
var imported ImportResult
|
||||
|
||||
if result.Delegated {
|
||||
// The external manager already placed and tagged these files in
|
||||
// its own library. Moving them out from under a system that is
|
||||
// still managing them would be worse than useless, so the files
|
||||
// are recorded where they are and the library scan picks them
|
||||
// up in place.
|
||||
imported = ImportResult{Paths: result.Files}
|
||||
|
||||
if job != nil {
|
||||
job.Logf(jobs.LevelInfo, fmt.Sprintf(
|
||||
"External manager imported %d files; recording them in place",
|
||||
len(result.Files),
|
||||
))
|
||||
}
|
||||
} else {
|
||||
opts := m.importOptions()
|
||||
opts.WriteTags = true
|
||||
|
||||
opts.LibraryRoot, err = m.library.LibraryPath(dl.LibraryID)
|
||||
if err != nil {
|
||||
m.failItem(ctx, job, item, dl.ID,
|
||||
fmt.Errorf("resolve library root: %w", err))
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
imported, err = m.importer.Import(ctx, dl, result, opts)
|
||||
if err != nil {
|
||||
m.failItem(ctx, job, item, dl.ID, err)
|
||||
|
||||
return
|
||||
return grabOutcome{
|
||||
item: item,
|
||||
imported: ImportResult{Paths: result.Files},
|
||||
}
|
||||
}
|
||||
|
||||
opts := m.importOptions()
|
||||
opts.WriteTags = true
|
||||
|
||||
opts.LibraryRoot, err = m.library.LibraryPath(dl.LibraryID)
|
||||
if err != nil {
|
||||
return fail(fmt.Errorf("resolve library root: %w", err), false)
|
||||
}
|
||||
|
||||
imported, err := m.importer.Import(ctx, dl, result, opts)
|
||||
if err != nil {
|
||||
return fail(err, errors.Is(err, ErrTooIncomplete))
|
||||
}
|
||||
|
||||
return grabOutcome{item: item, imported: imported}
|
||||
}
|
||||
|
||||
// nextCandidate picks the candidate to try after the ones in failed.
|
||||
//
|
||||
// It only ever offers a candidate auto-pick would have taken on its own
|
||||
// (autoAcceptable), so falling back cannot lower the bar an unattended
|
||||
// download is held to: the second choice has to clear the same gates
|
||||
// the first did.
|
||||
//
|
||||
// On Soulseek a failure belongs to the *peer* — offline, refusing, or
|
||||
// holding us in a queue — so every folder that peer offered is skipped
|
||||
// with it. Elsewhere a failure belongs to the release, and only that
|
||||
// candidate is.
|
||||
func (m *Manager) nextCandidate(
|
||||
ctx context.Context,
|
||||
dl Download,
|
||||
failed []Candidate,
|
||||
out grabOutcome,
|
||||
fallback bool,
|
||||
) (Candidate, bool) {
|
||||
if !fallback || !out.retryable || ctx.Err() != nil ||
|
||||
len(failed) >= maxGrabAttempts {
|
||||
return Candidate{}, false
|
||||
}
|
||||
|
||||
m.resMu.RLock()
|
||||
ranked := m.results[dl.ID]
|
||||
m.resMu.RUnlock()
|
||||
|
||||
prefs := m.preferences()
|
||||
|
||||
for _, c := range ranked {
|
||||
if ruledOutBy(c, failed) || !autoAcceptable(dl, c, prefs) {
|
||||
continue
|
||||
}
|
||||
|
||||
return c, true
|
||||
}
|
||||
|
||||
return Candidate{}, false
|
||||
}
|
||||
|
||||
// ruledOutBy reports whether a failure among failed also rules out c.
|
||||
func ruledOutBy(c Candidate, failed []Candidate) bool {
|
||||
for _, f := range failed {
|
||||
if c.ID == f.ID && c.ProviderID == f.ProviderID {
|
||||
return true
|
||||
}
|
||||
|
||||
if c.Kind == KindSlskd && f.Kind == KindSlskd &&
|
||||
c.ProviderID == f.ProviderID && c.Origin != "" &&
|
||||
c.Origin == f.Origin {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
// describeCandidate names a candidate for the job log.
|
||||
func describeCandidate(c Candidate) string {
|
||||
if c.Origin != "" {
|
||||
return fmt.Sprintf("%q from %s", c.Title, c.Origin)
|
||||
}
|
||||
|
||||
return fmt.Sprintf("%q", c.Title)
|
||||
}
|
||||
|
||||
// finishGrab records a successful import and retires what the request
|
||||
// was holding.
|
||||
func (m *Manager) finishGrab(
|
||||
ctx context.Context,
|
||||
dl Download,
|
||||
item DownloadItem,
|
||||
imported ImportResult,
|
||||
job *jobs.Handle,
|
||||
) {
|
||||
if err := m.store.SetItemImported(
|
||||
ctx, item.ID, imported.Paths,
|
||||
); err != nil {
|
||||
@@ -1177,23 +1341,6 @@ func (m *Manager) failDownload(
|
||||
}
|
||||
}
|
||||
|
||||
// failItem records an item-level failure and fails its download.
|
||||
func (m *Manager) failItem(
|
||||
ctx context.Context,
|
||||
job *jobs.Handle,
|
||||
item DownloadItem,
|
||||
downloadID string,
|
||||
err error,
|
||||
) {
|
||||
if serr := m.store.SetItemState(
|
||||
ctx, item.ID, StateFailed, err.Error(),
|
||||
); serr != nil {
|
||||
m.logger.Warn("could not record item failure", "error", serr)
|
||||
}
|
||||
|
||||
m.failDownload(ctx, job, downloadID, err)
|
||||
}
|
||||
|
||||
// startJob registers the request in the background jobs panel.
|
||||
func (m *Manager) startJob(dl Download) *jobs.Handle {
|
||||
if m.jobsReg == nil {
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/url"
|
||||
"os"
|
||||
"path"
|
||||
"path/filepath"
|
||||
@@ -75,6 +76,24 @@ const (
|
||||
|
||||
// slskdHTTPTimeout bounds one API call.
|
||||
slskdHTTPTimeout = 20 * time.Second
|
||||
|
||||
// slskdStallAfter is how long a grab may go without a byte arriving
|
||||
// before the peer is given up on. It is measured from enqueue, so
|
||||
// it covers a peer that queues us and never starts as well as one
|
||||
// that starts and stops. Ten minutes is long enough for a short
|
||||
// queue ahead of us to clear and short enough that one unresponsive
|
||||
// peer does not hold slskd's single transfer slot for an evening.
|
||||
slskdStallAfter = 10 * time.Minute
|
||||
|
||||
// slskdAbsentGrace is how long a requested file may be missing from
|
||||
// slskd's transfer list before it is counted as failed. slskd lists
|
||||
// a transfer as soon as it accepts it, so a file still absent after
|
||||
// a few polls was refused.
|
||||
slskdAbsentGrace = 30 * time.Second
|
||||
|
||||
// slskdCancelTimeout bounds the cleanup that cancels abandoned
|
||||
// transfers.
|
||||
slskdCancelTimeout = 15 * time.Second
|
||||
)
|
||||
|
||||
func init() {
|
||||
@@ -134,6 +153,8 @@ type slskd struct {
|
||||
searchPoll time.Duration
|
||||
searchWait time.Duration
|
||||
transferPoll time.Duration
|
||||
stallAfter time.Duration
|
||||
absentGrace time.Duration
|
||||
}
|
||||
|
||||
// newSlskd builds the provider from config.
|
||||
@@ -188,6 +209,8 @@ func newSlskd(
|
||||
searchPoll: slskdSearchPoll,
|
||||
searchWait: slskdSearchWait,
|
||||
transferPoll: slskdTransferPoll,
|
||||
stallAfter: slskdStallAfter,
|
||||
absentGrace: slskdAbsentGrace,
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -467,6 +490,15 @@ func (s *slskd) Grab(
|
||||
)
|
||||
}
|
||||
|
||||
// slskd keeps finished transfers listed until someone removes them,
|
||||
// and a transfer is matched to the request by filename. A record
|
||||
// left by an earlier attempt at the same file from the same peer
|
||||
// would otherwise be read as this attempt's answer the moment the
|
||||
// first poll came back — an old failure failing a transfer that has
|
||||
// not started. So what is already terminal is noted before enqueueing
|
||||
// and ignored after.
|
||||
stale := s.terminalTransferIDs(ctx, username)
|
||||
|
||||
wanted := make([]map[string]any, 0, len(c.Files))
|
||||
for _, f := range c.Files {
|
||||
wanted = append(wanted, map[string]any{
|
||||
@@ -476,24 +508,69 @@ func (s *slskd) Grab(
|
||||
}
|
||||
|
||||
if err := s.client.post(
|
||||
ctx, "/api/v0/transfers/downloads/"+username, wanted, nil,
|
||||
ctx, slskdDownloadsPath(username), wanted, nil,
|
||||
); err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
|
||||
if err := s.awaitTransfers(ctx, username, c, onProgress); err != nil {
|
||||
if err := s.awaitTransfers(
|
||||
ctx, username, stale, c, onProgress,
|
||||
); err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
|
||||
return s.collect(c, dst)
|
||||
}
|
||||
|
||||
// slskdDownloadsPath is the transfers endpoint for one peer. Soulseek
|
||||
// usernames may contain spaces and punctuation, so the name is escaped
|
||||
// rather than spliced into the path.
|
||||
func slskdDownloadsPath(username string) string {
|
||||
return "/api/v0/transfers/downloads/" + url.PathEscape(username)
|
||||
}
|
||||
|
||||
// terminalTransferIDs returns the ids of this peer's transfers that are
|
||||
// already finished. Best effort: slskd answers 404 for a peer it has no
|
||||
// transfers with, and any failure here means only that there is nothing
|
||||
// to ignore.
|
||||
func (s *slskd) terminalTransferIDs(
|
||||
ctx context.Context,
|
||||
username string,
|
||||
) map[string]bool {
|
||||
transfers, err := s.transfersFor(ctx, username)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
out := make(map[string]bool, len(transfers))
|
||||
|
||||
for _, t := range transfers {
|
||||
if finished, _ := t.done(); finished && t.ID != "" {
|
||||
out[t.ID] = true
|
||||
}
|
||||
}
|
||||
|
||||
return out
|
||||
}
|
||||
|
||||
// awaitTransfers polls until every requested file reaches a terminal
|
||||
// state. Soulseek queues are measured in hours, so the only deadline
|
||||
// is the caller's context.
|
||||
// state, the transfer stalls, or the caller gives up.
|
||||
//
|
||||
// Soulseek queues are measured in hours, so there is no deadline on the
|
||||
// transfer as a whole — but there is one on *progress*. slskd's
|
||||
// transfer limit is one, so a peer that holds us in its queue without
|
||||
// sending a byte is not only failing this download, it is holding every
|
||||
// other Soulseek download behind it. After stallAfter with nothing
|
||||
// moving the peer is given up on, and the manager tries another.
|
||||
//
|
||||
// Whatever way this ends short of every file finishing, the transfers
|
||||
// still live in slskd are cancelled there. Returning without doing so
|
||||
// leaves the daemon downloading into its own folder for a request
|
||||
// nobody is waiting on any more.
|
||||
func (s *slskd) awaitTransfers(
|
||||
ctx context.Context,
|
||||
username string,
|
||||
stale map[string]bool,
|
||||
c Candidate,
|
||||
onProgress ProgressFunc,
|
||||
) error {
|
||||
@@ -502,9 +579,18 @@ func (s *slskd) awaitTransfers(
|
||||
wanted[f.Path] = true
|
||||
}
|
||||
|
||||
var (
|
||||
started = time.Now()
|
||||
lastProgress = started
|
||||
lastBytes int64
|
||||
live []slskdTransfer
|
||||
)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
s.cancelTransfers(username, live)
|
||||
|
||||
return fmt.Errorf("%w: transfer cancelled", ErrSlskdTimeout)
|
||||
case <-time.After(s.transferPoll):
|
||||
}
|
||||
@@ -512,60 +598,177 @@ func (s *slskd) awaitTransfers(
|
||||
transfers, err := s.transfersFor(ctx, username)
|
||||
if err != nil {
|
||||
// A blip talking to the daemon should not abandon a
|
||||
// transfer that may be hours in.
|
||||
// transfer that may be hours in — but a daemon that stays
|
||||
// away is a stall like any other.
|
||||
s.logger.Debug("slskd transfer poll failed", "error", err)
|
||||
|
||||
if time.Since(lastProgress) >= s.stallAfter {
|
||||
s.cancelTransfers(username, live)
|
||||
|
||||
return fmt.Errorf(
|
||||
"%w: slskd has not answered for %s: %w",
|
||||
ErrSlskdTimeout, s.stallAfter, err,
|
||||
)
|
||||
}
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
var (
|
||||
done, failed int
|
||||
current int64
|
||||
tally := tallyTransfers(
|
||||
transfers, wanted, stale,
|
||||
time.Since(started) >= s.absentGrace,
|
||||
)
|
||||
live = tally.live
|
||||
|
||||
for _, t := range transfers {
|
||||
if !wanted[t.Filename] {
|
||||
continue
|
||||
}
|
||||
|
||||
current += t.BytesTransferred
|
||||
|
||||
finished, ok := t.done()
|
||||
if !finished {
|
||||
continue
|
||||
}
|
||||
|
||||
if ok {
|
||||
done++
|
||||
} else {
|
||||
failed++
|
||||
}
|
||||
if tally.bytes > lastBytes {
|
||||
lastBytes = tally.bytes
|
||||
lastProgress = time.Now()
|
||||
}
|
||||
|
||||
if onProgress != nil {
|
||||
onProgress(Progress{
|
||||
Current: current,
|
||||
Current: tally.bytes,
|
||||
Total: c.TotalSize,
|
||||
Phase: fmt.Sprintf(
|
||||
"Transferring from %s (%d/%d)", username, done, len(wanted),
|
||||
"Transferring from %s (%d/%d)",
|
||||
username, tally.done, len(wanted),
|
||||
),
|
||||
})
|
||||
}
|
||||
|
||||
if done+failed < len(wanted) {
|
||||
if tally.done+tally.failed >= len(wanted) {
|
||||
// Some files failing is normal — a peer goes offline
|
||||
// mid-folder. Let the importer's completeness check decide
|
||||
// whether what arrived is enough, rather than discarding it
|
||||
// here.
|
||||
if tally.done == 0 {
|
||||
return fmt.Errorf(
|
||||
"%w: all %d files failed",
|
||||
ErrSlskdTransferFailed, tally.failed,
|
||||
)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
if time.Since(lastProgress) < s.stallAfter {
|
||||
continue
|
||||
}
|
||||
|
||||
// Some files failing is normal — a peer goes offline mid-folder.
|
||||
// Let the importer's completeness check decide whether what
|
||||
// arrived is enough, rather than discarding it here.
|
||||
if done == 0 {
|
||||
return fmt.Errorf(
|
||||
"%w: all %d files failed", ErrSlskdTransferFailed, failed,
|
||||
s.cancelTransfers(username, live)
|
||||
|
||||
// A folder that stalls on its last track is the same shape as
|
||||
// one whose last track failed, and goes forward the same way.
|
||||
if tally.done > 0 {
|
||||
s.logger.Info(
|
||||
"slskd transfer stalled; keeping what arrived",
|
||||
"peer", username,
|
||||
"done", tally.done,
|
||||
"wanted", len(wanted),
|
||||
)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
return nil
|
||||
return fmt.Errorf(
|
||||
"%w: %s sent nothing in %s",
|
||||
ErrSlskdTimeout, username, s.stallAfter,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// transferTally is one poll's reading of the files a grab asked for.
|
||||
type transferTally struct {
|
||||
done, failed int
|
||||
bytes int64
|
||||
|
||||
// live are the requested transfers slskd is still working on,
|
||||
// which are what has to be cancelled if the grab is abandoned.
|
||||
live []slskdTransfer
|
||||
}
|
||||
|
||||
// tallyTransfers reads a peer's transfer list against the files a grab
|
||||
// asked for.
|
||||
//
|
||||
// A requested file slskd does not list at all is one it never accepted
|
||||
// — refused at enqueue, or dropped — and it will never reach a terminal
|
||||
// state to be counted by. Once absentExpired, such a file counts as
|
||||
// failed, or the grab would wait on it until the six-hour ceiling.
|
||||
func tallyTransfers(
|
||||
transfers []slskdTransfer,
|
||||
wanted map[string]bool,
|
||||
stale map[string]bool,
|
||||
absentExpired bool,
|
||||
) transferTally {
|
||||
seen := make(map[string]slskdTransfer, len(wanted))
|
||||
|
||||
for _, t := range transfers {
|
||||
if !wanted[t.Filename] || stale[t.ID] {
|
||||
continue
|
||||
}
|
||||
|
||||
seen[t.Filename] = t
|
||||
}
|
||||
|
||||
var out transferTally
|
||||
|
||||
for name := range wanted {
|
||||
t, ok := seen[name]
|
||||
if !ok {
|
||||
if absentExpired {
|
||||
out.failed++
|
||||
}
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
out.bytes += t.BytesTransferred
|
||||
|
||||
finished, succeeded := t.done()
|
||||
|
||||
switch {
|
||||
case !finished:
|
||||
out.live = append(out.live, t)
|
||||
case succeeded:
|
||||
out.done++
|
||||
default:
|
||||
out.failed++
|
||||
}
|
||||
}
|
||||
|
||||
return out
|
||||
}
|
||||
|
||||
// cancelTransfers asks slskd to cancel and forget transfers this grab
|
||||
// is abandoning. It runs on a context of its own: the usual reason to
|
||||
// be here is that the caller's context has just been cancelled, and a
|
||||
// cleanup that inherited it would never be sent.
|
||||
func (s *slskd) cancelTransfers(username string, live []slskdTransfer) {
|
||||
if len(live) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(
|
||||
context.Background(), slskdCancelTimeout,
|
||||
)
|
||||
defer cancel()
|
||||
|
||||
for _, t := range live {
|
||||
if t.ID == "" {
|
||||
continue
|
||||
}
|
||||
|
||||
endpoint := slskdDownloadsPath(username) + "/" +
|
||||
url.PathEscape(t.ID) + "?remove=true"
|
||||
|
||||
if err := s.client.delete(ctx, endpoint); err != nil {
|
||||
s.logger.Warn(
|
||||
"could not cancel slskd transfer",
|
||||
"peer", username,
|
||||
"file", t.Filename,
|
||||
"error", err,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -581,9 +784,7 @@ func (s *slskd) transfersFor(
|
||||
} `json:"directories"`
|
||||
}
|
||||
|
||||
if err := s.client.get(
|
||||
ctx, "/api/v0/transfers/downloads/"+username, &raw,
|
||||
); err != nil {
|
||||
if err := s.client.get(ctx, slskdDownloadsPath(username), &raw); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
|
||||
@@ -31,8 +31,18 @@ type slskdStub struct {
|
||||
transfers [][]slskdTransfer
|
||||
pollCount int
|
||||
|
||||
// before is what the downloads endpoint reports until something is
|
||||
// enqueued: records slskd already held from earlier attempts.
|
||||
before []slskdTransfer
|
||||
|
||||
// enqueued records what was requested for download.
|
||||
enqueued []map[string]any
|
||||
posted bool
|
||||
|
||||
// paths records the escaped path of every transfers call, and
|
||||
// cancelled the escaped request URI of every DELETE.
|
||||
paths []string
|
||||
cancelled []string
|
||||
|
||||
// unauthorized makes every call return 401.
|
||||
unauthorized bool
|
||||
@@ -87,7 +97,12 @@ func newSlskdStub(t *testing.T) *slskdStub {
|
||||
return
|
||||
}
|
||||
|
||||
if r.Method == http.MethodPost {
|
||||
s.mu.Lock()
|
||||
s.paths = append(s.paths, r.URL.EscapedPath())
|
||||
s.mu.Unlock()
|
||||
|
||||
switch r.Method {
|
||||
case http.MethodPost:
|
||||
var body []map[string]any
|
||||
|
||||
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
|
||||
@@ -96,15 +111,35 @@ func newSlskdStub(t *testing.T) *slskdStub {
|
||||
|
||||
s.mu.Lock()
|
||||
s.enqueued = body
|
||||
s.posted = true
|
||||
s.mu.Unlock()
|
||||
|
||||
w.WriteHeader(http.StatusCreated)
|
||||
|
||||
return
|
||||
case http.MethodDelete:
|
||||
s.mu.Lock()
|
||||
s.cancelled = append(s.cancelled, r.URL.RequestURI())
|
||||
s.mu.Unlock()
|
||||
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
|
||||
if !s.posted {
|
||||
before := s.before
|
||||
s.mu.Unlock()
|
||||
|
||||
writeJSON(t, w, map[string]any{
|
||||
"directories": []map[string]any{{"files": before}},
|
||||
})
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
idx := s.pollCount
|
||||
if idx >= len(s.transfers) {
|
||||
idx = len(s.transfers) - 1
|
||||
@@ -191,6 +226,11 @@ func newStubSlskd(t *testing.T, stub *slskdStub) (*slskd, string) {
|
||||
s.searchWait = 200 * time.Millisecond
|
||||
s.transferPoll = time.Millisecond
|
||||
|
||||
// Long enough that no existing test trips them by accident; the
|
||||
// tests about stalls and absences set their own.
|
||||
s.stallAfter = time.Minute
|
||||
s.absentGrace = time.Minute
|
||||
|
||||
return s, downloads
|
||||
}
|
||||
|
||||
@@ -565,3 +605,293 @@ func TestSlskdRequiresConfiguration(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// slskdAlbum is a two-file candidate from peer, with the files slskd
|
||||
// would have written already in place under downloads.
|
||||
func slskdAlbum(t *testing.T, downloads, peer string, arrived ...string) Candidate {
|
||||
t.Helper()
|
||||
|
||||
folder := filepath.Join(downloads, "Album")
|
||||
if err := os.MkdirAll(folder, 0o750); err != nil {
|
||||
t.Fatalf("mkdir: %v", err)
|
||||
}
|
||||
|
||||
for _, name := range arrived {
|
||||
if err := os.WriteFile(
|
||||
filepath.Join(folder, name), []byte("audio"), 0o600,
|
||||
); err != nil {
|
||||
t.Fatalf("write: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
return Candidate{
|
||||
Files: []CandidateFile{
|
||||
{Path: `\s\Album\01 A.flac`, Size: 500, IsAudio: true},
|
||||
{Path: `\s\Album\02 B.flac`, Size: 500, IsAudio: true},
|
||||
},
|
||||
TotalSize: 1000,
|
||||
Payload: map[string]string{"username": peer},
|
||||
}
|
||||
}
|
||||
|
||||
// cancelledURIs returns what the stub was asked to cancel.
|
||||
func (s *slskdStub) cancelledURIs() []string {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
return append([]string(nil), s.cancelled...)
|
||||
}
|
||||
|
||||
// A peer that queues us and never sends a byte is given up on, and the
|
||||
// queued transfers are cancelled in slskd rather than left to start
|
||||
// hours later for a request nobody is waiting on.
|
||||
func TestSlskdGrabGivesUpOnAStalledPeer(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
stub := newSlskdStub(t)
|
||||
stub.transfers = [][]slskdTransfer{{
|
||||
{ID: "t1", Filename: `\s\Album\01 A.flac`, State: "Queued, Remotely"},
|
||||
{ID: "t2", Filename: `\s\Album\02 B.flac`, State: "Queued, Remotely"},
|
||||
}}
|
||||
|
||||
s, downloads := newStubSlskd(t, stub)
|
||||
s.stallAfter = 30 * time.Millisecond
|
||||
|
||||
_, err := s.Grab(
|
||||
context.Background(), slskdAlbum(t, downloads, "peer"), t.TempDir(), nil,
|
||||
)
|
||||
if !errors.Is(err, ErrSlskdTimeout) {
|
||||
t.Fatalf("error = %v, want ErrSlskdTimeout", err)
|
||||
}
|
||||
|
||||
got := stub.cancelledURIs()
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("cancelled %v, want both queued transfers", got)
|
||||
}
|
||||
|
||||
for _, uri := range got {
|
||||
if !strings.HasSuffix(uri, "?remove=true") {
|
||||
t.Errorf("cancel %s does not remove the record", uri)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A folder that stalls on its last track goes forward with what
|
||||
// arrived, the same as one whose last track failed; the importer's
|
||||
// completeness check decides whether that is enough.
|
||||
func TestSlskdGrabKeepsWhatArrivedBeforeAStall(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
stub := newSlskdStub(t)
|
||||
stub.transfers = [][]slskdTransfer{{
|
||||
{
|
||||
ID: "t1", Filename: `\s\Album\01 A.flac`,
|
||||
State: "Completed, Succeeded", BytesTransferred: 500,
|
||||
},
|
||||
{ID: "t2", Filename: `\s\Album\02 B.flac`, State: "Queued, Remotely"},
|
||||
}}
|
||||
|
||||
s, downloads := newStubSlskd(t, stub)
|
||||
s.stallAfter = 30 * time.Millisecond
|
||||
|
||||
got, err := s.Grab(
|
||||
context.Background(),
|
||||
slskdAlbum(t, downloads, "peer", "01 A.flac"),
|
||||
t.TempDir(), nil,
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("Grab: %v", err)
|
||||
}
|
||||
|
||||
if len(got.Files) != 1 {
|
||||
t.Errorf("collected %d files, want the 1 that arrived", len(got.Files))
|
||||
}
|
||||
|
||||
if cancelled := stub.cancelledURIs(); len(cancelled) != 1 ||
|
||||
!strings.Contains(cancelled[0], "/t2") {
|
||||
t.Errorf("cancelled %v, want only the stalled t2", cancelled)
|
||||
}
|
||||
}
|
||||
|
||||
// Progress is what holds the stall timer off. A transfer that keeps
|
||||
// moving bytes is never abandoned, however long it takes.
|
||||
func TestSlskdGrabWaitsOnATransferThatIsMoving(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
stub := newSlskdStub(t)
|
||||
|
||||
for b := int64(1); b <= 100; b++ {
|
||||
stub.transfers = append(stub.transfers, []slskdTransfer{
|
||||
{ID: "t1", Filename: `\s\Album\01 A.flac`, State: "InProgress", BytesTransferred: b},
|
||||
{ID: "t2", Filename: `\s\Album\02 B.flac`, State: "Queued, Remotely"},
|
||||
})
|
||||
}
|
||||
|
||||
stub.transfers = append(stub.transfers, []slskdTransfer{
|
||||
{
|
||||
ID: "t1",
|
||||
Filename: `\s\Album\01 A.flac`,
|
||||
State: "Completed, Succeeded",
|
||||
BytesTransferred: 500,
|
||||
},
|
||||
{
|
||||
ID: "t2",
|
||||
Filename: `\s\Album\02 B.flac`,
|
||||
State: "Completed, Succeeded",
|
||||
BytesTransferred: 500,
|
||||
},
|
||||
})
|
||||
|
||||
s, downloads := newStubSlskd(t, stub)
|
||||
// A hundred polls take several times the stall window; each one
|
||||
// moves a byte. The window is kept well above one poll so a
|
||||
// descheduled test runner does not read as a stall.
|
||||
s.transferPoll = 5 * time.Millisecond
|
||||
s.stallAfter = 150 * time.Millisecond
|
||||
|
||||
got, err := s.Grab(
|
||||
context.Background(),
|
||||
slskdAlbum(t, downloads, "peer", "01 A.flac", "02 B.flac"),
|
||||
t.TempDir(), nil,
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("Grab: %v", err)
|
||||
}
|
||||
|
||||
if len(got.Files) != 2 {
|
||||
t.Errorf("collected %d files, want 2", len(got.Files))
|
||||
}
|
||||
}
|
||||
|
||||
// A file slskd never lists was refused at enqueue and will never reach
|
||||
// a terminal state. It counts as failed once the grace period is up,
|
||||
// rather than being waited on until the six-hour ceiling.
|
||||
func TestSlskdGrabCountsAnUnlistedFileAsFailed(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
stub := newSlskdStub(t)
|
||||
stub.transfers = [][]slskdTransfer{{
|
||||
{
|
||||
ID: "t1", Filename: `\s\Album\01 A.flac`,
|
||||
State: "Completed, Succeeded", BytesTransferred: 500,
|
||||
},
|
||||
}}
|
||||
|
||||
s, downloads := newStubSlskd(t, stub)
|
||||
s.absentGrace = 20 * time.Millisecond
|
||||
|
||||
got, err := s.Grab(
|
||||
context.Background(),
|
||||
slskdAlbum(t, downloads, "peer", "01 A.flac"),
|
||||
t.TempDir(), nil,
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("Grab: %v", err)
|
||||
}
|
||||
|
||||
if len(got.Files) != 1 {
|
||||
t.Errorf("collected %d files, want 1", len(got.Files))
|
||||
}
|
||||
}
|
||||
|
||||
// A finished record left by an earlier attempt at the same file is not
|
||||
// this attempt's answer. Without the snapshot it would fail the grab on
|
||||
// the first poll, before the new transfer had started.
|
||||
func TestSlskdGrabIgnoresAnEarlierAttemptsRecord(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
stale := slskdTransfer{
|
||||
ID: "old", Filename: `\s\Album\01 A.flac`, State: "Completed, Errored",
|
||||
}
|
||||
|
||||
stub := newSlskdStub(t)
|
||||
stub.before = []slskdTransfer{stale}
|
||||
stub.transfers = [][]slskdTransfer{
|
||||
{stale},
|
||||
{
|
||||
stale,
|
||||
{
|
||||
ID: "new", Filename: `\s\Album\01 A.flac`,
|
||||
State: "Completed, Succeeded", BytesTransferred: 500,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
s, downloads := newStubSlskd(t, stub)
|
||||
|
||||
c := slskdAlbum(t, downloads, "peer", "01 A.flac")
|
||||
c.Files = c.Files[:1]
|
||||
|
||||
got, err := s.Grab(context.Background(), c, t.TempDir(), nil)
|
||||
if err != nil {
|
||||
t.Fatalf("Grab: %v", err)
|
||||
}
|
||||
|
||||
if len(got.Files) != 1 {
|
||||
t.Errorf("collected %d files, want 1", len(got.Files))
|
||||
}
|
||||
}
|
||||
|
||||
// Cancelling the download cancels the transfer in slskd too. The
|
||||
// cleanup must not inherit the cancelled context, or it is never sent.
|
||||
func TestSlskdGrabCancelsTransfersWhenTheCallerGivesUp(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
stub := newSlskdStub(t)
|
||||
stub.transfers = [][]slskdTransfer{{
|
||||
{ID: "t1", Filename: `\s\Album\01 A.flac`, State: "InProgress", BytesTransferred: 10},
|
||||
{ID: "t2", Filename: `\s\Album\02 B.flac`, State: "Queued, Remotely"},
|
||||
}}
|
||||
|
||||
s, downloads := newStubSlskd(t, stub)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
|
||||
defer cancel()
|
||||
|
||||
_, err := s.Grab(ctx, slskdAlbum(t, downloads, "peer"), t.TempDir(), nil)
|
||||
if !errors.Is(err, ErrSlskdTimeout) {
|
||||
t.Fatalf("error = %v, want ErrSlskdTimeout", err)
|
||||
}
|
||||
|
||||
if got := stub.cancelledURIs(); len(got) != 2 {
|
||||
t.Errorf("cancelled %v, want both live transfers", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Soulseek usernames carry spaces and punctuation; spliced raw into the
|
||||
// path, a name with a slash addresses a different endpoint entirely.
|
||||
func TestSlskdEscapesTheUsername(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
stub := newSlskdStub(t)
|
||||
stub.transfers = [][]slskdTransfer{{
|
||||
{
|
||||
ID: "t1", Filename: `\s\Album\01 A.flac`,
|
||||
State: "Completed, Succeeded", BytesTransferred: 500,
|
||||
},
|
||||
{
|
||||
ID: "t2", Filename: `\s\Album\02 B.flac`,
|
||||
State: "Completed, Succeeded", BytesTransferred: 500,
|
||||
},
|
||||
}}
|
||||
|
||||
s, downloads := newStubSlskd(t, stub)
|
||||
|
||||
if _, err := s.Grab(
|
||||
context.Background(),
|
||||
slskdAlbum(t, downloads, "dj a/b", "01 A.flac", "02 B.flac"),
|
||||
t.TempDir(), nil,
|
||||
); err != nil {
|
||||
t.Fatalf("Grab: %v", err)
|
||||
}
|
||||
|
||||
stub.mu.Lock()
|
||||
paths := append([]string(nil), stub.paths...)
|
||||
stub.mu.Unlock()
|
||||
|
||||
for _, p := range paths {
|
||||
if p != "/api/v0/transfers/downloads/dj%20a%2Fb" {
|
||||
t.Errorf("transfers call went to %s", p)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -735,6 +735,45 @@ func AutoPickVeto(
|
||||
return ""
|
||||
}
|
||||
|
||||
// autoAcceptable reports whether auto-pick may take this one candidate
|
||||
// without asking: the request is anchored to a tracklist, and the
|
||||
// candidate is inside the user's guardrails and clears the match and
|
||||
// quality bars. It is AutoPickVeto's test applied to a single
|
||||
// candidate, which is what falling back to a second choice needs.
|
||||
func autoAcceptable(dl Download, c Candidate, prefs AutoDownloadPrefs) bool {
|
||||
return dl.Anchored() &&
|
||||
len(dl.Expected) > 0 &&
|
||||
prefs.eligible(c, dl.runtimeMillis()) &&
|
||||
c.Match.Overall >= minMatch &&
|
||||
c.Quality.Overall >= minQuality
|
||||
}
|
||||
|
||||
// autoPick returns the candidate auto-pick takes: the best-ranked one
|
||||
// it may take at all.
|
||||
//
|
||||
// That is not `ranked[0]`. AutoPickVeto judges the best candidate
|
||||
// *inside* the guardrails, so when the overall best is outside them —
|
||||
// over the size ceiling, say — the veto passes on the strength of the
|
||||
// second, and grabbing the first would download exactly the copy the
|
||||
// user said not to take unattended.
|
||||
func autoPick(
|
||||
dl Download,
|
||||
ranked []Candidate,
|
||||
prefs AutoDownloadPrefs,
|
||||
) (Candidate, bool) {
|
||||
if AutoPickVeto(dl, ranked, prefs) != "" {
|
||||
return Candidate{}, false
|
||||
}
|
||||
|
||||
for _, c := range ranked {
|
||||
if autoAcceptable(dl, c, prefs) {
|
||||
return c, true
|
||||
}
|
||||
}
|
||||
|
||||
return Candidate{}, false
|
||||
}
|
||||
|
||||
// mergeMatched copies MatchedTo assignments from the audio-only slice
|
||||
// back onto the full file list.
|
||||
func mergeMatched(all, matched []CandidateFile) []CandidateFile {
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
|
||||
"yellowjacket/backend/coverart"
|
||||
"yellowjacket/backend/database/sql/sqlcgen"
|
||||
"yellowjacket/internal/testfixtures"
|
||||
)
|
||||
|
||||
// TestScan_StoresOnlyCoverTiers pins the size decision: a scan writes
|
||||
@@ -26,10 +27,9 @@ func TestScan_StoresOnlyCoverTiers(t *testing.T) {
|
||||
|
||||
lib, db := setupTestLibrary(t)
|
||||
|
||||
root, err := filepath.Abs("../../test_data/music_library_test")
|
||||
if err != nil {
|
||||
t.Fatalf("resolve fixture path: %v", err)
|
||||
}
|
||||
// Load skips when the fixture library has not been generated, as
|
||||
// every other fixture test does.
|
||||
root := testfixtures.Load(t).Root()
|
||||
|
||||
library, err := db.Queries.CreateLibrary(lib.ctx, sqlcgen.CreateLibraryParams{
|
||||
Name: "Fixtures",
|
||||
|
||||
Reference in new issue
Block a user