From d6e5096681c7bc544788f044760afe8e64f1c4e8 Mon Sep 17 00:00:00 2001 From: deleteLater Date: Mon, 31 Aug 2026 20:23:54 +0800 Subject: [PATCH 1/2] initial implementation --- .../Transport/FbWebSocket.cs | 31 +++++-- .../Transport/FbWebSocketTests.cs | 82 +++++++++++++++---- 2 files changed, 92 insertions(+), 21 deletions(-) diff --git a/src/FeatBit.ServerSdk/Transport/FbWebSocket.cs b/src/FeatBit.ServerSdk/Transport/FbWebSocket.cs index 15216be..abf380a 100644 --- a/src/FeatBit.ServerSdk/Transport/FbWebSocket.cs +++ b/src/FeatBit.ServerSdk/Transport/FbWebSocket.cs @@ -13,6 +13,8 @@ namespace FeatBit.Sdk.Server.Transport { internal sealed partial class FbWebSocket { + internal const WebSocketCloseStatus ServerRejectedCloseStatus = (WebSocketCloseStatus)4003; + public event Func OnConnected; public event Func OnKeepAlive; public event Func OnReconnecting; @@ -28,6 +30,7 @@ internal sealed partial class FbWebSocket private readonly Func _webSocketUriResolver; private WebSocketTransport _transport; private Task _receiveTask; + private Task _reconnectTask; private readonly TimeSpan _keepAliveInterval; private Timer _keepAliveTimer; private Exception _closeException; @@ -85,7 +88,7 @@ public async Task ConnectAsync(bool isReconnecting = false, CancellationToken ca Log.ErrorStartingTransport(_logger, ex); // reconnect if we failed to start the transport - _ = ReconnectAsync(); + _reconnectTask = ReconnectAsync(); } throw; @@ -197,7 +200,7 @@ private async Task HandleConnectionCloseAsync() if (ShouldReconnect()) { - _ = ReconnectAsync(); + _reconnectTask = ReconnectAsync(); } else { @@ -205,10 +208,24 @@ private async Task HandleConnectionCloseAsync() } } - private bool ShouldReconnect() + private bool ShouldReconnect() => ShouldReconnect(_transport.CloseStatus, _stopCts.IsCancellationRequested); + + internal static bool ShouldReconnect(WebSocketCloseStatus? closeStatus, bool stopRequested) { - var closeStatus = _transport.CloseStatus; - return closeStatus != WebSocketCloseStatus.NormalClosure && closeStatus != (WebSocketCloseStatus)4003; + if (closeStatus == ServerRejectedCloseStatus) + { + // server rejected the connection explicitly, so we should not reconnect + return false; + } + + if (stopRequested) + { + // connection was stopped manually, so we should not reconnect + return false; + } + + // otherwise, we should reconnect + return true; } private async Task ReconnectAsync() @@ -343,10 +360,12 @@ public async Task CloseAsync() Log.WaitingForReceiveLoopToTerminate(_logger); await (_receiveTask ?? Task.CompletedTask).ConfigureAwait(false); + + await (_reconnectTask ?? Task.CompletedTask).ConfigureAwait(false); } Log.Closed(_logger, _transport?.CloseStatus); _transport = null; } } -} \ No newline at end of file +} diff --git a/tests/FeatBit.ServerSdk.Tests/Transport/FbWebSocketTests.cs b/tests/FeatBit.ServerSdk.Tests/Transport/FbWebSocketTests.cs index 773ae7a..26898c6 100644 --- a/tests/FeatBit.ServerSdk.Tests/Transport/FbWebSocketTests.cs +++ b/tests/FeatBit.ServerSdk.Tests/Transport/FbWebSocketTests.cs @@ -101,25 +101,19 @@ public async Task TestOnClosed_ClientInitiated() Assert.True(onClosedCalled); } - [Theory] - [InlineData("close-normally", WebSocketCloseStatus.NormalClosure, "", "")] - [InlineData("close-with-4003", (WebSocketCloseStatus)4003, "invalid request, close by server", "")] - public async Task TestOnClosed_ServerInitiated( - string op, - WebSocketCloseStatus expectCloseStatus, - string expectCloseDescription, - string expectCloseMessage) + [Fact] + public async Task TestOnClosed_ServerRejectedConnection() { - var fbWebSocket = _app.CreateFbWebSocket(op); + var fbWebSocket = _app.CreateFbWebSocket("close-with-4003"); var tcs = new TaskCompletionSource(); var onClosedTask = tcs.Task; fbWebSocket.OnClosed += (exception, closeStatus, closeDescription, closeMessage) => { Assert.Null(exception); - Assert.Equal(expectCloseStatus, closeStatus); - Assert.Equal(expectCloseDescription, closeDescription); - Assert.Equal(expectCloseMessage, closeMessage); + Assert.Equal((WebSocketCloseStatus)4003, closeStatus); + Assert.Equal("invalid request, close by server", closeDescription); + Assert.Equal(string.Empty, closeMessage); tcs.SetResult(); return Task.CompletedTask; @@ -129,6 +123,31 @@ public async Task TestOnClosed_ServerInitiated( await onClosedTask.WaitAsync(TimeSpan.FromSeconds(1)); } + [Theory] + [InlineData(null)] + [InlineData(WebSocketCloseStatus.NormalClosure)] + [InlineData(WebSocketCloseStatus.EndpointUnavailable)] + [InlineData(WebSocketCloseStatus.InternalServerError)] + public void ShouldReconnectWhenNotStopped(WebSocketCloseStatus? closeStatus) + { + Assert.True(FbWebSocket.ShouldReconnect(closeStatus, stopRequested: false)); + } + + [Fact] + public void ShouldNotReconnectWhenStopped() + { + Assert.False(FbWebSocket.ShouldReconnect(WebSocketCloseStatus.NormalClosure, stopRequested: true)); + } + + [Fact] + public void ShouldNotReconnectWhenServerRejectsConnection() + { + Assert.False(FbWebSocket.ShouldReconnect( + FbWebSocket.ServerRejectedCloseStatus, + stopRequested: false + )); + } + [Fact] public async Task TestOnKeepAlive() { @@ -216,15 +235,45 @@ public async Task TestReconnectOnUnexpectedClose() Assert.True(onReconnectedCalled); } + [Fact] + public async Task TestReconnectOnNormalClose() + { + var fbWebSocket = _app.CreateFbWebSocket("close-normally"); + var reconnecting = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + fbWebSocket.OnReconnecting += _ => + { + reconnecting.TrySetResult(); + return Task.CompletedTask; + }; + + try + { + await fbWebSocket.ConnectAsync(); + await reconnecting.Task.WaitAsync(TimeSpan.FromSeconds(1)); + } + finally + { + await fbWebSocket.CloseAsync(); + } + } + [Fact] public async Task CanStopFbWebSocketWhenReconnecting() { var fbWebSocket = _app.CreateFbWebSocket( "close-unexpectedly", // set reconnect delay as 1s - builder => builder.ReconnectRetryDelays(new[] { TimeSpan.FromSeconds(1) }) + builder => builder.ReconnectRetryDelays([TimeSpan.FromSeconds(1)]) ); + var reconnecting = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + fbWebSocket.OnReconnecting += _ => + { + reconnecting.TrySetResult(); + return Task.CompletedTask; + }; + var tcs = new TaskCompletionSource(); var onClosedTask = tcs.Task; fbWebSocket.OnClosed += (exception, closeStatus, closeDescription, closeMessage) => @@ -232,7 +281,6 @@ public async Task CanStopFbWebSocketWhenReconnecting() Assert.Null(exception); Assert.Equal(WebSocketCloseStatus.EndpointUnavailable, closeStatus); Assert.Equal("server going down", closeDescription); - Assert.Equal("FbWebSocket stopped during reconnect delay. Done reconnecting.", closeMessage); tcs.SetResult(); @@ -240,8 +288,12 @@ public async Task CanStopFbWebSocketWhenReconnecting() }; await fbWebSocket.ConnectAsync(); + await reconnecting.Task.WaitAsync(TimeSpan.FromSeconds(1)); + // OnReconnecting is raised immediately before the retry delay. Give the reconnect task + // enough time to enter that 1-second delay before stopping it. + await Task.Delay(TimeSpan.FromMilliseconds(100)); await fbWebSocket.CloseAsync(); await onClosedTask.WaitAsync(TimeSpan.FromSeconds(1)); } -} \ No newline at end of file +} From 031219ba1c9330e9d21ee9196abfab2d2bcb05f1 Mon Sep 17 00:00:00 2001 From: deleteLater Date: Mon, 31 Aug 2026 20:47:30 +0800 Subject: [PATCH 2/2] Reconnect after normal WebSocket closures --- reconnect-test.mjs | 328 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 328 insertions(+) create mode 100644 reconnect-test.mjs diff --git a/reconnect-test.mjs b/reconnect-test.mjs new file mode 100644 index 0000000..959fe68 --- /dev/null +++ b/reconnect-test.mjs @@ -0,0 +1,328 @@ +#!/usr/bin/env node + +import { spawn, spawnSync } from 'node:child_process'; +import { appendFileSync, createWriteStream, mkdirSync, rmSync, writeFileSync } from 'node:fs'; +import { join, resolve } from 'node:path'; +import process from 'node:process'; + +const repoRoot = resolve(import.meta.dirname); +const runId = new Date().toISOString().replaceAll(/[-:TZ.]/g, '').slice(0, 14); +const artifactRoot = join(repoRoot, 'artifacts', 'els-reconnect'); +const artifactDirectory = join(artifactRoot, runId); +const clientProjectDirectory = join(artifactDirectory, 'sdk-client'); +const sdkProject = join(repoRoot, 'src', 'FeatBit.ServerSdk', 'FeatBit.ServerSdk.csproj'); +const elsBaseUri = 'http://localhost:5100'; +const postgresContainer = 'featbit-infra-postgresql-1'; +const elsContainer = 'featbit-sdk-reconnect-els'; +const elsImage = 'featbit/evaluation-server:reconnect-comparison'; +const dockerNetwork = 'featbit-network'; + +const options = parseArguments(process.argv.slice(2)); +mkdirSync(artifactRoot, { recursive: true }); +mkdirSync(artifactDirectory, { recursive: true }); +const testLogPath = join(artifactRoot, `sdk-${runId}-test.log`); +const clientLogPath = join(artifactRoot, `sdk-${runId}-client.log`); + +let environmentSecret = ''; +let clientProcess = null; +let clientStatus = ''; +let issueReproduced = false; +let testStarted = false; +let sdkClosedAt = null; +let clientStoppedAt = null; + +function parseArguments(args) { + const parsed = { + environmentId: '', testDurationSeconds: 300, downtimeSeconds: 2, + recoveryTimeoutSeconds: 90, closedConfirmationSeconds: 10, + stopElsAfterTest: false, keepArtifacts: false, + }; + for (let index = 0; index < args.length; index += 1) { + const argument = args[index]; + const next = () => { + index += 1; + if (index >= args.length) throw new Error(`Missing value for ${argument}.`); + return args[index]; + }; + switch (argument) { + case '--environment-id': parsed.environmentId = next(); break; + case '--test-duration': parsed.testDurationSeconds = positiveInteger(next(), argument); break; + case '--downtime': parsed.downtimeSeconds = nonNegativeInteger(next(), argument); break; + case '--recovery-timeout': parsed.recoveryTimeoutSeconds = positiveInteger(next(), argument); break; + case '--closed-confirmation': parsed.closedConfirmationSeconds = positiveInteger(next(), argument); break; + case '--stop-els-after-test': parsed.stopElsAfterTest = true; break; + case '--keep-artifacts': parsed.keepArtifacts = true; break; + case '--help': printHelp(); process.exit(0); + default: throw new Error(`Unknown argument: ${argument}`); + } + } + return parsed; +} + +function positiveInteger(value, option) { + const number = Number.parseInt(value, 10); + if (!Number.isInteger(number) || number <= 0) throw new Error(`${option} must be a positive integer.`); + return number; +} + +function nonNegativeInteger(value, option) { + const number = Number.parseInt(value, 10); + if (!Number.isInteger(number) || number < 0) throw new Error(`${option} must be a non-negative integer.`); + return number; +} + +function printHelp() { + console.log(`Usage: node reconnect-test.mjs [options] + +Builds a temporary client against the SDK source in this checkout, repeatedly restarts ELS, +and detects whether FbClient becomes permanently Closed. + +Options: + --environment-id Environment whose server secret is used (default: first environment) + --test-duration Total restart-test interval (default: 300) + --downtime ELS downtime per cycle (default: 2) + --recovery-timeout Maximum Ready-state recovery wait per cycle (default: 90) + --closed-confirmation Time Closed must persist after ELS recovery (default: 10) + --stop-els-after-test Leave ELS stopped after the test + --keep-artifacts Keep the generated temporary client project + --help Show this help + +Exit codes: 0 = no stuck connection observed, 2 = permanent Closed state reproduced. +`); +} + +function step(message) { console.log(`\n==> ${message}`); } +function sleep(milliseconds) { return new Promise((resolvePromise) => setTimeout(resolvePromise, milliseconds)); } + +function timestamp(date = new Date()) { + const pad = (value, length = 2) => String(value).padStart(length, '0'); + const offsetMinutes = -date.getTimezoneOffset(); + const sign = offsetMinutes >= 0 ? '+' : '-'; + const offset = Math.abs(offsetMinutes); + return `${date.getFullYear()}-${pad(date.getMonth() + 1)}-${pad(date.getDate())} ` + + `${pad(date.getHours())}:${pad(date.getMinutes())}:${pad(date.getSeconds())}.` + + `${pad(date.getMilliseconds(), 3)} ${sign}${pad(Math.floor(offset / 60))}:${pad(offset % 60)}`; +} + +function logTest(message, date = new Date()) { + const line = `${timestamp(date)} [TEST] ${message}`; + console.log(line); + appendFileSync(testLogPath, `${line}\n`, 'utf8'); +} + +function run(command, args, { cwd = repoRoot, allowFailure = false } = {}) { + const result = spawnSync(command, args, { cwd, encoding: 'utf8', windowsHide: true }); + if (result.error) throw result.error; + if (!allowFailure && result.status !== 0) { + throw new Error(`${command} ${args.join(' ')} failed.\n${result.stdout ?? ''}${result.stderr ?? ''}`); + } + return result; +} + +function docker(args, settings = {}) { return run('docker', args, settings); } +function escapeSqlLiteral(value) { return value.replaceAll("'", "''"); } +function queryDatabase(sql) { + return docker(['exec', postgresContainer, 'psql', '-U', 'postgres', '-d', 'featbit', '-Atc', sql]) + .stdout.trim().split(/\r?\n/)[0]?.trim() ?? ''; +} + +async function waitFor(description, predicate, timeoutSeconds) { + const deadline = Date.now() + timeoutSeconds * 1_000; + while (Date.now() < deadline) { + if (await predicate()) return; + await sleep(250); + } + throw new Error(`${description} timed out after ${timeoutSeconds} seconds (last SDK status: ${clientStatus || 'Unavailable'}).`); +} + +async function endpointStatus(uri) { + try { return (await fetch(uri, { signal: AbortSignal.timeout(2_000) })).status; } + catch { return 0; } +} + +function resolveEnvironment() { + const idFilter = options.environmentId + ? `and e.id::text = '${escapeSqlLiteral(options.environmentId)}'` + : ''; + const row = queryDatabase(`select concat_ws('|', e.id::text, e.name, s.item->>'value') + from environments e cross join lateral jsonb_array_elements(e.secrets) s(item) + where s.item->>'type' = 'server' ${idFilter} + order by e.created_at, e.id limit 1;`); + const first = row.indexOf('|'); + const second = row.indexOf('|', first + 1); + if (first < 1 || second < 0 || second === row.length - 1) { + throw new Error(options.environmentId + ? `No server secret found for environment '${options.environmentId}'.` + : 'No environment server secret was found.'); + } + console.log(`Environment: ${row.slice(first + 1, second)} (${row.slice(0, first)})`); + environmentSecret = row.slice(second + 1); +} + +function ensureElsContainer() { + if (docker(['inspect', elsContainer], { allowFailure: true }).status === 0) { + docker(['start', elsContainer]); + return; + } + docker([ + 'run', '--detach', '--name', elsContainer, '--network', dockerNetwork, '--publish', '5100:5100', + '--env', 'ASPNETCORE_ENVIRONMENT=Production', '--env', 'DbProvider=Postgres', + '--env', 'MqProvider=Postgres', '--env', 'CacheProvider=None', + '--env', 'Postgres__ConnectionString=Host=postgresql;Port=5432;Username=postgres;Password=please_change_me;Database=featbit', + elsImage, + ]); +} + +async function waitForEls() { + await waitFor('ELS health check', async () => await endpointStatus(`${elsBaseUri}/health/liveness`) === 200, 90); +} + +function createClientProject() { + mkdirSync(clientProjectDirectory, { recursive: true }); + writeFileSync(join(clientProjectDirectory, 'ReconnectClient.csproj'), ` + + Exe + net8.0 + enable + enable + + + +`, 'utf8'); + writeFileSync(join(clientProjectDirectory, 'Program.cs'), `using FeatBit.Sdk.Server; +using FeatBit.Sdk.Server.Options; + +var secret = Environment.GetEnvironmentVariable("FEATBIT_RECONNECT_SECRET") + ?? throw new InvalidOperationException("FEATBIT_RECONNECT_SECRET is required."); +var options = new FbOptionsBuilder(secret) + .Streaming(new Uri("ws://localhost:5100")) + .Event(new Uri("http://localhost:5100")) + .DisableEvents(true) + .StartWaitTime(TimeSpan.FromSeconds(10)) + .Build(); +var client = new FbClient(options); +using var timer = new PeriodicTimer(TimeSpan.FromMilliseconds(250)); +FbClientStatus? lastStatus = null; +while (await timer.WaitForNextTickAsync()) +{ + if (lastStatus != client.Status) + { + lastStatus = client.Status; + Console.WriteLine(DateTimeOffset.Now.ToString("yyyy-MM-dd HH:mm:ss.fff zzz") + " [SDK] STATUS=" + lastStatus); + } +} +`, 'utf8'); +} + +function observeClientOutput(chunk, log) { + const text = chunk.toString(); + log.write(text); + for (const match of text.matchAll(/STATUS=(NotReady|Ready|Stale|Closed)/g)) { + if (match[1] !== clientStatus) { + clientStatus = match[1]; + logTest(`SDK status changed to ${clientStatus}.`); + if (clientStatus === 'Closed' && sdkClosedAt === null) sdkClosedAt = new Date(); + } + } +} + +function startClient() { + createClientProject(); + run('dotnet', ['build', 'ReconnectClient.csproj', '--configuration', 'Release'], { cwd: clientProjectDirectory }); + const assembly = join(clientProjectDirectory, 'bin', 'Release', 'net8.0', 'ReconnectClient.dll'); + const log = createWriteStream(clientLogPath); + const child = spawn('dotnet', [assembly], { + cwd: clientProjectDirectory, windowsHide: true, stdio: ['ignore', 'pipe', 'pipe'], + env: { ...process.env, FEATBIT_RECONNECT_SECRET: environmentSecret }, + }); + child.stdout.on('data', (chunk) => observeClientOutput(chunk, log)); + child.stderr.on('data', (chunk) => observeClientOutput(chunk, log)); + clientProcess = { child, log }; + console.log(`SDK client log: ${clientLogPath}`); +} + +async function stopClient(reason) { + if (!clientProcess) return; + clientStoppedAt = new Date(); + logTest(`Stopping SDK client process (${reason}).`, clientStoppedAt); + clientProcess.child.kill(); + await Promise.race([ + new Promise((resolvePromise) => clientProcess.child.once('exit', resolvePromise)), sleep(5_000), + ]); + clientProcess.log.end(); + clientProcess = null; + logTest('SDK client process stop completed.'); +} + +async function confirmClosed(cycle) { + if (clientStatus !== 'Closed') return false; + await sleep(options.closedConfirmationSeconds * 1_000); + if (clientStatus !== 'Closed') return false; + issueReproduced = true; + logTest(`Cycle ${cycle}: SDK remained Closed ${options.closedConfirmationSeconds} seconds after ELS recovered.`); + return true; +} + +async function runRestartTest() { + const deadline = Date.now() + options.testDurationSeconds * 1_000; + let cycle = 0; + while (Date.now() < deadline) { + cycle += 1; + logTest(`Cycle ${cycle}: stopping ELS.`); + docker(['stop', '--time', '10', elsContainer]); + await waitFor('ELS shutdown', async () => await endpointStatus(`${elsBaseUri}/health/liveness`) === 0, 15); + logTest(`Cycle ${cycle}: ELS is stopped; SDK status is ${clientStatus || 'Unavailable'}.`); + await sleep(options.downtimeSeconds * 1_000); + docker(['start', elsContainer]); + await waitForEls(); + logTest(`Cycle ${cycle}: ELS is healthy again; SDK status is ${clientStatus || 'Unavailable'}.`); + if (await confirmClosed(cycle)) return; + await waitFor(`Cycle ${cycle} SDK recovery`, + async () => clientStatus === 'Ready' || clientStatus === 'Closed', options.recoveryTimeoutSeconds); + if (await confirmClosed(cycle)) return; + logTest(`Cycle ${cycle}: SDK reconnected and returned to Ready.`); + } + logTest(`Completed ${cycle} restart cycles within the requested test interval.`); +} + +try { + step('Resolving test configuration'); + for (const command of ['node', 'dotnet', 'docker']) run(command, ['--version']); + resolveEnvironment(); + console.log(`SDK source: ${sdkProject}`); + console.log(`Test time: ${options.testDurationSeconds} seconds`); + console.log(`Temporary files: ${artifactDirectory}`); + + step('Preparing SDK client and ELS'); + ensureElsContainer(); + await waitForEls(); + startClient(); + environmentSecret = ''; + await waitFor('SDK initial synchronization', async () => clientStatus === 'Ready', 90); + + step('Running ELS restart cycles'); + testStarted = true; + await runRestartTest(); + step(issueReproduced ? 'Issue reproduced' : 'Test interval completed without reproducing the issue'); +} finally { + environmentSecret = ''; + await stopClient(issueReproduced ? 'after reproducing the reconnect issue' : 'after completing the test interval'); + if (issueReproduced) { + logTest(`RESULT: Issue reproduced. SDK entered terminal Closed state at ${timestamp(sdkClosedAt)}.`); + } else if (testStarted) { + logTest('RESULT: Test completed without observing a persistent SDK Closed state.'); + } else { + logTest('RESULT: Test aborted before the ELS restart phase.'); + } + if (clientStoppedAt !== null) logTest(`RESULT: Test harness stopped the SDK client at ${timestamp(clientStoppedAt)}.`); + if (!options.keepArtifacts) rmSync(artifactDirectory, { recursive: true, force: true }); + if (options.stopElsAfterTest) docker(['stop', elsContainer], { allowFailure: true }); + else { + docker(['start', elsContainer], { allowFailure: true }); + await waitForEls(); + } + console.log(`SDK client log: ${clientLogPath}`); + console.log(`Test log: ${testLogPath}`); +} + +if (issueReproduced) process.exitCode = 2;