rotor: performance optimizations #1192
Conversation
…f just 1 function rotor: added support for 'fastFunctions' option that disables copying of events between functions chain steps rotor: bulker-destination: switched to undici rotor: small tweaks
![mrge-io[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
| if (err.name !== DropRetryErrorName && err.event) { | ||
| const errEvent = err.event; | ||
| if (err.name !== DropRetryErrorName) { | ||
| const errEvent = err.event || event; |
There was a problem hiding this comment.
Behavior change in error handling may lead to unexpected processing. The original condition only continued with events if err.event was truthy, but now fallback to original event occurs. This could propagate events that might have been intentionally dropped in error cases.
| result = undefined | ||
| return result | ||
| } catch (err) { | ||
| throw err; |
There was a problem hiding this comment.
Removed retry policy annotation on errors
| const chainRes = await runChain(chain, iterator, user, ctx); | ||
| checkError(chainRes); | ||
| return chainRes.result; | ||
| return runChain(chain, iterator, user, ctx); |
There was a problem hiding this comment.
The execution log and error tracking have been completely removed
| } | ||
| if (isDropResult(result)) { | ||
| result = undefined | ||
| return result |
There was a problem hiding this comment.
Result handling has changed from {result, execLog} object to direct return of result
| dispatcher: undiciAgent, | ||
| }); | ||
| if (res.statusCode != 200) { | ||
| throw new HTTPError(`HTTP Error: ${res.statusCode}`, res.statusCode, await res.body.text()); |
There was a problem hiding this comment.
Error message is missing statusText, which was present in the original code. This reduces debugging information when errors occur
| ); | ||
| if (!res.ok) { | ||
| throw new HTTPError(`HTTP Error: ${res.status} ${res.statusText}`, res.status, await res.text()); | ||
| const res = await request(`${bulkerEndpoint}/post/${destinationId}?tableName=${table}`, { |
There was a problem hiding this comment.
The original fetch call included a { log: false } option that's been removed in the new implementation, which might affect logging behavior
| error: err, | ||
| }]; | ||
| } | ||
| if (!isDropResult(result)) { |
There was a problem hiding this comment.
In runSingle, when result is falsy but not a drop result, the original event is lost unlike in runChain
| error: err, | ||
| }]; | ||
| } | ||
| if (!isDropResult(result)) { |
There was a problem hiding this comment.
runSingle doesn't handle array results correctly. It just assigns events = result rather than unpacking arrays like runChain does
| if (!isDropResult(result)) { | ||
| events = result; | ||
| } | ||
| return {events, execLog}; |
There was a problem hiding this comment.
runSingle function doesn't ensure events is always an array, unlike runChain which always returns an array for events
| let execLog = []; | ||
| let events = []; | ||
| let result = undefined; | ||
| try { |
There was a problem hiding this comment.
The runSingle function doesn't respect the fastFunctions option that runChain uses, creating inconsistent behavior
| try { | ||
| result = await f.f(deepCopy(event), ctx); | ||
| result = await f.f(fastFunctions ? event : deepCopy(event), ctx); |
There was a problem hiding this comment.
Enabling fastFunctions option could lead to unexpected side effects if functions modify event objects
| const chainRes = await runChain(chain, iterator, user, ctx); | ||
| checkError(chainRes); | ||
| return chainRes.result; | ||
| return runChain(chain, iterator, user, ctx); |
There was a problem hiding this comment.
Changed return structure could break consumers expecting {result, execLog} format
| result = undefined | ||
| return result | ||
| } catch (err) { | ||
| throw err; |
There was a problem hiding this comment.
Loss of detailed error information from execLog
| @@ -273,6 +273,7 @@ const exports: Export[] = [ | |||
| options: { | |||
| ...data, | |||
| ...((workspace.featuresEnabled ?? []).includes("nofetchlogs") ? { fetchLogLevel: "debug" } : {}), | |||
| ...((workspace.featuresEnabled ?? []).includes("fastFunctions") ? { fastFunctions: true } : {}), | |||
There was a problem hiding this comment.
The optionsHash is calculated using only the original data object and doesn't include the newly added fastFunctions flag
| const chainRes = await runChain(chain, iterator, user, ctx); | ||
| checkError(chainRes); | ||
| return chainRes.result; | ||
| return runChain(chain, iterator, user, ctx); |
There was a problem hiding this comment.
Error handling flow has fundamentally changed, removing the intermediary execution log.
| error: err, | ||
| }]; | ||
| } | ||
| if (!isDropResult(result)) { |
There was a problem hiding this comment.
Missing handling for falsy non-drop results in runSingle
| }]; | ||
| } | ||
| if (!isDropResult(result)) { | ||
| events = result; |
There was a problem hiding this comment.
Direct assignment to events without ensuring it's an array
| try { | ||
| result = await f.f(deepCopy(event), ctx); | ||
| result = await f.f(fastFunctions ? event : deepCopy(event), ctx); |
There was a problem hiding this comment.
Potential side effects with fastFunctions optimization
| if (err.name !== DropRetryErrorName && err.event) { | ||
| const errEvent = err.event; | ||
| if (err.name !== DropRetryErrorName) { | ||
| const errEvent = err.event || event; |
There was a problem hiding this comment.
Falling back to the original event may bypass pipeline transformations
rotor: don't do excessive event copy if functions pipeline consists of just 1 function
rotor: added support for 'fastFunctions' option that disables copying of events between functions chain steps
rotor: bulker-destination: switched to undici
rotor: small tweaks