diff --git a/connection.go b/connection.go index d7ffd57..da63844 100644 --- a/connection.go +++ b/connection.go @@ -201,7 +201,19 @@ func (c *connection) triggerTimer(now time.Time) (err error) { return errors.New("network inactivity") } - if pk, t := c.retransmission.shift(now, c.rtt.RTO()); len(pk) > 0 { + if pk, t, gaveUp := c.retransmission.shift(now, c.rtt.RTO()); len(pk) > 0 { + if gaveUp { + // A reliable packet exhausted its retransmission attempts. The + // peer has a permanent gap in a stream — every subsequent frame + // on it would park in the reorder queue forever. Fail loudly so + // the application (spectrum) can fall back / redial instead of + // serving frozen sessions. This also releases the packet's bytes + // from the congestion window, which previously leaked and could + // mute the connection permanently. + c.sender.OnAck(now, t, c.rtt, uint64(len(pk))-protocol.PacketHeaderSize) + _ = c.CloseWithError(frame.ConnectionCloseTimeout, "retransmission exhausted") + return errors.New("retransmission exhausted") + } c.sender.OnCongestionEvent(now, t) if _, err := c.writeDatagram(pk); err != nil { return err @@ -212,11 +224,25 @@ func (c *connection) triggerTimer(now time.Time) (err error) { func (c *connection) receive(now, t time.Time, sequenceID uint32, frames []frame.Frame) (err error) { if sequenceID != 0 { - c.ack.add(t, sequenceID) - if !c.receiveQueue.add(sequenceID) { + switch c.receiveQueue.add(sequenceID) { + case receiveAccepted: + c.ack.add(t, sequenceID) + case receiveDuplicate: + // Re-ack: our previous acknowledgement may have been lost and the + // peer keeps retransmitting until it sees one. + c.ack.add(t, sequenceID) c.logger.Log("duplicate_receive", "sequenceID", sequenceID) releaseFrames(frames) return + case receiveRejected: + // Outside the receive window / queue full. Deliberately NOT + // acked: the previous behavior (ack before add) made the sender + // forget the packet while we dropped it — silent permanent data + // loss and a stuck stream. Unacked, the peer retransmits it once + // the window advances. + c.logger.Log("rejected_receive", "sequenceID", sequenceID) + releaseFrames(frames) + return } } diff --git a/receive_queue.go b/receive_queue.go index 09245e7..164bbfd 100644 --- a/receive_queue.go +++ b/receive_queue.go @@ -2,6 +2,21 @@ package spectral const maxReceiveQueueEntries = 8192 +type receiveResult int + +const ( + // receiveAccepted: fresh packet, deliver and acknowledge. + receiveAccepted receiveResult = iota + // receiveDuplicate: already delivered. Re-acknowledge (our previous ack + // may have been lost) but do not deliver again. + receiveDuplicate + // receiveRejected: fresh packet outside the receive window (or queue + // full). Must NOT be acknowledged: acking makes the sender drop it from + // its retransmission queue and the data is permanently lost. Dropping it + // unacked lets the sender retransmit once the window advances. + receiveRejected +) + type receiveQueue struct { expected uint32 queue map[uint32]bool @@ -14,24 +29,24 @@ func newReceiveQueue() *receiveQueue { } } -func (r *receiveQueue) add(sequenceID uint32) bool { +func (r *receiveQueue) add(sequenceID uint32) receiveResult { if r.exists(sequenceID) { - return false + return receiveDuplicate } if sequenceID > r.expected+maxReceiveQueueEntries { - return false + return receiveRejected } // Always admit the next expected sequence ID, even when the queue is full, // so merge() can progress and free queued entries. if len(r.queue) >= maxReceiveQueueEntries && sequenceID != r.expected { - return false + return receiveRejected } r.queue[sequenceID] = true r.merge() - return true + return receiveAccepted } func (r *receiveQueue) exists(sequenceID uint32) bool { diff --git a/receive_queue_test.go b/receive_queue_test.go index 5bcd59d..e9e8009 100644 --- a/receive_queue_test.go +++ b/receive_queue_test.go @@ -6,7 +6,7 @@ func TestReceiveQueueAllowsExpectedWhenFull(t *testing.T) { q := newReceiveQueue() for sequenceID := uint32(2); sequenceID <= maxReceiveQueueEntries+1; sequenceID++ { - if !q.add(sequenceID) { + if q.add(sequenceID) != receiveAccepted { t.Fatalf("expected sequence %d to be queued", sequenceID) } } @@ -15,7 +15,7 @@ func TestReceiveQueueAllowsExpectedWhenFull(t *testing.T) { t.Fatalf("expected queue length %d, got %d", maxReceiveQueueEntries, len(q.queue)) } - if !q.add(1) { + if q.add(1) != receiveAccepted { t.Fatal("expected missing sequence to be admitted when queue is full") } diff --git a/retransmission_queue.go b/retransmission_queue.go index 04f7eac..993229e 100644 --- a/retransmission_queue.go +++ b/retransmission_queue.go @@ -61,7 +61,13 @@ func (r *retransmissionQueue) next(rto time.Duration) (t time.Time) { return } -func (r *retransmissionQueue) shift(now time.Time, rto time.Duration) (p []byte, t time.Time) { +// shift returns the next packet due for retransmission. gaveUp is true when a +// reliable packet has exhausted retransmissionAttempts: the connection can no +// longer guarantee ordered delivery (the peer's streams have a permanent gap) +// and the caller must tear the connection down loudly instead of silently +// dropping the packet — the previous behavior left streams stalled forever +// while the connection stayed "healthy". +func (r *retransmissionQueue) shift(now time.Time, rto time.Duration) (p []byte, t time.Time, gaveUp bool) { r.mu.Lock() defer r.mu.Unlock() if len(r.queue) == 0 { @@ -88,11 +94,11 @@ func (r *retransmissionQueue) shift(now time.Time, rto time.Duration) (p []byte, if entry.attempts >= retransmissionAttempts { r.queue[0] = nil r.queue = r.queue[1:] - } else { - r.queue = append(r.queue[1:], entry) - r.sort() + return entry.payload, sent, true } - return entry.payload, sent + r.queue = append(r.queue[1:], entry) + r.sort() + return entry.payload, sent, false } return } diff --git a/stream.go b/stream.go index 3dd99f8..b3c3680 100644 --- a/stream.go +++ b/stream.go @@ -189,7 +189,13 @@ func (s *Stream) read(p []byte) (n int) { s.mu.Lock() defer s.mu.Unlock() if s.buffer.Len() > 0 { - return s.buffer.Read(p) + n = s.buffer.Read(p) + // Draining the buffer may have freed room for frames parked in the + // reorder queue. processFrames is otherwise only invoked from + // receive(), i.e. when a NEW frame arrives — with an idle sender + // (request/response protocols) parked in-order frames would never be + // delivered and Read would block forever on s.available. + s.processFrames() } return }