feat(openfeature): improve exposure buffering #9742
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -58,7 +58,8 @@ class BaseFFEWriter { | |||||
|
|
||||||
| this._buffer = [] | ||||||
| this._bufferLimit = 1000 | ||||||
| this._bufferStart = 0 | ||||||
| this._dropWarningLogged = false | ||||||
|
|
||||||
| this._config = config | ||||||
| this._endpoint = endpoint | ||||||
|
|
@@ -99,51 +100,87 @@ class BaseFFEWriter { | |||||
| const eventArray = Array.isArray(events) ? events : [events] | ||||||
|
|
||||||
| for (const event of eventArray) { | ||||||
| if (this._buffer.length < this._bufferLimit) { | ||||||
| this._buffer.push(event) | ||||||
| } else { | ||||||
| this._buffer[this._bufferStart] = event | ||||||
| this._bufferStart = (this._bufferStart + 1) % this._bufferLimit | ||||||
|
Comment on lines
+106
to
+107
Member
There was a problem hiding this comment. minor: is there a significant reason to prefer recent events? This seems to introduce more complexity and room for error. Dropping newer events is also very slightly faster (because allocators optimize for young objects dying young). |
||||||
| this._droppedEvents++ | ||||||
|
|
||||||
| if (!this._dropWarningLogged) { | ||||||
| this._dropWarningLogged = true | ||||||
| log.warn( | ||||||
| '%s dropped exposure event(s) at cap %d. This may invalidate experiment results.', | ||||||
|
Member
There was a problem hiding this comment. minor: exposure-specific logging in a potentially shared "base" writer. It doesn't look like it's actually shared, so maybe we should merge the two — this would make the interaction between |
||||||
| this.constructor.name, | ||||||
| this._bufferLimit | ||||||
| ) | ||||||
| } | ||||||
| } | ||||||
| } | ||||||
| } | ||||||
|
|
||||||
| /** | ||||||
| * Sizes, batches, and flushes all buffered events. | ||||||
| */ | ||||||
| flush () { | ||||||
| if (this._buffer.length === 0) { | ||||||
| return | ||||||
| } | ||||||
|
|
||||||
| const events = this._bufferStart === 0 | ||||||
| ? this._buffer | ||||||
| : [...this._buffer.slice(this._bufferStart), ...this._buffer.slice(0, this._bufferStart)] | ||||||
| this._buffer = [] | ||||||
| this._bufferStart = 0 | ||||||
|
|
||||||
| let batch = [] | ||||||
| let batchSize = 0 | ||||||
|
|
||||||
| for (const event of events) { | ||||||
| let eventSize | ||||||
| try { | ||||||
| eventSize = Buffer.byteLength(JSON.stringify(event)) | ||||||
| } catch (error) { | ||||||
| log.warn('%s could not serialize an event, dropping event: %s', this.constructor.name, error.message) | ||||||
|
There was a problem hiding this comment.
When an exposure or its AGENTS.md reference: AGENTS.md:L222-L225 Useful? React with 👍 / 👎. |
||||||
| this._droppedEvents++ | ||||||
| continue | ||||||
| } | ||||||
|
|
||||||
| if (this._eventSizeLimit && eventSize > this._eventSizeLimit) { | ||||||
| log.warn( | ||||||
| '%s event size %d bytes exceeds limit %d, dropping event', | ||||||
| this.constructor.name, | ||||||
| eventSize, | ||||||
| this._eventSizeLimit | ||||||
| ) | ||||||
| this._droppedEvents++ | ||||||
| continue | ||||||
| } | ||||||
|
|
||||||
| if (this._payloadSizeLimit && eventSize > this._payloadSizeLimit) { | ||||||
|
Member
There was a problem hiding this comment. nitpick: I was a bit confused by two almost identical checks. One way to generalize this is to make sure that the event size limit is ≤ payload size limit (which should always be the case), so we can drop the second check. In constructor: if (this._payloadSizeLimit) {
this._eventSizeLimit = Math.min(this._eventSizeLimit || Infinity, this._payloadSizeLimit)
} |
||||||
| log.warn( | ||||||
| '%s event size %d bytes exceeds payload limit %d, dropping event', | ||||||
| this.constructor.name, | ||||||
| eventSize, | ||||||
| this._payloadSizeLimit | ||||||
| ) | ||||||
| this._droppedEvents++ | ||||||
| continue | ||||||
| } | ||||||
|
|
||||||
| if (batch.length > 0 && this._payloadSizeLimit && batchSize + eventSize > this._payloadSizeLimit) { | ||||||
|
Member
There was a problem hiding this comment. nitpick: the piece here looks like it would allow a single event to exceed the whole payload size limit (when
Suggested change
|
||||||
| this.#send(batch) | ||||||
| batch = [] | ||||||
| batchSize = 0 | ||||||
| } | ||||||
|
|
||||||
| batch.push(event) | ||||||
| batchSize += eventSize | ||||||
| } | ||||||
|
|
||||||
| if (batch.length > 0) { | ||||||
| this.#send(batch) | ||||||
| } | ||||||
| } | ||||||
|
|
||||||
| /** | ||||||
|
|
@@ -194,6 +231,34 @@ class BaseFFEWriter { | |||||
| this._fallbackRoute = fallbackRoute ? this.#createRoute(fallbackRoute) : undefined | ||||||
| } | ||||||
|
|
||||||
| /** | ||||||
| * Sends one event batch. | ||||||
| * | ||||||
| * @param {Array<object>} events - Events in the batch | ||||||
| * @returns {void} | ||||||
| */ | ||||||
| #send (events) { | ||||||
| let payload | ||||||
| try { | ||||||
| payload = this._encode(this.makePayload(events)) | ||||||
| } catch (error) { | ||||||
| log.warn( | ||||||
| '%s could not encode %d event(s), dropping batch: %s', | ||||||
| this.constructor.name, | ||||||
| events.length, | ||||||
| error.message | ||||||
| ) | ||||||
| this._droppedEvents += events.length | ||||||
| return | ||||||
| } | ||||||
|
|
||||||
| // eslint-disable-next-line eslint-rules/eslint-log-printf-style | ||||||
| log.debug(() => `${this.constructor.name} flushing payload: ${safeJSONStringify(payload)}`) | ||||||
|
|
||||||
| const route = this.#createActiveRoute() | ||||||
| this.#sendRequest(payload, events.length, route, this._fallbackRoute) | ||||||
| } | ||||||
|
|
||||||
| /** | ||||||
| * Creates request state for a configured writer route. | ||||||
| * | ||||||
|
|
||||||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Because size validation now happens only during
flush(), events over the 1 MB limit still consume slots and replace valid older exposures in this ring. For example, if the queue contains 1,000 valid events and a synchronous burst of oversized evaluation contexts arrives before the timer runs, those invalid events evict the valid batch and are then themselves discarded during flushing, potentially delivering nothing; the count cap also no longer prevents a burst of large unique objects from consuming excessive memory. Oversized events must not be allowed to displace sendable entries in the bounded queue.Useful? React with 👍 / 👎.