fix(router): preserve event order while draining a previously failed job #2546
Conversation
atzoum
force-pushed
the
fix.drainRouterOrder
branch
2 times, most recently
from
81d1172
to
d7bb5d5
Compare
Codecov ReportBase: 43.90% // Head: 43.83% // Decreases project coverage by
Additional details and impacted files@@ Coverage Diff @@
## master #2546 +/- ##
==========================================
- Coverage 43.90% 43.83% -0.07%
==========================================
Files 187 187
Lines 39161 39075 -86
==========================================
- Hits 17195 17130 -65
+ Misses 20877 20861 -16
+ Partials 1089 1084 -5
Help us with your feedback. Take ten seconds to tell us how you rate us. Have a feature suggestion? Share it here. ☔ View full report at Codecov. |
atzoum
force-pushed
the
fix.drainRouterOrder
branch
from
d7bb5d5
to
3ffaf01
Compare
| @@ -1271,7 +1311,7 @@ func (rt *HandleT) findWorker(job *jobsdb.JobT, throttledAtTime time.Time) (toSe | |||
| userID := job.UserID | |||
| // checking if the user is in throttledMap. If yes, returning nil. | |||
| // this check is done to maintain order. | |||
| if _, ok := rt.throttledUserMap[userID]; ok { | |||
| if _, ok := rt.throttledUserMap[userID]; ok && rt.guaranteeUserEventOrder { | |||
There was a problem hiding this comment.
found this was missing
| defer func() { | ||
| if toSendWorker == nil { | ||
| rt.throttler.Dec(parameters.DestinationID, userID, 1, throttledAtTime, throttler.ALL_LEVELS) | ||
| } | ||
| }() |
There was a problem hiding this comment.
Added this since we need to decrement the throttler if no worker is found for the job.
As a general note, the current throttling logic (incrementing/decrementing) appears to be fragile.
cc @fracasula
atzoum
changed the title
atzoum
force-pushed
the
fix.drainRouterOrder
branch
4 times, most recently
from
4c3a44e
to
1660322
Compare
@chandumlg the initial intention was to leave it there as an optimization, i.e. sparing the extra loop for expired jobs that won't be picked up due to another previously failed job. But you are correct, this minor optimization doesn't justify all this extra code. Will simplify it by using a single draining strategy. Thanks! |
atzoum
force-pushed
the
fix.drainRouterOrder
branch
from
93836b2
to
ff8e0d0
Compare
|
LGTM |
Description
When a job needs to be drained from the router, we cannot drain it early, but we need to push it to the worker's queue, so that all other (waiting) jobs for the same userID have time to flush from the pipeline's buffers, before the job is actually drained.
Notion Ticket
Link
Security