fix(cli): concurrent deploy telemetry events share an event id

packages/aws-cdk/lib/cli/telemetry/session.ts

@@ -64,6 +64,7 @@ export class TelemetrySession {
   private sigintListener?: () => void;
   private _nextEventCounters?: Record<string, number>;
   private count = 0;
+  private lastStateUpdate: Promise<void> = Promise.resolve();
   private loadTime?: number;
 
   constructor(private readonly props: TelemetrySessionProps) {
@@ -241,6 +242,12 @@ export class TelemetrySession {
 
   public async emit(event: TelemetryEvent): Promise<void> {
     this.count += 1;
+    const { event: { command }, environment, project } = this.sessionInfo;
+    const identifiers = {
+      ...this.sessionInfo.identifiers,
+      eventId: `${this.sessionInfo.identifiers.sessionId}:${this.count}`,
+      timestamp: new Date().toISOString(),
+    };
 
     const counters = {
       ...this._nextEventCounters,
@@ -254,17 +261,13 @@ export class TelemetrySession {
 
     return this.client.emit({
       event: {
-        command: this.sessionInfo.event.command,
+        command,
         state: getState(event.error),
         eventType: event.eventType,
       },
-      identifiers: {
-        ...this.sessionInfo.identifiers,
-        eventId: `${this.sessionInfo.identifiers.sessionId}:${this.count}`,
-        timestamp: new Date().toISOString(),
-      },
-      environment: this.sessionInfo.environment,
-      project: this.sessionInfo.project,
+      identifiers,
+      environment,
+      project,
       duration: {
         total: event.duration,
       },
@@ -284,7 +287,8 @@ export class TelemetrySession {
    * failed DEPLOY events in a sequence.
    */
   private async trackDeployStatistics(isSuccessful: boolean, counters: Record<string, number>) {
-    await withTelemetryState((state) => {
+    // Concurrent deployments read and write the same state file, so update it one at a time.
+    const update = this.lastStateUpdate.then(() => withTelemetryState((state) => {
       const recentFailures = state.sequentialDeploymentFailures ?? 0;
 
       if (isSuccessful) {
@@ -294,7 +298,14 @@ export class TelemetrySession {
       }
 
       counters.sequentialDeploymentFailures = state.sequentialDeploymentFailures;
-    });
+    }));
+    this.lastStateUpdate = update.catch(() => undefined);
+
+    try {
+      await update;
+    } catch (e: any) {
+      await this.ioHost.defaults.trace(`Failed to update telemetry state: ${e.message}`);
+    }
   }
 
   private get sessionInfo(): SessionSchema {

packages/aws-cdk/test/cli/telemetry/session.test.ts

@@ -7,6 +7,7 @@ import { CliIoHost } from '../../../lib/cli/io-host';
 import { type TelemetrySchema } from '../../../lib/cli/telemetry/schema';
 import { TelemetrySession, isValidWrapperUserAgent } from '../../../lib/cli/telemetry/session';
 import { IoHostTelemetrySink } from '../../../lib/cli/telemetry/sink/io-host-sink';
+import * as telemetryState from '../../../lib/cli/telemetry/telemetry-state';
 import { withEnv } from '../../_helpers/with-env';
 
 let ioHost: CliIoHost;
@@ -34,6 +35,10 @@ describe('TelemetrySession', () => {
     clientFlushSpy = jest.spyOn(client, 'flush');
   });
 
+  afterEach(() => {
+    jest.restoreAllMocks();
+  });
+
   test('can emit data to the client', async () => {
     // WHEN
     await session.emit({
@@ -114,6 +119,63 @@ describe('TelemetrySession', () => {
     }));
   });
 
+  test('concurrent DEPLOY events get distinct event ids', async () => {
+    // GIVEN
+    jest.spyOn(telemetryState, 'withTelemetryState').mockResolvedValue(undefined);
+
+    // WHEN: two stacks finish deploying at the same time, as with --concurrency
+    const first = session.emit({ eventType: 'DEPLOY', duration: 1234 });
+    const second = session.emit({ eventType: 'DEPLOY', duration: 1234 });
+    await first;
+    await second;
+
+    // THEN
+    const sequenceNumbers = clientEmitSpy.mock.calls.map(([e]) => e.identifiers.eventId.split(':').pop());
+    expect(sequenceNumbers.sort()).toEqual(['1', '2']);
+  });
+
+  test('concurrent failed DEPLOY events count every sequential failure', async () => {
+    // GIVEN: an in-memory state file that yields between read and write, so updates can overlap
+    let state: telemetryState.TelemetryState = {};
+    jest.spyOn(telemetryState, 'withTelemetryState').mockImplementation(async (block) => {
+      const copy = { ...state };
+      await Promise.resolve();
+      const ret = block(copy);
+      state = copy;
+      return ret;
+    });
+
+    // WHEN
+    const error = { name: ToolkitError.name };
+    const first = session.emit({ eventType: 'DEPLOY', duration: 1234, error });
+    const second = session.emit({ eventType: 'DEPLOY', duration: 1234, error });
+    const third = session.emit({ eventType: 'DEPLOY', duration: 1234, error });
+    await first;
+    await second;
+    await third;
+
+    // THEN
+    const failures = clientEmitSpy.mock.calls.map(([e]) => e.counters?.sequentialDeploymentFailures);
+    expect(failures).toEqual([1, 2, 3]);
+    expect(state.sequentialDeploymentFailures).toEqual(3);
+  });
+
+  test('DEPLOY event is still sent when the state file cannot be updated, and later ones are counted', async () => {
+    // GIVEN
+    jest.spyOn(telemetryState, 'withTelemetryState')
+      .mockRejectedValueOnce(new Error('EACCES'))
+      .mockImplementation(async (block) => block({}));
+
+    // WHEN
+    await session.emit({ eventType: 'DEPLOY', duration: 1234 });
+    await session.emit({ eventType: 'DEPLOY', duration: 1234, error: { name: ToolkitError.name } });
+
+    // THEN
+    expect(clientEmitSpy).toHaveBeenCalledTimes(2);
+    expect(clientEmitSpy.mock.calls[0][0].counters?.sequentialDeploymentFailures).toBeUndefined();
+    expect(clientEmitSpy.mock.calls[1][0].counters?.sequentialDeploymentFailures).toEqual(1);
+  });
+
   test('calling end more than once results in no-op', async () => {
     // GIVEN
     const spanEndSpy = jest.spyOn(session.commandSpan!, 'end');