released first stable version of Spectral

- replaced nack-based retransmission with RTO-based retransmission

- improved congestion control with the NewReno algorithm, now the default

- added MTU discovery and MTU-based fragmentation

- implemented a dynamic event loop that processes everything connection-related, removing the need for much of the previous synchronization

- introduced delay-based ACKs to optimize network responsiveness

- introduced Spectral Log (SLOG) for logging network conditions and events

.github/workflows/staticcheck.yml

@@ -20,7 +20,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v5
         with:
-          go-version: '1.23.0'
+          go-version: '1.23.3'
 
       - name: Install Staticcheck
         run: go install honnef.co/go/tools/cmd/staticcheck@latest

ack_queue.go

@@ -1,53 +1,50 @@
 package spectral
 
 import (
-	"slices"
-	"sync"
 	"time"
+
+	"github.com/cooldogedev/spectral/internal/protocol"
 )
 
 type ackQueue struct {
-	sequenceID uint32
-	sort       bool
-	lastAck    time.Time
-	list       []uint32
-	mu         sync.Mutex
+	queue   []uint32
+	max     uint32
+	maxTime time.Time
+	nextAck time.Time
 }
 
 func newAckQueue() *ackQueue {
 	return &ackQueue{}
 }
 
-func (a *ackQueue) add(sequenceID uint32) {
-	a.mu.Lock()
-	a.sequenceID++
-	if !a.sort && sequenceID != a.sequenceID {
-		a.sort = true
+func (a *ackQueue) add(now time.Time, sequenceID uint32) {
+	a.queue = append(a.queue, sequenceID)
+	if sequenceID > a.max {
+		a.max = sequenceID
+		a.maxTime = now
+	}
+
+	if a.nextAck.IsZero() {
+		a.nextAck = now.Add(protocol.MaxAckDelay - protocol.TimerGranularity)
 	}
-	a.lastAck = time.Now()
-	a.list = append(a.list, sequenceID)
-	a.mu.Unlock()
 }
 
-func (a *ackQueue) addDuplicate(sequenceID uint32) {
-	a.mu.Lock()
-	a.sort = true
-	a.list = append(a.list, sequenceID)
-	a.mu.Unlock()
+func (a *ackQueue) next() (t time.Time) {
+	return a.nextAck
 }
 
-func (a *ackQueue) flush() (delay int64, list []uint32) {
-	a.mu.Lock()
-	if len(a.list) > 0 {
-		delay = time.Since(a.lastAck).Nanoseconds()
-		list = a.list
-		if a.sort {
-			slices.Sort(list)
+func (a *ackQueue) flush(now time.Time) (list []uint32, maxSequenceID uint32, delay time.Duration) {
+	if len(a.queue) > 0 && now.After(a.nextAck) {
+		list = a.queue
+		maxSequenceID = a.max
+		delay = now.Sub(a.maxTime)
+		if delay < 0 {
+			delay = 0
 		}
-		a.sort = false
-		a.lastAck = time.Time{}
-		a.list = a.list[:0]
+		a.queue = a.queue[:0]
+		a.max = 0
+		a.maxTime = time.Time{}
+		a.nextAck = time.Time{}
 	}
-	a.mu.Unlock()
 	return
 }

client.go

@@ -3,11 +3,11 @@ package spectral
 import (
 	"context"
 	"errors"
-	"fmt"
+	"net"
 	"sync"
 
-	"github.com/cooldogedev/spectral/internal"
 	"github.com/cooldogedev/spectral/internal/frame"
+	"github.com/cooldogedev/spectral/internal/log"
 	"github.com/cooldogedev/spectral/internal/protocol"
 )
 
@@ -16,38 +16,53 @@ type ClientConnection struct {
 	response        chan *frame.ConnectionResponse
 	streamResponses map[protocol.StreamID]chan *frame.StreamResponse
 	streamID        protocol.StreamID
-	mu              sync.Mutex
+	mu              sync.RWMutex
 }
 
-func newClientConnection(conn *internal.Conn, ctx context.Context) *ClientConnection {
+func newClientConnection(conn *udpConn, peerAddr *net.UDPAddr, ctx context.Context) *ClientConnection {
 	c := &ClientConnection{
-		connection:      newConnection(conn, -1, ctx),
-		response:        make(chan *frame.ConnectionResponse),
+		connection:      newConnection(conn, peerAddr, -1, ctx, log.PerspectiveClient),
+		response:        make(chan *frame.ConnectionResponse, 1),
 		streamResponses: make(map[protocol.StreamID]chan *frame.StreamResponse),
 	}
 	c.connection.handler = c.handle
 	return c
 }
 
 func (c *ClientConnection) OpenStream(ctx context.Context) (*Stream, error) {
+	ch := make(chan *frame.StreamResponse, 1)
 	c.mu.Lock()
 	streamID := c.streamID
 	c.streamID++
-	ch := make(chan *frame.StreamResponse, 1)
 	c.streamResponses[streamID] = ch
 	c.mu.Unlock()
-	if err := c.write(&frame.StreamRequest{StreamID: streamID}); err != nil {
+	defer func() {
+		close(ch)
+		c.mu.Lock()
+		delete(c.streamResponses, streamID)
+		c.mu.Unlock()
+	}()
+
+	c.logger.Log("stream_open_request", "streamID", streamID)
+	if err := c.writeControl(&frame.StreamRequest{StreamID: streamID}, true); err != nil {
 		return nil, err
 	}
 
 	select {
 	case <-ctx.Done():
-		return nil, ctx.Err()
+		return nil, context.Cause(ctx)
 	case response := <-ch:
 		if response.Response == frame.StreamResponseFailed {
+			c.logger.Log("stream_open_fail", "streamID", streamID)
 			return nil, errors.New("failed to open stream")
 		}
-		return c.createStream(streamID)
+
+		stream, err := c.createStream(streamID)
+		if err != nil {
+			return nil, err
+		}
+		c.logger.Log("stream_open_success", "streamID", streamID)
+		return stream, nil
 	}
 }
 
@@ -56,14 +71,17 @@ func (c *ClientConnection) handle(fr frame.Frame) (err error) {
 	case *frame.ConnectionResponse:
 		c.response <- fr
 	case *frame.StreamResponse:
-		c.mu.Lock()
+		c.mu.RLock()
 		ch, ok := c.streamResponses[fr.StreamID]
-		delete(c.streamResponses, fr.StreamID)
-		c.mu.Unlock()
-		if !ok {
-			return fmt.Errorf("received an unknown stream response for %v", fr.StreamID)
+		c.mu.RUnlock()
+		if ok {
+			select {
+			case ch <- fr:
+			default:
+			}
+		} else {
+			c.logger.Log("stream_response_unknown", "streamID", fr.StreamID)
 		}
-		ch <- fr
 	}
 	return
 }