From 0d5353da76aaf14cb927f9257b4ac121c3c3f620 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Sun, 1 Nov 2026 09:57:04 +0100 Subject: [PATCH 1/2] perf: skip unchanged organization writes in data-sink member updates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit findOrCreateOrganization rewrote every organization it found: the identity and attribute upserts ran on every call and markOrgAttributeDefault updated every attribute row of its name. It now reads the stored attributes and identities once, works out which writes would change anything, and returns before taking the lock when none would. markOrgAttributeDefault also skips rows whose flag is already right. Refs linuxfoundation/lfx-cdp-insights-ops#969 Signed-off-by: Uroš Marolt --- .../src/organizations/attributes.ts | 29 ++++ .../src/organizations/base.ts | 139 +++++++++++++----- .../src/organizations/identities.ts | 24 ++- 3 files changed, 153 insertions(+), 39 deletions(-) diff --git a/services/libs/data-access-layer/src/organizations/attributes.ts b/services/libs/data-access-layer/src/organizations/attributes.ts index 280e5f8937..4acb89d264 100644 --- a/services/libs/data-access-layer/src/organizations/attributes.ts +++ b/services/libs/data-access-layer/src/organizations/attributes.ts @@ -86,6 +86,34 @@ export const upsertOrgAttributes = async ( } } +const isSameOrgAttribute = ( + stored: IDbOrgAttribute, + attribute: { name: string; source: string; value: string }, +): boolean => + stored.name === attribute.name && + stored.source === attribute.source && + stored.value === attribute.value + +// Must match the conflict key of upsertOrgAttributes: organizationId, name, source, md5(value) +export function hasAllOrgAttributes( + stored: IDbOrgAttribute[], + attributes: IDbOrgAttributeInput[], +): boolean { + return attributes.every((attribute) => stored.some((s) => isSameOrgAttribute(s, attribute))) +} + +// Must match the flags markOrgAttributeDefault sets +export function isOrgAttributeDefault( + stored: IDbOrgAttribute[], + attribute: { name: string; source: string; value: string }, +): boolean { + const sameName = stored.filter((s) => s.name === attribute.name) + return ( + sameName.some((s) => isSameOrgAttribute(s, attribute)) && + sameName.every((s) => s.default === isSameOrgAttribute(s, attribute)) + ) +} + export async function markOrgAttributeDefault( qx: QueryExecutor, organizationId: string, @@ -109,6 +137,7 @@ export async function markOrgAttributeDefault( END WHERE "organizationId" = $(organizationId) AND name = $(name) + AND "default" IS DISTINCT FROM (source = $(source) AND value = $(value)) `, { organizationId, diff --git a/services/libs/data-access-layer/src/organizations/base.ts b/services/libs/data-access-layer/src/organizations/base.ts index dc4579f622..fe39e151ac 100644 --- a/services/libs/data-access-layer/src/organizations/base.ts +++ b/services/libs/data-access-layer/src/organizations/base.ts @@ -22,9 +22,23 @@ import { QueryExecutor } from '../queryExecutor' import { findManyLfSegmentsByNames } from '../segments' import { QueryOptions, QueryResult, prepareBulkInsert, queryTable, queryTableById } from '../utils' import { prepareSelectColumns } from '../utils' -import { findOrgAttributes, markOrgAttributeDefault, upsertOrgAttributes } from './attributes' -import { insertOrganizationIdentities, upsertOrgIdentities } from './identities' -import { IDbOrganization, IDbOrganizationInput, IFindOrCreateOrganizationResult } from './types' +import { + hasAllOrgAttributes, + isOrgAttributeDefault, + markOrgAttributeDefault, + upsertOrgAttributes, +} from './attributes' +import { + hasAllOrgIdentities, + insertOrganizationIdentities, + upsertOrgIdentities, +} from './identities' +import { + IDbOrgAttribute, + IDbOrganization, + IDbOrganizationInput, + IFindOrCreateOrganizationResult, +} from './types' import { prepareOrganizationData } from './utils' const log = getServiceChildLogger('data-access-layer/organizations') @@ -530,6 +544,52 @@ async function findOrgForUpdate(qe: QueryExecutor, id: string): Promise { + return qe.selectOne( + ` + select + coalesce((select jsonb_agg(oa) from "orgAttributes" oa where oa."organizationId" = $(id)), '[]') as attributes, + coalesce((select jsonb_agg(oi) from "organizationIdentities" oi where oi."organizationId" = $(id)), '[]') as identities + `, + { id }, + ) +} + +function planOrgUpdate( + data: IOrganization, + source: string, + existing: IDbOrganization, + stored: IStoredOrg, +) { + // prepareOrganizationData rewrites the default flags of the attributes it gets, so it gets copies + const processed = prepareOrganizationData( + data, + source, + existing, + stored.attributes.map((a) => ({ ...a })), + ) + const updateOrganization = Object.keys(processed.organization).length > 0 + const upsertIdentities = !hasAllOrgIdentities(stored.identities, data.identities) + const upsertAttributes = !hasAllOrgAttributes(stored.attributes, processed.attributes) + const markDefaults = processed.attributes.filter( + (a) => a.default && !isOrgAttributeDefault(stored.attributes, a), + ) + return { + processed, + updateOrganization, + upsertIdentities, + upsertAttributes, + markDefaults, + changesNothing: + !updateOrganization && !upsertIdentities && !upsertAttributes && markDefaults.length === 0, + } +} + // lockOrganization: a caller that runs this concurrently without transactions passes a lock, so // calls that reach the same organization through different identities update it one at a time export async function findOrCreateOrganization( @@ -618,24 +678,29 @@ export async function findOrCreateOrganization( if (existing) { log.trace(`Found existing organization, organization will be updated!`) + let stored = await logExecutionTimeV2( + async () => findOrgAttributesAndIdentities(qe, existing.id), + log, + 'organizationService -> findOrCreateOrganization -> findOrgAttributesAndIdentities', + ) + let plan = planOrgUpdate(data, source, existing, stored) + if (plan.changesNothing) { + return { id: existing.id, created: false } + } + const release = await lockOrganization?.(existing.id) try { if (release) { // read it again: the call that held the lock may have changed it existing = (await findOrgForUpdate(qe, existing.id)) ?? existing + stored = await findOrgAttributesAndIdentities(qe, existing.id) + plan = planOrgUpdate(data, source, existing, stored) } - const existingAttributes = await logExecutionTimeV2( - async () => findOrgAttributes(qe, existing.id), - log, - 'organizationService -> findOrCreateOrganization -> findOrgAttributes', - ) - - const processed = prepareOrganizationData(data, source, existing, existingAttributes) - + const { processed } = plan log.trace({ updateData: processed.organization }, `Updating organization!`) - if (Object.keys(processed.organization).length > 0) { + if (plan.updateOrganization) { log.info({ orgId: existing.id }, `Updating organization!`) await logExecutionTimeV2( async () => @@ -644,30 +709,32 @@ export async function findOrCreateOrganization( 'organizationService -> findOrCreateOrganization -> updateOrganization', ) } - await logExecutionTimeV2( - async () => - upsertOrgIdentities( - qe, - existing.id, - data.identities.map((i) => ({ ...i, source })), - integrationId, - ), - log, - 'organizationService -> findOrCreateOrganization -> upsertOrgIdentities', - ) - await logExecutionTimeV2( - async () => upsertOrgAttributes(qe, existing.id, processed.attributes), - log, - 'organizationService -> findOrCreateOrganization -> upsertOrgAttributes', - ) - for (const attr of processed.attributes) { - if (attr.default) { - await logExecutionTimeV2( - async () => markOrgAttributeDefault(qe, existing.id, attr), - log, - 'organizationService -> findOrCreateOrganization -> markOrgAttributeDefault', - ) - } + if (plan.upsertIdentities) { + await logExecutionTimeV2( + async () => + upsertOrgIdentities( + qe, + existing.id, + data.identities.map((i) => ({ ...i, source })), + integrationId, + ), + log, + 'organizationService -> findOrCreateOrganization -> upsertOrgIdentities', + ) + } + if (plan.upsertAttributes) { + await logExecutionTimeV2( + async () => upsertOrgAttributes(qe, existing.id, processed.attributes), + log, + 'organizationService -> findOrCreateOrganization -> upsertOrgAttributes', + ) + } + for (const attr of plan.markDefaults) { + await logExecutionTimeV2( + async () => markOrgAttributeDefault(qe, existing.id, attr), + log, + 'organizationService -> findOrCreateOrganization -> markOrgAttributeDefault', + ) } id = existing.id diff --git a/services/libs/data-access-layer/src/organizations/identities.ts b/services/libs/data-access-layer/src/organizations/identities.ts index df7d9d73b7..2606fc7915 100644 --- a/services/libs/data-access-layer/src/organizations/identities.ts +++ b/services/libs/data-access-layer/src/organizations/identities.ts @@ -121,6 +121,26 @@ export async function insertOrganizationIdentities( return qx.result(query) } +function findStoredOrgIdentity( + stored: IOrganizationIdentity[], + identity: Partial, +): IOrganizationIdentity | undefined { + return stored.find( + (s) => + s.value === identity.value && s.platform === identity.platform && s.type === identity.type, + ) +} + +export function hasAllOrgIdentities( + stored: IOrganizationIdentity[], + identities: Partial[], +): boolean { + return identities.every((identity) => { + const existing = findStoredOrgIdentity(stored, identity) + return existing && (identity.verified === undefined || existing.verified === identity.verified) + }) +} + export async function upsertOrgIdentities( qe: QueryExecutor, organizationId: string, @@ -132,9 +152,7 @@ export async function upsertOrgIdentities( const toUpdate: Partial[] = [] for (const i of identities) { - const existing = existingIdentities.find( - (ei) => ei.value === i.value && ei.platform === i.platform && ei.type === i.type, - ) + const existing = findStoredOrgIdentity(existingIdentities, i) if (!existing) { toCreate.push({ organizationId, From bcf57175171f77c102ed45fcd2c3febb97733909 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Mon, 2 Nov 2026 11:25:11 +0100 Subject: [PATCH 2/2] fix: run every organization write as before once anything changes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The writes after the first read used decisions taken from that read. Without the per-organization lock another transaction could change the organization meanwhile, so a skipped identity or default write could leave its change unapplied, and two concurrent default changes could both keep their row. Unchanged organizations still return early; otherwise the identity and attribute upserts and the default marking run as before. Refs linuxfoundation/lfx-cdp-insights-ops#969 Signed-off-by: Uroš Marolt --- .../src/organizations/attributes.ts | 1 - .../src/organizations/base.ts | 69 +++++++++---------- 2 files changed, 31 insertions(+), 39 deletions(-) diff --git a/services/libs/data-access-layer/src/organizations/attributes.ts b/services/libs/data-access-layer/src/organizations/attributes.ts index 4acb89d264..03c0a0e4b8 100644 --- a/services/libs/data-access-layer/src/organizations/attributes.ts +++ b/services/libs/data-access-layer/src/organizations/attributes.ts @@ -137,7 +137,6 @@ export async function markOrgAttributeDefault( END WHERE "organizationId" = $(organizationId) AND name = $(name) - AND "default" IS DISTINCT FROM (source = $(source) AND value = $(value)) `, { organizationId, diff --git a/services/libs/data-access-layer/src/organizations/base.ts b/services/libs/data-access-layer/src/organizations/base.ts index fe39e151ac..38b8ad886e 100644 --- a/services/libs/data-access-layer/src/organizations/base.ts +++ b/services/libs/data-access-layer/src/organizations/base.ts @@ -573,20 +573,13 @@ function planOrgUpdate( existing, stored.attributes.map((a) => ({ ...a })), ) - const updateOrganization = Object.keys(processed.organization).length > 0 - const upsertIdentities = !hasAllOrgIdentities(stored.identities, data.identities) - const upsertAttributes = !hasAllOrgAttributes(stored.attributes, processed.attributes) - const markDefaults = processed.attributes.filter( - (a) => a.default && !isOrgAttributeDefault(stored.attributes, a), - ) return { processed, - updateOrganization, - upsertIdentities, - upsertAttributes, - markDefaults, changesNothing: - !updateOrganization && !upsertIdentities && !upsertAttributes && markDefaults.length === 0, + Object.keys(processed.organization).length === 0 && + hasAllOrgIdentities(stored.identities, data.identities) && + hasAllOrgAttributes(stored.attributes, processed.attributes) && + processed.attributes.every((a) => !a.default || isOrgAttributeDefault(stored.attributes, a)), } } @@ -697,10 +690,12 @@ export async function findOrCreateOrganization( plan = planOrgUpdate(data, source, existing, stored) } + // Once anything changes, every write runs as before: a decision taken from the earlier read + // could be stale by now for callers without the lock const { processed } = plan log.trace({ updateData: processed.organization }, `Updating organization!`) - if (plan.updateOrganization) { + if (Object.keys(processed.organization).length > 0) { log.info({ orgId: existing.id }, `Updating organization!`) await logExecutionTimeV2( async () => @@ -709,32 +704,30 @@ export async function findOrCreateOrganization( 'organizationService -> findOrCreateOrganization -> updateOrganization', ) } - if (plan.upsertIdentities) { - await logExecutionTimeV2( - async () => - upsertOrgIdentities( - qe, - existing.id, - data.identities.map((i) => ({ ...i, source })), - integrationId, - ), - log, - 'organizationService -> findOrCreateOrganization -> upsertOrgIdentities', - ) - } - if (plan.upsertAttributes) { - await logExecutionTimeV2( - async () => upsertOrgAttributes(qe, existing.id, processed.attributes), - log, - 'organizationService -> findOrCreateOrganization -> upsertOrgAttributes', - ) - } - for (const attr of plan.markDefaults) { - await logExecutionTimeV2( - async () => markOrgAttributeDefault(qe, existing.id, attr), - log, - 'organizationService -> findOrCreateOrganization -> markOrgAttributeDefault', - ) + await logExecutionTimeV2( + async () => + upsertOrgIdentities( + qe, + existing.id, + data.identities.map((i) => ({ ...i, source })), + integrationId, + ), + log, + 'organizationService -> findOrCreateOrganization -> upsertOrgIdentities', + ) + await logExecutionTimeV2( + async () => upsertOrgAttributes(qe, existing.id, processed.attributes), + log, + 'organizationService -> findOrCreateOrganization -> upsertOrgAttributes', + ) + for (const attr of processed.attributes) { + if (attr.default) { + await logExecutionTimeV2( + async () => markOrgAttributeDefault(qe, existing.id, attr), + log, + 'organizationService -> findOrCreateOrganization -> markOrgAttributeDefault', + ) + } } id = existing.id