Fix silent stream stalls and permanent data loss

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
 		}
 	}
 

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 {

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")
 	}
 

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
 }

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
 }