rogee/creator-hub
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(controlplane): outbound gateway WS channel replaces inbound HTTP client

Browse Source
This commit is contained in:
rogee
2026-10-09 05:39:41 +08:00
parent 1f647b5b65
commit bee423fe51
17 changed files with 1090 additions and 1071 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+2
View File
@@ -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
go.sum
+4
View File
@@ -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=
internal/controlplane/api/douyin.go
+1 -1
View File
@@ -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) ||
internal/controlplane/api/environments.go
+71 -20
View File
@@ -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) {
internal/controlplane/api/event_channel.go
-255
View File
@@ -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"))
}
}
internal/controlplane/api/event_channel_test.go
-254
View File
@@ -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
}
}
internal/controlplane/api/event_listener_worker.go
+46 -28
View File
@@ -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
internal/controlplane/api/gateway_channel.go
+375
View File
@@ -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):
}
}
}
internal/controlplane/api/gateway_channel_test.go
+320
View File
@@ -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)
}
}
internal/controlplane/api/gateway_channel_test_kit_test.go
+254
View File
@@ -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://")
}
internal/controlplane/api/gateway_health.go
-83
View File
@@ -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)
}
internal/controlplane/api/gateway_health_test.go
-273
View File
@@ -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)
}
}
internal/controlplane/api/hub_native_unit_test.go
-84
View File
@@ -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)
}
}
internal/controlplane/app/app.go
+15 -6
View File
@@ -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)
internal/controlplane/app/app_test.go
+2
View File
@@ -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},
internal/controlplane/gatewayclient/client.go
-51
View File
@@ -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
}
internal/controlplane/workers/runtime.go
-16
View File
@@ -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")
}
}
}
}