-
Notifications
You must be signed in to change notification settings - Fork 20
/
defaultRetry.ts
290 lines (255 loc) · 10.2 KB
/
defaultRetry.ts
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
import { Event, Options, Transport, Payload, PayloadOptions, Status, Response, RetryClass } from '@amplitude/types';
import { setupDefaultTransport } from '../transports';
import { DEFAULT_OPTIONS, BASE_RETRY_TIMEOUT_DEPRECATED, BASE_RETRY_TIMEOUT_DEPRECATED_WARNING } from '../constants';
import { asyncSleep, collectInvalidEventIndices, logger } from '@amplitude/utils';
interface RetryMetadata {
shouldRetry: boolean;
shouldReduceEventCount: boolean;
eventIndicesToRemove: number[];
}
/**
* Converts deprecated maxRetries option to retryTimeouts
*/
function convertMaxRetries(maxRetries: number): number[] {
const retryTimeouts = [];
let currentTimeout = BASE_RETRY_TIMEOUT_DEPRECATED;
for (let i = 0; i < maxRetries; i++) {
retryTimeouts.push(currentTimeout);
currentTimeout *= 2;
}
return retryTimeouts;
}
export class RetryHandler implements RetryClass {
protected readonly _apiKey: string;
// A map of maps to event buffers for failed events
// The first key is userId (or ''), and second is deviceId (or '')
private readonly _idToBuffer: Map<string, Map<string, Event[]>> = new Map<string, Map<string, Event[]>>();
protected readonly _options: Options;
private readonly _transport: Transport;
private _eventsInRetry = 0;
public constructor(apiKey: string, options: Partial<Options> = {}) {
this._apiKey = apiKey;
this._options = { ...DEFAULT_OPTIONS, ...options };
this._transport = this._options.transportClass ?? setupDefaultTransport(this._options);
if (this._options.maxRetries !== undefined) {
logger.warn(BASE_RETRY_TIMEOUT_DEPRECATED_WARNING);
this._options.retryTimeouts = convertMaxRetries(this._options.maxRetries);
delete this._options.maxRetries;
}
}
/**
* @inheritDoc
*/
public async sendEventsWithRetry(events: readonly Event[]): Promise<Response> {
let response: Response = { status: Status.Unknown, statusCode: 0 };
const eventsToSend = this._pruneEvents(events);
try {
response = await this._transport.sendPayload(this._getPayload(eventsToSend));
if (response.status !== Status.Success) {
throw new Error(response.status);
}
} catch {
if (this._shouldRetryEvents()) {
this._onEventsError(events, response);
}
}
return response;
}
private _shouldRetryEvents(): boolean {
if (this._options.retryTimeouts.length === 0) {
return false;
}
// TODO: Refine logic of what happens when we reach the queue limit.
return this._eventsInRetry < this._options.maxCachedEvents;
}
// Sends events with ids currently in active retry buffers straight
// to the retry buffer they should be in
private _pruneEvents(events: readonly Event[]): Event[] {
const prunedEvents: Event[] = [];
events.forEach(event => {
const { user_id: userId = '', device_id: deviceId = '' } = event;
// We can ignore events with neither. They would fail anyways when sent as event.
if (userId.length > 0 || deviceId.length > 0) {
const retryBuffer = this._getRetryBuffer(userId, deviceId);
if (retryBuffer !== null) {
retryBuffer.push(event);
this._eventsInRetry++;
} else {
prunedEvents.push(event);
}
}
});
return prunedEvents;
}
protected _getPayloadOptions(): { options?: PayloadOptions } {
if (typeof this._options.minIdLength === 'number') {
return {
options: {
min_id_length: this._options.minIdLength,
},
};
}
return {};
}
protected _getPayload(events: readonly Event[]): Payload {
return {
api_key: this._apiKey,
events,
...this._getPayloadOptions(),
};
}
private _getRetryBuffer(userId: string, deviceId: string): Event[] | null {
const deviceToBufferMap = this._idToBuffer.get(userId);
if (deviceToBufferMap === undefined) {
return null;
}
return deviceToBufferMap.get(deviceId) ?? null;
}
// cleans up the id to buffer map if the job is done
private _cleanUpBuffer(userId: string, deviceId: string): void {
const deviceToBufferMap = this._idToBuffer.get(userId);
if (deviceToBufferMap === undefined) {
return;
}
const eventsToRetry = deviceToBufferMap.get(deviceId);
if (eventsToRetry !== undefined && eventsToRetry.length === 0) {
deviceToBufferMap.delete(deviceId);
}
if (deviceToBufferMap.size === 0) {
this._idToBuffer.delete(userId);
}
}
private _onEventsError(events: readonly Event[], response: Response): void {
let eventsToRetry: readonly Event[] = events;
// See if there are any events we can immediately throw out
if (response.status === Status.RateLimit && response.body !== undefined) {
const { exceededDailyQuotaUsers, exceededDailyQuotaDevices } = response.body;
eventsToRetry = events.filter(({ user_id: userId, device_id: deviceId }) => {
return (
!(userId !== undefined && userId in exceededDailyQuotaUsers) &&
!(deviceId !== undefined && deviceId in exceededDailyQuotaDevices)
);
});
} else if (response.status === Status.Invalid) {
if (typeof response.body?.missingField === 'string' || events.length === 1) {
// Return early if there's an issue with the entire payload
// or if there's only one event and its invalid
return;
} else if (response.body !== undefined) {
const invalidEventIndices = new Set<number>(collectInvalidEventIndices(response));
eventsToRetry = events.filter((_, index) => !invalidEventIndices.has(index));
}
} else if (response.status === Status.Success) {
// In case _onEventsError was called when we were actually successful
// In which case, why even start retrying events?
return;
}
eventsToRetry.forEach((event: Event) => {
const { user_id: userId = '', device_id: deviceId = '' } = event;
if (userId.length > 0 || deviceId.length > 0) {
let deviceToBufferMap = this._idToBuffer.get(userId);
if (deviceToBufferMap === undefined) {
deviceToBufferMap = new Map<string, Event[]>();
this._idToBuffer.set(userId, deviceToBufferMap);
}
let retryBuffer = deviceToBufferMap.get(deviceId);
if (retryBuffer === undefined) {
retryBuffer = [];
deviceToBufferMap.set(deviceId, retryBuffer);
// In the next event loop, start retrying these events
setTimeout(() => {
// eslint-disable-next-line @typescript-eslint/no-floating-promises
this._retryEventsOnLoop(userId, deviceId);
}, 0);
}
this._eventsInRetry++;
retryBuffer.push(event);
}
});
}
private async _retryEventsOnce(
userId: string,
deviceId: string,
eventsToRetry: readonly Event[],
): Promise<RetryMetadata> {
const response = await this._transport.sendPayload(this._getPayload(eventsToRetry));
let shouldRetry = true;
let shouldReduceEventCount = false;
let eventIndicesToRemove: number[] = [];
if (response.status === Status.RateLimit) {
// RateLimit: See if we hit the daily quota
if (response.body !== undefined) {
const { exceededDailyQuotaUsers, exceededDailyQuotaDevices } = response.body;
if (deviceId in exceededDailyQuotaDevices || userId in exceededDailyQuotaUsers) {
shouldRetry = false; // This device/user may not be retried for a while. Just give up.
}
}
shouldReduceEventCount = true; // Reduce the payload to reduce risk of throttling
} else if (response.status === Status.PayloadTooLarge) {
shouldReduceEventCount = true;
} else if (response.status === Status.Invalid) {
if (eventsToRetry.length === 1) {
shouldRetry = false; // If there's only one event, just toss it.
} else {
eventIndicesToRemove = collectInvalidEventIndices(response); // Figure out which events need to go.
}
} else if (response.status === Status.Success) {
// Success! We sent the events
shouldRetry = false; // End the retry loop
}
return { shouldRetry, shouldReduceEventCount, eventIndicesToRemove };
}
private async _retryEventsOnLoop(userId: string, deviceId: string): Promise<void> {
const eventsBuffer = this._getRetryBuffer(userId, deviceId);
if (eventsBuffer === null || eventsBuffer.length === 0) {
this._cleanUpBuffer(userId, deviceId);
return;
}
let eventCount = eventsBuffer.length;
for (let numRetries = 0; numRetries < this._options.retryTimeouts.length; numRetries++) {
const sleepDuration = this._options.retryTimeouts[numRetries];
await asyncSleep(sleepDuration);
const isLastTry = numRetries === this._options.retryTimeouts.length;
const eventsToRetry = eventsBuffer.slice(0, eventCount);
const { shouldRetry, shouldReduceEventCount, eventIndicesToRemove } = await this._retryEventsOnce(
userId,
deviceId,
eventsToRetry,
);
if (eventIndicesToRemove.length > 0) {
let numEventsRemoved = 0;
// Reverse the indices so that splicing doesn't cause any indexing issues.
Array.from(eventIndicesToRemove)
.reverse()
.forEach(index => {
if (index < eventCount) {
eventsBuffer.splice(index, 1);
numEventsRemoved += 1;
}
});
eventCount -= numEventsRemoved;
this._eventsInRetry -= eventCount;
if (eventCount < 1) {
break; // If we managed to remove all the events, break off early.
}
}
if (!shouldRetry) {
break; // We ended!
}
if (shouldReduceEventCount && !isLastTry) {
eventCount = Math.max(eventCount >> 1, 1);
}
}
// Clean up the events
// Either because they were sent, or because we decided to no longer try them
eventsBuffer.splice(0, eventCount);
this._eventsInRetry -= eventCount;
// if more events came in during this time,
// retry them on a new loop
// otherwise, this call will immediately return on the next event loop.
setTimeout(() => {
// eslint-disable-next-line @typescript-eslint/no-floating-promises
this._retryEventsOnLoop(userId, deviceId);
}, 0);
}
}