gsadmin/unleash
1
0
Fork 0
mirror of https://github.com/Unleash/unleash.git synced 2026-09-10 06:05:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: make client registration writes resilient to deadlocks (#12619)

Browse Source 
## About the changes

Concurrent registration flushes can acquire overlapping application and
instance row locks in different orders, causing deadlocks. A failed
flush also discards its buffered registrations.

Sort bulk writes by their conflict keys and coalesce application
observations by `app_name`, keeping the greatest `seen_at` within each
batch. Keep the original applications-first order: commit applications
and their usage rows together, then persist instance heartbeats
separately. Sorting addresses lock ordering; swapping these calls has no
demonstrated deadlock-prevention benefit.

Restore failed batches while preserving newer registrations received
during persistence. Correct application `lastSeen` mapping to the
persisted `seen_at` column and type the database row mappings.

Fixes #11390. Consistent ordering addresses opposite-order batch
deadlocks; it does not eliminate ordinary lock waits or reduce write
volume.

## OSS PR checklist

- [x] I have read and agree to the [Unleash Contributor License
Agreement](https://github.com/Unleash/unleash/blob/main/CLA.md).
- [x] I have added tests or explained why tests are not needed.
- [x] I have updated documentation where relevant. No configuration or
schema changes require documentation updates.
This commit is contained in:
Gastón Fournier
2026-09-09 12:01:42 +02:00
committed by GitHub
parent ac2cadad83
commit 1bebbafece
7 changed files with 195 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/db/client-applications-store.test.ts
+23
View File
@@ -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', () => {
src/lib/db/client-applications-store.ts
+106 -46
View File
@@ -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<IClientApplication, 'lastSeen'> & {
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<IClientApplication>) => {
const remapRow = (
input: Partial<IClientApplication>,
): PersistedClientApplicationWriteModel => {
const temp = {
app_name: input.appName,
updated_at: input.updatedAt || new Date(),
@@ -114,6 +146,27 @@ const remapRow = (input: Partial<IClientApplication>) => {
return temp;
};
const coalesceApplicationRows = (
rows: PersistedClientApplicationWriteModel[],
): ClientApplicationUpsertRow[] => {
const rowsByAppName = new Map<string, ClientApplicationUpsertRow>();
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<IClientApplication>[]): Promise<void> {
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<Record<string, ClientApplicationUsageWriteModel>>(
(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<IClientApplication>) => {
private remapUsageRow(
input: Partial<IClientApplication>,
): ClientApplicationUsageWriteModel[] {
if (!input.projects || input.projects.length === 0) {
return [
{
@@ -505,7 +565,7 @@ export default class ClientApplicationsStore
environment: input.environment || '*',
}));
}
};
}
async removeInactiveApplications(): Promise<number> {
const stopTimer = this.timer('removeInactiveApplications');
src/lib/db/client-instance-store.ts
+14 -11
View File
@@ -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>): 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<IClientInstance> => {
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<IClientInstance>;
};
export default class ClientInstanceStore implements IClientInstanceStore {
@@ -86,7 +82,14 @@ export default class ClientInstanceStore implements IClientInstanceStore {
async bulkUpsert(instances: INewClientInstance[]): Promise<void> {
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'])
src/lib/features/metrics/client-metrics/client-metrics-service.e2e.test.ts
-3
View File
@@ -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,
src/lib/features/metrics/instance/instance-service.test.ts
+38
View File
@@ -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();
src/lib/features/metrics/instance/instance-service.ts
+10 -2
View File
@@ -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<IClientApplication>[] = 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);
}
}
src/lib/features/metrics/instance/metrics.test.ts
+4 -1
View File
@@ -481,7 +481,10 @@ describe('bulk metrics', () => {
},
{
project: 'project-b',
environments: ['production', 'development'],
environments: expect.arrayContaining([
'development',
'production',
]),
},
{
project: 'project-c',