CUB-204: wire WS client as primary, REST poller as fallback
Some checks failed
Dev Build / build-test (pull_request) Failing after 1s
Build (Dev) / trigger-deploy (pull_request) Has been skipped
openclaw/grimm-review REJECTED — 6 blocking issues
Build (Dev) / build-go-backend (pull_request) Failing after 0s
Build (Dev) / build-frontend (pull_request) Failing after 1s

- Rename GatewayURL/GatewayPollInterval → GatewayRestURL/GatewayRestPollInterval
- Change Docker-aware defaults (host.docker.internal instead of localhost)
- Client.Start() waits for WS readiness (30s timeout), falls back to REST
- Client.SetWSClient()/MarkWSReady() for WS→REST coordination
- WSClient.SetRESTClient() so WS notifies REST on successful handshake
- main.go wires both clients: WS primary, REST fallback with cross-references
- .env.example documents WS_GATEWAY_URL, GATEWAY_TOKEN, REST fallback vars
- docker-compose.yml adds WS_GATEWAY_URL and GATEWAY_TOKEN env vars
- reference/CONTROL_CENTER_CONTEXT.md documents architecture and startup sequence
This commit is contained in:
Dex
2026-05-20 11:16:05 +00:00
parent 9062f8fa8d
commit e131798f3b
9 changed files with 369 additions and 38 deletions

View File

@@ -17,13 +17,17 @@ import (
)
// Client polls the OpenClaw gateway for agent status and keeps the database
// and SSE broker in sync.
// and SSE broker in sync. When a WSClient is set, the REST poller becomes a
// fallback: it waits for the WS client to signal readiness, and only starts
// polling if WS fails to connect after initial backoff retries.
type Client struct {
url string
pollInterval time.Duration
httpClient *http.Client
agents repository.AgentRepo
broker *handler.Broker
wsClient *WSClient // optional WS client; when set, REST is fallback only
wsReady chan struct{} // closed once WS connection is established
}
// Config holds gateway client configuration, typically loaded from environment.
@@ -48,22 +52,67 @@ func NewClient(cfg Config, agents repository.AgentRepo, broker *handler.Broker)
httpClient: &http.Client{Timeout: 10 * time.Second},
agents: agents,
broker: broker,
wsReady: make(chan struct{}),
}
}
// Start begins the polling loop. It runs until ctx is cancelled.
func (c *Client) Start(ctx context.Context) {
slog.Info("gateway client starting",
"url", c.url,
"pollInterval", c.pollInterval.String())
// SetWSClient wires the WebSocket client so the REST poller knows to defer
// to it. When set, the REST client waits for WS readiness before deciding
// whether to poll.
func (c *Client) SetWSClient(ws *WSClient) {
c.wsClient = ws
}
// MarkWSReady signals that the WS connection is live and the REST poller
// should stand down. Called by WSClient after a successful handshake.
func (c *Client) MarkWSReady() {
select {
case <-c.wsReady:
// already closed
default:
close(c.wsReady)
}
}
// Start begins the gateway client loop. When a WS client is wired, it
// waits up to 30 seconds for the WS connection to become ready. If WS
// connects, the REST poller stands down and only logs periodically. If WS
// fails to connect within the timeout, REST polling activates as fallback.
func (c *Client) Start(ctx context.Context) {
if c.wsClient != nil {
slog.Info("gateway client waiting for WS connection", "timeout", "30s")
select {
case <-c.wsReady:
slog.Info("gateway client using WS — REST poller standing down")
// WS is live; keep this goroutine alive but idle. If WS
// disconnects later, we could re-enter polling, but for now
// the WS client handles its own reconnection.
<-ctx.Done()
slog.Info("gateway client stopped (WS mode)")
return
case <-time.After(30 * time.Second):
slog.Warn("gateway client: WS not ready after 30s — falling back to REST polling",
"url", c.url,
"pollInterval", c.pollInterval.String())
case <-ctx.Done():
slog.Info("gateway client stopped while waiting for WS")
return
}
} else {
slog.Info("gateway client using REST polling (no WS client configured)",
"url", c.url,
"pollInterval", c.pollInterval.String())
}
// REST fallback polling
ticker := time.NewTicker(c.pollInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
slog.Info("gateway client stopped")
slog.Info("gateway client stopped (REST fallback)")
return
case <-ticker.C:
c.poll(ctx)