From 8fc22920f363dbc86b8256b8548d1a01eb46514e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Beno=C3=AEt=20Zugmeyer?= Date: Tue, 20 May 2025 15:38:27 +0200 Subject: [PATCH 1/5] =?UTF-8?q?=E2=9C=A8=20[RUM-10144]=20implement=20a=20b?= =?UTF-8?q?uffered=20observable?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/core/src/index.ts | 2 +- packages/core/src/tools/observable.spec.ts | 143 ++++++++++++++++++++- packages/core/src/tools/observable.ts | 93 ++++++++++++-- 3 files changed, 224 insertions(+), 14 deletions(-) diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 5fe2ddc543..a2751ecdad 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -56,7 +56,7 @@ export { } from './domain/telemetry' export { monitored, monitor, callMonitored, setDebugMode, monitorError } from './tools/monitor' export type { Subscription } from './tools/observable' -export { Observable } from './tools/observable' +export { Observable, BufferedObservable } from './tools/observable' export type { SessionManager } from './domain/session/sessionManager' export { startSessionManager, stopSessionManager } from './domain/session/sessionManager' export { diff --git a/packages/core/src/tools/observable.spec.ts b/packages/core/src/tools/observable.spec.ts index 5805343ef9..baffdae2f0 100644 --- a/packages/core/src/tools/observable.spec.ts +++ b/packages/core/src/tools/observable.spec.ts @@ -1,4 +1,4 @@ -import { mergeObservables, Observable } from './observable' +import { BufferedObservable, mergeObservables, Observable } from './observable' describe('observable', () => { let observable: Observable @@ -119,3 +119,144 @@ describe('mergeObservables', () => { expect(subscriber).not.toHaveBeenCalled() }) }) + +describe('BufferedObservable', () => { + it('invokes the observer with buffered data', async () => { + const observable = new BufferedObservable(100) + observable.notify('first') + observable.notify('second') + + const observer = jasmine.createSpy('observer') + observable.subscribe(observer) + + await nextMicroTask() + + expect(observer).toHaveBeenCalledTimes(2) + }) + + it('invokes the observer asynchronously', async () => { + const observable = new BufferedObservable(100) + observable.notify('first') + + const observer = jasmine.createSpy('observer') + observable.subscribe(observer) + + expect(observer).not.toHaveBeenCalled() + + await nextMicroTask() + + expect(observer).toHaveBeenCalledWith('first') + }) + + it('invokes the observer when new data is notified after subscription', async () => { + const observable = new BufferedObservable(100) + + const observer = jasmine.createSpy('observer') + observable.subscribe(observer) + + observable.notify('first') + + await nextMicroTask() + + observable.notify('second') + + expect(observer).toHaveBeenCalledTimes(2) + expect(observer).toHaveBeenCalledWith('first') + expect(observer).toHaveBeenCalledWith('second') + }) + + it('drops data when the buffer is full', async () => { + const observable = new BufferedObservable(2) + observable.notify('first') // This should be dropped + observable.notify('second') + observable.notify('third') + + const observer = jasmine.createSpy('observer') + observable.subscribe(observer) + + await nextMicroTask() + + expect(observer).toHaveBeenCalledTimes(2) + expect(observer).toHaveBeenCalledWith('second') + expect(observer).toHaveBeenCalledWith('third') + }) + + it('allows to unsubscribe from the observer, the middle of buffered data', async () => { + const observable = new BufferedObservable(100) + observable.notify('first') + observable.notify('second') + + const observer = jasmine.createSpy('observer').and.callFake(() => { + subscription.unsubscribe() + }) + const subscription = observable.subscribe(observer) + + await nextMicroTask() + + expect(observer).toHaveBeenCalledTimes(1) + }) + + it('allows to unsubscribe before the buffered data', async () => { + const observable = new BufferedObservable(100) + observable.notify('first') + + const observer = jasmine.createSpy('observer') + const subscription = observable.subscribe(observer) + + subscription.unsubscribe() + + await nextMicroTask() + + expect(observer).not.toHaveBeenCalled() + }) + + it('allows to unsubscribe after the buffered data', async () => { + const observable = new BufferedObservable(100) + + const observer = jasmine.createSpy('observer') + const subscription = observable.subscribe(observer) + + await nextMicroTask() + + subscription.unsubscribe() + + observable.notify('first') + + expect(observer).not.toHaveBeenCalled() + }) + + it('calling unbuffer() removes buffered data', async () => { + const observable = new BufferedObservable(2) + observable.notify('first') + observable.notify('second') + + observable.unbuffer() + await nextMicroTask() + + const observer = jasmine.createSpy('observer') + observable.subscribe(observer) + await nextMicroTask() + + expect(observer).not.toHaveBeenCalled() + }) + + it('when calling unbuffer() right after subscription, buffered data should still be notified', async () => { + const observable = new BufferedObservable(2) + observable.notify('first') + observable.notify('second') + + const observer = jasmine.createSpy('observer') + observable.subscribe(observer) + + observable.unbuffer() + await nextMicroTask() + + expect(observer).toHaveBeenCalledTimes(2) + expect(observer).toHaveBeenCalledWith('first') + expect(observer).toHaveBeenCalledWith('second') + }) +}) + +function nextMicroTask() { + return Promise.resolve() +} diff --git a/packages/core/src/tools/observable.ts b/packages/core/src/tools/observable.ts index 60b672126b..8d8b08ea2e 100644 --- a/packages/core/src/tools/observable.ts +++ b/packages/core/src/tools/observable.ts @@ -1,32 +1,42 @@ +import { monitorError } from './monitor' + export interface Subscription { unsubscribe: () => void } +type Observer = (data: T) => void + // eslint-disable-next-line no-restricted-syntax export class Observable { - private observers: Array<(data: T) => void> = [] + protected observers: Array> = [] private onLastUnsubscribe?: () => void constructor(private onFirstSubscribe?: (observable: Observable) => (() => void) | void) {} - subscribe(f: (data: T) => void): Subscription { - this.observers.push(f) - if (this.observers.length === 1 && this.onFirstSubscribe) { - this.onLastUnsubscribe = this.onFirstSubscribe(this) || undefined - } + subscribe(observer: Observer): Subscription { + this.addObserver(observer) return { - unsubscribe: () => { - this.observers = this.observers.filter((other) => f !== other) - if (!this.observers.length && this.onLastUnsubscribe) { - this.onLastUnsubscribe() - } - }, + unsubscribe: () => this.removeObserver(observer), } } notify(data: T) { this.observers.forEach((observer) => observer(data)) } + + protected addObserver(observer: Observer) { + this.observers.push(observer) + if (this.observers.length === 1 && this.onFirstSubscribe) { + this.onLastUnsubscribe = this.onFirstSubscribe(this) || undefined + } + } + + protected removeObserver(observer: Observer) { + this.observers = this.observers.filter((other) => observer !== other) + if (!this.observers.length && this.onLastUnsubscribe) { + this.onLastUnsubscribe() + } + } } export function mergeObservables(...observables: Array>) { @@ -37,3 +47,62 @@ export function mergeObservables(...observables: Array>) { return () => subscriptions.forEach((subscription) => subscription.unsubscribe()) }) } + +// eslint-disable-next-line no-restricted-syntax +export class BufferedObservable extends Observable { + private buffer: T[] = [] + + constructor(private maxBufferSize: number) { + // no onFirstSubscribe as it makes less sense with buffered data + super() + } + + notify(data: T) { + this.buffer.push(data) + if (this.buffer.length > this.maxBufferSize) { + this.buffer.shift() + } + super.notify(data) + } + + subscribe(observer: Observer): Subscription { + let closed = false + + const subscription = { + unsubscribe: () => { + closed = true + this.removeObserver(observer) + }, + } + + enqueueMicroTask(() => { + for (const data of this.buffer) { + if (closed) { + return + } + observer(data) + } + + if (!closed) { + this.addObserver(observer) + } + }) + + return subscription + } + + /** + * Drop buffered data and don't buffer future data. This is to avoid leaking memory when it's not + * needed anymore. This is not be required in most cases, but still useful to clarify our intent + * and lowering our memory impact. + */ + unbuffer() { + enqueueMicroTask(() => { + this.maxBufferSize = this.buffer.length = 0 + }) + } +} + +function enqueueMicroTask(callback: () => void) { + Promise.resolve().then(callback).catch(monitorError) +} From 6b29b3eb15d80c5ef6cdb539c44ec84e7607f503 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Beno=C3=AEt=20Zugmeyer?= Date: Tue, 3 Jun 2025 18:41:51 +0200 Subject: [PATCH 2/5] =?UTF-8?q?=F0=9F=93=9D=20deprecate=20BoundedBuffer?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/core/src/tools/boundedBuffer.ts | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/packages/core/src/tools/boundedBuffer.ts b/packages/core/src/tools/boundedBuffer.ts index 3b48796966..976ac26434 100644 --- a/packages/core/src/tools/boundedBuffer.ts +++ b/packages/core/src/tools/boundedBuffer.ts @@ -2,12 +2,18 @@ import { removeItem } from './utils/arrayUtils' const BUFFER_LIMIT = 500 +/** + * @deprecated Use `BufferedObservable` instead. + */ export interface BoundedBuffer { add: (callback: (arg: T) => void) => void remove: (callback: (arg: T) => void) => void drain: (arg: T) => void } +/** + * @deprecated Use `BufferedObservable` instead. + */ export function createBoundedBuffer(): BoundedBuffer { const buffer: Array<(arg: T) => void> = [] From 92253a5bb6732b7b6fd4a5592d9b722f7b6c567e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Beno=C3=AEt=20Zugmeyer?= Date: Tue, 3 Jun 2025 18:26:59 +0200 Subject: [PATCH 3/5] =?UTF-8?q?=E2=9C=85=20[RUM-10144]=20prepare=20tests?= =?UTF-8?q?=20for=20async=20assembly?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This commit is purely intending to ease review. It does not introduce any functional change or new test case, it just prepares the tests for the assembly being asynchronous coming in the next commit. --- packages/rum-core/src/boot/startRum.spec.ts | 43 ++-- packages/rum-core/src/domain/assembly.spec.ts | 242 +++++++++++------- 2 files changed, 168 insertions(+), 117 deletions(-) diff --git a/packages/rum-core/src/boot/startRum.spec.ts b/packages/rum-core/src/boot/startRum.spec.ts index 6dda5d59d1..1d6a53bcef 100644 --- a/packages/rum-core/src/boot/startRum.spec.ts +++ b/packages/rum-core/src/boot/startRum.spec.ts @@ -58,7 +58,7 @@ function collectServerEvents(lifeCycle: LifeCycle) { return serverRumEvents } -function startRumStub( +async function startRumStub( lifeCycle: LifeCycle, configuration: RumConfiguration, sessionManager: RumSessionManager, @@ -96,6 +96,10 @@ function startRumStub( ) startLongAnimationFrameCollection(lifeCycle, configuration) + + // Wait for assembly to start producing events + await Promise.resolve() + return { stop: () => { viewHistory.stop() @@ -111,7 +115,7 @@ describe('rum session', () => { let lifeCycle: LifeCycle let sessionManager: RumSessionManagerMock - beforeEach(() => { + beforeEach(async () => { lifeCycle = new LifeCycle() sessionManager = createRumSessionManagerMock().setId('42') const domMutationObservable = new Observable() @@ -119,7 +123,7 @@ describe('rum session', () => { const { locationChangeObservable } = setupLocationObserver() serverRumEvents = collectServerEvents(lifeCycle) - const { stop } = startRumStub( + const { stop } = await startRumStub( lifeCycle, mockRumConfiguration(), sessionManager, @@ -160,7 +164,7 @@ describe('rum session keep alive', () => { let sessionManager: RumSessionManagerMock let serverRumEvents: RumEvent[] - beforeEach(() => { + beforeEach(async () => { lifeCycle = new LifeCycle() clock = mockClock() sessionManager = createRumSessionManagerMock().setId('1234') @@ -169,7 +173,7 @@ describe('rum session keep alive', () => { const { locationChangeObservable } = setupLocationObserver() serverRumEvents = collectServerEvents(lifeCycle) - const { stop } = startRumStub( + const { stop } = await startRumStub( lifeCycle, mockRumConfiguration(), sessionManager, @@ -227,14 +231,14 @@ describe('rum events url', () => { let serverRumEvents: RumEvent[] let stop: () => void - function setupViewUrlTest() { + async function setupViewUrlTest() { const sessionManager = createRumSessionManagerMock().setId('1234') const domMutationObservable = new Observable() const windowOpenObservable = new Observable() const locationSetupResult = setupLocationObserver('http://foo.com/') changeLocation = locationSetupResult.changeLocation - const startResult = startRumStub( + const startResult = await startRumStub( lifeCycle, mockRumConfiguration(), sessionManager, @@ -258,8 +262,8 @@ describe('rum events url', () => { }) }) - it('should keep the same URL when updating a view ended by a URL change', () => { - setupViewUrlTest() + it('should keep the same URL when updating a view ended by a URL change', async () => { + await setupViewUrlTest() serverRumEvents.length = 0 changeLocation('/bar') @@ -269,11 +273,11 @@ describe('rum events url', () => { expect(serverRumEvents[1].view.url).toEqual('http://foo.com/bar') }) - it('should attach the url corresponding to the start of the event', () => { + it('should attach the url corresponding to the start of the event', async () => { clock = mockClock() const { notifyPerformanceEntries } = mockPerformanceObserver() - setupViewUrlTest() + await setupViewUrlTest() clock.tick(10) changeLocation('http://foo.com/?bar=bar') clock.tick(10) @@ -296,10 +300,10 @@ describe('rum events url', () => { expect(longTaskEvent.view.url).toBe('http://foo.com/?bar=bar') }) - it('should keep the same URL when updating an ended view', () => { + it('should keep the same URL when updating an ended view', async () => { clock = mockClock() const { triggerOnLoad } = mockDocumentReadyState() - setupViewUrlTest() + await setupViewUrlTest() clock.tick(VIEW_DURATION) @@ -320,7 +324,7 @@ describe('view events', () => { let interceptor: ReturnType let stop: () => void - function setupViewCollectionTest() { + async function setupViewCollectionTest() { const startResult = startRum( mockRumConfiguration(), noopRecorderApi, @@ -331,6 +335,9 @@ describe('view events', () => { createCustomVitalsState() ) + // Wait for assembly to start producing events + await Promise.resolve() + stop = startResult.stop interceptor = interceptRequests() } @@ -344,14 +351,14 @@ describe('view events', () => { }) }) - it('sends a view update on page unload when bridge is absent', () => { + it('sends a view update on page unload when bridge is absent', async () => { // Note: this test is intentionally very high level to make sure the view update is correctly // made right before flushing the Batch. // Arbitrary duration to simulate a non-zero view duration const VIEW_DURATION = ONE_SECOND as Duration - setupViewCollectionTest() + await setupViewCollectionTest() clock.tick(VIEW_DURATION - relativeNow()) window.dispatchEvent(createNewEvent('beforeunload')) @@ -367,13 +374,13 @@ describe('view events', () => { expect(lastRumViewEvent.view.time_spent).toBe(toServerDuration(VIEW_DURATION)) }) - it('sends a view update on page unload when bridge is present', () => { + it('sends a view update on page unload when bridge is present', async () => { const eventBridge = mockEventBridge() const sendSpy = spyOn(eventBridge, 'send') const VIEW_DURATION = ONE_SECOND as Duration - setupViewCollectionTest() + await setupViewCollectionTest() clock.tick(VIEW_DURATION - relativeNow()) window.dispatchEvent(createNewEvent('beforeunload')) diff --git a/packages/rum-core/src/domain/assembly.spec.ts b/packages/rum-core/src/domain/assembly.spec.ts index 5d0f0e9c52..be51adcfe9 100644 --- a/packages/rum-core/src/domain/assembly.spec.ts +++ b/packages/rum-core/src/domain/assembly.spec.ts @@ -27,8 +27,8 @@ describe('rum assembly', () => { describe('beforeSend', () => { describe('fields modification', () => { describe('modifiable fields', () => { - it('should allow modification', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('should allow modification', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: (event) => (event.view.url = 'modified'), }, @@ -38,11 +38,12 @@ describe('rum assembly', () => { rawRumEvent: createRawRumEvent(RumEventType.LONG_TASK, { view: { url: '/path?foo=bar' } }), }) - expect(serverRumEvents[0].view.url).toBe('modified') + const rumEvents = await getRumEvents() + expect(rumEvents[0].view.url).toBe('modified') }) - it('should allow addition', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('should allow addition', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: (event) => (event.view.name = 'added'), }, @@ -52,11 +53,12 @@ describe('rum assembly', () => { rawRumEvent: createRawRumEvent(RumEventType.LONG_TASK, { view: { url: '/path?foo=bar' } }), }) - expect(serverRumEvents[0].view.name).toBe('added') + const rumEvents = await getRumEvents() + expect(rumEvents[0].view.name).toBe('added') }) - it('should allow modification of view.performance.lcp.resource_url', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('should allow modification of view.performance.lcp.resource_url', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: (event) => (event.view.performance.lcp.resource_url = 'modified_url'), }, @@ -68,12 +70,13 @@ describe('rum assembly', () => { }), }) - expect((serverRumEvents[0].view as any).performance.lcp.resource_url).toBe('modified_url') + const rumEvents = await getRumEvents() + expect((rumEvents[0].view as any).performance.lcp.resource_url).toBe('modified_url') }) describe('field resource.graphql on Resource events', () => { - it('by default, it should not be modifiable', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('by default, it should not be modifiable', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: (event) => (event.resource!.graphql = { operationType: 'query' }), }, @@ -83,13 +86,14 @@ describe('rum assembly', () => { rawRumEvent: createRawRumEvent(RumEventType.RESOURCE, { resource: { url: '/path?foo=bar' } }), }) - expect((serverRumEvents[0] as RumResourceEvent).resource.graphql).toBeUndefined() + const rumEvents = await getRumEvents() + expect((rumEvents[0] as RumResourceEvent).resource.graphql).toBeUndefined() }) - it('with the writable_resource_graphql experimental flag is set, it should be modifiable', () => { + it('with the writable_resource_graphql experimental flag is set, it should be modifiable', async () => { mockExperimentalFeatures([ExperimentalFeature.WRITABLE_RESOURCE_GRAPHQL]) - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: (event) => (event.resource!.graphql = { operationType: 'query' }), }, @@ -99,14 +103,15 @@ describe('rum assembly', () => { rawRumEvent: createRawRumEvent(RumEventType.RESOURCE, { resource: { url: '/path?foo=bar' } }), }) - expect((serverRumEvents[0] as RumResourceEvent).resource.graphql).toEqual({ operationType: 'query' }) + const rumEvents = await getRumEvents() + expect((rumEvents[0] as RumResourceEvent).resource.graphql).toEqual({ operationType: 'query' }) }) }) }) describe('context field', () => { - it('should allow modification on context field for events other than views', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('should allow modification on context field for events other than views', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: (event) => { event.context.foo = 'bar' @@ -118,11 +123,12 @@ describe('rum assembly', () => { rawRumEvent: createRawRumEvent(RumEventType.LONG_TASK), }) - expect(serverRumEvents[0].context!.foo).toBe('bar') + const rumEvents = await getRumEvents() + expect(rumEvents[0].context!.foo).toBe('bar') }) - it('should allow replacing the context field for events other than views', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('should allow replacing the context field for events other than views', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: (event) => { event.context.foo = 'bar' @@ -134,11 +140,12 @@ describe('rum assembly', () => { rawRumEvent: createRawRumEvent(RumEventType.LONG_TASK), }) - expect(serverRumEvents[0].context!.foo).toBe('bar') + const rumEvents = await getRumEvents() + expect(rumEvents[0].context!.foo).toBe('bar') }) - it('should empty the context field if set to null', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('should empty the context field if set to null', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: (event) => { event.context = null @@ -151,11 +158,12 @@ describe('rum assembly', () => { customerContext: { foo: 'bar' }, }) - expect(serverRumEvents[0].context).toBeUndefined() + const rumEvents = await getRumEvents() + expect(rumEvents[0].context).toBeUndefined() }) - it('should empty the context field if set to undefined', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('should empty the context field if set to undefined', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: (event) => { event.context = undefined @@ -168,11 +176,12 @@ describe('rum assembly', () => { customerContext: { foo: 'bar' }, }) - expect(serverRumEvents[0].context).toBeUndefined() + const rumEvents = await getRumEvents() + expect(rumEvents[0].context).toBeUndefined() }) - it('should empty the context field if deleted', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('should empty the context field if deleted', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: (event) => { delete event.context @@ -185,11 +194,12 @@ describe('rum assembly', () => { customerContext: { foo: 'bar' }, }) - expect(serverRumEvents[0].context).toBeUndefined() + const rumEvents = await getRumEvents() + expect(rumEvents[0].context).toBeUndefined() }) - it('should define the context field even if the global context is empty', () => { - const { lifeCycle } = setupAssemblyTestWithDefaults({ + it('should define the context field even if the global context is empty', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: (event) => { expect(event.context).toEqual({}) @@ -200,10 +210,12 @@ describe('rum assembly', () => { notifyRawRumEvent(lifeCycle, { rawRumEvent: createRawRumEvent(RumEventType.LONG_TASK), }) + + await getRumEvents() }) - it('should accept modification on context field for view events', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('should accept modification on context field for view events', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: (event) => { event.context.foo = 'bar' @@ -215,11 +227,12 @@ describe('rum assembly', () => { rawRumEvent: createRawRumEvent(RumEventType.VIEW), }) - expect(serverRumEvents[0].context).toEqual({ foo: 'bar' }) + const rumEvents = await getRumEvents() + expect(rumEvents[0].context).toEqual({ foo: 'bar' }) }) - it('should reject replacing the context field to non-object', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('should reject replacing the context field to non-object', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: (event) => { event.context = 1 @@ -232,13 +245,14 @@ describe('rum assembly', () => { customerContext: { foo: 'bar' }, }) - expect(serverRumEvents[0].context!.foo).toBe('bar') + const rumEvents = await getRumEvents() + expect(rumEvents[0].context!.foo).toBe('bar') }) }) describe('allowed customer provided field', () => { - it('should allow modification of the error fingerprint', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('should allow modification of the error fingerprint', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: (event) => (event.error.fingerprint = 'my_fingerprint'), }, @@ -248,12 +262,13 @@ describe('rum assembly', () => { rawRumEvent: createRawRumEvent(RumEventType.ERROR), }) - expect((serverRumEvents[0] as RumErrorEvent).error.fingerprint).toBe('my_fingerprint') + const rumEvents = await getRumEvents() + expect((rumEvents[0] as RumErrorEvent).error.fingerprint).toBe('my_fingerprint') }) }) - it('should reject modification of field not sensitive, context or customer provided', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('should reject modification of field not sensitive, context or customer provided', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: (event: RumEvent) => ((event.view as any).id = 'modified'), }, @@ -265,11 +280,12 @@ describe('rum assembly', () => { }), }) - expect(serverRumEvents[0].view.id).toBe('aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee') + const rumEvents = await getRumEvents() + expect(rumEvents[0].view.id).toBe('aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee') }) - it('should not allow to add a sensitive field on the wrong event type', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('should not allow to add a sensitive field on the wrong event type', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: (event) => { event.error = { message: 'added' } @@ -281,13 +297,14 @@ describe('rum assembly', () => { rawRumEvent: createRawRumEvent(RumEventType.VIEW), }) - expect((serverRumEvents[0] as any).error?.message).toBeUndefined() + const rumEvents = await getRumEvents() + expect((rumEvents[0] as any).error?.message).toBeUndefined() }) }) describe('events dismission', () => { - it('should allow dismissing events other than views', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('should allow dismissing events other than views', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: () => false, }, @@ -317,11 +334,12 @@ describe('rum assembly', () => { }), }) - expect(serverRumEvents.length).toBe(0) + const rumEvents = await getRumEvents() + expect(rumEvents.length).toBe(0) }) - it('should not allow dismissing view events', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('should not allow dismissing view events', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: () => false, }, @@ -334,13 +352,14 @@ describe('rum assembly', () => { }), }) - expect(serverRumEvents[0].view.id).toBe('aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee') + const rumEvents = await getRumEvents() + expect(rumEvents[0].view.id).toBe('aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee') expect(displaySpy).toHaveBeenCalledWith("Can't dismiss view events using beforeSend!") }) }) - it('should not dismiss when true is returned', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('should not dismiss when true is returned', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: () => true, }, @@ -352,11 +371,12 @@ describe('rum assembly', () => { }), }) - expect(serverRumEvents.length).toBe(1) + const rumEvents = await getRumEvents() + expect(rumEvents.length).toBe(1) }) - it('should not dismiss when undefined is returned', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('should not dismiss when undefined is returned', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { beforeSend: () => undefined, }, @@ -368,19 +388,21 @@ describe('rum assembly', () => { }), }) - expect(serverRumEvents.length).toBe(1) + const rumEvents = await getRumEvents() + expect(rumEvents.length).toBe(1) }) }) describe('customer context', () => { - it('should be merged with event attributes', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults() + it('should be merged with event attributes', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults() notifyRawRumEvent(lifeCycle, { customerContext: { foo: 'bar' }, rawRumEvent: createRawRumEvent(RumEventType.VIEW), }) - expect((serverRumEvents[0].context as any).foo).toEqual('bar') + const rumEvents = await getRumEvents() + expect((rumEvents[0].context as any).foo).toEqual('bar') }) }) @@ -388,8 +410,8 @@ describe('rum assembly', () => { const extraConfigurationOptions = { service: 'default service', version: 'default version' } describe('fields service and version', () => { - it('it should be modifiable', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('it should be modifiable', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { ...extraConfigurationOptions, beforeSend: (event) => { @@ -404,21 +426,22 @@ describe('rum assembly', () => { notifyRawRumEvent(lifeCycle, { rawRumEvent: createRawRumEvent(RumEventType.RESOURCE), }) - expect((serverRumEvents[0] as RumResourceEvent).service).toBe('bar') - expect((serverRumEvents[0] as RumResourceEvent).version).toBe('0.2.0') + const rumEvents = await getRumEvents() + expect((rumEvents[0] as RumResourceEvent).service).toBe('bar') + expect((rumEvents[0] as RumResourceEvent).version).toBe('0.2.0') notifyRawRumEvent(lifeCycle, { rawRumEvent: createRawRumEvent(RumEventType.VIEW), }) - expect((serverRumEvents[1] as RumViewEvent).service).toBe('bar') - expect((serverRumEvents[1] as RumViewEvent).version).toBe('0.2.0') + expect((rumEvents[1] as RumViewEvent).service).toBe('bar') + expect((rumEvents[1] as RumViewEvent).version).toBe('0.2.0') }) }) }) describe('assemble hook', () => { - it('should add and override common properties', () => { - const { lifeCycle, hooks, serverRumEvents } = setupAssemblyTestWithDefaults({ + it('should add and override common properties', async () => { + const { lifeCycle, hooks, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { service: 'default service', version: 'default version' }, }) @@ -432,13 +455,14 @@ describe('rum assembly', () => { notifyRawRumEvent(lifeCycle, { rawRumEvent: createRawRumEvent(RumEventType.ACTION), }) - expect(serverRumEvents[0].service).toEqual('new service') - expect(serverRumEvents[0].version).toEqual('new version') - expect(serverRumEvents[0].view.id).toEqual('new view id') + const rumEvents = await getRumEvents() + expect(rumEvents[0].service).toEqual('new service') + expect(rumEvents[0].version).toEqual('new version') + expect(rumEvents[0].view.id).toEqual('new view id') }) - it('should not override customer context', () => { - const { lifeCycle, hooks, serverRumEvents } = setupAssemblyTestWithDefaults() + it('should not override customer context', async () => { + const { lifeCycle, hooks, getRumEvents } = setupAssemblyTestWithDefaults() hooks.register(HookNames.Assemble, ({ eventType }) => ({ type: eventType, @@ -449,39 +473,44 @@ describe('rum assembly', () => { rawRumEvent: createRawRumEvent(RumEventType.ACTION), customerContext: { foo: 'customer context' }, }) - expect(serverRumEvents[0].context).toEqual({ foo: 'customer context' }) + const rumEvents = await getRumEvents() + expect(rumEvents[0].context).toEqual({ foo: 'customer context' }) }) }) describe('event generation condition', () => { - it('when tracked, it should generate event', () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults() + it('when tracked, it should generate event', async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults() notifyRawRumEvent(lifeCycle, { rawRumEvent: createRawRumEvent(RumEventType.VIEW), }) - expect(serverRumEvents.length).toBe(1) + const rumEvents = await getRumEvents() + expect(rumEvents.length).toBe(1) }) - it('when not tracked, it should not generate event', () => { + it('when not tracked, it should not generate event', async () => { const sessionManager = createRumSessionManagerMock().setNotTracked() - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ sessionManager }) + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ sessionManager }) notifyRawRumEvent(lifeCycle, { rawRumEvent: createRawRumEvent(RumEventType.VIEW), }) - expect(serverRumEvents.length).toBe(0) + const rumEvents = await getRumEvents() + expect(rumEvents.length).toBe(0) }) - it('should get session state from event start', () => { + it('should get session state from event start', async () => { const sessionManager = createRumSessionManagerMock() spyOn(sessionManager, 'findTrackedSession').and.callThrough() - const { lifeCycle } = setupAssemblyTestWithDefaults({ sessionManager }) + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ sessionManager }) notifyRawRumEvent(lifeCycle, { rawRumEvent: createRawRumEvent(RumEventType.ACTION), startTime: 123 as RelativeTime, }) + await getRumEvents() + expect(sessionManager.findTrackedSession).toHaveBeenCalledWith(123 as RelativeTime) }) }) @@ -500,8 +529,8 @@ describe('rum assembly', () => { }, ].forEach(({ eventType, message }) => { describe(`${eventType} events limitation`, () => { - it(`stops sending ${eventType} events when reaching the limit`, () => { - const { lifeCycle, serverRumEvents, reportErrorSpy } = setupAssemblyTestWithDefaults({ + it(`stops sending ${eventType} events when reaching the limit`, async () => { + const { lifeCycle, getRumEvents, reportErrorSpy } = setupAssemblyTestWithDefaults({ partialConfiguration: { eventRateLimiterThreshold: 1 }, }) @@ -512,8 +541,9 @@ describe('rum assembly', () => { rawRumEvent: createRawRumEvent(eventType, { date: 200 as TimeStamp }), }) - expect(serverRumEvents.length).toBe(1) - expect(serverRumEvents[0].date).toBe(100) + const rumEvents = await getRumEvents() + expect(rumEvents.length).toBe(1) + expect(rumEvents[0].date).toBe(100) expect(reportErrorSpy).toHaveBeenCalledTimes(1) expect(reportErrorSpy.calls.argsFor(0)[0]).toEqual( jasmine.objectContaining({ @@ -523,8 +553,8 @@ describe('rum assembly', () => { ) }) - it(`does not take discarded ${eventType} events into account`, () => { - const { lifeCycle, serverRumEvents, reportErrorSpy } = setupAssemblyTestWithDefaults({ + it(`does not take discarded ${eventType} events into account`, async () => { + const { lifeCycle, getRumEvents, reportErrorSpy } = setupAssemblyTestWithDefaults({ partialConfiguration: { eventRateLimiterThreshold: 1, beforeSend: (event) => { @@ -547,8 +577,9 @@ describe('rum assembly', () => { notifyRawRumEvent(lifeCycle, { rawRumEvent: createRawRumEvent(eventType, { date: 200 as TimeStamp }), }) - expect(serverRumEvents.length).toBe(1) - expect(serverRumEvents[0].date).toBe(200) + const rumEvents = await getRumEvents() + expect(rumEvents.length).toBe(1) + expect(rumEvents[0].date).toBe(200) expect(reportErrorSpy).not.toHaveBeenCalled() }) @@ -558,11 +589,13 @@ describe('rum assembly', () => { clock = mockClock() }) - it(`allows to send new ${eventType} events after a minute`, () => { - const { lifeCycle, serverRumEvents } = setupAssemblyTestWithDefaults({ + it(`allows to send new ${eventType} events after a minute`, async () => { + const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults({ partialConfiguration: { eventRateLimiterThreshold: 1 }, }) + const rumEvents = await getRumEvents() + notifyRawRumEvent(lifeCycle, { rawRumEvent: createRawRumEvent(eventType, { date: 100 as TimeStamp }), }) @@ -574,9 +607,9 @@ describe('rum assembly', () => { rawRumEvent: createRawRumEvent(eventType, { date: 300 as TimeStamp }), }) - expect(serverRumEvents.length).toBe(2) - expect(serverRumEvents[0].date).toBe(100) - expect(serverRumEvents[1].date).toBe(300) + expect(rumEvents.length).toBe(2) + expect(rumEvents[0].date).toBe(100) + expect(rumEvents[1].date).toBe(300) }) }) }) @@ -618,7 +651,7 @@ function setupAssemblyTestWithDefaults({ }) const recorderApi = noopRecorderApi const viewHistory = { ...mockViewHistory(), findView: () => findView() } - startGlobalContext(hooks, mockRumConfiguration()) + const globalContext = startGlobalContext(hooks, mockRumConfiguration()) startSessionContext(hooks, rumSessionManager, recorderApi, viewHistory) startRumAssembly(mockRumConfiguration(partialConfiguration), lifeCycle, hooks, reportErrorSpy) @@ -626,5 +659,16 @@ function setupAssemblyTestWithDefaults({ subscription.unsubscribe() }) - return { lifeCycle, hooks, reportErrorSpy, serverRumEvents, recorderApi } + return { + lifeCycle, + hooks, + reportErrorSpy, + getRumEvents: async () => { + // Wait for assembly to start producing events + await Promise.resolve() + return serverRumEvents + }, + recorderApi, + globalContext, + } } From a3f665b0625ea4543c608c4f735739b34df68552 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Beno=C3=AEt=20Zugmeyer?= Date: Wed, 28 May 2025 15:13:05 +0200 Subject: [PATCH 4/5] =?UTF-8?q?=E2=9C=A8=20[RUM-10144]=20delay=20assembly?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/rum-core/src/domain/assembly.spec.ts | 28 +++++++++++ packages/rum-core/src/domain/assembly.ts | 47 +++++++++++-------- 2 files changed, 55 insertions(+), 20 deletions(-) diff --git a/packages/rum-core/src/domain/assembly.spec.ts b/packages/rum-core/src/domain/assembly.spec.ts index be51adcfe9..e87e1ef85a 100644 --- a/packages/rum-core/src/domain/assembly.spec.ts +++ b/packages/rum-core/src/domain/assembly.spec.ts @@ -478,6 +478,34 @@ describe('rum assembly', () => { }) }) + describe('global context', () => { + it('applies global context to events', async () => { + const { lifeCycle, globalContext, getRumEvents } = setupAssemblyTestWithDefaults() + + globalContext.setContext({ foo: 'bar' }) + + notifyRawRumEvent(lifeCycle, { + rawRumEvent: createRawRumEvent(RumEventType.ACTION), + }) + + const rumEvents = await getRumEvents() + expect(rumEvents[0].context).toEqual({ foo: 'bar' }) + }) + + it('applies global context to events generated before the global context is set', async () => { + const { lifeCycle, globalContext, getRumEvents } = setupAssemblyTestWithDefaults() + + notifyRawRumEvent(lifeCycle, { + rawRumEvent: createRawRumEvent(RumEventType.ACTION), + }) + + globalContext.setContext({ foo: 'bar' }) + + const rumEvents = await getRumEvents() + expect(rumEvents[0].context).toEqual({ foo: 'bar' }) + }) + }) + describe('event generation condition', () => { it('when tracked, it should generate event', async () => { const { lifeCycle, getRumEvents } = setupAssemblyTestWithDefaults() diff --git a/packages/rum-core/src/domain/assembly.ts b/packages/rum-core/src/domain/assembly.ts index 4eaa22a945..b994f007a0 100644 --- a/packages/rum-core/src/domain/assembly.ts +++ b/packages/rum-core/src/domain/assembly.ts @@ -6,13 +6,14 @@ import { createEventRateLimiter, isExperimentalFeatureEnabled, ExperimentalFeature, + BufferedObservable, HookNames, DISCARDED, } from '@datadog/browser-core' import type { RumEventDomainContext } from '../domainContext.types' import { RumEventType } from '../rawRumEvent.types' import type { RumEvent } from '../rumEvent.types' -import type { LifeCycle } from './lifeCycle' +import type { LifeCycle, RawRumEventCollectedData } from './lifeCycle' import { LifeCycleEventType } from './lifeCycle' import type { RumConfiguration } from './configuration' import type { ModifiableFieldPaths } from './limitModification' @@ -34,6 +35,10 @@ const ROOT_MODIFIABLE_FIELD_PATHS: ModifiableFieldPaths = { version: 'string', } +// The size of the buffer for events that are collected just after starting the assembly. This is a +// bit arbitrary, but should be large enough to avoid dropping events in most cases. +const BUFFERED_EVENT_SIZE = 100 + let modifiableFieldPathsByEvent: { [key in RumEventType]: ModifiableFieldPaths } export function startRumAssembly( @@ -102,30 +107,32 @@ export function startRumAssembly( ), } - lifeCycle.subscribe( - LifeCycleEventType.RAW_RUM_EVENT_COLLECTED, - ({ startTime, duration, rawRumEvent, domainContext, customerContext }) => { - const defaultRumEventAttributes = hooks.triggerHook(HookNames.Assemble, { - eventType: rawRumEvent.type, - startTime, - duration, - })! + const observable = new BufferedObservable(BUFFERED_EVENT_SIZE) - if (defaultRumEventAttributes === DISCARDED) { - return - } + lifeCycle.subscribe(LifeCycleEventType.RAW_RUM_EVENT_COLLECTED, (data) => observable.notify(data)) + + observable.subscribe(({ startTime, duration, rawRumEvent, domainContext, customerContext }) => { + const defaultRumEventAttributes = hooks.triggerHook(HookNames.Assemble, { + eventType: rawRumEvent.type, + startTime, + duration, + })! + + if (defaultRumEventAttributes === DISCARDED) { + return + } - const serverRumEvent = combine(defaultRumEventAttributes, { context: customerContext }, rawRumEvent) as RumEvent & - Context + const serverRumEvent = combine(defaultRumEventAttributes, { context: customerContext }, rawRumEvent) as RumEvent & + Context - if (shouldSend(serverRumEvent, configuration.beforeSend, domainContext, eventRateLimiters)) { - if (isEmptyObject(serverRumEvent.context!)) { - delete serverRumEvent.context - } - lifeCycle.notify(LifeCycleEventType.RUM_EVENT_COLLECTED, serverRumEvent) + if (shouldSend(serverRumEvent, configuration.beforeSend, domainContext, eventRateLimiters)) { + if (isEmptyObject(serverRumEvent.context!)) { + delete serverRumEvent.context } + lifeCycle.notify(LifeCycleEventType.RUM_EVENT_COLLECTED, serverRumEvent) } - ) + }) + observable.unbuffer() } function shouldSend( From 3d012c1d005d3bfca28c45d9694733ec5f48d64e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Beno=C3=AEt=20Zugmeyer?= Date: Tue, 3 Jun 2025 18:23:26 +0200 Subject: [PATCH 5/5] =?UTF-8?q?=E2=9C=85=20[RUM-10144]=20add=20an=20e2e=20?= =?UTF-8?q?test?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- test/e2e/scenario/rum/init.scenario.ts | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/test/e2e/scenario/rum/init.scenario.ts b/test/e2e/scenario/rum/init.scenario.ts index 2681010744..4250484001 100644 --- a/test/e2e/scenario/rum/init.scenario.ts +++ b/test/e2e/scenario/rum/init.scenario.ts @@ -183,6 +183,23 @@ test.describe('API calls and events around init', () => { const viewContext = await page.evaluate(() => window.DD_RUM?.getViewContext()) expect(viewContext).toEqual({ foo: 'bar' }) }) + + createTest('context set right after init should be applied to events generated during init') + .withRum() + .withRumSlim() + .withRumInit((configuration) => { + window.DD_RUM!.init(configuration) + window.DD_RUM!.setViewContext({ viewContext: true }) + window.DD_RUM!.setGlobalContext({ globalContext: true }) + window.DD_RUM!.setUser({ id: 'user-id' }) + }) + .run(async ({ intakeRegistry, flushEvents }) => { + await flushEvents() + + const initialView = intakeRegistry.rumViewEvents[0] + expect(initialView.context).toEqual({ viewContext: true, globalContext: true }) + expect(initialView.usr).toEqual({ id: 'user-id' }) + }) }) test.describe('beforeSend', () => {