From bee423fe5150b0d318cf4f980b7439153f84cec1 Mon Sep 17 00:00:00 2001 From: Rogee Date: Fri, 9 Oct 2026 05:39:41 +0800 Subject: [PATCH] feat(controlplane): outbound gateway WS channel replaces inbound HTTP client --- go.mod | 2 + go.sum | 4 + internal/controlplane/api/douyin.go | 2 +- internal/controlplane/api/environments.go | 91 ++++- internal/controlplane/api/event_channel.go | 255 ------------ .../controlplane/api/event_channel_test.go | 254 ------------ .../controlplane/api/event_listener_worker.go | 74 ++-- internal/controlplane/api/gateway_channel.go | 375 ++++++++++++++++++ .../controlplane/api/gateway_channel_test.go | 320 +++++++++++++++ .../api/gateway_channel_test_kit_test.go | 254 ++++++++++++ internal/controlplane/api/gateway_health.go | 83 ---- .../controlplane/api/gateway_health_test.go | 273 ------------- .../controlplane/api/hub_native_unit_test.go | 84 ---- internal/controlplane/app/app.go | 21 +- internal/controlplane/app/app_test.go | 2 + internal/controlplane/gatewayclient/client.go | 51 --- internal/controlplane/workers/runtime.go | 16 - 17 files changed, 1090 insertions(+), 1071 deletions(-) delete mode 100644 internal/controlplane/api/event_channel.go delete mode 100644 internal/controlplane/api/event_channel_test.go create mode 100644 internal/controlplane/api/gateway_channel.go create mode 100644 internal/controlplane/api/gateway_channel_test.go create mode 100644 internal/controlplane/api/gateway_channel_test_kit_test.go delete mode 100644 internal/controlplane/api/gateway_health.go delete mode 100644 internal/controlplane/api/gateway_health_test.go delete mode 100644 internal/controlplane/api/hub_native_unit_test.go delete mode 100644 internal/controlplane/gatewayclient/client.go diff --git a/go.mod b/go.mod index a52e81d..07b7064 100644 --- a/go.mod +++ b/go.mod @@ -14,6 +14,7 @@ require ( require ( github.com/andybalholm/brotli v1.2.2 // indirect + github.com/fasthttp/websocket v1.5.12 // indirect github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/go-viper/mapstructure/v2 v2.4.0 // indirect github.com/gofiber/schema v1.8.3 // indirect @@ -29,6 +30,7 @@ require ( github.com/pelletier/go-toml/v2 v2.2.4 // indirect github.com/philhofer/fwd v1.2.0 // indirect github.com/sagikazarmark/locafero v0.11.0 // indirect + github.com/savsgio/gotils v0.0.0-20240704082632-aef3928b8a38 // indirect github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 // indirect github.com/spf13/afero v1.15.0 // indirect github.com/spf13/cast v1.10.0 // indirect diff --git a/go.sum b/go.sum index a038df9..4b7fd9d 100644 --- a/go.sum +++ b/go.sum @@ -4,6 +4,8 @@ github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6N github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/fasthttp/websocket v1.5.12 h1:e4RGPpWW2HTbL3zV0Y/t7g0ub294LkiuXXUuTOUInlE= +github.com/fasthttp/websocket v1.5.12/go.mod h1:I+liyL7/4moHojiOgUOIKEWm9EIxHqxZChS+aMFltyg= github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8= github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0= github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k= @@ -55,6 +57,8 @@ github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/f github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= github.com/sagikazarmark/locafero v0.11.0 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc= github.com/sagikazarmark/locafero v0.11.0/go.mod h1:nVIGvgyzw595SUSUE6tvCp3YYTeHs15MvlmU87WwIik= +github.com/savsgio/gotils v0.0.0-20240704082632-aef3928b8a38 h1:D0vL7YNisV2yqE55+q0lFuGse6U8lxlg7fYTctlT5Gc= +github.com/savsgio/gotils v0.0.0-20240704082632-aef3928b8a38/go.mod h1:sM7Mt7uEoCeFSCBM+qBrqvEo+/9vdmj19wzp3yzUhmg= github.com/shamaton/msgpack/v3 v3.2.0 h1:1q2Ms+MWmuRju+PuDMSFDB7p7621npeX4zprJN5Zck8= github.com/shamaton/msgpack/v3 v3.2.0/go.mod h1:sgBYvEiyz8JR1NC3yGRoPVME9xXovpnh3l/plW1nfRo= github.com/sirupsen/logrus v1.10.0 h1:T8MxJJXVZkfcC5zSRMRAg2F8+lxjmUCGGWPzFxO+Msc= diff --git a/internal/controlplane/api/douyin.go b/internal/controlplane/api/douyin.go index f73f7c0..d7ed73c 100644 --- a/internal/controlplane/api/douyin.go +++ b/internal/controlplane/api/douyin.go @@ -108,7 +108,7 @@ type douyinMessageHistory struct { func (browser douyinGatewayBrowser) request() (douyinGatewayRequest, error) { environment := browser.environment - if browser.gateway.Endpoint == "" || browser.gateway.Token == "" || !accountRunnable(environment) || + if browser.gateway.ID <= 0 || !accountRunnable(environment) || !gatewayGenerationIDPattern.MatchString(environment.AccountID) || !gatewayGenerationIDPattern.MatchString(environment.Alias) || !gatewayGenerationIDPattern.MatchString(environment.Exit.ID) || environment.Exit.HealthStatus != "healthy" || environment.RuntimeCleanupPending || !gatewayGenerationIDPattern.MatchString(environment.RuntimeID) || diff --git a/internal/controlplane/api/environments.go b/internal/controlplane/api/environments.go index b955c19..8ebfbb1 100644 --- a/internal/controlplane/api/environments.go +++ b/internal/controlplane/api/environments.go @@ -1,7 +1,6 @@ package api import ( - "bytes" "context" "encoding/json" "errors" @@ -9,10 +8,10 @@ import ( "io" "net/http" "regexp" + "strconv" "strings" "time" - "git.ipao.vip/rogee/creator-hub/internal/controlplane/gatewayclient" hub "git.ipao.vip/rogee/creator-hub/internal/environment" "github.com/gofiber/fiber/v3" "github.com/sirupsen/logrus" @@ -21,12 +20,14 @@ import ( // HubStore 是控制面编排所需的存储能力；生产实现为 *hub.Store，测试使用内存桩。 type HubStore interface { LockResources(ctx context.Context, aliases, exitIDs []string, _ ...[]string) (func(), error) - CreateGateway(ctx context.Context, name, endpoint, token string) (hub.Gateway, error) - UpdateGateway(ctx context.Context, currentName, name, endpoint, token string) (hub.Gateway, error) + CreateGateway(ctx context.Context, name string) (hub.Gateway, error) + UpdateGateway(ctx context.Context, currentName, name string) (hub.Gateway, error) + ResetGatewayKey(ctx context.Context, name string) (hub.Gateway, error) ListGateways(ctx context.Context) ([]hub.Gateway, error) GetGateway(ctx context.Context, name string) (hub.Gateway, error) DeleteGateway(ctx context.Context, name string) error - RecordGatewayCheck(ctx context.Context, name, status, reason string) (hub.Gateway, error) + ListGatewayTasksPage(ctx context.Context, gatewayID int64, page, pageSize int) (hub.GatewayTaskPage, error) + CountPendingGatewayTasks(ctx context.Context, gatewayID int64) (int, error) ListEnvs(ctx context.Context) ([]hub.Env, error) GetEnv(ctx context.Context, alias string) (hub.Env, error) CreateNetworkExit(ctx context.Context, exit hub.NetworkExit) (hub.NetworkExit, error) @@ -88,13 +89,16 @@ const ( largeGatewayResponseLimit = 16 << 20 ) -// gatewayCall 调用某个网关的 /v1 路由；ok 为 false 时 status/body 携带网关错误。 +// gatewayCall 经出站通道以任务形式调用某个网关的 /v1 路由； +// 网关离线时任务立即终态化并返回明确错误，不伪装成功。 func gatewayCall(ctx context.Context, target hub.Gateway, method, path string, body any, timeout time.Duration) (status int, responseBody []byte, err error) { - return gatewayclient.Call(ctx, target, method, path, body, timeout) + return gatewayCallWithLimit(ctx, target, method, path, body, timeout, defaultGatewayResponseLimit) } +// gatewayCallWithLimit 的 responseLimit 仅用于兼容原 HTTP 直连签名的响应上限语义； +// 出站通道以帧承载响应，平台侧读上限由 WS 库统一约束。 func gatewayCallWithLimit(ctx context.Context, target hub.Gateway, method, path string, body any, timeout time.Duration, responseLimit int) (status int, responseBody []byte, err error) { - return gatewayclient.CallWithLimit(ctx, target, method, path, body, timeout, responseLimit) + return dispatchGatewayTask(ctx, target, method, path, body, timeout, true) } func gatewayCreatePayload(environment hub.EnvironmentContext, _ string, networkExit gatewayNetworkExit) map[string]any { @@ -325,6 +329,11 @@ type runtimeStatus struct { ProxyReady bool `json:"proxy_ready"` } +type gatewayView struct { + hub.Gateway + PendingTasks int `json:"pending_tasks"` +} + type envView struct { hub.Env State string `json:"state"` @@ -473,40 +482,61 @@ func registerHubWithNetwork(app *fiber.App, store HubStore, probe NetworkExitPro if err != nil { return hubError(c, err) } - return c.JSON(gateways) + views := make([]gatewayView, 0, len(gateways)) + for _, gateway := range gateways { + pending, pendingErr := store.CountPendingGatewayTasks(c.Context(), gateway.ID) + if pendingErr != nil { + return hubError(c, pendingErr) + } + views = append(views, gatewayView{Gateway: gateway, PendingTasks: pending}) + } + return c.JSON(views) }) app.Post("/api/gateways", func(c fiber.Ctx) error { input := struct { - Name string `json:"name"` - Endpoint string `json:"endpoint"` - Token string `json:"token"` + Name string `json:"name"` }{} if err := decodeHubJSON(c, &input); err != nil { return hubError(c, err) } - gateway, err := store.CreateGateway(c.Context(), input.Name, input.Endpoint, input.Token) + gateway, err := store.CreateGateway(c.Context(), input.Name) if err != nil { return hubError(c, err) } - probeGatewayAsync(c.Context(), store, gateway) return c.Status(fiber.StatusCreated).JSON(gateway) }) app.Put("/api/gateways/:name", func(c fiber.Ctx) error { input := struct { - Name string `json:"name"` - Endpoint string `json:"endpoint"` - Token string `json:"token"` + Name string `json:"name"` }{} if err := decodeHubJSON(c, &input); err != nil { return hubError(c, err) } - gateway, err := store.UpdateGateway(c.Context(), c.Params("name"), input.Name, input.Endpoint, input.Token) + gateway, err := store.UpdateGateway(c.Context(), c.Params("name"), input.Name) if err != nil { return hubError(c, err) } - probeGatewayAsync(c.Context(), store, gateway) return c.JSON(gateway) }) + app.Post("/api/gateways/:name/reset-key", func(c fiber.Ctx) error { + gateway, err := store.ResetGatewayKey(c.Context(), c.Params("name")) + if err != nil { + return hubError(c, err) + } + return c.JSON(gateway) + }) + app.Get("/api/gateways/:name/tasks", func(c fiber.Ctx) error { + gateway, err := store.GetGateway(c.Context(), c.Params("name")) + if err != nil { + return hubError(c, err) + } + page, pageSize := pageParams(c) + tasks, err := store.ListGatewayTasksPage(c.Context(), gateway.ID, page, pageSize) + if err != nil { + return hubError(c, err) + } + return c.JSON(tasks) + }) app.Delete("/api/gateways/:name", func(c fiber.Ctx) error { if err := store.DeleteGateway(c.Context(), c.Params("name")); err != nil { return hubError(c, err) @@ -2019,7 +2049,7 @@ func actionForEnvironment(actionName string, environment hub.EnvironmentContext) } func decodeHubJSON(c fiber.Ctx, destination any) error { - decoder := json.NewDecoder(bytes.NewReader(c.Body())) + decoder := json.NewDecoder(strings.NewReader(string(c.Body()))) decoder.DisallowUnknownFields() if err := decoder.Decode(destination); err != nil { return hub.ErrInvalid @@ -2030,6 +2060,27 @@ func decodeHubJSON(c fiber.Ctx, destination any) error { return nil } +func pageParams(c fiber.Ctx) (int, int) { + page := max(fiberQueryInt(c, "page", 1), 1) + pageSize := fiberQueryInt(c, "page_size", 25) + if pageSize < 1 || pageSize > 100 { + pageSize = 25 + } + return page, pageSize +} + +func fiberQueryInt(c fiber.Ctx, key string, fallback int) int { + raw := c.Query(key, "") + if raw == "" { + return fallback + } + value, err := strconv.Atoi(raw) + if err != nil { + return fallback + } + return value +} + func hubError(c fiber.Ctx, err error) error { var failure gatewayFailure if errors.As(err, &failure) { diff --git a/internal/controlplane/api/event_channel.go b/internal/controlplane/api/event_channel.go deleted file mode 100644 index 9fecf28..0000000 --- a/internal/controlplane/api/event_channel.go +++ /dev/null @@ -1,255 +0,0 @@ -package api - -import ( - "context" - "errors" - "fmt" - "net/http" - "net/url" - "strings" - "sync" - "sync/atomic" - "time" - - "git.ipao.vip/rogee/creator-hub/internal/creator" - "git.ipao.vip/rogee/creator-hub/internal/environment" - "github.com/gorilla/websocket" - "github.com/sirupsen/logrus" -) - -type channelMessage struct { - Type string `json:"type"` - Subscription string `json:"subscription"` - Error string `json:"error"` - Deliveries []creator.ListenerDelivery `json:"deliveries"` -} -type eventSubscription struct { - hub *gatewayEventChannel - id, alias string - messages chan channelMessage - done chan struct{} - failErr error -} -type gatewayEventChannel struct { - target environment.Gateway - once sync.Once - connectErr error - conn *websocket.Conn - write sync.Mutex - mu sync.Mutex - subs map[string]*eventSubscription - done chan struct{} - finish sync.Once -} - -var eventChannels = struct { - sync.Mutex - items map[string]*gatewayEventChannel -}{items: map[string]*gatewayEventChannel{}} -var subscriptionSequence atomic.Uint64 - -func acquireEventSubscription(ctx context.Context, target environment.Gateway, params map[string]any) (*eventSubscription, error) { - key := target.Endpoint + "\n" + target.Token - eventChannels.Lock() - hub := eventChannels.items[key] - if hub != nil { - select { - case <-hub.done: - hub = nil - default: - } - } - if hub == nil { - hub = &gatewayEventChannel{target: target, subs: map[string]*eventSubscription{}, done: make(chan struct{})} - eventChannels.items[key] = hub - } - alias, _ := params["alias"].(string) - sub := &eventSubscription{hub: hub, id: fmt.Sprint(subscriptionSequence.Add(1)), alias: alias, messages: make(chan channelMessage, 64), done: make(chan struct{})} - // Reserve the subscription before releasing the pool lock, so closing the - // previous last account cannot retire a channel another account is joining. - hub.mu.Lock() - hub.subs[sub.id] = sub - hub.mu.Unlock() - eventChannels.Unlock() - hub.once.Do(func() { - u, err := url.Parse(strings.TrimRight(target.Endpoint, "/") + "/v1/channel") - if err != nil { - hub.connectErr = err - hub.fail(err) - return - } - switch u.Scheme { - case "http": - u.Scheme = "ws" - case "https": - u.Scheme = "wss" - default: - hub.connectErr = errors.New("invalid gateway scheme") - hub.fail(hub.connectErr) - return - } - headers := http.Header{"Authorization": []string{"Bearer " + target.Token}} - dialer := websocket.Dialer{HandshakeTimeout: 10 * time.Second} - hub.conn, _, hub.connectErr = dialer.DialContext(ctx, u.String(), headers) - if hub.connectErr != nil { - hub.fail(hub.connectErr) - return - } - hub.conn.SetReadLimit(16 << 20) - hub.conn.SetPingHandler(func(data string) error { - _ = hub.conn.SetReadDeadline(time.Now().Add(45 * time.Second)) - return hub.conn.WriteControl(websocket.PongMessage, []byte(data), time.Now().Add(5*time.Second)) - }) - go hub.receive() - }) - if hub.connectErr != nil { - sub.close() - return nil, fmt.Errorf("gateway WS connection: %w", hub.connectErr) - } - value := make(map[string]any, len(params)+2) - for k, v := range params { - value[k] = v - } - value["type"] = "subscribe" - value["subscription"] = sub.id - if err := hub.send(value); err != nil { - sub.close() - return nil, err - } - select { - case message := <-sub.messages: - if message.Type != "subscribed" { - sub.close() - return nil, fmt.Errorf("gateway WS subscription: %s", message.Error) - } - return sub, nil - case <-hub.done: - sub.close() - return nil, errors.New("gateway WS disconnected") - case <-ctx.Done(): - sub.close() - return nil, ctx.Err() - } -} -func (h *gatewayEventChannel) send(value any) error { - h.write.Lock() - defer h.write.Unlock() - select { - case <-h.done: - return errors.New("gateway WS disconnected") - default: - } - if err := h.conn.SetWriteDeadline(time.Now().Add(5 * time.Second)); err != nil { - return err - } - if err := h.conn.WriteJSON(value); err != nil { - h.fail(err) - return err - } - return nil -} -func (h *gatewayEventChannel) receive() { - for { - _ = h.conn.SetReadDeadline(time.Now().Add(45 * time.Second)) - var message channelMessage - if err := h.conn.ReadJSON(&message); err != nil { - h.fail(err) - return - } - if message.Type != "subscribed" && message.Type != "deliveries" && message.Type != "error" { - h.fail(errors.New("invalid gateway WS message")) - return - } - if message.Type == "error" && message.Subscription == "" { - h.fail(errors.New(message.Error)) - return - } - h.mu.Lock() - sub := h.subs[message.Subscription] - h.mu.Unlock() - if sub == nil { - continue - } // A closed generation cannot receive delayed replies. - select { - case sub.messages <- message: - default: - // One slow account must not tear down the shared channel: retire only - // this subscription. The gateway keeps unacknowledged deliveries and - // replays them when the account session resubscribes. - spec := fmt.Errorf("event subscription queue overflow; the account session will resubscribe and replay unacknowledged deliveries") - logrus.WithFields(logrus.Fields{"alias": sub.alias, "subscription": sub.id}).Error("event subscription queue overflow; isolating subscription from shared channel") - sub.fail(spec) - } - } -} -func (h *gatewayEventChannel) fail(err error) { - h.finish.Do(func() { - close(h.done) - if h.conn != nil { - _ = h.conn.Close() - } - h.mu.Lock() - defer h.mu.Unlock() - for _, sub := range h.subs { - select { - case sub.messages <- channelMessage{Type: "error", Error: err.Error()}: - default: - } - } - }) -} - -// fail retires one subscription without touching the shared channel: the -// account session restarts and the gateway replays unacknowledged deliveries. -// Only the receive goroutine calls this, so close(done) happens exactly once. -func (s *eventSubscription) fail(err error) { - if err == nil { - err = errors.New("event subscription failed") - } - s.hub.mu.Lock() - delete(s.hub.subs, s.id) - s.hub.mu.Unlock() - s.failErr = err - close(s.done) -} -func (s *eventSubscription) poll(ctx context.Context, acks []string) ([]creator.ListenerDelivery, error) { - if len(acks) > 0 { - if err := s.hub.send(map[string]any{"type": "ack", "alias": s.alias, "subscription": s.id, "delivery_ids": acks}); err != nil { - return nil, err - } - } - select { - case message := <-s.messages: - if message.Type != "deliveries" || len(message.Deliveries) == 0 { - return nil, fmt.Errorf("gateway event delivery: %s", message.Error) - } - for _, d := range message.Deliveries { - if err := d.Validate(); err != nil { - return nil, err - } - } - return message.Deliveries, nil - case <-s.done: - return nil, s.failErr - case <-s.hub.done: - return nil, errors.New("gateway WS disconnected") - case <-ctx.Done(): - return nil, ctx.Err() - } -} -func (s *eventSubscription) close() { - _ = s.hub.send(map[string]any{"type": "unsubscribe", "alias": s.alias, "subscription": s.id}) - key := s.hub.target.Endpoint + "\n" + s.hub.target.Token - eventChannels.Lock() - s.hub.mu.Lock() - delete(s.hub.subs, s.id) - empty := len(s.hub.subs) == 0 - if empty && eventChannels.items[key] == s.hub { - delete(eventChannels.items, key) - } - s.hub.mu.Unlock() - eventChannels.Unlock() - if empty { - s.hub.fail(errors.New("gateway event channel stopped")) - } -} diff --git a/internal/controlplane/api/event_channel_test.go b/internal/controlplane/api/event_channel_test.go deleted file mode 100644 index a3c7249..0000000 --- a/internal/controlplane/api/event_channel_test.go +++ /dev/null @@ -1,254 +0,0 @@ -package api - -import ( - "context" - "errors" - "fmt" - "git.ipao.vip/rogee/creator-hub/internal/environment" - "github.com/gorilla/websocket" - "net/http" - "net/http/httptest" - "strings" - "sync" - "sync/atomic" - "testing" - "time" -) - -func TestEventChannelSharesConnectionAndAcknowledges(t *testing.T) { - var connections atomic.Int32 - acks := make(chan struct{}, 2) - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.URL.Path != "/v1/channel" || r.Header.Get("Authorization") != "Bearer test" { - t.Errorf("unexpected channel request") - w.WriteHeader(400) - return - } - c, err := (&websocket.Upgrader{}).Upgrade(w, r, nil) - if err != nil { - t.Error(err) - return - } - defer c.Close() - connections.Add(1) - for { - var v map[string]any - if c.ReadJSON(&v) != nil { - return - } - switch v["type"] { - case "subscribe": - _ = c.WriteJSON(map[string]any{"type": "subscribed", "subscription": v["subscription"]}) - _ = c.WriteJSON(map[string]any{"type": "deliveries", "subscription": v["subscription"], "deliveries": []any{map[string]any{"kind": "open", "delivery_id": v["alias"]}}}) - case "ack": - acks <- struct{}{} - } - } - })) - defer server.Close() - ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) - defer cancel() - target := environment.Gateway{Endpoint: server.URL, Token: "test"} - first, err := acquireEventSubscription(ctx, target, map[string]any{"alias": "a"}) - if err != nil { - t.Fatal(err) - } - defer first.close() - second, err := acquireEventSubscription(ctx, target, map[string]any{"alias": "b"}) - if err != nil { - t.Fatal(err) - } - defer second.close() - if connections.Load() != 1 { - t.Fatalf("connections=%d", connections.Load()) - } - items, err := first.poll(ctx, nil) - if err != nil || items[0].DeliveryID != "a" { - t.Fatalf("%+v %v", items, err) - } - items, err = second.poll(ctx, nil) - if err != nil || items[0].DeliveryID != "b" { - t.Fatalf("%+v %v", items, err) - } - short, c := context.WithTimeout(ctx, 30*time.Millisecond) - defer c() - _, _ = first.poll(short, []string{"a"}) - select { - case <-acks: - case <-ctx.Done(): - t.Fatal("ACK not received") - } -} -func TestEventChannelRejectsHandshakeAndSubscriptionTimeout(t *testing.T) { - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(503) })) - defer server.Close() - for _, endpoint := range []string{server.URL, "invalid://gateway", "http://%"} { - ctx, cancel := context.WithTimeout(context.Background(), time.Second) - _, err := acquireEventSubscription(ctx, environment.Gateway{Endpoint: endpoint}, map[string]any{"alias": "a"}) - cancel() - if err == nil { - t.Fatalf("invalid connection accepted: %s", endpoint) - } - } - waiting := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - c, e := (&websocket.Upgrader{}).Upgrade(w, r, nil) - if e != nil { - return - } - defer c.Close() - for { - var value map[string]any - if c.ReadJSON(&value) != nil { - return - } - } - })) - defer waiting.Close() - ctx, cancel := context.WithTimeout(context.Background(), 40*time.Millisecond) - defer cancel() - _, err := acquireEventSubscription(ctx, environment.Gateway{Endpoint: waiting.URL}, map[string]any{"alias": "a"}) - if !errors.Is(err, context.DeadlineExceeded) { - t.Fatalf("subscription wait did not honor cancellation: %v", err) - } -} - -func TestEventChannelReportsDisconnect(t *testing.T) { - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - c, e := (&websocket.Upgrader{}).Upgrade(w, r, nil) - if e != nil { - return - } - defer c.Close() - var v map[string]any - _ = c.ReadJSON(&v) - _ = c.WriteJSON(map[string]any{"type": "subscribed", "subscription": v["subscription"]}) - })) - defer server.Close() - ctx, cancel := context.WithTimeout(context.Background(), time.Second) - defer cancel() - sub, err := acquireEventSubscription(ctx, environment.Gateway{Endpoint: server.URL}, map[string]any{"alias": "a"}) - if err == nil { - defer sub.close() - _, err = sub.poll(ctx, nil) - } - if err == nil { - t.Fatal("disconnect hidden") - } -} - -func TestEventChannelSubscriptionOverflowIsolatesAccounts(t *testing.T) { - var connections atomic.Int32 - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - c, err := (&websocket.Upgrader{}).Upgrade(w, r, nil) - if err != nil { - t.Error(err) - return - } - defer c.Close() - connections.Add(1) - var write sync.Mutex - send := func(v map[string]any) { - write.Lock() - defer write.Unlock() - _ = c.WriteJSON(v) - } - for { - var v map[string]any - if c.ReadJSON(&v) != nil { - return - } - if v["type"] != "subscribe" { - continue - } - send(map[string]any{"type": "subscribed", "subscription": v["subscription"]}) - if v["alias"] == "flooded" { - for i := 0; i < 80; i++ { - send(map[string]any{"type": "deliveries", "subscription": v["subscription"], - "deliveries": []any{map[string]any{"kind": "open", "delivery_id": fmt.Sprintf("flood-%d", i)}}}) - } - } - if v["alias"] == "late" { - send(map[string]any{"type": "deliveries", "subscription": v["subscription"], - "deliveries": []any{map[string]any{"kind": "open", "delivery_id": "late-1"}}}) - } - } - })) - defer server.Close() - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - target := environment.Gateway{Endpoint: server.URL, Token: "test"} - flooded, err := acquireEventSubscription(ctx, target, map[string]any{"alias": "flooded"}) - if err != nil { - t.Fatal(err) - } - defer flooded.close() - late, err := acquireEventSubscription(ctx, target, map[string]any{"alias": "late"}) - if err != nil { - t.Fatalf("shared channel refused a joining account after another account overflowed: %v", err) - } - defer late.close() - items, err := late.poll(ctx, nil) - if err != nil || items[0].DeliveryID != "late-1" { - t.Fatalf("other account deliveries broke after one account overflowed: %+v %v", items, err) - } - if connections.Load() != 1 { - t.Fatalf("overflow tore down the shared channel, connections=%d", connections.Load()) - } - // Buffered deliveries sent before the overflow fired remain valid work; the - // subscription keeps reporting them until the buffer drains, then surfaces - // the explicit overflow error. - for { - _, err = flooded.poll(ctx, nil) - if err != nil { - if !strings.Contains(err.Error(), "overflow") { - t.Fatalf("overflowed subscription must fail with an explicit overflow error, got %v", err) - } - break - } - } -} - -func TestEventChannelClosingLastAccountDoesNotRetireJoiningAccount(t *testing.T) { - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - c, err := (&websocket.Upgrader{}).Upgrade(w, r, nil) - if err != nil { - return - } - defer c.Close() - for { - var value map[string]any - if c.ReadJSON(&value) != nil { - return - } - if value["type"] == "subscribe" { - _ = c.WriteJSON(map[string]any{"type": "subscribed", "subscription": value["subscription"]}) - } - } - })) - defer server.Close() - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - target := environment.Gateway{Endpoint: server.URL} - previous, err := acquireEventSubscription(ctx, target, map[string]any{"alias": "a"}) - if err != nil { - t.Fatal(err) - } - defer func() { previous.close() }() - for i := 0; i < 50; i++ { - type result struct { - sub *eventSubscription - err error - } - joined := make(chan result, 1) - go func() { - sub, err := acquireEventSubscription(ctx, target, map[string]any{"alias": "b"}) - joined <- result{sub, err} - }() - previous.close() - next := <-joined - if next.err != nil { - t.Fatalf("joining account interrupted at iteration %d: %v", i, next.err) - } - previous = next.sub - } -} diff --git a/internal/controlplane/api/event_listener_worker.go b/internal/controlplane/api/event_listener_worker.go index 5b54cf4..d4bc74b 100644 --- a/internal/controlplane/api/event_listener_worker.go +++ b/internal/controlplane/api/event_listener_worker.go @@ -36,17 +36,18 @@ type browserAliasResourceLocker interface { } type gatewayListenerSession struct { - target hub.Gateway - environment hub.EnvironmentContext - uid string - generation string - sessionID string - enabled bool - resources browserAliasResourceLocker - boundaryAt *time.Time - checkpoints map[string]string - coverFiles eventCoverFiles - subscription *eventSubscription + target hub.Gateway + environment hub.EnvironmentContext + uid string + generation string + sessionID string + enabled bool + resources browserAliasResourceLocker + boundaryAt *time.Time + checkpoints map[string]string + coverFiles eventCoverFiles + session *gatewaySession + events <-chan EventBatch } func (s *gatewayListenerSession) request(ctx context.Context, action string) ([]creator.ListenerDelivery, error) { @@ -110,27 +111,44 @@ func (s *gatewayListenerSession) request(ctx context.Context, action string) ([] return nil, creator.ErrInvalid } func (s *gatewayListenerSession) Start(ctx context.Context) ([]creator.ListenerDelivery, error) { + // 先订阅事件队列再启动监听任务，保证网关首批投递有接收者； + // Start 失败时反注册，避免会话里残留无消费者的事件队列。 + session := sessionByGateway(s.target.ID) + if session == nil { + return nil, hub.ErrGatewayOffline + } + queue := session.subscribeEvents(s.environment.Alias) if _, err := s.request(ctx, "start"); err != nil { + session.unsubscribeEvents(s.environment.Alias) return nil, err } - params := gatewayGenerationPayload(s.environment) - params["alias"], params["uid"] = s.environment.Alias, s.uid - var err error - params["listener_generation"] = s.generation - params["session_id"] = s.sessionID - subscribeCtx, cancel := context.WithTimeout(ctx, 10*time.Second) - defer cancel() - s.subscription, err = acquireEventSubscription(subscribeCtx, s.target, params) - if err != nil { - cleanupCtx, cleanupCancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) - defer cleanupCancel() - _, cleanupErr := s.request(cleanupCtx, "stop") - return nil, errors.Join(err, cleanupErr) - } + s.session = session + s.events = queue return nil, nil } func (s *gatewayListenerSession) Poll(ctx context.Context, ack []string) ([]creator.ListenerDelivery, error) { - return s.subscription.poll(ctx, ack) + if len(ack) > 0 { + if err := s.session.acknowledgeEvents(s.environment.Alias, ack); err != nil { + return nil, err + } + } + select { + case batch := <-s.events: + if batch.Subscription == "" { + // 队列溢出信号：本账号订阅被隔离，批次仍在网关待重推。 + return nil, errors.New("event subscription queue overflow; the account session will resubscribe and replay unacknowledged deliveries") + } + for _, d := range batch.Deliveries { + if err := d.Validate(); err != nil { + return nil, err + } + } + return batch.Deliveries, nil + case <-s.session.done: + return nil, errors.New("gateway session disconnected") + case <-ctx.Done(): + return nil, ctx.Err() + } } func (s *gatewayListenerSession) enrich(ctx context.Context, items []creator.ListenerDelivery) ([]creator.ListenerDelivery, error) { if s.coverFiles != nil { @@ -154,8 +172,8 @@ func (s *gatewayListenerSession) enrich(ctx context.Context, items []creator.Lis return items, nil } func (s *gatewayListenerSession) Stop(ctx context.Context) error { - if s.subscription != nil { - s.subscription.close() + if s.session != nil { + s.session.unsubscribeEvents(s.environment.Alias) } _, err := s.request(ctx, "stop") return err diff --git a/internal/controlplane/api/gateway_channel.go b/internal/controlplane/api/gateway_channel.go new file mode 100644 index 0000000..1eb0377 --- /dev/null +++ b/internal/controlplane/api/gateway_channel.go @@ -0,0 +1,375 @@ +package api + +// 平台侧网关出站通道：网关主动连入的 WS 会话、统一任务模型（持久化 + 下发 + 结果等待） +// 与事件流路由。方向反转后平台不再拨号连网关；设计见 docs/gateway-outbound-channel-plan.md。 + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sync" + "time" + + "git.ipao.vip/rogee/creator-hub/internal/creator" + hub "git.ipao.vip/rogee/creator-hub/internal/environment" + "github.com/fasthttp/websocket" + "github.com/gofiber/fiber/v3" + "github.com/sirupsen/logrus" +) + +// gatewayChannelStore 是出站通道所需的存储能力；生产实现为 *hub.Store，测试用内存桩。 +type gatewayChannelStore interface { + CreateGatewayTask(ctx context.Context, gatewayID int64, method, path string, payload []byte, timeout time.Duration) (hub.GatewayTask, error) + DispatchGatewayTask(ctx context.Context, taskID int64) (bool, error) + FinishGatewayTask(ctx context.Context, taskID int64, status string, responseStatus int, taskError string, responseBytes int64) (bool, error) + InterruptDispatchedTasks(ctx context.Context, gatewayID int64, reason string) error + NextPendingGatewayTasks(ctx context.Context, gatewayID int64, limit int) ([]hub.GatewayTask, error) + ResetGatewayConnections(ctx context.Context) error + TouchGatewaySession(ctx context.Context, gatewayID int64, version string) error + SetGatewayOnline(ctx context.Context, gatewayID int64, online bool) error +} + +// gatewayChannelStoreHolder 由应用启动时注入生产 store；测试可直接替换。 +var gatewayChannelStoreHolder gatewayChannelStore + +// SetGatewayChannelStore 注入通道存储实现。 +func SetGatewayChannelStore(store gatewayChannelStore) { gatewayChannelStoreHolder = store } + +func channelStore() gatewayChannelStore { + if gatewayChannelStoreHolder == nil { + panic("gateway channel store is not configured") + } + return gatewayChannelStoreHolder +} + +// TaskResult 是任务 result 帧的载荷。 +type TaskResult struct { + Status int `json:"status"` + Body []byte `json:"body"` + Error string `json:"error"` +} + +// EventBatch 是网关事件投递帧的载荷；语义沿用原 /v1/channel deliveries。 +type EventBatch struct { + Subscription string `json:"subscription"` + Alias string `json:"alias"` + Deliveries []creator.ListenerDelivery `json:"deliveries"` +} + +// gatewaySession 是一个网关的活动出站连接。 +type gatewaySession struct { + gateway hub.Gateway + conn *websocket.Conn + write sync.Mutex + done chan struct{} + finish sync.Once + + mu sync.Mutex + waiter chan TaskResult // 当前在等结果的同步任务（单连接串行下发场景按 id 匹配） + events map[string]chan EventBatch // alias -> 事件队列 +} + +var gatewaySessions = struct { + sync.Mutex + items map[int64]*gatewaySession +}{items: map[int64]*gatewaySession{}} + +func sessionByGateway(gatewayID int64) *gatewaySession { + gatewaySessions.Lock() + defer gatewaySessions.Unlock() + return gatewaySessions.items[gatewayID] +} + +func (s *gatewaySession) send(value any) error { + s.write.Lock() + defer s.write.Unlock() + select { + case <-s.done: + return errors.New("gateway session disconnected") + default: + } + if err := s.conn.SetWriteDeadline(time.Now().Add(10 * time.Second)); err != nil { + return err + } + return s.conn.WriteJSON(value) +} + +type taskFrame struct { + Type string `json:"type"` + ID int64 `json:"id"` + Method string `json:"method"` + Path string `json:"path"` + Payload json.RawMessage `json:"payload"` +} + +type resultFrame struct { + Type string `json:"type"` + ID int64 `json:"id"` + TaskResult +} + +// dispatchGatewayTask 创建持久化任务；网关在线时立即下发并（按需）等待结果。 +// wait=false 仅入队/补发，由通道下发器异步推进。 +func dispatchGatewayTask(ctx context.Context, gateway hub.Gateway, method, path string, body any, timeout time.Duration, wait bool) (int, []byte, error) { + payload, err := json.Marshal(body) + if err != nil { + return 0, nil, fmt.Errorf("encode gateway task payload: %w", err) + } + task, err := channelStore().CreateGatewayTask(ctx, gateway.ID, method, path, payload, timeout) + if err != nil { + return 0, nil, fmt.Errorf("persist gateway task: %w", err) + } + session := sessionByGateway(gateway.ID) + if session == nil { + // 同步等待的任务立即终态化；异步任务保持 pending 由下发器补发。 + if wait { + if _, markErr := channelStore().FinishGatewayTask(ctx, task.ID, hub.TaskFailed, 0, "网关离线", 0); markErr != nil { + logrus.WithError(markErr).Error("persist gateway offline task failure failed") + } + } + return 0, nil, hub.ErrGatewayOffline + } + result, err := session.execute(ctx, task, wait) + if err != nil { + return 0, nil, err + } + return result.Status, result.Body, nil +} + +// execute 下发单个任务并等待结果；下发失败立即终态化，等待按任务超时。 +func (s *gatewaySession) execute(ctx context.Context, task hub.GatewayTask, wait bool) (*TaskResult, error) { + var waiter chan TaskResult + if _, err := channelStore().DispatchGatewayTask(ctx, task.ID); err != nil { + return nil, fmt.Errorf("mark gateway task dispatched: %w", err) + } + frame := taskFrame{Type: "task", ID: task.ID, Method: task.Method, Path: task.Path, Payload: task.Payload} + if wait { + // 先建 waiter 再下发：网关可能秒回 result，晚建会丢结果导致同步调用假超时。 + s.mu.Lock() + s.waiter = make(chan TaskResult, 1) + waiter = s.waiter + s.mu.Unlock() + defer func() { + s.mu.Lock() + if s.waiter == waiter { + s.waiter = nil + } + s.mu.Unlock() + }() + } + if err := s.send(frame); err != nil { + _, _ = channelStore().FinishGatewayTask(ctx, task.ID, hub.TaskFailed, 0, "任务下发失败："+err.Error(), 0) + return nil, fmt.Errorf("send gateway task: %w", err) + } + if !wait { + return &TaskResult{}, nil + } + waitCtx, cancel := context.WithTimeout(ctx, time.Duration(task.TimeoutSeconds)*time.Second) + defer cancel() + for { + select { + case r := <-waiter: + if r.Error != "" { + _, _ = channelStore().FinishGatewayTask(ctx, task.ID, hub.TaskFailed, r.Status, r.Error, int64(len(r.Body))) + return nil, fmt.Errorf("gateway task %d %s %s failed: %s", task.ID, task.Method, task.Path, r.Error) + } + _, _ = channelStore().FinishGatewayTask(ctx, task.ID, hub.TaskSucceeded, r.Status, "", int64(len(r.Body))) + return &r, nil + case <-s.done: + _, _ = channelStore().FinishGatewayTask(ctx, task.ID, hub.TaskFailed, 0, "网关连接中断，结果未确认", 0) + return nil, errors.New("gateway session disconnected") + case <-waitCtx.Done(): + _, _ = channelStore().FinishGatewayTask(ctx, task.ID, hub.TaskFailed, 0, "网关任务等待超时", 0) + return nil, fmt.Errorf("gateway task %d timed out after %ds", task.ID, task.TimeoutSeconds) + } + } +} + +// receiveLoop 读取网关帧：result 唤醒等待者，event 路由到订阅队列，heartbeat 刷新健康。 +func (s *gatewaySession) receiveLoop(store gatewaySessionStore) { + for { + _ = s.conn.SetReadDeadline(time.Now().Add(60 * time.Second)) + var frame struct { + Type string `json:"type"` + ID int64 `json:"id"` + Version string `json:"version"` + *TaskResult + Event *EventBatch `json:"event"` + } + if err := s.conn.ReadJSON(&frame); err != nil { + s.fail(fmt.Sprintf("网关连接断开：%v", err), store) + return + } + switch frame.Type { + case "result": + s.mu.Lock() + waiter := s.waiter + s.mu.Unlock() + if waiter != nil && frame.TaskResult != nil { + select { + case waiter <- *frame.TaskResult: + default: + } + } + case "heartbeat": + if err := store.TouchGatewaySession(context.Background(), s.gateway.ID, frame.Version); err != nil { + logrus.WithError(err).Warn("gateway heartbeat persistence failed") + } + case "event": + if frame.Event != nil { + s.deliverEvent(*frame.Event) + } + default: + s.fail("网关通道消息类型无效:"+frame.Type, store) + return + } + } +} + +func (s *gatewaySession) deliverEvent(batch EventBatch) { + s.mu.Lock() + queue := s.events[batch.Alias] + s.mu.Unlock() + if queue == nil { + // 无订阅者的投递直接确认丢弃：worker 注册订阅后才允许网关启动监听， + // 该场景只出现在 worker 刚失败重启的间隙，批次仍留在网关待重推。 + logrus.WithFields(logrus.Fields{"gateway": s.gateway.Name, "alias": batch.Alias}).Warn("gateway event batch has no subscriber; asking gateway to retain it") + _ = s.send(map[string]any{"type": "event_nack", "subscription": batch.Subscription, "alias": batch.Alias}) + return + } + select { + case queue <- batch: + default: + // 单账号消费过载只隔离该订阅；批次未确认，网关保留并重推。 + logrus.WithFields(logrus.Fields{"gateway": s.gateway.Name, "alias": batch.Alias}).Error("gateway event queue overflow; isolating subscription") + s.removeEventQueue(batch.Alias) + s.failSubscription(batch.Alias, queue) + } +} + +func (s *gatewaySession) failSubscription(alias string, queue chan EventBatch) { + queue <- EventBatch{Subscription: "", Alias: alias} +} + +func (s *gatewaySession) removeEventQueue(alias string) { + s.mu.Lock() + delete(s.events, alias) + s.mu.Unlock() +} + +func (s *gatewaySession) fail(reason string, store gatewaySessionStore) { + s.finish.Do(func() { + close(s.done) + _ = s.conn.Close() + gatewaySessions.Lock() + if gatewaySessions.items[s.gateway.ID] == s { + delete(gatewaySessions.items, s.gateway.ID) + } + gatewaySessions.Unlock() + logrus.WithFields(logrus.Fields{"gateway": s.gateway.Name, "reason": reason}).Error("gateway session closed") + if store != nil { + _ = store.InterruptDispatchedTasks(context.Background(), s.gateway.ID, reason) + _ = store.SetGatewayOnline(context.Background(), s.gateway.ID, false) + } + }) +} + +// subscribeEvents 注册某账号事件队列；订阅先于监听启动任务，保证网关首批投递不丢失。 +func (s *gatewaySession) subscribeEvents(alias string) <-chan EventBatch { + queue := make(chan EventBatch, 64) + s.mu.Lock() + s.events[alias] = queue + s.mu.Unlock() + return queue +} + +func (s *gatewaySession) unsubscribeEvents(alias string) { + s.mu.Lock() + queue := s.events[alias] + delete(s.events, alias) + s.mu.Unlock() + if queue != nil { + close(queue) + } +} + +func (s *gatewaySession) acknowledgeEvents(alias string, deliveryIDs []string) error { + return s.send(map[string]any{"type": "event_ack", "alias": alias, "delivery_ids": deliveryIDs}) +} + +// HandleGatewayChannel 对外暴露网关出站通道 WS 路由注册。 +func HandleGatewayChannel(app *fiber.App, store *hub.Store) { + handleGatewayChannel(app, store) +} + +// gatewaySessionStore 是会话读循环所需的存储能力；生产实现为 *hub.Store。 +type gatewaySessionStore interface { + TouchGatewaySession(ctx context.Context, gatewayID int64, version string) error + SetGatewayOnline(ctx context.Context, gatewayID int64, online bool) error + InterruptDispatchedTasks(ctx context.Context, gatewayID int64, reason string) error + GetGatewayByAccessKey(ctx context.Context, accessKey string) (hub.Gateway, error) +} + +// serveGatewaySession 完成认证、会话注册与读循环；fiber 路由与测试共用。 +func serveGatewaySession(conn *websocket.Conn, authorization string, store gatewaySessionStore) { + gateway, err := store.GetGatewayByAccessKey(context.Background(), authorization) + if err != nil { + logrus.Warn("gateway agent rejected: invalid access key") + _ = conn.Close() + return + } + session := &gatewaySession{gateway: gateway, conn: conn, done: make(chan struct{}), events: map[string]chan EventBatch{}} + gatewaySessions.Lock() + gatewaySessions.items[gateway.ID] = session + gatewaySessions.Unlock() + if err := store.SetGatewayOnline(context.Background(), gateway.ID, true); err != nil { + logrus.WithError(err).Warn("gateway online persistence failed") + } + logrus.WithFields(logrus.Fields{"gateway": gateway.Name, "gateway_id": gateway.ID}).Info("gateway agent connected") + session.receiveLoop(store) +} + +// handleGatewayChannel 处理网关出站 WS 升级与会话生命周期。 +func handleGatewayChannel(app *fiber.App, store *hub.Store) { + app.Get("/v1/agent", func(c fiber.Ctx) error { + upgrader := websocket.FastHTTPUpgrader{} + return upgrader.Upgrade(c.RequestCtx(), func(conn *websocket.Conn) { + authorization := string(c.RequestCtx().Request.Header.Peek("Authorization")) + serveGatewaySession(conn, authorization, store) + }) + }) +} + +// RunGatewayTaskDispatcher 周期扫描 pending 任务，对在线网关按 id 序补发。 +func RunGatewayTaskDispatcher(ctx context.Context, store *hub.Store) { + for ctx.Err() == nil { + gateways, err := store.ListGateways(ctx) + if err != nil { + logrus.WithError(err).Error("list gateways for task dispatch failed") + } else { + for _, gateway := range gateways { + session := sessionByGateway(gateway.ID) + if session == nil { + continue + } + tasks, err := store.NextPendingGatewayTasks(ctx, gateway.ID, 64) + if err != nil { + logrus.WithError(err).WithField("gateway", gateway.Name).Error("read pending gateway tasks failed") + continue + } + for _, task := range tasks { + if _, err := session.execute(ctx, task, false); err != nil { + logrus.WithError(err).WithField("gateway", gateway.Name).Error("dispatch pending gateway task failed") + break + } + } + } + } + select { + case <-ctx.Done(): + return + case <-time.After(2 * time.Second): + } + } +} diff --git a/internal/controlplane/api/gateway_channel_test.go b/internal/controlplane/api/gateway_channel_test.go new file mode 100644 index 0000000..dcae7f7 --- /dev/null +++ b/internal/controlplane/api/gateway_channel_test.go @@ -0,0 +1,320 @@ +package api + +import ( + "context" + "encoding/json" + "errors" + "testing" + "time" + + hub "git.ipao.vip/rogee/creator-hub/internal/environment" + "github.com/fasthttp/websocket" +) + +func TestGatewayChannelDispatchesTaskAndWaitsForResult(t *testing.T) { + store := newMemoryChannelStore(newTestGateway(1, "gw-main", "test-access-key-0001")) + server := (&gatewayChannelTestServer{store: store, URL: startChannelServer(t, store)}) + + SetGatewayChannelStore(store) + + fake := connectFakeGateway(t, server.URL, "test-access-key-0001", func(task taskFrame) *TaskResult { + if task.Method != "POST" || task.Path != "/v1/browsers/alias/douyin/get" { + return &TaskResult{Status: 404, Error: "unexpected path " + task.Path} + } + return &TaskResult{Status: 200, Body: []byte(`{"ok":true}`)} + }) + _ = fake + <-waitForSession(t, 1) + + status, body, err := gatewayCall(context.Background(), newTestGateway(1, "gw-main", "unused"), "POST", "/v1/browsers/alias/douyin/get", map[string]any{"url": "https://x"}, 5*time.Second) + if err != nil || status != 200 || string(body) != `{"ok":true}` { + t.Fatalf("gateway call failed: status=%d body=%s err=%v", status, body, err) + } +} + +func waitForSession(t *testing.T, gatewayID int64) <-chan struct{} { + t.Helper() + done := make(chan struct{}) + go func() { + defer close(done) + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if sessionByGateway(gatewayID) != nil { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Error("gateway session never registered") + }() + return done +} + +func TestGatewayChannelRejectsInvalidAccessKey(t *testing.T) { + store := newMemoryChannelStore(newTestGateway(1, "gw-main", "test-access-key-0001")) + server := (&gatewayChannelTestServer{store: store, URL: startChannelServer(t, store)}) + + header := map[string][]string{"Authorization": {"Bearer wrong-key"}} + conn, _, err := websocket.DefaultDialer.Dial(serverWSURL(server.URL)+"/v1/agent", header) + if err != nil { + t.Fatalf("dial before auth rejection is transport level: %v", err) + } + defer conn.Close() + // 服务端应立即关闭连接：读操作收尾即失败。 + _ = conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + if _, _, err = conn.ReadMessage(); err == nil { + t.Fatal("invalid access key must not establish a session") + } + if sessionByGateway(1) != nil { + t.Fatal("rejected gateway must not register a session") + } +} + +func TestGatewayChannelOfflineFailsSyncTaskImmediately(t *testing.T) { + store := newMemoryChannelStore(newTestGateway(2, "gw-offline", "test-access-key-0002")) + SetGatewayChannelStore(store) + + _, _, err := gatewayCall(context.Background(), newTestGateway(2, "gw-offline", "unused"), "POST", "/v1/browsers", nil, time.Second) + if !errors.Is(err, hub.ErrGatewayOffline) { + t.Fatalf("offline gateway must fail immediately: %v", err) + } + tasks, _ := store.NextPendingGatewayTasks(context.Background(), 2, 10) + if len(tasks) != 0 { + t.Fatalf("offline sync task must be terminal, pending=%d", len(tasks)) + } +} + +func TestGatewayChannelTaskTimeoutMarksFailure(t *testing.T) { + store := newMemoryChannelStore(newTestGateway(3, "gw-slow", "test-access-key-0003")) + server := (&gatewayChannelTestServer{store: store, URL: startChannelServer(t, store)}) + SetGatewayChannelStore(store) + + // 假网关收到任务后不回结果。 + _ = connectFakeGateway(t, server.URL, "test-access-key-0003", func(task taskFrame) *TaskResult { return nil }) + <-waitForSession(t, 3) + + _, _, err := gatewayCall(context.Background(), newTestGateway(3, "gw-slow", "unused"), "POST", "/v1/browsers/alias/stop", nil, 300*time.Millisecond) + if err == nil { + t.Fatal("unanswered task must time out") + } + var offline hub.GatewayTask + found := false + for _, task := range func() map[int64]*hub.GatewayTask { + store.mu.Lock() + defer store.mu.Unlock() + return store.tasks + }() { + if task.Error == "网关任务等待超时" { + offline = *task + found = true + } + } + if !found || offline.Status != hub.TaskFailed { + t.Fatalf("timeout must be recorded as failed task: %+v", offline) + } +} + +func TestGatewayChannelDisconnectInterruptsInflightTask(t *testing.T) { + store := newMemoryChannelStore(newTestGateway(4, "gw-drop", "test-access-key-0004")) + server := (&gatewayChannelTestServer{store: store, URL: startChannelServer(t, store)}) + SetGatewayChannelStore(store) + + fake := connectFakeGateway(t, server.URL, "test-access-key-0004", func(task taskFrame) *TaskResult { return nil }) + <-waitForSession(t, 4) + + callErr := make(chan error, 1) + go func() { + _, _, err := gatewayCall(context.Background(), newTestGateway(4, "gw-drop", "unused"), "POST", "/v1/browsers/alias/start", nil, 10*time.Second) + callErr <- err + }() + deadline := time.After(2 * time.Second) + var taskID int64 + select { + case taskID = <-fake.taskIDs: + case <-deadline: + t.Fatal("task was never dispatched") + } + // 网关断开：等待方必须以连接中断收尾。 + _ = fake.conn.Close() + select { + case err := <-callErr: + if err == nil { + t.Fatal("disconnect must fail the waiting task") + } + case <-time.After(2 * time.Second): + t.Fatal("waiting task did not finish after disconnect") + } + _ = taskID + deadline = time.After(2 * time.Second) + for { + store.mu.Lock() + interrupted := store.interrupted + store.mu.Unlock() + if interrupted >= 1 { + break + } + select { + case <-deadline: + t.Fatal("disconnect must interrupt dispatched tasks") + default: + time.Sleep(5 * time.Millisecond) + } + } +} + +func TestGatewayChannelEventRoutingAndAck(t *testing.T) { + store := newMemoryChannelStore(newTestGateway(5, "gw-events", "test-access-key-0005")) + server := (&gatewayChannelTestServer{store: store, URL: startChannelServer(t, store)}) + + acks := make(chan []string, 4) + var conn *websocket.Conn + ackReady := make(chan struct{}) + go func() { + header := map[string][]string{"Authorization": {"Bearer test-access-key-0005"}} + c, _, err := websocket.DefaultDialer.Dial(serverWSURL(server.URL)+"/v1/agent", header) + if err != nil { + t.Errorf("fake gateway dial failed: %v", err) + return + } + conn = c + go func() { + for { + var frame struct { + Type string `json:"type"` + Alias string `json:"alias"` + DeliveryIDs []string `json:"delivery_ids"` + } + if err := c.ReadJSON(&frame); err != nil { + return + } + if frame.Type == "event_ack" { + acks <- frame.DeliveryIDs + } + } + }() + close(ackReady) + }() + <-ackReady + defer func() { _ = conn.Close() }() + <-waitForSession(t, 5) + session := sessionByGateway(5) + queue := session.subscribeEvents("alias-a") + defer session.unsubscribeEvents("alias-a") + + // 平台侧尚未消费前，网关先推两批：队列应缓冲而非丢弃。 + batch1 := eventBatchFor("alias-a", "d-1") + batch2 := eventBatchFor("alias-a", "d-2") + if err := conn.WriteJSON(map[string]any{"type": "event", "event": batch1}); err != nil { + t.Fatal(err) + } + if err := conn.WriteJSON(map[string]any{"type": "event", "event": batch2}); err != nil { + t.Fatal(err) + } + + got := <-queue + if got.Deliveries[0].DeliveryID != "d-1" { + t.Fatalf("first batch lost: %+v", got) + } + if err := session.acknowledgeEvents("alias-a", []string{"d-1"}); err != nil { + t.Fatalf("ack send failed: %v", err) + } + select { + case ids := <-acks: + if len(ids) != 1 || ids[0] != "d-1" { + t.Fatalf("ack frame mismatch: %v", ids) + } + case <-time.After(2 * time.Second): + t.Fatal("ack frame never reached the gateway") + } + + // 心跳帧刷新会话健康。 + if err := conn.WriteJSON(map[string]any{"type": "heartbeat", "version": "2026.10.0"}); err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + store.mu.Lock() + heartbeats := store.heartbeats + store.mu.Unlock() + if heartbeats > 0 { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatal("heartbeat was not persisted") +} + +func TestGatewayChannelUnsubscribedEventBatchIsNacked(t *testing.T) { + store := newMemoryChannelStore(newTestGateway(6, "gw-nack", "test-access-key-0006")) + server := (&gatewayChannelTestServer{store: store, URL: startChannelServer(t, store)}) + + nacks := make(chan string, 4) + ready := make(chan struct{}) + var conn *websocket.Conn + go func() { + header := map[string][]string{"Authorization": {"Bearer test-access-key-0006"}} + c, _, err := websocket.DefaultDialer.Dial(serverWSURL(server.URL)+"/v1/agent", header) + if err != nil { + t.Errorf("dial failed: %v", err) + return + } + conn = c + go func() { + for { + var frame struct { + Type string `json:"type"` + Alias string `json:"alias"` + } + if err := c.ReadJSON(&frame); err != nil { + return + } + if frame.Type == "event_nack" { + nacks <- frame.Alias + } + } + }() + close(ready) + }() + <-ready + defer func() { _ = conn.Close() }() + <-waitForSession(t, 6) + + if err := conn.WriteJSON(map[string]any{"type": "event", "event": eventBatchFor("ghost", "d-9")}); err != nil { + t.Fatal(err) + } + select { + case alias := <-nacks: + if alias != "ghost" { + t.Fatalf("nack alias mismatch: %s", alias) + } + case <-time.After(2 * time.Second): + t.Fatal("unsubscribed batch was silently consumed instead of nacked") + } +} + +func TestGatewayChannelTaskPayloadRoundTrip(t *testing.T) { + store := newMemoryChannelStore(newTestGateway(7, "gw-payload", "test-access-key-0007")) + server := (&gatewayChannelTestServer{store: store, URL: startChannelServer(t, store)}) + SetGatewayChannelStore(store) + + received := make(chan json.RawMessage, 1) + fake := connectFakeGateway(t, server.URL, "test-access-key-0007", func(task taskFrame) *TaskResult { + received <- task.Payload + return &TaskResult{Status: 204} + }) + _ = fake + <-waitForSession(t, 7) + status, _, err := gatewayCall(context.Background(), newTestGateway(7, "gw-payload", "unused"), "POST", "/v1/browsers", map[string]any{"alias": "a", "nested": map[string]int{"seed": 7}}, 5*time.Second) + if err != nil || status != 204 { + fake.mu.Lock() + t.Logf("diagnose: taskIDs=%d results=%#v", len(fake.taskIDs), fake.results) + fake.mu.Unlock() + t.Fatalf("payload task failed: status=%d err=%v", status, err) + } + var payload map[string]any + if err := json.Unmarshal(<-received, &payload); err != nil { + t.Fatalf("payload must stay valid JSON: %v", err) + } + if payload["alias"] != "a" { + t.Fatalf("payload content mismatch: %v", payload) + } +} diff --git a/internal/controlplane/api/gateway_channel_test_kit_test.go b/internal/controlplane/api/gateway_channel_test_kit_test.go new file mode 100644 index 0000000..ff0ceae --- /dev/null +++ b/internal/controlplane/api/gateway_channel_test_kit_test.go @@ -0,0 +1,254 @@ +package api + +// 网关出站通道测试基建：内存桩 store + 平台侧 WS 会话服务器 + 假网关客户端。 + +import ( + "context" + "encoding/json" + "net" + "net/http" + "strings" + "sync" + "testing" + "time" + + "git.ipao.vip/rogee/creator-hub/internal/creator" + hub "git.ipao.vip/rogee/creator-hub/internal/environment" + "github.com/fasthttp/websocket" + "github.com/gofiber/fiber/v3" +) + +type memoryChannelStore struct { + mu sync.Mutex + gateways map[string]hub.Gateway // access_key -> gateway + tasks map[int64]*hub.GatewayTask + taskSeq int64 + heartbeats int + onlineCalls int + interrupted int +} + +func newMemoryChannelStore(gateways ...hub.Gateway) *memoryChannelStore { + store := &memoryChannelStore{gateways: map[string]hub.Gateway{}, tasks: map[int64]*hub.GatewayTask{}} + for _, gateway := range gateways { + store.gateways[gateway.AccessKey] = gateway + } + return store +} + +func (s *memoryChannelStore) TouchGatewaySession(_ context.Context, gatewayID int64, version string) error { + s.mu.Lock() + defer s.mu.Unlock() + s.heartbeats++ + for key, gateway := range s.gateways { + if gateway.ID == gatewayID { + gateway.Version = version + gateway.Online = true + now := time.Now() + gateway.LastSeenAt = &now + s.gateways[key] = gateway + } + } + return nil +} + +func (s *memoryChannelStore) SetGatewayOnline(_ context.Context, gatewayID int64, online bool) error { + s.mu.Lock() + defer s.mu.Unlock() + s.onlineCalls++ + for key, gateway := range s.gateways { + if gateway.ID == gatewayID { + gateway.Online = online + s.gateways[key] = gateway + } + } + return nil +} + +func (s *memoryChannelStore) InterruptDispatchedTasks(_ context.Context, _ int64, _ string) error { + s.mu.Lock() + defer s.mu.Unlock() + s.interrupted++ + return nil +} + +func (s *memoryChannelStore) GetGatewayByAccessKey(_ context.Context, accessKey string) (hub.Gateway, error) { + s.mu.Lock() + defer s.mu.Unlock() + accessKey = strings.TrimPrefix(accessKey, "Bearer ") + if gateway, ok := s.gateways[accessKey]; ok { + return gateway, nil + } + return hub.Gateway{}, hub.ErrNotFound +} + +func (s *memoryChannelStore) CreateGatewayTask(_ context.Context, gatewayID int64, method, path string, payload []byte, timeout time.Duration) (hub.GatewayTask, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.taskSeq++ + timeoutSeconds := int(timeout / time.Second) + if timeoutSeconds < 1 { + timeoutSeconds = 1 + } + task := hub.GatewayTask{ID: s.taskSeq, GatewayID: gatewayID, Method: method, Path: path, + Payload: payload, Status: hub.TaskPending, TimeoutSeconds: timeoutSeconds} + s.tasks[task.ID] = &task + return task, nil +} + +func (s *memoryChannelStore) DispatchGatewayTask(_ context.Context, taskID int64) (bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + if task, ok := s.tasks[taskID]; ok && task.Status == hub.TaskPending { + task.Status = hub.TaskDispatched + return true, nil + } + return false, nil +} + +func (s *memoryChannelStore) FinishGatewayTask(_ context.Context, taskID int64, status string, responseStatus int, taskError string, responseBytes int64) (bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + if task, ok := s.tasks[taskID]; ok && (task.Status == hub.TaskDispatched || task.Status == hub.TaskPending) { + task.Status = status + task.ResponseStatus = responseStatus + task.Error = taskError + task.ResponseBytes = responseBytes + return true, nil + } + return false, nil +} + +func (s *memoryChannelStore) NextPendingGatewayTasks(_ context.Context, gatewayID int64, limit int) ([]hub.GatewayTask, error) { + s.mu.Lock() + defer s.mu.Unlock() + tasks := []hub.GatewayTask{} + for id := int64(1); id <= s.taskSeq && len(tasks) < limit; id++ { + if task, ok := s.tasks[id]; ok && task.GatewayID == gatewayID && task.Status == hub.TaskPending { + tasks = append(tasks, *task) + } + } + return tasks, nil +} + +func (s *memoryChannelStore) ResetGatewayConnections(_ context.Context) error { return nil } + +func (s *memoryChannelStore) CreateTaskRow(gatewayID int64, method, path string, payload []byte) hub.GatewayTask { + task, _ := s.CreateGatewayTask(context.Background(), gatewayID, method, path, payload, 30*time.Second) + return task +} + +// startChannelServer 启动与生产一致的 fiber WS 服务（真实 listener + FastHTTPUpgrader）。 +func startChannelServer(t *testing.T, store gatewaySessionStore) string { + t.Helper() + app := fiber.New() + app.Get("/v1/agent", func(c fiber.Ctx) error { + authorization := string(c.RequestCtx().Request.Header.Peek("Authorization")) + upgrader := websocket.FastHTTPUpgrader{} + return upgrader.Upgrade(c.RequestCtx(), func(conn *websocket.Conn) { + serveGatewaySession(conn, authorization, store) + }) + }) + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("channel listener: %v", err) + } + go func() { _ = app.Listener(listener, fiber.ListenConfig{DisableStartupMessage: true}) }() + t.Cleanup(func() { _ = listener.Close() }) + return "http://" + listener.Addr().String() +} + +// gatewayChannelTestServer 旧签名兼容：URL 指向真实 fiber listener。 +type gatewayChannelTestServer struct { + store *memoryChannelStore + URL string +} + +type fakeChannelGateway struct { + mu sync.Mutex + conn *websocket.Conn + taskIDs chan int64 + results map[int64]TaskResult // 已发送的 result，供测试诊断 +} + +// WriteJSON 串行化对连接的写，避免与任务 result/心跳并发写帧。 +func (g *fakeChannelGateway) WriteJSON(value any) error { + g.mu.Lock() + defer g.mu.Unlock() + return g.conn.WriteJSON(value) +} + +// connectFakeGateway 模拟网关出站连接：收到 task 帧后用 handler 生成 result 帧。 +func connectFakeGateway(t *testing.T, serverURL, accessKey string, handle func(task taskFrame) *TaskResult) *fakeChannelGateway { + t.Helper() + header := http.Header{"Authorization": []string{"Bearer " + accessKey}} + conn, _, err := websocket.DefaultDialer.Dial(serverWSURL(serverURL)+"/v1/agent", header) + if err != nil { + t.Fatalf("fake gateway dial failed: %v", err) + } + gateway := &fakeChannelGateway{conn: conn, taskIDs: make(chan int64, 8), results: map[int64]TaskResult{}} + go func() { + for { + var raw struct { + Type string `json:"type"` + ID int64 `json:"id"` + Method string `json:"method"` + Path string `json:"path"` + Payload json.RawMessage `json:"payload"` + } + if err := conn.ReadJSON(&raw); err != nil { + return + } + if raw.Type != "task" { + continue + } + select { + case gateway.taskIDs <- raw.ID: + default: + } + if result := handle(taskFrame{ID: raw.ID, Method: raw.Method, Path: raw.Path, Payload: raw.Payload}); result != nil { + gateway.mu.Lock() + writeErr := conn.WriteJSON(resultFrame{Type: "result", ID: raw.ID, TaskResult: *result}) + gateway.results[raw.ID] = *result + gateway.mu.Unlock() + if writeErr != nil { + t.Logf("fake gateway result write failed for task %d: %v", raw.ID, writeErr) + } + } + } + }() + // 平台 60s 读超时依赖心跳；假网关必须周期回心跳保持会话。 + tickerStop := make(chan struct{}) + go func() { + ticker := time.NewTicker(20 * time.Second) + defer ticker.Stop() + for { + select { + case <-ticker.C: + gateway.mu.Lock() + _ = conn.WriteJSON(map[string]any{"type": "heartbeat", "version": "test"}) + gateway.mu.Unlock() + case <-tickerStop: + return + } + } + }() + t.Cleanup(func() { close(tickerStop) }) + t.Cleanup(func() { _ = conn.Close() }) + return gateway +} + +func newTestGateway(id int64, name, accessKey string) hub.Gateway { + return hub.Gateway{ID: id, Name: name, AccessKey: accessKey} +} + +func eventBatchFor(alias, deliveryID string) EventBatch { + return EventBatch{Subscription: "sub-1", Alias: alias, Deliveries: []creator.ListenerDelivery{{ + DeliveryID: deliveryID, Kind: "open", + }}} +} + +// serverWSURL 把 http:// 地址转为 ws:// 地址。 +func serverWSURL(httpURL string) string { + return "ws://" + strings.TrimPrefix(httpURL, "http://") +} diff --git a/internal/controlplane/api/gateway_health.go b/internal/controlplane/api/gateway_health.go deleted file mode 100644 index 6c87300..0000000 --- a/internal/controlplane/api/gateway_health.go +++ /dev/null @@ -1,83 +0,0 @@ -package api - -import ( - "context" - "net/http" - "sync" - "time" - - hub "git.ipao.vip/rogee/creator-hub/internal/environment" - "github.com/sirupsen/logrus" -) - -// gatewayHealthProbeTimeout 是单次 /healthz 探活的超时；测试可临时替换。 -var gatewayHealthProbeTimeout = 5 * time.Second - -// gatewayHealthStore 是网关健康检查所需的存储能力；生产实现为 *hub.Store，测试使用内存桩。 -type gatewayHealthStore interface { - ListGateways(ctx context.Context) ([]hub.Gateway, error) - RecordGatewayCheck(ctx context.Context, name, status, reason string) (hub.Gateway, error) -} - -// classifyGatewayHealth 将一次 /healthz 探活结果归一为持久化状态与原因。 -// 连接失败或 5xx 判为 unhealthy；其余（含 401 令牌不匹配）说明进程可达，判为 healthy -// 并把异常状态写进 reason 供排查。 -func classifyGatewayHealth(status int, callErr error) (string, string) { - if callErr != nil { - reason := callErr.Error() - if len(reason) > 300 { - reason = reason[:300] - } - return "unhealthy", reason - } - if status >= 500 { - return "unhealthy", "gateway returned status " + http.StatusText(status) - } - if status >= 400 { - return "healthy", "gateway responded with status " + http.StatusText(status) - } - return "healthy", "" -} - -// RecordGatewayHealthChecks 执行一轮网关健康探活并落库；供 workers 定时任务与测试调用。 -func RecordGatewayHealthChecks(ctx context.Context, store *hub.Store) error { - return recordGatewayHealthChecks(ctx, store) -} - -// recordGatewayHealthChecks 对全部注册网关并发探活一次，每个网关只覆盖写入最近一次结果。 -func recordGatewayHealthChecks(ctx context.Context, store gatewayHealthStore) error { - gateways, err := store.ListGateways(ctx) - if err != nil { - return err - } - var wg sync.WaitGroup - for _, gateway := range gateways { - wg.Add(1) - go func() { - defer wg.Done() - probeAndRecordGateway(ctx, store, gateway) - }() - } - wg.Wait() - return nil -} - -// probeAndRecordGateway 对单个网关探活一次并覆盖写入最近一次结果；落库失败不阻断其余网关，仅记日志。 -func probeAndRecordGateway(ctx context.Context, store gatewayHealthStore, gateway hub.Gateway) { - status, _, callErr := gatewayCall(ctx, gateway, http.MethodGet, "/healthz", nil, gatewayHealthProbeTimeout) - healthStatus, reason := classifyGatewayHealth(status, callErr) - if _, recordErr := store.RecordGatewayCheck(ctx, gateway.Name, healthStatus, reason); recordErr != nil { - logrus.WithFields(logrus.Fields{ - "service": "control-plane", - "event_type": "gateway_health_check", - "gateway": gateway.Name, - "probed": healthStatus, - }).WithError(recordErr).Warn("gateway health check result was not persisted") - } -} - -// probeGatewayAsync 在后台对单个网关立即探活一次并落库，不阻塞当前请求； -// 用 WithoutCancel 脱离请求生命周期，请求结束后探活与落库仍会完成。 -func probeGatewayAsync(ctx context.Context, store gatewayHealthStore, gateway hub.Gateway) { - go probeAndRecordGateway(context.WithoutCancel(ctx), store, gateway) -} diff --git a/internal/controlplane/api/gateway_health_test.go b/internal/controlplane/api/gateway_health_test.go deleted file mode 100644 index e14f825..0000000 --- a/internal/controlplane/api/gateway_health_test.go +++ /dev/null @@ -1,273 +0,0 @@ -package api - -import ( - "context" - "errors" - "net/http" - "net/http/httptest" - "sync" - "testing" - "time" - - hub "git.ipao.vip/rogee/creator-hub/internal/environment" - "github.com/gofiber/fiber/v3" -) - -// setGatewayHealthProbeTimeout 临时替换探活超时，返回旧值供 restore 恢复。 -func setGatewayHealthProbeTimeout(d time.Duration) time.Duration { - old := gatewayHealthProbeTimeout - gatewayHealthProbeTimeout = d - return old -} - -// restoreGatewayHealthProbeTimeout 恢复探活超时旧值。 -func restoreGatewayHealthProbeTimeout(old time.Duration) func() { - return func() { gatewayHealthProbeTimeout = old } -} - -type recordedGatewayCheck struct { - name string - status string - reason string -} - -// healthMemoryStore 在 memoryStore 之上记录 RecordGatewayCheck 调用，供健康检查断言。 -type healthMemoryStore struct { - *memoryStore - mu sync.Mutex - checks []recordedGatewayCheck -} - -func (s *healthMemoryStore) RecordGatewayCheck(_ context.Context, name, status, reason string) (hub.Gateway, error) { - s.mu.Lock() - defer s.mu.Unlock() - s.checks = append(s.checks, recordedGatewayCheck{name: name, status: status, reason: reason}) - return hub.Gateway{Name: name, HealthStatus: status, LastCheckReason: reason}, nil -} - -func TestClassifyGatewayHealth(t *testing.T) { - for _, test := range []struct { - name string - status int - callErr error - wantStatus string - wantReasonHad bool // 是否应携带非空 reason - }{ - {name: "204 healthy without reason", status: http.StatusNoContent, wantStatus: "healthy", wantReasonHad: false}, - {name: "200 healthy without reason", status: http.StatusOK, wantStatus: "healthy", wantReasonHad: false}, - {name: "401 reachable but unauthorized", status: http.StatusUnauthorized, wantStatus: "healthy", wantReasonHad: true}, - {name: "404 reachable but misrouted", status: http.StatusNotFound, wantStatus: "healthy", wantReasonHad: true}, - {name: "503 unhealthy", status: http.StatusServiceUnavailable, wantStatus: "unhealthy", wantReasonHad: true}, - {name: "connection error unhealthy", callErr: errors.New(`Get "http://gw:8081/healthz": dial tcp: connection refused`), wantStatus: "unhealthy", wantReasonHad: true}, - } { - t.Run(test.name, func(t *testing.T) { - status, reason := classifyGatewayHealth(test.status, test.callErr) - if status != test.wantStatus { - t.Fatalf("classify status = %q, want %q (reason=%q)", status, test.wantStatus, reason) - } - if test.wantReasonHad != (reason != "") { - t.Fatalf("classify reason = %q, wantReasonHad=%v", reason, test.wantReasonHad) - } - }) - } -} - -func TestClassifyGatewayHealthTruncatesReason(t *testing.T) { - long := errors.New(string(make([]byte, 1024))) - _, reason := classifyGatewayHealth(0, long) - if len(reason) != 300 { - t.Fatalf("reason length = %d, want 300", len(reason)) - } -} - -func TestRecordGatewayHealthChecksPersistsLatestResult(t *testing.T) { - reachable := &fakeGateway{token: "token-a"} - reachableServer := httptest.NewServer(reachable.handler(t)) - t.Cleanup(reachableServer.Close) - - closedServer := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})) - closedServer.Close() // 端口已释放，保证连接失败 - - store := &healthMemoryStore{memoryStore: newMemoryStore()} - store.gateways["gw-live"] = hub.Gateway{Name: "gw-live", Endpoint: reachableServer.URL, Token: reachable.token} - store.gateways["gw-down"] = hub.Gateway{Name: "gw-down", Endpoint: closedServer.URL, Token: "unused-token"} - - if err := recordGatewayHealthChecks(context.Background(), store); err != nil { - t.Fatalf("record gateway health checks returned error: %v", err) - } - - byName := map[string]recordedGatewayCheck{} - for _, check := range store.checks { - byName[check.name] = check - } - if len(store.checks) != 2 { - t.Fatalf("expected one check per gateway, got %#v", store.checks) - } - if byName["gw-live"].status != "healthy" { - t.Fatalf("reachable gateway status = %q, want healthy", byName["gw-live"].status) - } - if byName["gw-live"].reason != "" { - t.Fatalf("reachable gateway reason = %q, want empty", byName["gw-live"].reason) - } - if byName["gw-down"].status != "unhealthy" || byName["gw-down"].reason == "" { - t.Fatalf("unreachable gateway = %#v, want unhealthy with reason", byName["gw-down"]) - } - if requests := reachable.recorded(); len(requests) != 1 || requests[0].path != "/healthz" { - t.Fatalf("reachable gateway probes = %#v, want exactly one /healthz", requests) - } -} - -func TestRecordGatewayHealthChecksEmptyListIsNoop(t *testing.T) { - store := &healthMemoryStore{memoryStore: newMemoryStore()} - if err := recordGatewayHealthChecks(context.Background(), store); err != nil { - t.Fatalf("empty list returned error: %v", err) - } - if len(store.checks) != 0 { - t.Fatalf("empty list recorded checks: %#v", store.checks) - } -} - -// 编译期约束：探活超时常量必须明显小于调度间隔，避免探活互相堆叠。 -func TestGatewayHealthProbeTimeoutBounded(t *testing.T) { - if gatewayHealthProbeTimeout > 10*time.Second { - t.Fatalf("probe timeout %v must stay below scheduling cadence", gatewayHealthProbeTimeout) - } -} - -// waitForGatewayCheck 轮询等待指定网关的立即探活结果落库；超时判失败。 -func waitForGatewayCheck(t *testing.T, store *healthMemoryStore, name string) recordedGatewayCheck { - t.Helper() - deadline := time.Now().Add(2 * time.Second) - for time.Now().Before(deadline) { - store.mu.Lock() - for _, check := range store.checks { - if check.name == name { - store.mu.Unlock() - return check - } - } - store.mu.Unlock() - time.Sleep(5 * time.Millisecond) - } - t.Fatalf("gateway %q health check was not recorded within deadline", name) - return recordedGatewayCheck{} -} - -// healthzProbeServer 返回校验 Bearer token 后立即响应 204 的 /healthz 测试服务。 -func healthzProbeServer(t *testing.T, token string, probeStarted chan struct{}, respond func(http.ResponseWriter)) *httptest.Server { - t.Helper() - server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { - if request.URL.Path != "/healthz" { - response.WriteHeader(http.StatusNotFound) - return - } - if request.Header.Get("Authorization") != "Bearer "+token { - response.WriteHeader(http.StatusUnauthorized) - return - } - if probeStarted != nil { - close(probeStarted) - } - if respond != nil { - respond(response) - return - } - response.WriteHeader(http.StatusNoContent) - })) - t.Cleanup(server.Close) - return server -} - -// TestCreateGatewayProbesHealthImmediately：注册网关成功后立即按注册的 endpoint 与 token -// 对 /healthz 探活并落库，结果为 healthy。 -func TestCreateGatewayProbesHealthImmediately(t *testing.T) { - store := &healthMemoryStore{memoryStore: newMemoryStore()} - probeStarted := make(chan struct{}) - server := healthzProbeServer(t, "unit-test-gateway-token", probeStarted, nil) - app := fiber.New() - registerHubWithNetwork(app, store, fakeExitProbe{}, func(hub.NetworkExitAccess) (string, error) { return "", nil }) - - response := do(app, http.MethodPost, "/api/gateways", - `{"name":"gw-new","endpoint":"`+server.URL+`","token":"unit-test-gateway-token"}`) - if response.Code != http.StatusCreated { - t.Fatalf("create gateway returned %d: %s", response.Code, response.Body.String()) - } - - select { - case <-probeStarted: - case <-time.After(2 * time.Second): - t.Fatal("gateway health probe was not triggered after create") - } - check := waitForGatewayCheck(t, store, "gw-new") - if check.status != "healthy" || check.reason != "" { - t.Fatalf("created gateway check = %#v, want healthy without reason", check) - } -} - -// TestCreateGatewayDoesNotBlockOnSlowGateway：网关探活被拖住时注册请求必须立即返回， -// 后台探活超时后把 unhealthy 结果落库。 -func TestCreateGatewayDoesNotBlockOnSlowGateway(t *testing.T) { - t.Cleanup(restoreGatewayHealthProbeTimeout(setGatewayHealthProbeTimeout(250 * time.Millisecond))) - store := &healthMemoryStore{memoryStore: newMemoryStore()} - server := healthzProbeServer(t, "unit-test-gateway-token", nil, func(response http.ResponseWriter) { - time.Sleep(800 * time.Millisecond) // 超过探活超时，若探活同步执行则注册请求会被拖住 - response.WriteHeader(http.StatusNoContent) - }) - app := fiber.New() - registerHubWithNetwork(app, store, fakeExitProbe{}, func(hub.NetworkExitAccess) (string, error) { return "", nil }) - - started := time.Now() - response := do(app, http.MethodPost, "/api/gateways", - `{"name":"gw-slow","endpoint":"`+server.URL+`","token":"unit-test-gateway-token"}`) - if response.Code != http.StatusCreated { - t.Fatalf("create gateway returned %d: %s", response.Code, response.Body.String()) - } - if elapsed := time.Since(started); elapsed > 500*time.Millisecond { - t.Fatalf("create gateway blocked %v on health probe", elapsed) - } - check := waitForGatewayCheck(t, store, "gw-slow") - if check.status != "unhealthy" || check.reason == "" { - t.Fatalf("slow gateway check = %#v, want unhealthy with reason", check) - } -} - -// TestUpdateGatewayProbesHealthImmediately：编辑网关成功后立即按新 endpoint 探活并覆盖落库。 -func TestUpdateGatewayProbesHealthImmediately(t *testing.T) { - store := &healthMemoryStore{memoryStore: newMemoryStore()} - store.gateways["gw-1"] = hub.Gateway{Name: "gw-1", Endpoint: "http://gw-old:8081", Token: "unit-test-gateway-token"} - server := healthzProbeServer(t, "unit-test-gateway-token", nil, nil) - app := fiber.New() - registerHubWithNetwork(app, store, fakeExitProbe{}, func(hub.NetworkExitAccess) (string, error) { return "", nil }) - - response := do(app, http.MethodPut, "/api/gateways/gw-1", - `{"name":"gw-1","endpoint":"`+server.URL+`","token":""}`) - if response.Code != http.StatusOK { - t.Fatalf("update gateway returned %d: %s", response.Code, response.Body.String()) - } - check := waitForGatewayCheck(t, store, "gw-1") - if check.status != "healthy" || check.reason != "" { - t.Fatalf("updated gateway check = %#v, want healthy without reason", check) - } -} - -// TestUpdateGatewayUnreachableRecordsUnhealthy：编辑指向不可达地址时保存照常成功， -// 后台探活把 unhealthy 结果落库供页面展示。 -func TestUpdateGatewayUnreachableRecordsUnhealthy(t *testing.T) { - t.Cleanup(restoreGatewayHealthProbeTimeout(setGatewayHealthProbeTimeout(300 * time.Millisecond))) - store := &healthMemoryStore{memoryStore: newMemoryStore()} - store.gateways["gw-1"] = hub.Gateway{Name: "gw-1", Endpoint: "http://gw-old:8081", Token: "unit-test-gateway-token"} - closedServer := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})) - closedServer.Close() // 端口已释放，保证连接失败 - app := fiber.New() - registerHubWithNetwork(app, store, fakeExitProbe{}, func(hub.NetworkExitAccess) (string, error) { return "", nil }) - - response := do(app, http.MethodPut, "/api/gateways/gw-1", - `{"name":"gw-1","endpoint":"`+closedServer.URL+`","token":""}`) - if response.Code != http.StatusOK { - t.Fatalf("update gateway returned %d: %s", response.Code, response.Body.String()) - } - check := waitForGatewayCheck(t, store, "gw-1") - if check.status != "unhealthy" || check.reason == "" { - t.Fatalf("unreachable gateway check = %#v, want unhealthy with reason", check) - } -} diff --git a/internal/controlplane/api/hub_native_unit_test.go b/internal/controlplane/api/hub_native_unit_test.go deleted file mode 100644 index 62d2088..0000000 --- a/internal/controlplane/api/hub_native_unit_test.go +++ /dev/null @@ -1,84 +0,0 @@ -package api - -import ( - "context" - "errors" - "net/http" - "net/http/httptest" - "testing" - - "git.ipao.vip/rogee/creator-hub/internal/creator" - hub "git.ipao.vip/rogee/creator-hub/internal/environment" -) - -func TestCreatorGatewayBrowserResolveAndSharePlatformValidation(t *testing.T) { - environment := hub.EnvironmentContext{Env: hub.Env{Alias: "account-1"}, RuntimeID: "runtime-1", RuntimeNetworkID: "network-1", BindingVersion: 3} - cases := []struct { - name string - status int - body string - wantURL string - wantErr bool - }{ - {name: "success", status: http.StatusOK, body: `{"url":"https://www.douyin.com/video/1"}`, wantURL: "https://www.douyin.com/video/1"}, - {name: "invalid json", status: http.StatusOK, body: `{`, wantErr: true}, - {name: "empty url", status: http.StatusOK, body: `{"url":""}`, wantErr: true}, - {name: "rejected", status: http.StatusBadGateway, body: `gateway down`, wantErr: true}, - } - for _, testCase := range cases { - t.Run(testCase.name, func(t *testing.T) { - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.URL.Path != "/v1/browsers/account-1/douyin/resolve" { - t.Fatalf("resolve path = %q", r.URL.Path) - } - w.WriteHeader(testCase.status) - _, _ = w.Write([]byte(testCase.body)) - })) - defer server.Close() - got, err := (creatorGatewayBrowser{gateway: hub.Gateway{Endpoint: server.URL}, environment: environment}).Resolve(context.Background(), "https://www.douyin.com/video/1") - if (err != nil) != testCase.wantErr || got != testCase.wantURL { - t.Fatalf("resolve = %q, err = %v", got, err) - } - }) - } - - for _, testCase := range []struct { - value string - platform string - wantErr bool - }{ - {value: "https://www.douyin.com/video/1", platform: creator.PlatformDouyin}, - {value: "https://v.douyin.com/abc", platform: creator.PlatformDouyin}, - {value: "https://www.xiaohongshu.com/explore/1", wantErr: true}, - {value: "https://xhslink.com/abc", wantErr: true}, - {value: "http://www.douyin.com/video/1", wantErr: true}, - {value: "https://user@www.douyin.com/video/1", wantErr: true}, - {value: "https://www.douyin.com:443/video/1", wantErr: true}, - {value: "https://www.douyin.com/video/1#part", wantErr: true}, - {value: "https://example.com/video/1", wantErr: true}, - } { - platform, err := competitorSharePlatform(testCase.value) - if testCase.wantErr { - if !errors.Is(err, creator.ErrInvalid) { - t.Errorf("competitorSharePlatform(%q) error = %v", testCase.value, err) - } - continue - } - if err != nil || platform != testCase.platform { - t.Errorf("competitorSharePlatform(%q) = %q, %v", testCase.value, platform, err) - } - } -} - -func TestRebindRecoveryErrorPreservesFailureStates(t *testing.T) { - sentinel := errors.New("rebind failed") - if got := rebindRecoveryError(true, sentinel); !errors.Is(got, sentinel) { - t.Fatalf("existing error = %v", got) - } - if got := rebindRecoveryError(false, nil); got == nil { - t.Fatal("incomplete recovery was reported as successful") - } - if got := rebindRecoveryError(true, nil); got != nil { - t.Fatalf("complete recovery error = %v", got) - } -} diff --git a/internal/controlplane/app/app.go b/internal/controlplane/app/app.go index c7fd7c2..e8fd0a0 100644 --- a/internal/controlplane/app/app.go +++ b/internal/controlplane/app/app.go @@ -102,11 +102,16 @@ func newCommand() *cobra.Command { defer close(listenerDone) api.RunAccountEventListeners(listenerContext, creatorStore, hubStore) }() - gatewayHealthContext, stopGatewayHealth := context.WithCancel(command.Context()) - gatewayHealthDone := make(chan struct{}) + // 出站通道：启动时清空上次进程遗留的在线标记，任务下发器补发 pending 任务。 + if err := hubStore.ResetGatewayConnections(command.Context()); err != nil { + logrus.WithError(err).Error("reset gateway connections failed") + } + api.SetGatewayChannelStore(hubStore) + taskDispatchContext, stopTaskDispatch := context.WithCancel(command.Context()) + taskDispatchDone := make(chan struct{}) go func() { - defer close(gatewayHealthDone) - workers.RunGatewayHealthCheck(gatewayHealthContext, hubStore) + defer close(taskDispatchDone) + api.RunGatewayTaskDispatcher(taskDispatchContext, hubStore) }() privateMessagesContext, stopPrivateMessages := context.WithCancel(command.Context()) privateMessagesDone := make(chan struct{}) @@ -121,12 +126,12 @@ func newCommand() *cobra.Command { stopCreatorScheduler() stopListeners() stopPrivateMessages() - stopGatewayHealth() + stopTaskDispatch() stopHeartbeat() <-creatorScheduleDone <-listenerDone <-privateMessagesDone - <-gatewayHealthDone + <-taskDispatchDone <-heartbeatDone return listenErr }, @@ -251,6 +256,10 @@ func newHandler(webDirectory, username, password string, phaseAStore *account.St IdleTimeout: 60 * time.Second, }) system.RegisterHealth(app, phaseAStore, hubStore, creatorStore) + // 网关出站通道在认证中间件之外：连接层用网关 access_key 自认证。 + if hubStore != nil { + api.HandleGatewayChannel(app, hubStore) + } app.Use(system.Authenticate(username, password)) if hubStore != nil { environmentsapi.Register(app, hubStore) diff --git a/internal/controlplane/app/app_test.go b/internal/controlplane/app/app_test.go index a7f958a..4a3f1f6 100644 --- a/internal/controlplane/app/app_test.go +++ b/internal/controlplane/app/app_test.go @@ -359,6 +359,8 @@ func controlPlaneRouteMatrix() []controlPlaneRouteCase { {http.MethodPost, "/api/gateways", "/api/gateways", "", http.StatusBadRequest}, {http.MethodPut, "/api/gateways/:name", "/api/gateways/missing", `{"name":"gw-missing","endpoint":"http://gw-missing:8081","token":""}`, http.StatusNotFound}, {http.MethodDelete, "/api/gateways/:name", "/api/gateways/missing", "", http.StatusNotFound}, + {http.MethodPost, "/api/gateways/:name/reset-key", "/api/gateways/missing/reset-key", "", http.StatusNotFound}, + {http.MethodGet, "/api/gateways/:name/tasks", "/api/gateways/missing/tasks", "", http.StatusNotFound}, {http.MethodGet, "/api/phase-a/accounts", "/api/phase-a/accounts", "", http.StatusOK}, {http.MethodGet, "/api/phase-a/accounts/:id", "/api/phase-a/accounts/missing", "", http.StatusNotFound}, diff --git a/internal/controlplane/gatewayclient/client.go b/internal/controlplane/gatewayclient/client.go deleted file mode 100644 index 24cb170..0000000 --- a/internal/controlplane/gatewayclient/client.go +++ /dev/null @@ -1,51 +0,0 @@ -package gatewayclient - -import ( - "bytes" - "context" - "encoding/json" - "io" - "net/http" - "time" - - "git.ipao.vip/rogee/creator-hub/internal/environment" -) - -const ( - DefaultResponseLimit = 1 << 20 - LargeResponseLimit = 16 << 20 -) - -// Call performs one authenticated native gateway request with a bounded response body. -func Call(ctx context.Context, target environment.Gateway, method, path string, body any, timeout time.Duration) (int, []byte, error) { - return CallWithLimit(ctx, target, method, path, body, timeout, DefaultResponseLimit) -} - -// CallWithLimit performs one authenticated native gateway request with an explicit body limit. -func CallWithLimit(ctx context.Context, target environment.Gateway, method, path string, body any, timeout time.Duration, responseLimit int) (int, []byte, error) { - callCtx, cancel := context.WithTimeout(ctx, timeout) - defer cancel() - var payload io.Reader - if body != nil { - encoded, err := json.Marshal(body) - if err != nil { - return 0, nil, err - } - payload = bytes.NewReader(encoded) - } - request, err := http.NewRequestWithContext(callCtx, method, target.Endpoint+path, payload) - if err != nil { - return 0, nil, err - } - request.Header.Set("Authorization", "Bearer "+target.Token) - if body != nil { - request.Header.Set("Content-Type", "application/json") - } - response, err := http.DefaultClient.Do(request) - if err != nil { - return 0, nil, err - } - defer response.Body.Close() - responseBody, err := io.ReadAll(io.LimitReader(response.Body, int64(responseLimit))) - return response.StatusCode, responseBody, err -} diff --git a/internal/controlplane/workers/runtime.go b/internal/controlplane/workers/runtime.go index 86a9ac8..a3aec7a 100644 --- a/internal/controlplane/workers/runtime.go +++ b/internal/controlplane/workers/runtime.go @@ -29,19 +29,3 @@ func RunRuntimeLeaseHeartbeat(ctx context.Context, store *environment.Store) { func RunCreatorScheduler(ctx context.Context, store *creatorDomain.Store, accountStore *account.Store, environmentStore *environment.Store) { api.RunCreatorScheduler(ctx, store, accountStore, environmentStore) } - -// RunGatewayHealthCheck 每 30 秒对全部注册网关探活一次，只保留最近一次结果。 -func RunGatewayHealthCheck(ctx context.Context, store *environment.Store) { - ticker := time.NewTicker(30 * time.Second) - defer ticker.Stop() - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - if err := api.RecordGatewayHealthChecks(ctx, store); err != nil && ctx.Err() == nil { - logrus.WithField("service", "control-plane").WithError(err).Warn("gateway health check failed") - } - } - } -}