feat(tcp): вернуть поддержку bond

2026-10-12 09:41:53 +02:00 · 2026-09-22 22:33:56 +03:00
parent a8b8363b79
commit 6f84a18b43
25 changed files with 1099 additions and 10 deletions
@@ -0,0 +1,21 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/samosvalishe/free-turn-proxy/internal/clientsdb"
+	"github.com/samosvalishe/free-turn-proxy/internal/config"
+)
+
+func TestAdmitTCPAndBond(t *testing.T) {
+	for _, mode := range []config.ProxyMode{config.ProxyModeTCP, config.ProxyModeUDP} {
+		cfg := &config.Server{Proxy: config.ProxyOpts{Mode: mode}}
+		for _, tag := range []byte{clientsdb.ModeUnset, clientsdb.ModeUDP, clientsdb.ModeTCP, clientsdb.ModeTCPBond, 255} {
+			err := admitClient(cfg, nil, "test-client", tag)
+			want := (mode == config.ProxyModeTCP && (tag == clientsdb.ModeTCP || tag == clientsdb.ModeTCPBond)) || (mode == config.ProxyModeUDP && (tag == clientsdb.ModeUnset || tag == clientsdb.ModeUDP))
+			if (err == nil) != want {
+				t.Fatalf("mode=%s tag=%d err=%v", mode, tag, err)
+			}
+		}
+	}
+}
@@ -15,6 +15,7 @@ import (
 	"github.com/samosvalishe/free-turn-proxy/internal/clientsdb"
 	"github.com/samosvalishe/free-turn-proxy/internal/config"
 	"github.com/samosvalishe/free-turn-proxy/internal/logx"
+	"github.com/samosvalishe/free-turn-proxy/internal/proxy/bond"
 	"github.com/samosvalishe/free-turn-proxy/internal/proxy/tcpserver"
 	"github.com/samosvalishe/free-turn-proxy/internal/proxy/udpserver"
 	"github.com/samosvalishe/free-turn-proxy/internal/safego"
@@ -124,6 +125,7 @@ func listen(cfg *config.Server, addr *net.UDPAddr, logger logx.Logger) (net.List
 }

 func serve(ctx context.Context, logger logx.Logger, listener net.Listener, db *clientsdb.DB, cfg *config.Server) {
+	bonds := &bond.Server{}
 	var wg sync.WaitGroup
 	var backoff time.Duration
 	for {
@@ -152,7 +154,7 @@ func serve(ctx context.Context, logger logx.Logger, listener net.Listener, db *c
 		}
 		backoff = 0
 		wg.Go(func() {
-			_ = safego.Run(logger, func() { handleAccepted(ctx, logger, db, conn, cfg) })
+			_ = safego.Run(logger, func() { handleAccepted(ctx, logger, db, conn, cfg, bonds) })
 		})
 	}
 }
@@ -180,13 +182,16 @@ func wireMode(m config.ProxyMode) byte {
 }

 func modeName(b byte) string {
+	if b == clientsdb.ModeTCPBond {
+		return "tcp-bond"
+	}
 	if b == clientsdb.ModeTCP {
 		return string(config.ProxyModeTCP)
 	}
 	return string(config.ProxyModeUDP)
 }

-func handleAccepted(ctx context.Context, logger logx.Logger, db *clientsdb.DB, conn net.Conn, cfg *config.Server) {
+func handleAccepted(ctx context.Context, logger logx.Logger, db *clientsdb.DB, conn net.Conn, cfg *config.Server, bonds *bond.Server) {
 	defer func() {
 		if closeErr := conn.Close(); closeErr != nil {
 			logger.Warnf("failed to close incoming connection: %s", closeErr)
@@ -211,7 +216,9 @@ func handleAccepted(ctx context.Context, logger logx.Logger, db *clientsdb.DB, c
 	}
 	logger.Debugf("Handshake done")

+	var clientMode byte
 	clientID, data, err := clientsdb.AcceptClientID(dtlsConn, func(id string, mode byte) error {
+		clientMode = mode
 		return admitClient(cfg, db, id, mode)
 	})
 	if err != nil {
@@ -220,9 +227,12 @@ func handleAccepted(ctx context.Context, logger logx.Logger, db *clientsdb.DB, c
 	}

 	logger.Infof("Session up: client=%s from=%s", clientID, conn.RemoteAddr())
-	if cfg.Proxy.Mode == config.ProxyModeTCP {
+	switch {
+	case clientMode == clientsdb.ModeTCPBond:
+		tcpserver.HandleBond(ctx, logger, data, cfg.Proxy.Connect, cfg.KCP.Profile, bonds, clientID)
+	case cfg.Proxy.Mode == config.ProxyModeTCP:
 		tcpserver.Handle(ctx, logger, data, cfg.Proxy.Connect, cfg.KCP.Profile)
-	} else {
+	default:
 		udpserver.Handle(ctx, logger, data, cfg.Proxy.Connect)
 	}
 	logger.Infof("Session down: client=%s from=%s", clientID, conn.RemoteAddr())
@@ -230,6 +240,9 @@ func handleAccepted(ctx context.Context, logger logx.Logger, db *clientsdb.DB, c

 // admitClient: режим клиента обязан совпасть с сервером, ID - быть в allowlist (если он есть).
 func admitClient(cfg *config.Server, db *clientsdb.DB, id string, mode byte) error {
+	if mode == clientsdb.ModeTCPBond {
+		mode = clientsdb.ModeTCP
+	}
 	if want := wireMode(cfg.Proxy.Mode); mode != clientsdb.ModeUnset && mode != want {
 		return fmt.Errorf("mode mismatch: клиент %s, сервер %s - приведите -mode к одному значению",
 			modeName(mode), cfg.Proxy.Mode)
@@ -12,6 +12,7 @@
 | `-n`                | `12` (`direct`: `1`)                              | параллельных потоков; в `-mode tcp` это число сессий пула, по которым round-robin раскладываются локальные TCP-соединения                                                                                                                                 |
 | `-transport`        | `tcp`                                             | транспорт до TURN-реле: `tcp` (TCP/TLS) \| `udp`                                                                                                                                                                                                          |
 | `-mode`             | `udp`                                             | режим туннеля: `udp` (UDP-релей для WireGuard) \| `tcp` (TCP-форвардер для Xray/sing-box)                                                                                                                                                                 |
+| `-bond` | `false` | объединять сессии для одного TCP-соединения; только `-mode tcp`, до 256 сессий суммарно; нужен обновлённый сервер |
 | `-turn`             | из creds                                          | переопределить IP TURN-сервера                                                                                                                                                                                                                            |
 | `-port`             | из creds                                          | переопределить порт TURN-сервера                                                                                                                                                                                                                          |
 | `-obf-profile`      | `none`                                            | wire-профиль обфускации payload: `none` \| `rtpopus` (RTP/opus + ChaCha20-Poly1305 AEAD) \| `rtpopus2` (+ RTP header extension, ближе к WebRTC) \| `rtpopus3` (+ abs-send-time, VAD, имитация потерь, вариативный timestamp); должен совпадать с сервером |
@@ -32,6 +32,18 @@ Relayed-данные TURN - всегда датаграммы, даже при `

 `-n` в этом режиме задаёт число сессий пула: каждое локальное TCP-соединение уходит целиком в одну из них (round-robin).

+### Bond для одного TCP-соединения
+
+Добавьте клиенту `-bond`: данные каждого TCP-соединения будут распределяться между доступными сессиями. На сервере достаточно `-mode tcp`, он одновременно принимает обычные и bond-клиенты. По умолчанию bond выключен.
+
+```bash
+./client -listen 127.0.0.1:9000 -peer <vps>:56000 -links "<call-link>" -mode tcp -bond -n 4
+```
+
+Состав каналов фиксируется при открытии TCP-соединения; с одной доступной сессией bond тоже работает. Поддерживается до 256 сессий суммарно по всем ссылкам. Новые сессии используются следующими соединениями. При отказе участвующей сессии текущее соединение закрывается, а Xray может открыть новое через оставшиеся. Бесшовного переноса уже переданных данных между каналами нет. Медленный канал задерживает сборку потока; прирост скорости зависит от сети.
+
+Для bond обновите и клиент, и сервер: сервер без его поддержки отклонит такой клиент при проверке режима. Обычный TCP сохраняет прежний протокол. В JSON клиента настройка называется `proxy.bond`, в `freeturn://` - `bond`.
+
 Встроенный WireGuard (`-tunnel`) с `-mode tcp` не сочетается: он гонит датаграммы.

 Режим должен совпадать на обеих сторонах: клиент помечает им первую запись, и сервер с другим `-mode` закрывает соединение с явной ошибкой вместо молчаливого зависания.
@@ -26,6 +26,7 @@ Payload - JSON-объект, закодированный `base64url` (без pa
 | `peer` | `-peer` | адрес сервера на VPS (`ip:port`). |
 | `transport` | `-transport` | транспорт до TURN-реле: `tcp` \| `udp`. |
 | `mode` | `-mode` | режим туннеля: `udp` \| `tcp`; отсутствует - `udp`. |
+| `bond` | `-bond` | объединять сессии для одного TCP-соединения (`true`); только `mode: tcp`, по умолчанию выключено. |
 | `obf` | `-obf-profile` | профиль обфускации (`rtpopus` \| `rtpopus2` \| `rtpopus3`); `none` опускается. |
 | `key` | `-obf-key` | ключ обфускации (hex), только при заданном `obf`. |
 | `n` | `-n` | число TURN-потоков; отсутствует - дефолт клиента. |
@@ -190,9 +190,10 @@ func syncDir(dir string) {
 }

 const (
-	ModeUnset byte = 0
-	ModeUDP   byte = 1
-	ModeTCP   byte = 2
+	ModeUnset   byte = 0
+	ModeUDP     byte = 1
+	ModeTCP     byte = 2
+	ModeTCPBond byte = 3

 	// idVersionAck - клиент ждёт подтверждения ID.
 	idVersionAck byte = 2
@@ -30,6 +30,7 @@ func ClientArgs(c *Client) []string {
 	addIf(c.VK.StreamsPerCred != def.VK.StreamsPerCred, "-streams-per-cred", strconv.Itoa(c.VK.StreamsPerCred))
 	addIf(c.TURN.TransportUDP != def.TURN.TransportUDP, "-transport", TransportUDP)
 	addIf(c.Proxy.Mode != def.Proxy.Mode, "-mode", string(c.Proxy.Mode))
+	addIf(c.Proxy.Bond, "-bond")
 	args = append(args, kcpArgs(c.KCP.Profile, def.KCP.Profile)...)
 	addIf(c.Obf.Enabled(), "-obf-profile", string(c.Obf.Profile))
 	addIf(c.Obf.Enabled(), "-obf-key", hex.EncodeToString(c.Obf.Key))
@@ -0,0 +1,54 @@
+package config
+
+import (
+	"errors"
+	"io"
+	"strings"
+	"testing"
+
+	"github.com/samosvalishe/free-turn-proxy/internal/uri"
+)
+
+func TestBondConfigRoundTrip(t *testing.T) {
+	args := []string{"-provider", "direct", "-peer", "127.0.0.1:56000", "-mode", "tcp", "-bond", "-n", "3"}
+	c, err := ParseClient(args, io.Discard)
+	if err != nil {
+		t.Fatal(err)
+	}
+	round, err := ParseClient(ClientArgs(c), io.Discard)
+	if err != nil || !round.Proxy.Bond {
+		t.Fatalf("args round trip: %v", err)
+	}
+	link := (&uri.Config{Provider: "direct", Peer: "127.0.0.1:56000", Mode: "tcp", Bond: true, N: 3}).String()
+	parsed, err := uri.Parse(link)
+	if err != nil || !parsed.Bond {
+		t.Fatalf("URI round trip: %v", err)
+	}
+	c, err = ParseClient([]string{link}, io.Discard)
+	if err != nil || !c.Proxy.Bond {
+		t.Fatalf("URI to CLI: %v", err)
+	}
+	c, err = ParseClientJSON([]byte(`{"provider":"direct","peer":"127.0.0.1:56000","proxy":{"mode":"tcp","bond":true}}`), "")
+	if err != nil || !c.Proxy.Bond {
+		t.Fatalf("JSON: %v", err)
+	}
+	c, err = ParseClientJSON([]byte(`{}`), link)
+	if err != nil || !c.Proxy.Bond {
+		t.Fatalf("URI overlay: %v", err)
+	}
+	if strings.Contains(DefaultClientJSON(), `"bond"`) {
+		t.Fatal("default JSON changed")
+	}
+}
+
+func TestBondValidation(t *testing.T) {
+	for _, extra := range [][]string{{}, {"-mode", "tcp", "-n", "257"}} {
+		args := append([]string{"-provider", "direct", "-peer", "127.0.0.1:56000", "-bond"}, extra...)
+		if _, err := ParseClient(args, io.Discard); !errors.Is(err, ErrBondConfig) {
+			t.Fatalf("accepted %v: %v", args, err)
+		}
+	}
+	if _, err := ParseClientJSON([]byte(`{"provider":"direct","peer":"127.0.0.1:1","proxy":{"mode":"tcp","bonds":true}}`), ""); err == nil {
+		t.Fatal("unknown JSON field accepted")
+	}
+}
@@ -47,6 +47,7 @@ func ParseClient(args []string, errOut io.Writer) (*Client, error) {
 	fs.IntVar(&r.N, "n", r.N, "число параллельных потоков; 0 - по провайдеру (vk 12, direct 1)")
 	fs.StringVar(&r.Transport, "transport", r.Transport, "транспорт до TURN-реле: tcp | udp")
 	fs.StringVar(&r.Mode, "mode", r.Mode, "режим туннеля: udp (WireGuard) | tcp (Xray/sing-box)")
+	fs.BoolVar(&r.Bond, "bond", r.Bond, "объединять сессии для одного TCP-соединения; только -mode tcp")
 	fs.StringVar(&r.ObfProfile, "obf-profile", r.ObfProfile, "wire-профиль обфускации: none | rtpopus | rtpopus2 | rtpopus3; должен совпадать с сервером")
 	fs.StringVar(&r.ObfKey, "obf-key", r.ObfKey, "ключ для -obf-profile != none: 32 байта hex (64 символа)")
 	fs.BoolVar(&r.GenObfKey, "gen-obf-key", r.GenObfKey, "напечатать новый -obf-key и выйти")
@@ -42,6 +42,7 @@ const (
 )

 type ProxyOpts struct {
+	Bond    bool
 	Mode    ProxyMode
 	Listen  string
 	Connect string
@@ -43,6 +43,7 @@ type turnJSON struct {
 }

 type proxyJSON struct {
+	Bond   bool   `json:"bond,omitempty"`
 	Mode   string `json:"mode"`
 	Listen string `json:"listen"`
 }
@@ -167,6 +168,7 @@ func (j ClientJSON) toRaw() raw {
 		StreamsPerCred: j.VK.StreamsPerCred,
 		Transport:      j.TURN.Transport,
 		Mode:           j.Proxy.Mode,
+		Bond:           j.Proxy.Bond,

 		ObfProfile: j.Obf.Profile,
 		ObfKey:     j.Obf.Key,
@@ -26,6 +26,7 @@ type raw struct {
 	StreamsPerCred int
 	Transport      string
 	Mode           string
+	Bond           bool

 	ObfProfile string
 	ObfKey     string
@@ -55,6 +56,7 @@ func (r *raw) applyURI(u *uri.Config) {
 	override(&r.Provider, u.Provider)
 	override(&r.Transport, u.Transport)
 	override(&r.Mode, u.Mode)
+	r.Bond = r.Bond || u.Bond
 	if u.KCP != nil {
 		r.KCP = kcpmux.Profile{
 			NoDelay:    u.KCP.NoDelay,
@@ -142,6 +144,7 @@ func assemble(r raw) (*Client, error) {
 			Timing:  r.ObfTiming,
 		},
 		Proxy: ProxyOpts{
+			Bond:   r.Bond,
 			Mode:   ProxyMode(r.Mode),
 			Listen: r.Listen,
 			Peer:   r.Peer,
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"strings"

+	"github.com/samosvalishe/free-turn-proxy/internal/proxy/bond"
 	"github.com/samosvalishe/free-turn-proxy/internal/transport/kcpmux"
 	"github.com/samosvalishe/free-turn-proxy/internal/tunnel"
 )
@@ -30,6 +31,9 @@ func Validate(c *Client) error {
 	if err := validateProxyMode(c.Proxy.Mode); err != nil {
 		return err
 	}
+	if err := validateBond(c); err != nil {
+		return err
+	}
 	if err := validateTunnel(c.Tunnel, c.Proxy.Mode); err != nil {
 		return err
 	}
@@ -39,6 +43,22 @@ func Validate(c *Client) error {
 	return validateObfTiming(c.Obf)
 }

+var ErrBondConfig = errors.New("invalid bond configuration")
+
+func validateBond(c *Client) error {
+	if !c.Proxy.Bond {
+		return nil
+	}
+	groups := 1
+	if c.Provider.Name == ProviderVK {
+		groups = max(len(c.VK.Links), 1)
+	}
+	if c.Proxy.Mode != ProxyModeTCP || c.TURN.N > bond.MaxLanes/groups {
+		return fmt.Errorf("%w: -bond requires -mode tcp and at most %d sessions in total", ErrBondConfig, bond.MaxLanes)
+	}
+	return nil
+}
+
 func validateProvider(c *Client) error {
 	switch c.Provider.Name {
 	case ProviderDirect:
@@ -0,0 +1,300 @@
+package bond
+
+import (
+	"bytes"
+	"context"
+	"encoding/hex"
+	"errors"
+	"io"
+	"net"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/samosvalishe/free-turn-proxy/internal/stats"
+	"github.com/xtaci/smux"
+)
+
+func tcpPair(t *testing.T) (*net.TCPConn, net.Conn) {
+	t.Helper()
+	ln, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = ln.Close() }()
+	a, err := (&net.Dialer{}).DialContext(t.Context(), "tcp", ln.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	b, err := ln.Accept()
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = a.Close(); _ = b.Close() })
+	if err := a.SetDeadline(time.Now().Add(15 * time.Second)); err != nil {
+		t.Fatal(err)
+	}
+	tcp, ok := a.(*net.TCPConn)
+	if !ok {
+		t.Fatal("expected TCP")
+	}
+	return tcp, b
+}
+
+func TestSessionFailureAfterFINStopsCopy(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	a, b := net.Pipe()
+	clientSession, err := smux.Client(a, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = clientSession.Close() }()
+	serverSession, err := smux.Server(b, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = serverSession.Close() }()
+	lane, err := clientSession.OpenStream()
+	if err != nil {
+		t.Fatal(err)
+	}
+	remote, err := serverSession.AcceptStream()
+	if err != nil {
+		t.Fatal(err)
+	}
+	client, local := tcpPair(t)
+	done := make(chan error, 1)
+	go func() { done <- Copy(ctx, local, []net.Conn{lane}, nil) }()
+	if err = writeFrame(remote, frameFIN, 0, nil); err != nil {
+		t.Fatal(err)
+	}
+	if _, err = client.Read(make([]byte, 1)); !errors.Is(err, io.EOF) {
+		t.Fatalf("half-close: %v", err)
+	}
+	_ = serverSession.Close()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		cancel()
+		<-done
+		t.Fatal("session failure after FIN left local read blocked")
+	}
+}
+
+type delayedRead struct {
+	net.Conn
+	once sync.Once
+}
+
+type blockedWriter struct {
+	net.Conn
+	writing chan struct{}
+}
+
+func (w *blockedWriter) Write(b []byte) (int, error) {
+	close(w.writing)
+	return w.Conn.Write(b)
+}
+
+func TestLaneFailureUnblocksLocalWrite(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	a, b := net.Pipe()
+	defer func() { _ = b.Close() }()
+	local := &blockedWriter{Conn: a, writing: make(chan struct{})}
+	lane, remote := net.Pipe()
+	defer func() { _ = remote.Close() }()
+	done := make(chan error, 1)
+	go func() { done <- Copy(ctx, local, []net.Conn{lane}, nil) }()
+	if err := writeFrame(remote, frameData, 0, []byte("blocked")); err != nil {
+		t.Fatal(err)
+	}
+	<-local.writing
+	_ = remote.Close()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		cancel()
+		<-done
+		t.Fatal("lane failure left local write blocked")
+	}
+}
+
+func (d *delayedRead) Read(p []byte) (int, error) {
+	d.once.Do(func() { time.Sleep(30 * time.Millisecond) })
+	return d.Conn.Read(p)
+}
+
+func TestCopyReordersAndPreservesHalfClose(t *testing.T) {
+	for _, count := range []int{1, 3} {
+		t.Run(strconv.Itoa(count), func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(t.Context(), 15*time.Second)
+			defer cancel()
+			client, clientLocal := tcpPair(t)
+			backend, serverLocal := tcpPair(t)
+			left, right := make([]net.Conn, count), make([]net.Conn, count)
+			for i := range left {
+				left[i], right[i] = net.Pipe()
+			}
+			right[0] = &delayedRead{Conn: right[0]}
+			traffic := stats.New(true)
+			done := make(chan error, 2)
+			go func() { done <- Copy(ctx, clientLocal, left, traffic) }()
+			go func() { done <- Copy(ctx, serverLocal, right, nil) }()
+			request := bytes.Repeat([]byte("request contents"), 128*1024)
+			response := bytes.Repeat([]byte("response after EOF"), 128*1024)
+			writer := make(chan error, 1)
+			go func() {
+				_, err := client.Write(request)
+				writer <- errors.Join(err, client.CloseWrite())
+			}()
+			got, err := io.ReadAll(backend)
+			if err != nil || !bytes.Equal(got, request) {
+				t.Fatalf("request: len=%d err=%v", len(got), err)
+			}
+			if err = <-writer; err != nil {
+				t.Fatal(err)
+			}
+			go func() {
+				_, werr := backend.Write(response)
+				writer <- errors.Join(werr, backend.CloseWrite())
+			}()
+			got, err = io.ReadAll(client)
+			if err != nil || !bytes.Equal(got, response) {
+				t.Fatalf("response: len=%d err=%v", len(got), err)
+			}
+			if err := <-writer; err != nil {
+				t.Fatal(err)
+			}
+			for range 2 {
+				if err := <-done; err != nil {
+					t.Fatal(err)
+				}
+			}
+			tx, rx := traffic.Counters()
+			if tx != uint64(len(request)) || rx != uint64(len(response)) {
+				t.Fatalf("traffic=%d/%d", tx, rx)
+			}
+		})
+	}
+}
+
+func TestCopyLaneFailureAndCancel(t *testing.T) {
+	for _, failure := range []bool{false, true} {
+		t.Run(map[bool]string{false: "cancel", true: "lane failure"}[failure], func(t *testing.T) {
+			ctx, cancel := context.WithCancel(t.Context())
+			defer cancel()
+			local, peer := net.Pipe()
+			defer func() { _ = peer.Close() }()
+			lanes, remote := make([]net.Conn, 3), make([]net.Conn, 3)
+			for i := range lanes {
+				lanes[i], remote[i] = net.Pipe()
+				defer func() { _ = remote[i].Close() }()
+			}
+			done := make(chan error, 1)
+			go func() { done <- Copy(ctx, local, lanes, nil) }()
+			if failure {
+				_ = remote[1].Close()
+			} else {
+				cancel()
+			}
+			select {
+			case err := <-done:
+				if err == nil {
+					t.Fatal("expected failure")
+				}
+			case <-time.After(3 * time.Second):
+				t.Fatal("copy did not stop")
+			}
+			if _, err := peer.Read(make([]byte, 1)); !errors.Is(err, io.EOF) {
+				t.Fatalf("local not closed: %v", err)
+			}
+		})
+	}
+}
+
+func TestHelloWireAndValidation(t *testing.T) {
+	var b bytes.Buffer
+	h := Hello{ID: 0x0102030405060708, Index: 1, Count: 3}
+	if err := WriteHello(&b, h); err != nil {
+		t.Fatal(err)
+	}
+	if got := hex.EncodeToString(b.Bytes()); got != "564c423101010203040506070800010003" {
+		t.Fatal(got)
+	}
+	got, err := ReadHello(&b)
+	if err != nil || got != h {
+		t.Fatalf("hello=%+v, %v", got, err)
+	}
+	for _, invalid := range []Hello{{Count: 0}, {Count: MaxLanes + 1}, {Count: 2, Index: 2}} {
+		b.Reset()
+		if err := WriteHello(&b, invalid); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := ReadHello(&b); !errors.Is(err, ErrProtocol) {
+			t.Fatalf("accepted %+v: %v", invalid, err)
+		}
+	}
+}
+
+func TestReorderRejectsInvalidSequences(t *testing.T) {
+	for name, frames := range map[string][]frame{
+		"gap":             {{typ: frameFIN, seq: 1}},
+		"duplicate":       {{typ: frameData, seq: 1}, {typ: frameData, seq: 1}},
+		"overflow":        {{typ: frameData, seq: pendingCap + 1}},
+		"conflicting FIN": {{typ: frameData, seq: 1}, {typ: frameFIN, seq: 0}},
+	} {
+		t.Run(name, func(t *testing.T) {
+			ch := make(chan received, len(frames))
+			for _, f := range frames {
+				ch <- received{frame: f}
+			}
+			ctx, cancel := context.WithTimeout(t.Context(), time.Second)
+			defer cancel()
+			local, peer := net.Pipe()
+			defer func() { _ = local.Close(); _ = peer.Close() }()
+			if err := reorder(ctx, local, ch, 1, nil); !errors.Is(err, ErrProtocol) {
+				t.Fatalf("error=%v", err)
+			}
+		})
+	}
+}
+
+func TestServerSeparatesClientsAndRejectsDuplicateLane(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	s := &Server{}
+	laneA, remoteA := net.Pipe()
+	laneB, remoteB := net.Pipe()
+	defer func() { _ = remoteA.Close(); _ = remoteB.Close() }()
+	h := Hello{ID: 7, Index: 0, Count: 2}
+	key := connectionKey{client: "a", id: h.ID}
+	a, err := s.attach(ctx, key, h, laneA, "127.0.0.1:1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	b, err := s.attach(ctx, connectionKey{client: "b", id: h.ID}, h, laneB, "127.0.0.1:1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if a == b {
+		t.Fatal("different clients share group")
+	}
+	if _, err := s.attach(ctx, key, h, laneA, "127.0.0.1:1"); !errors.Is(err, ErrProtocol) {
+		t.Fatal("duplicate accepted")
+	}
+	cancel()
+	for _, g := range []*group{a, b} {
+		select {
+		case <-g.done:
+		case <-time.After(3 * time.Second):
+			t.Fatal("group did not stop")
+		}
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if len(s.groups) != 0 {
+		t.Fatal("groups leaked")
+	}
+}
@@ -0,0 +1,205 @@
+package bond
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"sync"
+
+	"github.com/samosvalishe/free-turn-proxy/internal/stats"
+)
+
+// Copy владеет local и lanes; ошибка любого направления закрывает всё соединение.
+func Copy(ctx context.Context, local net.Conn, lanes []net.Conn, traffic *stats.Stats) error {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	stop := context.AfterFunc(ctx, func() {
+		_ = local.Close()
+		for _, lane := range lanes {
+			_ = lane.Close()
+		}
+	})
+	defer stop()
+	defer func() {
+		_ = local.Close()
+		for _, lane := range lanes {
+			_ = lane.Close()
+		}
+	}()
+	if len(lanes) == 0 || len(lanes) > MaxLanes {
+		return fmt.Errorf("%w: lane count", ErrProtocol)
+	}
+	results := make(chan error, 2)
+	go func() { results <- send(local, lanes, traffic) }()
+	var receiver sync.WaitGroup
+	receiver.Go(func() {
+		err := receive(ctx, local, lanes, traffic, cancel)
+		results <- err
+		if err == nil {
+			watchClosure(lanes, cancel)
+		}
+	})
+	defer func() { cancel(); receiver.Wait() }()
+	err := <-results
+	if err != nil {
+		cancel()
+	}
+	return errors.Join(err, <-results)
+}
+
+// После FIN чтение EOF/ошибки всё ещё должно прерывать заблокированный local.Read.
+func watchClosure(lanes []net.Conn, cancel context.CancelFunc) {
+	var wg sync.WaitGroup
+	for _, lane := range lanes {
+		wg.Go(func() {
+			var b [1]byte
+			_, _ = lane.Read(b[:])
+			cancel()
+		})
+	}
+	wg.Wait()
+}
+
+func send(local net.Conn, lanes []net.Conn, traffic *stats.Stats) error {
+	buf := make([]byte, maxChunk)
+	var seq uint64
+	for {
+		n, err := local.Read(buf)
+		if n > 0 {
+			if werr := writeFrame(lanes[seq%uint64(len(lanes))], frameData, seq, buf[:n]); werr != nil {
+				return werr
+			}
+			seq++
+			if traffic != nil {
+				traffic.AddTx(n)
+			}
+		}
+		if errors.Is(err, io.EOF) {
+			for _, lane := range lanes {
+				if werr := writeFrame(lane, frameFIN, seq, nil); werr != nil {
+					return werr
+				}
+			}
+			return nil
+		}
+		if err != nil {
+			return fmt.Errorf("bond local read: %w", err)
+		}
+	}
+}
+
+type received struct {
+	frame
+	err error
+}
+
+func receive(ctx context.Context, local net.Conn, lanes []net.Conn, traffic *stats.Stats, abort context.CancelFunc) error {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	frames := make(chan received, len(lanes))
+	var readers sync.WaitGroup
+	for _, lane := range lanes {
+		readers.Go(func() { readLane(ctx, lane, frames, abort) })
+	}
+	err := reorder(ctx, local, frames, len(lanes), traffic)
+	cancel()
+	// При ошибке чтения другие lanes могут ждать данных бесконечно.
+	if err != nil {
+		for _, lane := range lanes {
+			_ = lane.Close()
+		}
+	}
+	readers.Wait()
+	return err
+}
+
+func readLane(ctx context.Context, lane net.Conn, frames chan<- received, abort context.CancelFunc) {
+	for {
+		f, err := readFrame(lane)
+		if err != nil {
+			// Получатель может стоять в local.Write, не читая frames.
+			abort()
+		}
+		select {
+		case frames <- received{frame: f, err: err}:
+		case <-ctx.Done():
+			return
+		}
+		if err != nil || f.typ == frameFIN {
+			return
+		}
+	}
+}
+
+func reorder(ctx context.Context, local net.Conn, frames <-chan received, count int, traffic *stats.Stats) error {
+	pending := make(map[uint64][]byte)
+	var expect, end uint64
+	var fins int
+	for {
+		select {
+		case <-ctx.Done():
+			return fmt.Errorf("bond receive: %w", ctx.Err())
+		case f := <-frames:
+			if f.err != nil {
+				return f.err
+			}
+			if f.typ == frameFIN {
+				if f.seq < expect || (fins > 0 && end != f.seq) {
+					return fmt.Errorf("%w: conflicting FIN", ErrProtocol)
+				}
+				end = f.seq
+				fins++
+			} else {
+				if err := queueFrame(pending, f.frame, expect, end, fins); err != nil {
+					return err
+				}
+			}
+			if err := drain(local, pending, &expect, traffic); err != nil {
+				return err
+			}
+			if fins == count {
+				if expect != end || len(pending) != 0 {
+					return fmt.Errorf("%w: missing data before FIN", ErrProtocol)
+				}
+				return closeWrite(local)
+			}
+		}
+	}
+}
+
+func closeWrite(local net.Conn) error {
+	if cw, ok := local.(interface{ CloseWrite() error }); ok {
+		if err := cw.CloseWrite(); err != nil {
+			return fmt.Errorf("bond half-close: %w", err)
+		}
+	}
+	return nil
+}
+
+func queueFrame(pending map[uint64][]byte, f frame, expect, end uint64, fins int) error {
+	_, duplicate := pending[f.seq]
+	if f.seq < expect || duplicate || f.seq-expect > pendingCap || (fins > 0 && f.seq >= end) {
+		return fmt.Errorf("%w: data sequence", ErrProtocol)
+	}
+	pending[f.seq] = f.data
+	return nil
+}
+
+func drain(local net.Conn, pending map[uint64][]byte, expect *uint64, traffic *stats.Stats) error {
+	for {
+		data, ok := pending[*expect]
+		if !ok {
+			return nil
+		}
+		if err := writeFull(local, data); err != nil {
+			return err
+		}
+		if traffic != nil {
+			traffic.AddRx(len(data))
+		}
+		delete(pending, *expect)
+		*expect++
+	}
+}
@@ -0,0 +1,99 @@
+// Package bond объединяет smux-потоки разных сессий в одно TCP-соединение.
+package bond
+
+import (
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"io"
+	"time"
+)
+
+const (
+	MaxLanes          = 256
+	SetupTimeout      = 10 * time.Second
+	maxChunk          = 16 * 1024
+	pendingCap        = 256
+	frameData    byte = 1
+	frameFIN     byte = 2
+)
+
+var ErrProtocol = errors.New("bond: invalid protocol")
+
+// Сохранён формат VLB1: magic(4), version(1), connection(8), index(2), count(2); big endian.
+type Hello struct {
+	ID    uint64
+	Index uint16
+	Count uint16
+}
+
+func WriteHello(w io.Writer, h Hello) error {
+	var b [17]byte
+	copy(b[:], "VLB1")
+	b[4] = 1
+	binary.BigEndian.PutUint64(b[5:13], h.ID)
+	binary.BigEndian.PutUint16(b[13:15], h.Index)
+	binary.BigEndian.PutUint16(b[15:17], h.Count)
+	return writeFull(w, b[:])
+}
+
+func ReadHello(r io.Reader) (Hello, error) {
+	var b [17]byte
+	if _, err := io.ReadFull(r, b[:]); err != nil {
+		return Hello{}, fmt.Errorf("bond hello: %w", err)
+	}
+	h := Hello{ID: binary.BigEndian.Uint64(b[5:13]), Index: binary.BigEndian.Uint16(b[13:15]), Count: binary.BigEndian.Uint16(b[15:17])}
+	if string(b[:4]) != "VLB1" || b[4] != 1 || h.Count == 0 || h.Count > MaxLanes || h.Index >= h.Count {
+		return Hello{}, fmt.Errorf("%w: hello", ErrProtocol)
+	}
+	return h, nil
+}
+
+// Кадр VLB1: type(1), sequence(8), size(4), payload(size); FIN обозначает следующий sequence.
+type frame struct {
+	typ  byte
+	seq  uint64
+	data []byte
+}
+
+func writeFrame(w io.Writer, typ byte, seq uint64, data []byte) error {
+	var b [13]byte
+	b[0] = typ
+	binary.BigEndian.PutUint64(b[1:9], seq)
+	binary.BigEndian.PutUint32(b[9:13], uint32(len(data))) //nolint:gosec // payload ограничен maxChunk
+	if err := writeFull(w, b[:]); err != nil {
+		return err
+	}
+	return writeFull(w, data)
+}
+
+func readFrame(r io.Reader) (frame, error) {
+	var b [13]byte
+	if _, err := io.ReadFull(r, b[:]); err != nil {
+		return frame{}, fmt.Errorf("bond frame: %w", err)
+	}
+	size := binary.BigEndian.Uint32(b[9:13])
+	f := frame{typ: b[0], seq: binary.BigEndian.Uint64(b[1:9])}
+	if (f.typ != frameData && f.typ != frameFIN) || size > maxChunk || (f.typ == frameFIN && size != 0) {
+		return frame{}, fmt.Errorf("%w: frame", ErrProtocol)
+	}
+	f.data = make([]byte, size)
+	if _, err := io.ReadFull(r, f.data); err != nil {
+		return frame{}, fmt.Errorf("bond payload: %w", err)
+	}
+	return f, nil
+}
+
+func writeFull(w io.Writer, b []byte) error {
+	if len(b) == 0 {
+		return nil
+	}
+	n, err := w.Write(b)
+	if err != nil {
+		return fmt.Errorf("bond write: %w", err)
+	}
+	if n != len(b) {
+		return fmt.Errorf("bond write: %w", io.ErrShortWrite)
+	}
+	return nil
+}
@@ -0,0 +1,112 @@
+package bond
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+)
+
+type connectionKey struct {
+	client string
+	id     uint64
+}
+
+type group struct {
+	lanes    []net.Conn
+	attached int
+	ready    chan struct{}
+	done     chan struct{}
+	cancel   context.CancelFunc
+	err      error
+}
+
+// Server общий для всех DTLS-сессий одного listener; Client ID изолирует группы клиентов.
+type Server struct {
+	mu     sync.Mutex
+	groups map[connectionKey]*group
+}
+
+func (s *Server) Handle(ctx context.Context, lane net.Conn, client, backend string) error {
+	defer func() { _ = lane.Close() }()
+	stop := context.AfterFunc(ctx, func() { _ = lane.Close() })
+	defer stop()
+	if err := lane.SetReadDeadline(time.Now().Add(SetupTimeout)); err != nil {
+		return fmt.Errorf("bond hello deadline: %w", err)
+	}
+	h, err := ReadHello(lane)
+	if err != nil {
+		return err
+	}
+	if err = lane.SetReadDeadline(time.Time{}); err != nil {
+		return fmt.Errorf("bond hello deadline: %w", err)
+	}
+	g, err := s.attach(ctx, connectionKey{client: client, id: h.ID}, h, lane, backend)
+	if err != nil {
+		return err
+	}
+	select {
+	case <-ctx.Done():
+		g.cancel()
+		<-g.done
+	case <-g.done:
+	}
+	return g.err
+}
+
+func (s *Server) attach(ctx context.Context, key connectionKey, h Hello, lane net.Conn, backend string) (*group, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.groups == nil {
+		s.groups = make(map[connectionKey]*group)
+	}
+	g := s.groups[key]
+	if g == nil {
+		groupCtx, cancel := context.WithCancel(ctx)
+		g = &group{lanes: make([]net.Conn, h.Count), ready: make(chan struct{}), done: make(chan struct{}), cancel: cancel}
+		s.groups[key] = g
+		go s.run(groupCtx, key, g, backend)
+	}
+	if len(g.lanes) != int(h.Count) || g.lanes[h.Index] != nil {
+		return nil, fmt.Errorf("%w: duplicate lane or count mismatch", ErrProtocol)
+	}
+	g.lanes[h.Index] = lane
+	g.attached++
+	if g.attached == len(g.lanes) {
+		close(g.ready)
+	}
+	return g, nil
+}
+
+func (s *Server) run(ctx context.Context, key connectionKey, g *group, backend string) {
+	g.err = serveGroup(ctx, g, backend)
+	g.cancel()
+	s.mu.Lock()
+	delete(s.groups, key)
+	lanes := append([]net.Conn(nil), g.lanes...)
+	s.mu.Unlock()
+	for _, lane := range lanes {
+		if lane != nil {
+			_ = lane.Close()
+		}
+	}
+	close(g.done)
+}
+
+func serveGroup(ctx context.Context, g *group, backend string) error {
+	timer := time.NewTimer(SetupTimeout)
+	defer timer.Stop()
+	select {
+	case <-g.ready:
+	case <-ctx.Done():
+		return fmt.Errorf("bond attach: %w", ctx.Err())
+	case <-timer.C:
+		return fmt.Errorf("bond attach: %w", context.DeadlineExceeded)
+	}
+	conn, err := (&net.Dialer{Timeout: SetupTimeout}).DialContext(ctx, "tcp", backend)
+	if err != nil {
+		return fmt.Errorf("bond backend: %w", err)
+	}
+	return Copy(ctx, conn, g.lanes, nil)
+}
@@ -0,0 +1,79 @@
+package tcprelay
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/binary"
+	"net"
+	"time"
+
+	"github.com/samosvalishe/free-turn-proxy/internal/logx"
+	"github.com/samosvalishe/free-turn-proxy/internal/proxy/bond"
+	"github.com/samosvalishe/free-turn-proxy/internal/stats"
+)
+
+func proxyBond(ctx context.Context, log logx.Logger, conn net.Conn, pool *sessionPool) {
+	defer func() { _ = conn.Close() }()
+	lanes, traffic := openBondLanes(pool)
+	defer func() {
+		for _, lane := range lanes {
+			_ = lane.Close()
+		}
+	}()
+	if len(lanes) == 0 {
+		return
+	}
+	stop := context.AfterFunc(ctx, func() {
+		for _, lane := range lanes {
+			_ = lane.Close()
+		}
+	})
+	defer stop()
+	var id [8]byte
+	_, _ = rand.Read(id[:])
+	deadline := time.Now().Add(bond.SetupTimeout)
+	for i, lane := range lanes {
+		h := bond.Hello{ID: binary.BigEndian.Uint64(id[:]), Index: uint16(i), Count: uint16(len(lanes))} //nolint:gosec // len(lanes) <= MaxLanes
+		if err := helloBondLane(lane, h, deadline); err != nil {
+			log.Debugf("TCP bond setup: %v", err)
+			return
+		}
+	}
+	log.Debugf("TCP bond connected: lanes=%d", len(lanes))
+	if err := bond.Copy(ctx, conn, lanes, traffic); err != nil && ctx.Err() == nil {
+		log.Debugf("TCP bond closed: %v", err)
+	}
+}
+
+func helloBondLane(lane net.Conn, h bond.Hello, deadline time.Time) error {
+	if err := lane.SetWriteDeadline(deadline); err != nil {
+		return err
+	}
+	if err := bond.WriteHello(lane, h); err != nil {
+		return err
+	}
+	return lane.SetWriteDeadline(time.Time{})
+}
+
+func openBondLanes(pool *sessionPool) ([]net.Conn, *stats.Stats) {
+	pool.mu.RLock()
+	candidates := append([]*pooledSession(nil), pool.sessions...)
+	pool.mu.RUnlock()
+	lanes := make([]net.Conn, 0, len(candidates))
+	var traffic *stats.Stats
+	for _, ps := range candidates {
+		if len(lanes) == bond.MaxLanes {
+			break
+		}
+		if ps.sess.IsClosed() {
+			continue
+		}
+		stream, err := ps.sess.OpenStream()
+		if err != nil {
+			continue
+		}
+		lanes = append(lanes, stream)
+		traffic = ps.traffic
+	}
+	return lanes, traffic
+}
@@ -0,0 +1,125 @@
+package tcprelay
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/samosvalishe/free-turn-proxy/internal/logx"
+	"github.com/samosvalishe/free-turn-proxy/internal/netconn"
+	"github.com/samosvalishe/free-turn-proxy/internal/proxy/bond"
+	"github.com/samosvalishe/free-turn-proxy/internal/proxy/tcpserver"
+	"github.com/samosvalishe/free-turn-proxy/internal/transport/kcpmux"
+	"github.com/xtaci/smux"
+)
+
+func pairedBondSession(t *testing.T, ctx context.Context, server *bond.Server, backend string) *smux.Session {
+	t.Helper()
+	a, b := netconn.DatagramPipe(2048, 1024)
+	t.Cleanup(func() { _ = a.Close(); _ = b.Close() })
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		tcpserver.HandleBond(ctx, logx.Nop(), b, backend, kcpmux.DefaultProfile(), server, "test-client")
+	}()
+	kcp, err := kcpmux.Dial(a, kcpmux.DefaultProfile())
+	if err != nil {
+		t.Fatal(err)
+	}
+	sess, err := smux.Client(kcp, kcpmux.SmuxConfig())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		_ = sess.Close()
+		select {
+		case <-done:
+		case <-time.After(5 * time.Second):
+			t.Error("bond server leaked")
+		}
+	})
+	return sess
+}
+
+func TestBondAcrossSessionsAndAfterSessionFailure(t *testing.T) {
+	ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second)
+	defer cancel()
+	pool := newSessionPool(nil)
+	pool.bond = true
+	server := &bond.Server{}
+	backend := echoBackend(t)
+	for i := range 3 {
+		pool.Add(i+1, pairedBondSession(t, ctx, server, backend), nil)
+	}
+	ln, err := (&net.ListenConfig{}).Listen(ctx, "tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = ln.Close() }()
+	stop := context.AfterFunc(ctx, func() { _ = ln.Close() })
+	defer stop()
+	loopDone := make(chan struct{})
+	go func() { defer close(loopDone); acceptLoop(ctx, &Deps{Log: logx.Nop()}, ln, pool) }()
+	conn := dialBondTest(t, ctx, ln.Addr().String())
+	payload := bytes.Repeat([]byte("bond payload"), 128*1024)
+	checkBondEcho(t, conn, payload)
+	for _, ps := range pool.sessions {
+		if ps.sess.NumStreams() != 1 {
+			t.Fatalf("session %d: expected one bond lane", ps.id)
+		}
+	}
+	_ = pool.sessions[1].sess.Close()
+	if _, err := conn.Read(make([]byte, 1)); err == nil {
+		t.Fatal("broken bond still open")
+	}
+	_ = conn.Close()
+	// Следующее соединение использует оставшиеся сессии без перезапуска listener.
+	conn = dialBondTest(t, ctx, ln.Addr().String())
+	checkBondEcho(t, conn, payload)
+	_ = conn.Close()
+	cancel()
+	select {
+	case <-loopDone:
+	case <-time.After(5 * time.Second):
+		t.Fatal("accept loop leaked")
+	}
+}
+
+func dialBondTest(t *testing.T, ctx context.Context, addr string) net.Conn {
+	t.Helper()
+	conn, err := (&net.Dialer{}).DialContext(ctx, "tcp", addr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = conn.Close() })
+	if err := conn.SetDeadline(time.Now().Add(10 * time.Second)); err != nil {
+		t.Fatal(err)
+	}
+	return conn
+}
+
+func checkBondEcho(t *testing.T, conn net.Conn, payload []byte) {
+	t.Helper()
+	done := make(chan error, 1)
+	go func() {
+		n, err := conn.Write(payload)
+		if err == nil && n != len(payload) {
+			err = fmt.Errorf("short write %d", n)
+		}
+		done <- err
+	}()
+	got := make([]byte, len(payload))
+	if _, err := io.ReadFull(conn, got); err != nil {
+		t.Fatal(err)
+	}
+	if err := <-done; err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(got, payload) {
+		t.Fatal("corrupted echo")
+	}
+}
@@ -21,6 +21,7 @@ type pooledSession struct {

 // sessionPool - конкурентно-безопасный round-robin пул живых сессий.
 type sessionPool struct {
+	bond     bool
 	mu       sync.RWMutex
 	sessions []*pooledSession
 	counter  atomic.Uint64
@@ -45,6 +45,7 @@ type DialFunc = udprelay.DialFunc
 type AuthHandler = udprelay.AuthHandler

 type Params struct {
+	Bond         bool
 	Dial         DialFunc
 	Profile      string
 	ObfKey       []byte
@@ -86,6 +87,7 @@ func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, lis
 	}
 	log := deps.log()
 	pool := newSessionPool(deps.ConnectedStreams)
+	pool.bond = params.Bond

 	runCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
@@ -98,7 +100,7 @@ func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, lis
 	}
 	stopCloser := context.AfterFunc(runCtx, func() { _ = listener.Close() })
 	defer stopCloser()
-	log.Infof("TCP mode: listening on %s (round-robin across %d sessions)", listenAddr, numSessions)
+	log.Infof("TCP mode: listening on %s (sessions=%d bond=%t)", listenAddr, numSessions, params.Bond)

 	fatalCh := make(chan error, 1)
 	fatal := func(err error) {
@@ -201,6 +203,12 @@ func acceptLoop(ctx context.Context, deps *Deps, listener net.Listener, pool *se
 			continue
 		}
 		backoff = 0
+		if pool.bond {
+			wg.Go(func() {
+				_ = safego.Run(log, func() { proxyBond(ctx, log, conn, pool) })
+			})
+			continue
+		}

 		ps := pool.Pick()
 		if ps == nil {
@@ -387,7 +395,11 @@ func createSession(ctx context.Context, deps *Deps, params *Params, peer *net.UD
 	closers = append(closers, func() { _ = dtlsConn.Close() })

 	// Wire-контракт: Client ID первой app-record, до KCP.
-	if err = clientsdb.WriteClientID(ctx, dtlsConn, params.ClientID, clientsdb.ModeTCP); err != nil {
+	mode := clientsdb.ModeTCP
+	if params.Bond {
+		mode = clientsdb.ModeTCPBond
+	}
+	if err = clientsdb.WriteClientID(ctx, dtlsConn, params.ClientID, mode); err != nil {
 		cleanup()
 		return nil, fmt.Errorf("send client ID: %w", err)
 	}
@@ -11,6 +11,7 @@ import (

 	"github.com/samosvalishe/free-turn-proxy/internal/logx"
 	"github.com/samosvalishe/free-turn-proxy/internal/netconn"
+	"github.com/samosvalishe/free-turn-proxy/internal/proxy/bond"
 	"github.com/samosvalishe/free-turn-proxy/internal/safego"
 	"github.com/samosvalishe/free-turn-proxy/internal/transport/kcpmux"
 	"github.com/xtaci/smux"
@@ -20,6 +21,22 @@ const backendDialTimeout = 10 * time.Second

 // Handle блокирует вызывающую горутину до закрытия сессии клиентом или ctx.
 func Handle(ctx context.Context, logger logx.Logger, dtlsConn net.Conn, connectAddr string, profile kcpmux.Profile) {
+	handle(ctx, logger, dtlsConn, profile, func(ctx context.Context, stream *smux.Stream) {
+		handleStream(ctx, logger, stream, connectAddr)
+	})
+}
+
+func HandleBond(ctx context.Context, logger logx.Logger, dtlsConn net.Conn, connectAddr string, profile kcpmux.Profile, server *bond.Server, clientID string) {
+	handle(ctx, logger, dtlsConn, profile, func(ctx context.Context, stream *smux.Stream) {
+		if err := server.Handle(ctx, stream, clientID, connectAddr); err != nil && ctx.Err() == nil {
+			logger.Debugf("tcpserver bond: %v", err)
+		}
+	})
+}
+
+func handle(ctx context.Context, logger logx.Logger, dtlsConn net.Conn, profile kcpmux.Profile, streamHandler func(context.Context, *smux.Stream)) {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
 	kcpSess, err := kcpmux.Accept(dtlsConn, profile)
 	if err != nil {
 		logger.Errorf("tcpserver: %s", err)
@@ -57,9 +74,10 @@ func Handle(ctx context.Context, logger logx.Logger, dtlsConn net.Conn, connectA
 			break
 		}
 		wg.Go(func() {
-			_ = safego.Run(logger, func() { handleStream(ctx, logger, stream, connectAddr) })
+			_ = safego.Run(logger, func() { streamHandler(ctx, stream) })
 		})
 	}
+	cancel()
 	wg.Wait()
 }

@@ -483,6 +483,7 @@ func (s *Session) relayTCP(ctx context.Context, auth udprelay.AuthHandler, dial
 		Recycle:          s.recycleCh,
 	}
 	params := &tcprelay.Params{
+		Bond:         s.cfg.Proxy.Bond,
 		Dial:         dial,
 		Profile:      string(s.cfg.Obf.Profile),
 		ObfKey:       s.cfg.Obf.Key,
@@ -18,6 +18,7 @@ type Config struct {
 	Peer           string
 	Transport      string
 	Mode           string
+	Bond           bool
 	ObfProfile     string
 	ObfKey         string
 	ObfTimingMs    int
@@ -41,6 +42,7 @@ type wire struct {
 	Peer           string `json:"peer"`
 	Transport      string `json:"transport,omitempty"`
 	Mode           string `json:"mode,omitempty"`
+	Bond           bool   `json:"bond,omitempty"`
 	Obf            string `json:"obf,omitempty"`
 	Key            string `json:"key,omitempty"`
 	TimingMs       int    `json:"timing,omitempty"`
@@ -92,6 +94,7 @@ func Parse(s string) (*Config, error) {
 		Peer:           w.Peer,
 		Transport:      w.Transport,
 		Mode:           w.Mode,
+		Bond:           w.Bond,
 		ObfProfile:     w.Obf,
 		ObfKey:         w.Key,
 		ObfTimingMs:    w.TimingMs,
@@ -117,6 +120,7 @@ func (c *Config) String() string {
 		Peer:           c.Peer,
 		Transport:      c.Transport,
 		Mode:           c.Mode,
+		Bond:           c.Bond,
 		N:              c.N,
 		StreamsPerCred: c.StreamsPerCred,
 		ClientID:       c.ClientID,
@@ -69,6 +69,8 @@ func Version() string

 `proxy.mode`: `udp` (по умолчанию, UDP-релей для WireGuard) или `tcp` (TCP-форвардер для Xray/sing-box).

+`proxy.bond: true` включает объединение сессий для каждого TCP-соединения; допускается только с `proxy.mode: "tcp"`, по умолчанию выключено. Нужны обновлённые сервер и AAR. При отказе одной участвующей сессии соединение закрывается; новое использует оставшиеся сессии. Публичные методы gomobile не меняются.
+
 В `tcp` ядро слушает `proxy.listen` как TCP-порт, и VLESS-клиент ходит туда своим outbound. Своего tun при этом нет:

 *   запускать через `Start(configJSON)`, не через `StartTunnel`; `VpnService.prepare`/`establish` звать не нужно - иначе tun отбирается у чужого VPN;