diff --git a/src/lib/db/client-applications-store.test.ts b/src/lib/db/client-applications-store.test.ts index 3f6f4df1f7..8bad26bc5d 100644 --- a/src/lib/db/client-applications-store.test.ts +++ b/src/lib/db/client-applications-store.test.ts @@ -23,6 +23,29 @@ describe('ClientApplicationsStore', () => { service = db.stores.clientApplicationsStore as ClientApplicationsStore; }); + test('bulkUpsert keeps the latest observation for an application', async () => { + const appName = 'coalesced-application'; + const latestSeenAt = new Date('2026-09-07T08:00:00.000Z'); + + await service.bulkUpsert([ + { appName, lastSeen: latestSeenAt, description: 'Latest metadata' }, + { + appName, + lastSeen: new Date('2026-09-06T08:00:00.000Z'), + description: 'Older metadata', + }, + ]); + + const application = await db + .rawDatabase('client_applications') + .select('seen_at', 'description') + .where({ app_name: appName }) + .first(); + + expect(application.seen_at).toEqual(latestSeenAt); + expect(application.description).toBe('Latest metadata'); + }); + describe('mapApplicationOverviewData()', () => { describe('handling deprecated strategies', () => { test('should not count any of the four deprecated strategies as missing', () => { diff --git a/src/lib/db/client-applications-store.ts b/src/lib/db/client-applications-store.ts index 74b1e431cb..72835239f7 100644 --- a/src/lib/db/client-applications-store.ts +++ b/src/lib/db/client-applications-store.ts @@ -5,6 +5,7 @@ import type { IClientApplications, IClientApplicationsSearchParams, IClientApplicationsStore, + IClientApplicationUsage, } from '../types/stores/client-applications-store.js'; import type { Logger, LogProvider } from '../logger.js'; import type { Db } from './db.js'; @@ -13,6 +14,7 @@ import { applySearchFilters } from '../features/feature-search/search-utils.js'; import type { IFlagResolver } from '../types/index.js'; import metricsHelper from '../util/metrics-helper.js'; import { DB_TIME } from '../metric-events.js'; +import type { Row } from '../server-impl.js'; const COLUMNS = [ 'app_name', @@ -24,6 +26,7 @@ const COLUMNS = [ 'url', 'color', 'icon', + 'seen_at', ]; const TABLE = 'client_applications'; @@ -36,21 +39,48 @@ const DEPRECATED_STRATEGIES = [ 'userWithId', ]; -const mapRow: (any) => IClientApplication = (row) => ({ - appName: row.app_name, - createdAt: row.created_at, - updatedAt: row.updated_at, - description: row.description, - strategies: row.strategies || [], - createdBy: row.created_by, - url: row.url, - color: row.color, - icon: row.icon, - lastSeen: row.last_seen, - announced: row.announced, - project: row.project, - environment: row.environment, -}); +type PersistedClientApplicationReadModel = Row< + Omit & { + seenAt: Date; + project: string; // project is stored in the usage table, but's legacy + } +>; +type PersistedClientApplicationWriteModel = { + app_name?: string; + seen_at: Date; + updated_at: Date; + description?: string; + created_by?: string; + announced?: boolean; + url?: string; + color?: string; + icon?: string; + strategies?: string; +}; +type ClientApplicationUpsertRow = PersistedClientApplicationWriteModel & { + app_name: string; +}; +type ClientApplicationUsageWriteModel = { + app_name?: string; + project: string; + environment: string; +}; +const mapRow: (row: PersistedClientApplicationReadModel) => IClientApplication = + (row) => ({ + appName: row.app_name, + createdAt: row.created_at, + updatedAt: row.updated_at, + description: row.description, + strategies: row.strategies || [], + createdBy: row.created_by, + url: row.url, + color: row.color, + icon: row.icon, + lastSeen: row.seen_at, + announced: row.announced, + project: row.project, + environment: row.environment, + }); const reduceRows = (rows: any[]): IClientApplication[] => { const appsObj = rows.reduce((acc, row) => { @@ -60,7 +90,7 @@ const reduceRows = (rows: any[]): IClientApplication[] => { if (existingApp) { const existingProject = existingApp.usage.find( - (usage) => usage.project === project, + (usage: IClientApplicationUsage) => usage.project === project, ); if (existingProject) { @@ -92,7 +122,9 @@ const reduceRows = (rows: any[]): IClientApplication[] => { return Object.values(appsObj); }; -const remapRow = (input: Partial) => { +const remapRow = ( + input: Partial, +): PersistedClientApplicationWriteModel => { const temp = { app_name: input.appName, updated_at: input.updatedAt || new Date(), @@ -114,6 +146,27 @@ const remapRow = (input: Partial) => { return temp; }; +const coalesceApplicationRows = ( + rows: PersistedClientApplicationWriteModel[], +): ClientApplicationUpsertRow[] => { + const rowsByAppName = new Map(); + + for (const row of rows) { + if (!row.app_name) { + continue; + } + + const existing = rowsByAppName.get(row.app_name); + const winner = + !existing || row.seen_at >= existing.seen_at ? row : existing; + rowsByAppName.set(row.app_name, { ...winner, app_name: row.app_name }); + } + + return [...rowsByAppName.values()].sort((a, b) => + a.app_name.localeCompare(b.app_name), + ); +}; + export default class ClientApplicationsStore implements IClientApplicationsStore { @@ -155,38 +208,43 @@ export default class ClientApplicationsStore async bulkUpsert(apps: Partial[]): Promise { const stopTimer = this.timer('bulkUpsert'); - const rows = apps.map(remapRow); - const uniqueRows = Object.values( - rows.reduce((acc, row) => { - if (row.app_name) { - acc[row.app_name] = row; - } - return acc; - }, {}), + const uniqueSortedRows = coalesceApplicationRows( + apps.map((app) => remapRow(app)), ); const usageRows = apps.flatMap(this.remapUsageRow); - const uniqueUsageRows = Object.values( - usageRows.reduce((acc, row) => { - if (row.app_name) { - acc[`${row.app_name} ${row.project} ${row.environment}`] = - row; - } - return acc; - }, {}), + const uniqueSortedUsageRows = Object.values( + usageRows.reduce>( + (acc, row) => { + if (row.app_name) { + acc[ + `${row.app_name} ${row.project} ${row.environment}` + ] = row; + } + return acc; + }, + {}, + ), + ).sort( + (a, b) => + (a.app_name ?? '').localeCompare(b.app_name ?? '') || + a.project.localeCompare(b.project) || + a.environment.localeCompare(b.environment), ); - await this.db(TABLE) - .insert(uniqueRows) - .onConflict('app_name') - .merge({ - updated_at: this.db.raw('EXCLUDED.updated_at'), - seen_at: this.db.raw('EXCLUDED.seen_at'), - }); + await this.db.transaction(async (transaction) => { + await transaction(TABLE) + .insert(uniqueSortedRows) + .onConflict('app_name') + .merge({ + updated_at: transaction.raw('EXCLUDED.updated_at'), + seen_at: transaction.raw('EXCLUDED.seen_at'), + }); - await this.db(TABLE_USAGE) - .insert(uniqueUsageRows) - .onConflict(['app_name', 'project', 'environment']) - .ignore(); + await transaction(TABLE_USAGE) + .insert(uniqueSortedUsageRows) + .onConflict(['app_name', 'project', 'environment']) + .ignore(); + }); stopTimer(); } @@ -489,7 +547,9 @@ export default class ClientApplicationsStore }; } - private remapUsageRow = (input: Partial) => { + private remapUsageRow( + input: Partial, + ): ClientApplicationUsageWriteModel[] { if (!input.projects || input.projects.length === 0) { return [ { @@ -505,7 +565,7 @@ export default class ClientApplicationsStore environment: input.environment || '*', })); } - }; + } async removeInactiveApplications(): Promise { const stopTimer = this.timer('removeInactiveApplications'); diff --git a/src/lib/db/client-instance-store.ts b/src/lib/db/client-instance-store.ts index a59cc4c01f..22e9614283 100644 --- a/src/lib/db/client-instance-store.ts +++ b/src/lib/db/client-instance-store.ts @@ -9,6 +9,7 @@ import { subDays } from 'date-fns'; import type { Db } from './db.js'; import metricsHelper from '../util/metrics-helper.js'; import { DB_TIME } from '../metric-events.js'; +import type { Row } from '../server-impl.js'; const COLUMNS = [ 'app_name', @@ -21,7 +22,7 @@ const COLUMNS = [ ]; const TABLE = 'client_instances'; -const mapRow = (row): IClientInstance => ({ +const mapRow = (row: Row): IClientInstance => ({ appName: row.app_name, instanceId: row.instance_id, sdkVersion: row.sdk_version, @@ -32,7 +33,7 @@ const mapRow = (row): IClientInstance => ({ environment: row.environment, }); -const mapToDb = (client: INewClientInstance) => { +const mapToDb = (client: INewClientInstance): Row => { const temp = { app_name: client.appName, instance_id: client.instanceId, @@ -43,14 +44,9 @@ const mapToDb = (client: INewClientInstance) => { environment: client.environment, }; - const result = {}; - for (const [key, value] of Object.entries(temp)) { - if (value !== undefined) { - result[key] = value; - } - } - - return result; + return Object.fromEntries( + Object.entries(temp).filter(([_, value]) => value !== undefined), + ) as Row; }; export default class ClientInstanceStore implements IClientInstanceStore { @@ -86,7 +82,14 @@ export default class ClientInstanceStore implements IClientInstanceStore { async bulkUpsert(instances: INewClientInstance[]): Promise { const stopTimer = this.metricTimer('bulkUpsert'); - const rows = instances.map(mapToDb); + const rows = [...instances] + .sort( + (a, b) => + a.appName.localeCompare(b.appName) || + a.instanceId.localeCompare(b.instanceId) || + (a.environment ?? '').localeCompare(b.environment ?? ''), + ) + .map(mapToDb); await this.db(TABLE) .insert(rows) .onConflict(['app_name', 'instance_id', 'environment']) diff --git a/src/lib/features/metrics/client-metrics/client-metrics-service.e2e.test.ts b/src/lib/features/metrics/client-metrics/client-metrics-service.e2e.test.ts index 8ec9b4e688..82504a6a0b 100644 --- a/src/lib/features/metrics/client-metrics/client-metrics-service.e2e.test.ts +++ b/src/lib/features/metrics/client-metrics/client-metrics-service.e2e.test.ts @@ -1,6 +1,5 @@ import ClientInstanceService from '../instance/instance-service.js'; import type { IClientApp } from '../../../types/model.js'; -import { secondsToMilliseconds } from 'date-fns'; import { createTestConfig } from '../../../../test/config/test-config.js'; import type { IUnleashConfig, IUnleashStores } from '../../../types/index.js'; import { APPLICATION_CREATED } from '../../../events/index.js'; @@ -17,8 +16,6 @@ beforeAll(async () => { db = await dbInit('client_metrics_service_serial', getLogger); stores = db.stores; config = createTestConfig({}); - const _bulkInterval = secondsToMilliseconds(0.5); - const _announcementInterval = secondsToMilliseconds(2); clientInstanceService = new ClientInstanceService( stores, diff --git a/src/lib/features/metrics/instance/instance-service.test.ts b/src/lib/features/metrics/instance/instance-service.test.ts index a952df7e7d..084fe5ac16 100644 --- a/src/lib/features/metrics/instance/instance-service.test.ts +++ b/src/lib/features/metrics/instance/instance-service.test.ts @@ -198,6 +198,44 @@ test('No registrations during a time period will not call stores', async () => { expect(bulkSpy).toHaveBeenCalledTimes(0); }); +test('restores a failed batch without overwriting newer registrations', async () => { + let clientMetrics: ClientInstanceService; + const instanceStoreSpy = vi.fn(async () => { + await clientMetrics.registerInstance( + { appName: 'test-app', instanceId: 'test-instance' }, + 'new-client-ip', + 'new-environment', + ); + throw new Error('deadlock detected'); + }); + clientMetrics = new ClientInstanceService( + { + clientMetricsStoreV2: new FakeClientMetricsStoreV2(), + strategyStore: new FakeStrategiesStore(), + featureToggleStore: new FakeFeatureToggleStore(), + clientApplicationsStore: { bulkUpsert: vi.fn() } as any, + clientInstanceStore: { bulkUpsert: instanceStoreSpy } as any, + eventStore: new FakeEventStore(), + }, + config, + new FakePrivateProjectChecker(), + ); + + await clientMetrics.registerInstance( + { appName: 'test-app', instanceId: 'test-instance' }, + 'old-client-ip', + 'old-environment', + ); + await clientMetrics.bulkAdd(); + + expect(clientMetrics.seenClients['test-app_test-instance']).toMatchObject({ + appName: 'test-app', + instanceId: 'test-instance', + clientIp: 'new-client-ip', + environment: 'new-environment', + }); +}); + test('registrations without an app name are ignored without dropping valid registrations', async () => { const appStoreSpy = vi.fn(); const instanceStoreSpy = vi.fn(); diff --git a/src/lib/features/metrics/instance/instance-service.ts b/src/lib/features/metrics/instance/instance-service.ts index bc55965106..fd52dad5b2 100644 --- a/src/lib/features/metrics/instance/instance-service.ts +++ b/src/lib/features/metrics/instance/instance-service.ts @@ -217,7 +217,9 @@ export default class ClientInstanceService { this.clientApplicationsStore && this.clientInstanceStore ) { - const uniqueRegistrations = Object.values(this.seenClients).filter( + const pendingClients = this.seenClients; + this.seenClients = {}; + const uniqueRegistrations = Object.values(pendingClients).filter( (client) => Boolean(client.appName), ); const uniqueApps: Partial[] = Object.values( @@ -238,7 +240,6 @@ export default class ClientInstanceService { return soFar; }, {}), ); - this.seenClients = {}; try { if (uniqueRegistrations.length > 0) { await this.clientApplicationsStore.bulkUpsert(uniqueApps); @@ -247,6 +248,13 @@ export default class ClientInstanceService { ); } } catch (err) { + // restore on error + for (const [key, client] of Object.entries(pendingClients)) { + this.seenClients[key] = { + ...client, + ...this.seenClients[key], + }; + } this.logger.warn('Failed to register clients', err); } } diff --git a/src/lib/features/metrics/instance/metrics.test.ts b/src/lib/features/metrics/instance/metrics.test.ts index cbe62640c3..85b4293b85 100644 --- a/src/lib/features/metrics/instance/metrics.test.ts +++ b/src/lib/features/metrics/instance/metrics.test.ts @@ -481,7 +481,10 @@ describe('bulk metrics', () => { }, { project: 'project-b', - environments: ['production', 'development'], + environments: expect.arrayContaining([ + 'development', + 'production', + ]), }, { project: 'project-c',