From 4c3627aa928bc3589abe315ff6b2a8ab840b8b8c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Mon, 2 Nov 2026 12:39:15 +0100 Subject: [PATCH 1/7] fix: apply every distinct organization payload in a data-sink batch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The batch's organization cache was keyed by identities only, so when two payloads in one batch described the same organization differently, only the first one's data was written. Each distinct payload is now applied once. Payloads that share identities wait for each other and land on the same organization. Once the batch is done, for each organization and source that got more than one payload, the one from the newest activity (then the one most activities carry) is applied again so its values become the defaults of its source. addOrgsToSegments sorts its rows so concurrent inserts can't deadlock. Closes linuxfoundation/lfx-cdp-insights-ops#972 Signed-off-by: Uroš Marolt --- .../src/service/activity.service.ts | 36 +++- .../src/service/member.service.ts | 187 +++++++++++++++--- .../src/organizations/base.ts | 10 +- 3 files changed, 203 insertions(+), 30 deletions(-) diff --git a/services/apps/data_sink_worker/src/service/activity.service.ts b/services/apps/data_sink_worker/src/service/activity.service.ts index c212856ad5..65307eb634 100644 --- a/services/apps/data_sink_worker/src/service/activity.service.ts +++ b/services/apps/data_sink_worker/src/service/activity.service.ts @@ -63,7 +63,12 @@ import { import { IActivityUpdateData, ISentimentActivityInput } from './activity.data' import { ActivityTimestampsByEmailDomain, IMemberUpdateData } from './member.data' -import MemberService, { getEarliestMemberJoinedAt, mergeIfAllowed } from './member.service' +import MemberService, { + applyNewestOrgPayloads, + getEarliestMemberJoinedAt, + mergeIfAllowed, + registerOrgPayloadActivities, +} from './member.service' import MemberAttributeService, { isEmptyAttributeValue } from './memberAttribute.service' import { IProcessActivityResult } from './types' @@ -1454,10 +1459,21 @@ export default class ActivityService extends LoggerBase { const preparedActivities: IActivityPrepareForUpsertResult[] = [] const memberService = new MemberService(this.pgStore, this.redisClient, this.temporal, this.log) - // Shared org promise cache: ensures findOrCreateOrganization is called at most once per - // unique org per batch. Concurrent member creates that reference the same org await the - // same promise instead of firing redundant DB round trips. + // One findOrCreateOrganization per distinct organization payload per batch: members that carry + // the same payload await the same promise const orgPromiseCache = new Map>() + registerOrgPayloadActivities( + orgPromiseCache, + relevantPayloads.flatMap((p) => + [p.activity.member, p.activity.objectMember] + .filter((member) => member) + .map((member) => ({ + source: p.platform, + timestamp: p.activity.timestamp, + organizations: member.organizations, + })), + ), + ) // find distinct members to create const payloadsWithoutDbMembers: IActivityProcessData[] = relevantPayloads.filter( @@ -1722,6 +1738,18 @@ export default class ActivityService extends LoggerBase { 'activityService -> processActivities -> updateMembers', ) + // Only decides which organization values are the defaults, so a failure here doesn't fail the + // activities whose members were already updated + try { + await logExecutionTimeV2( + async () => applyNewestOrgPayloads(orgPromiseCache, this.log), + this.log, + 'activityService -> processActivities -> applyNewestOrgPayloads', + ) + } catch (err) { + this.log.error(err, 'Failed to apply the newest organization payloads of the batch!') + } + for (const payload of relevantPayloads) { const memberFailure = memberUpdates.failures.member.get(payload) if (memberFailure) { diff --git a/services/apps/data_sink_worker/src/service/member.service.ts b/services/apps/data_sink_worker/src/service/member.service.ts index 2dcae3a476..c3943d4f35 100644 --- a/services/apps/data_sink_worker/src/service/member.service.ts +++ b/services/apps/data_sink_worker/src/service/member.service.ts @@ -154,14 +154,130 @@ function lockOrganizationFor(orgPromiseCache: OrgPromiseCache) { } } +const stableJson = (value: unknown): string => + JSON.stringify(value, (_key, v) => + v && typeof v === 'object' && !Array.isArray(v) + ? Object.fromEntries(Object.entries(v).sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0))) + : v, + ) + +// Domains are normalized like findOrCreateOrganization normalizes them in place, so the key is the +// same before and after the payload is applied +function orgPayloadKey(source: string, org: IOrganization): string { + const content: Record = { + ...org, + identities: (org.identities ?? []) + .map((i) => + [ + OrganizationIdentityType.PRIMARY_DOMAIN, + OrganizationIdentityType.ALTERNATIVE_DOMAIN, + ].includes(i.type) + ? { ...i, value: i.value ? normalizeHostname(i.value, false) : i.value } + : i, + ) + .filter((i) => i.value !== undefined), + } + // the source of the member's link to the organization, not organization data + delete content.source + return stableJson({ source, content }) +} + +interface IOrgPayloadActivities { + newestMs: number + count: number +} + +interface IAppliedOrgPayload { + organizationId: string + source: string + payloadKey: string + activities: IOrgPayloadActivities + apply: () => Promise +} + +const orgPayloadsOfBatch = new WeakMap< + OrgPromiseCache, + { activities: Map; applied: IAppliedOrgPayload[] } +>() + +function batchOrgPayloads(orgPromiseCache: OrgPromiseCache) { + let payloads = orgPayloadsOfBatch.get(orgPromiseCache) + if (!payloads) { + payloads = { activities: new Map(), applied: [] } + orgPayloadsOfBatch.set(orgPromiseCache, payloads) + } + return payloads +} + +export function registerOrgPayloadActivities( + orgPromiseCache: OrgPromiseCache, + activities: { source: string; timestamp: string; organizations?: IOrganization[] }[], +): void { + const payloads = batchOrgPayloads(orgPromiseCache).activities + for (const { source, timestamp, organizations } of activities) { + const ms = new Date(timestamp).getTime() + for (const org of organizations ?? []) { + const key = orgPayloadKey(source, org) + const known = payloads.get(key) + payloads.set(key, { + newestMs: Math.max(known?.newestMs ?? ms, ms), + count: (known?.count ?? 0) + 1, + }) + } + } +} + +const winsOver = (a: IAppliedOrgPayload, b: IAppliedOrgPayload): boolean => + a.activities.newestMs !== b.activities.newestMs + ? a.activities.newestMs > b.activities.newestMs + : a.activities.count !== b.activities.count + ? a.activities.count > b.activities.count + : a.payloadKey > b.payloadKey + +// The payload applied last sets the defaults of its source, and which one that is depends on timing, +// so the winner (newest activity, then most activities) is applied again +export async function applyNewestOrgPayloads( + orgPromiseCache: OrgPromiseCache, + log: Logger, +): Promise { + const byOrganizationAndSource = new Map() + for (const payload of batchOrgPayloads(orgPromiseCache).applied) { + const key = `${payload.organizationId}#${payload.source}` + const payloads = byOrganizationAndSource.get(key) ?? [] + payloads.push(payload) + byOrganizationAndSource.set(key, payloads) + } + + for (const payloads of byOrganizationAndSource.values()) { + if (new Set(payloads.map((p) => p.payloadKey)).size < 2) { + continue + } + const winner = payloads.reduce((a, b) => (winsOver(b, a) ? b : a)) + try { + await winner.apply() + } catch (err) { + log.error( + err, + { organizationId: winner.organizationId, source: winner.source }, + 'Failed to apply the newest organization payload!', + ) + } + } +} + +// Payloads that share identities wait for each other, so they all land on the same organization +// and each adds what it carries function findOrCreateOrgOnce( orgPromiseCache: OrgPromiseCache | undefined, + source: string, org: IOrganization, findOrCreate: ( lockOrganization?: (organizationId: string) => Promise<() => void>, ) => Promise, ): Promise { - const key = orgCacheKey(org) + const payloadKey = orgPayloadKey(source, org) + const identityKey = orgCacheKey(org) + const key = identityKey ? `${identityKey}#${payloadKey}` : null const cached = key ? orgPromiseCache?.get(key) : undefined if (cached) { return cached @@ -173,9 +289,22 @@ function findOrCreateOrgOnce( const locks = orgIdentityLocks.get(orgPromiseCache) ?? new Map>() orgIdentityLocks.set(orgPromiseCache, locks) const lockKeys = orgLockKeys(org) - const promise = Promise.all(lockKeys.map((k) => locks.get(k))).then(() => - findOrCreate(lockOrganizationFor(orgPromiseCache)), - ) + const apply = () => findOrCreate(lockOrganizationFor(orgPromiseCache)) + const promise = Promise.all(lockKeys.map((k) => locks.get(k))) + .then(apply) + .then((result) => { + if (result) { + const payloads = batchOrgPayloads(orgPromiseCache) + payloads.applied.push({ + organizationId: result.id, + source, + payloadKey, + activities: payloads.activities.get(payloadKey) ?? { newestMs: 0, count: 0 }, + apply, + }) + } + return result + }) const settled = promise.catch(() => undefined) for (const k of lockKeys) { locks.set(k, settled) @@ -622,12 +751,16 @@ export default class MemberService extends LoggerBase { continue } - const result = await findOrCreateOrgOnce(orgPromiseCache, org, (lockOrganization) => - logExecutionTimeV2( - () => orgService.findOrCreate(platform, integrationId, org, lockOrganization), - this.log, - 'memberService -> create -> orgService.findOrCreate', - ), + const result = await findOrCreateOrgOnce( + orgPromiseCache, + platform, + org, + (lockOrganization) => + logExecutionTimeV2( + () => orgService.findOrCreate(platform, integrationId, org, lockOrganization), + this.log, + 'memberService -> create -> orgService.findOrCreate', + ), ) if (result) { organizations.push({ @@ -879,12 +1012,16 @@ export default class MemberService extends LoggerBase { this.log.trace({ memberId: id }, 'Finding or creating organization!') - const result = await findOrCreateOrgOnce(orgPromiseCache, org, (lockOrganization) => - logExecutionTimeV2( - () => orgService.findOrCreate(platform, integrationId, org, lockOrganization), - this.log, - 'memberService -> update -> orgService.findOrCreate', - ), + const result = await findOrCreateOrgOnce( + orgPromiseCache, + platform, + org, + (lockOrganization) => + logExecutionTimeV2( + () => orgService.findOrCreate(platform, integrationId, org, lockOrganization), + this.log, + 'memberService -> update -> orgService.findOrCreate', + ), ) if (result) { organizations.push({ @@ -1010,13 +1147,17 @@ export default class MemberService extends LoggerBase { }, ], } - const result = await findOrCreateOrgOnce(orgPromiseCache, org, (lockOrganization) => - orgService.findOrCreate( - OrganizationAttributeSource.EMAIL, - integrationId, - org, - lockOrganization, - ), + const result = await findOrCreateOrgOnce( + orgPromiseCache, + OrganizationAttributeSource.EMAIL, + org, + (lockOrganization) => + orgService.findOrCreate( + OrganizationAttributeSource.EMAIL, + integrationId, + org, + lockOrganization, + ), ) if (result) { organizations.push({ diff --git a/services/libs/data-access-layer/src/organizations/base.ts b/services/libs/data-access-layer/src/organizations/base.ts index feda98cfbd..76a870cf35 100644 --- a/services/libs/data-access-layer/src/organizations/base.ts +++ b/services/libs/data-access-layer/src/organizations/base.ts @@ -254,13 +254,17 @@ export async function markOrganizationEnriched( export async function addOrgsToSegments( qe: QueryExecutor, - segmentIds: string[], - orgIds: string[], + unsortedSegmentIds: string[], + unsortedOrgIds: string[], ): Promise { - if (orgIds.length === 0 || segmentIds.length === 0) { + if (unsortedOrgIds.length === 0 || unsortedSegmentIds.length === 0) { return } + // Concurrent inserts of the same new rows in opposite order can deadlock + const segmentIds = [...unsortedSegmentIds].sort() + const orgIds = [...unsortedOrgIds].sort() + const parameters: Record = { tenantId: DEFAULT_TENANT_ID, } From 85cd9db0b18b3f6c1701b4a01bc1e168a905a010 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Mon, 2 Nov 2026 16:18:45 +0100 Subject: [PATCH 2/7] fix: count each org payload once per activity and replay every payload MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An activity whose member and object member carry the same organization payload counted twice. When an organization got several payloads from one source, only the winner was applied again, so attributes the winner lacks kept whichever payload landed last. Every distinct payload is now applied again from the lowest ranked to the winner. Refs linuxfoundation/lfx-cdp-insights-ops#972 Signed-off-by: Uroš Marolt --- .../src/service/activity.service.ts | 17 +++++------ .../src/service/member.service.ts | 29 ++++++++++--------- 2 files changed, 24 insertions(+), 22 deletions(-) diff --git a/services/apps/data_sink_worker/src/service/activity.service.ts b/services/apps/data_sink_worker/src/service/activity.service.ts index 65307eb634..d71383d891 100644 --- a/services/apps/data_sink_worker/src/service/activity.service.ts +++ b/services/apps/data_sink_worker/src/service/activity.service.ts @@ -1464,15 +1464,14 @@ export default class ActivityService extends LoggerBase { const orgPromiseCache = new Map>() registerOrgPayloadActivities( orgPromiseCache, - relevantPayloads.flatMap((p) => - [p.activity.member, p.activity.objectMember] - .filter((member) => member) - .map((member) => ({ - source: p.platform, - timestamp: p.activity.timestamp, - organizations: member.organizations, - })), - ), + relevantPayloads.map((p) => ({ + source: p.platform, + timestamp: p.activity.timestamp, + organizations: [ + ...(p.activity.member?.organizations ?? []), + ...(p.activity.objectMember?.organizations ?? []), + ], + })), ) // find distinct members to create diff --git a/services/apps/data_sink_worker/src/service/member.service.ts b/services/apps/data_sink_worker/src/service/member.service.ts index c3943d4f35..2b1ad6d328 100644 --- a/services/apps/data_sink_worker/src/service/member.service.ts +++ b/services/apps/data_sink_worker/src/service/member.service.ts @@ -216,8 +216,8 @@ export function registerOrgPayloadActivities( const payloads = batchOrgPayloads(orgPromiseCache).activities for (const { source, timestamp, organizations } of activities) { const ms = new Date(timestamp).getTime() - for (const org of organizations ?? []) { - const key = orgPayloadKey(source, org) + // an activity counts once for each payload, however many times it carries it + for (const key of new Set((organizations ?? []).map((org) => orgPayloadKey(source, org)))) { const known = payloads.get(key) payloads.set(key, { newestMs: Math.max(known?.newestMs ?? ms, ms), @@ -235,7 +235,7 @@ const winsOver = (a: IAppliedOrgPayload, b: IAppliedOrgPayload): boolean => : a.payloadKey > b.payloadKey // The payload applied last sets the defaults of its source, and which one that is depends on timing, -// so the winner (newest activity, then most activities) is applied again +// so they are applied again from the lowest ranked to the winner (newest activity, most activities) export async function applyNewestOrgPayloads( orgPromiseCache: OrgPromiseCache, log: Logger, @@ -249,18 +249,21 @@ export async function applyNewestOrgPayloads( } for (const payloads of byOrganizationAndSource.values()) { - if (new Set(payloads.map((p) => p.payloadKey)).size < 2) { + const distinct = Array.from(new Map(payloads.map((p) => [p.payloadKey, p])).values()) + if (distinct.length < 2) { continue } - const winner = payloads.reduce((a, b) => (winsOver(b, a) ? b : a)) - try { - await winner.apply() - } catch (err) { - log.error( - err, - { organizationId: winner.organizationId, source: winner.source }, - 'Failed to apply the newest organization payload!', - ) + distinct.sort((a, b) => (winsOver(a, b) ? 1 : winsOver(b, a) ? -1 : 0)) + for (const payload of distinct) { + try { + await payload.apply() + } catch (err) { + log.error( + err, + { organizationId: payload.organizationId, source: payload.source }, + 'Failed to apply an organization payload again!', + ) + } } } } From 3560fec4b171e981fcf0d6e52e383d2275d42a26 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Mon, 2 Nov 2026 17:47:24 +0100 Subject: [PATCH 3/7] fix: apply an organization payload again to the organization it went to MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A later payload of the batch can unverify the identity an earlier one was found by, so looking it up again could miss and create a second organization. findOrCreateOrganization takes an optional organization id that skips the lookup, and the replay passes the recorded one. Refs linuxfoundation/lfx-cdp-insights-ops#972 Signed-off-by: Uroš Marolt --- .../src/service/member.service.ts | 31 ++++++++++++++----- .../src/service/organization.service.ts | 11 ++++++- .../src/organizations/base.ts | 11 ++++++- 3 files changed, 43 insertions(+), 10 deletions(-) diff --git a/services/apps/data_sink_worker/src/service/member.service.ts b/services/apps/data_sink_worker/src/service/member.service.ts index 2b1ad6d328..3b6c3639e9 100644 --- a/services/apps/data_sink_worker/src/service/member.service.ts +++ b/services/apps/data_sink_worker/src/service/member.service.ts @@ -276,6 +276,7 @@ function findOrCreateOrgOnce( org: IOrganization, findOrCreate: ( lockOrganization?: (organizationId: string) => Promise<() => void>, + organizationId?: string, ) => Promise, ): Promise { const payloadKey = orgPayloadKey(source, org) @@ -292,9 +293,8 @@ function findOrCreateOrgOnce( const locks = orgIdentityLocks.get(orgPromiseCache) ?? new Map>() orgIdentityLocks.set(orgPromiseCache, locks) const lockKeys = orgLockKeys(org) - const apply = () => findOrCreate(lockOrganizationFor(orgPromiseCache)) const promise = Promise.all(lockKeys.map((k) => locks.get(k))) - .then(apply) + .then(() => findOrCreate(lockOrganizationFor(orgPromiseCache))) .then((result) => { if (result) { const payloads = batchOrgPayloads(orgPromiseCache) @@ -303,7 +303,7 @@ function findOrCreateOrgOnce( source, payloadKey, activities: payloads.activities.get(payloadKey) ?? { newestMs: 0, count: 0 }, - apply, + apply: () => findOrCreate(lockOrganizationFor(orgPromiseCache), result.id), }) } return result @@ -758,9 +758,16 @@ export default class MemberService extends LoggerBase { orgPromiseCache, platform, org, - (lockOrganization) => + (lockOrganization, organizationId) => logExecutionTimeV2( - () => orgService.findOrCreate(platform, integrationId, org, lockOrganization), + () => + orgService.findOrCreate( + platform, + integrationId, + org, + lockOrganization, + organizationId, + ), this.log, 'memberService -> create -> orgService.findOrCreate', ), @@ -1019,9 +1026,16 @@ export default class MemberService extends LoggerBase { orgPromiseCache, platform, org, - (lockOrganization) => + (lockOrganization, organizationId) => logExecutionTimeV2( - () => orgService.findOrCreate(platform, integrationId, org, lockOrganization), + () => + orgService.findOrCreate( + platform, + integrationId, + org, + lockOrganization, + organizationId, + ), this.log, 'memberService -> update -> orgService.findOrCreate', ), @@ -1154,12 +1168,13 @@ export default class MemberService extends LoggerBase { orgPromiseCache, OrganizationAttributeSource.EMAIL, org, - (lockOrganization) => + (lockOrganization, organizationId) => orgService.findOrCreate( OrganizationAttributeSource.EMAIL, integrationId, org, lockOrganization, + organizationId, ), ) if (result) { diff --git a/services/apps/data_sink_worker/src/service/organization.service.ts b/services/apps/data_sink_worker/src/service/organization.service.ts index d01060cc6f..69fda381dc 100644 --- a/services/apps/data_sink_worker/src/service/organization.service.ts +++ b/services/apps/data_sink_worker/src/service/organization.service.ts @@ -36,10 +36,19 @@ export class OrganizationService extends LoggerBase { integrationId: string, data: IOrganization, lockOrganization?: (organizationId: string) => Promise<() => void>, + organizationId?: string, ): Promise { return this.store.transactionally(async (txStore) => { const qe = dbStoreQx(txStore) - return findOrCreateOrganization(qe, source, data, integrationId, true, lockOrganization) + return findOrCreateOrganization( + qe, + source, + data, + integrationId, + true, + lockOrganization, + organizationId, + ) }) } diff --git a/services/libs/data-access-layer/src/organizations/base.ts b/services/libs/data-access-layer/src/organizations/base.ts index 76a870cf35..efe615cbd9 100644 --- a/services/libs/data-access-layer/src/organizations/base.ts +++ b/services/libs/data-access-layer/src/organizations/base.ts @@ -602,6 +602,7 @@ export async function findOrCreateOrganization( integrationId?: string, throttleUpdatedAt = false, lockOrganization?: (organizationId: string) => Promise<() => void>, + organizationId?: string, ): Promise { if (isPlaceholderEmployerName(data.displayName)) { log.debug({ displayName: data.displayName }, 'Skipping placeholder employer name.') @@ -641,8 +642,16 @@ export async function findOrCreateOrganization( } let existing + // A payload applied again goes to the organization it went to before: a later payload may have + // unverified the identity it was found by + if (organizationId) { + existing = await findOrgForUpdate(qe, organizationId) + if (!existing) { + return undefined + } + } // find existing org by sent verified identities - for (const identity of verifiedIdentities) { + for (const identity of existing ? [] : verifiedIdentities) { existing = await logExecutionTimeV2( async () => findOrgByVerifiedIdentity(qe, identity), log, From c1d99d8201548d4d45e9f1606a4135f39e4cee42 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Mon, 2 Nov 2026 18:52:25 +0100 Subject: [PATCH 4/7] fix: send later org payloads with the same identities to the same org MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A payload can unverify the domain an earlier payload of the batch was found by. A later variant with that domain and no display name then missed every lookup and created a second organization. The org id each identity key resolved to is now kept per batch, and later variants with the same identities go to it without a lookup, like the identity-keyed cache did. Refs linuxfoundation/lfx-cdp-insights-ops#972 Signed-off-by: Uroš Marolt --- .../src/service/member.service.ts | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/services/apps/data_sink_worker/src/service/member.service.ts b/services/apps/data_sink_worker/src/service/member.service.ts index 3b6c3639e9..4024baa640 100644 --- a/services/apps/data_sink_worker/src/service/member.service.ts +++ b/services/apps/data_sink_worker/src/service/member.service.ts @@ -235,7 +235,7 @@ const winsOver = (a: IAppliedOrgPayload, b: IAppliedOrgPayload): boolean => : a.payloadKey > b.payloadKey // The payload applied last sets the defaults of its source, and which one that is depends on timing, -// so they are applied again from the lowest ranked to the winner (newest activity, most activities) +// so they are applied again from lowest ranked to the winner (newest activity, most activities) export async function applyNewestOrgPayloads( orgPromiseCache: OrgPromiseCache, log: Logger, @@ -268,6 +268,10 @@ export async function applyNewestOrgPayloads( } } +// Per batch: the organization each identity key resolved to. Later payloads with the same identities +// go there without a lookup: an earlier payload may have unverified their identity +const orgIdsOfIdentityKeys = new WeakMap>() + // Payloads that share identities wait for each other, so they all land on the same organization // and each adds what it carries function findOrCreateOrgOnce( @@ -293,10 +297,20 @@ function findOrCreateOrgOnce( const locks = orgIdentityLocks.get(orgPromiseCache) ?? new Map>() orgIdentityLocks.set(orgPromiseCache, locks) const lockKeys = orgLockKeys(org) + const resolvedIds = orgIdsOfIdentityKeys.get(orgPromiseCache) ?? new Map() + orgIdsOfIdentityKeys.set(orgPromiseCache, resolvedIds) const promise = Promise.all(lockKeys.map((k) => locks.get(k))) - .then(() => findOrCreate(lockOrganizationFor(orgPromiseCache))) + .then(() => + findOrCreate( + lockOrganizationFor(orgPromiseCache), + identityKey ? resolvedIds.get(identityKey) : undefined, + ), + ) .then((result) => { if (result) { + if (identityKey && !resolvedIds.has(identityKey)) { + resolvedIds.set(identityKey, result.id) + } const payloads = batchOrgPayloads(orgPromiseCache) payloads.applied.push({ organizationId: result.id, From 1bc76143249df4964ec0e9657425c9b3fedad096 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Mon, 2 Nov 2026 20:43:08 +0100 Subject: [PATCH 5/7] fix: normalize domains in the org cache key MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The cache key lowercased identity values but kept www and other hostname differences, so www.croxel.com and croxel.com got different keys while the lookup, the identity locks and the payload key treat them as one. A later variant could then miss the organization the first one resolved to. The cache key and the lock keys now share one normalized identity key. Refs linuxfoundation/lfx-cdp-insights-ops#972 Signed-off-by: Uroš Marolt --- .../src/service/member.service.ts | 55 ++++++++++++------- 1 file changed, 34 insertions(+), 21 deletions(-) diff --git a/services/apps/data_sink_worker/src/service/member.service.ts b/services/apps/data_sink_worker/src/service/member.service.ts index 4024baa640..737e52805a 100644 --- a/services/apps/data_sink_worker/src/service/member.service.ts +++ b/services/apps/data_sink_worker/src/service/member.service.ts @@ -77,15 +77,40 @@ type EmailDomainOrganization = { created: boolean } +// Keyed the way findOrCreateOrganization looks organizations up: domains after hostname +// normalization and a primary domain on any platform +function verifiedOrgIdentityKey( + identity: NonNullable[number], +): { key: string; value: string } | undefined { + if (!identity.verified || !identity.value) { + return undefined + } + const isDomain = [ + OrganizationIdentityType.PRIMARY_DOMAIN, + OrganizationIdentityType.ALTERNATIVE_DOMAIN, + ].includes(identity.type) + const value = isDomain ? normalizeHostname(identity.value, false) : identity.value + if (!value) { + return undefined + } + return { + value, + key: + identity.type === OrganizationIdentityType.PRIMARY_DOMAIN + ? `${identity.type}:${value.toLowerCase()}` + : `${identity.platform}:${identity.type}:${value.toLowerCase()}`, + } +} + /** * Returns a stable cache key for an org based on its verified identities, falling back to * displayName. Used by the org promise cache to deduplicate `findOrCreateOrganization` calls * across members in the same batch. */ function orgCacheKey(org: IOrganization): string | null { - const verified = (org.identities ?? []) - .filter((i) => i.verified) - .map((i) => `${i.platform}:${i.type}:${i.value.toLowerCase()}`) + const verified = Array.from( + new Set((org.identities ?? []).map((i) => verifiedOrgIdentityKey(i)?.key).filter(Boolean)), + ) .sort() .join('|') if (verified) return verified @@ -98,30 +123,18 @@ function orgCacheKey(org: IOrganization): string | null { // organization they created instead of creating a duplicate when members are updated concurrently. const orgIdentityLocks = new WeakMap>>() -// Keyed the way findOrCreateOrganization looks organizations up: domains after hostname -// normalization, a primary domain on any platform, and the name as the fallback. Without a -// display name, an organization is created under its first verified identity's value. +// The verified identities, and the name as the fallback. Without a display name, an organization +// is created under its first verified identity's value. function orgLockKeys(org: IOrganization): string[] { const keys: string[] = [] let creationName: string | undefined for (const identity of org.identities ?? []) { - if (!identity.verified || !identity.value) { - continue - } - const isDomain = [ - OrganizationIdentityType.PRIMARY_DOMAIN, - OrganizationIdentityType.ALTERNATIVE_DOMAIN, - ].includes(identity.type) - const value = isDomain ? normalizeHostname(identity.value, false) : identity.value - if (!value) { + const verified = verifiedOrgIdentityKey(identity) + if (!verified) { continue } - creationName ??= value - keys.push( - identity.type === OrganizationIdentityType.PRIMARY_DOMAIN - ? `${identity.type}:${value.toLowerCase()}` - : `${identity.platform}:${identity.type}:${value.toLowerCase()}`, - ) + creationName ??= verified.value + keys.push(verified.key) } const name = (org.displayName || creationName)?.trim().toLowerCase() From 2a85ae88719860e82693d379ed713929bf8251cf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Mon, 2 Nov 2026 21:40:32 +0100 Subject: [PATCH 6/7] fix: keep the platform in the org cache key MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two organizations can own the same domain on different platforms. The lookup tries the payload's platform first and only falls back to any platform, so a platform-free cache key could send a payload to the other organization. The cache key keeps the platform; the identity locks still take a primary domain on any platform. Refs linuxfoundation/lfx-cdp-insights-ops#972 Signed-off-by: Uroš Marolt --- .../data_sink_worker/src/service/member.service.ts | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/services/apps/data_sink_worker/src/service/member.service.ts b/services/apps/data_sink_worker/src/service/member.service.ts index 737e52805a..518e50b912 100644 --- a/services/apps/data_sink_worker/src/service/member.service.ts +++ b/services/apps/data_sink_worker/src/service/member.service.ts @@ -77,11 +77,11 @@ type EmailDomainOrganization = { created: boolean } -// Keyed the way findOrCreateOrganization looks organizations up: domains after hostname -// normalization and a primary domain on any platform +// Domains after hostname normalization, like findOrCreateOrganization looks them up. The lock key +// takes a primary domain on any platform, as the lookup's fallback does function verifiedOrgIdentityKey( identity: NonNullable[number], -): { key: string; value: string } | undefined { +): { key: string; lockKey: string; value: string } | undefined { if (!identity.verified || !identity.value) { return undefined } @@ -93,12 +93,14 @@ function verifiedOrgIdentityKey( if (!value) { return undefined } + const key = `${identity.platform}:${identity.type}:${value.toLowerCase()}` return { value, - key: + key, + lockKey: identity.type === OrganizationIdentityType.PRIMARY_DOMAIN ? `${identity.type}:${value.toLowerCase()}` - : `${identity.platform}:${identity.type}:${value.toLowerCase()}`, + : key, } } @@ -134,7 +136,7 @@ function orgLockKeys(org: IOrganization): string[] { continue } creationName ??= verified.value - keys.push(verified.key) + keys.push(verified.lockKey) } const name = (org.displayName || creationName)?.trim().toLowerCase() From be35606185d32e275ffc22fe721eec6a6bc1f9fb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Tue, 3 Nov 2026 08:52:23 +0100 Subject: [PATCH 7/7] fix: keep a malformed organization from failing the whole batch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Organization payloads are registered before the per-member error handling, so an entry like organizations: [null] threw there and failed every activity of the batch. Malformed entries and invalid timestamps are now skipped in that pass and left to the member-level handling. Refs linuxfoundation/lfx-cdp-insights-ops#972 Signed-off-by: Uroš Marolt --- .../src/service/activity.service.ts | 7 +++---- .../data_sink_worker/src/service/member.service.ts | 14 +++++++++++--- 2 files changed, 14 insertions(+), 7 deletions(-) diff --git a/services/apps/data_sink_worker/src/service/activity.service.ts b/services/apps/data_sink_worker/src/service/activity.service.ts index d71383d891..e49c72e461 100644 --- a/services/apps/data_sink_worker/src/service/activity.service.ts +++ b/services/apps/data_sink_worker/src/service/activity.service.ts @@ -1467,10 +1467,9 @@ export default class ActivityService extends LoggerBase { relevantPayloads.map((p) => ({ source: p.platform, timestamp: p.activity.timestamp, - organizations: [ - ...(p.activity.member?.organizations ?? []), - ...(p.activity.objectMember?.organizations ?? []), - ], + organizations: [p.activity.member, p.activity.objectMember].flatMap((member) => + Array.isArray(member?.organizations) ? member.organizations : [], + ), })), ) diff --git a/services/apps/data_sink_worker/src/service/member.service.ts b/services/apps/data_sink_worker/src/service/member.service.ts index 518e50b912..6634bec05d 100644 --- a/services/apps/data_sink_worker/src/service/member.service.ts +++ b/services/apps/data_sink_worker/src/service/member.service.ts @@ -230,9 +230,17 @@ export function registerOrgPayloadActivities( ): void { const payloads = batchOrgPayloads(orgPromiseCache).activities for (const { source, timestamp, organizations } of activities) { - const ms = new Date(timestamp).getTime() - // an activity counts once for each payload, however many times it carries it - for (const key of new Set((organizations ?? []).map((org) => orgPayloadKey(source, org)))) { + const time = new Date(timestamp).getTime() + const ms = Number.isFinite(time) ? time : 0 + let keys: Set + try { + // an activity counts once for each payload, however many times it carries it + keys = new Set((organizations ?? []).map((org) => orgPayloadKey(source, org))) + } catch { + // runs before the per-member error handling, which reports a malformed organization + continue + } + for (const key of keys) { const known = payloads.get(key) payloads.set(key, { newestMs: Math.max(known?.newestMs ?? ms, ms),