Refactor replicate object with async await - #2780
Conversation
Hello sylvainsenechal,My role is to assist you with the merge of this Available options
Available commands
Status report is not available. |
| /** | ||
| * Runs an async task for each item in a collection, up to `limit` at a time. | ||
| * On error, no new tasks are started, but already-running tasks are awaited | ||
| * before the function resolves. Returns [firstError, results] so that callers |
There was a problem hiding this comment.
Return the first error only : Keeping same behavior as before with mapLimitWaitPendingIfError.
Could've returned all errors as a list but not needed now and prefer to keep existing behavior.
Cannot aggregate errors neither, as we need to keep the original error to check its properties to determine if its retryable or not
There was a problem hiding this comment.
but already-running tasks are awaited what happens if an error is triggered here ? Can make sense to stop new flows on error, but for the current one we can miss an error ?
There was a problem hiding this comment.
Ok I saw your comment. I'm not sure it's the right approach and we should return all errors ?
There was a problem hiding this comment.
Mhh its replicating initial pre migration behavior, I think its fine anyways as replicateObject will try to delete all orphans as soon as a single error is found
Although maybe there is a question raised about reusability of this function, maybe its built to be too custom for replicateObject and should be more generic (return array of errors) ?
There was a problem hiding this comment.
refactored with async await, in the file runTaskWithConcurrency
Waiting for approvalThe following approvals are needed before I can proceed with the merge:
|
Codecov Report❌ Patch coverage is
Additional details and impacted files
... and 3 files with indirect coverage changes
@@ Coverage Diff @@
## development/9.5 #2780 +/- ##
================================================
Coverage 75.72% 75.73%
================================================
Files 201 201
Lines 13937 13937
================================================
+ Hits 10554 10555 +1
+ Misses 3373 3372 -1
Partials 10 10
Flags with carried forward coverage won't be shown. Click here to find out more. 🚀 New features to boost your workflow:
|
406a3d4 to
7f5778f
Compare
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
7f5778f to
5680800
Compare
5680800 to
64afb79
Compare
There was a problem hiding this comment.
Migrations here aren't the main point of this pr, but were kinda forced as MultipleBackend extends ReplicateObject.
I did the minimum required migration on this file
| await this._deleteOrphans(destEntry, destLocations, log); | ||
| throw err; | ||
| } | ||
| return this._handleReplicationOutcome(null, sourceEntry, destEntry, kafkaEntry, log, done); |
There was a problem hiding this comment.
_handleReplicationOutcome(null, …, done) is called from inside the try: if it throws after calling done, the catch calls it again and done fires twice. We can compute the error in the try/catch and call the handler exactly once after it. (same at line 909 and in _processQueueEntryRetryFull)
Waiting for approvalThe following approvals are needed before I can proceed with the merge:
The following reviewers are expecting changes from the author, or must review again: |
64afb79 to
7467f4b
Compare
|
Sorry I took some time to respond, you can review again next week @benzekrimaha @maeldonn |
449cafe to
05fd0ff
Compare
I didnt run the linter though, the diff is just re indentation mostly 🤔 |
05fd0ff to
03574e0
Compare
03574e0 to
6dd6d10
Compare
| }); | ||
| throw errors.BadRole; | ||
| } | ||
| return [this.sourceRole, entryRoles[1]]; |
There was a problem hiding this comment.
entryRoles[1] is undefined for the single-role external-backend case , the value is discarded by _setupClients, but returning a knowingly-undefined slot is confusing; a comment or returning only what's meaningful would help.
There was a problem hiding this comment.
Thanks for the reviews, i didnt expect to mess it up so much -_-
There was a problem hiding this comment.
I'm gonna remove it, just like it was before as it's not used
6dd6d10 to
5070c36
Compare
| actionDesc: 'get bucket replication configuration', | ||
| logFields: { entry: entry.getLogInfo() }, | ||
| actionFunc: done => this._setupRolesOnce(entry, log, done), | ||
| actionFunc: done => this._setupRolesOnce(entry, log) |
There was a problem hiding this comment.
The actionFunc callbacks bridge async methods back to callback-based retry(), but retry() now supports a promise mode (when done is omitted). Since you already added the promise overload in BackbeatTask.retry, these wrappers could use it directly to eliminate the manual .then(r => done(null, r), done) bridging. For example:
async _setupRoles(entry, log) {
return await this.retry({
actionDesc: 'get bucket replication configuration',
logFields: { entry: entry.getLogInfo() },
actionFunc: async () => this._setupRolesOnce(entry, log),
shouldRetryFunc: err => err.retryable,
log,
});
}This applies to all four retry wrappers (_setupRoles, _setTargetAccountMd, _getAndPutPart, _putMetadata). Not a blocker — could be a follow-up.
| logFields: { entry: entry.getLogInfo() }, | ||
| actionFunc: done => this._setupRolesOnce(entry, log, done), | ||
| actionFunc: done => this._setupRolesOnce(entry, log) | ||
| .then(roles => done(null, roles), done), |
There was a problem hiding this comment.
Correction to my previous comment: actionFunc must still accept a callback since retry() internally calls actionFunc(done, nbRetries). The .then(r => done(null, r), done) bridging pattern is necessary with the current retry() implementation. Disregard the suggestion above — the current code is correct.
A cleaner solution would require refactoring retry() to natively accept an async actionFunc, which is out of scope for this PR.
| return doneOnce(...args); | ||
| }; | ||
| actionFunc(_handleRes, nbRetries); | ||
| return actionFunc(_handleRes, nbRetries); |
There was a problem hiding this comment.
nothing consumes that return value, why change it?
There was a problem hiding this comment.
I added it because of the "consistent-return" linter but yeah better to return undefined
| actionFunc: done => this._setupRolesOnce(entry, log) | ||
| .then(roles => done(null, roles), done), |
There was a problem hiding this comment.
all these done => this._xxxOnce(...).then(r => done(null, r), done) bridges keep retry callback-based inside an otherwise async class , fine as a transition, but can we please create a followup ticket to convert retry internals so we don't keep both paradigms. (applies to all similar cases)
There was a problem hiding this comment.
yeah and i think i would even create a follow to fully move retry to be async, and to take an async action function because its really annoying to deal with this transition state
| if (err.$metadata?.httpStatusCode === 404) { | ||
| return doneOnce(err); | ||
| // eslint-disable-next-line no-param-reassign | ||
| err.origin = 'source'; |
There was a problem hiding this comment.
behavior change: the old code rejected errors.ObjNotFound without origin, so _handleReplicationOutcome took the “target object not found, retrying with full data write” path; setting origin = 'source' here makes it take the skip path instead. Skipping looks correct for a source-side 404 mid-stream, but is the change intended? IMO it deserves a test.
There was a problem hiding this comment.
yes this was discussed and the change is small and intended, i added 2 tests
| err.ObjNotFound = true; | ||
| // eslint-disable-next-line no-param-reassign | ||
| err.name = 'ObjNotFound'; |
There was a problem hiding this comment.
why mutate the SDK error instead of rejecting errors.ObjNotFound like before? this satisfies the err.name === 'ObjNotFound' checks but loses err.is.*, so any future err.is.ObjNotFound check will silently miss it. arsenal errors is the canonical module...
There was a problem hiding this comment.
Yeah I think i did this because right after the if statement use similar error rewrite pattern but we can use the arsenal error
There was a problem hiding this comment.
Addressed in another similar comment, I messed up, I think wanted to follow similar pattern to the else branch but yeah we can use arsenal error
|
|
||
| processQueueEntry(_sourceEntry, kafkaEntry, done) { | ||
| this._processQueueEntry(_sourceEntry, kafkaEntry).then( | ||
| result => result === null ? done() : done(null, result), |
There was a problem hiding this comment.
does the queue processor actually distinguish done() from done(null, undefined)? if not, the null sentinel and this ternary can go away I think
There was a problem hiding this comment.
yeah its fine
| * @param {Array} coll - collection to iterate over | ||
| * @return {Promise<[Error|null, Array]>} - always resolves, never rejects | ||
| */ | ||
| async function runTasksWithConcurrency(task, limit, coll) { |
There was a problem hiding this comment.
two things:
- why swap the parameter order? both
async.mapLimitand the oldmapLimitWaitPendingIfErroruse(coll, limit, iteratee),keeping the conventional order would avoid silent argument swaps. - a never-rejecting
[err, results]tuple is easy to misuse: a caller who forgets to check[0]silently ignores errors... The partial-results need is real, so OK if kept, but the doc comment should be kept prominent
There was a problem hiding this comment.
Yes both legit and addressed
| .then(() => assert.fail('expected error')) | ||
| .catch(err => { |
There was a problem hiding this comment.
use await assert.rejects(...): as written, when the task unexpectedly succeeds, the assert.fail AssertionError is swallowed by the following .catch and re-asserted, so the test fails with a confusing message instead of “expected error”. (applies to all similar cases)
There was a problem hiding this comment.
yeah and many similar patterns actually
| next => this._refreshSourceEntry(sourceEntry, log, (err, res) => { | ||
| if (err && err.name === 'ObjNotFound' && | ||
| sourceEntry.getReplicationIsNFS() && !sourceEntry.getIsDeleteMarker()) { | ||
| next => this._refreshSourceEntry(sourceEntry, log) |
There was a problem hiding this comment.
not really just a bit of habits to have. After 3/4 migrations you'll be an expoert
Waiting for approvalThe following approvals are needed before I can proceed with the merge:
The following reviewers are expecting changes from the author, or must review again: |
this commit is mandatory as multipleBackend is a class that extends replicateObject and some functions are overloaded Issue: BB-803
5070c36 to
ae7ea53
Compare
| if (err.$metadata?.httpStatusCode === 404) { | ||
| return doneOnce(err); | ||
| const objNotFound = errors.ObjNotFound; | ||
| objNotFound.origin = 'source'; |
There was a problem hiding this comment.
errors.ObjNotFound is a singleton — const objNotFound = errors.ObjNotFound does not clone it. Setting .origin = 'source' here permanently mutates the shared object for the lifetime of the process, so every future consumer of errors.ObjNotFound (in any code path) will see origin === 'source'.
| objNotFound.origin = 'source'; | |
| return reject(errors.ObjNotFound); |
There was a problem hiding this comment.
false, errors.ObjNotFound invokes this getter :
static errors() → { get: () => new ArsenalError(...) })
Waiting for approvalThe following approvals are needed before I can proceed with the merge:
|
|
/approve |
|
I have successfully merged the changeset of this pull request
The following branches have NOT changed:
This pull request did not target the following hotfix branch(es) so they
Please check the status of the associated issue BB-803. Goodbye sylvainsenechal. The following options are set: approve |
4 participants
Issue: BB-803
Review hints :
Imo this refactor is super overdue and not so complicated to do.
Refactor replicate object with async await