[themarolt]

Summary

#5125 updates each member once per batch, but it applies all member updates one after another. Meetings batches on -high carry about 50 different members each, so their batches got slower after the deploy (p50 1.71 → 2.04 s, p90 2.09 → 3.31 s per batch).

This PR keeps the order inside each group of linked members and updates the groups concurrently.

Changes

• updateMembers builds the timestamp-ordered runs per group of linked members (members that share a verified identity or can merge) and runs the groups through mapWithConcurrency. Within a group nothing changes.
• Member merges and redirects stay inside a group: they only happen on a verified identity conflict (the unique index is where verified = true), and the owner of every sent verified identity is already put in the sender's group.
• Organizations can be shared between groups, and the data sink runs without transactions, so the batch orders organization work in process:
  • Find-or-create calls whose payloads share a verified identity (a primary domain on any platform, or the name the organization is created under) wait for each other, so two groups can't both create the same new organization.
  • Calls that reach the same existing organization through different identities update it one at a time: findOrCreateOrganization takes an optional lock, which the data sink keys by organization id per batch, and reads the organization again once it holds it.
  • When two groups send the same organization, the first payload to arrive is applied, as before perf: combine a member's updates within a data sink batch #5125.
• If a group fails, the others stop before their next run and the batch fails only once their in-flight updates are done, so the batch retry can't overlap them.
• The limit is MEMBER_UPDATE_CONCURRENCY, default 5, which is the worker's DB pool size (main.ts). A higher value only makes updates wait for a connection.
• mapWithConcurrency (services/libs/common) now rejects a non-integer limit. A NaN limit, for example from a mistyped env var, started no workers and skipped every item without an error. Its other callers pass integer constants.

Local end-to-end runs of this PR against main (8b8fae2), interleaved on the same machine:

• github/git: 92.5k real results fetched through Nango, on a baseline seeded read-only from prod.
• meetings: 39,659 real activities: 7 days of Snowflake attendance rows run through the real MeetingAttendanceTransformer, with their members, organizations and existing relations seeded read-only from prod, sent the way the Snowflake consumer sends them. The second pass re-sends the same activities, like the nightly re-send.

	main	this PR
github/git, s per batch	0.252 / 0.242 / 0.360	0.191 / 0.227 / 0.185
meetings, s per batch (first pass / re-send)	0.716 / 0.697	0.227–0.259 / 0.222–0.249
meetings, updateMembers s per batch	0.575 / 0.560	0.135–0.147 / 0.132–0.141

Every table has the same rows, and the Kafka activity messages (88,871 and 79,284), search-sync messages and Redis keys are the same. The remaining differences are generated ids, timestamps, stack traces and ordering, which also differ between two runs of main. One exception: in 2 of 3 meetings runs, an organization sent by two payloads with the same identities got 2 extra non-default attributes (logo, industry) from the payload that main never applies. The organization row itself is the same.

Type of change

• Bug fix
• New feature
• Refactor / cleanup
• Performance improvement
• Chore / dependency update
• Documentation

Related issue

Closes linuxfoundation/lfx-cdp-insights-ops#959

[cursor]

PR Summary

Medium Risk
Concurrent member updates plus org locking change ordering and timing around merges and org upserts; failure coordination must stay correct for safe batch retries.

Overview
Speeds up data-sink batches by processing independent linked-member groups in parallel instead of updating every member run sequentially (addressing slowdown after batch-once-per-member behavior).

Member updates (activity.service.ts): Runs are still built and ordered inside each linked group (merge/redirect semantics unchanged). Groups are executed via mapWithConcurrency with limit MEMBER_UPDATE_CONCURRENCY (default 5, aligned with the DB pool). On failure, other groups stop before their next run and the batch throws only after in-flight work finishes so retries do not overlap partial work.

Organizations (concurrent safety): Parallel groups can share orgs. findOrCreateOrgOnce adds per-batch identity ordering (wait on shared verified identity keys before create) and per-org-id serialization for updates when the same org is reached via different identities. findOrCreateOrganization accepts an optional lockOrganization, re-reads the org after acquiring the lock, and wraps the update path in release logic. Member create/update/email-domain paths route through this helper.

mapWithConcurrency: Rejects non-integer limits so a bad env var cannot silently skip all work.

^{Reviewed by Cursor Bugbot for commit bbe6ac3. Bugbot is set up for automated code reviews on this repo. Configure here.}

[Copilot]

🟡 Changes recommended

Unresolved concurrency issues can create duplicate organizations and leave writes running after batch failure, allowing retries to overlap and update or merge members from stale state.

3 open issues

• Concurrent resolutions can create duplicate organizations · New
• Failed workers can leave writes running during batch retry · New
• Concurrent groups make organization data timing-dependent · New

What changed in this PR

This PR reduces data-sink batch processing time by updating independent member groups concurrently, building on the per-member batching introduced in #5125.

Changes:

• Runs linked-member groups concurrently while preserving timestamp order within each group.
• Adds MEMBER_UPDATE_CONCURRENCY, defaulting to 5.
• Rejects non-integer concurrency limits to prevent silently skipped work.

File	Description
services/​libs/​common/​src/​concurrency.ts	Requires a positive integer concurrency limit.
services/​apps/​data_sink_worker/​src/​service/​activity.service.ts	Processes member groups with configurable concurrency while retaining sequential updates within each group.

_{Review effort: Balanced}

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit db1ef87. Configure here.}

[Copilot]

🟡 Changes recommended

Critical locking gaps allow equivalent domain variants and matching display names with different verified domains to acquire disjoint locks, risking duplicate organization creation.

2 open issues

• Normalize domains before constructing organization lock keys · New
• Lock organization names when resolving verified identities · New

_{Review effort: Balanced}

[Copilot]

🔵 Needs a closer look

Critical locking gaps allow concurrent calls to create duplicate organizations or overwrite higher-priority attributes with lower-priority values.

No open issues 🎉

2 previously missed issues

In code that hasn't changed since last review

Lock identity-derived organization names to prevent duplicate creation

services/​apps/​data_sink_worker/​src/​service/​member.service.ts:124

An email-derived organization has no displayName, but creation names it after its first normalized verified identity (organizations/base.ts:659–661). It therefore takes no name:example.com lock, so a concurrent payload named example.com with a different verified domain can create a duplicate instead of reusing the email-created row. Include the identity-derived creation name in the name lock when displayName is absent.

Serialize updates to organizations resolved through different locks

services/​apps/​data_sink_worker/​src/​service/​member.service.ts:148

Different identities and supplied names can still resolve to the same existing organization while taking disjoint locks. Both calls can read its attributes before either writes, allowing a later lower-priority source to overwrite the higher-priority default computed by the other call (organizations/base.ts:610–654). Serialize by resolved organization ID, or lock and reload the organization and its attributes before preparing the update.

2 resolved since last review

• Lock organization names when resolving verified identities
• Normalize domains before constructing organization lock keys

_{Review effort: Balanced}

[themarolt]

Re: Copilot review · findings outside review threads

• Lock identity-derived organization names to prevent duplicate creation · services/apps/data_sink_worker/src/service/member.service.ts:124
  Fixed in 3d5d013. Without a display name, the name lock uses the first normalized verified identity, which is the name findOrCreateOrganization creates the organization under.
• Serialize updates to organizations resolved through different locks · services/apps/data_sink_worker/src/service/member.service.ts:148
  Fixed in bbe6ac3. findOrCreateOrganization takes an optional lock, and the data sink passes one keyed by organization id per batch. Once the lock is held, the organization is read again before its attributes. A row lock wouldn't work here: data-sink-worker runs without transactions, so it would be released as soon as the statement ends.

[Copilot]

🔵 Needs a closer look

Human assessment is needed for concurrency risks involving mutex release before transaction commit and inconsistent organization–segment insert ordering, which can cause stale overwrites and deadlocks.

No open issues 🎉

_{Review effort: Balanced}

[themarolt]

Re: Copilot review · concerns in the summary

• Lock released before a transaction commits: data-sink-worker runs without transactions. Its stores are created with withTransactions = false, so transactionally() runs the callback directly and every write in findOrCreateOrganization is committed when its statement returns. The lock is released in a finally after all of those writes, so the next holder reads committed data. The callers that do use transactions (members enrichment worker, public API) don't pass a lock, so their behavior is unchanged.
• Organization–segment insert order: addOrgsToSegments inserts rows in the order a member's organizations arrive. Two member updates that insert the same new rows in opposite order could deadlock, and Postgres would then abort one statement and its activity would be retried. The same race exists between pods and existed before perf: combine a member's updates within a data sink batch #5125, and no deadlock appeared in local runs. Sorting the organization ids there is tracked in linuxfoundation/lfx-cdp-insights-ops#972, together with the related organization payload fix.