TaskScheduler: include the bucket and the content in the action dedupe key - #2798
TaskScheduler: include the bucket and the content in the action dedupe key#2798delthas wants to merge 2 commits into
Conversation
Hello delthas,My role is to assist you with the merge of this Available options
Available commands
Status report is not available. |
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files
... and 3 files with indirect coverage changes
@@ Coverage Diff @@
## development/9.5 #2798 +/- ##
===================================================
- Coverage 75.76% 75.56% -0.20%
===================================================
Files 200 200
Lines 13922 13925 +3
===================================================
- Hits 10548 10523 -25
- Misses 3364 3392 +28
Partials 10 10
Flags with carried forward coverage won't be shown. Click here to find out more. 🚀 New features to boost your workflow:
|
Waiting for approvalThe following approvals are needed before I can proceed with the merge:
|
dc3d703 to
e307e22
Compare
| // Both entry types share a single dedupe cache: keep the | ||
| // namespaces of their keys disjoint |
There was a problem hiding this comment.
| // Both entry types share a single dedupe cache: keep the | |
| // namespaces of their keys disjoint |
| // contentMd5 and eTag hold the same value, the latter quoted | ||
| const content = (contentMd5 || eTag || '').replace(/"/g, ''); |
There was a problem hiding this comment.
| // contentMd5 and eTag hold the same value, the latter quoted | |
| const content = (contentMd5 || eTag || '').replace(/"/g, ''); | |
| const content = contentMd5 || eTag.replace(/"/g, '') || ''; |
Better then ?
There was a problem hiding this comment.
That form throws a TypeError when target.eTag is unset: .replace is evaluated before the || '' can catch it, and unset eTags do occur (tests/functional/replication/queueProcessor.js:730 builds a copyLocation action without one).
It is moot now anyway: contentMd5 is never set by any producer, so the read is gone and it reads (eTag || '').replace(/"/g, '').
There was a problem hiding this comment.
(not specific to this file - using inline comment to allow threaded replies...)
No producer sets target.contentMd5; createCopyLocationAction sets target.eTag, which holds the same value, quoted. Taken from either attribute, unquoted. Without it bucket/key is a mutable identity: a non-versioned object overwritten between two scans collides with its own pending action, and the pending one then fails CopyLocationTask._checkObjectState with "object contents have changed", so neither action transitions the object.
I don't understand this, can you clarify?
Do you mean there is a second problem, i.e. createCopyLoation may be used in non-versioned buckets, where bucket/key is not "unique" (it may relate to different revision/content of the object, if it was replaced) - yet dedup would engage anyway?
→ Is this really an issue? Dedup happens only if both entries are in the queue I guess (so nothing processed yet); and there is only a single "object" to process anyway (if it was replaced, we don't have access to previous content anymore): so whatever message is processed with replicate whatever is there?
→ Or do we have an extra guard when processing the message, to ensure we only replicate exactly what we tried to replicate? (i.e. we actually expect to fail the first replication with incorrect eTag; and succeed the second one if it happens - to avoid some potential race conditions maybe?). In that case dedup should actually still happen - but make sure we keep the latest message only (that is the one with the most "up to date" eTag, so the only one which may succeed) ?
There was a problem hiding this comment.
Yes, this is a second problem, and your second reading is the right one: there is an extra guard, and that is exactly what makes the drop harmful.
CopyLocationTask._checkObjectState compares target.eTag with the object's current content-md5 and rejects a mismatch with InvalidObjectState ("object contents have changed"). So an action published before the overwrite can never succeed — only the most recent one can. Without the eTag in the dedupe key, the two actions collide, dedupe keeps the older one and drops the newer: the old one then fails its own state check and is skipped as committable, and nothing transitions the object until LifecycleResetTransitionInProgressTask clears the stale flag. With the eTag, both run — the stale one costs one metadata read and is skipped, the newest one succeeds.
The preconditions are narrow, to be clear: a non-versioned bucket, the object re-qualifying after the overwrite (immediate with a Date-based rule, since getTransitionTimestamp ignores LastModified), and a data-mover backlog spanning a scan interval.
Your "keep the latest message only" is strictly better than letting both run, and I filed BB-855 for it. It needs TaskScheduler to replace an already-queued task, which it cannot do today, and the class is shared by every consumer — so it did not feel right to fold into this fix.
There was a problem hiding this comment.
Both entry types share a single dedupe cache. Now that both keys carry an unquoted content, this prefix is what keeps a replication entry and a copyLocation on the same object version apart — they are distinct work.
What do you mean "distinct work" ?
Either they are in the same kafka topic / BackbeatConsumer → they are the same work (a single entry may be replicated to multiple destinations) ; or they are indeed distinct work (separate topic / consumer group), should use a distinct dedup cache instead?
There was a problem hiding this comment.
Also wondering about Both entry types share a single dedupe cache : the code indeed supports both kinds of message, but I don't think (at least I hope...) that a single kafka topic can actually contains both.
So each instance if the cache would only get a single type, and the "content" prefix is really just defensive, with no real-life impact?
There was a problem hiding this comment.
You are right that a single topic holds a single type — but the cache is per process, not per topic, and one process consumes both topics.
The backbeat-replication-data-processor deployment runs npm run queue_processor with no topic argument (zenko-operator pkg/controller/zenko/reconcile_backbeat_replication_data_processor_deployment.go:87-96). QueueProcessor then creates _consumer on the replication topic (:744) and _dataMoverConsumer on the data-mover topic (:757), and both push into the single this.taskScheduler (:934 for replication entries, :991 for actions). The replay processor is the exception: it gets a topic argument, so :752 skips the data-mover consumer and it only ever sees ObjectQueueEntry.
That said, I agree on the practical impact: a live collision would also need the same object/version/eTag in flight on both paths at once, and _applyTransitionRule skips objects whose replication status is PENDING/PROCESSING/FAILED. So this is hygiene rather than a bug I can demonstrate.
| @@ -13,16 +13,21 @@ function getTaskSchedulerQueueKey(entry) { | |||
| } | |||
|
|
|||
| function getTaskSchedulerDedupeKey(entry) { | |||
There was a problem hiding this comment.
can we consistently & systematically use "eTag" instead of "content" ?
That is the actual business vocabulary here, unambiguously:
- content could be the data or just about anything;
- contentMD5 is not necessarily the eTag, esp. in case of MPU;
eTagis the S3 standard field designed explicitly to identify "changes" to the content (however it may be computed)
There was a problem hiding this comment.
Adopted, the local is now objectETag.
One correction for the record though: in this codebase content-md5 is the ETag, MPU included. completeMultipartUpload stores createAggregateETag()'s <md5>-<N> into it (arsenal lib/s3middleware/processMpuParts.ts:14-40, cloudserver lib/api/completeMultipartUpload.js:802), and ObjectMD.isMultipartUpload() is literally getContentMd5().includes('-'). The ETag is that same string, quoted at response time (collectResponseHeaders.js:76). So there is no MPU divergence between the two — but agreed that eTag is the unambiguous vocabulary.
| const { bucket, key, version, contentMd5, eTag } = | ||
| entry.getAttribute('target'); | ||
| return `${key}:${version || ''}:${contentMd5}`; | ||
| // contentMd5 and eTag hold the same value, the latter quoted |
There was a problem hiding this comment.
so the eTag/contentMd5 was already (theorically) handled - just not working consistently ? Not a real change (like the bucket part), but more of a fix to standardize the format and fallback to eTag?
In which case would we have eTag but not contentMd5 ?
There was a problem hiding this comment.
Always — no producer has ever set target.contentMd5, in any version (zero occurrences in ReplicationAPI.js on 7.70, 8.6, 9.0 and 9.5, and git log --all -S"contentMd5" on that file returns no commit).
So it was not inconsistent, it was dead: the third component of the key was the literal string undefined for every action ever published. I removed the read entirely, and the key now takes the eTag.
| const version = entry.getVersionId(); | ||
| const contentMd5 = entry.getContentMd5(); | ||
| return `${key}:${version || ''}:${contentMd5}`; | ||
| return `object:${key}:${version || ''}:${contentMd5}`; |
There was a problem hiding this comment.
shouldn't we have ${bucket} here as well?
There was a problem hiding this comment.
It is already there: getCanonicalKey() is ${bucket}/${key} (lib/models/ObjectQueueEntry.js:152-154). I renamed the local to canonicalKey so that reads as such.
| (ctx, done) => this._processTask(ctx.entry, done), | ||
| orderByFunc, | ||
| null, | ||
| this._concurrency, |
There was a problem hiding this comment.
not your change, but it is weird that _concurrency was not used ; can you double check what happened (in git history), to make sure we did not unexpectedly break something there? (i.e. it is really just an incomplete change, not a broken feature...)
There was a problem hiding this comment.
Checked: the argument is simply unused — TaskScheduler never had a fourth parameter. The limit itself was moved, not lost.
Before BB-645 the processing queue was async.queue(this._queueProcessor, this._concurrency), so the concurrency lived in the queue. #2626 replaced it with TaskScheduler and moved the bound into BackbeatConsumer._getAvailableSlotsInPipeline(), which now gates how many messages are fetched: this._concurrency - this._processingQueue.running() - this._nConsumePendingRequests. Your own thread on that PR (r1995146317) is where it was settled.
So nothing is broken, only the argument was left behind at the call site — and the logger now takes its slot.
| } | ||
| if (entry instanceof ActionQueueEntry) { | ||
| const { key, version, contentMd5 } = | ||
| const { bucket, key, version, contentMd5, eTag } = |
There was a problem hiding this comment.
is there always a bucket ? (i.e. accross all uses of "ActionQueueEntry")
does it degrade cleanly if that is not the case?
There was a problem hiding this comment.
Only createCopyLocationAction produces this target shape, and it always sets bucket from params.bucketName. The other ActionQueueEntry producers (deleteData, deleteObject, deleteMPU, deleteArchivedSourceData) go to other topics and processors, and processDataMoverEntry only builds a task for copyLocation with toLocation === this.site.
The assumption is pre-existing too: getTaskSchedulerQueueKey already builds ${bucket}/${key} for actions. And it degrades cleanly — a missing bucket gives undefined/<key>, i.e. the grouping we had before, rather than throwing.
|
Reading back Francois review, whats happens with old data and new data ? I mean, can we have conflicts and/or dedup not working because the key changes at deployment time ? |
The TaskScheduler dedupe key of an ActionQueueEntry omitted the bucket, so two actions targeting the same object key in different buckets were considered duplicates, and the second one was silently dropped. Its third component read target.contentMd5, which no producer has ever set: the content of a copyLocation action is carried by target.eTag. The key of the transition of an object in a non-versioned bucket thus degenerated to `<objectKey>::undefined`, and such an object overwritten with new contents also collided with its own pending action. Build the key from the bucket, the version and the eTag, stripped of its quotes. Both entry types share a single dedupe cache -- a queue processor consumes the replication and the data-mover topics in the same process -- so prefix the keys to keep their namespaces disjoint: a replication entry and an action on the same object version are distinct work and must never be deduplicated against each other. Issue: BB-810
e307e22 to
77767b3
Compare
Dropping a task on a dedupe key match was so far entirely silent, which leaves no trace to diagnose from when a task is skipped wrongly. The logger takes the constructor slot of a concurrency parameter that TaskScheduler never had: that limit is enforced by the consumer, in BackbeatConsumer._getAvailableSlotsInPipeline(). Issue: BB-810
77767b3 to
04da66b
Compare
No conflict is possible: the dedupe cache is per-process and in-memory, and it only holds tasks that are currently queued or in flight. Keys are computed at push time, from the message, by whichever process consumes it — they are never persisted, and never compared between processes. A process that restarts or is upgraded starts with an empty cache. Mixed producer versions are not a problem either. Nothing has ever written The worst case during a rolling upgrade is that an old and a new process each deduplicate within their own window, so a duplicate pair could be processed twice instead of once. That is the at-least-once behaviour the pipeline already handles: |
Request integration branchesWaiting for integration branch creation to be requested by the user. To request integration branches, please comment on this pull request with the following command: Alternatively, the |
4 participants
The
TaskSchedulerdedupe key of anActionQueueEntryomitted the bucket, so two actions targeting the same object key in different buckets were considered duplicates and the second one was silently dropped — it is acknowledged and its offset committed, so nothing replays it.What the key is made of, before and after:
QueueProcessor.js:744and:757, both feedingthis.taskScheduler). The prefix keeps a replication entry and acopyLocationon the same object version apart — they are distinct work.index.htmlin two buckets gave the same key, and the second action was dropped.undefinedfor objects in non-versioned buckets, hence the eTag below.contentMd5, alwaysundefinedtarget.contentMd5, in any version: the content of acopyLocationaction is carried bytarget.eTag. Without itbucket/keyis a mutable identity — a non-versioned object overwritten between two scans collides with its own pending action, and the pending one then failsCopyLocationTask._checkObjectStatewith "object contents have changed", so neither action transitions the object. Quotes are stripped so the value matches thecontent-md5used on the other branch.toLocationprocessDataMoverEntryonly builds a task whentoLocation === this.site, so it is constant within a given scheduler.The
ObjectQueueEntrybranch gets the matching prefix; it is otherwise unchanged, as it already carried the bucket (viagetCanonicalKey()) and a realcontentMd5.Nothing to migrate: the dedupe cache is per-process and in-memory, cleared when the task ends, and the key is derived from attributes already published.
Tasks skipped by deduplication are now logged, which was so far silent. The logger takes the constructor slot of a concurrency parameter that
TaskSchedulernever had — that limit is enforced by the consumer, inBackbeatConsumer._getAvailableSlotsInPipeline().Issue: BB-810