-
-
+
+
diff --git a/internal/health/checker.go b/internal/health/checker.go
index 618cd4c..4afbbb2 100644
--- a/internal/health/checker.go
+++ b/internal/health/checker.go
@@ -2,12 +2,14 @@ package health
import (
"context"
+ "crypto/rand"
"encoding/binary"
"fmt"
"io"
"log"
"net"
"sync"
+ "sync/atomic"
"time"
"github.com/ParsaKSH/SlipStream-Plus/internal/config"
@@ -34,6 +36,8 @@ type Checker struct {
mu sync.Mutex
failures map[int]int
+
+ probeSeq atomic.Uint32 // unique probe counter to avoid ConnID collisions
}
func NewChecker(mgr *engine.Manager, cfg *config.HealthCheckConfig) *Checker {
@@ -371,8 +375,14 @@ func (c *Checker) probeFramingProtocol(inst *engine.Instance) (time.Duration, er
synPayload = append(synPayload, []byte(domain)...) // domain
synPayload = append(synPayload, 0x00, 0x50) // port 80
- // Use a unique probe ConnID (high range to avoid collision)
- probeConnID := uint32(0xFFFF0000) + uint32(inst.ID())
+ // Use a unique probe ConnID combining high-range prefix, instance ID,
+ // monotonic counter, and random bits to avoid collisions with real connections
+ // and previous probes that haven't been cleaned up yet.
+ seq := c.probeSeq.Add(1)
+ var rndBuf [2]byte
+ rand.Read(rndBuf[:])
+ rnd := uint32(binary.BigEndian.Uint16(rndBuf[:]))
+ probeConnID := uint32(0xFE000000) | (uint32(inst.ID())&0xFF)<<16 | (seq&0xFF)<<8 | (rnd & 0xFF)
synFrame := &tunnel.Frame{
ConnID: probeConnID,
diff --git a/internal/proxy/socks5.go b/internal/proxy/socks5.go
index f715233..17f5931 100644
--- a/internal/proxy/socks5.go
+++ b/internal/proxy/socks5.go
@@ -34,7 +34,7 @@ type Server struct {
// Packet-split mode fields
packetSplit bool
tunnelPool *tunnel.TunnelPool
- connIDGen tunnel.ConnIDGenerator
+ connIDGen *tunnel.ConnIDGenerator
chunkSize int
}
@@ -53,6 +53,7 @@ func NewServer(listenAddr string, bufferSize int, maxConns int, mgr *engine.Mana
func (s *Server) EnablePacketSplit(pool *tunnel.TunnelPool, chunkSize int) {
s.packetSplit = true
s.tunnelPool = pool
+ s.connIDGen = tunnel.NewConnIDGenerator()
s.chunkSize = chunkSize
log.Printf("[proxy] packet-split mode enabled (chunk_size=%d)", chunkSize)
}
@@ -281,6 +282,13 @@ func (s *Server) handlePacketSplit(clientConn net.Conn, connID uint64, atyp byte
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
+ // When context is cancelled (either direction finished), close the client
+ // connection so that blocking Read/Write calls unblock immediately.
+ go func() {
+ <-ctx.Done()
+ clientConn.Close()
+ }()
+
var txN, rxN int64
var wg sync.WaitGroup
diff --git a/internal/tunnel/pool.go b/internal/tunnel/pool.go
index 5db735d..df0e070 100644
--- a/internal/tunnel/pool.go
+++ b/internal/tunnel/pool.go
@@ -230,8 +230,12 @@ func (p *TunnelPool) readLoop(tc *TunnelConn) {
ch := v.(chan *Frame)
select {
case ch <- frame:
- default:
- // Buffer full — drop silently
+ case <-time.After(5 * time.Second):
+ // Buffer full for too long — connection is stuck, log and drop
+ log.Printf("[tunnel-pool] instance %d: frame buffer full for conn=%d, dropping frame seq=%d",
+ tc.inst.ID(), frame.ConnID, frame.SeqNum)
+ case <-p.stopCh:
+ return
}
}
}
diff --git a/internal/tunnel/protocol.go b/internal/tunnel/protocol.go
index 922d9d5..4fb93b8 100644
--- a/internal/tunnel/protocol.go
+++ b/internal/tunnel/protocol.go
@@ -1,6 +1,7 @@
package tunnel
import (
+ "crypto/rand"
"encoding/binary"
"fmt"
"io"
@@ -104,10 +105,27 @@ func ReadFrame(r io.Reader) (*Frame, error) {
}
// ConnIDGenerator produces unique connection IDs.
+// Starts from a random offset to avoid collisions after process restarts
+// while CentralServer still holds state from the previous session.
type ConnIDGenerator struct {
counter atomic.Uint32
}
+// NewConnIDGenerator creates a generator with a random starting offset.
+func NewConnIDGenerator() *ConnIDGenerator {
+ g := &ConnIDGenerator{}
+ // Random start in [1, 0x7FFFFFFF) — avoids 0 and leaves room before wrap
+ var buf [4]byte
+ if _, err := rand.Read(buf[:]); err == nil {
+ start := binary.BigEndian.Uint32(buf[:]) & 0x7FFFFFFF
+ if start == 0 {
+ start = 1
+ }
+ g.counter.Store(start)
+ }
+ return g
+}
+
// Next returns the next unique connection ID.
func (g *ConnIDGenerator) Next() uint32 {
return g.counter.Add(1)
diff --git a/internal/tunnel/splitter.go b/internal/tunnel/splitter.go
index 456343a..9577b36 100644
--- a/internal/tunnel/splitter.go
+++ b/internal/tunnel/splitter.go
@@ -233,25 +233,28 @@ func (ps *PacketSplitter) pickInstanceExcluding(excludeID int) *engine.Instance
}
// Reorderer buffers out-of-order frames and delivers them in sequence order.
+// If a frame is missing for longer than gapTimeout, it is skipped to prevent
+// permanent stalls from lost frames.
type Reorderer struct {
- nextSeq uint32
- buffer map[uint32][]byte
- timeout time.Duration
+ nextSeq uint32
+ buffer map[uint32][]byte
+ gapTimeout time.Duration
+ waitingSince time.Time // when we first started waiting for nextSeq
}
func NewReorderer() *Reorderer {
return &Reorderer{
- nextSeq: 0,
- buffer: make(map[uint32][]byte),
- timeout: 10 * time.Second,
+ nextSeq: 0,
+ buffer: make(map[uint32][]byte),
+ gapTimeout: 2 * time.Second,
}
}
func NewReordererAt(startSeq uint32) *Reorderer {
return &Reorderer{
- nextSeq: startSeq,
- buffer: make(map[uint32][]byte),
- timeout: 10 * time.Second,
+ nextSeq: startSeq,
+ buffer: make(map[uint32][]byte),
+ gapTimeout: 2 * time.Second,
}
}
@@ -263,13 +266,59 @@ func (r *Reorderer) Insert(seq uint32, data []byte) {
}
func (r *Reorderer) Next() []byte {
+ // Fast path: next seq is available
data, ok := r.buffer[r.nextSeq]
- if !ok {
+ if ok {
+ delete(r.buffer, r.nextSeq)
+ r.nextSeq++
+ r.waitingSince = time.Time{} // reset wait timer
+ return data
+ }
+
+ // Nothing buffered at all — nothing to skip to
+ if len(r.buffer) == 0 {
+ r.waitingSince = time.Time{}
return nil
}
- delete(r.buffer, r.nextSeq)
- r.nextSeq++
- return data
+
+ // There are buffered frames but nextSeq is missing.
+ // Start or check the gap timer.
+ now := time.Now()
+ if r.waitingSince.IsZero() {
+ r.waitingSince = now
+ return nil
+ }
+
+ if now.Sub(r.waitingSince) < r.gapTimeout {
+ return nil // still within grace period
+ }
+
+ // Gap timeout expired — skip to the lowest available seq
+ r.skipToLowest()
+ r.waitingSince = time.Time{}
+
+ data, ok = r.buffer[r.nextSeq]
+ if ok {
+ delete(r.buffer, r.nextSeq)
+ r.nextSeq++
+ return data
+ }
+ return nil
+}
+
+// skipToLowest advances nextSeq to the lowest seq number in the buffer.
+func (r *Reorderer) skipToLowest() {
+ minSeq := r.nextSeq
+ found := false
+ for seq := range r.buffer {
+ if !found || seq < minSeq {
+ minSeq = seq
+ found = true
+ }
+ }
+ if found && minSeq > r.nextSeq {
+ r.nextSeq = minSeq
+ }
}
func (r *Reorderer) Pending() int {