Add RRPIT - Rapid Reliable Packet Interactive Transport (#3658)

* Add rrpit(unreleased) transport

* rrpit(unreleased) transport: boost reconstruction for the front lane

* rrpit(unreleased) transport: avoid blocking delivery of packets with smux packet reordering

* rrpit(unreleased) transport: add dual channel with differential prioritization

* rrpit(unreleased) transport: add remoteMaxDataShardsPerLane

* rrpit(unreleased) transport: add remoteMaxDataShardsPerLane(fix)

* rrpit(unreleased) transport: add auto-reconnect for transport connection

* rrpit(unreleased) transport: add auto-reconnect for transport connection

* rrpit(unreleased) transport: update tests

* rrpit(unreleased) transport: limit frame size for framed packet write closer

* rrpit(unreleased) transport: remove packet length wrapper

* rrpit(unreleased) transport: update go mod

* rrpit(unreleased) transport: update dtls error propagation

* rrpit(unreleased) transport: update broken tests

* rrpit(unreleased) transport: fix deadlock when closing persistent channel

* rrpit(unreleased) transport: add server disconnection detection for persistent connection

* rrpit(unreleased) transport: fix broken lints

* rrpit(unreleased) transport: fix broken lints

* rrpit(unreleased) transport: fix broken lints

* rrpit(unreleased) transport: fix broken lints

* rrpit(unreleased) transport: fix broken lints

Author: xiaokangwang

app/dispatcher/default.go

@@ -313,6 +313,19 @@ func (d *DefaultDispatcher) routedDispatch(ctx context.Context, link *transport.
 		}
 	} else if d.router != nil {
 		if route, err := d.router.PickRoute(routing_session.AsRoutingContext(ctx)); err == nil {
+			if routeWithAttributes, ok := route.(interface{ GetSessionAttributes() map[string]string }); ok {
+				attrs := routeWithAttributes.GetSessionAttributes()
+				if len(attrs) > 0 {
+					content := session.ContentFromContext(ctx)
+					if content == nil {
+						content = new(session.Content)
+						ctx = session.ContextWithContent(ctx, content)
+					}
+					for key, value := range attrs {
+						content.SetAttribute(key, value)
+					}
+				}
+			}
 			tag := route.GetOutboundTag()
 			if h := d.ohm.GetHandler(tag); h != nil {
 				newError("taking detour [", tag, "] for [", destination, "]").WriteToLog(session.ExportIDToError(ctx))

app/router/config.go

@@ -18,9 +18,10 @@ import (
 )
 
 type Rule struct {
-	Tag       string
-	Balancer  *Balancer
-	Condition Condition
+	Tag           string
+	Balancer      *Balancer
+	Condition     Condition
+	SetAttributes []*SetAttribute
 }
 
 func (r *Rule) GetTag() (string, error) {

app/router/config.pb.go

@@ -12,6 +12,10 @@ import "common/net/network.proto";
 import "common/protoext/extensions.proto";
 import "app/router/routercommon/common.proto";
 
+message SetAttribute {
+  string key = 1;
+  string value = 2;
+}
 
 message RoutingRule {
   oneof target_tag {
@@ -66,6 +70,8 @@ message RoutingRule {
 
   string domain_matcher = 17;
 
+  repeated SetAttribute set_attribute = 18;
+
   // geo_domain instruct simplified config loader to load geo domain rule and fill in domain field.
   repeated v2ray.core.app.router.routercommon.GeoSite geo_domain = 68001;
 }
@@ -178,6 +184,8 @@ message SimplifiedRoutingRule {
 
   string domain_matcher = 17;
 
+  repeated SetAttribute set_attribute = 18;
+
   // geo_domain instruct simplified config loader to load geo domain rule and fill in domain field.
   repeated v2ray.core.app.router.routercommon.GeoSite geo_domain = 68001;
 }
@@ -189,4 +197,4 @@ message SimplifiedConfig {
   DomainStrategy domain_strategy = 1;
   repeated SimplifiedRoutingRule rule = 2;
   repeated BalancingRule balancing_rule = 3;
-}
+}

app/router/config.proto

@@ -29,6 +29,7 @@ type Route struct {
 	routing.Context
 	outboundGroupTags []string
 	outboundTag       string
+	sessionAttributes map[string]string
 }
 
 // Init initializes the Router.
@@ -53,8 +54,9 @@ func (r *Router) Init(ctx context.Context, config *Config, d dns.Client, ohm out
 			return err
 		}
 		rr := &Rule{
-			Condition: cond,
-			Tag:       rule.GetTag(),
+			Condition:     cond,
+			Tag:           rule.GetTag(),
+			SetAttributes: rule.GetSetAttribute(),
 		}
 		btag := rule.GetBalancingTag()
 		if len(btag) > 0 {
@@ -80,7 +82,20 @@ func (r *Router) PickRoute(ctx routing.Context) (routing.Route, error) {
 	if err != nil {
 		return nil, err
 	}
-	return &Route{Context: ctx, outboundTag: tag}, nil
+	route := &Route{Context: ctx, outboundTag: tag}
+	if len(rule.SetAttributes) > 0 {
+		route.sessionAttributes = make(map[string]string, len(rule.SetAttributes))
+		for _, attr := range rule.SetAttributes {
+			if attr == nil || attr.Key == "" {
+				continue
+			}
+			route.sessionAttributes[attr.Key] = attr.Value
+		}
+		if len(route.sessionAttributes) == 0 {
+			route.sessionAttributes = nil
+		}
+	}
+	return route, nil
 }
 
 func (r *Router) pickRouteInternal(ctx routing.Context) (*Rule, routing.Context, error) {
@@ -140,6 +155,17 @@ func (r *Route) GetOutboundTag() string {
 	return r.outboundTag
 }
 
+func (r *Route) GetSessionAttributes() map[string]string {
+	if r == nil || len(r.sessionAttributes) == 0 {
+		return nil
+	}
+	out := make(map[string]string, len(r.sessionAttributes))
+	for key, value := range r.sessionAttributes {
+		out[key] = value
+	}
+	return out
+}
+
 func init() {
 	common.Must(common.RegisterConfig((*Config)(nil), func(ctx context.Context, config interface{}) (interface{}, error) {
 		r := new(Router)
@@ -242,6 +268,7 @@ func init() {
 			rule.Networks = v.Networks.GetNetwork()
 			rule.Protocol = v.Protocol
 			rule.Attributes = v.Attributes
+			rule.SetAttribute = append(rule.SetAttribute, v.SetAttribute...)
 			rule.UserEmail = v.UserEmail
 			rule.InboundTag = v.InboundTag
 			rule.DomainMatcher = v.DomainMatcher

app/router/router.go

@@ -54,6 +54,50 @@ func TestSimpleRouter(t *testing.T) {
 	}
 }
 
+func TestRouteSetAttribute(t *testing.T) {
+	config := &Config{
+		Rule: []*RoutingRule{
+			{
+				TargetTag: &RoutingRule_Tag{
+					Tag: "test",
+				},
+				Networks: []net.Network{net.Network_TCP},
+				SetAttribute: []*SetAttribute{
+					{
+						Key:   "rrpitSessionClass",
+						Value: "background",
+					},
+				},
+			},
+		},
+	}
+
+	mockCtl := gomock.NewController(t)
+	defer mockCtl.Finish()
+
+	mockDNS := mocks.NewDNSClient(mockCtl)
+	mockOhm := mocks.NewOutboundManager(mockCtl)
+	mockHs := mocks.NewOutboundHandlerSelector(mockCtl)
+
+	r := new(Router)
+	common.Must(r.Init(context.TODO(), config, mockDNS, &mockOutboundManager{
+		Manager:         mockOhm,
+		HandlerSelector: mockHs,
+	}, nil))
+
+	ctx := session.ContextWithOutbound(context.Background(), &session.Outbound{Target: net.TCPDestination(net.DomainAddress("v2fly.org"), 80)})
+	route, err := r.PickRoute(routing_session.AsRoutingContext(ctx))
+	common.Must(err)
+	attrRoute, ok := route.(interface{ GetSessionAttributes() map[string]string })
+	if !ok {
+		t.Fatal("route does not expose session attributes")
+	}
+	attrs := attrRoute.GetSessionAttributes()
+	if attrs["rrpitSessionClass"] != "background" {
+		t.Fatalf("unexpected session attribute map: %#v", attrs)
+	}
+}
+
 func TestSimpleBalancer(t *testing.T) {
 	config := &Config{
 		Rule: []*RoutingRule{

app/router/router_test.go

@@ -1,4 +1,4 @@
-package abstactOutbount
+package abstactOutbound
 
 import "github.com/v2fly/v2ray-core/v5/common/errors"
 

…/net/abstactOutbount/errors.generated.go …/net/abstactOutbound/errors.generated.gocommon/net/abstactOutbount/errors.generated.go renamed to common/net/abstactOutbound/errors.generated.go common/net/abstactOutbount/errors.generated.go renamed to common/net/abstactOutbound/errors.generated.go

@@ -1,4 +1,4 @@
-package abstactOutbount
+package abstactOutbound
 
 //go:generate go run github.com/v2fly/v2ray-core/v5/common/errors/errorgen
 

common/net/abstactOutbount/outbound.go common/net/abstactOutbound/outbound.gocommon/net/abstactOutbount/outbound.go renamed to common/net/abstactOutbound/outbound.go

@@ -18,10 +18,11 @@ require (
 	github.com/gorilla/websocket v1.5.3
 	github.com/improbable-eng/grpc-web v0.15.0
 	github.com/jhump/protoreflect v1.18.0
+	github.com/lunixbochs/struc v0.0.0-20200707160740-784aaebc1d40
 	github.com/miekg/dns v1.1.72
 	github.com/mustafaturan/bus v1.0.2
 	github.com/pelletier/go-toml v1.9.5
-	github.com/pion/dtls/v2 v2.2.12
+	github.com/pion/dtls/v3 v3.0.11
 	github.com/pion/stun/v3 v3.1.1
 	github.com/pion/transport/v2 v2.2.10
 	github.com/pires/go-proxyproto v0.11.0
@@ -36,6 +37,8 @@ require (
 	github.com/v2fly/struc v0.0.0-20241227015403-8e8fa1badfd6
 	github.com/vincent-petithory/dataurl v1.0.0
 	github.com/xiaokangwang/VLite v0.0.0-20220418190619-cff95160a432
+	github.com/xssnick/raptorq v1.3.0
+	github.com/xtaci/smux v1.5.24
 	go.starlark.net v0.0.0-20230612165344-9532f5667272
 	go4.org/netipx v0.0.0-20230303233057-f1b76eb4bb35
 	golang.org/x/crypto v0.49.0
@@ -71,10 +74,9 @@ require (
 	github.com/klauspost/cpuid/v2 v2.2.5 // indirect
 	github.com/klauspost/reedsolomon v1.11.7 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect
-	github.com/lunixbochs/struc v0.0.0-20200707160740-784aaebc1d40 // indirect
 	github.com/mustafaturan/monoton v1.0.0 // indirect
 	github.com/patrickmn/go-cache v2.1.0+incompatible // indirect
-	github.com/pion/dtls/v3 v3.0.11 // indirect
+	github.com/pion/dtls/v2 v2.2.12 // indirect
 	github.com/pion/logging v0.2.4 // indirect
 	github.com/pion/randutil v0.1.0 // indirect
 	github.com/pion/sctp v1.8.7 // indirect
@@ -87,7 +89,6 @@ require (
 	github.com/secure-io/siv-go v0.0.0-20180922214919-5ff40651e2c4 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/wlynxg/anet v0.0.5 // indirect
-	github.com/xtaci/smux v1.5.24 // indirect
 	golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842 // indirect
 	golang.org/x/mod v0.33.0 // indirect
 	golang.org/x/text v0.35.0 // indirect