Skip to content

Commit 224cdb5

Browse files
bjoergebjoerge
authoredMar 17, 2025··
feat(studio): detect and report high listener roundtrip latency (#8943)
·
v3.80.1v3.80.0
1 parent 7613d72 commit 224cdb5

16 files changed

+218
-50
lines changed
 

‎packages/sanity/src/core/store/_legacy/datastores.ts

+23-2
Original file line numberDiff line numberDiff line change
@@ -16,8 +16,9 @@ import {
1616
type ConnectionStatusStore,
1717
createConnectionStatusStore,
1818
} from './connection-status/connection-status-store'
19-
import {createDocumentStore, type DocumentStore} from './document'
19+
import {createDocumentStore, type DocumentStore, type LatencyReportEvent} from './document'
2020
import {DocumentDesynced} from './document/__telemetry__/documentOutOfSyncEvents.telemetry'
21+
import {HighListenerLatencyOccurred} from './document/__telemetry__/listenerLatency.telemetry'
2122
import {fetchFeatureToggle} from './document/document-pair/utils/fetchFeatureToggle'
2223
import {type OutOfSyncError} from './document/utils/sequentializeListenerEvents'
2324
import {createGrantsStore, type GrantsStore} from './grants'
@@ -27,6 +28,11 @@ import {createProjectStore, type ProjectStore} from './project'
2728
import {useResourceCache} from './ResourceCacheProvider'
2829
import {createUserStore, type UserStore} from './user'
2930

31+
/**
32+
* Latencies below this value will not be logged
33+
*/
34+
const IGNORE_LATENCY_BELOW_MS = 1000
35+
3036
/**
3137
* @hidden
3238
* @beta */
@@ -155,6 +161,19 @@ export function useDocumentStore(): DocumentStore {
155161
[telemetry],
156162
)
157163

164+
const handleReportLatency = useCallback(
165+
(event: LatencyReportEvent) => {
166+
if (event.latencyMs > IGNORE_LATENCY_BELOW_MS) {
167+
telemetry.log(HighListenerLatencyOccurred, {
168+
latency: event.latencyMs,
169+
shard: event.shard,
170+
transactionId: event.transactionId,
171+
})
172+
}
173+
},
174+
[telemetry],
175+
)
176+
158177
return useMemo(() => {
159178
const documentStore =
160179
resourceCache.get<DocumentStore>({
@@ -169,7 +188,8 @@ export function useDocumentStore(): DocumentStore {
169188
schema,
170189
i18n,
171190
serverActionsEnabled,
172-
pairListenerOptions: {
191+
extraOptions: {
192+
onReportLatency: handleReportLatency,
173193
onSyncErrorRecovery: handleSyncErrorRecovery,
174194
},
175195
})
@@ -191,6 +211,7 @@ export function useDocumentStore(): DocumentStore {
191211
workspace,
192212
templates,
193213
serverActionsEnabled,
214+
handleReportLatency,
194215
handleSyncErrorRecovery,
195216
])
196217
}

‎packages/sanity/src/core/store/_legacy/document/__telemetry__/listenerLatency.telemetry.ts

+12
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
import {defineEvent} from '@sanity/telemetry'
2+
3+
type Samples = {
4+
latency: number
5+
transactionId: string
6+
shard?: string
7+
}
8+
export const HighListenerLatencyOccurred = defineEvent<Samples>({
9+
name: 'High Listener Latency Detected',
10+
version: 1,
11+
description: 'Emits when a high listener latency is detected',
12+
})

‎packages/sanity/src/core/store/_legacy/document/document-pair/checkoutPair.ts

+132-11
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,14 @@
1-
import {type Action, type SanityClient} from '@sanity/client'
1+
import {
2+
type Action,
3+
type MultipleActionResult,
4+
type SanityClient,
5+
type SingleMutationResult,
6+
} from '@sanity/client'
27
import {type Mutation} from '@sanity/mutator'
38
import {type SanityDocument} from '@sanity/types'
49
import {omit} from 'lodash'
5-
import {EMPTY, from, merge, type Observable} from 'rxjs'
6-
import {filter, map, mergeMap, share, take, tap} from 'rxjs/operators'
10+
import {defer, EMPTY, from, merge, type Observable} from 'rxjs'
11+
import {filter, map, mergeMap, scan, share, take, tap, withLatestFrom} from 'rxjs/operators'
712

813
import {type DocumentVariantType} from '../../../../util/getDocumentVariantType'
914
import {
@@ -13,11 +18,19 @@ import {
1318
type MutationPayload,
1419
type RemoteSnapshotEvent,
1520
} from '../buffered-doc'
16-
import {getPairListener, type ListenerEvent, type PairListenerOptions} from '../getPairListener'
21+
import {
22+
type DocumentStoreExtraOptions,
23+
getPairListener,
24+
type LatencyReportEvent,
25+
type ListenerEvent,
26+
} from '../getPairListener'
1727
import {type IdPair, type PendingMutationsEvent, type ReconnectEvent} from '../types'
1828
import {actionsApiClient} from './utils/actionsApiClient'
1929
import {operationsApiClient} from './utils/operationsApiClient'
2030

31+
/** Timeout on request that fetches shard name before reporting latency */
32+
const FETCH_SHARD_TIMEOUT = 20_000
33+
2134
const isMutationEventForDocId =
2235
(id: string) =>
2336
(
@@ -41,6 +54,13 @@ export type DocumentVersionEvent = WithVersion<ReconnectEvent | BufferedDocument
4154
* @beta */
4255
export type RemoteSnapshotVersionEvent = WithVersion<RemoteSnapshotEvent>
4356

57+
/**
58+
* @hidden
59+
* @beta
60+
* The SingleMutationResult type from `@sanity/client` doesn't reflect what we get back from POST /mutate
61+
*/
62+
export type MutationResult = Omit<SingleMutationResult, 'documentId'>
63+
4464
/**
4565
* @hidden
4666
* @beta */
@@ -156,7 +176,11 @@ function commitActions(client: SanityClient, idPair: IdPair, mutationParams: Mut
156176
})
157177
}
158178

159-
function commitMutations(client: SanityClient, idPair: IdPair, mutationParams: Mutation['params']) {
179+
function commitMutations(
180+
client: SanityClient,
181+
idPair: IdPair,
182+
mutationParams: Mutation['params'],
183+
): Promise<MutationResult> {
160184
const {resultRev, ...mutation} = mutationParams
161185
return operationsApiClient(client, idPair).dataRequest('mutate', mutation, {
162186
visibility: 'async',
@@ -173,7 +197,7 @@ function submitCommitRequest(
173197
idPair: IdPair,
174198
request: CommitRequest,
175199
serverActionsEnabled: boolean,
176-
) {
200+
): Observable<MultipleActionResult | MutationResult> {
177201
return from(
178202
serverActionsEnabled
179203
? commitActions(client, idPair, request.mutation.params)
@@ -197,16 +221,30 @@ function submitCommitRequest(
197221
)
198222
}
199223

224+
type LatencyTrackingEvent = {
225+
transactionId: string
226+
submittedAt: Date
227+
receivedAt: Date
228+
deltaMs: number
229+
}
230+
231+
type LatencyTrackingState = {
232+
pending: {type: 'receive' | 'submit'; transactionId: string; timestamp: Date}[]
233+
event: LatencyTrackingEvent | undefined
234+
}
235+
200236
/** @internal */
201237
export function checkoutPair(
202238
client: SanityClient,
203239
idPair: IdPair,
204240
serverActionsEnabled: Observable<boolean>,
205-
pairListenerOptions?: PairListenerOptions,
241+
options: DocumentStoreExtraOptions = {},
206242
): Pair {
207243
const {publishedId, draftId, versionId} = idPair
208244

209-
const listenerEvents$ = getPairListener(client, idPair, pairListenerOptions).pipe(share())
245+
const {onReportLatency, onSyncErrorRecovery, tag} = options
246+
247+
const listenerEvents$ = getPairListener(client, idPair, {onSyncErrorRecovery, tag}).pipe(share())
210248

211249
const reconnect$ = listenerEvents$.pipe(
212250
filter((ev) => ev.type === 'reconnect'),
@@ -248,6 +286,19 @@ export function checkoutPair(
248286
),
249287
),
250288
),
289+
)
290+
291+
// Note: we're only subscribing to this for the side-effect
292+
const combinedEvents = defer(() =>
293+
onReportLatency
294+
? reportLatency({
295+
commits$: commits$,
296+
listenerEvents$: listenerEvents$,
297+
client,
298+
onReportLatency,
299+
})
300+
: merge(commits$, listenerEvents$),
301+
).pipe(
251302
mergeMap(() => EMPTY),
252303
share(),
253304
)
@@ -256,22 +307,92 @@ export function checkoutPair(
256307
transactionsPendingEvents$,
257308
draft: {
258309
...draft,
259-
events: merge(commits$, reconnect$, draft.events).pipe(map(setVersion('draft'))),
310+
events: merge(combinedEvents, reconnect$, draft.events).pipe(map(setVersion('draft'))),
260311
remoteSnapshot$: draft.remoteSnapshot$.pipe(map(setVersion('draft'))),
261312
},
262313
...(typeof version === 'undefined'
263314
? {}
264315
: {
265316
version: {
266317
...version,
267-
events: merge(commits$, reconnect$, version.events).pipe(map(setVersion('version'))),
318+
events: merge(combinedEvents, reconnect$, version.events).pipe(
319+
map(setVersion('version')),
320+
),
268321
remoteSnapshot$: version.remoteSnapshot$.pipe(map(setVersion('version'))),
269322
},
270323
}),
271324
published: {
272325
...published,
273-
events: merge(commits$, reconnect$, published.events).pipe(map(setVersion('published'))),
326+
events: merge(combinedEvents, reconnect$, published.events).pipe(
327+
map(setVersion('published')),
328+
),
274329
remoteSnapshot$: published.remoteSnapshot$.pipe(map(setVersion('published'))),
275330
},
276331
}
277332
}
333+
334+
function reportLatency(options: {
335+
commits$: Observable<MultipleActionResult | MutationResult>
336+
listenerEvents$: Observable<ListenerEvent>
337+
client: SanityClient
338+
onReportLatency: (event: LatencyReportEvent) => void
339+
}) {
340+
const {client, commits$, listenerEvents$, onReportLatency} = options
341+
// Note: this request happens once and the result is then cached indefinitely
342+
const shardInfo = fetch(client.getUrl(client.getDataUrl('ping')), {
343+
signal: AbortSignal.timeout(FETCH_SHARD_TIMEOUT),
344+
})
345+
.then((response) => response.headers.get('X-Sanity-Shard') || undefined)
346+
.catch(() => undefined)
347+
348+
const submittedMutations = commits$.pipe(
349+
map((ev) => ({
350+
type: 'submit' as const,
351+
transactionId: ev.transactionId,
352+
timestamp: new Date(),
353+
})),
354+
share(),
355+
)
356+
357+
const receivedMutations = listenerEvents$.pipe(
358+
filter((ev) => ev.type === 'mutation'),
359+
map((ev) => ({
360+
type: 'receive' as const,
361+
transactionId: ev.transactionId,
362+
timestamp: new Date(),
363+
})),
364+
share(),
365+
)
366+
367+
return merge(submittedMutations, receivedMutations).pipe(
368+
scan(
369+
(state: LatencyTrackingState, event): LatencyTrackingState => {
370+
const matchingIndex = state.pending.findIndex(
371+
(e) => e.transactionId === event.transactionId,
372+
)
373+
if (matchingIndex > -1) {
374+
const matching = state.pending[matchingIndex]
375+
const [submitEvent, receiveEvent] =
376+
matching.type == 'submit' ? [matching, event] : [event, matching]
377+
return {
378+
event: {
379+
transactionId: event.transactionId,
380+
submittedAt: submitEvent.timestamp,
381+
receivedAt: submitEvent.timestamp,
382+
deltaMs: receiveEvent.timestamp.getTime() - submitEvent.timestamp.getTime(),
383+
},
384+
pending: state.pending.toSpliced(matchingIndex, 1),
385+
}
386+
}
387+
return {event: undefined, pending: state.pending.concat(event)}
388+
},
389+
{event: undefined, pending: []},
390+
),
391+
map((state) => state.event),
392+
filter((event) => !!event),
393+
withLatestFrom(shardInfo),
394+
tap(([event, shard]) =>
395+
onReportLatency?.({latencyMs: event.deltaMs, shard, transactionId: event.transactionId}),
396+
),
397+
)
398+
}

‎packages/sanity/src/core/store/_legacy/document/document-pair/consistencyStatus.ts

+4-4
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@ import {type SanityClient} from '@sanity/client'
22
import {combineLatest, type Observable} from 'rxjs'
33
import {distinctUntilChanged, map, publishReplay, refCount, switchMap} from 'rxjs/operators'
44

5-
import {type PairListenerOptions} from '../getPairListener'
5+
import {type DocumentStoreExtraOptions} from '../getPairListener'
66
import {type IdPair} from '../types'
77
import {memoize} from '../utils/createMemoizer'
88
import {memoizedPair} from './memoizedPair'
@@ -15,16 +15,16 @@ export const consistencyStatus: (
1515
idPair: IdPair,
1616
typeName: string,
1717
serverActionsEnabled: Observable<boolean>,
18-
pairListenerOptions?: PairListenerOptions,
18+
extraOptions?: DocumentStoreExtraOptions,
1919
) => Observable<boolean> = memoize(
2020
(
2121
client: SanityClient,
2222
idPair: IdPair,
2323
typeName: string,
2424
serverActionsEnabled: Observable<boolean>,
25-
pairListenerOptions?: PairListenerOptions,
25+
extraOptions?: DocumentStoreExtraOptions,
2626
) => {
27-
return memoizedPair(client, idPair, typeName, serverActionsEnabled, pairListenerOptions).pipe(
27+
return memoizedPair(client, idPair, typeName, serverActionsEnabled, extraOptions).pipe(
2828
switchMap(({draft, published}) =>
2929
combineLatest([draft.consistency$, published.consistency$]),
3030
),

‎packages/sanity/src/core/store/_legacy/document/document-pair/documentEvents.ts

+2-2
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@ import {type SanityClient} from '@sanity/client'
22
import {merge, type Observable} from 'rxjs'
33
import {switchMap} from 'rxjs/operators'
44

5-
import {type PairListenerOptions} from '../getPairListener'
5+
import {type DocumentStoreExtraOptions} from '../getPairListener'
66
import {type IdPair} from '../types'
77
import {memoize} from '../utils/createMemoizer'
88
import {type DocumentVersionEvent} from './checkoutPair'
@@ -17,7 +17,7 @@ export const documentEvents = memoize(
1717
idPair: IdPair,
1818
typeName: string,
1919
serverActionsEnabled: Observable<boolean>,
20-
pairListenerOptions?: PairListenerOptions,
20+
pairListenerOptions?: DocumentStoreExtraOptions,
2121
): Observable<DocumentVersionEvent> => {
2222
return memoizedPair(client, idPair, typeName, serverActionsEnabled, pairListenerOptions).pipe(
2323
switchMap(({draft, published}) => merge(draft.events, published.events)),

‎packages/sanity/src/core/store/_legacy/document/document-pair/editOperations.ts

+2-2
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ import {concat, EMPTY, merge, type Observable, of} from 'rxjs'
44
import {map, mergeMap, shareReplay} from 'rxjs/operators'
55

66
import {type HistoryStore} from '../../history'
7-
import {type PairListenerOptions} from '../getPairListener'
7+
import {type DocumentStoreExtraOptions} from '../getPairListener'
88
import {type IdPair} from '../types'
99
import {memoize} from '../utils/createMemoizer'
1010
import {memoizeKeyGen} from './memoizeKeyGen'
@@ -20,7 +20,7 @@ export const editOperations = memoize(
2020
historyStore: HistoryStore
2121
schema: Schema
2222
serverActionsEnabled: Observable<boolean>
23-
pairListenerOptions?: PairListenerOptions
23+
pairListenerOptions?: DocumentStoreExtraOptions
2424
},
2525
idPair: IdPair,
2626
typeName: string,

‎packages/sanity/src/core/store/_legacy/document/document-pair/editState.ts

+3-3
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ import {map, publishReplay, refCount, startWith, switchMap} from 'rxjs/operators
55

66
import {getVersionFromId} from '../../../../util'
77
import {createSWR} from '../../../../util/rxSwr'
8-
import {type PairListenerOptions} from '../getPairListener'
8+
import {type DocumentStoreExtraOptions} from '../getPairListener'
99
import {type IdPair, type PendingMutationsEvent} from '../types'
1010
import {memoize} from '../utils/createMemoizer'
1111
import {memoizeKeyGen} from './memoizeKeyGen'
@@ -57,7 +57,7 @@ export const editState = memoize(
5757
client: SanityClient
5858
schema: Schema
5959
serverActionsEnabled: Observable<boolean>
60-
pairListenerOptions?: PairListenerOptions
60+
extraOptions?: DocumentStoreExtraOptions
6161
},
6262
idPair: IdPair,
6363
typeName: string,
@@ -70,7 +70,7 @@ export const editState = memoize(
7070
idPair,
7171
typeName,
7272
ctx.serverActionsEnabled,
73-
ctx.pairListenerOptions,
73+
ctx.extraOptions,
7474
).pipe(
7575
switchMap((versions) =>
7676
combineLatest([

0 commit comments

Comments
 (0)
Please sign in to comment.