diff --git a/packages/core/src/domain/session/sessionManager.ts b/packages/core/src/domain/session/sessionManager.ts index 789d0d5487..68c4d9d7a4 100644 --- a/packages/core/src/domain/session/sessionManager.ts +++ b/packages/core/src/domain/session/sessionManager.ts @@ -33,6 +33,8 @@ export interface SessionContext extends Context { * just because the user moved to another page. */ hasError: boolean + /** Where the detail stored for this session starts, when its events were withheld for a while. */ + detailSampledFrom: number | undefined anonymousId: string | undefined } @@ -99,6 +101,7 @@ export function startSessionManager( trackingType: sessionStore.getSession()[productKey] as TrackingType, isReplayForced: !!sessionStore.getSession().forcedReplay, hasError: !!sessionStore.getSession().hasError, + detailSampledFrom: Number(sessionStore.getSession().detailFrom) || undefined, anonymousId: sessionStore.getSession().anonymousId, } } diff --git a/packages/rum-core/src/boot/startRum.ts b/packages/rum-core/src/boot/startRum.ts index 0c1af632b9..b1872e08a2 100644 --- a/packages/rum-core/src/boot/startRum.ts +++ b/packages/rum-core/src/boot/startRum.ts @@ -111,6 +111,9 @@ export function startRum( ? startRumSessionManager(configuration, lifeCycle, trackingConsentState) : startRumSessionManagerStub() + // Subscribed before the batch below, and it has to stay that way: the withheld event buffer runs + // on the same event, and only sees a session as released if this has already marked it. Reorder + // them and the release waits for whatever event happens to come next. const sessionErrorTracking = startSessionErrorTracking(lifeCycle, session) cleanupTasks.push(() => sessionErrorTracking.stop()) @@ -121,7 +124,7 @@ export function startRum( telemetry.observable, reportError, pageMayExitObservable, - session.expireObservable, + session, createEncoder ) cleanupTasks.push(() => batch.stop()) diff --git a/packages/rum-core/src/domain/configuration/configuration.spec.ts b/packages/rum-core/src/domain/configuration/configuration.spec.ts index a2c4c48875..558b208d23 100644 --- a/packages/rum-core/src/domain/configuration/configuration.spec.ts +++ b/packages/rum-core/src/domain/configuration/configuration.spec.ts @@ -533,6 +533,7 @@ describe('serializeRumConfiguration', () => { subdomain: 'foo', sessionReplaySampleRate: 60, sessionReplayOnErrorSampleRate: 40, + sessionOnErrorSampleRate: 30, startSessionReplayRecordingManually: true, trackUserInteractions: true, actionNameAttribute: 'test-id', @@ -560,6 +561,7 @@ describe('serializeRumConfiguration', () => { | 'propagateTraceBaggage' // not reported yet: needs a rum-events-format schema change first | 'sessionReplayOnErrorSampleRate' + | 'sessionOnErrorSampleRate' ? never : CamelToSnakeCase // By specifying the type here, we can ensure that serializeConfiguration is returning an diff --git a/packages/rum-core/src/domain/configuration/configuration.ts b/packages/rum-core/src/domain/configuration/configuration.ts index 33743b5d9e..85476d86b7 100644 --- a/packages/rum-core/src/domain/configuration/configuration.ts +++ b/packages/rum-core/src/domain/configuration/configuration.ts @@ -110,6 +110,19 @@ export interface RumInitConfiguration extends InitConfiguration { * the withheld minute is uploaded and recording continues normally for the rest of the session. */ sessionReplayOnErrorSampleRate?: number | undefined + /** + * The percentage of tracked sessions that collect events but only upload them if the session + * reports an error: 100 for all, 0 for none. Drawn only for sessions that the plain + * `sessionSampleRate` draw missed, so a session is never counted by both rates. + * + * Such a session collects from the start and keeps at most the last minute of it in memory. If it + * never reports an error, nothing is uploaded and the session is not stored. On the first error, + * the withheld minute is uploaded and collection continues normally. + * + * A session sampled this way never uploads its replay ahead of its events: until the events are + * released the session does not exist yet, and a replay sent then would have nothing to attach to. + */ + sessionOnErrorSampleRate?: number | undefined /** * If the session is sampled for Session Replay, only start the recording when `startSessionReplayRecording()` is called, instead of at the beginning of the session. Default: if startSessionReplayRecording is 0, true; otherwise, false. * See [Session Replay Usage](https://docs.datadoghq.com/real_user_monitoring/session_replay/browser/#usage) for further information. @@ -186,6 +199,7 @@ export interface RumConfiguration extends Configuration { enablePrivacyForActionName: boolean sessionReplaySampleRate: number sessionReplayOnErrorSampleRate: number + sessionOnErrorSampleRate: number startSessionReplayRecordingManually: boolean trackUserInteractions: boolean trackViewsManually: boolean @@ -219,6 +233,7 @@ export function validateAndBuildRumConfiguration( if ( !isSampleRate(initConfiguration.sessionReplaySampleRate, 'Session Replay') || !isSampleRate(initConfiguration.sessionReplayOnErrorSampleRate, 'Session Replay on Error') || + !isSampleRate(initConfiguration.sessionOnErrorSampleRate, 'Session on Error') || !isSampleRate(initConfiguration.traceSampleRate, 'Trace') ) { return @@ -243,6 +258,7 @@ export function validateAndBuildRumConfiguration( const sessionReplaySampleRate = initConfiguration.sessionReplaySampleRate ?? 0 const sessionReplayOnErrorSampleRate = initConfiguration.sessionReplayOnErrorSampleRate ?? 0 + const sessionOnErrorSampleRate = initConfiguration.sessionOnErrorSampleRate ?? 0 return { applicationId: initConfiguration.applicationId, @@ -250,6 +266,7 @@ export function validateAndBuildRumConfiguration( actionNameAttribute: initConfiguration.actionNameAttribute, sessionReplaySampleRate, sessionReplayOnErrorSampleRate, + sessionOnErrorSampleRate, startSessionReplayRecordingManually: initConfiguration.startSessionReplayRecordingManually !== undefined ? !!initConfiguration.startSessionReplayRecordingManually diff --git a/packages/rum-core/src/domain/contexts/sessionContext.ts b/packages/rum-core/src/domain/contexts/sessionContext.ts index a520a8524b..317bea5300 100644 --- a/packages/rum-core/src/domain/contexts/sessionContext.ts +++ b/packages/rum-core/src/domain/contexts/sessionContext.ts @@ -26,11 +26,17 @@ export function startSessionContext( let hasReplay let sampledForReplay + let sampledForError + let detailSampledFrom let sampledForErrorReplay let isActive if (eventType === RumEventType.VIEW) { hasReplay = !isReplayWithheld && recorderApi.getReplayStats(view.id) ? true : undefined sampledForReplay = session.sessionReplay === SessionReplayState.SAMPLED + // Tells the backend that this session's detail only starts where the buffer reached, so the + // gap before it reads as "not collected" rather than as missing data. + sampledForError = session.sampledOnError || undefined + detailSampledFrom = session.detailSampledFrom // Tells a replay collected only because the session errored apart from one collected // unconditionally - the two cost differently and are answered by different questions. sampledForErrorReplay = session.sampledOnErrorReplay || undefined @@ -46,6 +52,8 @@ export function startSessionContext( type: SessionType.USER, has_replay: hasReplay, sampled_for_replay: sampledForReplay, + sampled_for_error: sampledForError, + detail_sampled_from: detailSampledFrom, sampled_for_error_replay: sampledForErrorReplay, is_active: isActive, }, diff --git a/packages/rum-core/src/domain/rumSessionManager.spec.ts b/packages/rum-core/src/domain/rumSessionManager.spec.ts index 21dd0a499d..d688666efc 100644 --- a/packages/rum-core/src/domain/rumSessionManager.spec.ts +++ b/packages/rum-core/src/domain/rumSessionManager.spec.ts @@ -290,6 +290,79 @@ describe('rum session manager', () => { }) }) + describe('on-error session sampling', () => { + const ON_ERROR_ONLY = { + sessionSampleRate: 0, + sessionOnErrorSampleRate: 100, + sessionReplaySampleRate: 0, + sessionReplayOnErrorSampleRate: 0, + } + + it('draws the on-error type only when the plain session draw missed', () => { + startRumSessionManagerWithDefaults({ + configuration: { ...ON_ERROR_ONLY, sessionSampleRate: 100 }, + }) + + expect(getSessionState(SESSION_STORE_KEY)[RUM_SESSION_KEY]).toBe(RumTrackingType.TRACKED_WITHOUT_SESSION_REPLAY) + }) + + it('withholds the events of a session drawn on error', () => { + const sessionManager = startRumSessionManagerWithDefaults({ configuration: ON_ERROR_ONLY }) + + expect(getSessionState(SESSION_STORE_KEY)[RUM_SESSION_KEY]).toBe( + RumTrackingType.TRACKED_ON_ERROR_WITHOUT_SESSION_REPLAY + ) + expect(sessionManager.findTrackedSession()!.eventsWithheld).toBeTrue() + + sessionManager.setSessionHasError() + + expect(sessionManager.findTrackedSession()!.eventsWithheld).toBeFalse() + }) + + it('withholds the replay alongside the events, even when the plain replay rate was drawn', () => { + // a replay uploaded while the events are withheld would have no session to attach to + const sessionManager = startRumSessionManagerWithDefaults({ + configuration: { ...ON_ERROR_ONLY, sessionReplaySampleRate: 100 }, + }) + + expect(getSessionState(SESSION_STORE_KEY)[RUM_SESSION_KEY]).toBe( + RumTrackingType.TRACKED_ON_ERROR_WITH_SESSION_REPLAY + ) + expect(sessionManager.findTrackedSession()!.sessionReplay).toBe(SessionReplayState.BUFFERED_ON_ERROR) + }) + + it('releases events and replay together on the first error', () => { + const sessionManager = startRumSessionManagerWithDefaults({ + configuration: { ...ON_ERROR_ONLY, sessionReplaySampleRate: 100 }, + }) + + sessionManager.setSessionHasError() + + const session = sessionManager.findTrackedSession()! + expect(session.eventsWithheld).toBeFalse() + expect(session.sessionReplay).toBe(SessionReplayState.SAMPLED) + }) + + it('releases the events when capture is forced, so the forced replay is not left orphaned', () => { + const sessionManager = startRumSessionManagerWithDefaults({ + configuration: { ...ON_ERROR_ONLY, sessionReplaySampleRate: 100 }, + }) + + sessionManager.setForcedReplay() + + const session = sessionManager.findTrackedSession()! + expect(session.eventsWithheld).toBeFalse() + expect(session.sessionReplay).toBe(SessionReplayState.FORCED) + }) + + it('keeps marking the session as on-error once its events have been released', () => { + const sessionManager = startRumSessionManagerWithDefaults({ configuration: ON_ERROR_ONLY }) + sessionManager.setSessionHasError() + + expect(sessionManager.findTrackedSession()!.sampledOnError).toBeTrue() + }) + }) + function startRumSessionManagerWithDefaults({ configuration }: { configuration?: Partial } = {}) { return startRumSessionManager( mockRumConfiguration({ diff --git a/packages/rum-core/src/domain/rumSessionManager.ts b/packages/rum-core/src/domain/rumSessionManager.ts index 02ddb45388..a8c9d4fd95 100644 --- a/packages/rum-core/src/domain/rumSessionManager.ts +++ b/packages/rum-core/src/domain/rumSessionManager.ts @@ -29,17 +29,34 @@ export interface RumSessionManager { * `sessionReplayOnErrorSampleRate`, this is what releases the withheld replay. */ setSessionHasError: () => void + /** Records how far back the detail released for this session actually reaches. */ + setSessionDetailSampledFrom: (timestamp: number) => void } export type RumSession = { id: string sessionReplay: SessionReplayState /** - * Whether the replay of this session is only kept if it reports an error. Unlike - * {@link sessionReplay} this stays true once the error has been reported, so a replay collected - * that way can be told apart from one collected unconditionally. + * Whether the session collects events but withholds them until it reports an error. Nothing is + * uploaded while this is true, and if the session never errors nothing ever is. + */ + eventsWithheld: boolean + /** + * Whether the session was drawn by `sessionOnErrorSampleRate`. Unlike {@link eventsWithheld} this + * stays true once the error has been reported, so what is stored can be told apart from a plainly + * sampled session - its detail only starts where the buffer reached. + */ + sampledOnError: boolean + /** + * Whether the replay of this session is only kept if it reports an error. Same idea as + * {@link sampledOnError}, for the replay rather than the events. */ sampledOnErrorReplay: boolean + /** + * Where the detail stored for this session starts, for a session whose events were withheld. The + * gap before it is data that was never collected rather than data that went missing. + */ + detailSampledFrom?: number anonymousId?: string } @@ -48,6 +65,8 @@ export const enum RumTrackingType { TRACKED_WITH_SESSION_REPLAY = '1', TRACKED_WITHOUT_SESSION_REPLAY = '2', TRACKED_WITH_ERROR_SESSION_REPLAY = '3', + TRACKED_ON_ERROR_WITHOUT_SESSION_REPLAY = '4', + TRACKED_ON_ERROR_WITH_SESSION_REPLAY = '5', } export const enum SessionReplayState { @@ -95,6 +114,12 @@ export function startRumSessionManager( sessionEntity.hasError = true } } + if (!previousState.detailFrom && newState.detailFrom) { + const sessionEntity = sessionManager.findSession() + if (sessionEntity) { + sessionEntity.detailSampledFrom = Number(newState.detailFrom) || undefined + } + } }) return { findTrackedSession: (startTime) => { @@ -105,7 +130,10 @@ export function startRumSessionManager( return { id: session.id, sessionReplay: computeSessionReplayState(session.trackingType, session.hasError, session.isReplayForced), + eventsWithheld: computeEventsWithheld(session.trackingType, session.hasError, session.isReplayForced), + sampledOnError: withholdsEvents(session.trackingType), sampledOnErrorReplay: withholdsReplay(session.trackingType), + detailSampledFrom: session.detailSampledFrom, anonymousId: session.anonymousId, } }, @@ -113,11 +141,25 @@ export function startRumSessionManager( expireObservable: sessionManager.expireObservable, setForcedReplay: () => sessionManager.updateSessionState({ forcedReplay: '1' }), setSessionHasError: () => sessionManager.updateSessionState({ hasError: '1' }), + // Kept on the session rather than stamped on the released view events: the batch upserts views + // by id, so the next ordinary view update - which arrives within seconds - would replace the + // stamped one before the batch is ever sent. + setSessionDetailSampledFrom: (timestamp) => sessionManager.updateSessionState({ detailFrom: String(timestamp) }), } } export function withholdsReplay(trackingType: RumTrackingType) { - return trackingType === RumTrackingType.TRACKED_WITH_ERROR_SESSION_REPLAY + return ( + trackingType === RumTrackingType.TRACKED_WITH_ERROR_SESSION_REPLAY || + trackingType === RumTrackingType.TRACKED_ON_ERROR_WITH_SESSION_REPLAY + ) +} + +export function withholdsEvents(trackingType: RumTrackingType) { + return ( + trackingType === RumTrackingType.TRACKED_ON_ERROR_WITHOUT_SESSION_REPLAY || + trackingType === RumTrackingType.TRACKED_ON_ERROR_WITH_SESSION_REPLAY + ) } export function computeSessionReplayState( @@ -142,6 +184,19 @@ export function computeSessionReplayState( return SessionReplayState.OFF } +export function computeEventsWithheld( + trackingType: RumTrackingType, + hasError: boolean, + isReplayForced: boolean +): boolean { + // Forcing capture asks for this user's whole session, so it releases the events too - otherwise + // the forced replay would be uploaded for a session that does not exist yet. + if (hasError || isReplayForced) { + return false + } + return withholdsEvents(trackingType) +} + /** * Start a tracked replay session stub */ @@ -149,6 +204,8 @@ export function startRumSessionManagerStub(): RumSessionManager { const session: RumSession = { id: '00000000-aaaa-0000-aaaa-000000000000', sessionReplay: bridgeSupports(BridgeCapability.RECORDS) ? SessionReplayState.SAMPLED : SessionReplayState.OFF, + eventsWithheld: false, + sampledOnError: false, sampledOnErrorReplay: false, } return { @@ -157,6 +214,7 @@ export function startRumSessionManagerStub(): RumSessionManager { expireObservable: new Observable(), setForcedReplay: noop, setSessionHasError: noop, + setSessionDetailSampledFrom: noop, } } @@ -164,15 +222,26 @@ function computeSessionState(configuration: RumConfiguration, rawTrackingType?: let trackingType: RumTrackingType if (hasValidRumSession(rawTrackingType)) { trackingType = rawTrackingType - } else if (!performDraw(configuration.sessionSampleRate)) { - trackingType = RumTrackingType.NOT_TRACKED - } else if (performDraw(configuration.sessionReplaySampleRate)) { - trackingType = RumTrackingType.TRACKED_WITH_SESSION_REPLAY - } else if (performDraw(configuration.sessionReplayOnErrorSampleRate)) { - // Drawn only when the plain replay draw missed, so a session is never counted by both rates. - trackingType = RumTrackingType.TRACKED_WITH_ERROR_SESSION_REPLAY + } else if (performDraw(configuration.sessionSampleRate)) { + if (performDraw(configuration.sessionReplaySampleRate)) { + trackingType = RumTrackingType.TRACKED_WITH_SESSION_REPLAY + } else if (performDraw(configuration.sessionReplayOnErrorSampleRate)) { + // Drawn only when the plain replay draw missed, so a session is never counted by both rates. + trackingType = RumTrackingType.TRACKED_WITH_ERROR_SESSION_REPLAY + } else { + trackingType = RumTrackingType.TRACKED_WITHOUT_SESSION_REPLAY + } + } else if (performDraw(configuration.sessionOnErrorSampleRate)) { + // Drawn only when the plain session draw missed, so a session is never counted by both rates. + // Such a session never uploads its replay ahead of its events: whichever replay rate it draws, + // the replay is withheld alongside them, because until they are released the session does not + // exist yet and a replay sent then would have nothing to attach to. + trackingType = + performDraw(configuration.sessionReplaySampleRate) || performDraw(configuration.sessionReplayOnErrorSampleRate) + ? RumTrackingType.TRACKED_ON_ERROR_WITH_SESSION_REPLAY + : RumTrackingType.TRACKED_ON_ERROR_WITHOUT_SESSION_REPLAY } else { - trackingType = RumTrackingType.TRACKED_WITHOUT_SESSION_REPLAY + trackingType = RumTrackingType.NOT_TRACKED } return { trackingType, @@ -185,7 +254,9 @@ function hasValidRumSession(trackingType?: string): trackingType is RumTrackingT trackingType === RumTrackingType.NOT_TRACKED || trackingType === RumTrackingType.TRACKED_WITH_SESSION_REPLAY || trackingType === RumTrackingType.TRACKED_WITHOUT_SESSION_REPLAY || - trackingType === RumTrackingType.TRACKED_WITH_ERROR_SESSION_REPLAY + trackingType === RumTrackingType.TRACKED_WITH_ERROR_SESSION_REPLAY || + trackingType === RumTrackingType.TRACKED_ON_ERROR_WITHOUT_SESSION_REPLAY || + trackingType === RumTrackingType.TRACKED_ON_ERROR_WITH_SESSION_REPLAY ) } @@ -193,6 +264,8 @@ function isTypeTracked(rumSessionType: RumTrackingType | undefined) { return ( rumSessionType === RumTrackingType.TRACKED_WITHOUT_SESSION_REPLAY || rumSessionType === RumTrackingType.TRACKED_WITH_SESSION_REPLAY || - rumSessionType === RumTrackingType.TRACKED_WITH_ERROR_SESSION_REPLAY + rumSessionType === RumTrackingType.TRACKED_WITH_ERROR_SESSION_REPLAY || + rumSessionType === RumTrackingType.TRACKED_ON_ERROR_WITHOUT_SESSION_REPLAY || + rumSessionType === RumTrackingType.TRACKED_ON_ERROR_WITH_SESSION_REPLAY ) } diff --git a/packages/rum-core/src/transport/startRumBatch.ts b/packages/rum-core/src/transport/startRumBatch.ts index 3f7238ca65..aa03cf23d2 100644 --- a/packages/rum-core/src/transport/startRumBatch.ts +++ b/packages/rum-core/src/transport/startRumBatch.ts @@ -1,4 +1,11 @@ -import type { Context, TelemetryEvent, Observable, RawError, PageMayExitEvent, Encoder } from '@flashcatcloud/browser-core' +import type { + Context, + TelemetryEvent, + Observable, + RawError, + PageMayExitEvent, + Encoder, +} from '@flashcatcloud/browser-core' import { DeflateEncoderStreamId, combine, @@ -7,9 +14,9 @@ import { } from '@flashcatcloud/browser-core' import type { RumConfiguration } from '../domain/configuration' import type { LifeCycle } from '../domain/lifeCycle' -import { LifeCycleEventType } from '../domain/lifeCycle' +import type { RumSessionManager } from '../domain/rumSessionManager' import { RumEventType } from '../rawRumEvent.types' -import type { RumEvent } from '../rumEvent.types' +import { startWithheldEventBuffer } from './withheldEventBuffer' export function startRumBatch( configuration: RumConfiguration, @@ -17,7 +24,7 @@ export function startRumBatch( telemetryEventObservable: Observable, reportError: (error: RawError) => void, pageMayExitObservable: Observable, - sessionExpireObservable: Observable, + sessionManager: RumSessionManager, createEncoder: (streamId: DeflateEncoderStreamId) => Encoder ) { const replica = configuration.replica @@ -35,10 +42,12 @@ export function startRumBatch( }, reportError, pageMayExitObservable, - sessionExpireObservable + sessionManager.expireObservable ) - lifeCycle.subscribe(LifeCycleEventType.RUM_EVENT_COLLECTED, (serverRumEvent: RumEvent & Context) => { + // Events reach the batch through the buffer, which either forwards them straight away or withholds + // them until the session reports an error. A session that never errors uploads nothing at all. + const withheldEventBuffer = startWithheldEventBuffer(lifeCycle, sessionManager, (serverRumEvent) => { if (serverRumEvent.type === RumEventType.VIEW) { batch.upsert(serverRumEvent, serverRumEvent.view.id) } else { @@ -48,5 +57,13 @@ export function startRumBatch( telemetryEventObservable.subscribe((event) => batch.add(event, isTelemetryReplicationAllowed(configuration))) - return batch + return { + ...batch, + stop: () => { + // Stops the buffer too, so a release waiting on its jitter cannot fire into a batch that is + // no longer flushing. + withheldEventBuffer.stop() + batch.stop() + }, + } } diff --git a/packages/rum-core/src/transport/withheldEventBuffer.spec.ts b/packages/rum-core/src/transport/withheldEventBuffer.spec.ts new file mode 100644 index 0000000000..51ddc90a58 --- /dev/null +++ b/packages/rum-core/src/transport/withheldEventBuffer.spec.ts @@ -0,0 +1,324 @@ +import type { Context } from '@flashcatcloud/browser-core' +import { ONE_SECOND, PageExitReason } from '@flashcatcloud/browser-core' +import type { Clock } from '@flashcatcloud/browser-core/test' +import { mockClock, registerCleanupTask } from '@flashcatcloud/browser-core/test' +import { createRumSessionManagerMock } from '../../test' +import { RumEventType } from '../rawRumEvent.types' +import type { RumEvent } from '../rumEvent.types' +import { LifeCycle, LifeCycleEventType } from '../domain/lifeCycle' +import { + WITHHELD_BUFFER_DURATION, + WITHHELD_BUFFER_EVENTS_LIMIT, + WITHHELD_BUFFER_RELEASE_MAX_DELAY, + computeReleaseDelay, + startWithheldEventBuffer, +} from './withheldEventBuffer' + +describe('startWithheldEventBuffer', () => { + let clock: Clock + let lifeCycle: LifeCycle + let sessionManager: ReturnType + let forwarded: Array + + function collect(type: RumEventType, overrides: Context = {}) { + const event = { + type, + date: 1234, + view: { id: 'view-1' }, + session: {}, + ...(type === RumEventType.RESOURCE ? { resource: { status_code: 200 } } : {}), + ...overrides, + } as unknown as RumEvent & Context + lifeCycle.notify(LifeCycleEventType.RUM_EVENT_COLLECTED, event) + return event + } + + /** Everything the buffer released, once the release jitter has elapsed. */ + function releasedAfterJitter() { + clock.tick(WITHHELD_BUFFER_RELEASE_MAX_DELAY) + return forwarded + } + + beforeEach(() => { + clock = mockClock() + lifeCycle = new LifeCycle() + forwarded = [] + sessionManager = createRumSessionManagerMock().setTrackedOnError() + const { stop } = startWithheldEventBuffer(lifeCycle, sessionManager, (event) => forwarded.push(event)) + registerCleanupTask(() => { + stop() + clock.cleanup() + }) + }) + + it('forwards immediately when the session is not withholding', () => { + sessionManager.setTrackedWithSessionReplay() + + collect(RumEventType.VIEW) + collect(RumEventType.RESOURCE) + + expect(forwarded.length).toBe(2) + }) + + it('uploads nothing while the session has not reported an error', () => { + collect(RumEventType.VIEW) + collect(RumEventType.RESOURCE) + collect(RumEventType.ACTION) + clock.tick(30 * ONE_SECOND) + + expect(forwarded.length).toBe(0) + }) + + it('releases the buffer once the session reports an error', () => { + collect(RumEventType.VIEW) + collect(RumEventType.RESOURCE) + collect(RumEventType.ACTION) + + sessionManager.setSessionHasError() + collect(RumEventType.ERROR) + + const released = releasedAfterJitter() + expect(released.map((event) => event.type)).toEqual([ + RumEventType.VIEW, + RumEventType.RESOURCE, + RumEventType.ACTION, + RumEventType.ERROR, + ]) + }) + + it('marks how far back the released detail reaches', () => { + collect(RumEventType.VIEW) + collect(RumEventType.RESOURCE, { date: 4321 }) + + sessionManager.setSessionHasError() + collect(RumEventType.ERROR) + + const view = releasedAfterJitter().find((event) => event.type === RumEventType.VIEW)! + expect((view.session as Context).detail_sampled_from).toBe(4321) + }) + + it('keeps only the latest event of a view, since a view event supersedes the ones before it', () => { + collect(RumEventType.VIEW, { documentVersion: 1 }) + collect(RumEventType.VIEW, { documentVersion: 2 }) + collect(RumEventType.VIEW, { documentVersion: 3 }) + + sessionManager.setSessionHasError() + collect(RumEventType.ERROR) + + const views = releasedAfterJitter().filter((event) => event.type === RumEventType.VIEW) + expect(views.length).toBe(1) + expect((views[0] as unknown as Context).documentVersion).toBe(3) + }) + + it('drops detail that has aged out of the window', () => { + collect(RumEventType.VIEW) + collect(RumEventType.RESOURCE) + clock.tick(WITHHELD_BUFFER_DURATION + ONE_SECOND) + collect(RumEventType.ACTION) + + sessionManager.setSessionHasError() + collect(RumEventType.ERROR) + + const types = releasedAfterJitter().map((event) => event.type) + expect(types).not.toContain(RumEventType.RESOURCE) + expect(types).toContain(RumEventType.ACTION) + }) + + it('drops the buffer when the session expires without ever reporting an error', () => { + collect(RumEventType.VIEW) + collect(RumEventType.RESOURCE) + + sessionManager.setNotTracked() + collect(RumEventType.RESOURCE) + + expect(releasedAfterJitter().filter((event) => event.type === RumEventType.RESOURCE).length).toBe(1) + }) + + it('releases on page exit when the session errored without the buffer having noticed yet', () => { + // the event arrives synchronously, but the session state behind it is written through a lock + // that can defer the write - so the buffer can still read the session as withholding + collect(RumEventType.VIEW) + collect(RumEventType.RESOURCE) + sessionManager.setSessionHasError() + // no further event, so nothing re-reads the session before the page goes + + lifeCycle.notify(LifeCycleEventType.PAGE_MAY_EXIT, { reason: PageExitReason.UNLOADING }) + + expect(forwarded.map((event) => event.type)).toEqual([RumEventType.VIEW, RumEventType.RESOURCE]) + }) + + it('keeps the buffer when the page is only hidden, since it comes back', () => { + collect(RumEventType.VIEW) + collect(RumEventType.RESOURCE, { date: 111 }) + + lifeCycle.notify(LifeCycleEventType.PAGE_MAY_EXIT, { reason: PageExitReason.HIDDEN }) + expect(forwarded.length).toBe(0) + + sessionManager.setSessionHasError() + collect(RumEventType.ERROR) + + const dates = releasedAfterJitter().map((event) => event.date) + expect(dates).toContain(111) + }) + + it('records on the session how far back the released detail reaches', () => { + const spy = spyOn(sessionManager, 'setSessionDetailSampledFrom').and.callThrough() + collect(RumEventType.VIEW) + collect(RumEventType.RESOURCE, { date: 4321 }) + + sessionManager.setSessionHasError() + collect(RumEventType.ERROR) + releasedAfterJitter() + + // kept on the session, because the batch upserts views by id and the next ordinary view update + // would otherwise replace the stamped one before anything is sent + expect(spy).toHaveBeenCalledWith(4321) + }) + + it('drops the buffer when the session ends without ever having errored', () => { + collect(RumEventType.VIEW) + collect(RumEventType.RESOURCE) + + lifeCycle.notify(LifeCycleEventType.SESSION_EXPIRED) + + expect(releasedAfterJitter().length).toBe(0) + }) + + it('sends a release that is still waiting on jitter when the page is about to go', () => { + collect(RumEventType.VIEW) + collect(RumEventType.RESOURCE) + sessionManager.setSessionHasError() + collect(RumEventType.ERROR) + // still inside the jitter window: the error rides along with the buffer, so nothing left yet + expect(forwarded.length).toBe(0) + + lifeCycle.notify(LifeCycleEventType.PAGE_MAY_EXIT, { reason: PageExitReason.UNLOADING }) + + expect(forwarded.map((event) => event.type)).toEqual([RumEventType.VIEW, RumEventType.RESOURCE, RumEventType.ERROR]) + }) + + it('drops long tasks before actions when it runs out of room', () => { + collect(RumEventType.VIEW) + for (let i = 0; i < WITHHELD_BUFFER_EVENTS_LIMIT; i++) { + collect(RumEventType.LONG_TASK) + } + collect(RumEventType.ACTION) + + sessionManager.setSessionHasError() + collect(RumEventType.ERROR) + + expect(releasedAfterJitter().map((event) => event.type)).toContain(RumEventType.ACTION) + }) + + it('never drops errors, however full the buffer gets', () => { + collect(RumEventType.VIEW) + collect(RumEventType.ERROR, { date: 1 }) + for (let i = 0; i < WITHHELD_BUFFER_EVENTS_LIMIT * 2; i++) { + collect(RumEventType.LONG_TASK) + } + + sessionManager.setSessionHasError() + collect(RumEventType.ERROR, { date: 2 }) + + const errors = releasedAfterJitter().filter((event) => event.type === RumEventType.ERROR) + expect(errors.some((event) => event.date === 1)).toBeTrue() + }) + + it('gives up the newest error rather than the first one when only errors are left', () => { + collect(RumEventType.VIEW) + for (let i = 0; i < WITHHELD_BUFFER_EVENTS_LIMIT + 20; i++) { + collect(RumEventType.ERROR, { date: i }) + } + + sessionManager.setSessionHasError() + collect(RumEventType.ERROR, { date: 9999 }) + + const dates = releasedAfterJitter() + .filter((event) => event.type === RumEventType.ERROR) + .map((event) => event.date) + // the first error - the one the session is about - survives + expect(dates).toContain(0) + }) + + it('releases every detail alongside the view it hangs from', () => { + // the backend builds the session row out of view events, so a detail without its view would be + // unreachable however the view came to be missing + for (let i = 0; i < 60; i++) { + collect(RumEventType.VIEW, { view: { id: `view-${i}` } }) + collect(RumEventType.RESOURCE, { view: { id: `view-${i}` } }) + } + + sessionManager.setSessionHasError() + collect(RumEventType.ERROR, { view: { id: 'view-59' } }) + + const released = releasedAfterJitter() + const releasedViewIds = new Set( + released.filter((event) => event.type === RumEventType.VIEW).map((event) => event.view.id) + ) + released + .filter((event) => event.type !== RumEventType.VIEW) + .forEach((event) => expect(releasedViewIds.has(event.view.id)).toBeTrue()) + }) + + it('lets a view go once none of its detail is left inside the window', () => { + collect(RumEventType.VIEW, { view: { id: 'old-view' } }) + collect(RumEventType.RESOURCE, { view: { id: 'old-view' } }) + clock.tick(WITHHELD_BUFFER_DURATION + ONE_SECOND) + collect(RumEventType.VIEW, { view: { id: 'current-view' } }) + + sessionManager.setSessionHasError() + collect(RumEventType.ERROR, { view: { id: 'current-view' } }) + + const releasedViewIds = releasedAfterJitter() + .filter((event) => event.type === RumEventType.VIEW) + .map((event) => event.view.id) + expect(releasedViewIds).not.toContain('old-view') + expect(releasedViewIds).toContain('current-view') + }) +}) + +describe('computeReleaseDelay', () => { + function randomSessionId() { + const hex = '0123456789abcdef' + let id = '' + for (let i = 0; i < 36; i++) { + id += i === 8 || i === 13 || i === 18 || i === 23 ? '-' : hex[Math.floor(Math.random() * 16)] + } + return id + } + + it('is stable for a given session', () => { + const id = randomSessionId() + + expect(computeReleaseDelay(id)).toBe(computeReleaseDelay(id)) + }) + + it('stays within the release window', () => { + for (let i = 0; i < 1000; i++) { + const delay = computeReleaseDelay(randomSessionId()) + expect(delay).toBeGreaterThanOrEqual(0) + expect(delay).toBeLessThan(WITHHELD_BUFFER_RELEASE_MAX_DELAY) + } + }) + + it('spreads sessions across the window rather than bunching them up', () => { + // session ids are same-length strings over one small alphabet, so a running sum of their + // character codes lands nearly all of them within a few hundred ms of each other - which delays + // the herd instead of spreading it + const bucketCount = 10 + const buckets = new Array(bucketCount).fill(0) + const samples = 10000 + for (let i = 0; i < samples; i++) { + const bucket = Math.floor( + (computeReleaseDelay(randomSessionId()) / WITHHELD_BUFFER_RELEASE_MAX_DELAY) * bucketCount + ) + buckets[bucket] += 1 + } + + buckets.forEach((count) => { + // a flat spread puts 10% in each; allow a wide margin and still catch bunching + expect(count / samples).toBeGreaterThan(0.05) + expect(count / samples).toBeLessThan(0.2) + }) + }) +}) diff --git a/packages/rum-core/src/transport/withheldEventBuffer.ts b/packages/rum-core/src/transport/withheldEventBuffer.ts new file mode 100644 index 0000000000..d06e314ca0 --- /dev/null +++ b/packages/rum-core/src/transport/withheldEventBuffer.ts @@ -0,0 +1,320 @@ +import type { Context, RelativeTime, TimeoutId } from '@flashcatcloud/browser-core' +import { + ONE_KIBI_BYTE, + ONE_SECOND, + addTelemetryDebug, + clearTimeout, + computeBytesCount, + jsonStringify, + relativeNow, + setTimeout, +} from '@flashcatcloud/browser-core' +import type { LifeCycle } from '../domain/lifeCycle' +import { LifeCycleEventType } from '../domain/lifeCycle' +import type { RumSessionManager } from '../domain/rumSessionManager' +import { RumEventType } from '../rawRumEvent.types' +import type { RumEvent } from '../rumEvent.types' + +/** + * How much history a withheld buffer may span. Same number as the replay side, because it is the + * same promise to the customer: an error session shows the minute leading up to the error. + */ +export const WITHHELD_BUFFER_DURATION = 60 * ONE_SECOND + +/** Memory bound. Above it the least valuable events are dropped first, see {@link EvictionTier}. */ +const WITHHELD_BUFFER_BYTES_LIMIT = 64 * ONE_KIBI_BYTE +export const WITHHELD_BUFFER_EVENTS_LIMIT = 200 + +/** + * A view is the container its events hang from: the backend builds the session row out of view + * events, so a detail released without its view would be unreachable. Views are kept out of the + * eviction budget for that reason, and this only bounds pathological single-page navigation counts. + */ +const WITHHELD_BUFFER_VIEWS_LIMIT = 50 + +/** + * Correlated errors make every client release at the same instant, right when whatever caused them + * is already under strain. Releases are spread over this window instead. + */ +export const WITHHELD_BUFFER_RELEASE_MAX_DELAY = 3 * ONE_SECOND + +/** What gets dropped first when the buffer is over budget. Lower goes first. */ +const enum EvictionTier { + /** Long tasks, and requests that succeeded without complaint. */ + FIRST, + /** Actions and vitals: they explain what the user was doing. */ + LAST, + /** + * Errors are the reason the session is kept at all, so they go only once nothing else is left - + * and even then the newest goes first, because the earliest error is the one that releases the + * buffer and the one the session is about. + */ + LAST_RESORT, +} + +interface WithheldEvent { + event: RumEvent & Context + viewId: string + time: RelativeTime + bytes: number + tier: EvictionTier +} + +export function startWithheldEventBuffer( + lifeCycle: LifeCycle, + sessionManager: RumSessionManager, + forward: (event: RumEvent & Context) => void +) { + /** Latest event per view, in insertion order. */ + let views = new Map() + let details: WithheldEvent[] = [] + let bytes = 0 + let currentViewId: string | undefined + let withheldForSessionId: string | undefined + let releaseTimeoutId: TimeoutId | undefined + let droppedCount = 0 + + const eventSubscription = lifeCycle.subscribe(LifeCycleEventType.RUM_EVENT_COLLECTED, (event) => { + const session = sessionManager.findTrackedSession() + + if (session?.eventsWithheld) { + if (withheldForSessionId !== undefined && withheldForSessionId !== session.id) { + // A renewed session is a different session: it draws its own sampling and starts without an + // error, so what the previous one collected must not ride along. + clearBuffer() + } + withheldForSessionId = session.id + hold(event) + return + } + + if (withheldForSessionId !== undefined) { + if (session && session.id === withheldForSessionId) { + // The session just reported its error. This event - typically the error itself - joins what + // is held so that the whole history leaves in order, and behind the same jitter. + hold(event) + scheduleRelease() + return + } + // The session that was withholding is gone without ever reporting an error. + clearBuffer() + } + + forward(event) + }) + + /** + * Called when what is held may not get another chance to leave: the page is going away, or the + * session ended (which is also how a withdrawn tracking consent arrives here). + * + * `discardIfUnreleased` says whether the buffer has anything left to wait for. A session that + * ended is over, so what it never released goes no further. A page being hidden is not: it comes + * back, and dropping the minute it had collected would leave the error that follows with nothing. + */ + function settleBuffer(discardIfUnreleased: boolean) { + if (withheldForSessionId === undefined) { + return + } + // A release already scheduled goes out now rather than being lost to the jitter window. The + // session is also re-read, because it may have reported its error without the buffer noticing: + // the event arrives synchronously but the state behind it is written through a lock that can + // defer the write, and "an error, then the user leaves" is exactly what this feature is for. + const session = sessionManager.findTrackedSession() + const hasSinceErrored = !!session && session.id === withheldForSessionId && !session.eventsWithheld + + if (releaseTimeoutId !== undefined || hasSinceErrored) { + release() + } else if (discardIfUnreleased) { + clearBuffer() + } + } + + // Kept on a page exit: switching tabs raises one and the page comes straight back, while a page + // that is really unloading takes the buffer with it either way - so there is nothing to gain by + // dropping it, and a minute of history to lose. The replay side reasons the same way. + const pageMayExitSubscription = lifeCycle.subscribe(LifeCycleEventType.PAGE_MAY_EXIT, () => settleBuffer(false)) + const sessionExpireSubscription = lifeCycle.subscribe(LifeCycleEventType.SESSION_EXPIRED, () => settleBuffer(true)) + + function hold(event: RumEvent & Context) { + if (event.type === RumEventType.VIEW) { + // Upsert: a view event is cumulative, so the latest one supersedes the ones before it. This + // mirrors what the batch already does with view events. The delete is deliberate - setting an + // existing key leaves its insertion order untouched, so without it the oldest entry would be + // the first view seen rather than the least recently updated one. + views.delete(event.view.id) + views.set(event.view.id, event) + currentViewId = event.view.id + while (views.size > WITHHELD_BUFFER_VIEWS_LIMIT) { + views.delete(views.keys().next().value!) + } + prune() + return + } + + const held: WithheldEvent = { + event, + viewId: event.view.id, + time: relativeNow(), + bytes: computeBytesCount(jsonStringify(event) ?? ''), + tier: getEvictionTier(event), + } + details.push(held) + bytes += held.bytes + + prune() + while (details.length > WITHHELD_BUFFER_EVENTS_LIMIT || bytes > WITHHELD_BUFFER_BYTES_LIMIT) { + if (!evictOne()) { + break + } + } + } + + /** Drops what has aged out of the window, so the span kept is the one we promise. */ + function prune() { + const oldestAllowed = (relativeNow() - WITHHELD_BUFFER_DURATION) as RelativeTime + let cutoff = 0 + while (cutoff < details.length && details[cutoff].time < oldestAllowed) { + bytes -= details[cutoff].bytes + droppedCount += 1 + cutoff += 1 + } + if (cutoff > 0) { + details = details.slice(cutoff) + } + + // A view is kept as the container of the detail hanging from it, so once none of its detail is + // left inside the window it has nothing left to contain. Without this the map would grow with + // every route change for as long as the page lives, holding more than the detail budget itself. + // The view in progress always stays: it is the container the error will hang from. + const viewsWithDetail = new Set(details.map((held) => held.viewId)) + views.forEach((_, viewId) => { + if (viewId !== currentViewId && !viewsWithDetail.has(viewId)) { + views.delete(viewId) + } + }) + } + + /** Removes one event of the least valuable tier present. Returns false when there is none left. */ + function evictOne() { + for (const tier of [EvictionTier.FIRST, EvictionTier.LAST]) { + const index = details.findIndex((held) => held.tier === tier) + if (index !== -1) { + evictAt(index) + return true + } + } + + // Only errors are left. One still has to go to stay within budget, and it is the newest: an + // error storm would otherwise push out the first error, which is the one that released the + // buffer and the one the session is really about. + for (let index = details.length - 1; index >= 0; index -= 1) { + if (details[index].tier === EvictionTier.LAST_RESORT) { + evictAt(index) + return true + } + } + return false + } + + function evictAt(index: number) { + bytes -= details[index].bytes + droppedCount += 1 + details.splice(index, 1) + } + + function scheduleRelease() { + if (releaseTimeoutId !== undefined) { + return + } + releaseTimeoutId = setTimeout(release, computeReleaseDelay(withheldForSessionId!)) + } + + function release() { + prune() + + // A detail whose view is gone has no container to hang from, so it would be unreachable. + const releasable = details.filter((held) => views.has(held.viewId)) + const detailSampledFrom = releasable.length > 0 ? releasable[0].event.date : undefined + + if (detailSampledFrom !== undefined) { + // Recorded on the session so that every view update from here on carries it - the batch + // upserts views by id, so the next ordinary update would otherwise replace these ones before + // the batch is ever sent. These were assembled too early to pick it up, so they are given the + // same value directly, which is what the backend sees if the page goes before the next update. + sessionManager.setSessionDetailSampledFrom(detailSampledFrom) + } + + views.forEach((view) => { + if (detailSampledFrom !== undefined) { + view.session.detail_sampled_from = detailSampledFrom + } + forward(view) + }) + releasable.forEach((held) => forward(held.event)) + + addTelemetryDebug('Error session event buffer released', { + 'buffer.views_count': views.size, + 'buffer.events_count': releasable.length, + 'buffer.dropped_count': droppedCount, + 'buffer.bytes': bytes, + }) + + clearBuffer() + } + + /** Empties the buffer, whether it was just released or is being thrown away. */ + function clearBuffer() { + clearTimeout(releaseTimeoutId) + releaseTimeoutId = undefined + views = new Map() + details = [] + bytes = 0 + droppedCount = 0 + currentViewId = undefined + withheldForSessionId = undefined + } + + return { + stop: () => { + clearBuffer() + eventSubscription.unsubscribe() + pageMayExitSubscription.unsubscribe() + sessionExpireSubscription.unsubscribe() + }, + } +} + +function getEvictionTier(event: RumEvent): EvictionTier { + switch (event.type) { + case RumEventType.ERROR: + return EvictionTier.LAST_RESORT + case RumEventType.LONG_TASK: + return EvictionTier.FIRST + case RumEventType.RESOURCE: { + // A request that failed is part of how the error happened; one that succeeded rarely is. + // -1 stands for an unknown status code, which is treated like an ordinary success + const statusCode = event.resource?.status_code ?? -1 + return statusCode === 0 || statusCode >= 400 ? EvictionTier.LAST : EvictionTier.FIRST + } + default: + return EvictionTier.LAST + } +} + +/** Keeps the running hash inside the range `Math.imul` is exact over. */ +const LARGEST_INT32_PRIME = 2147483647 + +/** + * Deterministic per session, so a client always spreads to the same offset. + * + * Multiplicative rather than a running sum: session ids are same-length strings drawn from the same + * small alphabet, so summing their character codes lands almost every session within a few hundred + * milliseconds of the same value - which delays the herd instead of spreading it. + */ +export function computeReleaseDelay(sessionId: string) { + let hash = 0 + for (let i = 0; i < sessionId.length; i += 1) { + hash = (Math.imul(hash, 31) + sessionId.charCodeAt(i)) % LARGEST_INT32_PRIME + } + return Math.abs(hash) % WITHHELD_BUFFER_RELEASE_MAX_DELAY +} diff --git a/packages/rum-core/test/mockRumSessionManager.ts b/packages/rum-core/test/mockRumSessionManager.ts index a97a96fba0..0ce90f5175 100644 --- a/packages/rum-core/test/mockRumSessionManager.ts +++ b/packages/rum-core/test/mockRumSessionManager.ts @@ -1,7 +1,9 @@ import { Observable } from '@flashcatcloud/browser-core' import { RumTrackingType, + computeEventsWithheld, computeSessionReplayState, + withholdsEvents, withholdsReplay, type RumSessionManager, } from '../src/domain/rumSessionManager' @@ -12,8 +14,10 @@ export interface RumSessionManagerMock extends RumSessionManager { setTrackedWithoutSessionReplay(): RumSessionManagerMock setTrackedWithSessionReplay(): RumSessionManagerMock setTrackedWithErrorSessionReplay(): RumSessionManagerMock + setTrackedOnError(): RumSessionManagerMock setForcedReplay(): RumSessionManagerMock setSessionHasError(): RumSessionManagerMock + setSessionDetailSampledFrom(timestamp: number): RumSessionManagerMock } const DEFAULT_ID = 'session-id' @@ -21,6 +25,7 @@ const enum SessionStatus { TRACKED_WITH_SESSION_REPLAY, TRACKED_WITHOUT_SESSION_REPLAY, TRACKED_WITH_ERROR_SESSION_REPLAY, + TRACKED_ON_ERROR, NOT_TRACKED, EXPIRED, } @@ -29,6 +34,7 @@ const TRACKING_TYPES: { [key in SessionStatus]?: RumTrackingType } = { [SessionStatus.TRACKED_WITH_SESSION_REPLAY]: RumTrackingType.TRACKED_WITH_SESSION_REPLAY, [SessionStatus.TRACKED_WITHOUT_SESSION_REPLAY]: RumTrackingType.TRACKED_WITHOUT_SESSION_REPLAY, [SessionStatus.TRACKED_WITH_ERROR_SESSION_REPLAY]: RumTrackingType.TRACKED_WITH_ERROR_SESSION_REPLAY, + [SessionStatus.TRACKED_ON_ERROR]: RumTrackingType.TRACKED_ON_ERROR_WITH_SESSION_REPLAY, } export function createRumSessionManagerMock(): RumSessionManagerMock { @@ -36,6 +42,7 @@ export function createRumSessionManagerMock(): RumSessionManagerMock { let sessionStatus: SessionStatus = SessionStatus.TRACKED_WITH_SESSION_REPLAY let forcedReplay: boolean = false let hasError: boolean = false + let detailSampledFrom: number | undefined return { findTrackedSession() { const trackingType = TRACKING_TYPES[sessionStatus] @@ -46,6 +53,9 @@ export function createRumSessionManagerMock(): RumSessionManagerMock { id, // Derived the same way as in production, so the mock cannot drift from the real state machine sessionReplay: computeSessionReplayState(trackingType, hasError, forcedReplay), + eventsWithheld: computeEventsWithheld(trackingType, hasError, forcedReplay), + sampledOnError: withholdsEvents(trackingType), + detailSampledFrom, sampledOnErrorReplay: withholdsReplay(trackingType), anonymousId: 'device-123', } @@ -75,6 +85,10 @@ export function createRumSessionManagerMock(): RumSessionManagerMock { sessionStatus = SessionStatus.TRACKED_WITH_ERROR_SESSION_REPLAY return this }, + setTrackedOnError() { + sessionStatus = SessionStatus.TRACKED_ON_ERROR + return this + }, setForcedReplay() { forcedReplay = true return this @@ -83,5 +97,9 @@ export function createRumSessionManagerMock(): RumSessionManagerMock { hasError = true return this }, + setSessionDetailSampledFrom(timestamp) { + detailSampledFrom = timestamp + return this + }, } }