增量更新节点状态

Dockerfile

@@ -24,6 +24,7 @@ RUN apt-get update && apt-get install -y ca-certificates
 
 # 从构建阶段复制编译好的二进制文件
 COPY --from=builder /build/bin/proxy_linux_amd64 /app/proxy
+COPY cmd/gateway/ip2region.xdb /app/id2region.xdb
 
 # 设置可执行权限
 RUN chmod +x /app/proxy

README.md

@@ -1,3 +1,21 @@
+## TODO
+
+数据通道池化
+
+协程池化
+
+访问失败的也返回访问记录
+
+解决节点断开立即重连会导致 status 覆盖的问题，维护一个修改表，每次提交后清空，每个节点的修改在同一阶段内只有一次
+
+节点每次发送心跳后提供或计算 loss 和 rtt 信息，并对比 host 和 status，将需要更新的数据提交到更新队列
+
+### 长期
+
+实现一个接口以手动全量同步数据库中的节点信息，防止网关节点数据与数据库出现偏差
+
+将协议内容抽离出公共包，gateway 和 edge 节点共同调用
+
 ## 开发相关
 
 ### 目录结构

cmd/gateway/main.go

@@ -5,8 +5,7 @@ import (
 )
 
 func main() {
-	var app = gateway.New()
-	var err = app.Run()
+	var err = gateway.Start()
 	if err != nil {
 		println(err.Error())
 	}

edge/edge.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"context"
 	"encoding/binary"
+	"encoding/hex"
 	"errors"
 	"fmt"
 	"io"
@@ -14,13 +15,14 @@ import (
 	"os/signal"
 	"proxy-server/edge/core"
 	"proxy-server/edge/env"
-	"proxy-server/edge/geo"
 	"proxy-server/edge/report"
 	"proxy-server/utils"
 	"time"
 )
 
 func Start() error {
+	var ctx, cancel = signal.NotifyContext(context.Background(), os.Interrupt, os.Kill)
+	defer cancel()
 
 	// 初始化环境变量
 	slog.Debug("初始化环境变量...")
@@ -29,64 +31,31 @@ func Start() error {
 		return fmt.Errorf("初始化环境变量失败: %w", err)
 	}
 
-	// 获取归属地
-	slog.Debug("获取节点归属地...")
-	err = geo.Query()
-	if err != nil {
-		return fmt.Errorf("获取节点归属地失败: %w", err)
-	}
-
 	// 注册节点
 	slog.Debug("注册节点...")
-	id, host, err := report.Online(geo.Prov, geo.City, geo.Isp)
+	id, host, err := report.Online()
 	if err != nil {
 		return fmt.Errorf("注册节点失败: %w", err)
 	}
 
 	// 连接到网关
-	var ctx, cancel = signal.NotifyContext(context.Background(), os.Interrupt, os.Kill)
-	defer cancel()
-
 	var errCh = make(chan error)
 	go func() {
-		for {
-			err = ctrl(ctx, id, host)
-			if err == nil {
-				errCh <- nil
-				return
-			}
-
-			select {
-			case <-ctx.Done():
-				return
-			default:
-				slog.Error("建立控制通道失败", "err", err)
-				slog.Info(fmt.Sprintf("%d 秒后重试", core.RetryInterval))
-			}
-
-			select {
-			case <-ctx.Done():
-				return
-			case <-time.After(time.Duration(core.RetryInterval) * time.Second):
-			}
+		err = ctrl(ctx, id, host)
+		if err == nil {
+			errCh <- err
 		}
 	}()
 
 	// 等待退出
 	select {
+	case <-ctx.Done():
 	case err := <-errCh:
 		if err != nil {
 			slog.Error("控制通道发生错误", "err", err)
 		}
 	}
 
-	// 下线节点
-	slog.Debug("下线节点...")
-	err = report.Offline()
-	if err != nil {
-		slog.Error("下线节点失败", "err", err)
-	}
-
 	return ctx.Err()
 }
 
@@ -102,114 +71,91 @@ func ctrl(ctx context.Context, id int32, host string) error {
 	defer utils.Close(conn)
 	var reader = bufio.NewReader(conn)
 
-	// 发送节点连接命令
-	slog.Debug("发送节点连接命令")
+	// 发送开启连接
 	err = sendOpen(conn, id)
 	if err != nil {
 		return fmt.Errorf("发送节点信息失败: %w", err)
 	}
 
 	// 异步定时发送心跳
+	var writeErr = make(chan error)
 	go func() {
-		ticker := time.NewTicker(time.Duration(core.HeartbeatInterval) * time.Second)
-		defer ticker.Stop()
+		ticker := time.Tick(time.Duration(core.HeartbeatInterval) * time.Second)
 		for {
 			select {
 			case <-ctx.Done():
+				writeErr <- nil
 				return
-			case tick := <-ticker.C:
+			case _ = <-ticker:
 				err := sendPing(conn)
 				if err != nil {
-					slog.Error("发送心跳失败", "time", tick, "err", err)
+					writeErr <- fmt.Errorf("发送心跳失败: %w", err)
+					return
 				}
 			}
 		}
 	}()
 
-	// 异步等待连接命令
-	slog.Info("等待用户连接")
-	var cmdCh = make(chan ConnCmd)
-	var errCh = make(chan error)
+	// 异步读取节点命令
+	var readErr = make(chan error)
 	go func() {
 		for {
+			// 读取命令
 			cmd, err := reader.ReadByte()
-			if err != nil {
-				switch {
-				case errors.Is(err, net.ErrClosed):
-					err = fmt.Errorf("控制通道关闭: %w", err)
-				case errors.Is(err, io.EOF):
-					err = fmt.Errorf("网关关闭了控制通道: %w", err)
-				default:
-					err = fmt.Errorf("读取命令失败: %w", err)
-				}
-				errCh <- err
+			if _, err := utils.WarpConnErr(err); err != nil {
+				readErr <- err
 				return
 			}
-
 			switch cmd {
+
+			// pong 命令，忽略
 			case 1:
-				// 忽略网关响应的 pong 命令
+
+			// 代理命令
 			case 5:
-				tag, addr, err := onConn(reader)
+				err := onConn(reader, dataAddr)
 				if err != nil {
-					slog.Error("接收连接命令失败", "err", err)
+					readErr <- fmt.Errorf("处理代理命令失败: %w", err)
 					return
 				}
-				cmdCh <- ConnCmd{
-					Tag:  tag,
-					Addr: addr,
-				}
 			}
 		}
 	}()
 
 	// 等待建立数据通道
-	for loop := true; loop; {
-		select {
-		case <-ctx.Done():
-			loop = false
-		case err = <-errCh:
-			slog.Error("读取控制命令失败", "err", err)
-			loop = false
-		case cmd := <-cmdCh:
-			slog.Debug("建立数据通道", "tag", cmd.Tag, "addr", cmd.Addr)
-			go func() {
-				err := data(dataAddr, cmd.Addr, cmd.Tag)
-				if err != nil {
-					slog.Error("建立数据通道失败", "err", err)
-				}
-			}()
-		}
+	select {
+	case <-ctx.Done():
+	case err = <-readErr:
 	}
 
-	// 发送关闭连接（不 return err，否则会重新连接）
+	// 发送关闭连接
 	slog.Debug("发送关闭连接")
 	err = sendClose(conn)
 	if err != nil {
-		slog.Error("发送关闭连接失败", "err", err)
+		return fmt.Errorf("发送关闭连接失败: %w", err)
 	}
 
 	return nil
 }
 
-func data(proxy string, dest string, tag [16]byte) error {
+func data(proxy string, destination string, tag [16]byte) error {
+	var tagStr = hex.EncodeToString(tag[:])
+	slog.Debug("建立数据通道", "tag", tagStr, "addr", destination)
 
 	// 向目标地址建立连接
 	var result = 1
 	var dstErr error
-	dst, err := net.Dial("tcp", dest)
+	dst, err := net.Dial("tcp", destination)
 	if err != nil {
-		dstErr = fmt.Errorf("连接目标地址失败: %w", dstErr)
+		dstErr = ErrDstConnFailed
 		result = 0
 	}
-	defer utils.Close(dst)
 
 	// 向服务端建立连接
 	src, err := net.Dial("tcp", proxy)
 	if err != nil {
 		return errors.New("连接服务端失败")
 	}
-	defer utils.Close(src)
 
 	// 发送连接状态
 	var buf = make([]byte, 17)
@@ -224,39 +170,29 @@ func data(proxy string, dest string, tag [16]byte) error {
 		return dstErr
 	}
 
-	var waitSrc = make(chan error)
+	var waitSrc = make(chan struct{}, 1)
 	go func() {
+		defer utils.Close(dst)
 		_, err := io.Copy(dst, src)
-		switch {
-		case errors.Is(err, net.ErrClosed):
-			slog.Debug("网关连接意外关闭")
-		case err != nil:
-			slog.Error("读取网关数据失败", "err", err)
-		default:
-			slog.Debug("网关数据读取完成")
+		if ok, err := utils.WarpConnErr(err); !ok {
+			slog.Error("读取网关数据失败", "tag", tagStr, "err", err)
 		}
-		waitSrc <- err
+		waitSrc <- struct{}{}
 	}()
 
-	var waitDst = make(chan error)
+	var waitDst = make(chan struct{}, 1)
 	go func() {
+		defer utils.Close(src)
 		_, err := io.Copy(src, dst)
-		switch {
-		case errors.Is(err, net.ErrClosed):
-			slog.Debug("目标连接意外关闭")
-		case err != nil && !errors.Is(err, net.ErrClosed):
+		if ok, err := utils.WarpConnErr(err); !ok {
 			slog.Error("读取目标数据失败", "err", err)
-		default:
-			slog.Debug("目标数据读取完成")
 		}
-		waitDst <- err
+		waitDst <- struct{}{}
 	}()
 
-	// 等待任意一方关闭数据连接
-	select {
-	case <-waitSrc:
-	case <-waitDst:
-	}
+	// 等待连接全部安全关闭或超时
+	<-waitSrc
+	<-waitDst
 
 	return nil
 }
@@ -295,27 +231,50 @@ func sendPing(writer io.Writer) error {
 	return nil
 }
 
-func onConn(reader io.Reader) (tag [16]byte, addr string, err error) {
+func onConn(reader io.Reader, dataAddr string) (err error) {
+
+	// 读取连接命令
 	var buf = make([]byte, 16+2)
 	_, err = io.ReadFull(reader, buf)
 	if err != nil {
-		return [16]byte{}, "", err
+		return err
 	}
 
-	tag = [16]byte(buf[0:16])
+	var tag = [16]byte(buf[0:16])
 
 	var addrLen = binary.BigEndian.Uint16(buf[16:18])
 	var addrBuf = make([]byte, addrLen)
 	_, err = io.ReadFull(reader, addrBuf)
 	if err != nil {
-		return [16]byte{}, "", err
+		return err
 	}
+	var addr = string(addrBuf)
 
-	addr = string(addrBuf)
-	return tag, addr, nil
+	// 异步建立数据通道
+	go func() {
+		err := data(dataAddr, addr, tag)
+		switch {
+		case errors.Is(err, ErrDstConnFailed):
+			slog.Debug("目标地址连接失败", "tag", hex.EncodeToString(tag[:]), "addr", addr)
+		case err != nil:
+			slog.Error("建立数据通道失败", "err", err)
+		}
+	}()
+
+	return nil
 }
 
 type ConnCmd struct {
 	Tag  [16]byte
 	Addr string
 }
+
+type Err string
+
+func (e Err) Error() string {
+	return string(e)
+}
+
+const (
+	ErrDstConnFailed = Err("目标地址连接失败")
+)

edge/env/env.go

@@ -9,15 +9,13 @@ import (
 var Mode = "dev"
 var Name = "dev-edge"
 
-var EndpointOnline = "https://api.lanhuip.com/api/edge/online"
-var EndpointOffline = "https://api.lanhuip.com/api/edge/offline"
+var EndpointAssign = "https://api.lanhuip.com/api/edge/assign"
 
 func Init() error {
 
 	var env = flag.String("e", "dev", "环境变量，可选值 dev 或 prod")
 	var name = flag.String("n", "", "节点唯一标识")
-	var online = flag.String("online", "", "服务注册地址")
-	var offline = flag.String("offline", "", "服务注销地址")
+	var assign = flag.String("assign", "", "服务发现地址")
 
 	flag.Parse()
 
@@ -35,18 +33,14 @@ func Init() error {
 		return errors.New("节点唯一标识不能为空")
 	}
 
-	if online != nil && *online != "" {
-		EndpointOnline = *online
-	}
-
-	if offline != nil && *offline != "" {
-		EndpointOffline = *offline
+	if assign != nil && *assign != "" {
+		EndpointAssign = *assign
 	}
 
 	if Mode == "dev" {
 		slog.SetLogLoggerLevel(slog.LevelDebug)
 	} else {
-		slog.SetLogLoggerLevel(slog.LevelWarn)
+		slog.SetLogLoggerLevel(slog.LevelInfo)
 	}
 
 	return nil

edge/geo/geo.go

@@ -1,159 +0,0 @@
-package geo
-
-import (
-	"bufio"
-	"encoding/json"
-	"fmt"
-	"io"
-	"net/http"
-	"net/textproto"
-	"proxy-server/edge/env"
-	"strings"
-)
-
-var (
-	Ip   string
-	Prov string
-	City string
-	Isp  string
-)
-
-func Query() (err error) {
-
-	switch env.Mode {
-	case "dev":
-		err = dev()
-	default:
-		err = ipapi()
-	}
-	return err
-}
-
-func dev() (err error) {
-	Prov = "河南省"
-	City = "郑州市"
-	Isp = "电信"
-	return nil
-}
-
-func cip() (err error) {
-	//goland:noinspection HttpUrlsUsage
-	const endpoint = "http://cip.cc"
-
-	req, err := http.NewRequest("GET", endpoint, nil)
-	if err != nil {
-		return fmt.Errorf("创建请求失败: %w", err)
-	}
-	req.Header.Set("User-Agent", "curl/8.9.1")
-	req.Header.Set("Accept", "*/*")
-
-	resp, err := http.DefaultClient.Do(req)
-	if err != nil {
-		return fmt.Errorf("执行请求失败: %w", err)
-	}
-	defer func(Body io.ReadCloser) {
-		_ = Body.Close()
-	}(resp.Body)
-	if resp.StatusCode != http.StatusOK {
-		return fmt.Errorf("状态码: %s", resp.Status)
-	}
-
-	reader := textproto.NewReader(bufio.NewReader(resp.Body))
-	ipLine, err := reader.ReadLine()
-	if err != nil {
-		return fmt.Errorf("读取响应失败: %w", err)
-	}
-	Ip = strings.TrimSpace(strings.Split(ipLine, ":")[1])
-
-	addrLine, err := reader.ReadLine()
-	if err != nil {
-		return fmt.Errorf("读取响应失败: %w", err)
-	}
-	addr := strings.Split(strings.Split(addrLine, ":")[1], " ")
-	Prov = strings.TrimSpace(addr[1])
-	City = strings.TrimSpace(addr[2])
-
-	ispLine, err := reader.ReadLine()
-	if err != nil {
-		return fmt.Errorf("读取响应失败: %w", err)
-	}
-	Isp = strings.TrimSpace(strings.Split(ispLine, ":")[1])
-
-	return nil
-}
-
-func ipapi() (err error) {
-	//goland:noinspection HttpUrlsUsage
-	const endpoint = "http://ip-api.com/json/?fields=regionName,city,as,query&lang=zh-CN"
-
-	resp, err := http.Get(endpoint)
-	if err != nil {
-		return fmt.Errorf("执行请求失败: %w", err)
-	}
-	defer func(Body io.ReadCloser) {
-		_ = Body.Close()
-	}(resp.Body)
-
-	if resp.StatusCode != http.StatusOK {
-		return fmt.Errorf("状态码: %s", resp.Status)
-	}
-
-	var data struct {
-		RegionName string `json:"regionName"`
-		City       string `json:"city"`
-		As         string `json:"as"`
-		Query      string `json:"query"`
-	}
-	err = json.NewDecoder(resp.Body).Decode(&data)
-	if err != nil {
-		return fmt.Errorf("解析响应失败: %w", err)
-	}
-
-	Ip = data.Query
-	Prov = data.RegionName
-	City = data.City
-
-	var telecom = []string{"AS4134", "AS4812", "AS134419", "AS140292"}
-	var unicom = []string{"AS4837", "AS17621", "AS17816"}
-	var mobile = []string{
-		"AS9808", "AS24444", "AS24445", "AS24547", "AS38019",
-		"AS56040", "AS56041", "AS56042", "AS56044", "AS56046", "AS56047",
-		"AS132525", "AS134810",
-	}
-	var foreign = []string{
-		"AS9299",
-	}
-
-	for _, telecomAsn := range telecom {
-		if strings.HasPrefix(data.As, telecomAsn) {
-			Isp = "电信"
-			break
-		}
-	}
-	if Isp == "" {
-		for _, unicomAsn := range unicom {
-			if strings.HasPrefix(data.As, unicomAsn) {
-				Isp = "联通"
-				break
-			}
-		}
-	}
-	if Isp == "" {
-		for _, mobileAsn := range mobile {
-			if strings.HasPrefix(data.As, mobileAsn) {
-				Isp = "移动"
-				break
-			}
-		}
-	}
-	if Isp == "" {
-		for _, foreignAsn := range foreign {
-			if strings.HasPrefix(data.As, foreignAsn) {
-				Isp = "国外"
-				break
-			}
-		}
-	}
-
-	return nil
-}

edge/report/report.go

@@ -11,22 +11,8 @@ import (
 	"strings"
 )
 
-func Online(prov, city, isp string) (id int32, host string, err error) {
-
-	var ispInt = 0
-	switch isp {
-	case "电信":
-		ispInt = 1
-	case "联通":
-		ispInt = 2
-	case "移动":
-		ispInt = 3
-	}
-
+func Online() (id int32, host string, err error) {
 	bytes, err := json.Marshal(map[string]any{
-		"prov":    prov,
-		"city":    city,
-		"isp":     ispInt,
 		"name":    env.Name,
 		"version": core.Version,
 	})
@@ -35,11 +21,14 @@ func Online(prov, city, isp string) (id int32, host string, err error) {
 	}
 	var body = strings.NewReader(string(bytes))
 
-	resp, err := http.Post(env.EndpointOnline, "application/json", body)
+	resp, err := http.Post(env.EndpointAssign, "application/json", body)
 	if err != nil {
 		return 0, "", fmt.Errorf("执行请求失败: %w", err)
 	}
-	defer resp.Body.Close()
+	defer func(Body io.ReadCloser) {
+		_ = Body.Close()
+	}(resp.Body)
+
 	if resp.StatusCode != http.StatusOK {
 		return 0, "", errors.New("状态码: " + resp.Status)
 	}
@@ -62,24 +51,3 @@ func Online(prov, city, isp string) (id int32, host string, err error) {
 
 	return respBody.Id, respBody.Host, nil
 }
-
-func Offline() error {
-	var bytes, err = json.Marshal(map[string]any{
-		"name": env.Name,
-	})
-	if err != nil {
-		return err
-	}
-	var body = strings.NewReader(string(bytes))
-
-	resp, err := http.Post(env.EndpointOffline, "application/json", body)
-	if err != nil {
-		return fmt.Errorf("执行请求失败: %w", err)
-	}
-	defer resp.Body.Close()
-	if resp.StatusCode != http.StatusOK {
-		return errors.New("状态码: " + resp.Status)
-	}
-
-	return nil
-}

gateway/app/app.go

@@ -1,8 +1,11 @@
 package app
 
 import (
+	"fmt"
+	"net"
 	"proxy-server/gateway/core"
 	"proxy-server/utils"
+	"sync"
 )
 
 type Stoppable interface {
@@ -14,25 +17,99 @@ var (
 	Name           string
 	PlatformSecret string // 平台密钥，验证接收的请求是否属于平台
 
-	Assigns = core.SyncMap[uint16, int32]{}       // 转发端口 -> 节点 ID
-	Edges   = core.SyncMap[int32, uint16]{}       // 节点 ID -> 转发端口
-	Permits = core.SyncMap[int32, *core.Permit]{} // 转发端口 -> 权限配置
+	Edges   = core.SyncMap[int32, *core.Edge]{} // 节点信息表		（包外只读！数据存储有关联性，所有写入操作只在包内进行）
+	Assigns = core.SyncMap[uint16, int32]{}     // 分配索引		（包外只读！数据存储有关联性，所有写入操作只在包内进行）
+
+	Permits = core.SyncMap[int32, *core.Permit]{} // 节点权限表	（包外只读！数据存储有关联性，所有写入操作只在包内进行）
 
 	CtrlConnWg  utils.CountWaitGroup             // 控制通道计数器
 	DataConnWg  utils.CountWaitGroup             // 数据通道计数器
 	FwdLesWg    utils.CountWaitGroup             // 转发监听端口计数器
 	UserConnWg  utils.CountWaitGroup             // 用户连接计数器
 	UserConnMap core.SyncMap[string, *core.Conn] // 用户连接暂存
+
+	LockPortAssign = sync.Mutex{}               // 锁定端口分配，防止并发冲突
+	EdgeUpdates    = make(chan *core.Edge, 100) // 节点更新通知通道
 )
 
-func AddEdge(id int32, port uint16) {
-	Edges.Store(id, port)
+func OnlineEdgeNew(id int32, port uint16, addr *net.TCPAddr) error {
+	if addr == nil {
+		return fmt.Errorf("边缘节点 %d 地址无效", id)
+	}
+	info, err := IpGeo(addr.IP)
+	if err != nil {
+		return fmt.Errorf("查询 geo 信息失败：%w", err)
+	}
+
+	var host = addr.IP.String()
+	var edge = &core.Edge{
+		Id:     id,
+		Host:   &host,
+		Port:   &port,
+		Prov:   &info.Prov,
+		City:   &info.City,
+		Isp:    &info.Isp,
+		Status: &core.EdgeOnline,
+	}
+
+	Edges.Store(id, edge)
 	Assigns.Store(port, id)
+	EdgeUpdates <- edge
+
+	return nil
 }
 
-func DelEdge(port uint16) {
-	id, _ := Assigns.LoadAndDelete(port)
-	Edges.Delete(id)
+func OnlineEdgeUpdate(id int32, addr *net.TCPAddr) error {
+	if addr == nil {
+		return fmt.Errorf("边缘节点 %d 地址无效", id)
+	}
+	edge, ok := Edges.Load(id)
+	if !ok {
+		return fmt.Errorf("边缘节点 %d 不存在", id)
+	}
+
+	edge.Status = &core.EdgeOnline
+	toUpdate := &core.Edge{
+		Id:     edge.Id,
+		Status: &core.EdgeOnline,
+	}
+
+	host := addr.IP.String()
+	if edge.Host == nil || *edge.Host != host {
+		edge.Host = &host
+		toUpdate.Host = &host
+	}
+
+	EdgeUpdates <- toUpdate
+
+	return nil
+}
+
+func OfflineEdge(id int32) error {
+	edge, ok := Edges.Load(id)
+	if !ok || edge == nil {
+		return fmt.Errorf("边缘节点 %d 不存在", id)
+	}
+
+	edge.Status = &core.EdgeOffline
+	EdgeUpdates <- &core.Edge{
+		Id:     id,
+		Status: &core.EdgeOffline,
+	}
+
+	return nil
+}
+
+func StoreEdge(edge *core.Edge) error {
+	if edge == nil || edge.Id == 0 {
+		return fmt.Errorf("无效的边缘节点: %+v", edge)
+	}
+
+	Edges.Store(edge.Id, edge)
+	if edge.Port != nil {
+		Assigns.Store(*edge.Port, edge.Id)
+	}
+	return nil
 }
 
 func LoadPermit(port uint16) *core.Permit {
@@ -48,3 +125,7 @@ func LoadPermit(port uint16) *core.Permit {
 
 	return permit
 }
+
+func StorePermit(def *core.PermitDef) {
+	Permits.Store(def.Id, &def.Permit)
+}

gateway/app/geo.go

@@ -0,0 +1,53 @@
+package app
+
+import (
+	"fmt"
+	"log/slog"
+	"net"
+	g "proxy-server/gateway/globals"
+	"strings"
+)
+
+func IpGeo(ip net.IP) (info *IpGeoInfo, err error) {
+	defer func() {
+		var rs = recover()
+		if reErr, ok := rs.(error); ok {
+			err = fmt.Errorf("执行归属地查询异常 %w", reErr)
+		}
+	}()
+
+	// 本地归属地查询
+	str, err := g.Geo.SearchByStr(ip.String())
+	if err != nil {
+		return nil, err
+	}
+
+	if str != "" {
+		slog.Debug("本地解析归属地结果", "str", str)
+		values := strings.Split(str, "|")
+		if len(values) != 5 {
+			return nil, fmt.Errorf("本地归属地查询解析失败")
+		}
+
+		var info = &IpGeoInfo{}
+
+		if values[2] != "0" {
+			info.Prov = values[2]
+		}
+		if values[3] != "0" {
+			info.City = values[3]
+		}
+		if values[4] != "0" {
+			info.Isp = values[4]
+		}
+		return info, nil
+	}
+
+	return nil, fmt.Errorf("本地归属地查询为空")
+}
+
+type IpGeoInfo struct {
+	Prov string
+	City string
+	Isp  string
+}

gateway/app/report.go

@@ -1,12 +1,12 @@
-package report
+package app
 
 import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"io"
+	"log/slog"
 	"net/http"
-	"proxy-server/gateway/app"
 	"proxy-server/gateway/core"
 	"proxy-server/gateway/env"
 	"strings"
@@ -23,34 +23,45 @@ func Online(name string) (err error) {
 	}
 
 	var body struct {
-		Id     int32  `json:"id"`
-		Secret string `json:"secret"`
+		Id      int32             `json:"id"`
+		Secret  string            `json:"secret"`
+		Permits []*core.PermitDef `json:"permits"`
+		Edges   []*core.Edge      `json:"edges"`
 	}
 	err = json.Unmarshal([]byte(resp), &body)
 	if err != nil {
 		return err
 	}
 
-	app.Id = body.Id
-	app.PlatformSecret = body.Secret
-
+	Id = body.Id
+	PlatformSecret = body.Secret
+	for _, def := range body.Permits {
+		StorePermit(def)
+	}
+	for _, edge := range body.Edges {
+		err := StoreEdge(edge)
+		if err != nil {
+			slog.Error("存储边缘节点失败", "err", err, "edge", edge)
+		}
+	}
 	return nil
 }
 
-func Offline(name string) (err error) {
+func Offline() (err error) {
 	_, err = call(env.EndpointOffline, map[string]any{
-		"name":    name,
-		"version": core.Version,
+		"id": Id,
 	})
 	return err
 }
 
-func Assigned(edgeId int32, port uint16) (err error) {
-	_, err = call(env.EndpointAssigned, map[string]any{
-		"proxy": app.Id,
-		"edge":  edgeId,
-		"port":  port,
+func Update(data []*core.Edge) (err error) {
+	_, err = call(env.EndpointUpdate, map[string]any{
+		"id":    Id,
+		"edges": data,
 	})
+	if err != nil {
+		err = fmt.Errorf("更新节点数据失败：%w", err)
+	}
 	return err
 }
 
@@ -74,7 +85,10 @@ func call(endpoint string, body any) (string, error) {
 	if err != nil {
 		return "", err
 	}
-	defer resp.Body.Close()
+	defer func(Body io.ReadCloser) {
+		_ = Body.Close()
+	}(resp.Body)
+
 	if resp.StatusCode != http.StatusOK {
 		return "", fmt.Errorf("请求失败，状态码：%d", resp.StatusCode)
 	}

gateway/core/auth.go

@@ -8,3 +8,8 @@ type Permit struct {
 	Username   *string   `json:"username"`
 	Password   *string   `json:"password"`
 }
+
+type PermitDef struct {
+	Id int32 `json:"id"`
+	Permit
+}

gateway/core/edge.go

@@ -0,0 +1,18 @@
+package core
+
+type Edge struct {
+	Id     int32   `json:"id"`
+	Host   *string `json:"host,omitempty"`
+	Port   *uint16 `json:"port,omitempty"`
+	Prov   *string `json:"prov,omitempty"`
+	City   *string `json:"city,omitempty"`
+	Isp    *string `json:"isp,omitempty"`
+	Status *int    `json:"status,omitempty"`
+	Rtt    *int    `json:"rtt,omitempty"`  // 节点响应时间，单位毫秒
+	Loss   *int    `json:"loss,omitempty"` // 节点丢包率，单位百分比
+}
+
+var (
+	EdgeOffline = 0
+	EdgeOnline  = 1
+)

gateway/env/env.go

@@ -14,13 +14,16 @@ import (
 var (
 	RunMode = "dev" // 运行模式，dev: 开发模式，prod: 生产模式
 
-	AppCtrlPort    uint16 = 18080
-	AppDataPort    uint16 = 18081
-	AppWebPort     uint16 = 8848
-	AppLogMode            = "dev"
-	AppExitTimeout        = 5  // 等待服务停止的超时时间
-	AppDataTimeout        = 10 // 等待数据通道连接的超时时间
-	AppUserTimeout        = 10 // 等待用户发送数据的超时时间（端口复用需要分析协议，如果用户长期不发送数据，将会阻塞分析协程）
+	AppCtrlPort      uint16 = 18080
+	AppDataPort      uint16 = 18081
+	AppWebPort       uint16 = 8848
+	AppLogMode              = "dev"
+	AppExitTimeout          = 5  // 等待服务停止的超时时间
+	AppDataTimeout          = 20 // 等待数据通道连接的超时时间
+	AppUserRWTimeout        = 10 // 等待用户连接读写超时时间
+	AppDataRWTimeout        = 10 // 等待数据通道读写超时时间
+	AppCtrlRWTimeout        = 10 // 等待控制通道读写超时时间
+	AppCtrlHBTimeout        = 30 // 控制通道心跳超时时间（断开连接等待时间为：心跳等待时间 * 2 + 读写等待时间）
 
 	AuthWhitelist []net.IP // 全局白名单，可以将白名单 IP 视为一个可信任代理
 
@@ -32,9 +35,9 @@ var (
 	RedisDb   = 0
 	RedisPass = ""
 
-	EndpointOnline   string
-	EndpointOffline  string
-	EndpointAssigned string
+	EndpointOnline  string
+	EndpointOffline string
+	EndpointUpdate  string
 )
 
 func Init() {
@@ -103,13 +106,40 @@ func Init() {
 		AppDataTimeout = appDataTimeout
 	}
 
-	value = os.Getenv("APP_USER_TIMEOUT")
+	value = os.Getenv("APP_USER_RW_TIMEOUT")
 	if value != "" {
 		appUserTimeout, err := strconv.Atoi(value)
 		if err != nil {
-			panic(fmt.Sprintf("环境变量 APP_USER_TIMEOUT 格式错误: %v", err))
+			panic(fmt.Sprintf("环境变量 APP_USER_RW_TIMEOUT 格式错误: %v", err))
 		}
-		AppUserTimeout = appUserTimeout
+		AppUserRWTimeout = appUserTimeout
+	}
+
+	value = os.Getenv("APP_DATA_RW_TIMEOUT")
+	if value != "" {
+		appDataRWTimeout, err := strconv.Atoi(value)
+		if err != nil {
+			panic(fmt.Sprintf("环境变量 APP_DATA_RW_TIMEOUT 格式错误: %v", err))
+		}
+		AppDataRWTimeout = appDataRWTimeout
+	}
+
+	value = os.Getenv("APP_CTRL_RW_TIMEOUT")
+	if value != "" {
+		appCtrlRWTimeout, err := strconv.Atoi(value)
+		if err != nil {
+			panic(fmt.Sprintf("环境变量 APP_CTRL_RW_TIMEOUT 格式错误: %v", err))
+		}
+		AppCtrlRWTimeout = appCtrlRWTimeout
+	}
+
+	value = os.Getenv("APP_CTRL_HB_TIMEOUT")
+	if value != "" {
+		appCtrlHBTimeout, err := strconv.Atoi(value)
+		if err != nil {
+			panic(fmt.Sprintf("环境变量 APP_CTRL_HB_TIMEOUT 格式错误: %v", err))
+		}
+		AppCtrlHBTimeout = appCtrlHBTimeout
 	}
 
 	value = os.Getenv("AUTH_WHITELIST")
@@ -179,10 +209,11 @@ func Init() {
 	} else {
 		panic("环境变量 ENDPOINT_OFFLINE 未设置")
 	}
-	value = os.Getenv("ENDPOINT_ASSIGNED")
+
+	value = os.Getenv("ENDPOINT_UPDATE")
 	if value != "" {
-		EndpointAssigned = value
+		EndpointUpdate = value
 	} else {
-		panic("环境变量 ENDPOINT_ASSIGNED 未设置")
+		panic("环境变量 ENDPOINT_UPDATE 未设置")
 	}
 }

gateway/fwd/ctrl.go

@@ -11,10 +11,9 @@ import (
 	"net"
 	"proxy-server/gateway/app"
 	"proxy-server/gateway/env"
-	"proxy-server/gateway/report"
 	"proxy-server/utils"
 	"strconv"
-	"syscall"
+	"time"
 )
 
 type CtrlCmdType int
@@ -37,8 +36,7 @@ func ListenCtrl(ctx context.Context) error {
 	}
 	defer utils.Close(ls)
 
-	// 处理连接
-	// 异步等待连接
+	// 异步等待处理连接
 	var connCh = make(chan net.Conn)
 	go func() {
 		for {
@@ -80,123 +78,153 @@ func ListenCtrl(ctx context.Context) error {
 }
 
 func processCtrlConn(_ctx context.Context, conn net.Conn) (err error) {
-	// 通道上下文
-	ctx, cancel := context.WithCancel(_ctx)
+	reader := bufio.NewReader(conn)
 
-	// 结束后清理资源
-	var fwdPort uint16
+	// 上下文与通道信息
+	ctx, cancel := context.WithCancel(_ctx)
+	defer cancel()
+
+	// 结束时清理
+	var edgeId int32
 	defer func() {
-		slog.Debug("关闭控制通道", "port", fwdPort)
-		app.DelEdge(fwdPort)
+		err := app.OfflineEdge(edgeId)
+		if err != nil {
+			slog.Error("管理节点下线失败", "err", err)
+		}
 	}()
 
-	// 处理控制命令
-	defer cancel()
-	reader := bufio.NewReader(conn)
-	for {
-		// 循环等待直到服务关闭
-		select {
-		case <-ctx.Done():
-			return nil
-		default:
-		}
-
-		// 读取命令
-		cmd, err := reader.ReadByte()
-		if errors.Is(err, syscall.WSAECONNRESET) {
-			slog.Debug("节点重置了控制通道连接(WSAECONNRESET)")
-			return nil
-		}
-		if errors.Is(err, io.EOF) {
-			slog.Debug("节点关闭了控制通道")
-			return nil
-		}
-		if err != nil {
-			return fmt.Errorf("读取节点命令失败: %w", err)
-		}
-
-		// 处理节点命令
-		switch CtrlCmdType(cmd) {
-
-		// 连接建立命令
-		case CtrlCmdOpen:
-			var recv = make([]byte, 4)
-			_, err = io.ReadFull(reader, recv)
+	// 处理连接命令
+	var errCh = make(chan error)
+	go func() {
+		var err error
+		for {
+			// 读取命令
+			var timeout = time.Duration(env.AppCtrlHBTimeout*2+env.AppCtrlRWTimeout) * time.Second
+			err = conn.SetReadDeadline(time.Now().Add(timeout))
 			if err != nil {
-				return fmt.Errorf("读取节点 ID 失败: %w", err)
-			}
-			var client = int32(binary.BigEndian.Uint32(recv))
-			fwdPort, err = onOpen(ctx, conn, client)
-			if err != nil {
-				return fmt.Errorf("处理连接建立命令失败: %w", err)
+				errCh <- fmt.Errorf("设置读取超时失败: %w", err)
+				return
 			}
 
-		// 心跳命令
-		case CtrlCmdPing:
-			err = onPing(conn)
-			if err != nil {
-				return fmt.Errorf("处理心跳命令失败: %w", err)
+			cmd, err := reader.ReadByte()
+			if _, err := utils.WarpConnErr(err); err != nil {
+				errCh <- err
+				return
 			}
 
-		// 连接关闭命令
-		case CtrlCmdClose:
-			err = onClose(conn)
+			// 处理节点命令
+			err = conn.SetReadDeadline(time.Now().Add(time.Duration(env.AppCtrlRWTimeout) * time.Second))
 			if err != nil {
-				return fmt.Errorf("处理关闭命令失败: %w", err)
+				errCh <- fmt.Errorf("设置读取超时失败: %w", err)
+				return
 			}
-			return nil
+			switch CtrlCmdType(cmd) {
 
-		// 忽略其他不应该由节点发起的命令
-		default:
-			return fmt.Errorf("无法处理控制命令: %d", cmd)
+			// 连接建立命令
+			case CtrlCmdOpen:
+				var recv = make([]byte, 4)
+				_, err = io.ReadFull(reader, recv)
+				if err != nil {
+					errCh <- fmt.Errorf("读取节点 ID 失败: %w", err)
+					return
+				}
+				edgeId = int32(binary.BigEndian.Uint32(recv))
+				err = onOpen(ctx, conn, edgeId, conn.RemoteAddr())
+				if err != nil {
+					errCh <- fmt.Errorf("处理连接建立命令失败: %w", err)
+					return
+				}
+
+			// 心跳命令
+			case CtrlCmdPing:
+				err = onPing(conn)
+				if err != nil {
+					errCh <- fmt.Errorf("处理心跳命令失败: %w", err)
+					return
+				}
+
+			// 连接关闭命令
+			case CtrlCmdClose:
+				err = onClose(conn)
+				if err != nil {
+					errCh <- fmt.Errorf("处理关闭命令失败: %w", err)
+					return
+				}
+
+			// 忽略其他不应该由节点发起的命令
+			default:
+				errCh <- fmt.Errorf("无法处理控制命令: %d", cmd)
+				return
+			}
 		}
+	}()
+
+	// 等待处理结束
+	select {
+	case <-ctx.Done():
+	case err = <-errCh:
 	}
+
+	return
 }
 
-func onOpen(ctx context.Context, writer io.Writer, edge int32) (port uint16, err error) {
-	// open 命令全局只执行一次
-	_, ok := app.Edges.Load(edge)
-	if ok {
-		return 0, fmt.Errorf("节点 ID %d 已经连接", edge)
+func onOpen(ctx context.Context, writer io.Writer, edgeId int32, addr net.Addr) (err error) {
+	tcpAddr, ok := addr.(*net.TCPAddr)
+	if !ok {
+		return fmt.Errorf("无效的地址类型: %T", addr)
 	}
 
-	// 分配端口
-	var minim uint16 = 20000
-	var maxim uint16 = 60000
-	for i := minim; i < maxim; i++ {
-		var _, ok = app.Assigns.Load(i)
-		if !ok {
-			port = i
-			app.AddEdge(edge, port)
-			break
+	var port uint16
+	edge, ok := app.Edges.Load(edgeId)
+	if !ok || edge.Port == nil {
+		// 分配端口
+		app.LockPortAssign.Lock()
+
+		var minim uint16 = 20000
+		var maxim uint16 = 60000
+		for i := minim; i < maxim; i++ {
+			var _, ok = app.Assigns.Load(i)
+			if !ok {
+				port = i
+				break
+			}
+		}
+		if port == 0 {
+			return errors.New("没有可用的端口")
 		}
-	}
-	if port == 0 {
-		return 0, errors.New("没有可用的端口")
-	}
 
-	// 报告端口分配
-	if err = report.Assigned(edge, port); err != nil {
-		return 0, fmt.Errorf("报告端口分配失败: %w", err)
-	}
+		err := app.OnlineEdgeNew(edgeId, port, tcpAddr)
+		if err != nil {
+			return fmt.Errorf("新增边缘节点失败：%w", err)
+		}
 
-	// 响应节点
-	if err = sendPong(writer); err != nil {
-		return 0, fmt.Errorf("响应节点失败: %w", err)
+		app.LockPortAssign.Unlock()
+	} else {
+		// 更新边缘节点地址
+		port = *edge.Port
+		err := app.OnlineEdgeUpdate(edgeId, tcpAddr)
+		if err != nil {
+			return fmt.Errorf("尝试更新边缘节点失败: %w", err)
+		}
 	}
 
 	// 启动转发服务
 	app.FwdLesWg.Add(1)
 	go func() {
 		defer app.FwdLesWg.Done()
-		slog.Info("监听转发端口", "port", port, "edge", edge)
+		slog.Info("监听转发端口", "port", port, "edge", edgeId)
 		err = ListenUser(ctx, port, writer)
 		if err != nil {
-			slog.Error("监听转发端口失败", "port", port, "edge", edge, "err", err)
+			slog.Error("监听转发端口失败", "port", port, "edge", edgeId, "err", err)
 		}
 	}()
 
-	return port, nil
+	// 响应节点
+	if err = sendPong(writer); err != nil {
+		return fmt.Errorf("响应节点失败: %w", err)
+	}
+
+	return nil
 }
 
 func onPing(writer io.Writer) (err error) {

gateway/fwd/data.go

@@ -60,10 +60,7 @@ func ListenData(ctx context.Context) error {
 			app.DataConnWg.Add(1)
 			go func() {
 				defer app.DataConnWg.Done()
-				defer func() {
-					utils.Close(conn)
-					slog.Debug("关闭数据通道连接")
-				}()
+				defer utils.Close(conn)
 				err := processDataConn(ctx, conn)
 				if err != nil {
 					slog.Error("处理数据通道连接失败", "err", err)
@@ -73,8 +70,8 @@ func ListenData(ctx context.Context) error {
 	}
 }
 
-func processDataConn(ctx context.Context, client net.Conn) error {
-	var reader = bufio.NewReader(client)
+func processDataConn(ctx context.Context, edge net.Conn) error {
+	var reader = bufio.NewReader(edge)
 
 	// 接收连接结果
 	var buf = make([]byte, 17)
@@ -89,12 +86,13 @@ func processDataConn(ctx context.Context, client net.Conn) error {
 	// 加载用户连接
 	user, ok := app.UserConnMap.LoadAndDelete(tag)
 	if !ok {
-		return fmt.Errorf("用户连接已关闭，tag：%s", tag)
+		if status == 1 {
+			return fmt.Errorf("用户连接已关闭，tag：%s", tag)
+		} else {
+			return nil
+		}
 	}
-	defer func() {
-		utils.Close(user)
-		slog.Debug("关闭用户连接")
-	}()
+	defer utils.Close(user)
 
 	// 检查状态
 	if status != 1 {
@@ -119,13 +117,8 @@ func processDataConn(ctx context.Context, client net.Conn) error {
 	var waitEdge = make(chan error)
 	go func() {
 		_, err := io.Copy(user, reader)
-		switch {
-		case errors.Is(err, net.ErrClosed):
-			slog.Debug("节点连接意外关闭")
-		case err != nil:
+		if ok, err := utils.WarpConnErr(err); !ok {
 			slog.Error("读取节点数据失败", "err", err)
-		default:
-			slog.Debug("节点数据读取完成")
 		}
 		waitEdge <- err
 	}()
@@ -133,14 +126,9 @@ func processDataConn(ctx context.Context, client net.Conn) error {
 	// 复制用户数据到节点
 	var waitUser = make(chan error)
 	go func() {
-		_, err := io.Copy(client, teeUser)
-		switch {
-		case errors.Is(err, net.ErrClosed):
-			slog.Debug("用户连接意外关闭")
-		case err != nil:
+		_, err := io.Copy(edge, teeUser)
+		if ok, err := utils.WarpConnErr(err); !ok {
 			slog.Error("读取用户数据失败", "err", err)
-		default:
-			slog.Debug("用户数据读取完成")
 		}
 		waitUser <- err
 	}()

gateway/fwd/user.go

@@ -17,7 +17,7 @@ import (
 )
 
 func ListenUser(ctx context.Context, port uint16, ctrl io.Writer) error {
-	dspt, err := dispatcher.New(port, time.Duration(env.AppUserTimeout)*time.Second)
+	dspt, err := dispatcher.New(port, time.Duration(env.AppUserRWTimeout)*time.Second)
 	if err != nil {
 		return err
 	}
@@ -85,7 +85,7 @@ func processUserConn(ctx context.Context, user *core.Conn, ctrl io.Writer) (err
 	if ok {
 		utils.Close(user)
 		if errors.Is(err, context.DeadlineExceeded) {
-			slog.Error("用户连接超时", "tag", tag, "addr", user.RemoteAddr().String())
+			slog.Warn("建立数据通道超时", "tag", tag, "addr", user.RemoteAddr().String())
 		}
 	}
 

gateway/gateway.go

@@ -13,7 +13,6 @@ import (
 	"proxy-server/gateway/fwd"
 	g "proxy-server/gateway/globals"
 	"proxy-server/gateway/log"
-	"proxy-server/gateway/report"
 	"proxy-server/gateway/web"
 	"proxy-server/utils"
 	"sync"
@@ -21,28 +20,16 @@ import (
 
 	"github.com/google/uuid"
 
-	"github.com/joho/godotenv"
-
 	_ "net/http/pprof"
 )
 
-type server struct {
-}
-
-func New() *server {
-	return &server{}
-}
-
-func (s *server) Run() (err error) {
+func Start() (err error) {
 
 	// 初始化
-	err = s.init()
-	if err != nil {
-		return err
-	}
+	setup()
 
 	// 恢复服务状态
-	err = s.restore()
+	err = restore()
 	if err != nil {
 		return err
 	}
@@ -59,18 +46,28 @@ func (s *server) Run() (err error) {
 	defer close(fwdQuit)
 	go func() {
 		defer wg.Done()
-		err = s.startFwd(ctx)
+		err = startFwd(ctx)
 		fwdQuit <- err
 	}()
 
 	// 接口服务
 	wg.Add(1)
-	apiQuit := make(chan error, 1)
-	defer close(apiQuit)
+	webQuit := make(chan error, 1)
+	defer close(webQuit)
 	go func() {
 		defer wg.Done()
-		err := s.startWeb(ctx)
-		apiQuit <- err
+		err := startWeb(ctx)
+		webQuit <- err
+	}()
+
+	// 调度任务服务
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		err := startTask(ctx)
+		if err != nil {
+			slog.Error("调度任务服务异常退出", "err", err)
+		}
 	}()
 
 	// debug
@@ -89,7 +86,7 @@ func (s *server) Run() (err error) {
 
 	// 报告上线
 	slog.Info("报告服务上线")
-	err = report.Online(app.Name)
+	err = app.Online(app.Name)
 	if err != nil {
 		return fmt.Errorf("服务上线失败: %w", err)
 	}
@@ -100,7 +97,7 @@ func (s *server) Run() (err error) {
 		if err != nil {
 			slog.Warn("fwd 服务异常退出", "err", err)
 		}
-	case err := <-apiQuit:
+	case err := <-webQuit:
 		if err != nil {
 			slog.Warn("web 服务异常退出", "err", err)
 		}
@@ -113,13 +110,14 @@ func (s *server) Run() (err error) {
 		defer wg.Done()
 		// 报告下线
 		slog.Debug("报告服务下线")
-		err = report.Offline(app.Name)
+		err = app.Offline()
 		if err != nil {
 			slog.Error("服务下线失败", "err", err)
 		}
 
-		// 关闭 redis
+		// 关闭服务
 		g.ExitRedis()
+		g.ExitGeo()
 	}()
 
 	// 等待其它服务关闭
@@ -133,21 +131,14 @@ func (s *server) Run() (err error) {
 	return nil
 }
 
-func (s *server) init() error {
-
-	err := godotenv.Load()
-	if err != nil {
-		println("没有本地环境变量文件")
-	}
-
+func setup() {
 	log.Init()
 	env.Init()
 	g.InitRedis()
-
-	return nil
+	g.InitGeo()
 }
 
-func (s *server) restore() error {
+func restore() error {
 	var file = "proxy.lock"
 
 	bytes, err := os.ReadFile(file)
@@ -176,7 +167,7 @@ func (s *server) restore() error {
 	return nil
 }
 
-func (s *server) startFwd(ctx context.Context) error {
+func startFwd(ctx context.Context) error {
 	server := fwd.New()
 	go func() {
 		<-ctx.Done()
@@ -185,7 +176,7 @@ func (s *server) startFwd(ctx context.Context) error {
 	return server.Run()
 }
 
-func (s *server) startWeb(ctx context.Context) error {
+func startWeb(ctx context.Context) error {
 	server := web.New()
 	go func() {
 		<-ctx.Done()
@@ -196,3 +187,49 @@ func (s *server) startWeb(ctx context.Context) error {
 	}()
 	return server.Run()
 }
+
+func startTask(ctx context.Context) error {
+
+	// 维护一个修改表，每次提交后清空，每个节点的修改在同一阶段内只有一次
+	var lock = sync.Mutex{}
+	var updates = make([]*core.Edge, 0)
+	var updatesMap = make(map[int32]*core.Edge)
+	go func() {
+		for data := range app.EdgeUpdates {
+			lock.Lock()
+			if update, ok := updatesMap[data.Id]; ok {
+				if data.Status != update.Status {
+					// 如果状态发生变化，则更新
+					updatesMap[data.Id] = data
+					updates = append(updates, data)
+				}
+			} else {
+				updatesMap[data.Id] = data
+				updates = append(updates, data)
+			}
+			lock.Unlock()
+		}
+	}()
+
+	// 每 30 秒批量提交一次更新
+	var scheduler = time.Tick(30 * time.Second)
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-scheduler:
+			if len(updates) == 0 {
+				continue
+			}
+			lock.Lock()
+			err := app.Update(updates)
+			if err != nil {
+				slog.Error("调度更新任务失败", "err", err)
+			} else {
+				clear(updates)
+				updates = updates[:0]
+			}
+			lock.Unlock()
+		}
+	}
+}

gateway/globals/geo.go

@@ -0,0 +1,23 @@
+package globals
+
+import "github.com/lionsoul2014/ip2region/binding/golang/xdb"
+
+var Geo *xdb.Searcher
+
+func InitGeo() {
+	var err error
+
+	buff, err := xdb.LoadContentFromFile("ip2region.xdb")
+	if err != nil {
+		panic("读取 geo 数据库文件失败")
+	}
+
+	Geo, err = xdb.NewWithBuffer(buff)
+	if err != nil {
+		panic("初始化 geo 查询工具失败")
+	}
+}
+
+func ExitGeo() {
+	Geo.Close()
+}

gateway/web/handlers/auth.go

@@ -6,10 +6,7 @@ import (
 	"proxy-server/gateway/core"
 )
 
-type PermitReq struct {
-	Id int32 `json:"id"`
-	core.Permit
-}
+type PermitReq []core.PermitDef
 
 func Permit(ctx *fiber.Ctx) (err error) {
 
@@ -20,14 +17,14 @@ func Permit(ctx *fiber.Ctx) (err error) {
 	}
 
 	// 获取请求参数
-	req, err := core.Decrypt[[]PermitReq](&sec, app.PlatformSecret)
+	req, err := core.Decrypt[PermitReq](&sec, app.PlatformSecret)
 	if err != nil {
 		return err
 	}
 
 	// 保存授权配置
 	for _, permit := range *req {
-		app.Permits.Store(permit.Id, &permit.Permit)
+		app.StorePermit(&permit)
 	}
 
 	return nil

gateway/web/handlers/info.go

@@ -16,8 +16,7 @@ type InfoResp struct {
 	DataConnections int `json:"data_connections"`
 
 	// Edges []EdgeResp `json:"edges"`
-	Assigns map[uint16]int32       `json:"assigns"`
-	Edges   map[int32]uint16       `json:"edges"`
+	Edges   map[int32]*core.Edge   `json:"edges"`
 	Permits map[int32]*core.Permit `json:"permits"`
 }
 
@@ -29,30 +28,9 @@ type EdgeResp struct {
 
 func Info(c *fiber.Ctx) error {
 
-	// var edges = make([]EdgeResp, 0)
-	// app.Edges.Range(func(id int32, port uint16) bool {
-	// 	permit, ok := app.Permits.Load(id)
-	// 	if !ok {
-	// 		return true
-	// 	}
-	//
-	// 	edges = append(edges, EdgeResp{
-	// 		Id:     id,
-	// 		Port:   port,
-	// 		Permit: permit,
-	// 	})
-	// 	return true
-	// })
-
-	var assigns = make(map[uint16]int32)
-	app.Assigns.Range(func(port uint16, id int32) bool {
-		assigns[port] = id
-		return true
-	})
-
-	var edges = make(map[int32]uint16)
-	app.Edges.Range(func(id int32, port uint16) bool {
-		edges[id] = port
+	var edges = make(map[int32]*core.Edge)
+	app.Edges.Range(func(id int32, edge *core.Edge) bool {
+		edges[id] = edge
 		return true
 	})
 
@@ -69,7 +47,6 @@ func Info(c *fiber.Ctx) error {
 		UserConnections: int(app.UserConnWg.Count()),
 		CtrlConnections: int(app.CtrlConnWg.Count()),
 		DataConnections: int(app.DataConnWg.Count()),
-		Assigns:         assigns,
 		Edges:           edges,
 		Permits:         permits,
 	})

go.mod

@@ -17,8 +17,10 @@ require (
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/klauspost/compress v1.18.0 // indirect
+	github.com/lionsoul2014/ip2region/binding/golang v0.0.0-20250508043914-ed57fa5c5274 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/mattn/go-runewidth v0.0.16 // indirect
+	github.com/mitchellh/go-homedir v1.1.0 // indirect
 	github.com/rivo/uniseg v0.4.7 // indirect
 	github.com/valyala/bytebufferpool v1.0.0 // indirect
 	github.com/valyala/fasthttp v1.59.0 // indirect

go.sum

@@ -1,4 +1,5 @@
 github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7XdTA=
+github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA=
 github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
 github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
 github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
@@ -14,6 +15,9 @@ github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+
 github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
 github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
 github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
+github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
+github.com/lionsoul2014/ip2region/binding/golang v0.0.0-20250508043914-ed57fa5c5274 h1:Vslec/nYvO2TdLdhwex8/1x64OZoQNsUzG79WABQaWg=
+github.com/lionsoul2014/ip2region/binding/golang v0.0.0-20250508043914-ed57fa5c5274/go.mod h1:C5LA5UO2ZXJrLaPLYtE1wUJMiyd/nwWaCO5cw/2pSHs=
 github.com/lmittmann/tint v1.0.7 h1:D/0OqWZ0YOGZ6AyC+5Y2kD8PBEzBk6rFHVSfOqCkF9Y=
 github.com/lmittmann/tint v1.0.7/go.mod h1:HIS3gSy7qNwGCj+5oRjAutErFBl4BzdQP6cJZ0NfMwE=
 github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE=
@@ -22,27 +26,35 @@ github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWE
 github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
 github.com/mattn/go-runewidth v0.0.16 h1:E5ScNMtiwvlvB5paMFdw9p4kSQzbXFikJ5SQO6TULQc=
 github.com/mattn/go-runewidth v0.0.16/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
+github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y=
+github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/redis/go-redis/v9 v9.8.0 h1:q3nRvjrlge/6UD7eTu/DSg2uYiU2mCL0G/uzBWqhicI=
 github.com/redis/go-redis/v9 v9.8.0/go.mod h1:huWgSWd8mW6+m0VPhJjSSQ+d6Nh1VICQ6Q5lHuCH/Iw=
 github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
 github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ=
+github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
 github.com/soheilhy/cmux v0.1.5 h1:jjzc5WVemNEDTLwv9tlmemhC73tI08BNOIGwBOo10Js=
 github.com/soheilhy/cmux v0.1.5/go.mod h1:T7TcVDs9LWfQgPlPsdngu6I6QIoyIFZDDC6sNE1GqG0=
 github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
 github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
 github.com/valyala/fasthttp v1.59.0 h1:Qu0qYHfXvPk1mSLNqcFtEk6DpxgA26hy6bmydotDpRI=
+github.com/valyala/fasthttp v1.59.0/go.mod h1:GTxNb9Bc6r2a9D0TWNSPwDz78UxnTGBViY3xZNEqyYU=
 github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU=
+github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.37.0 h1:1zLorHbz+LYj7MQlSf1+2tPIIgibq2eL5xkrGk6f+2c=
+golang.org/x/net v0.37.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.31.0 h1:ioabZlmFYtWhL+TRYpcnNlLwhyxaM9kWTDEmfnprqik=
+golang.org/x/sys v0.31.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY=
+golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=

utils/conn.go

@@ -0,0 +1,35 @@
+package utils
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"syscall"
+)
+
+func WarpConnErr(err error) (bool, error) {
+	if err == nil {
+		return true, nil
+	}
+
+	var opErr *net.OpError
+	switch {
+
+	case errors.Is(err, net.ErrClosed):
+		return true, errors.New("连接已关闭")
+	case errors.Is(err, io.EOF):
+		return true, errors.New("连接已结束")
+
+	case errors.As(err, &opErr):
+		switch {
+		case
+			errors.Is(opErr.Err, syscall.WSAECONNRESET), errors.Is(opErr.Err, syscall.WSAECONNABORTED),
+			errors.Is(opErr.Err, syscall.ECONNRESET), errors.Is(opErr.Err, syscall.ECONNABORTED):
+			return false, errors.New("连接已重置或中止")
+		case opErr.Timeout():
+			return false, errors.New("连接已超时")
+		}
+	}
+	return false, fmt.Errorf("连接发生未处理的错误: %w", err)
+}

utils/utils.go

@@ -1,8 +1,10 @@
 package utils
 
 import (
+	"fmt"
 	"io"
 	"log/slog"
+	"runtime"
 )
 
 func ReadByte(reader io.Reader) (byte, error) {
@@ -29,10 +31,15 @@ func ReadBuffer(reader io.Reader, size int) ([]byte, error) {
 
 // Close 关闭对象，传入值绝对不能为 nil
 func Close(v any) {
-	if v, ok := v.(io.Closer); ok {
-		err := v.Close()
+	if c, ok := v.(io.Closer); ok {
+		err := c.Close()
 		if err != nil {
-			slog.Warn("对象关闭失败", "err", err)
+			var logger = slog.Default()
+			_, file, line, ok := runtime.Caller(1)
+			if ok {
+				logger = logger.With("source", fmt.Sprintf("%s:%d", file, line))
+			}
+			logger.Warn("对象关闭失败", "err", err)
 		}
 	}
 }