P2.1: Add GatewayConnector BackgroundService with WebSocket client
- GatewayConnector: BackgroundService maintaining persistent WebSocket connection to OpenClaw Gateway with Bearer token auth from backend env only - IGatewayConnector: Interface exposing connection state, version info, health - Exponential backoff reconnect: 1s initial, max 5min, with jitter - Version pinning: RequiredVersion from config; fail-fast or warn on mismatch - Pre-flight HTTP version check before WebSocket connect - GatewayHealthController: GET /api/health/gateway with connection status - GatewayConnectorOptions: Configurable via GatewayConnector config section - Token never logged; read from Integrations:OpenClaw:Password/Token only - 16 tests: options defaults, config binding, health endpoint responses, connection states, DI registration, routing attributes
This commit is contained in:
@@ -0,0 +1,511 @@
|
||||
using System.Net.Http.Headers;
|
||||
using System.Net.WebSockets;
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
using Microsoft.Extensions.Options;
|
||||
|
||||
namespace Nexus.Api.Services;
|
||||
|
||||
/// <summary>
|
||||
/// BackgroundService that maintains a persistent WebSocket connection to the
|
||||
/// OpenClaw Gateway for real-time event streaming and health monitoring.
|
||||
///
|
||||
/// Uses exponential backoff for reconnects, never logs the gateway token,
|
||||
/// and reports connection state via <see cref="IGatewayConnector"/>.
|
||||
/// </summary>
|
||||
public sealed class GatewayConnector : BackgroundService, IGatewayConnector
|
||||
{
|
||||
private readonly IHttpClientFactory _httpClientFactory;
|
||||
private readonly IConfiguration _configuration;
|
||||
private readonly ILogger<GatewayConnector> _logger;
|
||||
private readonly GatewayConnectorOptions _options;
|
||||
|
||||
// ── Connection state (lock-protected) ──
|
||||
private readonly object _lock = new();
|
||||
private GatewayConnectionState _state = GatewayConnectionState.Initializing;
|
||||
private string? _gatewayVersion;
|
||||
private string? _requiredVersion;
|
||||
private DateTimeOffset? _lastConnectedAt;
|
||||
private int _reconnectAttempts;
|
||||
private string? _statusMessage;
|
||||
|
||||
public GatewayConnector(
|
||||
IHttpClientFactory httpClientFactory,
|
||||
IConfiguration configuration,
|
||||
IOptions<GatewayConnectorOptions> options,
|
||||
ILogger<GatewayConnector> logger)
|
||||
{
|
||||
_httpClientFactory = httpClientFactory;
|
||||
_configuration = configuration;
|
||||
_logger = logger;
|
||||
_options = options.Value;
|
||||
}
|
||||
|
||||
// ── IGatewayConnector ──
|
||||
|
||||
public GatewayConnectionState ConnectionState
|
||||
{
|
||||
get { lock (_lock) return _state; }
|
||||
}
|
||||
|
||||
public string? GatewayVersion
|
||||
{
|
||||
get { lock (_lock) return _gatewayVersion; }
|
||||
}
|
||||
|
||||
public string? RequiredVersion
|
||||
{
|
||||
get { lock (_lock) return _requiredVersion; }
|
||||
}
|
||||
|
||||
public DateTimeOffset? LastConnectedAt
|
||||
{
|
||||
get { lock (_lock) return _lastConnectedAt; }
|
||||
}
|
||||
|
||||
public int ReconnectAttempts
|
||||
{
|
||||
get { lock (_lock) return _reconnectAttempts; }
|
||||
}
|
||||
|
||||
public string? StatusMessage
|
||||
{
|
||||
get { lock (_lock) return _statusMessage; }
|
||||
}
|
||||
|
||||
// ── BackgroundService ──
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
// Load version pin from configuration once at startup
|
||||
var pinnedVersion = _configuration["Integrations:OpenClaw:RequiredVersion"];
|
||||
lock (_lock)
|
||||
{
|
||||
_requiredVersion = string.IsNullOrWhiteSpace(pinnedVersion) ? null : pinnedVersion.Trim();
|
||||
}
|
||||
|
||||
if (_requiredVersion is null)
|
||||
{
|
||||
_logger.LogWarning(
|
||||
"Gateway version is UNPINNED (Integrations:OpenClaw:RequiredVersion is empty). "
|
||||
+ "Drift detection is disabled; set a required version to enable fail-fast on mismatch.");
|
||||
}
|
||||
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
try
|
||||
{
|
||||
await ConnectAndReceiveAsync(stoppingToken);
|
||||
}
|
||||
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
break;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "GatewayConnector connection loop exception (will retry)");
|
||||
}
|
||||
|
||||
// Exponential backoff before next reconnect attempt
|
||||
if (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
var delay = ComputeBackoff();
|
||||
_logger.LogInformation(
|
||||
"GatewayConnector reconnecting in {DelayMs}ms (attempt {Attempt})",
|
||||
(int)delay.TotalMilliseconds, GetReconnectAttempts() + 1);
|
||||
|
||||
SetState(GatewayConnectionState.Reconnecting);
|
||||
|
||||
try { await Task.Delay(delay, stoppingToken); }
|
||||
catch (OperationCanceledException) { break; }
|
||||
}
|
||||
}
|
||||
|
||||
_logger.LogInformation("GatewayConnector background service stopped");
|
||||
}
|
||||
|
||||
private async Task ConnectAndReceiveAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
using var ws = new ClientWebSocket();
|
||||
ConfigureWebSocketWithAuth(ws);
|
||||
|
||||
var baseUrl = _configuration["Integrations:OpenClaw:BaseUrl"] ?? "http://127.0.0.1:18789";
|
||||
var wsBase = ConvertToWebSocketUrl(baseUrl);
|
||||
|
||||
// ── Step 1: Pre-flight version check via HTTP ──
|
||||
string? detectedVersion = null;
|
||||
try
|
||||
{
|
||||
detectedVersion = await FetchGatewayVersionAsync(stoppingToken);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Pre-flight version check failed (non-fatal)");
|
||||
}
|
||||
|
||||
// Version check against the pin
|
||||
var (versionOk, versionWarning) = EvaluateVersion(detectedVersion);
|
||||
|
||||
if (!versionOk)
|
||||
{
|
||||
if (_options.FailFastOnVersionMismatch)
|
||||
{
|
||||
_logger.LogError(
|
||||
"Gateway version mismatch — fail-fast enabled. Detected: {Detected}, Required: {Required}",
|
||||
detectedVersion ?? "none", _requiredVersion);
|
||||
|
||||
IncrementReconnectAttempts();
|
||||
SetState(GatewayConnectionState.Failed,
|
||||
$"Version mismatch — fail-fast: detected '{detectedVersion}', required '{_requiredVersion}'.");
|
||||
return;
|
||||
}
|
||||
|
||||
_logger.LogWarning("Gateway version mismatch (non-fatal): {Warning}", versionWarning);
|
||||
}
|
||||
|
||||
// ── Step 2: Establish WebSocket connection ──
|
||||
var wsUri = new Uri(new Uri(wsBase), _options.WebSocketPath);
|
||||
_logger.LogInformation("GatewayConnector connecting to {Uri}", wsUri);
|
||||
|
||||
try
|
||||
{
|
||||
await ws.ConnectAsync(wsUri, stoppingToken);
|
||||
}
|
||||
catch (WebSocketException ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "WebSocket connection failed to {Uri} — gateway may not expose a WS endpoint", wsUri);
|
||||
IncrementReconnectAttempts();
|
||||
SetState(GatewayConnectionState.Disconnected,
|
||||
$"Connection refused — WebSocket endpoint may not be available at {wsUri}. Error: {ex.Message}");
|
||||
return;
|
||||
}
|
||||
catch (HttpRequestException ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "WebSocket HTTP upgrade failed to {Uri}", wsUri);
|
||||
IncrementReconnectAttempts();
|
||||
SetState(GatewayConnectionState.Disconnected,
|
||||
$"HTTP upgrade failed: {ex.Message}");
|
||||
return;
|
||||
}
|
||||
|
||||
// ── Step 3: Submit authentication token (if configured) ──
|
||||
var token = ResolveToken();
|
||||
if (token is not null)
|
||||
{
|
||||
try
|
||||
{
|
||||
var authFrame = Encoding.UTF8.GetBytes(
|
||||
JsonSerializer.Serialize(new { type = "auth", token }));
|
||||
await ws.SendAsync(
|
||||
new ArraySegment<byte>(authFrame),
|
||||
WebSocketMessageType.Text,
|
||||
endOfMessage: true,
|
||||
cancellationToken: stoppingToken);
|
||||
|
||||
_logger.LogDebug("WebSocket auth token sent");
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Failed to send WebSocket auth frame");
|
||||
}
|
||||
}
|
||||
|
||||
// ── After successful connection ──
|
||||
lock (_lock)
|
||||
{
|
||||
_state = GatewayConnectionState.Connected;
|
||||
_gatewayVersion = detectedVersion;
|
||||
_lastConnectedAt = DateTimeOffset.UtcNow;
|
||||
_reconnectAttempts = 0;
|
||||
_statusMessage = versionWarning;
|
||||
}
|
||||
|
||||
_logger.LogInformation(
|
||||
"GatewayConnector connected. Version: {Version}, Required: {Required}",
|
||||
detectedVersion ?? "unknown", _requiredVersion ?? "unpinned");
|
||||
|
||||
// ── Step 4: Receive loop (heartbeat / event processing) ──
|
||||
await ReceiveLoopAsync(ws, stoppingToken);
|
||||
}
|
||||
|
||||
private async Task ReceiveLoopAsync(ClientWebSocket ws, CancellationToken stoppingToken)
|
||||
{
|
||||
var buffer = new byte[4096];
|
||||
|
||||
try
|
||||
{
|
||||
while (ws.State == WebSocketState.Open && !stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
var result = await ws.ReceiveAsync(new ArraySegment<byte>(buffer), stoppingToken);
|
||||
|
||||
if (result.MessageType == WebSocketMessageType.Close)
|
||||
{
|
||||
_logger.LogInformation(
|
||||
"GatewayConnector received close frame: {Description}",
|
||||
result.CloseStatusDescription ?? "no description");
|
||||
break;
|
||||
}
|
||||
|
||||
// Process text frames (events, heartbeats, status updates)
|
||||
if (result.MessageType == WebSocketMessageType.Text)
|
||||
{
|
||||
var message = Encoding.UTF8.GetString(buffer, 0, result.Count);
|
||||
ProcessGatewayMessage(message);
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
// Expected on shutdown
|
||||
}
|
||||
catch (WebSocketException ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "WebSocket connection lost");
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (ws.State == WebSocketState.Open || ws.State == WebSocketState.CloseReceived)
|
||||
{
|
||||
try
|
||||
{
|
||||
await ws.CloseAsync(
|
||||
WebSocketCloseStatus.NormalClosure, "Connector shutdown", CancellationToken.None);
|
||||
}
|
||||
catch { /* Best effort */ }
|
||||
}
|
||||
|
||||
SetState(GatewayConnectionState.Disconnected, "Connection closed");
|
||||
}
|
||||
}
|
||||
|
||||
private void ProcessGatewayMessage(string message)
|
||||
{
|
||||
try
|
||||
{
|
||||
using var doc = JsonDocument.Parse(message);
|
||||
var root = doc.RootElement;
|
||||
var type = root.TryGetProperty("type", out var typeEl) ? typeEl.GetString() : null;
|
||||
|
||||
switch (type)
|
||||
{
|
||||
case "heartbeat":
|
||||
case "pong":
|
||||
_logger.LogDebug("Gateway heartbeat received");
|
||||
break;
|
||||
|
||||
case "event":
|
||||
var eventName = root.TryGetProperty("name", out var nameEl)
|
||||
? nameEl.GetString() : "unknown";
|
||||
_logger.LogDebug("Gateway event: {EventName}", eventName);
|
||||
// Future: fan-out to other services via a channel/event bus
|
||||
break;
|
||||
|
||||
case "error":
|
||||
var errorMsg = root.TryGetProperty("message", out var msgEl)
|
||||
? msgEl.GetString() : "unknown error";
|
||||
_logger.LogWarning("Gateway error frame received: {Error}", errorMsg);
|
||||
break;
|
||||
|
||||
default:
|
||||
_logger.LogDebug("Gateway message (type={Type}): {Preview}",
|
||||
type ?? "none", Truncate(message, 120));
|
||||
break;
|
||||
}
|
||||
}
|
||||
catch (JsonException)
|
||||
{
|
||||
_logger.LogDebug("Non-JSON gateway message: {Preview}", Truncate(message, 80));
|
||||
}
|
||||
}
|
||||
|
||||
// ── Helpers ──
|
||||
|
||||
private (bool Ok, string? Warning) EvaluateVersion(string? detectedVersion)
|
||||
{
|
||||
if (_requiredVersion is null)
|
||||
return (true, null); // Unpinned — always ok
|
||||
|
||||
if (detectedVersion is null)
|
||||
return (_options.FailFastOnMissingVersion ? false : true,
|
||||
"Gateway version not detected while version pin is set.");
|
||||
|
||||
if (!string.Equals(detectedVersion, _requiredVersion, StringComparison.OrdinalIgnoreCase))
|
||||
return (false, $"Version drift: detected '{detectedVersion}', required '{_requiredVersion}'.");
|
||||
|
||||
return (true, null);
|
||||
}
|
||||
|
||||
private async Task<string?> FetchGatewayVersionAsync(CancellationToken ct)
|
||||
{
|
||||
var client = _httpClientFactory.CreateClient("gateway");
|
||||
try
|
||||
{
|
||||
using var request = new HttpRequestMessage(HttpMethod.Get, "/health");
|
||||
ApplyAuth(request);
|
||||
|
||||
using var response = await client.SendAsync(request, ct);
|
||||
if (!response.IsSuccessStatusCode)
|
||||
return null;
|
||||
|
||||
// Check X-OpenClaw-Version header first
|
||||
if (response.Headers.TryGetValues("X-OpenClaw-Version", out var headerValues))
|
||||
{
|
||||
var version = headerValues.FirstOrDefault();
|
||||
if (!string.IsNullOrWhiteSpace(version))
|
||||
return version.Trim();
|
||||
}
|
||||
|
||||
// Try JSON body
|
||||
var body = await response.Content.ReadAsStringAsync(ct);
|
||||
if (!string.IsNullOrWhiteSpace(body))
|
||||
{
|
||||
try
|
||||
{
|
||||
using var doc = JsonDocument.Parse(body);
|
||||
var root = doc.RootElement;
|
||||
var v = TryGetString(root, "version")
|
||||
?? TryGetString(root, "gatewayVersion")
|
||||
?? TryGetString(root, "openclawVersion");
|
||||
if (v is not null) return v;
|
||||
}
|
||||
catch { /* Not JSON */ }
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
catch
|
||||
{
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private TimeSpan ComputeBackoff()
|
||||
{
|
||||
var attempts = GetReconnectAttempts();
|
||||
var backoffMs = (int)Math.Min(
|
||||
_options.ReconnectInitialDelayMs * Math.Pow(2, attempts),
|
||||
_options.ReconnectMaxDelayMs);
|
||||
|
||||
// Add jitter (±25%)
|
||||
var jitter = (int)(backoffMs * 0.25 * (Random.Shared.NextDouble() * 2 - 1));
|
||||
return TimeSpan.FromMilliseconds(Math.Max(100, backoffMs + jitter));
|
||||
}
|
||||
|
||||
private void IncrementReconnectAttempts()
|
||||
{
|
||||
lock (_lock) { _reconnectAttempts++; }
|
||||
}
|
||||
|
||||
private int GetReconnectAttempts()
|
||||
{
|
||||
lock (_lock) { return _reconnectAttempts; }
|
||||
}
|
||||
|
||||
private void SetState(GatewayConnectionState state, string? message = null)
|
||||
{
|
||||
lock (_lock)
|
||||
{
|
||||
_state = state;
|
||||
if (message is not null) _statusMessage = message;
|
||||
}
|
||||
}
|
||||
|
||||
private string? ResolveToken()
|
||||
{
|
||||
// Token NEVER logged — only read from config here
|
||||
var token = _configuration["Integrations:OpenClaw:Password"]
|
||||
?? _configuration["Integrations:OpenClaw:Token"];
|
||||
|
||||
return string.IsNullOrWhiteSpace(token) ? null : token;
|
||||
}
|
||||
|
||||
private void ApplyAuth(HttpRequestMessage request)
|
||||
{
|
||||
var token = ResolveToken();
|
||||
if (token is not null)
|
||||
request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", token);
|
||||
}
|
||||
|
||||
private static void ConfigureWebSocket(ClientWebSocket ws)
|
||||
{
|
||||
// Add auth header to the initial HTTP upgrade request
|
||||
ws.Options.KeepAliveInterval = TimeSpan.FromSeconds(30);
|
||||
ws.Options.UseDefaultCredentials = false;
|
||||
}
|
||||
|
||||
// Overload with auth header via cookie (WebSocket can't do Authorization header natively in all runtimes)
|
||||
private void ConfigureWebSocketWithAuth(ClientWebSocket ws)
|
||||
{
|
||||
var token = ResolveToken();
|
||||
if (token is not null)
|
||||
{
|
||||
ws.Options.SetRequestHeader("Authorization", "Bearer " + token);
|
||||
}
|
||||
}
|
||||
|
||||
private static string ConvertToWebSocketUrl(string baseUrl)
|
||||
{
|
||||
var trimmed = baseUrl.TrimEnd('/');
|
||||
if (trimmed.StartsWith("https://", StringComparison.OrdinalIgnoreCase))
|
||||
return "wss://" + trimmed["https://".Length..];
|
||||
if (trimmed.StartsWith("http://", StringComparison.OrdinalIgnoreCase))
|
||||
return "ws://" + trimmed["http://".Length..];
|
||||
return "ws://" + trimmed;
|
||||
}
|
||||
|
||||
private static string? TryGetString(JsonElement root, string property)
|
||||
=> root.ValueKind == JsonValueKind.Object
|
||||
&& root.TryGetProperty(property, out var value)
|
||||
&& value.ValueKind == JsonValueKind.String
|
||||
? value.GetString()
|
||||
: null;
|
||||
|
||||
private static string Truncate(string value, int maxLength)
|
||||
=> value.Length <= maxLength ? value : value[..maxLength] + "\u2026";
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Configuration options for the GatewayConnector background service.
|
||||
/// </summary>
|
||||
public sealed class GatewayConnectorOptions
|
||||
{
|
||||
public const string SectionName = "GatewayConnector";
|
||||
|
||||
/// <summary>
|
||||
/// WebSocket path relative to the gateway base URL.
|
||||
/// Default: "/ws"
|
||||
/// </summary>
|
||||
public string WebSocketPath { get; set; } = "/ws";
|
||||
|
||||
/// <summary>
|
||||
/// Initial reconnect delay in milliseconds.
|
||||
/// Default: 1000 (1 second)
|
||||
/// </summary>
|
||||
public int ReconnectInitialDelayMs { get; set; } = 1000;
|
||||
|
||||
/// <summary>
|
||||
/// Maximum reconnect delay in milliseconds.
|
||||
/// Default: 300_000 (5 minutes)
|
||||
/// </summary>
|
||||
public int ReconnectMaxDelayMs { get; set; } = 300_000;
|
||||
|
||||
/// <summary>
|
||||
/// Maximum number of consecutive reconnect attempts before entering Failed state.
|
||||
/// Default: 0 (no limit — keep retrying forever)
|
||||
/// </summary>
|
||||
public int ReconnectMaxAttempts { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// When true, the connector enters Failed state immediately if the gateway version
|
||||
/// doesn't match the pinned required version (instead of connecting with a warning).
|
||||
/// Default: true
|
||||
/// </summary>
|
||||
public bool FailFastOnVersionMismatch { get; set; } = true;
|
||||
|
||||
/// <summary>
|
||||
/// When true, missing version (gateway doesn't report one) with a version pin set
|
||||
/// also triggers fail-fast. Default: false (allows pinned-version gateways that
|
||||
/// don't expose a version endpoint).
|
||||
/// </summary>
|
||||
public bool FailFastOnMissingVersion { get; set; }
|
||||
}
|
||||
Reference in New Issue
Block a user