Batch: explore catalog + UI, Go 1.26, Soulseek downloads, CI fixes #273

Merged
yonlu merged 21 commits from batch/258-272 into main 2026-09-27 01:45:19 +00:00
6 changed files with 420 additions and 24 deletions
Showing only changes of commit 9710c11476 - Show all commits

No files matched your search

+12 -11
View File
@@ -79,9 +79,9 @@ func TestConcurrencyForPrefersOverrideThenKind(t *testing.T) {
want int
}{
{
name: "slskd defaults to one",
name: "slskd defaults to a few peers",
cfg: Config{Kind: KindSlskd},
want: 1,
want: 3,
},
{
name: "usenet defaults higher",
@@ -92,9 +92,9 @@ func TestConcurrencyForPrefersOverrideThenKind(t *testing.T) {
name: "explicit override wins",
cfg: Config{
Kind: KindSlskd,
Settings: map[string]string{concurrencyKey: "3"},
Settings: map[string]string{concurrencyKey: "1"},
},
want: 3,
want: 1,
},
{
name: "nonsense override falls back",
@@ -102,7 +102,7 @@ func TestConcurrencyForPrefersOverrideThenKind(t *testing.T) {
Kind: KindSlskd,
Settings: map[string]string{concurrencyKey: "not a number"},
},
want: 1,
want: 3,
},
{
name: "zero override falls back",
@@ -110,7 +110,7 @@ func TestConcurrencyForPrefersOverrideThenKind(t *testing.T) {
Kind: KindSlskd,
Settings: map[string]string{concurrencyKey: "0"},
},
want: 1,
want: 3,
},
{
name: "unknown kind falls back to the global default",
@@ -126,9 +126,9 @@ func TestConcurrencyForPrefersOverrideThenKind(t *testing.T) {
}
}
// The reason the per-provider cap exists: a Soulseek daemon capped at
// one transfer must serialize, even when the global cap would allow
// more and the user has queued several albums at once.
// The reason the per-provider cap exists: a daemon capped at one
// transfer must serialize, even when the global cap would allow more and
// the user has queued several albums at once.
func TestPerProviderCapSerializesTransfers(t *testing.T) {
t.Parallel()
@@ -142,6 +142,7 @@ func TestPerProviderCapSerializesTransfers(t *testing.T) {
ID: 1,
Kind: KindSlskd,
Priority: 50,
Settings: map[string]string{concurrencyKey: "1"},
}, slow)
// Three requests against the same one-at-a-time provider.
@@ -210,8 +211,8 @@ func TestSyncSemaphoresReplacesChangedLimits(t *testing.T) {
f.manager.installProvider(Config{ID: 1, Kind: KindSlskd}, nil)
first := f.manager.semaphoreFor(1)
if cap(first) != 1 {
t.Fatalf("slskd semaphore cap = %d, want 1", cap(first))
if want := kindConcurrency[KindSlskd]; cap(first) != want {
t.Fatalf("slskd semaphore cap = %d, want %d", cap(first), want)
}
// Same limit: the semaphore is kept, so in-flight accounting is not
+77
View File
@@ -0,0 +1,77 @@
package download
import (
"context"
"sync"
)
// keyedLock is a set of mutexes created on demand, one per key, that
// honour a context while waiting. An entry lives only while someone
// holds or waits on it, so a key per Soulseek peer or per folder name
// does not accumulate for the life of the process.
type keyedLock[K comparable] struct {
mu sync.Mutex
held map[K]*keyedEntry
}
type keyedEntry struct {
ch chan struct{}
// refs counts holders and waiters; the entry is dropped at zero.
refs int
}
// acquire blocks until k is free or ctx ends, and returns the function
// that frees it.
func (l *keyedLock[K]) acquire(ctx context.Context, k K) (func(), error) {
l.mu.Lock()
if l.held == nil {
l.held = map[K]*keyedEntry{}
}
e, ok := l.held[k]
if !ok {
e = &keyedEntry{ch: make(chan struct{}, 1)}
l.held[k] = e
}
e.refs++
l.mu.Unlock()
select {
case e.ch <- struct{}{}:
case <-ctx.Done():
l.drop(k, e)
return nil, ctx.Err()
}
var once sync.Once
return func() {
once.Do(func() {
<-e.ch
l.drop(k, e)
})
}, nil
}
func (l *keyedLock[K]) drop(k K, e *keyedEntry) {
l.mu.Lock()
defer l.mu.Unlock()
e.refs--
if e.refs == 0 {
delete(l.held, k)
}
}
// size reports how many keys are held or awaited, for tests.
func (l *keyedLock[K]) size() int {
l.mu.Lock()
defer l.mu.Unlock()
return len(l.held)
}
+39 -6
View File
@@ -59,13 +59,15 @@ const concurrencyKey = "maxConcurrent"
// A single global cap is the wrong shape here: usenet and torrent
// clients are built to run many transfers at once and are throttled by
// bandwidth, while Soulseek transfers come from one person's home
// upload slot. Hitting the same peer with parallel requests gets you
// queued behind everyone else at best and banned at worst, so slskd is
// capped at one — the polite number, and the one that actually
// completes fastest, because a Soulseek peer serves one file at a time
// regardless of how many you ask for.
// upload slot. Politeness there is per *peer* — asking one user for two
// folders at once gets you queued behind everyone else at best and
// banned at worst — and the manager holds that line separately, one
// grab per peer (peerLocks). Two different users do not compete for
// anyone's slot, so the daemon-wide number only bounds how many peers
// are asked at once, and one slow peer no longer serialises every other
// Soulseek download behind it.
var kindConcurrency = map[Kind]int{
KindSlskd: 1,
KindSlskd: 3,
KindYtDlp: 2,
KindQBittorrent: 4,
KindSABnzbd: 4,
@@ -155,6 +157,11 @@ type Manager struct {
semMu sync.Mutex
provSem map[int64]chan struct{}
// peerLocks holds one grab per Soulseek peer, taken before any
// slot: a grab waiting for a busy peer must not sit on a provider
// slot another peer could be using.
peerLocks keyedLock[peerKey]
// delegatePoll is how often delegating managers are asked for
// status. A field rather than the constant so tests can drive the
// full delegate flow without sleeping through it.
@@ -790,6 +797,23 @@ func (m *Manager) grab(
// whole list for six hours.
const maxGrabAttempts = 3
// peerKey names one Soulseek user on one daemon. The same username on
// two daemons is two logins and two queues.
type peerKey struct {
provider int64
peer string
}
// peerKeyFor returns the peer a candidate is fetched from, when the
// source is one where asking a peer for two things at once is rude.
func peerKeyFor(c Candidate) (peerKey, bool) {
if c.Kind != KindSlskd || c.Origin == "" {
return peerKey{}, false
}
return peerKey{provider: c.ProviderID, peer: c.Origin}, true
}
// grabOutcome is how one candidate's attempt ended.
type grabOutcome struct {
item DownloadItem
@@ -825,6 +849,15 @@ func (m *Manager) attemptGrab(
}
if !plan.delegated() {
if key, ok := peerKeyFor(c); ok {
release, err := m.peerLocks.acquire(ctx, key)
if err != nil {
return grabOutcome{err: err}
}
defer release()
}
provSem := m.semaphoreFor(plan.transportID)
select {
+224
View File
@@ -0,0 +1,224 @@
package download
import (
"context"
"errors"
"path/filepath"
"sync"
"testing"
"time"
)
// Soulseek politeness is per peer, not per daemon (#272).
func TestKeyedLockSerialisesOneKeyOnly(t *testing.T) {
t.Parallel()
var l keyedLock[string]
ctx := context.Background()
releaseA, err := l.acquire(ctx, "a")
if err != nil {
t.Fatalf("acquire a: %v", err)
}
// Another key is free while "a" is held.
releaseB, err := l.acquire(ctx, "b")
if err != nil {
t.Fatalf("acquire b: %v", err)
}
releaseB()
// The same key waits, and gives up with its context.
short, cancel := context.WithTimeout(ctx, 20*time.Millisecond)
defer cancel()
if _, err := l.acquire(short, "a"); !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("second acquire of a held key = %v, want the deadline", err)
}
releaseA()
releaseA() // Idempotent: a second call must not free someone else's hold.
if n := l.size(); n != 0 {
t.Errorf("%d keys left behind, want none once nobody holds or waits", n)
}
}
// grabEach runs one grab per candidate and returns a function that waits
// for all of them; grabAll's reasons for waiting apply.
func grabEach(t *testing.T, f managerFixture, cands []Candidate) func() {
t.Helper()
ctx := context.Background()
var wg sync.WaitGroup
for i, c := range cands {
dl := fourTrackDownload()
dl.ID = "dl-" + string(rune('a'+i))
if err := f.store.CreateDownload(ctx, dl); err != nil {
t.Fatalf("CreateDownload: %v", err)
}
wg.Add(1)
go func() {
defer wg.Done()
f.manager.grab(ctx, dl, c, nil, false)
}()
}
return func() {
done := make(chan struct{})
go func() {
wg.Wait()
close(done)
}()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Error("transfers did not finish")
}
}
}
func slskdCandidates(p *FakeProvider, peers ...string) []Candidate {
out := make([]Candidate, 0, len(peers))
for i, peer := range peers {
c := p.Candidates[0]
c.ID = c.ID + "-" + itoa(i)
c.Kind = KindSlskd
c.ProviderID = 1
c.Origin = peer
out = append(out, c)
}
return out
}
// Three albums from one user are asked for one at a time, even though
// the daemon would allow three transfers.
func TestOnePeerIsAskedForOneThingAtATime(t *testing.T) {
t.Parallel()
f := newManagerFixture(t)
f.manager.SetMaxConcurrent(4)
p := fakeWithAlbum(1, "slskd", ".flac")
p.GrabGate = make(chan struct{})
f.manager.installProvider(Config{ID: 1, Kind: KindSlskd, Priority: 50}, p)
wait := grabEach(t, f, slskdCandidates(p, "alice", "alice", "alice"))
waitFor(t, func() bool { return p.GrabCallCount() >= 1 }, "no grab started")
time.Sleep(150 * time.Millisecond)
if got := p.MaxParallelGrabs(); got != 1 {
t.Errorf("%d simultaneous grabs from one peer, want 1", got)
}
close(p.GrabGate)
waitFor(t, func() bool { return p.GrabCallCount() == 3 }, "queued grabs never ran")
wait()
if n := f.manager.peerLocks.size(); n != 0 {
t.Errorf("%d peer locks left behind", n)
}
}
// Different users run at once, up to the daemon's cap — the point of
// the change: one slow peer no longer holds up every other.
func TestDifferentPeersRunTogether(t *testing.T) {
t.Parallel()
f := newManagerFixture(t)
f.manager.SetMaxConcurrent(8)
p := fakeWithAlbum(1, "slskd", ".flac")
p.GrabGate = make(chan struct{})
f.manager.installProvider(Config{ID: 1, Kind: KindSlskd, Priority: 50}, p)
wait := grabEach(t, f, slskdCandidates(p, "alice", "bob", "carol", "dave"))
waitFor(
t,
func() bool { return p.MaxParallelGrabs() >= kindConcurrency[KindSlskd] },
"different peers were serialised",
)
time.Sleep(100 * time.Millisecond)
if got := p.MaxParallelGrabs(); got != kindConcurrency[KindSlskd] {
t.Errorf("%d simultaneous grabs, want the daemon cap %d", got, kindConcurrency[KindSlskd])
}
close(p.GrabGate)
wait()
}
func TestSlskdLocalFolders(t *testing.T) {
t.Parallel()
s := &slskd{downloadsPath: "/dl"}
got := s.localFolders(Candidate{Files: []CandidateFile{
{Path: `\m\The Wall\CD2\01 Hey You.flac`},
{Path: `\m\The Wall\CD1\01 In The Flesh.flac`},
{Path: `\m\The Wall\CD1\02 The Thin Ice.flac`},
}})
want := []string{filepath.Join("/dl", "CD1"), filepath.Join("/dl", "CD2")}
if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] {
t.Errorf("localFolders = %q, want %q", got, want)
}
}
// Two peers' "Greatest Hits" land in one slskd directory, so the second
// grab does not enqueue until the first has collected its files.
func TestSlskdSameFolderNameWaits(t *testing.T) {
t.Parallel()
stub := newSlskdStub(t)
s, downloads := newStubSlskd(t, stub)
c := Candidate{
Payload: map[string]string{"username": "bob"},
Files: []CandidateFile{
{Path: `\music\Greatest Hits\01 Intro.flac`, Size: 1, IsAudio: true},
},
}
release, err := lockSlskdFolders(
context.Background(), []string{filepath.Join(downloads, "Greatest Hits")},
)
if err != nil {
t.Fatalf("lock: %v", err)
}
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel()
if _, err := s.Grab(ctx, c, t.TempDir(), nil); !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("Grab = %v, want it to wait on the held folder", err)
}
release()
stub.mu.Lock()
posted := stub.posted
stub.mu.Unlock()
if posted {
t.Error("enqueued transfers into a folder another grab held")
}
}
+8 -7
View File
@@ -236,17 +236,18 @@ func Register(d Descriptor, c Constructor) {
}
// concurrencyField describes the per-provider transfer limit, with help
// text explaining why the default is what it is — a user who raises
// slskd from 1 to 8 and gets themselves queued behind every other
// Soulseek user deserves to have been warned.
// text explaining what the number means where it means something
// unusual: on slskd it counts peers, since each peer is only ever asked
// for one folder at a time whatever it is set to.
func concurrencyField(k Kind) Field {
help := "Maximum simultaneous transfers from this client."
if k == KindSlskd {
help = "Maximum simultaneous transfers. Soulseek peers serve " +
"one file at a time and queue or ban clients that ask for " +
"more, so 1 is both the polite setting and usually the " +
"fastest."
help = "How many Soulseek users to download from at once. " +
"Each user is only ever asked for one album at a time, " +
"since peers queue or ban clients that ask for more; " +
"this bounds how many different users are asked in " +
"parallel."
}
return Field{
+60
View File
@@ -10,6 +10,7 @@ import (
"path"
"path/filepath"
"regexp"
"slices"
"strings"
"time"
@@ -742,6 +743,12 @@ func (s *slskd) Grab(
// 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.
release, err := lockSlskdFolders(ctx, s.localFolders(c))
if err != nil {
return Result{}, err
}
defer release()
stale := s.terminalTransferIDs(ctx, username)
wanted := make([]map[string]any, 0, len(c.Files))
@@ -767,6 +774,59 @@ func (s *slskd) Grab(
return s.collect(c, dst)
}
// slskdFolders serialises grabs that land in the same local folder.
//
// slskd names a download's directory after the remote *leaf* folder, so
// two different albums both shared as "Greatest Hits" — or any two
// multi-disc rips, whose leaves are "CD1" and "CD2" — are written into
// one directory, and collect finds files by name there. Run at once,
// a file one peer never sent is filled by the other peer's file of the
// same name. One grab per peer made that impossible; several peers at
// once makes it likely. It is package-level and keyed on the full
// path because two configured clients can share one daemon.
var slskdFolders keyedLock[string]
// localFolders returns the directories under downloadsPath a candidate's
// files will be written to, sorted so every grab takes them in the same
// order and two cannot each hold what the other waits for.
func (s *slskd) localFolders(c Candidate) []string {
var out []string
for _, f := range c.Files {
norm := strings.ReplaceAll(f.Path, `\`, "/")
out = append(out, filepath.Join(s.downloadsPath, path.Base(path.Dir(norm))))
}
slices.Sort(out)
return slices.Compact(out)
}
// lockSlskdFolders takes every folder in order, releasing what it holds
// if the context ends part way.
func lockSlskdFolders(ctx context.Context, folders []string) (func(), error) {
releases := make([]func(), 0, len(folders))
releaseAll := func() {
for _, r := range slices.Backward(releases) {
r()
}
}
for _, f := range folders {
r, err := slskdFolders.acquire(ctx, f)
if err != nil {
releaseAll()
return nil, err
}
releases = append(releases, r)
}
return releaseAll, nil
}
// 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.