perf(team): skip redundant task interval resumes for stable alive members
readPersistedStatuses каждый раз делал полный sync-scan всех task JSON под file lock и звал resumeActiveIntervalsForMember для каждого member с runtimeAlive=true — на больших командах блокировал main до 8s. Теперь маркируем member как 'resume applied' пока он остаётся alive, сбрасываем маркер при переходе в not-alive (через syncMemberTaskActivityForRuntimeTransition и в readPersistedStatuses loop). Resume остаётся идемпотентным и материализует интервалы из истории один раз за цикл alive.
This commit is contained in:
parent
068399543e
commit
d9479b5c61
2 changed files with 141 additions and 6 deletions
|
|
@ -3549,6 +3549,7 @@ export class TeamProvisioningService {
|
||||||
private readonly transcriptProjectResolver: TeamTranscriptProjectResolver;
|
private readonly transcriptProjectResolver: TeamTranscriptProjectResolver;
|
||||||
private readonly taskActivityIntervalService = new TeamTaskActivityIntervalService();
|
private readonly taskActivityIntervalService = new TeamTaskActivityIntervalService();
|
||||||
private readonly leadTaskActivitySyncedRunKeys = new Set<string>();
|
private readonly leadTaskActivitySyncedRunKeys = new Set<string>();
|
||||||
|
private readonly memberTaskActivityResumeAppliedByTeam = new Map<string, Set<string>>();
|
||||||
private readonly crashRepairedActivityIntervalsByTeam = new Set<string>();
|
private readonly crashRepairedActivityIntervalsByTeam = new Set<string>();
|
||||||
private readonly pendingCrashRepairSnapshotByTeam = new Map<
|
private readonly pendingCrashRepairSnapshotByTeam = new Map<
|
||||||
string,
|
string,
|
||||||
|
|
@ -3782,6 +3783,49 @@ export class TeamProvisioningService {
|
||||||
this.runtimeProcessRowsForUsageSnapshotByTeam.delete(teamName);
|
this.runtimeProcessRowsForUsageSnapshotByTeam.delete(teamName);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private normalizeMemberKeyForTaskActivity(memberName: string): string {
|
||||||
|
return memberName.trim().toLowerCase();
|
||||||
|
}
|
||||||
|
|
||||||
|
private getMemberTaskActivityResumeAppliedSet(teamName: string): Set<string> {
|
||||||
|
let applied = this.memberTaskActivityResumeAppliedByTeam.get(teamName);
|
||||||
|
if (!applied) {
|
||||||
|
applied = new Set();
|
||||||
|
this.memberTaskActivityResumeAppliedByTeam.set(teamName, applied);
|
||||||
|
}
|
||||||
|
return applied;
|
||||||
|
}
|
||||||
|
|
||||||
|
private resumeTaskActivityIntervalsForAliveMember(
|
||||||
|
teamName: string,
|
||||||
|
memberName: string,
|
||||||
|
at: string
|
||||||
|
): void {
|
||||||
|
const memberKey = this.normalizeMemberKeyForTaskActivity(memberName);
|
||||||
|
if (!memberKey) return;
|
||||||
|
const applied = this.getMemberTaskActivityResumeAppliedSet(teamName);
|
||||||
|
if (applied.has(memberKey)) return;
|
||||||
|
const result = this.taskActivityIntervalService.resumeActiveIntervalsForMember(
|
||||||
|
teamName,
|
||||||
|
memberName,
|
||||||
|
at
|
||||||
|
);
|
||||||
|
if (!result.failed) {
|
||||||
|
applied.add(memberKey);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private dropTaskActivityResumeMarkerForMember(teamName: string, memberName: string): void {
|
||||||
|
const memberKey = this.normalizeMemberKeyForTaskActivity(memberName);
|
||||||
|
if (!memberKey) return;
|
||||||
|
const applied = this.memberTaskActivityResumeAppliedByTeam.get(teamName);
|
||||||
|
if (!applied) return;
|
||||||
|
applied.delete(memberKey);
|
||||||
|
if (applied.size === 0) {
|
||||||
|
this.memberTaskActivityResumeAppliedByTeam.delete(teamName);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private cloneMemberSpawnStatusesSnapshot(
|
private cloneMemberSpawnStatusesSnapshot(
|
||||||
snapshot: MemberSpawnStatusesSnapshot
|
snapshot: MemberSpawnStatusesSnapshot
|
||||||
): MemberSpawnStatusesSnapshot {
|
): MemberSpawnStatusesSnapshot {
|
||||||
|
|
@ -13587,6 +13631,7 @@ export class TeamProvisioningService {
|
||||||
): void {
|
): void {
|
||||||
if (previous.runtimeAlive === true && next.runtimeAlive !== true) {
|
if (previous.runtimeAlive === true && next.runtimeAlive !== true) {
|
||||||
this.pauseMemberTaskActivityForRuntimeLoss(run, memberName, previous, observedAt);
|
this.pauseMemberTaskActivityForRuntimeLoss(run, memberName, previous, observedAt);
|
||||||
|
this.dropTaskActivityResumeMarkerForMember(run.teamName, memberName);
|
||||||
} else if (previous.runtimeAlive !== true && next.runtimeAlive === true) {
|
} else if (previous.runtimeAlive !== true && next.runtimeAlive === true) {
|
||||||
const nextUpdatedMs = parseOptionalIsoMs(next.updatedAt);
|
const nextUpdatedMs = parseOptionalIsoMs(next.updatedAt);
|
||||||
const previousUpdatedMs = parseOptionalIsoMs(previous.updatedAt);
|
const previousUpdatedMs = parseOptionalIsoMs(previous.updatedAt);
|
||||||
|
|
@ -13594,7 +13639,7 @@ export class TeamProvisioningService {
|
||||||
nextUpdatedMs > 0 && (previousUpdatedMs <= 0 || nextUpdatedMs > previousUpdatedMs)
|
nextUpdatedMs > 0 && (previousUpdatedMs <= 0 || nextUpdatedMs > previousUpdatedMs)
|
||||||
? next.updatedAt
|
? next.updatedAt
|
||||||
: nowIso();
|
: nowIso();
|
||||||
this.taskActivityIntervalService.resumeActiveIntervalsForMember(
|
this.resumeTaskActivityIntervalsForAliveMember(
|
||||||
run.teamName,
|
run.teamName,
|
||||||
memberName,
|
memberName,
|
||||||
deriveTaskActivityResumeAt(previous, observedAt, resumeFallbackAt)
|
deriveTaskActivityResumeAt(previous, observedAt, resumeFallbackAt)
|
||||||
|
|
@ -13940,11 +13985,9 @@ export class TeamProvisioningService {
|
||||||
const runtimeObservedAt = nowIso();
|
const runtimeObservedAt = nowIso();
|
||||||
for (const [memberName, entry] of Object.entries(nextStatuses)) {
|
for (const [memberName, entry] of Object.entries(nextStatuses)) {
|
||||||
if (entry.runtimeAlive === true) {
|
if (entry.runtimeAlive === true) {
|
||||||
this.taskActivityIntervalService.resumeActiveIntervalsForMember(
|
this.resumeTaskActivityIntervalsForAliveMember(teamName, memberName, runtimeObservedAt);
|
||||||
teamName,
|
} else {
|
||||||
memberName,
|
this.dropTaskActivityResumeMarkerForMember(teamName, memberName);
|
||||||
runtimeObservedAt
|
|
||||||
);
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
const expectedMembers = snapshot ? this.getPersistedLaunchMemberNames(snapshot) : undefined;
|
const expectedMembers = snapshot ? this.getPersistedLaunchMemberNames(snapshot) : undefined;
|
||||||
|
|
|
||||||
|
|
@ -2663,6 +2663,98 @@ describe('TeamProvisioningService', () => {
|
||||||
pauseSpy.mockRestore();
|
pauseSpy.mockRestore();
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it('skips redundant resumeActiveIntervalsForMember calls while a member stays alive', () => {
|
||||||
|
const resumeSpy = vi
|
||||||
|
.spyOn(TeamTaskActivityIntervalService.prototype, 'resumeActiveIntervalsForMember')
|
||||||
|
.mockReturnValue({ changedTasks: 0 });
|
||||||
|
try {
|
||||||
|
const svc = new TeamProvisioningService();
|
||||||
|
const internals = svc as unknown as {
|
||||||
|
resumeTaskActivityIntervalsForAliveMember: (
|
||||||
|
teamName: string,
|
||||||
|
memberName: string,
|
||||||
|
at: string
|
||||||
|
) => void;
|
||||||
|
dropTaskActivityResumeMarkerForMember: (teamName: string, memberName: string) => void;
|
||||||
|
};
|
||||||
|
const teamName = 'member-task-activity-resume-dedup-team';
|
||||||
|
|
||||||
|
internals.resumeTaskActivityIntervalsForAliveMember(
|
||||||
|
teamName,
|
||||||
|
'alice',
|
||||||
|
'2026-05-29T00:00:00.000Z'
|
||||||
|
);
|
||||||
|
internals.resumeTaskActivityIntervalsForAliveMember(
|
||||||
|
teamName,
|
||||||
|
'alice',
|
||||||
|
'2026-05-29T00:00:05.000Z'
|
||||||
|
);
|
||||||
|
internals.resumeTaskActivityIntervalsForAliveMember(
|
||||||
|
teamName,
|
||||||
|
'ALICE',
|
||||||
|
'2026-05-29T00:00:10.000Z'
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(resumeSpy).toHaveBeenCalledTimes(1);
|
||||||
|
expect(resumeSpy).toHaveBeenNthCalledWith(
|
||||||
|
1,
|
||||||
|
teamName,
|
||||||
|
'alice',
|
||||||
|
'2026-05-29T00:00:00.000Z'
|
||||||
|
);
|
||||||
|
|
||||||
|
internals.dropTaskActivityResumeMarkerForMember(teamName, 'alice');
|
||||||
|
internals.resumeTaskActivityIntervalsForAliveMember(
|
||||||
|
teamName,
|
||||||
|
'alice',
|
||||||
|
'2026-05-29T00:01:00.000Z'
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(resumeSpy).toHaveBeenCalledTimes(2);
|
||||||
|
expect(resumeSpy).toHaveBeenNthCalledWith(
|
||||||
|
2,
|
||||||
|
teamName,
|
||||||
|
'alice',
|
||||||
|
'2026-05-29T00:01:00.000Z'
|
||||||
|
);
|
||||||
|
} finally {
|
||||||
|
resumeSpy.mockRestore();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
it('retries resumeActiveIntervalsForMember when the previous call reported failure', () => {
|
||||||
|
const resumeSpy = vi
|
||||||
|
.spyOn(TeamTaskActivityIntervalService.prototype, 'resumeActiveIntervalsForMember')
|
||||||
|
.mockReturnValueOnce({ changedTasks: 0, failed: true })
|
||||||
|
.mockReturnValueOnce({ changedTasks: 1 });
|
||||||
|
try {
|
||||||
|
const svc = new TeamProvisioningService();
|
||||||
|
const internals = svc as unknown as {
|
||||||
|
resumeTaskActivityIntervalsForAliveMember: (
|
||||||
|
teamName: string,
|
||||||
|
memberName: string,
|
||||||
|
at: string
|
||||||
|
) => void;
|
||||||
|
};
|
||||||
|
const teamName = 'member-task-activity-resume-retry-team';
|
||||||
|
|
||||||
|
internals.resumeTaskActivityIntervalsForAliveMember(
|
||||||
|
teamName,
|
||||||
|
'bob',
|
||||||
|
'2026-05-29T00:00:00.000Z'
|
||||||
|
);
|
||||||
|
internals.resumeTaskActivityIntervalsForAliveMember(
|
||||||
|
teamName,
|
||||||
|
'bob',
|
||||||
|
'2026-05-29T00:00:05.000Z'
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(resumeSpy).toHaveBeenCalledTimes(2);
|
||||||
|
} finally {
|
||||||
|
resumeSpy.mockRestore();
|
||||||
|
}
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
describe('member spawn status launch reads', () => {
|
describe('member spawn status launch reads', () => {
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue