From a2f7367671be608c38955dce7841005955f0116f Mon Sep 17 00:00:00 2001 From: Dominik Ferber Date: Sat, 15 Aug 2026 08:37:32 +0300 Subject: [PATCH] [flags-core] re-wire source events when a source restarts shutdown() unwires the stream and polling event handlers, but nothing wired them again. A client initialized after a shutdown opened a new connection and then ignored every datafile that connection delivered, so evaluations returned the default value with an error reason. The failure was silent: the socket was healthy and the fetch happened, only the handlers were missing. wireSourceEvents() now runs at the start of tryInitializeStream() and tryInitializePolling(), the only two places that start a source. Repeat calls cannot duplicate a subscription, because the handlers are stable instance properties and TypedEmitter keeps them in a Set. Two tests cover that invariant: one across a shutdown, and one for a source that starts twice within a single lifecycle, which happens when a failed initialization is followed by an evaluation. This also affects the VercelProvider OpenFeature provider, whose onClose() hook shuts the client down, since section 2.5.2 of the provider spec allows a provider to be initialized again after it is closed. Co-Authored-By: Claude Opus 5 (1M context) --- .changeset/lucky-pandas-restart.md | 11 + packages/vercel-flags-core/CLAUDE.md | 6 + .../vercel-flags-core/src/black-box.test.ts | 255 ++++++++++++++++++ .../vercel-flags-core/src/controller/index.ts | 11 + 4 files changed, 283 insertions(+) create mode 100644 .changeset/lucky-pandas-restart.md diff --git a/.changeset/lucky-pandas-restart.md b/.changeset/lucky-pandas-restart.md new file mode 100644 index 00000000..8ee35552 --- /dev/null +++ b/.changeset/lucky-pandas-restart.md @@ -0,0 +1,11 @@ +--- +'@vercel/flags-core': patch +--- + +Apply flag data again after a shutdown and a new initialization + +`shutdown()` removes the event handlers of the stream source and the polling source. But no code added these handlers again. Thus a client that you initialize after a shutdown opens a new connection, but it ignores all data from that connection. Evaluations then give the default value, with the reason `error`. + +This condition also applies to the `VercelProvider` OpenFeature provider. Its `onClose()` hook shuts the client down, and the OpenFeature specification permits a new initialization after a shutdown. + +The handlers are now added each time a source starts. A client that you initialize again thus gets new data, as expected. diff --git a/packages/vercel-flags-core/CLAUDE.md b/packages/vercel-flags-core/CLAUDE.md index e2adcaa4..ee5782c1 100644 --- a/packages/vercel-flags-core/CLAUDE.md +++ b/packages/vercel-flags-core/CLAUDE.md @@ -273,6 +273,12 @@ The Controller tags all data with its origin using `tagData(data, origin)` from - Supports multiple simultaneous clients - Necessary as we can't pass functions to `'use cache'` wrappers +### Restart After Shutdown + +`shutdown()` calls `unwireSourceEvents()`, so the source event handlers are no longer subscribed once it returns. `wireSourceEvents()` therefore runs at the start of both `tryInitializeStream()` and `tryInitializePolling()` — the only two places that start a source — so a client that is initialized again after a shutdown applies the data its new connection delivers. Without that, the stream reconnects but every datafile it sends is silently ignored. + +Re-wiring is safe to repeat: the handlers are stable instance properties and `TypedEmitter` keeps them in a `Set`, so a handler can not be registered twice. + ### configUpdatedAt Guard The Controller rejects incoming data (from stream or poll) if its `configUpdatedAt` is older than or equal to the current in-memory data. This prevents stale updates from overwriting newer data. Accepts the update if either side lacks a `configUpdatedAt`. diff --git a/packages/vercel-flags-core/src/black-box.test.ts b/packages/vercel-flags-core/src/black-box.test.ts index 281524a7..0afe19ab 100644 --- a/packages/vercel-flags-core/src/black-box.test.ts +++ b/packages/vercel-flags-core/src/black-box.test.ts @@ -33,6 +33,8 @@ const fetchMock = vi.fn(); * Returns a controller object that lets you gradually push messages * and a `response` promise suitable for use with a fetch mock. */ +type MockStream = ReturnType; + function createMockStream() { const encoder = new TextEncoder(); let controller: ReadableStreamDefaultController; @@ -3602,6 +3604,259 @@ describe('Controller (black-box)', () => { }); }); + // --------------------------------------------------------------------------- + // Restart after shutdown + // --------------------------------------------------------------------------- + describe('restart after shutdown', () => { + /** Builds a flagA datafile whose single environment resolves to `value`. */ + function makeFlagA(revision: number, value: boolean): BundledDefinitions { + return makeBundled({ + revision, + configUpdatedAt: revision, + definitions: { + flagA: { + environments: { production: value ? 1 : 0 }, + variants: [false, true], + }, + }, + }); + } + + it('should apply stream data after initialize, shutdown, initialize', async () => { + const streams = [createMockStream(), createMockStream()]; + let streamIndex = 0; + + fetchMock.mockImplementation((input) => { + const url = typeof input === 'string' ? input : input.toString(); + if (url.includes('/v1/stream')) { + const stream = streams[streamIndex++]; + if (!stream) { + return Promise.reject( + new Error(`Unexpected stream connection #${streamIndex}`), + ); + } + return stream.response; + } + if (url.includes('/v1/ingest')) return Promise.resolve(new Response()); + return Promise.reject(new Error(`Unexpected fetch: ${url}`)); + }); + + const [firstStream, secondStream] = streams as [MockStream, MockStream]; + const client = createClient(sdkKey, { fetch: fetchMock, polling: false }); + + const firstInit = client.initialize(); + await vi.advanceTimersByTimeAsync(0); + firstStream.push({ type: 'datafile', data: makeFlagA(1, true) }); + await firstInit; + + expect((await client.evaluate('flagA', false)).value).toBe(true); + + await client.shutdown(); + + // The second connection must be wired up again, so its datafile applies. + const secondInit = client.initialize(); + await vi.advanceTimersByTimeAsync(0); + secondStream.push({ type: 'datafile', data: makeFlagA(2, false) }); + await secondInit; + + const result = await client.evaluate('flagA', true); + expect(result.value).toBe(false); + expect(result.reason).not.toBe('error'); + expect(result.metrics?.source).toBe('in-memory'); + expect(result.metrics?.connectionState).toBe('connected'); + expect(streamIndex).toBe(2); + + firstStream.close(); + secondStream.close(); + await client.shutdown(); + }); + + it('should apply poll data after initialize, shutdown, initialize', async () => { + let revision = 1; + const datafileCalls: string[] = []; + + fetchMock.mockImplementation((input) => { + const url = typeof input === 'string' ? input : input.toString(); + if (url.includes('/v1/datafile')) { + datafileCalls.push(url); + return Promise.resolve( + new Response( + JSON.stringify(makeFlagA(revision, revision % 2 === 1)), + { status: 200, headers: { 'Content-Type': 'application/json' } }, + ), + ); + } + if (url.includes('/v1/ingest')) return Promise.resolve(new Response()); + return Promise.reject(new Error(`Unexpected fetch: ${url}`)); + }); + + const client = createClient(sdkKey, { fetch: fetchMock, stream: false }); + + await client.initialize(); + expect((await client.evaluate('flagA', false)).value).toBe(true); + + await client.shutdown(); + + revision = 2; + await client.initialize(); + + const result = await client.evaluate('flagA', true); + expect(result.value).toBe(false); + expect(result.reason).not.toBe('error'); + expect(result.metrics?.source).toBe('in-memory'); + expect(datafileCalls).toHaveLength(2); + + await client.shutdown(); + }); + + it('should subscribe poll handlers only once across a restart', async () => { + const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}); + let failPoll = false; + + fetchMock.mockImplementation((input) => { + const url = typeof input === 'string' ? input : input.toString(); + if (url.includes('/v1/datafile')) { + if (failPoll) return Promise.reject(new Error('poll boom')); + return Promise.resolve( + new Response(JSON.stringify(makeFlagA(1, true)), { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }), + ); + } + if (url.includes('/v1/ingest')) return Promise.resolve(new Response()); + return Promise.reject(new Error(`Unexpected fetch: ${url}`)); + }); + + const client = createClient(sdkKey, { + fetch: fetchMock, + stream: false, + polling: { intervalMs: 30_000, initTimeoutMs: 3000 }, + }); + + await client.initialize(); + await client.shutdown(); + await client.initialize(); + + // A single failing interval poll must be reported exactly once. A second + // subscription of the same handler would log it twice. + failPoll = true; + await vi.advanceTimersByTimeAsync(30_000); + + expect(errorSpy).toHaveBeenCalledTimes(1); + expect(errorSpy).toHaveBeenCalledWith( + '@vercel/flags-core: Poll failed:', + expect.objectContaining({ message: 'poll boom' }), + ); + + await client.shutdown(); + }); + + it('should keep working across repeated restarts', async () => { + const streams = [ + createMockStream(), + createMockStream(), + createMockStream(), + ]; + let streamIndex = 0; + + fetchMock.mockImplementation((input) => { + const url = typeof input === 'string' ? input : input.toString(); + if (url.includes('/v1/stream')) { + const stream = streams[streamIndex++]; + if (!stream) { + return Promise.reject( + new Error(`Unexpected stream connection #${streamIndex}`), + ); + } + return stream.response; + } + if (url.includes('/v1/ingest')) return Promise.resolve(new Response()); + return Promise.reject(new Error(`Unexpected fetch: ${url}`)); + }); + + const client = createClient(sdkKey, { fetch: fetchMock, polling: false }); + + for (const [round, stream] of streams.entries()) { + const initPromise = client.initialize(); + await vi.advanceTimersByTimeAsync(0); + stream.push({ + type: 'datafile', + data: makeFlagA(round + 1, round % 2 === 0), + }); + await initPromise; + + const result = await client.evaluate('flagA', false); + expect(result.value).toBe(round % 2 === 0); + + await client.shutdown(); + } + + expect(streamIndex).toBe(3); + for (const stream of streams) stream.close(); + }); + + it('should subscribe handlers once when a source starts twice without a shutdown', async () => { + const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}); + const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); + vi.mocked(readBundledDefinitions).mockRejectedValue( + new Error('no bundled definitions'), + ); + + let failPoll = true; + fetchMock.mockImplementation((input) => { + const url = typeof input === 'string' ? input : input.toString(); + if (url.includes('/v1/datafile')) { + if (failPoll) return Promise.reject(new Error('poll boom')); + return Promise.resolve( + new Response(JSON.stringify(makeFlagA(1, true)), { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }), + ); + } + if (url.includes('/v1/ingest')) return Promise.resolve(new Response()); + return Promise.reject(new Error(`Unexpected fetch: ${url}`)); + }); + + const pollErrorCount = () => + errorSpy.mock.calls.filter( + (call) => call[0] === '@vercel/flags-core: Poll failed:', + ).length; + + // Metrics are disabled so the assertions below see only poll errors and + // no usage batch outlives the test. + const client = createClient(sdkKey, { + fetch: fetchMock, + stream: false, + disableMetrics: true, + }); + + // First source start: the poll fails and no fallback data exists, so + // initialization rejects without ever caching data. + await expect(client.initialize()).rejects.toThrow( + 'no bundled definitions', + ); + expect(pollErrorCount()).toBe(1); + + // Second source start, with no shutdown in between: evaluating without + // cached data runs the fallback chain, which polls again. + failPoll = false; + expect((await client.evaluate('flagA', false)).value).toBe(true); + + // One failing interval poll must still be reported exactly once. A + // second subscription of the same handler would report it twice. + failPoll = true; + await vi.advanceTimersByTimeAsync(30_000); + + expect(pollErrorCount()).toBe(2); + expect(errorSpy).toHaveBeenCalledTimes(2); + expect(warnSpy).not.toHaveBeenCalled(); + + await client.shutdown(); + }); + }); + // --------------------------------------------------------------------------- // Lazy initialization // --------------------------------------------------------------------------- diff --git a/packages/vercel-flags-core/src/controller/index.ts b/packages/vercel-flags-core/src/controller/index.ts index 5f55c126..19a29ce7 100644 --- a/packages/vercel-flags-core/src/controller/index.ts +++ b/packages/vercel-flags-core/src/controller/index.ts @@ -183,6 +183,11 @@ export class Controller implements ControllerInterface { // Source event wiring // --------------------------------------------------------------------------- + /** + * Subscribes to source events. Safe to call repeatedly: the handlers are + * stable instance properties and the emitter stores them in a Set, so + * re-wiring after a shutdown cannot register a handler twice. + */ private wireSourceEvents(): void { this.streamSource.on('data', this.onStreamData); this.streamSource.on('primed', this.onStreamPrimed); @@ -458,6 +463,9 @@ export class Controller implements ControllerInterface { * Returns true if stream connected successfully within timeout. */ private async tryInitializeStream(): Promise { + // A shutdown unwired the handlers, so re-wire before the stream can emit. + this.wireSourceEvents(); + if (this.options.stream.initTimeoutMs <= 0) { try { await this.streamSource.start(); @@ -518,6 +526,9 @@ export class Controller implements ControllerInterface { * Only used when streaming is disabled and polling is the primary source. */ private async tryInitializePolling(): Promise { + // A shutdown unwired the handlers, so re-wire before the first poll emits. + this.wireSourceEvents(); + const pollPromise = this.pollingSource.poll(); if (this.options.polling.initTimeoutMs <= 0) {