Conversation
Contributor
Hello delthas,My role is to assist you with the merge of this Available options
Available commands
Status report is not available. |
Codecov Report❌ Patch coverage is
Additional details and impacted files
... and 3 files with indirect coverage changes
@@ Coverage Diff @@
## development/9.6 #2833 +/- ##
===================================================
- Coverage 76.50% 76.37% -0.13%
===================================================
Files 204 209 +5
Lines 14120 14191 +71
===================================================
+ Hits 10802 10839 +37
- Misses 3308 3342 +34
Partials 10 10
Flags with carried forward coverage won't be shown. Click here to find out more. 🚀 New features to boost your workflow:
|
Contributor
Waiting for approvalThe following approvals are needed before I can proceed with the merge:
|
delthas
force-pushed
the
improvement/BB-811/dr-mode-mongo-processor
branch
2 times, most recently
from
August 27, 2026 13:37
1a74a03 to
8675518
Compare
Contributor
Waiting for approvalThe following approvals are needed before I can proceed with the merge:
|
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
delthas
force-pushed
the
improvement/BB-811/dr-mode-mongo-processor
branch
from
August 31, 2026 08:31
e77010b to
5112a3a
Compare
delthas
force-pushed
the
improvement/BB-811/dr-mode-mongo-processor
branch
from
August 31, 2026 09:24
5112a3a to
0eb82b1
Compare
delthas
force-pushed
the
improvement/BB-811/dr-mode-mongo-processor
branch
2 times, most recently
from
September 4, 2026 15:18
2956946 to
d161e55
Compare
delthas
force-pushed
the
improvement/BB-811/dr-mode-mongo-processor
branch
from
September 7, 2026 15:44
d161e55 to
486c8e2
Compare
delthas
marked this pull request as ready for review
September 7, 2026 15:55
Data on an `isCRR` location belongs to a remote site: it may be read, but never deleted, and a version whose data still lives there has not been localized yet. Several places need to ask that question. Lifted verbatim from BB-813 (#2828), which introduces the same helper and carries its unit test and the location fixture. Kept byte-identical so that whichever branch lands second has this commit dropped as already applied, rather than conflicting. Issue: BB-811
mongoProcessorTask takes the mongo client config from config.queuePopulator.mongo, which works when a queue populator is deployed beside the processor. The D/R metadata sink deploys none, so its configuration would have to carry a queuePopulator block it never runs. Accept extensions.mongoProcessor.mongodb, validated with the shared mongoJoi, and fall back to the queue populator's config when absent. The fallback is guarded: a config omitting queuePopulator entirely would otherwise throw before validation could report anything useful. Issue: BB-811
The `server` section is optional, but the whitelist of addresses allowed to reach the health checks was extended unconditionally, so a process serving no API exited before starting: a D/R sink runs the mongo-processor alone, and configuring a section it never reads to get past this is no answer. Issue: BB-811
delthas
force-pushed
the
improvement/BB-811/dr-mode-mongo-processor
branch
from
September 10, 2026 12:52
486c8e2 to
8b693e6
Compare
The mongo-processor was written as the out-of-band ingestion consumer, and the D/R metadata sink now reuses it. The two disagree about most of what it does to an object, so put those decisions behind a mode: an abstract ProcessorMode whose methods assert, an implementation per mode, and an index mapping the configured name to the class, as the notification extension does for its destinations. IngestionMode carries today's behaviour verbatim, so this commit changes nothing. The default lives beside the mode map, so a processor built programmatically gets the same mode as one built from a config file. Issue: BB-811
The D/R metadata sink replicates production's objects, accounts included, so it applies what the source-side pipeline sends rather than rewriting it into something local: a new object is written as it arrives, and an update takes the entry's tags, object-lock state and ACLs while keeping the placement already stored. Keeping the stored placement is the point. The copy engine rewrites location and dataStoreName to a local location after the first write, and applying the entry's would send reads back to the source and leak a local copy that is never garbage-collected. Cleared values are applied like any other: removing a legal hold is an update. Three things follow from that and were unreachable before it: - the stored document is always read, because it is what distinguishes a first write from an update, and the entry cannot -- an insert is redelivered on replay and overlaps the bootstrap dump, so it is no promise that the object is absent here; - object-lock and ACL changes count as changes, where the ingestion diff looks only at tags and dropped them as duplicates; - a delete always applies, where the ingestion guard skips one whose object has moved location, which for a replicated object it always has. Issue: BB-811
Backbeat validates every extension listed under `extensions` against that extension's own schema, so naming the service account in a partial `extensions.ingestion` block makes the whole configuration invalid: the ingestion schema also requires a topic, a zookeeper path and a source list. A D/R sink runs no extension but this one, so it now carries the account in its own configuration, falling back to the ingestion extension for the deployments that enable both. Issue: BB-811
Change detection compared the metadata an update is allowed to carry, which left out the content itself: an object overwritten in place kept its version id, so the entry diffed to nothing and was dropped as a duplicate, and the sink went on describing the object that had been replaced. An object with no version of its own is rewritten in place, and so is the master a versioning suspended bucket marks null. Such an entry does not update the stored document, it replaces it: content, headers and user metadata all move, and the merge kept none of them. The Kafka Connect sink this replaces wrote the whole document -- the source pipeline turned a whole value write into a replace -- so the merge was a regression against it. Decide in the processor whether an entry replaces the stored object or updates it, and apply a replacing one as a first write. A version is immutable, so an entry for one still merges: clean room replicates versioned buckets only and keeps the semantics it had. Nothing of the stored document survives a replacement. Placement is the one field this site could own, and it cannot here: a cold object holds what the source pipeline derives from its storage class, and a restore this site performed describes bytes that are gone -- its expiration is sorbet's own to reconcile. Duplicate detection compares the document a write would leave with the one stored, so a replay is still a no-op while a rewrite that moved only a header is not. The content digest alone cannot tell the two apart: a PUT replaces the whole metadata whatever the bytes do. Issue: BB-811
delthas
force-pushed
the
improvement/BB-811/dr-mode-mongo-processor
branch
from
September 11, 2026 09:19
8b693e6 to
e1dd485
Compare
delthas
requested review from
maeldonn
and removed request for
SylvainSenechal
September 11, 2026 15:07
Contributor
Author
|
Requested @maeldonn in place of Sylvain Senechal, who is currently on PTO. |
delthas
requested review from
SylvainSenechal
and removed request for
maeldonn
September 14, 2026 00:32
Contributor
Author
|
Requested @SylvainSenechal in place of Mael Donnart, who is currently on PTO. |
Contributor
|
nit: PR description is way too long to read, schema way too verobose... so it does not help so much.
→ not sure if/what we should do in this PR, but something to consider anyway |
| topic: joi.string().required(), | ||
| groupId: joi.string().required(), | ||
| mode: joi.string().valid(...Object.keys(modes)).default(defaultMode), | ||
| mongodb: mongoJoi.when('mode', { is: 'dr', then: joi.required() }), |
Contributor
There was a problem hiding this comment.
mongodb config is already available in the "global" config, should not be duplicated here
| const ProcessorMode = require('./ProcessorMode'); | ||
| const getContentType = require('../utils/contentTypeHelper'); | ||
|
|
||
| class DRMode extends ProcessorMode { |
Contributor
There was a problem hiding this comment.
maybe more PullReplication than DR ?
(e.g. this is actually the metadata part of the pull replication feature of backbeat)
Comment on lines
+21
to
+22
| // the queue populator's mongo config is the fallback for a deployment that | ||
| // runs one beside this process; a D/R sink runs none, so it configures its own |
Contributor
There was a problem hiding this comment.
I don't understand this comment at all.
In any case, mongo-processor runs next to a mongo instance, and it is passed through the usual config.queuePopulator.mongo...
Comment on lines
+67
to
+70
| // the ingestion extension's auth is the fallback for a deployment that | ||
| // configures it; a D/R sink enables no extension but this one | ||
| const serviceAuth = | ||
| mongoProcessorConfig.auth ?? config.extensions.ingestion?.auth; |
Contributor
There was a problem hiding this comment.
this one may be trickier, not sure if/how we want to hack around...
mongo-processor was (technically) extension, but kind of "part" of ingestion → so using that extension's config.
now you are making the mongoProcessor cover multiple extensions (both replication and ingestion, basically) ; and introducing lots of conditional to address this...
- does this mean it should be separate extensions, instead of "modes" ?
- or should these ternary be handled in the "mode" class ?
- or do we really need to have different service account? (btw, you kept the serviceName: so the difference does not seem so critical?)
- or just handle it at "runtime", e.g. add extra env variable in ZKOP ?
| // not have repInfo set and the header is not set, then we can skip fetching. | ||
| const bucketRepInfo = bucketInfo.getReplicationConfiguration(); | ||
| if (!scalVersionId && !bucketRepInfo?.rules?.some(r => r.enabled)) { | ||
| if (!this._mode.needsExistingMetadata(sourceEntry, bucketInfo)) { |
Contributor
There was a problem hiding this comment.
nit: I wonder if we need this condition as well... it is an optimization, but not sure how relevant it is...
Comment on lines
+45
to
+88
| /** | ||
| * Apply the metadata fields an object gets when it is first written here. | ||
| * | ||
| * This method must be implemented by subclasses of ProcessorMode | ||
| * @param {ObjectQueueEntry} entry - object queue entry object | ||
| * @param {string} location - zenko storage location name | ||
| * @param {BucketInfo} bucketInfo - bucket info object | ||
| * @return {undefined} | ||
| */ | ||
| applyNewObjectMetadata(entry, location, bucketInfo) { // eslint-disable-line no-unused-vars | ||
| assert(false, | ||
| 'sub-classes of ProcessorMode must implement ' + | ||
| 'the applyNewObjectMetadata() method'); | ||
| } | ||
|
|
||
| /** | ||
| * Whether the entry replaces the object already stored rather than | ||
| * updating it, in which case it is applied as a first write. | ||
| * | ||
| * This method must be implemented by subclasses of ProcessorMode | ||
| * @param {ObjectQueueEntry} entry - object queue entry object | ||
| * @param {Object} zenkoObjMd - metadata fetched from mongo | ||
| * @return {boolean} true if the entry replaces the stored object | ||
| */ | ||
| replacesExistingMetadata(entry, zenkoObjMd) { // eslint-disable-line no-unused-vars | ||
| assert(false, | ||
| 'sub-classes of ProcessorMode must implement ' + | ||
| 'the replacesExistingMetadata() method'); | ||
| } | ||
|
|
||
| /** | ||
| * Merge the entry into the object already stored, deciding which fields the | ||
| * entry brings and which the stored document keeps. | ||
| * | ||
| * This method must be implemented by subclasses of ProcessorMode | ||
| * @param {ObjectQueueEntry} entry - object queue entry object | ||
| * @param {Object} zenkoObjMd - metadata fetched from mongo | ||
| * @return {undefined} | ||
| */ | ||
| mergeExistingMetadata(entry, zenkoObjMd) { // eslint-disable-line no-unused-vars | ||
| assert(false, | ||
| 'sub-classes of ProcessorMode must implement ' + | ||
| 'the mergeExistingMetadata() method'); | ||
| } |
Contributor
There was a problem hiding this comment.
maybe the logic would be simpler with a single function here mergeObjMD : receiving both the entry and zenkoObjMD (which may or may not be null) ?
we may not even need getChangedContent in that case...
| * @param {string|undefined} versionId - decoded version id of the entry | ||
| * @return {boolean} true if the object should be deleted | ||
| */ | ||
| shouldProcessDelete(zenkoObjMd, location, key, versionId) { // eslint-disable-line no-unused-vars |
Contributor
There was a problem hiding this comment.
same, would it be better to call this function _skipDelete ?
| delete written.replicationInfo; | ||
| delete stored.replicationInfo; | ||
|
|
||
| return !isDeepStrictEqual(written, stored); |
Contributor
There was a problem hiding this comment.
that is not quite costly, and not really safe anyway (could be a concurrent update in mongo)
- there is no real "replace" : we should always merge, i.e. keep the existing replicationInfo, dataLocation, etc...
- in theory could be some corner case with non-versionned: but versioning is required (should be enforced) for pull replication, just like it is for replication
in general, the goal is to have the "same metadata": so I think we shoudl do similar to replication (in cloudserver) -and unlike ingestion-, i.e. keep all the metadata except the few field we must keep: dataLocation, dataStoreName (though a story of its own, e.g. what is the object was transitioned in between?...), replicationInfo, archive status... and not try to optimize mongo writes at all cost
| * @param {Object} zenkoObjMd - metadata fetched from mongo | ||
| * @return {boolean} true if the entry changes mutable metadata | ||
| */ | ||
| _hasMutableChange(entry, zenkoObjMd) { |
Contributor
There was a problem hiding this comment.
an efficient way to check for changes could be the microVersionId (added for CRR cascade), bumped on every metadata write which must be replicated?
| } | ||
|
|
||
| replacesExistingMetadata(entry, zenkoObjMd) { // eslint-disable-line no-unused-vars | ||
| // an object with no version of its own is rewritten in place, so the |
Contributor
There was a problem hiding this comment.
- the bucket is versioned. so only MD can change
- if it were not, we could not just replace the object, as it may create an orphan...
overall this point to the point already raised in ProcessorMode: having a single "mergeMetadata", to keep the whole merging logic in each "mode", may be clearer? (and may help defining more precisely what a mode is - or a more precise name/concept)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
The mongo-processor was written as the out-of-band ingestion consumer, and the D/R metadata sink now reuses it. The two disagree about most of what it does to an object: ingestion rewrites identity and placement to local values, because the source system's accounts and locations do not exist here, while D/R replicates both and so applies what the source-side pipeline sends.
Put those decisions behind a mode, following the shape the notification extension uses for its destinations — an abstract
ProcessorMode, an implementation per mode, and an index mapping the configured name to the class.IngestionModecarries today's behaviour verbatim andmodedefaults toingestion, so nothing changes for an existing deployment.What the mongo-processor does
flowchart LR K[("Kafka topic")] --> E["read entry"] E --> B["look up the bucket"] B --> Z["read the stored object metadata"] Z --> A["apply the entry"] A --> M[("sink MongoDB")]Where the two modes diverge
Every decision the two modes disagree about is a hook on
ProcessorMode, so the surrounding flow stays single-sourced.flowchart TD E["entry"] --> T{"type"} T -->|del| G0["shouldProcessDelete"] G0 -->|ingestion| I0["only while the stored placement<br/>still matches the bucket"] G0 -->|D/R| D0["always"] I0 --> DEL[("delete in MongoDB")] D0 --> DEL T -->|put| G1["needsExistingMetadata"] G1 -->|ingestion| I1["only with a scal header or<br/>an enabled replication rule"] G1 -->|D/R| D1["always"] I1 --> G2["resolveVersionId"] D1 --> G2 G2 -->|ingestion| I2["the scal version id when present"] G2 -->|D/R| D2["the entry's own"] I2 --> G3["getChangedContent"] D2 --> G3 G3 -->|ingestion| I3["tags"] G3 -->|D/R| D3["tags, object lock, and placement<br/>while the data is still remote"] I3 --> S{"stored document"} D3 --> S S -->|absent| G4["applyNewObjectMetadata"] S -->|present| G6["replacesExistingMetadata"] G6 -->|ingestion| G5["mergeExistingMetadata"] G6 -->|"D/R, versioned"| G5 G6 -->|"D/R, no version id<br/>or a null version"| G4 G4 -->|ingestion| I4["local owner, location and data part,<br/>ACLs reset"] G4 -->|D/R| D4["ACLs reset"] G5 -->|ingestion| I5["keep the stored document,<br/>take the tags"] G5 -->|D/R| D5["keep the stored document, take tags<br/>and object lock, placement by locality"] I4 --> W[("write to MongoDB")] D4 --> W I5 --> W D5 --> WThe hooks
needsExistingMetadataresolveVersionIdx-amz-meta-scal-version-idwhen present, otherwise the entry'sgetChangedContentapplyNewObjectMetadatadataStoreNameand the data part to local values, resets ACLsreplacesExistingMetadatamergeExistingMetadatashouldProcessDeleteThe three marked beyond the ticket are not in the ticket text but follow from the design:
getObjectper entry.getContentTypeonly diffs tags, so a retention or legal-hold change with unchanged tags was dropped as a duplicate before the merge could run. The requirement is the matrix's "Merged on update"; the mechanism is ours.† The one decision the design does not cover.
x-amz-meta-scal-version-idis ignored in D/R mode. Both call sites used it mode-independently to choose which version to read and to delete. A production object that was itself OOB-ingested or cold-restored carries that header, naming a version of the system production ingested from — so honouring it here reads and deletes the wrong version. Version ids are the source's own and identical on both sides, so the entry's own id is authoritative. The design never considers a production site that itself ingests, so this is a guess, and the easiest thing in the PR to reverse.What happens to each field
owner-id,owner-display-namedataStoreNamelocationdata partsacltagsretentionMode,retentionDate,legalHoldversionIdx-amz-meta-scal-version-idreplicationInfocontent-length,content-md5, and the restConfiguration
The mongo-processor now reads two settings from its own extension config instead of from blocks it does not own: the mongo client config, so a D/R sink no longer carries a
queuePopulatorblock it never runs, and the service account, so it no longer carries a partialextensions.ingestionblock. The second was not optional — backbeat validates every extension listed underextensionsagainst that extension's own schema, and the ingestion schema also requires a topic, a zookeeper path and a source list, so naming the account there made the whole configuration invalid. Both fall back to the old location for a deployment that enables both extensions, andauthis required in D/R mode, which enables no other extension.Also stops extending the health-check address whitelist when no
serversection is configured. The section is optional, but it was dereferenced unconditionally, so a process serving no API exited before starting.Follow-ups and exclusions
deleteDatawhen a localized version is deleted__metastoreand Vault entitiestransitionInProgresssuppressionlocationConfig.jsonhas to stop being{}isCRRLocationreads it. Harmless for metadata-only D/R, where every lookup is false and every object is cold, which is the correct answer therelib/util/locations.jsreplicationInfoMongoProcessorMetricshas no such gauge. Every metric here is alsoingestion*-prefixed, which will read oddly on a D/R dashboardUnblocks ZKOP-562, whose generated configuration sets
mode,mongodbandauth, and which cannot start a mongo-processor without this.Issue: BB-811