v3

2026-08-16 02:50:24 -03:00
parent 6dac260155
commit 5621de243a
8 changed files with 628 additions and 353 deletions
@@ -238,20 +238,25 @@ type vpnClient struct {
 	ipv4, ipv6                                 netip.Addr
 	mtu                                        int
 	timeout                                    time.Duration
+	batchDelay                                 time.Duration
 	reconnectEvery                             int
 	upSizer, downSizer                         *adaptiveSizer
 	control, upload, download                  *txnLane
 	upPackets, downPackets, upBytes, downBytes atomic.Uint64
+	upBatches, downBatches, localDropped       atomic.Uint64
 	stopped                                    chan struct{}
 	stopOnce                                   sync.Once
 }

-func newVPNClient(tun *os.File, addr, token string, v4, v6 netip.Addr, mtu, start, min, max, growAfter, reconnectEvery int, timeout time.Duration, adaptLog bool) (*vpnClient, error) {
+func newVPNClient(tun *os.File, addr, token string, v4, v6 netip.Addr, mtu, start, min, max, growAfter, reconnectEvery int, timeout, batchDelay time.Duration, adaptLog bool) (*vpnClient, error) {
 	sid, err := randomSID()
 	if err != nil {
 		return nil, err
 	}
-	return &vpnClient{tun: tun, sid: sid, serverAddr: addr, token: token, ipv4: v4, ipv6: v6, mtu: mtu, timeout: timeout, reconnectEvery: reconnectEvery,
+	if batchDelay < 0 {
+		batchDelay = 0
+	}
+	return &vpnClient{tun: tun, sid: sid, serverAddr: addr, token: token, ipv4: v4, ipv6: v6, mtu: mtu, timeout: timeout, batchDelay: batchDelay, reconnectEvery: reconnectEvery,
 		upSizer: newSizer("upload", start, min, max, growAfter, adaptLog), downSizer: newSizer("download", start, min, max, growAfter, adaptLog),
 		control: newTxnLane(addr, timeout, reconnectEvery), upload: newTxnLane(addr, timeout, reconnectEvery), download: newTxnLane(addr, timeout, reconnectEvery), stopped: make(chan struct{})}, nil
 }
@@ -296,30 +301,113 @@ func (v *vpnClient) close() {
 	})
 }

-func (v *vpnClient) uploadLoop(errs chan<- error) {
-	buf := make([]byte, 65535)
-	var seq uint32
+func (v *vpnClient) logLocalDrop(reason string) {
+	n := v.localDropped.Add(1)
+	// Link-local/control traffic can be noisy. Keep it visible without filling
+	// the Android live log or making a harmless packet fatal to the VPN.
+	if n <= 8 || n%256 == 0 {
+		fmt.Printf("VPN DROP local packet (%s) dropped=%d\n", reason, n)
+	}
+}
+
+func (v *vpnClient) tunReadLoop(out chan<- []byte, errs chan<- error) {
+	buf := make([]byte, protocol.VPNMaxPacket)
 	for {
 		n, err := v.tun.Read(buf)
 		if err != nil {
 			errs <- err
 			return
 		}
-		if n < 1 {
+		if n < 1 || n > protocol.VPNMaxPacket {
 			continue
 		}
 		packet := append([]byte(nil), buf[:n]...)
-		if n > 65535 {
+		src, _, err := protocol.PacketAddresses(packet)
+		if err != nil {
+			v.logLocalDrop(err.Error())
 			continue
 		}
+		if src != v.ipv4 && src != v.ipv6 {
+			v.logLocalDrop(fmt.Sprintf("source %s is not assigned VPN address", src))
+			continue
+		}
+		select {
+		case out <- packet:
+		case <-v.stopped:
+			return
+		}
+	}
+}
+
+func batchWireSize(packets [][]byte) int {
+	n := 1
+	for _, p := range packets {
+		n += 2 + len(p)
+	}
+	return n
+}
+
+func (v *vpnClient) uploadLoop(in <-chan []byte, errs chan<- error) {
+	var seq uint32
+	var carry []byte
+	for {
+		var first []byte
+		if carry != nil {
+			first, carry = carry, nil
+		} else {
+			select {
+			case first = <-in:
+			case <-v.stopped:
+				return
+			}
+		}
+
+		packets := [][]byte{first}
+		encodedSize := 1 + 2 + len(first)
+		timer := time.NewTimer(v.batchDelay)
+	collect:
+		for encodedSize < protocol.VPNMaxBatch {
+			select {
+			case p := <-in:
+				need := 2 + len(p)
+				if encodedSize+need > protocol.VPNMaxBatch {
+					carry = p
+					break collect
+				}
+				packets = append(packets, p)
+				encodedSize += need
+			case <-timer.C:
+				break collect
+			case <-v.stopped:
+				if !timer.Stop() {
+					select {
+					case <-timer.C:
+					default:
+					}
+				}
+				return
+			}
+		}
+		if !timer.Stop() {
+			select {
+			case <-timer.C:
+			default:
+			}
+		}
+
+		batch, err := protocol.BuildVPNBatch(packets)
+		if err != nil {
+			errs <- err
+			return
+		}
 		offset := 0
-		for offset < n {
+		for offset < len(batch) {
 			limit := v.upSizer.Current()
-			size := n - offset
+			size := len(batch) - offset
 			if size > limit {
 				size = limit
 			}
-			req, e := protocol.BuildVPNPush(v.sid, seq, offset, n, packet[offset:offset+size])
+			req, e := protocol.BuildVPNPush(v.sid, seq, offset, len(batch), batch[offset:offset+size])
 			if e != nil {
 				errs <- e
 				return
@@ -334,16 +422,20 @@ func (v *vpnClient) uploadLoop(errs chan<- error) {
 				errs <- e
 				return
 			}
-			if rseq != seq || accepted < offset || accepted > n {
+			if rseq != seq || accepted < offset || accepted > len(batch) {
 				errs <- errors.New("bad server upload ACK")
 				return
 			}
-			fullRecord := size == limit
-			v.upSizer.Success(size, fullRecord)
+			v.upSizer.Success(size, size == limit)
 			offset = accepted
 		}
-		v.upPackets.Add(1)
-		v.upBytes.Add(uint64(n))
+		var rawBytes uint64
+		for _, p := range packets {
+			rawBytes += uint64(len(p))
+		}
+		v.upPackets.Add(uint64(len(packets)))
+		v.upBytes.Add(rawBytes)
+		v.upBatches.Add(1)
 		seq++
 	}
 }
@@ -352,7 +444,7 @@ func (v *vpnClient) downloadLoop(errs chan<- error) {
 	var want uint32
 	ack := protocol.VPNNoAck
 	offset := 0
-	var packet []byte
+	var transfer []byte
 	total := 0
 	for {
 		limit := v.downSizer.Current()
@@ -374,42 +466,58 @@ func (v *vpnClient) downloadLoop(errs chan<- error) {
 		if wait {
 			continue
 		}
-		if seq != want || roff != offset || rtotal < 1 || rtotal > 65535 {
+		if seq != want || roff != offset || rtotal < 1 || rtotal > protocol.VPNMaxBatch {
 			errs <- errors.New("bad server download sequence")
 			return
 		}
 		if offset == 0 {
 			total = rtotal
-			packet = make([]byte, 0, total)
+			transfer = make([]byte, 0, total)
 		} else if rtotal != total {
-			errs <- errors.New("download packet size changed")
+			errs <- errors.New("download transfer size changed")
 			return
 		}
-		packet = append(packet, data...)
+		transfer = append(transfer, data...)
 		offset += len(data)
 		v.downSizer.Success(len(data), len(data) == limit)
 		if offset < total {
 			continue
 		}
 		if offset != total {
-			errs <- errors.New("download packet overflow")
+			errs <- errors.New("download transfer overflow")
 			return
 		}
-		n, e := v.tun.Write(packet)
+		packets, e := protocol.ParseVPNBatch(transfer)
 		if e != nil {
-			errs <- e
-			return
+			// Compatibility with the first packet-VPN build, which used one raw
+			// IP packet as each transfer object.
+			if len(transfer) > 0 && (transfer[0]>>4 == 4 || transfer[0]>>4 == 6) {
+				packets = [][]byte{transfer}
+			} else {
+				errs <- e
+				return
+			}
 		}
-		if n != len(packet) {
-			errs <- io.ErrShortWrite
-			return
+		var rawBytes uint64
+		for _, packet := range packets {
+			n, e := v.tun.Write(packet)
+			if e != nil {
+				errs <- e
+				return
+			}
+			if n != len(packet) {
+				errs <- io.ErrShortWrite
+				return
+			}
+			rawBytes += uint64(n)
 		}
-		v.downPackets.Add(1)
-		v.downBytes.Add(uint64(n))
+		v.downPackets.Add(uint64(len(packets)))
+		v.downBytes.Add(rawBytes)
+		v.downBatches.Add(1)
 		ack = want
 		want++
 		offset = 0
-		packet = nil
+		transfer = nil
 		total = 0
 	}
 }
@@ -419,8 +527,10 @@ func (v *vpnClient) run() error {
 		return err
 	}
 	fmt.Println("VPN READY")
-	errs := make(chan error, 2)
-	go v.uploadLoop(errs)
+	errs := make(chan error, 3)
+	packets := make(chan []byte, 256)
+	go v.tunReadLoop(packets, errs)
+	go v.uploadLoop(packets, errs)
 	go v.downloadLoop(errs)
 	ticker := time.NewTicker(5 * time.Second)
 	defer ticker.Stop()
@@ -429,7 +539,7 @@ func (v *vpnClient) run() error {
 		case err := <-errs:
 			return err
 		case <-ticker.C:
-			fmt.Printf("STATS up_packets=%d down_packets=%d up_bytes=%d down_bytes=%d upload_chunk=%d download_chunk=%d pollers=1\n", v.upPackets.Load(), v.downPackets.Load(), v.upBytes.Load(), v.downBytes.Load(), v.upSizer.Current(), v.downSizer.Current())
+			fmt.Printf("STATS up_packets=%d down_packets=%d up_batches=%d down_batches=%d up_bytes=%d down_bytes=%d local_dropped=%d upload_chunk=%d download_chunk=%d pollers=1\n", v.upPackets.Load(), v.downPackets.Load(), v.upBatches.Load(), v.downBatches.Load(), v.upBytes.Load(), v.downBytes.Load(), v.localDropped.Load(), v.upSizer.Current(), v.downSizer.Current())
 		case <-v.stopped:
 			return nil
 		}
@@ -445,11 +555,12 @@ func main() {
 	ipv4Text := flag.String("vpn-ipv4", "10.123.0.2", "client VPN IPv4 address")
 	ipv6Text := flag.String("vpn-ipv6", "fd7a:4472:6167:6f6e::2", "client VPN IPv6 address")
 	mtu := flag.Int("vpn-mtu", 1280, "VPN interface MTU")
-	chunkMax := flag.Int("chunk-max", 65535, "maximum adaptive record bytes")
+	chunkMax := flag.Int("chunk-max", protocol.VPNMaxFragment, "maximum adaptive record bytes (up to 1 MiB)")
 	chunkMin := flag.Int("chunk-min", 32, "minimum adaptive record bytes")
-	chunkStart := flag.Int("chunk-start", 65535, "starting record bytes; app sets this equal to max")
+	chunkStart := flag.Int("chunk-start", protocol.VPNMaxFragment, "starting record bytes; app sets this equal to max")
 	growAfter := flag.Int("chunk-grow-after", 64, "full successful records before increasing chunk size")
 	timeout := flag.Duration("chunk-timeout", 2*time.Second, "framed transaction timeout")
+	batchDelay := flag.Duration("batch-delay", time.Millisecond, "maximum delay used to combine adjacent TUN packets into one transfer object")
 	reconnectEvery := flag.Int("chunk-reconnect-every", 32, "reconnect a TCP/53 lane after this many transactions; 0 keeps it open")
 	adaptLog := flag.Bool("chunk-adapt-log", false, "log adaptive chunk changes")
 	flag.Parse()
@@ -496,7 +607,7 @@ func main() {
 		}
 	}
 	addr := net.JoinHostPort(*serverHost, strconv.Itoa(*serverPort))
-	client, err := newVPNClient(tun, addr, *token, v4, v6, *mtu, *chunkStart, *chunkMin, *chunkMax, *growAfter, *reconnectEvery, *timeout, *adaptLog)
+	client, err := newVPNClient(tun, addr, *token, v4, v6, *mtu, *chunkStart, *chunkMin, *chunkMax, *growAfter, *reconnectEvery, *timeout, *batchDelay, *adaptLog)
 	if err != nil {
 		fmt.Fprintln(os.Stderr, err)
 		os.Exit(1)
@@ -1,6 +1,7 @@
 package main

 import (
+	"bytes"
 	"crypto/subtle"
 	"encoding/hex"
 	"errors"
@@ -36,6 +37,8 @@ type debugStats struct {
 	activeSessions atomic.Int64
 	upPackets      atomic.Uint64
 	downPackets    atomic.Uint64
+	upBatches      atomic.Uint64
+	downBatches    atomic.Uint64
 	upBytes        atomic.Uint64
 	downBytes      atomic.Uint64
 	dropped        atomic.Uint64
@@ -69,20 +72,27 @@ func tokenEqual(a, b string) bool {
 }

 type vpnSession struct {
-	sid        protocol.VPNSessionID
-	ipv4       netip.Addr
-	ipv6       netip.Addr
-	mtu        int
-	maxChunk   int
-	maxPackets int
-	manager    *vpnManager
+	sid           protocol.VPNSessionID
+	ipv4          netip.Addr
+	ipv6          netip.Addr
+	mtu           int
+	maxChunk      int
+	maxPackets    int
+	maxQueueBytes int
+	batchDelay    time.Duration
+	manager       *vpnManager

-	mu       sync.Mutex
-	notify   chan struct{}
-	packets  map[uint32][]byte
-	nextDown uint32
-	closed   bool
-	lastSeen time.Time
+	mu                sync.Mutex
+	notify            chan struct{}
+	packets           map[uint32]*downTransfer
+	nextDown          uint32
+	closed            bool
+	lastSeen          time.Time
+	pendingPackets    [][]byte
+	pendingEncoded    int
+	pendingTimer      *time.Timer
+	queuedPacketCount int
+	queuedBytes       int

 	upMu              sync.Mutex
 	expectedUp        uint32
@@ -95,10 +105,16 @@ type vpnSession struct {
 	haveLastComplete  bool
 }

-func newVPNSession(m *vpnManager, sid protocol.VPNSessionID, v4, v6 netip.Addr, mtu, maxChunk, maxPackets int) *vpnSession {
+type downTransfer struct {
+	data        []byte
+	packetCount int
+	rawBytes    int
+}
+
+func newVPNSession(m *vpnManager, sid protocol.VPNSessionID, v4, v6 netip.Addr, mtu, maxChunk, maxPackets, maxQueueBytes int, batchDelay time.Duration) *vpnSession {
 	return &vpnSession{
-		sid: sid, ipv4: v4, ipv6: v6, mtu: mtu, maxChunk: maxChunk, maxPackets: maxPackets,
-		manager: m, notify: make(chan struct{}), packets: make(map[uint32][]byte, maxPackets), lastSeen: time.Now(),
+		sid: sid, ipv4: v4, ipv6: v6, mtu: mtu, maxChunk: maxChunk, maxPackets: maxPackets, maxQueueBytes: maxQueueBytes, batchDelay: batchDelay,
+		manager: m, notify: make(chan struct{}), packets: make(map[uint32]*downTransfer, maxPackets), lastSeen: time.Now(),
 	}
 }

@@ -109,8 +125,54 @@ func (s *vpnSession) signalLocked() {
 func (s *vpnSession) touchLocked() { s.lastSeen = time.Now() }
 func (s *vpnSession) touch()       { s.mu.Lock(); s.touchLocked(); s.mu.Unlock() }

+func (s *vpnSession) flushPendingLocked() {
+	if len(s.pendingPackets) == 0 {
+		return
+	}
+	if s.pendingTimer != nil {
+		s.pendingTimer.Stop()
+		s.pendingTimer = nil
+	}
+	batch, err := protocol.BuildVPNBatch(s.pendingPackets)
+	if err != nil {
+		if s.manager.debug != nil {
+			s.manager.debug.dropped.Add(uint64(len(s.pendingPackets)))
+			s.manager.debug.errorf("BATCH sid=%s: %v", shortSID(s.sid), err)
+		}
+		s.queuedPacketCount -= len(s.pendingPackets)
+		for _, p := range s.pendingPackets {
+			s.queuedBytes -= len(p)
+		}
+		s.pendingPackets = nil
+		s.pendingEncoded = 0
+		return
+	}
+	rawBytes := 0
+	for _, p := range s.pendingPackets {
+		rawBytes += len(p)
+	}
+	seq := s.nextDown
+	s.nextDown++
+	s.packets[seq] = &downTransfer{data: batch, packetCount: len(s.pendingPackets), rawBytes: rawBytes}
+	if s.manager.debug != nil {
+		s.manager.debug.downBatches.Add(1)
+		s.manager.debug.packetf("BATCH QUEUE sid=%s seq=%d packets=%d raw_bytes=%d transfer_bytes=%d", shortSID(s.sid), seq, len(s.pendingPackets), rawBytes, len(batch))
+	}
+	s.pendingPackets = nil
+	s.pendingEncoded = 0
+	s.signalLocked()
+}
+
+func (s *vpnSession) flushPending() {
+	s.mu.Lock()
+	if !s.closed {
+		s.flushPendingLocked()
+	}
+	s.mu.Unlock()
+}
+
 func (s *vpnSession) enqueue(packet []byte) bool {
-	if len(packet) == 0 || len(packet) > 65535 {
+	if len(packet) == 0 || len(packet) > protocol.VPNMaxPacket {
 		return false
 	}
 	s.mu.Lock()
@@ -118,21 +180,39 @@ func (s *vpnSession) enqueue(packet []byte) bool {
 	if s.closed {
 		return false
 	}
-	if len(s.packets) >= s.maxPackets {
+	need := 2 + len(packet)
+	if len(s.pendingPackets) > 0 && s.pendingEncoded+need > protocol.VPNMaxBatch {
+		s.flushPendingLocked()
+	}
+	if s.queuedPacketCount >= s.maxPackets || s.queuedBytes+len(packet) > s.maxQueueBytes {
 		if s.manager.debug != nil {
 			s.manager.debug.dropped.Add(1)
 		}
 		return false
 	}
-	seq := s.nextDown
-	s.nextDown++
-	s.packets[seq] = append([]byte(nil), packet...)
+	p := append([]byte(nil), packet...)
+	if len(s.pendingPackets) == 0 {
+		s.pendingEncoded = 1
+	}
+	s.pendingPackets = append(s.pendingPackets, p)
+	s.pendingEncoded += 2 + len(p)
+	s.queuedPacketCount++
+	s.queuedBytes += len(p)
 	s.touchLocked()
-	s.signalLocked()
 	if s.manager.debug != nil {
 		s.manager.debug.downPackets.Add(1)
 		s.manager.debug.downBytes.Add(uint64(len(packet)))
-		s.manager.debug.packetf("QUEUE sid=%s seq=%d bytes=%d", shortSID(s.sid), seq, len(packet))
+		s.manager.debug.packetf("QUEUE sid=%s bytes=%d pending_packets=%d pending_transfer=%d", shortSID(s.sid), len(packet), len(s.pendingPackets), s.pendingEncoded)
+	}
+	if s.pendingEncoded >= protocol.VPNMaxBatch {
+		s.flushPendingLocked()
+	} else if s.pendingTimer == nil {
+		delay := s.batchDelay
+		if delay <= 0 {
+			s.flushPendingLocked()
+		} else {
+			s.pendingTimer = time.AfterFunc(delay, s.flushPending)
+		}
 	}
 	return true
 }
@@ -140,8 +220,8 @@ func (s *vpnSession) enqueue(packet []byte) bool {
 func (s *vpnSession) push(seq uint32, offset, total int, data []byte) (int, error) {
 	s.upMu.Lock()
 	defer s.upMu.Unlock()
-	if total < 1 || total > 65535 || len(data) < 1 || len(data) > s.maxChunk || offset < 0 || offset+len(data) > total {
-		return 0, errors.New("invalid packet fragment")
+	if total < 1 || total > protocol.VPNMaxBatch || len(data) < 1 || len(data) > s.maxChunk || offset < 0 || offset+len(data) > total {
+		return 0, errors.New("invalid transfer fragment")
 	}

 	if s.haveLastComplete && seq == s.lastComplete {
@@ -165,14 +245,14 @@ func (s *vpnSession) push(seq uint32, offset, total int, data []byte) (int, erro
 		s.currentBuf = make([]byte, 0, total)
 	}
 	if s.currentSeq != seq || s.currentTotal != total {
-		return 0, errors.New("packet fragment metadata changed")
+		return 0, errors.New("transfer fragment metadata changed")
 	}

 	// Idempotent retry: if this exact offset was already accepted, acknowledge
 	// the existing bytes instead of appending duplicate data.
 	if offset < len(s.currentBuf) {
 		end := offset + len(data)
-		if end <= len(s.currentBuf) && string(s.currentBuf[offset:end]) == string(data) {
+		if end <= len(s.currentBuf) && bytes.Equal(s.currentBuf[offset:end], data) {
 			return len(s.currentBuf), nil
 		}
 		return 0, errors.New("retry fragment does not match accepted data")
@@ -188,11 +268,11 @@ func (s *vpnSession) push(seq uint32, offset, total int, data []byte) (int, erro
 		return accepted, nil
 	}

-	packet := append([]byte(nil), s.currentBuf...)
+	transfer := append([]byte(nil), s.currentBuf...)
 	s.haveCurrent = false
 	s.currentBuf = nil

-	if err := s.manager.acceptClientPacket(s, packet); err != nil {
+	if err := s.manager.acceptClientTransfer(s, transfer); err != nil {
 		return 0, err
 	}

@@ -202,9 +282,8 @@ func (s *vpnSession) push(seq uint32, offset, total int, data []byte) (int, erro
 	s.expectedUp++
 	s.touch()
 	if s.manager.debug != nil {
-		s.manager.debug.upPackets.Add(1)
-		s.manager.debug.upBytes.Add(uint64(len(packet)))
-		s.manager.debug.packetf("UP sid=%s seq=%d bytes=%d", shortSID(s.sid), seq, len(packet))
+		s.manager.debug.upBatches.Add(1)
+		s.manager.debug.packetf("UP BATCH sid=%s seq=%d transfer_bytes=%d", shortSID(s.sid), seq, len(transfer))
 	}
 	return accepted, nil
 }
@@ -221,21 +300,26 @@ func (s *vpnSession) pull(ack, want uint32, offset, limit int, wait time.Duratio
 		if ack != protocol.VPNNoAck {
 			for seq := range s.packets {
 				if seq <= ack {
+					rec := s.packets[seq]
+					if rec != nil {
+						s.queuedPacketCount -= rec.packetCount
+						s.queuedBytes -= rec.rawBytes
+					}
 					delete(s.packets, seq)
 				}
 			}
 		}
-		if packet, ok := s.packets[want]; ok {
-			if offset >= len(packet) {
+		if rec, ok := s.packets[want]; ok {
+			if offset >= len(rec.data) {
 				s.mu.Unlock()
-				return nil, len(packet), false, errors.New("pull offset beyond packet")
+				return nil, len(rec.data), false, errors.New("pull offset beyond transfer")
 			}
 			end := offset + limit
-			if end > len(packet) {
-				end = len(packet)
+			if end > len(rec.data) {
+				end = len(rec.data)
 			}
-			out := append([]byte(nil), packet[offset:end]...)
-			total := len(packet)
+			out := append([]byte(nil), rec.data[offset:end]...)
+			total := len(rec.data)
 			s.mu.Unlock()
 			return out, total, false, nil
 		}
@@ -257,35 +341,41 @@ func (s *vpnSession) close() {
 	s.mu.Lock()
 	if !s.closed {
 		s.closed = true
+		if s.pendingTimer != nil {
+			s.pendingTimer.Stop()
+			s.pendingTimer = nil
+		}
 		s.signalLocked()
 	}
 	s.mu.Unlock()
 }

 type vpnManager struct {
-	mu           sync.RWMutex
-	sessions     map[protocol.VPNSessionID]*vpnSession
-	byIPv4       map[netip.Addr]*vpnSession
-	byIPv6       map[netip.Addr]*vpnSession
-	maxChunk     int
-	maxPackets   int
-	pollWait     time.Duration
-	timeout      time.Duration
-	tun          *os.File
-	tunWriteMu   sync.Mutex
-	mockEcho     bool
-	allowPrivate bool
-	debug        *debugStats
-	v4Prefix     netip.Prefix
-	v6Prefix     netip.Prefix
+	mu            sync.RWMutex
+	sessions      map[protocol.VPNSessionID]*vpnSession
+	byIPv4        map[netip.Addr]*vpnSession
+	byIPv6        map[netip.Addr]*vpnSession
+	maxChunk      int
+	maxPackets    int
+	maxQueueBytes int
+	batchDelay    time.Duration
+	pollWait      time.Duration
+	timeout       time.Duration
+	tun           *os.File
+	tunWriteMu    sync.Mutex
+	mockEcho      bool
+	allowPrivate  bool
+	debug         *debugStats
+	v4Prefix      netip.Prefix
+	v6Prefix      netip.Prefix
 }

-func newVPNManager(tun *os.File, mockEcho bool, maxChunk, maxPackets int, pollWait, timeout time.Duration, allowPrivate bool, debug *debugStats) *vpnManager {
+func newVPNManager(tun *os.File, mockEcho bool, maxChunk, maxPackets, maxQueueBytes int, pollWait, timeout, batchDelay time.Duration, allowPrivate bool, debug *debugStats) *vpnManager {
 	v4p := netip.MustParsePrefix(defaultVPNv4Prefix)
 	v6p := netip.MustParsePrefix(defaultVPNv6Prefix)
 	m := &vpnManager{
 		sessions: make(map[protocol.VPNSessionID]*vpnSession), byIPv4: make(map[netip.Addr]*vpnSession), byIPv6: make(map[netip.Addr]*vpnSession),
-		maxChunk: maxChunk, maxPackets: maxPackets, pollWait: pollWait, timeout: timeout, tun: tun, mockEcho: mockEcho, allowPrivate: allowPrivate, debug: debug,
+		maxChunk: maxChunk, maxPackets: maxPackets, maxQueueBytes: maxQueueBytes, pollWait: pollWait, timeout: timeout, batchDelay: batchDelay, tun: tun, mockEcho: mockEcho, allowPrivate: allowPrivate, debug: debug,
 		v4Prefix: v4p, v6Prefix: v6p,
 	}
 	if tun != nil {
@@ -318,7 +408,7 @@ func (m *vpnManager) addOrGet(sid protocol.VPNSessionID, v4, v6 netip.Addr, mtu
 	if m.byIPv4[v4] != nil || m.byIPv6[v6] != nil {
 		return nil, errors.New("client VPN address already in use")
 	}
-	s := newVPNSession(m, sid, v4, v6, mtu, m.maxChunk, m.maxPackets)
+	s := newVPNSession(m, sid, v4, v6, mtu, m.maxChunk, m.maxPackets, m.maxQueueBytes, m.batchDelay)
 	m.sessions[sid] = s
 	m.byIPv4[v4] = s
 	m.byIPv6[v6] = s
@@ -376,40 +466,6 @@ func (m *vpnManager) cleanupLoop() {
 	}
 }

-func packetAddresses(packet []byte) (src, dst netip.Addr, err error) {
-	if len(packet) < 1 {
-		return src, dst, errors.New("empty IP packet")
-	}
-	switch packet[0] >> 4 {
-	case 4:
-		if len(packet) < 20 {
-			return src, dst, errors.New("short IPv4 packet")
-		}
-		total := int(packet[2])<<8 | int(packet[3])
-		if total < 20 || total > len(packet) {
-			return src, dst, errors.New("invalid IPv4 total length")
-		}
-		var a, b [4]byte
-		copy(a[:], packet[12:16])
-		copy(b[:], packet[16:20])
-		return netip.AddrFrom4(a), netip.AddrFrom4(b), nil
-	case 6:
-		if len(packet) < 40 {
-			return src, dst, errors.New("short IPv6 packet")
-		}
-		total := 40 + (int(packet[4])<<8 | int(packet[5]))
-		if total > len(packet) {
-			return src, dst, errors.New("invalid IPv6 payload length")
-		}
-		var a, b [16]byte
-		copy(a[:], packet[8:24])
-		copy(b[:], packet[24:40])
-		return netip.AddrFrom16(a), netip.AddrFrom16(b), nil
-	default:
-		return src, dst, errors.New("unsupported IP version")
-	}
-}
-
 func destinationAllowed(dst netip.Addr, allowPrivate bool) bool {
 	if dst.IsUnspecified() || dst.IsMulticast() {
 		return false
@@ -423,32 +479,68 @@ func destinationAllowed(dst netip.Addr, allowPrivate bool) bool {
 	return true
 }

-func (m *vpnManager) acceptClientPacket(s *vpnSession, packet []byte) error {
-	src, dst, err := packetAddresses(packet)
+func (m *vpnManager) dropClientPacket(s *vpnSession, packet []byte, reason string) {
+	if m.debug != nil {
+		m.debug.dropped.Add(1)
+		m.debug.packetf("DROP sid=%s bytes=%d reason=%s", shortSID(s.sid), len(packet), reason)
+		// A source mismatch can be normal Android link-local/control traffic.
+		// Never tear down the whole VPN session for one such packet.
+		m.debug.logf("DROP sid=%s reason=%s", shortSID(s.sid), reason)
+	}
+}
+
+func (m *vpnManager) acceptClientPacket(s *vpnSession, packet []byte) (bool, error) {
+	src, dst, err := protocol.PacketAddresses(packet)
 	if err != nil {
-		return err
+		m.dropClientPacket(s, packet, err.Error())
+		return false, nil
 	}
 	if src != s.ipv4 && src != s.ipv6 {
-		return fmt.Errorf("source %s does not match session address", src)
+		m.dropClientPacket(s, packet, fmt.Sprintf("source %s does not match session address", src))
+		return false, nil
 	}
 	if !destinationAllowed(dst, m.allowPrivate) {
-		return fmt.Errorf("destination %s is blocked; use --allow-private to permit it", dst)
+		m.dropClientPacket(s, packet, fmt.Sprintf("destination %s is blocked", dst))
+		return false, nil
 	}
 	if m.mockEcho {
 		s.enqueue(packet)
-		return nil
+		return true, nil
 	}
 	if m.tun == nil {
-		return errors.New("VPN TUN is unavailable")
+		return false, errors.New("VPN TUN is unavailable")
 	}
 	m.tunWriteMu.Lock()
 	n, err := m.tun.Write(packet)
 	m.tunWriteMu.Unlock()
 	if err != nil {
-		return err
+		return false, err
 	}
 	if n != len(packet) {
-		return io.ErrShortWrite
+		return false, io.ErrShortWrite
+	}
+	return true, nil
+}
+
+func (m *vpnManager) acceptClientTransfer(s *vpnSession, transfer []byte) error {
+	packets, err := protocol.ParseVPNBatch(transfer)
+	if err != nil {
+		// Compatibility with the first packet-VPN build.
+		if len(transfer) > 0 && (transfer[0]>>4 == 4 || transfer[0]>>4 == 6) {
+			packets = [][]byte{transfer}
+		} else {
+			return err
+		}
+	}
+	for _, packet := range packets {
+		accepted, err := m.acceptClientPacket(s, packet)
+		if err != nil {
+			return err
+		}
+		if accepted && m.debug != nil {
+			m.debug.upPackets.Add(1)
+			m.debug.upBytes.Add(uint64(len(packet)))
+		}
 	}
 	return nil
 }
@@ -467,7 +559,7 @@ func (m *vpnManager) tunReadLoop() {
 			continue
 		}
 		packet := append([]byte(nil), buf[:n]...)
-		_, dst, e := packetAddresses(packet)
+		_, dst, e := protocol.PacketAddresses(packet)
 		if e != nil {
 			continue
 		}
@@ -667,8 +759,10 @@ func main() {
 	port := flag.Int("port", 53, "listen TCP port")
 	token := flag.String("token", "change-this-token", "shared token")
 	maxConnections := flag.Int("max-connections", 20000, "maximum simultaneous TCP/53 connections")
-	maxChunk := flag.Int("chunk-max", 65535, "maximum VPN fragment payload bytes (32-65535)")
+	maxChunk := flag.Int("chunk-max", protocol.VPNMaxFragment, "maximum DragonTCP transport fragment bytes (32-1048576)")
 	maxPackets := flag.Int("vpn-buffered-packets", 2048, "maximum queued return IP packets per client")
+	maxQueueBytes := flag.Int("vpn-buffer-bytes", 8*1024*1024, "maximum queued raw return bytes per client")
+	batchDelay := flag.Duration("batch-delay", time.Millisecond, "maximum delay to combine adjacent TUN packets into one transfer object")
 	pollWait := flag.Duration("poll-wait", 100*time.Millisecond, "long-poll wait for a return packet")
 	sessionTimeout := flag.Duration("session-timeout", 5*time.Minute, "idle VPN session timeout")
 	tunName := flag.String("tun", "dragontcp0", "Linux TUN interface name")
@@ -684,6 +778,10 @@ func main() {
 		fmt.Fprintf(os.Stderr, "--chunk-max must be 32-%d\n", protocol.VPNMaxFragment)
 		os.Exit(2)
 	}
+	if *maxPackets < 1 || *maxQueueBytes < protocol.VPNMaxPacket {
+		fmt.Fprintln(os.Stderr, "invalid VPN buffer limits")
+		os.Exit(2)
+	}
 	if *mtu < 576 || *mtu > 9000 {
 		fmt.Fprintln(os.Stderr, "--mtu must be 576-9000")
 		os.Exit(2)
@@ -699,7 +797,7 @@ func main() {
 		}
 		defer tun.Close()
 	}
-	manager := newVPNManager(tun, *mockEcho, *maxChunk, *maxPackets, *pollWait, *sessionTimeout, *allowPrivate, debug)
+	manager := newVPNManager(tun, *mockEcho, *maxChunk, *maxPackets, *maxQueueBytes, *pollWait, *sessionTimeout, *batchDelay, *allowPrivate, debug)
 	addr := net.JoinHostPort(*host, strconv.Itoa(*port))
 	ln, err := net.Listen("tcp", addr)
 	if err != nil {
@@ -713,13 +811,13 @@ func main() {
 	} else {
 		fmt.Printf("tun=%s mtu=%d IPv4=10.123.0.1/16 IPv6=fd7a:4472:6167:6f6e::1/64 auto_nat=%t\n", *tunName, *mtu, *autoNAT)
 	}
-	fmt.Printf("chunk_max=%d poll_wait=%s buffered_packets=%d\n", *maxChunk, pollWait.String(), *maxPackets)
+	fmt.Printf("chunk_max=%d batch_max=%d batch_delay=%s poll_wait=%s buffered_packets=%d buffer_bytes=%d\n", *maxChunk, protocol.VPNMaxBatch, batchDelay.String(), pollWait.String(), *maxPackets, *maxQueueBytes)
 	if debug.enabled && *statsEvery > 0 {
 		go func() {
 			t := time.NewTicker(*statsEvery)
 			defer t.Stop()
 			for range t.C {
-				fmt.Printf("[DEBUG] STATS uptime=%s conns=%d sessions=%d up_packets=%d down_packets=%d up_bytes=%d down_bytes=%d dropped=%d errors=%d\n", time.Since(debug.started).Round(time.Second), debug.activeConns.Load(), debug.activeSessions.Load(), debug.upPackets.Load(), debug.downPackets.Load(), debug.upBytes.Load(), debug.downBytes.Load(), debug.dropped.Load(), debug.errors.Load())
+				fmt.Printf("[DEBUG] STATS uptime=%s conns=%d sessions=%d up_packets=%d down_packets=%d up_batches=%d down_batches=%d up_bytes=%d down_bytes=%d dropped=%d errors=%d\n", time.Since(debug.started).Round(time.Second), debug.activeConns.Load(), debug.activeSessions.Load(), debug.upPackets.Load(), debug.downPackets.Load(), debug.upBatches.Load(), debug.downBatches.Load(), debug.upBytes.Load(), debug.downBytes.Load(), debug.dropped.Load(), debug.errors.Load())
 			}
 		}()
 	}