diff --git a/ap/ap.go b/ap/ap.go index ee348cc3..c2fc8507 100644 --- a/ap/ap.go +++ b/ap/ap.go @@ -14,6 +14,7 @@ import ( "io" "net" "sync" + "sync/atomic" "time" "github.com/cenkalti/backoff/v4" @@ -26,7 +27,15 @@ import ( "google.golang.org/protobuf/proto" ) -const pongAckInterval = 120 * time.Second +// The accesspoint pings the device about every pongAckInterval, at times a +// little later, and answers each Pong with a PongAck. pongAckMargin is the +// only slack for that lateness: a connection that has carried no PongAck for +// pongAckInterval+pongAckMargin, and that long since it was established, has +// gone silent. +const ( + pongAckInterval = 120 * time.Second + pongAckMargin = 20 * time.Second +) // handshakeTimeout bounds the key exchange, challenge and authentication when the caller's // context has no deadline: an accesspoint that stops answering mid-handshake would otherwise @@ -43,6 +52,10 @@ func (e *AccesspointLoginError) Error() string { return fmt.Sprintf("accesspoint login failed: %s %v", e.Message.ErrorCode.String(), e.Message.ErrorDescription) } +// ConnID identifies one connection to the accesspoint. A reconnection replaces +// it with a connection of another ConnID. +type ConnID uint64 + type Accesspoint struct { log librespot.Logger @@ -65,11 +78,18 @@ type Accesspoint struct { recvChans map[PacketType][]chan Packet recvChansLock sync.RWMutex lastPongAck time.Time + pongAckConn ConnID lastPongAckLock sync.Mutex - // connMu protects conn, encConn, and welcome pointer state. + // connMu protects conn, encConn, connID, replaced, and welcome pointer state. connMu sync.RWMutex + connID ConnID welcome *pb.APWelcome + // replaced is closed when the connection connID is replaced. + replaced chan struct{} + // closedConn is the last connection this side closed, which recvLoop is + // replacing unless the accesspoint is closing. + closedConn atomic.Uint64 } func NewAccesspoint(log librespot.Logger, addr librespot.GetAddressFunc, deviceId string) *Accesspoint { @@ -110,13 +130,8 @@ func (ap *Accesspoint) init(ctx context.Context) (err error) { conn, err := proxy.Dial(ctx_, "tcp", addr) cancel() if err == nil { - // close previous connection if any - if ap.conn != nil { - _ = ap.conn.Close() - } - // we assign to ap.conn after because if Dial fails we'll have a nil ap.conn which we don't want - ap.conn = conn + ap.setConn(conn) ap.log.Debugf("connected to %s", addr) return nil } else if attempts >= 6 { @@ -128,6 +143,20 @@ func (ap *Accesspoint) init(ctx context.Context) (err error) { } } +// setConn replaces the connection. Its caller holds connMu for writing. +func (ap *Accesspoint) setConn(conn net.Conn) { + if ap.conn != nil { + _ = ap.conn.Close() + } + if ap.replaced != nil { + close(ap.replaced) + } + + ap.conn = conn + ap.connID++ + ap.replaced = make(chan struct{}) +} + func (ap *Accesspoint) ConnectSpotifyToken(ctx context.Context, username, token string) error { creds := &pb.LoginCredentials{ Typ: pb.AuthenticationType_AUTHENTICATION_SPOTIFY_TOKEN.Enum(), @@ -269,27 +298,71 @@ func (ap *Accesspoint) Close() { } func (ap *Accesspoint) Send(ctx context.Context, pktType PacketType, payload []byte) error { - ap.connMu.RLock() - select { - case <-ap.done: + _, err := ap.SendConn(ctx, pktType, payload) + return err +} + +// SendConn is Send, and also returns the connection the packet was written to, +// for DropConn. +// +// A packet written to a connection this side has closed is sent again on its +// replacement: recvLoop takes connMu for the reconnection only once its read +// has failed, so a packet sent right after DropConn or closeConn would +// otherwise fail on the closed socket. +func (ap *Accesspoint) SendConn(ctx context.Context, pktType PacketType, payload []byte) (ConnID, error) { + for { + ap.connMu.RLock() + select { + case <-ap.done: + ap.connMu.RUnlock() + return 0, ErrAccesspointClosed + default: + } + + encConn, id, replaced := ap.encConn, ap.connID, ap.replaced ap.connMu.RUnlock() - return ErrAccesspointClosed - default: - } - encConn := ap.encConn - ap.connMu.RUnlock() + err := encConn.sendPacket(ctx, pktType, payload) + if err == nil { + return id, nil + } - if err := encConn.sendPacket(ctx, pktType, payload); err != nil { select { case <-ap.done: - return ErrAccesspointClosed + return id, ErrAccesspointClosed default: - return err + } + + if id == 0 || ConnID(ap.closedConn.Load()) != id { + return id, err + } + + select { + case <-replaced: + case <-ap.done: + return id, ErrAccesspointClosed + case <-ctx.Done(): + return id, ctx.Err() } } +} - return nil +// DropConn closes the connection id when an answer expected on it did not +// come, so that it is replaced. It does nothing if that connection was already +// replaced, and never waits: only a connection in progress (Connect, or +// recvLoop's reconnection) holds connMu for writing, and it is already +// replacing the connection. +func (ap *Accesspoint) DropConn(id ConnID) { + if !ap.connMu.TryRLock() { + return + } + conn, current := ap.conn, ap.connID + ap.connMu.RUnlock() + + if conn != nil && current == id { + ap.closedConn.Store(uint64(id)) + _ = conn.Close() + } } // recvChanSize buffers each receiver so that recvLoop can keep draining the @@ -355,7 +428,9 @@ loop: switch pkt { case PacketTypePing: - if err := ap.Send(ap.ctx, PacketTypePong, payload); err != nil { + // Not Send: recvLoop is what replaces a closed connection, so + // it must not wait for that. + if err := ap.encConn.sendPacket(ap.ctx, PacketTypePong, payload); err != nil { ap.log.WithError(err).Errorf("failed sending Pong packet") break loop } @@ -421,27 +496,29 @@ loop: } func (ap *Accesspoint) pongAckTicker() { - ticker := time.NewTicker(pongAckInterval) + const timeout = pongAckInterval + pongAckMargin + timer := time.NewTimer(timeout) + defer timer.Stop() -loop: for { select { case <-ap.done: - break loop - case <-ticker.C: - timePassed := ap.timeSinceLastPongAck() - if timePassed > pongAckInterval { - ap.log.Errorf("did not receive last pong ack from accesspoint, %.0fs passed", timePassed.Seconds()) - - // closing the connection should make the read on the "recvLoop" fail, - // continue hoping for a new connection - ap.closeConn() + return + case <-timer.C: + conn, timePassed := ap.sinceLastPongAck() + if timePassed < timeout { + timer.Reset(timeout - timePassed) continue } + + ap.log.Errorf("did not receive last pong ack from accesspoint, %.0fs passed", timePassed.Seconds()) + + // closing the connection should make the read on the "recvLoop" fail, + // continue hoping for a new connection + ap.DropConn(conn) + timer.Reset(timeout) } } - - ticker.Stop() } func (ap *Accesspoint) reconnect() (err error) { @@ -466,9 +543,12 @@ func (ap *Accesspoint) reconnect() (err error) { return nil } +// resetPongAckDeadline counts a new connection as a PongAck. Its caller holds +// connMu. func (ap *Accesspoint) resetPongAckDeadline() { ap.lastPongAckLock.Lock() - ap.lastPongAck = time.Now().Add(pongAckInterval) + ap.lastPongAck = time.Now() + ap.pongAckConn = ap.connID ap.lastPongAckLock.Unlock() } @@ -478,18 +558,19 @@ func (ap *Accesspoint) notePongAck() { ap.lastPongAckLock.Unlock() } -func (ap *Accesspoint) timeSinceLastPongAck() time.Duration { +func (ap *Accesspoint) sinceLastPongAck() (ConnID, time.Duration) { ap.lastPongAckLock.Lock() defer ap.lastPongAckLock.Unlock() - return time.Since(ap.lastPongAck) + return ap.pongAckConn, time.Since(ap.lastPongAck) } func (ap *Accesspoint) closeConn() { ap.connMu.RLock() - conn := ap.conn + conn, id := ap.conn, ap.connID ap.connMu.RUnlock() if conn != nil { + ap.closedConn.Store(uint64(id)) _ = conn.Close() } } diff --git a/ap/ap_test.go b/ap/ap_test.go index f6af4683..a3990b24 100644 --- a/ap/ap_test.go +++ b/ap/ap_test.go @@ -61,7 +61,7 @@ func TestPongAckTickerStopsWhenConnNil(t *testing.T) { ap.pongAckTicker() }() - time.Sleep(pongAckInterval + time.Nanosecond) + time.Sleep(pongAckInterval + pongAckMargin + time.Nanosecond) synctest.Wait() ap.Close() @@ -223,3 +223,127 @@ func TestConnectGivesUpOnASilentAccesspoint(t *testing.T) { t.Fatal("the handshake with a silent accesspoint never gave up") } } + +// A request that went unanswered drops the connection it was sent on. By the +// time it gives up, that connection may already have been replaced, and the +// replacement must survive. +func TestDropConnClosesOnlyTheConnectionItNames(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + accesspoint := NewAccesspoint(&librespot.NullLogger{}, nil, "") + defer accesspoint.Close() + + first := attachFakeAP(accesspoint, time.Hour, time.Hour) + sentOn, err := accesspoint.SendConn(context.Background(), PacketTypeRequestKey, []byte("key request")) + if err != nil { + t.Fatalf("send: %v", err) + } + second := attachFakeAP(accesspoint, time.Hour, time.Hour) + synctest.Wait() + if !first.isClosed() { + t.Fatal("the replaced connection is still open") + } + + accesspoint.DropConn(sentOn) + synctest.Wait() + if second.isClosed() { + t.Fatal("dropping a replaced connection closed its replacement") + } + + current, err := accesspoint.SendConn(context.Background(), PacketTypeRequestKey, []byte("key request")) + if err != nil { + t.Fatalf("send: %v", err) + } + + accesspoint.DropConn(current) + synctest.Wait() + if !second.isClosed() { + t.Fatal("dropping the current connection left it open") + } + }) +} + +// A reconnection holds connMu through its dials and backoff, and is already +// replacing the connection: DropConn must not wait for it, since its caller is +// waiting for an answer, not recvLoop. +func TestDropConnDoesNotWaitForAReconnection(t *testing.T) { + accesspoint := NewAccesspoint(&librespot.NullLogger{}, nil, "") + defer accesspoint.Close() + + accesspoint.connMu.Lock() + defer accesspoint.connMu.Unlock() + + dropped := make(chan struct{}) + go func() { + accesspoint.DropConn(0) + close(dropped) + }() + + select { + case <-dropped: + case <-time.After(time.Second): + t.Fatal("DropConn waited for a reconnection in progress") + } +} + +// DropConn closes the socket and returns before recvLoop has noticed and +// started the reconnection. A packet sent in between, such as the key +// provider's retry sent right after the drop, must wait for that reconnection +// instead of being written to the closed socket and failing. Here no +// reconnection is possible (no APWelcome), so the accesspoint closes, and that +// is the only way the send may end. +func TestSendAfterDropConnWaitsForTheReplacement(t *testing.T) { + for i := 0; i < 300; i++ { + accesspoint := NewAccesspoint(&librespot.NullLogger{}, nil, "") + attachFakeAP(accesspoint, time.Hour, time.Hour) + accesspoint.Receive(PacketTypeAesKey) + + current, err := accesspoint.SendConn(context.Background(), PacketTypeRequestKey, []byte("key request")) + if err != nil { + t.Fatalf("send: %v", err) + } + accesspoint.DropConn(current) + + sent := make(chan error, 1) + go func() { + _, err := accesspoint.SendConn(context.Background(), PacketTypeRequestKey, []byte("key request")) + sent <- err + }() + if err := <-sent; !errors.Is(err, ErrAccesspointClosed) { + t.Fatalf("after %d drops, a send right after DropConn failed with %q instead of waiting for the replacement", i+1, err) + } + accesspoint.Close() + } +} + +// The accesspoint's pings are the only traffic on an idle connection, so a +// PongAck that does not come is the only sign that it went silent: until it is +// dropped, every audio key and mercury request is sent into the void. It must +// be dropped a bounded time after the last PongAck, whatever the phase of the +// check, and never while PongAcks keep coming. +func TestSilentConnectionIsDroppedSoonAfterItsLastPongAck(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + accesspoint := NewAccesspoint(&librespot.NullLogger{}, nil, "") + defer accesspoint.Close() + + f := attachFakeAP(accesspoint, time.Second, pongAckInterval) + start := time.Now() + accesspoint.Receive(PacketTypeAesKey) + + time.Sleep(10 * time.Minute) + synctest.Wait() + select { + case <-f.closed: + t.Fatalf("a connection whose pongs were acknowledged was dropped after %v", f.closedAfter(start)) + default: + } + + f.cut.Store(true) + lastAck := f.lastAckAt() + + <-f.closed + if silent := f.closedAfter(lastAck); silent > pongAckInterval+pongAckMargin { + t.Fatalf("a silent connection was dropped %v after its last PongAck, want at most %v", + silent, pongAckInterval+pongAckMargin) + } + }) +} diff --git a/ap/fakeap_test.go b/ap/fakeap_test.go new file mode 100644 index 00000000..5502e26e --- /dev/null +++ b/ap/fakeap_test.go @@ -0,0 +1,126 @@ +//go:build test_unit + +package ap + +import ( + "bytes" + "context" + "net" + "sync" + "sync/atomic" + "time" +) + +// cuttableConn drops everything read and written once cut, like a connection +// whose network path went away without a FIN or an RST. +type cuttableConn struct { + net.Conn + cut *atomic.Bool +} + +func (c *cuttableConn) Read(b []byte) (int, error) { + for { + n, err := c.Conn.Read(b) + if err != nil || !c.cut.Load() { + return n, err + } + } +} + +func (c *cuttableConn) Write(b []byte) (int, error) { + if c.cut.Load() { + return len(b), nil + } + return c.Conn.Write(b) +} + +// fakeAP is the accesspoint's end of an authenticated connection: it pings the +// device every pingEvery and acknowledges each pong, until the test cuts it. +type fakeAP struct { + enc *shannonConn + cut atomic.Bool + closed chan struct{} + + mu sync.Mutex + lastAck time.Time + closedAt time.Time + pending sync.WaitGroup +} + +// attachFakeAP gives the accesspoint a connection to a fakeAP, as a completed +// handshake would. The accesspoint has no APWelcome, so once that connection +// is closed it gives up instead of reconnecting. +func attachFakeAP(accesspoint *Accesspoint, firstPing, pingEvery time.Duration) *fakeAP { + device, server := net.Pipe() + sendKey, recvKey := bytes.Repeat([]byte{1}, 32), bytes.Repeat([]byte{2}, 32) + + f := &fakeAP{enc: newShannonConn(server, recvKey, sendKey), closed: make(chan struct{})} + accesspoint.setConn(&cuttableConn{Conn: device, cut: &f.cut}) + accesspoint.encConn = newShannonConn(accesspoint.conn, sendKey, recvKey) + + go f.read() + go f.ping(firstPing, pingEvery) + return f +} + +func (f *fakeAP) read() { + for { + pkt, _, err := f.enc.receivePacket(context.Background()) + if err != nil { + f.mu.Lock() + f.closedAt = time.Now() + f.mu.Unlock() + close(f.closed) + f.pending.Wait() + return + } + if pkt != PacketTypePong { + continue + } + // net.Pipe has no buffer: answer from another goroutine so this loop + // keeps reading while the device is not, as a socket buffer would. + f.pending.Add(1) + go func() { + defer f.pending.Done() + if f.enc.sendPacket(context.Background(), PacketTypePongAck, nil) == nil && !f.cut.Load() { + f.mu.Lock() + f.lastAck = time.Now() + f.mu.Unlock() + } + }() + } +} + +func (f *fakeAP) ping(first, every time.Duration) { + next := first + for { + select { + case <-f.closed: + return + case <-time.After(next): + _ = f.enc.sendPacket(context.Background(), PacketTypePing, []byte{0, 0, 0, 0}) + next = every + } + } +} + +func (f *fakeAP) lastAckAt() time.Time { + f.mu.Lock() + defer f.mu.Unlock() + return f.lastAck +} + +func (f *fakeAP) closedAfter(t time.Time) time.Duration { + f.mu.Lock() + defer f.mu.Unlock() + return f.closedAt.Sub(t) +} + +func (f *fakeAP) isClosed() bool { + select { + case <-f.closed: + return true + default: + return false + } +} diff --git a/audio/provider.go b/audio/provider.go index 24be05c4..e69a87c6 100644 --- a/audio/provider.go +++ b/audio/provider.go @@ -5,6 +5,7 @@ import ( "context" "encoding/binary" "encoding/hex" + "errors" "fmt" "io" "sync" @@ -22,8 +23,27 @@ func (e KeyProviderError) Error() string { return fmt.Sprintf("failed retrieving aes key with code %d", e.Code) } +// A live accesspoint answers a key request within a round trip, so a request +// left unanswered for keyRequestTimeout was most likely sent on a connection +// that went silent: that connection is dropped and the request sent again on +// the one that replaces it, which keyRetryTimeout leaves time to establish. +const ( + keyRequestTimeout = 5 * time.Second + keyRetryTimeout = 15 * time.Second +) + +var errKeyRequestTimeout = fmt.Errorf("aes key request timed out: %w", context.DeadlineExceeded) + +// accesspoint is what KeyProvider needs of *ap.Accesspoint. +type accesspoint interface { + Done() <-chan struct{} + Receive(types ...ap.PacketType) <-chan ap.Packet + SendConn(ctx context.Context, pktType ap.PacketType, payload []byte) (ap.ConnID, error) + DropConn(id ap.ConnID) +} + type KeyProvider struct { - ap *ap.Accesspoint + ap accesspoint log librespot.Logger recvLoopOnce sync.Once @@ -35,6 +55,8 @@ type keyRequest struct { gid []byte fileId []byte resp chan keyResponse + // sent receives the connection the request was written to. + sent chan ap.ConnID } type keyResponse struct { @@ -42,7 +64,7 @@ type keyResponse struct { err error } -func NewAudioKeyProvider(log librespot.Logger, ap *ap.Accesspoint) *KeyProvider { +func NewAudioKeyProvider(log librespot.Logger, ap accesspoint) *KeyProvider { p := &KeyProvider{log: log, ap: ap} p.reqChan = make(chan keyRequest) return p @@ -111,12 +133,14 @@ func (p *KeyProvider) recvLoop() { // so one caller's cancellation must not abort the others. Request // applies the caller's deadline to the response wait instead, and // Send already fails once the accesspoint is closed. - if err := p.ap.Send(context.Background(), ap.PacketTypeRequestKey, buf.Bytes()); err != nil { + conn, err := p.ap.SendConn(context.Background(), ap.PacketTypeRequestKey, buf.Bytes()) + if err != nil { delete(reqs, reqSeq) req.resp <- keyResponse{err: fmt.Errorf("failed sending key request for file %s, gid: %s: %w", hex.EncodeToString(req.fileId), librespot.GidToBase62(req.gid), err)} continue } + req.sent <- conn p.log.Debugf("requested aes key for file %s, gid: %s", hex.EncodeToString(req.fileId), librespot.GidToBase62(req.gid)) } @@ -124,23 +148,46 @@ func (p *KeyProvider) recvLoop() { } func (p *KeyProvider) Request(ctx context.Context, gid []byte, fileId []byte) ([]byte, error) { + key, err := p.request(ctx, gid, fileId, keyRequestTimeout) + if errors.Is(err, errKeyRequestTimeout) { + p.log.Warnf("no aes key for file %s after %.0fs, retrying on a new accesspoint connection", + hex.EncodeToString(fileId), keyRequestTimeout.Seconds()) + key, err = p.request(ctx, gid, fileId, keyRetryTimeout) + } + + return key, err +} + +// request sends one key request, which gets its own sequence number and so +// can only be answered by the answer to that request. +func (p *KeyProvider) request(ctx context.Context, gid []byte, fileId []byte, timeout time.Duration) ([]byte, error) { done := p.ap.Done() p.startReceiving() - req := keyRequest{gid: gid, fileId: fileId, resp: make(chan keyResponse, 1)} + req := keyRequest{gid: gid, fileId: fileId, resp: make(chan keyResponse, 1), sent: make(chan ap.ConnID, 1)} select { case <-done: return nil, ap.ErrAccesspointClosed + case <-ctx.Done(): + return nil, ctx.Err() case p.reqChan <- req: } - ctx, cancel := context.WithTimeout(ctx, 15*time.Second) - defer cancel() + timer := time.NewTimer(timeout) + defer timer.Stop() var resp keyResponse select { case resp = <-req.resp: + case <-timer.C: + select { + case conn := <-req.sent: + p.ap.DropConn(conn) + default: + // Still being written, which waits for a reconnection in progress. + } + return nil, errKeyRequestTimeout case <-ctx.Done(): return nil, ctx.Err() case <-done: diff --git a/audio/provider_test.go b/audio/provider_test.go index 0030c527..fb82ad27 100644 --- a/audio/provider_test.go +++ b/audio/provider_test.go @@ -3,8 +3,13 @@ package audio_test import ( + "bytes" "context" + "slices" + "sync" "testing" + "testing/synctest" + "time" librespot "github.com/devgianlu/go-librespot" "github.com/devgianlu/go-librespot/ap" @@ -21,3 +26,105 @@ func TestRequestFailsWhenAccesspointIsClosed(t *testing.T) { _, err := provider.Request(context.Background(), nil, nil) require.ErrorIs(t, err, ap.ErrAccesspointClosed) } + +// fakeAccesspoint answers key requests as Spotify's accesspoint does on a live +// connection, and not at all on one that went silent, until DropConn replaces +// that connection with a live one. +type fakeAccesspoint struct { + done chan struct{} + recv chan ap.Packet + + mu sync.Mutex + conn ap.ConnID + silent bool + sentOn []ap.ConnID + dropped []ap.ConnID +} + +func newFakeAccesspoint(silent bool) *fakeAccesspoint { + return &fakeAccesspoint{done: make(chan struct{}), recv: make(chan ap.Packet, 8), silent: silent} +} + +func (f *fakeAccesspoint) Done() <-chan struct{} { return f.done } + +func (f *fakeAccesspoint) Receive(...ap.PacketType) <-chan ap.Packet { return f.recv } + +func (f *fakeAccesspoint) SendConn(_ context.Context, pktType ap.PacketType, payload []byte) (ap.ConnID, error) { + f.mu.Lock() + defer f.mu.Unlock() + + f.sentOn = append(f.sentOn, f.conn) + if pktType == ap.PacketTypeRequestKey && !f.silent { + // file id, gid, sequence number, 0 + fileId, seq := payload[:16], payload[32:36] + f.recv <- ap.Packet{Type: ap.PacketTypeAesKey, Payload: append(slices.Clone(seq), keyFor(fileId)...)} + } + return f.conn, nil +} + +func (f *fakeAccesspoint) DropConn(id ap.ConnID) { + f.mu.Lock() + defer f.mu.Unlock() + + f.dropped = append(f.dropped, id) + if id == f.conn { + f.conn++ + f.silent = false + } +} + +func (f *fakeAccesspoint) close() { close(f.done) } + +func keyFor(fileId []byte) []byte { + key := slices.Clone(fileId) + slices.Reverse(key) + return key +} + +var ( + testGid = bytes.Repeat([]byte{0x11}, 16) + testFileId = []byte("0123456789abcdef") +) + +// A connection that went silent answers no key request: the track load waiting +// on it failed, and every later one too until the accesspoint noticed, minutes +// later. The request must drop that connection and get its key on the next. +func TestKeyRequestIsRetriedOnANewConnectionWhenTheFirstWentSilent(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + accesspoint := newFakeAccesspoint(true) + defer accesspoint.close() + provider := audio.NewAudioKeyProvider(&librespot.NullLogger{}, accesspoint) + + start := time.Now() + key, err := provider.Request(context.Background(), testGid, testFileId) + require.NoError(t, err) + require.Equal(t, keyFor(testFileId), key) + require.LessOrEqual(t, time.Since(start), 5*time.Second) + + accesspoint.mu.Lock() + defer accesspoint.mu.Unlock() + require.Equal(t, []ap.ConnID{0}, accesspoint.dropped, "the silent connection was not dropped") + require.Equal(t, []ap.ConnID{0, 1}, accesspoint.sentOn, "the request was not sent again on the new connection") + }) +} + +// A caller that gives up first, such as a load cancelled by a skip, says +// nothing about the connection: it must not be dropped, nor the request sent +// again past the caller's deadline. +func TestKeyRequestEndedByItsCallerKeepsTheConnection(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + accesspoint := newFakeAccesspoint(true) + defer accesspoint.close() + provider := audio.NewAudioKeyProvider(&librespot.NullLogger{}, accesspoint) + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + _, err := provider.Request(ctx, testGid, testFileId) + require.ErrorIs(t, err, context.DeadlineExceeded) + + accesspoint.mu.Lock() + defer accesspoint.mu.Unlock() + require.Empty(t, accesspoint.dropped) + require.Equal(t, []ap.ConnID{0}, accesspoint.sentOn) + }) +}