From 7326e7268f1bbff99633317ae02d574dbcf5928b Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Mon, 24 Aug 2026 14:09:00 +0200 Subject: [PATCH 1/6] fix: tombston in inferMemberOrganizationStintChanges Signed-off-by: Umberto Sgueglia --- .../src/services/member-organization.ts | 10 ++++++++++ .../data-access-layer/src/members/organizations.ts | 3 ++- services/libs/types/src/organizations.ts | 1 + 3 files changed, 13 insertions(+), 1 deletion(-) diff --git a/services/libs/common_services/src/services/member-organization.ts b/services/libs/common_services/src/services/member-organization.ts index 5c2d9083f1..c5503b59ee 100644 --- a/services/libs/common_services/src/services/member-organization.ts +++ b/services/libs/common_services/src/services/member-organization.ts @@ -175,6 +175,12 @@ export function inferMemberOrganizationStintChanges( const activeRows = normalizedRows.filter((row) => !row.deletedAt) + const tombstonedOrgIds = new Set( + normalizedRows + .filter((row) => !!row.deletedAt && !!row.deletedBy) + .map((row) => row.organizationId), + ) + // Deleted dated rows suppress recreation for dates the user removed const deletedRows = normalizedRows.filter( (row): row is typeof row & { dateStart: string } => !!row.deletedAt && !!row.dateStart, @@ -199,6 +205,10 @@ export function inferMemberOrganizationStintChanges( })) for (const { organizationId, date: targetDate } of sortedDates) { + if (tombstonedOrgIds.has(organizationId)) { + continue + } + if ( deletedRows.some( (row) => diff --git a/services/libs/data-access-layer/src/members/organizations.ts b/services/libs/data-access-layer/src/members/organizations.ts index e8b1f46432..de6346fe57 100644 --- a/services/libs/data-access-layer/src/members/organizations.ts +++ b/services/libs/data-access-layer/src/members/organizations.ts @@ -83,7 +83,8 @@ export async function fetchMemberOrganizationsBySource( "title", "memberId", "source", - "deletedAt" + "deletedAt", + "deletedBy" FROM "memberOrganizations" WHERE "memberId" = $(memberId) AND "source" = $(source) diff --git a/services/libs/types/src/organizations.ts b/services/libs/types/src/organizations.ts index 78e74f4ed5..abd2dd5ead 100644 --- a/services/libs/types/src/organizations.ts +++ b/services/libs/types/src/organizations.ts @@ -63,6 +63,7 @@ export interface IMemberOrganization { verified?: boolean verifiedBy?: string deletedAt?: string + deletedBy?: string displayName?: string affiliationOverride?: IMemberOrganizationAffiliationOverride } From 078b33ee06015016a5d0cf48ae845518a04b102d Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Tue, 25 Aug 2026 10:42:43 +0200 Subject: [PATCH 2/6] fix: skipping foreing key violation Signed-off-by: Umberto Sgueglia --- .../src/jobs/inferMemberOrganizationStintChanges.job.ts | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts b/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts index 369fc4d6bb..63c3bb34a4 100644 --- a/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts +++ b/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts @@ -104,6 +104,15 @@ const job: IJobDefinition = { processed++ } catch (err) { + if ((err as { code?: string })?.code === '23503') { + ctx.log.warn( + { memberId, err }, + 'Stint change referenced a missing organization, purging from queue.', + ) + await purgeMember(redis, memberId) + continue + } + ctx.log.error(err, { memberId }, 'Failed to process member stint inference.') throw err } From a03d035a1ff343ec98b125e541bc9271998b429e Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Tue, 25 Aug 2026 11:00:02 +0200 Subject: [PATCH 3/6] fix: skipping foreing key violation Signed-off-by: Umberto Sgueglia --- ...inferMemberOrganizationStintChanges.job.ts | 62 ++++++++++++++++--- .../src/services/member-organization.ts | 4 +- 2 files changed, 56 insertions(+), 10 deletions(-) diff --git a/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts b/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts index 63c3bb34a4..8a529312ba 100644 --- a/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts +++ b/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts @@ -15,14 +15,21 @@ import { fetchManyOrganizationAffiliationPolicies, fetchMemberOrganizationsBySource, findMemberById, + findOrgsByIds, updateMemberOrganization, } from '@crowd/data-access-layer' import { WRITE_DB_CONFIG, getDbConnection } from '@crowd/data-access-layer/src/database' import { deleteMemberSegmentAffiliations } from '@crowd/data-access-layer/src/member_segment_affiliations' import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor' +import { Logger } from '@crowd/logging' import { REDIS_CONFIG, RedisCache, RedisClient, getRedisClient } from '@crowd/redis' import { TEMPORAL_CONFIG, getTemporalClient } from '@crowd/temporal' -import { MemberOrgDate, MemberOrgStintChange, OrganizationSource } from '@crowd/types' +import { + IMemberOrganization, + MemberOrgDate, + MemberOrgStintChange, + OrganizationSource, +} from '@crowd/types' import { IJobDefinition } from '../types' @@ -78,7 +85,15 @@ const job: IJobDefinition = { { withDeleted: true }, ) - const changes = inferMemberOrganizationStintChanges(memberId, existingOrgs, orgDates) + const validOrgDates = await dropStaleOrganizationDates( + qx, + existingOrgs, + orgDates, + memberId, + ctx.log, + ) + + const changes = inferMemberOrganizationStintChanges(memberId, existingOrgs, validOrgDates) if (changes.length > 0) { ctx.log.debug({ memberId, changes }, 'Stint changes identified.') @@ -104,12 +119,9 @@ const job: IJobDefinition = { processed++ } catch (err) { + // Rare race past dropStaleOrganizationDates(): leave the entry queued, next tick self-heals. if ((err as { code?: string })?.code === '23503') { - ctx.log.warn( - { memberId, err }, - 'Stint change referenced a missing organization, purging from queue.', - ) - await purgeMember(redis, memberId) + ctx.log.warn(err, { memberId }, 'Stint change referenced missing organization.') continue } @@ -138,6 +150,42 @@ function parseSetMembers(members: string[]): MemberOrgDate[] { return results } +/** + * Drops queued dates for organizations that no longer exist (deleted/merged after being + * queued), so a single stale reference can't FK-violate and roll back the whole member's + * transaction, taking other, still-valid queued dates down with it. + */ +async function dropStaleOrganizationDates( + qx: QueryExecutor, + existingOrgs: IMemberOrganization[], + orgDates: MemberOrgDate[], + memberId: string, + log: Logger, +): Promise { + const knownOrgIds = new Set(existingOrgs.map((o) => o.organizationId)) + const orgIdsToVerify = [...new Set(orgDates.map((d) => d.organizationId))].filter( + (id) => !knownOrgIds.has(id), + ) + + if (orgIdsToVerify.length === 0) { + return orgDates + } + + const existingOrgIds = new Set((await findOrgsByIds(qx, orgIdsToVerify)).map((o) => o.id)) + const staleOrgIds = orgIdsToVerify.filter((id) => !existingOrgIds.has(id)) + + if (staleOrgIds.length === 0) { + return orgDates + } + + log.warn( + { memberId, staleOrgIds }, + 'Dropping queued stint dates for organizations that no longer exist.', + ) + + return orgDates.filter((d) => !staleOrgIds.includes(d.organizationId)) +} + /** * Purges a member from the queue and their associated Redis entries. */ diff --git a/services/libs/common_services/src/services/member-organization.ts b/services/libs/common_services/src/services/member-organization.ts index c5503b59ee..7e53b4302b 100644 --- a/services/libs/common_services/src/services/member-organization.ts +++ b/services/libs/common_services/src/services/member-organization.ts @@ -176,9 +176,7 @@ export function inferMemberOrganizationStintChanges( const activeRows = normalizedRows.filter((row) => !row.deletedAt) const tombstonedOrgIds = new Set( - normalizedRows - .filter((row) => !!row.deletedAt && !!row.deletedBy) - .map((row) => row.organizationId), + normalizedRows.filter((row) => row.deletedAt && row.deletedBy).map((row) => row.organizationId), ) // Deleted dated rows suppress recreation for dates the user removed From 2752fe15590483967320eca78f361beded9d5ae2 Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Tue, 25 Aug 2026 11:00:16 +0200 Subject: [PATCH 4/6] fix: skipping foreing key violation Signed-off-by: Umberto Sgueglia --- .../src/jobs/inferMemberOrganizationStintChanges.job.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts b/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts index 8a529312ba..9cc930b7ff 100644 --- a/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts +++ b/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts @@ -119,7 +119,6 @@ const job: IJobDefinition = { processed++ } catch (err) { - // Rare race past dropStaleOrganizationDates(): leave the entry queued, next tick self-heals. if ((err as { code?: string })?.code === '23503') { ctx.log.warn(err, { memberId }, 'Stint change referenced missing organization.') continue From 6fd8bac950d682c042f8cfe11f8eda23cc8a33e4 Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Wed, 26 Aug 2026 09:53:10 +0200 Subject: [PATCH 5/6] fix: skipping foreing key violation Signed-off-by: Umberto Sgueglia --- ...inferMemberOrganizationStintChanges.job.ts | 62 +++++++++++++------ .../src/mergeActions/repo.ts | 23 +++++++ 2 files changed, 65 insertions(+), 20 deletions(-) diff --git a/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts b/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts index 9cc930b7ff..27cf011d21 100644 --- a/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts +++ b/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts @@ -20,6 +20,7 @@ import { } from '@crowd/data-access-layer' import { WRITE_DB_CONFIG, getDbConnection } from '@crowd/data-access-layer/src/database' import { deleteMemberSegmentAffiliations } from '@crowd/data-access-layer/src/member_segment_affiliations' +import { findMergedPrimaryIds } from '@crowd/data-access-layer/src/mergeActions/repo' import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor' import { Logger } from '@crowd/logging' import { REDIS_CONFIG, RedisCache, RedisClient, getRedisClient } from '@crowd/redis' @@ -28,6 +29,7 @@ import { IMemberOrganization, MemberOrgDate, MemberOrgStintChange, + MergeActionType, OrganizationSource, } from '@crowd/types' @@ -85,7 +87,7 @@ const job: IJobDefinition = { { withDeleted: true }, ) - const validOrgDates = await dropStaleOrganizationDates( + const reconciledOrgDates = await reconcileOrganizationDates( qx, existingOrgs, orgDates, @@ -93,7 +95,11 @@ const job: IJobDefinition = { ctx.log, ) - const changes = inferMemberOrganizationStintChanges(memberId, existingOrgs, validOrgDates) + const changes = inferMemberOrganizationStintChanges( + memberId, + existingOrgs, + reconciledOrgDates, + ) if (changes.length > 0) { ctx.log.debug({ memberId, changes }, 'Stint changes identified.') @@ -120,7 +126,12 @@ const job: IJobDefinition = { processed++ } catch (err) { if ((err as { code?: string })?.code === '23503') { - ctx.log.warn(err, { memberId }, 'Stint change referenced missing organization.') + const constraint = (err as { constraint?: string })?.constraint + ctx.log.warn( + err, + { memberId, constraint }, + 'Stint change referenced a missing related record.', + ) continue } @@ -149,12 +160,9 @@ function parseSetMembers(members: string[]): MemberOrgDate[] { return results } -/** - * Drops queued dates for organizations that no longer exist (deleted/merged after being - * queued), so a single stale reference can't FK-violate and roll back the whole member's - * transaction, taking other, still-valid queued dates down with it. - */ -async function dropStaleOrganizationDates( +// Merged orgs are rewritten to their primary id instead of dropped; only genuinely +// deleted orgs are dropped, since those can never resolve to a valid target. +async function reconcileOrganizationDates( qx: QueryExecutor, existingOrgs: IMemberOrganization[], orgDates: MemberOrgDate[], @@ -171,18 +179,29 @@ async function dropStaleOrganizationDates( } const existingOrgIds = new Set((await findOrgsByIds(qx, orgIdsToVerify)).map((o) => o.id)) - const staleOrgIds = orgIdsToVerify.filter((id) => !existingOrgIds.has(id)) + const missingOrgIds = orgIdsToVerify.filter((id) => !existingOrgIds.has(id)) - if (staleOrgIds.length === 0) { + if (missingOrgIds.length === 0) { return orgDates } - log.warn( - { memberId, staleOrgIds }, - 'Dropping queued stint dates for organizations that no longer exist.', - ) + const mergedPrimaryIds = await findMergedPrimaryIds(qx, MergeActionType.ORG, missingOrgIds) + const deletedOrgIds = new Set(missingOrgIds.filter((id) => !mergedPrimaryIds.has(id))) + + if (deletedOrgIds.size > 0) { + log.warn( + { memberId, deletedOrgIds: [...deletedOrgIds] }, + 'Dropping queued stint dates for organizations that no longer exist.', + ) + } - return orgDates.filter((d) => !staleOrgIds.includes(d.organizationId)) + return orgDates + .filter((d) => !deletedOrgIds.has(d.organizationId)) + .map((d) => + mergedPrimaryIds.has(d.organizationId) + ? { ...d, organizationId: mergedPrimaryIds.get(d.organizationId) } + : d, + ) } /** @@ -198,6 +217,13 @@ async function purgeMember(redis: RedisClient, memberId: string): Promise * Applies the stint changes to the database. */ async function applyStintChanges(qx: QueryExecutor, changes: MemberOrgStintChange[]) { + const insertOrgIds = [ + ...new Set( + changes.filter((c) => c.type === 'insert').map((c) => c.organizationId), + ), + ] + const orgAffiliationPolicies = await fetchManyOrganizationAffiliationPolicies(qx, insertOrgIds) + for (const change of changes) { if (change.type === 'insert') { const memberOrganizationId = await createMemberOrganization(qx, change.memberId, { @@ -207,10 +233,6 @@ async function applyStintChanges(qx: QueryExecutor, changes: MemberOrgStintChang source: OrganizationSource.EMAIL_DOMAIN, }) - const orgAffiliationPolicies = await fetchManyOrganizationAffiliationPolicies(qx, [ - change.organizationId, - ]) - if (memberOrganizationId && orgAffiliationPolicies.get(change.organizationId)) { await changeMemberOrganizationAffiliationOverrides(qx, [ { diff --git a/services/libs/data-access-layer/src/mergeActions/repo.ts b/services/libs/data-access-layer/src/mergeActions/repo.ts index 629ca69c26..bfc0324b60 100644 --- a/services/libs/data-access-layer/src/mergeActions/repo.ts +++ b/services/libs/data-access-layer/src/mergeActions/repo.ts @@ -80,6 +80,29 @@ export async function findEntityMergeActions( return result } +export async function findMergedPrimaryIds( + qx: QueryExecutor, + type: MergeActionType, + secondaryIds: string[], +): Promise> { + if (!secondaryIds.length) { + return new Map() + } + + const rows = await qx.select( + ` + SELECT ma."secondaryId", ma."primaryId" + FROM "mergeActions" ma + WHERE ma.type = $(type) + AND ma.state = $(state) + AND ma."secondaryId" = ANY($(secondaryIds)::uuid[]) + `, + { type, state: MergeActionState.MERGED, secondaryIds }, + ) + + return new Map(rows.map((r) => [r.secondaryId, r.primaryId])) +} + export async function setMergeAction( qx: QueryExecutor, type: MergeActionType, From 4b2128f45111e1ff700ec76560d3444448bd9816 Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Wed, 26 Aug 2026 10:31:32 +0200 Subject: [PATCH 6/6] fix: simplify retry Signed-off-by: Umberto Sgueglia --- ...inferMemberOrganizationStintChanges.job.ts | 41 +++++++++++++++---- 1 file changed, 32 insertions(+), 9 deletions(-) diff --git a/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts b/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts index 27cf011d21..0ebd5ef21e 100644 --- a/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts +++ b/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts @@ -35,6 +35,8 @@ import { import { IJobDefinition } from '../types' +const MAX_FK_VIOLATION_RETRIES = 3 + const job: IJobDefinition = { name: 'infer-member-organization-stint-changes', cronTime: CronTime.every(5).minutes(), @@ -122,16 +124,29 @@ const job: IJobDefinition = { memberId, rawMembers, ) + await redis.del(fkViolationRetryKey(memberId)) processed++ } catch (err) { if ((err as { code?: string })?.code === '23503') { const constraint = (err as { constraint?: string })?.constraint - ctx.log.warn( - err, - { memberId, constraint }, - 'Stint change referenced a missing related record.', - ) + const retryKey = fkViolationRetryKey(memberId) + const retries = await redis.incr(retryKey) + + if (retries >= MAX_FK_VIOLATION_RETRIES) { + ctx.log.error( + err, + { memberId, constraint, retries }, + 'Stint change repeatedly referenced a missing related record; purging poisoned queue entry.', + ) + await purgeMember(redis, memberId) + } else { + ctx.log.warn( + err, + { memberId, constraint, retries }, + 'Stint change referenced a missing related record, will retry.', + ) + } continue } @@ -209,8 +224,18 @@ async function reconcileOrganizationDates( */ async function purgeMember(redis: RedisClient, memberId: string): Promise { const datesKey = `${MEMBER_ORG_STINT_CHANGES_DATES_PREFIX}:${memberId}` + const retryKey = fkViolationRetryKey(memberId) + + await redis + .multi() + .del(datesKey) + .del(retryKey) + .sRem(MEMBER_ORG_STINT_CHANGES_QUEUE, memberId) + .exec() +} - await redis.multi().del(datesKey).sRem(MEMBER_ORG_STINT_CHANGES_QUEUE, memberId).exec() +function fkViolationRetryKey(memberId: string): string { + return `${MEMBER_ORG_STINT_CHANGES_DATES_PREFIX}:fk-violation-retries:${memberId}` } /** @@ -218,9 +243,7 @@ async function purgeMember(redis: RedisClient, memberId: string): Promise */ async function applyStintChanges(qx: QueryExecutor, changes: MemberOrgStintChange[]) { const insertOrgIds = [ - ...new Set( - changes.filter((c) => c.type === 'insert').map((c) => c.organizationId), - ), + ...new Set(changes.filter((c) => c.type === 'insert').map((c) => c.organizationId)), ] const orgAffiliationPolicies = await fetchManyOrganizationAffiliationPolicies(qx, insertOrgIds)