You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
perf: update each member once per data sink batch - #5125
data-sink-worker updates the member of every activity one at a time, so in a 50-activity batch about 63% of memberService.update calls repeat work for a member already updated in the same batch. Member updates take about 40% of processActivities time on -normal, and the repeats add to row-lock contention on busy members and bots.
This PR updates each repeated member once per batch, with the combined data of all its activities.
Changes
New pre-pass updateRepeatedMembers, timed as activityService -> processActivities -> updateRepeatedMembers, runs before the per-activity loop for every member that appears more than once in the batch (as member or object member; an activity whose member and object member are the same person counts once):
sorts the member's activities by timestamp;
splits them into runs of consecutive activities that share a segment, integration and platform;
calls memberService.update once per run, in order, so the newest activity's data is written last.
combineMemberUpdates merges a run's member data:
attributes from the oldest to the newest activity;
identities from all activities (update normalizes and dedupes them, as before);
organizations from all activities;
the earliest valid joinedAt;
reach merged.
The rule that ignores incoming attributes flagging a non-bot member as a GitHub bot is applied per activity, as mergeData does today.
The member row and identities are reloaded before a member is updated again, so every update starts from the stored state instead of the row loaded before the batch. That row used to keep only the last update's attributes and could move joinedAt later.
Merges: when an update merges the member away, its later runs update the surviving member (also reloaded first), and results of runs that ran earlier follow the redirect, through chains. memberMap is also seeded with these redirects, as a guard for payloads that resolve the same username to another member.
If a run's update fails, its activities are retried one by one in place. An activity that still fails is resolved there with handleMemberIdentityError, as the per-activity updates do: a merge redirect is followed, otherwise the error and its metadata go to the activity's result. The per-activity updates don't retry it, since they run after newer activities and would write its older data over theirs. Results are tracked per activity.
After the pre-pass, rows and identities of every updated member are reloaded for the per-activity loop.
joinedAt ignores dates within five days of the epoch (getEarliestValidDate uses the same tolerance for the stored date).
Email-domain stint dates stay per activity: update and assignOrganizationByEmailDomain take either one activity timestamp or the timestamps per email domain, and each domain's organization gets only the dates of the activities carrying an email on it. bufferMemberOrganizationActivityDates writes an organization's distinct dates in one Redis round trip and skips the write when there are none.
Members that appear once in the batch go through the per-activity path unchanged.
Notes for reviewers:
A member active in two segments or integrations in the same batch still gets one update per run, as update is per segment and integration.
Older activities in a batch now read isTeamMember/isBot defaults from the member as updated by the batch, as the next batch would.
Part of the member-update time moves from Promise.all (memberService.update) to the new updateRepeatedMembers timer. Compare the sum of both against the old metric, or memberService -> update.
Repeated members are updated one after another, as the per-activity loop did.
An activity whose update keeps failing (for example an identity conflict blocked by a no-merge pair, linuxfoundation/lfx-cdp-insights-ops#921) costs two update attempts instead of one: its run's and its own.
A batch of 50 activities has only about 18 distinct members, but every
activity updated its member again. Members with several activities in a
batch are now updated once per segment with their combined data:
attributes from oldest to newest, identities and organizations
together, the earliest joinedAt, and every activity date for the
email-domain organization dates. A combined update that fails falls back
to the per-activity updates, so error handling is unchanged.
Refs linuxfoundation/lfx-cdp-insights-ops#916
Signed-off-by: Uroš Marolt <uros@marolt.me>
Apply the bot-attribute check per activity so one flagged activity
doesn't drop the others' attributes, ignore empty, invalid and pre-1970
joinedAt values, and refresh a member's identities after a combined
update so later updates of the same member don't hit stale conflicts.
Refs linuxfoundation/lfx-cdp-insights-ops#916
Signed-off-by: Uroš Marolt <uros@marolt.me>
A combined member update buffered every activity date of the group for
every email domain organization, so a member with an old and a new work
email got both stints spanning both dates. Each domain now gets only the
dates of the activities that carry it, written in one Redis round trip.
A group for a member that an earlier group merged away now reuses the
redirect instead of updating the absorbed member again. Identity dedup is
left to update(), which already normalizes and dedupes them.
Refs linuxfoundation/lfx-cdp-insights-ops#916
Signed-off-by: Uroš Marolt <uros@marolt.me>
A combined member update recorded before a later update merged the same
member away kept the absorbed member's id, and the per-activity loop
used it without checking memberMap. Group results now follow redirects
found later in the pre-pass, and a known redirect in memberMap wins over
a group's own member id in the loop.
Refs linuxfoundation/lfx-cdp-insights-ops#916
Signed-off-by: Uroš Marolt <uros@marolt.me>
Every update of a member in a batch started from the member row loaded
before the batch, and updateMember writes the whole attributes object, so
a member updated for several segments kept only the last update's
attributes and could get a later joinedAt. The pre-pass now reloads the
member row along with its identities before updating the same member
again, and hands the reloaded rows to the per-activity updates.
Dates within five days of the epoch no longer count towards joinedAt,
matching getEarliestValidDate.
Refs linuxfoundation/lfx-cdp-insights-ops#916
Signed-off-by: Uroš Marolt <uros@marolt.me>
When a combined update merged a member into another, the target member's
own update later in the batch still started from its row and identities
from before the merge, and could overwrite what the merge stored.
Refs linuxfoundation/lfx-cdp-insights-ops#916
Signed-off-by: Uroš Marolt <uros@marolt.me>
Single-activity updates of a member ran after its combined updates, so an
older activity could overwrite a newer one's attributes. Every member that
appears more than once in a batch now goes through the pre-pass, its
groups ordered by their newest activity, and redirect chains are followed
to the end.
Refs linuxfoundation/lfx-cdp-insights-ops#916
Signed-off-by: Uroš Marolt <uros@marolt.me>
Combined groups of one member could interleave in time, so a group with
an older activity could still be written after a newer one, a group of
a member merged away earlier was skipped, and a failed group's
per-activity retry ran after newer updates.
The pre-pass now sorts each repeated member's activities by timestamp and
updates runs of consecutive activities that share a segment, integration
and platform, in order, refreshing the member between them. Runs of a
merged member update the surviving member, and a failed run is retried
per activity in place. Results are tracked per activity, so only
activities that fail on their own fall back to the per-activity loop.
Refs linuxfoundation/lfx-cdp-insights-ops#916
Signed-off-by: Uroš Marolt <uros@marolt.me>
An activity whose member update kept failing was tried again by the
per-activity updates after newer activities of the same member, and each
attempt wrote its older attributes over theirs. An activity left out of
the pre-pass could also skip its update through a merge redirect and be
reported as stored.
A failed run is now retried per activity in place, and an activity that
still fails is resolved there with handleMemberIdentityError: a redirect
is followed, otherwise the error and its metadata go to the result, and
the per-activity updates no longer touch it.
Also groups the pure helpers at module level and names the run ordering.
Refs linuxfoundation/lfx-cdp-insights-ops#916
Signed-off-by: Uroš Marolt <uros@marolt.me>
High Risk
Changes core activity ingestion member-update ordering, merging, and redirect handling; incorrect batch combining could corrupt member attributes, joinedAt, or stint dates despite extensive edge-case logic.
Overview Replaces per-activity memberService.update calls with a batch updateMembers pass that groups work by member, merges multiple activities via combineMemberUpdates (attributes, identities, orgs, earliest valid joinedAt, reach), and issues one update per timestamp-ordered run (split by segment/integration/platform). Merge redirects are tracked and member rows are refreshed before subsequent updates; combined runs fall back to per-activity updates on failure.
Also rejects activities with invalid timestamps during member prep, fixes memberAttValue when incoming attributes are plain values vs per-platform objects, and passes per-email-domain activity timestamps into org stint buffering (ActivityTimestampsByEmailDomain, multi-date Redis sAdd). Identity redirect sync always attaches survivingId on post-merge constraint errors.
Reviewed by Cursor Bugbot for commit 1978741. Bugbot is set up for automated code reviews on this repo. Configure here.
The reason will be displayed to describe this comment to others. Learn more.
🟡 Changes recommended
Unresolved merge recovery, cross-member update ordering, and attribute-handling issues can fail valid activities or produce incorrect member and activity data.
This PR reduces repeated member updates in the data-sink worker by combining activity data into chronological runs, aiming to reduce database work and row-lock contention.
Changes:
Adds a repeated-member pre-pass with combined updates, merge redirects, refreshes, and per-activity retries.
Preserves email-domain activity timestamps for organization stint tracking.
Batches distinct organization activity dates into Redis writes.
Members that merge into each other mid-batch could have an older run of
the survivor written after the newer merged data, so all repeated members'
runs now go in one timestamp order. A combined update that merged the
member and then failed records the survivor from the error before retrying
per activity. Redirected activities keep the survivor's row for the bot and
team-member flags, and a plain isBot value on GitHub counts as a bot flag,
as it does once validateAttributes normalizes it.
Refs linuxfoundation/lfx-cdp-insights-ops#916
Signed-off-by: Uroš Marolt <uros@marolt.me>
A repeated member could merge into a member that appears once in the
batch, and that member's own update, still run per activity after the
pre-pass, then wrote its older attributes and reach over the newer merged
data. Every existing member's update now goes through the pre-pass, in one
timestamp order across the batch. Members that appear once get a single
attempt, as before; the loop only applies the results, so its
per-activity updates and username redirect map are gone.
Refs linuxfoundation/lfx-cdp-insights-ops#916
Signed-off-by: Uroš Marolt <uros@marolt.me>
Combining attributes into a new object leaves primitive flags in the original activity payloads unnormalized, but memberAttValue only reads platform-keyed values. With repeated updates carrying isTeamMember: true then isTeamMember: false, the older activity's explicit true is ignored and it receives the refreshed member's false default. Preserve per-activity attribute normalization before reading flags, or make memberAttValue handle validated primitive values.
The per-activity update used to normalize the activity's own attributes in
place, turning a plain isTeamMember or isBot value into { [platform]: value }
before memberAttValue read it. The combined update validates a copy, so
memberAttValue now reads a plain value directly.
Refs linuxfoundation/lfx-cdp-insights-ops#916
Signed-off-by: Uroš Marolt <uros@marolt.me>
Primitive activity flags are ignored during attribute refresh · services/apps/data_sink_worker/src/service/activity.service.ts:228
Fixed in f7506c8: memberAttValue now reads a plain isTeamMember/isBot value on the activity as that value, the way validateAttributes stores it ({ [platform]: value }). The per-activity update used to normalize the activity's own attributes in place; the combined update validates a copy, so the activity kept the plain value.
update() removes URL values from location after the combined attributes
are merged, so a valid location from an older activity was lost when a
newer activity sent a URL for the same platform. Each activity's location
is now filtered before the merge, as its own update would have done.
Refs linuxfoundation/lfx-cdp-insights-ops#916
Signed-off-by: Uroš Marolt <uros@marolt.me>
The timestamp map includes invalid email identities that MemberService.update later drops. A run with verified @acme.com at t1 and user@acme.com at t2 therefore buffers both dates for acme.com, incorrectly extending its inferred organization stint. Apply normalizeMemberIdentities with dropInvalidEmails: true before collecting timestamps so only accepted emails contribute dates.
update() drops invalid emails before assigning organizations by email
domain, but the per-domain timestamps still counted them, so an invalid
verified email could add its activity's date to a domain that a valid
email in the same run also carries.
Refs linuxfoundation/lfx-cdp-insights-ops#916
Signed-off-by: Uroš Marolt <uros@marolt.me>
Exclude invalid emails from organization timestamp tracking · services/apps/data_sink_worker/src/service/activity.service.ts:240
Fixed in 9924553: emailDomainTimestamps now skips emails that fail isValidEmail after normalizeMemberIdentityValue, the same check normalizeMemberIdentities(…, { dropInvalidEmails: true }) applies in update(), so only emails update() keeps add dates to a domain.
An unparseable activity timestamp used to fail only that activity when
its per-activity member update threw on the joinedAt derived from it. The
combined update skips invalid dates, so the error surfaced later in
prepareForUpsert and failed the whole batch. Such activities now get an
UnrepeatableError before any member work, and a redirect recorded after
an earlier one follows the chain to the current survivor.
Refs linuxfoundation/lfx-cdp-insights-ops#916
Signed-off-by: Uroš Marolt <uros@marolt.me>
…tivity
syncIdentitiesAfterRedirect only attached survivingId to an insert
conflict after a second redirect. After the first one the member that
started the update is already absorbed, so a run that merged its member
away and then hit another conflict retried every activity on the
absorbed member and failed them all. The conflict now always carries the
survivor.
Runs are now ordered by their newest activity instead of their oldest,
so when members merge mid-batch the merged member's newest data, which
the merge carries into the survivor, is not followed by an older run of
the survivor.
Refs linuxfoundation/lfx-cdp-insights-ops#916
Signed-off-by: Uroš Marolt <uros@marolt.me>
The merge-limit error reports an absorbed member as the survivor because syncIdentitiesAfterRedirect checks the limit after a successful merge but before advancing survivingId. Recording that ID here sends retries and later runs to the wrong row, where another merge can fail. Change the merge-limit error in member.service.ts:319–324 to report survivingId: nextRedirectId, so this redirect targets the actual survivor.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
Summary
data-sink-worker updates the member of every activity one at a time, so in a 50-activity batch about 63% of
memberService.updatecalls repeat work for a member already updated in the same batch. Member updates take about 40% ofprocessActivitiestime on-normal, and the repeats add to row-lock contention on busy members and bots.This PR updates each repeated member once per batch, with the combined data of all its activities.
Changes
New pre-pass
updateRepeatedMembers, timed asactivityService -> processActivities -> updateRepeatedMembers, runs before the per-activity loop for every member that appears more than once in the batch (as member or object member; an activity whose member and object member are the same person counts once):memberService.updateonce per run, in order, so the newest activity's data is written last.combineMemberUpdatesmerges a run's member data:updatenormalizes and dedupes them, as before);joinedAt;reachmerged.The rule that ignores incoming attributes flagging a non-bot member as a GitHub bot is applied per activity, as
mergeDatadoes today.The member row and identities are reloaded before a member is updated again, so every update starts from the stored state instead of the row loaded before the batch. That row used to keep only the last update's attributes and could move
joinedAtlater.Merges: when an update merges the member away, its later runs update the surviving member (also reloaded first), and results of runs that ran earlier follow the redirect, through chains.
memberMapis also seeded with these redirects, as a guard for payloads that resolve the same username to another member.If a run's update fails, its activities are retried one by one in place. An activity that still fails is resolved there with
handleMemberIdentityError, as the per-activity updates do: a merge redirect is followed, otherwise the error and its metadata go to the activity's result. The per-activity updates don't retry it, since they run after newer activities and would write its older data over theirs. Results are tracked per activity.After the pre-pass, rows and identities of every updated member are reloaded for the per-activity loop.
joinedAtignores dates within five days of the epoch (getEarliestValidDateuses the same tolerance for the stored date).Email-domain stint dates stay per activity:
updateandassignOrganizationByEmailDomaintake either one activity timestamp or the timestamps per email domain, and each domain's organization gets only the dates of the activities carrying an email on it.bufferMemberOrganizationActivityDateswrites an organization's distinct dates in one Redis round trip and skips the write when there are none.Members that appear once in the batch go through the per-activity path unchanged.
Notes for reviewers:
updateis per segment and integration.isTeamMember/isBotdefaults from the member as updated by the batch, as the next batch would.Promise.all (memberService.update)to the newupdateRepeatedMemberstimer. Compare the sum of both against the old metric, ormemberService -> update.Type of change
Related issue
Closes linuxfoundation/lfx-cdp-insights-ops#916