From c6c49afb833a0b94b40699cc9e175d161dd6f029 Mon Sep 17 00:00:00 2001 From: Leo Durand Date: Sat, 7 Nov 2026 12:34:50 +0100 Subject: [PATCH 1/2] fix: drop an accesspoint connection that leaves an audio key unanswered Measured on a device after a network change (its wired link disconnected, wifi still up): three track loads in a row failed with "failed retrieving audio key: context deadline exceeded", and nothing played for 130 s. The key requests went out on the accesspoint connection bound to the address that had gone, which nothing answered and nothing closed until the pong ack check fired, 173 s after the last PongAck. A new connection then authenticated within 70 ms and the next load succeeded. Request waited 15 s for an answer and failed. Nothing closed the connection on that timeout and the player does not retry a load, so every load failed the same way until the pong ack check noticed. A key request left unanswered for 5 s now drops the connection it was written to, so that recvLoop's reconnection replaces it, and is sent once more with 15 s to be answered. That covers the reconnection, since writing the retry waits for connMu while a reconnection holds it. The first wait is shorter because a retry follows: a live accesspoint answers in a round trip (librespot gives up on a key after 1.5 s), and a false alarm costs one reconnection. - The connection is named by the ConnID that SendConn returns, so a request that gives up after its connection was replaced does not close the replacement. DropConn only closes the socket, which has no handshake, and takes connMu with TryRLock: only a connection in progress (Connect, or recvLoop's reconnection) holds it for writing, and it is already replacing the connection. So the caller never waits out a backoff, and no lock order is set against recvLoop. - A packet written to a connection this side closed (by DropConn or closeConn) is sent again once that connection is replaced. recvLoop only takes connMu for the reconnection after its read has failed, so the retry, sent right after the drop, would otherwise often reach the closed socket, fail as a send error, and fail the load the old way. recvLoop's own Pong is written directly, since recvLoop is what does the replacing. - Each attempt is a new request with its own sequence number and response channel, so a late answer to the first attempt reaches only that attempt's channel, which nothing reads any more. - A request ended by its caller, such as a load cancelled by a skip, says nothing about the connection: it drops nothing and is not retried. Handing the request to the pump also stops when the caller's context ends, because the pump's write waits for any reconnection in progress. - NewAudioKeyProvider takes the four methods it uses as an interface, which *ap.Accesspoint satisfies, so the retry can be tested against a fake accesspoint. Tests: a fake accesspoint for the key provider, whose connection is either live or silent until DropConn replaces it, and a fake accesspoint end over net.Pipe for DropConn. The tests do not compile against the parent, which has no SendConn or DropConn. Each was then seen failing on this commit with the matching part undone: - the parent's single 15 s attempt, with no drop: TestKeyRequestIsRetriedOnANewConnectionWhenTheFirstWentSilent fails with "aes key request timed out: context deadline exceeded"; - dropping and giving up on the caller's cancellation: TestKeyRequestEndedByItsCallerKeepsTheConnection fails with "Should be empty, but was [0 1]"; - RLock in place of TryRLock: TestDropConnDoesNotWaitForAReconnection fails; - no ConnID check: TestDropConnClosesOnlyTheConnectionItNames fails; - the parent's SendConn: TestSendAfterDropConnWaitsForTheReplacement, which drops the connection of a running recvLoop and sends at once, 300 times, fails on the first drop with "failed writing packet: io: read/write on closed pipe". It passes 20 runs in a row with the fix, and 10 with GOMAXPROCS=1. Floor: gofmt -l . (empty), go vet -tags "test_unit test_integration" ./..., go test -count=1 -tags "test_unit test_integration" ./..., all passing. Not run under -race: the race detector does not run on the build host (TSan rejects its 47-bit arm64 address space). On a device, with this and the next commit backported to v0.10.5, after the same network change: the first key request after the cut logged "no aes key for file ... after 5s, retrying on a new accesspoint connection", a new connection authenticated within 70 ms, and the track loaded 5.8 s after the skip that asked for it. Co-Authored-By: Claude Opus 5.5 --- ap/ap.go | 109 +++++++++++++++++++++++++++++++++-------- ap/ap_test.go | 91 ++++++++++++++++++++++++++++++++++ ap/fakeap_test.go | 104 +++++++++++++++++++++++++++++++++++++++ audio/provider.go | 59 +++++++++++++++++++--- audio/provider_test.go | 107 ++++++++++++++++++++++++++++++++++++++++ 5 files changed, 443 insertions(+), 27 deletions(-) create mode 100644 ap/fakeap_test.go diff --git a/ap/ap.go b/ap/ap.go index ee348cc3..e93bbd59 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" @@ -43,6 +44,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 @@ -67,9 +72,15 @@ type Accesspoint struct { lastPongAck time.Time 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 +121,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 +134,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 +289,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 +419,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 } @@ -486,10 +552,11 @@ func (ap *Accesspoint) timeSinceLastPongAck() time.Duration { 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..e044e79d 100644 --- a/ap/ap_test.go +++ b/ap/ap_test.go @@ -223,3 +223,94 @@ 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() + } +} diff --git a/ap/fakeap_test.go b/ap/fakeap_test.go new file mode 100644 index 00000000..dcd8f367 --- /dev/null +++ b/ap/fakeap_test.go @@ -0,0 +1,104 @@ +//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{} + + 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 { + 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() + _ = f.enc.sendPacket(context.Background(), PacketTypePongAck, nil) + }() + } +} + +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) 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) + }) +} From 12cba571658180e182f1c5609b260bb299d2a32e Mon Sep 17 00:00:00 2001 From: Leo Durand Date: Sat, 7 Nov 2026 12:35:57 +0100 Subject: [PATCH 2/2] fix: drop a silent accesspoint connection 140 s after its last PongAck The pong ack check ran on a 120 s ticker and closed the connection once the last PongAck was more than 120 s old, and a new connection set that PongAck 120 s in the future. So a connection that went silent was dropped at the second tick after its last PongAck, 120 s to 240 s later, or 240 s to 360 s after a reconnection that had not carried one yet. Measured on a device after a network change: the check fired 173 s after the last PongAck it had recorded, and the three track loads tried before then failed on audio keys that never came. In this protocol the accesspoint pings and the device answers, so the device cannot probe a connection itself: it can only bound the wait for the next PongAck. The check expected one at least every 120 s and allowed nothing past that, so a PongAck that came even a moment late dropped a working connection whenever a tick fell in between. librespot (core/src/session.rs) adds a 20 s safety margin to each of its keep-alive deadlines. The connection is now dropped 120 s plus a 20 s margin after its last PongAck, a new connection counting as one. A timer armed for that deadline replaces the ticker, so the phase of a tick no longer adds up to another 120 s. The check drops the connection its deadline belongs to through DropConn, so it can no longer close a replacement. It used to do that: closeConn waited for a reconnection in progress, then closed the connection that reconnection had just made. Tests: TestSilentConnectionIsDroppedSoonAfterItsLastPongAck uses a fake accesspoint end that pings every 120 s and acknowledges each Pong for 10 minutes, during which the connection must stay open. The fake then goes silent, and the connection must be closed at most 140 s after its last PongAck. With the parent's ticker (and pongAckMargin declared so the test compiles), it fails with "a silent connection was dropped 3m59s after its last PongAck, want at most 2m20s". With the margin set to -30 s, its live half fails ("a connection whose pongs were acknowledged was dropped after 1m31s"), so it also guards a bound tighter than the ping cadence. TestPongAckTickerStopsWhenConnNil now sleeps past the new deadline, so it still exercises the check firing with no connection. Floor: gofmt -l . (empty), go vet -tags "test_unit test_integration" ./..., go test -count=1 -tags "test_unit test_integration" ./..., all passing. Not run under -race: the race detector does not run on the build host. Co-Authored-By: Claude Opus 5.5 --- ap/ap.go | 48 ++++++++++++++++++++++++++++++----------------- ap/ap_test.go | 35 +++++++++++++++++++++++++++++++++- ap/fakeap_test.go | 26 +++++++++++++++++++++++-- 3 files changed, 89 insertions(+), 20 deletions(-) diff --git a/ap/ap.go b/ap/ap.go index e93bbd59..c2fc8507 100644 --- a/ap/ap.go +++ b/ap/ap.go @@ -27,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 @@ -70,6 +78,7 @@ type Accesspoint struct { recvChans map[PacketType][]chan Packet recvChansLock sync.RWMutex lastPongAck time.Time + pongAckConn ConnID lastPongAckLock sync.Mutex // connMu protects conn, encConn, connID, replaced, and welcome pointer state. @@ -487,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) { @@ -532,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() } @@ -544,10 +558,10 @@ 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() { diff --git a/ap/ap_test.go b/ap/ap_test.go index e044e79d..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() @@ -314,3 +314,36 @@ func TestSendAfterDropConnWaitsForTheReplacement(t *testing.T) { 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 index dcd8f367..5502e26e 100644 --- a/ap/fakeap_test.go +++ b/ap/fakeap_test.go @@ -41,7 +41,10 @@ type fakeAP struct { cut atomic.Bool closed chan struct{} - pending sync.WaitGroup + mu sync.Mutex + lastAck time.Time + closedAt time.Time + pending sync.WaitGroup } // attachFakeAP gives the accesspoint a connection to a fakeAP, as a completed @@ -64,6 +67,9 @@ 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 @@ -76,7 +82,11 @@ func (f *fakeAP) read() { f.pending.Add(1) go func() { defer f.pending.Done() - _ = f.enc.sendPacket(context.Background(), PacketTypePongAck, nil) + if f.enc.sendPacket(context.Background(), PacketTypePongAck, nil) == nil && !f.cut.Load() { + f.mu.Lock() + f.lastAck = time.Now() + f.mu.Unlock() + } }() } } @@ -94,6 +104,18 @@ func (f *fakeAP) ping(first, every time.Duration) { } } +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: