From 6e783fdaccd25b7c4a795f430c71341947f6fc38 Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Mon, 24 Aug 2026 09:31:45 +0200 Subject: [PATCH 1/5] Trigger localization of clean room objects In a clean room, objects are created locally but their metadata still points at the source cluster's location: the data itself has not been copied over yet. Something has to notice those objects and ask for the data to be pulled in. The queue populator is the natural place for it, since bootstrap, re-bootstrap and streamed updates all go through the same oplog. When an object lands on a location flagged isCRR, publish a copyLocation action on the data mover topic and let the existing data mover + transition merge pipeline do the actual copy. This is unrelated to replicationInfo, which describes replication of a *local* object to remote sites, so the check sits before any replication condition. Activation is simply the presence of extensions.replication.localization.toLocation in the config; without it nothing changes. The populator only knows the object owner's canonical id, and resolving an account id per entry would throttle the whole populator, so the lookup is deferred to the transition processor, which now resolves it once up front and stamps it back on the action - the same way the garbage collector already does. Issue: BB-814 --- conf/locationConfig.json | 7 + docker-entrypoint.sh | 10 + .../lifecycle/LifecycleConfigValidator.js | 1 + .../LifecycleObjectProcessor.js | 56 ++++- .../tasks/LifecycleUpdateTransitionTask.js | 42 +++- .../replication/ReplicationConfigValidator.js | 7 + .../replication/ReplicationQueuePopulator.js | 136 ++++++++++- lib/Config.js | 3 + lib/queuePopulator/QueuePopulator.js | 16 ++ .../LifecycleUpdateTransitionTask.spec.js | 43 ++++ tests/unit/mocks.js | 17 ++ .../ReplicationQueuePopulator.spec.js | 224 ++++++++++++++++++ 12 files changed, 553 insertions(+), 9 deletions(-) diff --git a/conf/locationConfig.json b/conf/locationConfig.json index 8ba3dc334c..dceb31ddac 100644 --- a/conf/locationConfig.json +++ b/conf/locationConfig.json @@ -42,5 +42,12 @@ "legacyAwsBehavior": false, "isCold": true, "details": {} + }, + "location-crr-source": { + "type": "scality", + "objectId": "location-crr-source", + "legacyAwsBehavior": false, + "isCRR": true, + "details": {} } } diff --git a/docker-entrypoint.sh b/docker-entrypoint.sh index 883146ff62..42aea6bdec 100755 --- a/docker-entrypoint.sh +++ b/docker-entrypoint.sh @@ -154,6 +154,16 @@ if [[ "$EXTENSIONS_REPLICATION_DEST_BOOTSTRAPLIST" ]]; then fi fi +# Clean room: localize objects whose data still lives on the source (isCRR) +# location. Setting the target location enables the trigger. +if [[ "$EXTENSIONS_REPLICATION_LOCALIZATION_TO_LOCATION" ]]; then + JQ_FILTERS_CONFIG="$JQ_FILTERS_CONFIG | .extensions.replication.localization.toLocation=\"$EXTENSIONS_REPLICATION_LOCALIZATION_TO_LOCATION\"" +fi + +if [[ "$EXTENSIONS_REPLICATION_LOCALIZATION_RESULTS_TOPIC" ]]; then + JQ_FILTERS_CONFIG="$JQ_FILTERS_CONFIG | .extensions.replication.localization.resultsTopic=\"$EXTENSIONS_REPLICATION_LOCALIZATION_RESULTS_TOPIC\"" +fi + # START Retry config # AWS_S3 diff --git a/extensions/lifecycle/LifecycleConfigValidator.js b/extensions/lifecycle/LifecycleConfigValidator.js index b6487862e5..0d5a1c1a0b 100644 --- a/extensions/lifecycle/LifecycleConfigValidator.js +++ b/extensions/lifecycle/LifecycleConfigValidator.js @@ -71,6 +71,7 @@ const joiSchema = joi.object({ concurrency: joi.number().greater(0).default(10), maxQueued: joi.number().greater(0).default(MAX_QUEUED_DEFAULT), probeServer: probeServerJoi.default(), + vaultAdmin: hostPortJoi, circuitBreaker: joi.object().optional(), }, coldStorageArchiveTopicPrefix: joi.string().default('cold-archive-req-'), diff --git a/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js b/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js index 794449bb52..39a8b8ae03 100644 --- a/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js +++ b/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js @@ -7,6 +7,9 @@ const BackbeatConsumerManager = require('../../../lib/BackbeatConsumerManager'); const ActionQueueEntry = require('../../../lib/models/ActionQueueEntry'); const ClientManager = require('../../../lib/clients/ClientManager'); const BackbeatTask = require('../../../lib/tasks/BackbeatTask'); +const VaultClientWrapper = require('../../utils/VaultClientWrapper'); +const { AccountIdCache } = require('../../utils/AccountIdCache'); +const { authTypeAssumeRole } = require('../../../lib/constants'); const logIdFromType = { 'object-processor': 'Backbeat:Lifecycle:ObjectProcessor', @@ -61,9 +64,55 @@ class LifecycleObjectProcessor extends EventEmitter { transport, }, this._log); + this.vaultClientWrapper = new VaultClientWrapper( + `lifecycle:${this.getProcessorType()}`, + this._processConfig.vaultAdmin, + this.getAuthConfig(this._lcConfig), + this._log, + ); + this._accountIdCache = new AccountIdCache( + this._processConfig.concurrency); + this.retryWrapper = new BackbeatTask(this._processConfig.retry); } + /** + * Resolve the account id of a canonical id. Actions published by the + * lifecycle conductor already carry the account id; those published by the + * queue populator (clean room localization) only know the canonical id. + * @param {String} ownerId - canonical id of the object owner + * @param {Logger} log - logger instance + * @param {Function} cb - callback: cb(err, accountId) + * @return {undefined} + */ + getAccountId(ownerId, log, cb) { + if (this.getAuthConfig(this._lcConfig).type !== authTypeAssumeRole) { + log.debug('skipping: not assume role auth type'); + return process.nextTick(cb); + } + + if (this._accountIdCache.isKnown(ownerId)) { + return process.nextTick(cb, null, this._accountIdCache.get(ownerId)); + } + + return this.vaultClientWrapper.getAccountId(ownerId, (err, accountId) => { + if (err) { + if (err.NoSuchEntity) { + log.error('canonical id does not exist', { error: err, ownerId }); + this._accountIdCache.miss(ownerId); + } else { + log.error('could not get account id', { error: err, ownerId }); + } + return cb(err); + } + + this._accountIdCache.set(ownerId, accountId); + this._accountIdCache.expireOldest(); + + return cb(null, accountId); + }); + } + getProcessorType() { return 'object-processor'; } @@ -130,6 +179,9 @@ class LifecycleObjectProcessor extends EventEmitter { start(done) { this.clientManager.initSTSConfig(); this.clientManager.initCredentialsManager(); + if (this.getAuthConfig(this._lcConfig).type === authTypeAssumeRole) { + this.vaultClientWrapper.init(); + } this._setupConsumers(done); } @@ -225,12 +277,14 @@ class LifecycleObjectProcessor extends EventEmitter { this.clientManager.getBackbeatClient.bind(this.clientManager), getBackbeatMetadataProxy: this.clientManager.getBackbeatMetadataProxy.bind(this.clientManager), + getAccountId: this.getAccountId.bind(this), logger: this._log, }; } isReady() { - return this._consumers && this._consumers.isReady(); + return this._consumers && this._consumers.isReady() && + this.vaultClientWrapper.tempCredentialsReady(); } } diff --git a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js index 4f57c33c7f..a3450f48ad 100644 --- a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js +++ b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js @@ -252,6 +252,34 @@ class LifecycleUpdateTransitionTask extends BackbeatTask { ], done); } + /** + * Actions published by the lifecycle conductor carry the account id; + * those published by the queue populator (clean room localization) only + * know the object owner's canonical id. Resolve it once, up-front, so the + * rest of the task - and the garbage collection entry it emits - can use + * `target.accountId` as usual. + * @param {ActionQueueEntry} entry - action entry to execute + * @param {Logger} log - logger instance + * @param {Function} cb - callback function + * @return {undefined} + */ + _resolveAccountId(entry, log, cb) { + const { accountId, owner } = this.getTargetAttribute(entry); + if (accountId || !owner) { + return process.nextTick(cb); + } + + log.debug('no account id in entry, resolving from canonical id', + { owner }); + return this.getAccountId(owner, log, (err, resolvedAccountId) => { + if (err) { + return cb(err); + } + entry.setAttribute('target.accountId', resolvedAccountId); + return cb(); + }); + } + /** * * @param {ActionQueueEntry} entry - action entry to execute @@ -268,11 +296,17 @@ class LifecycleUpdateTransitionTask extends BackbeatTask { lastModified: 'target.lastModified', }); log.addDefaultFields(entry.getLogInfo()); - if (entry.getStatus() === 'success') { - return this.handleSuccessfullTransition(entry, log, done); - } - return this.handleFailedTransition(entry, log, done); + return this._resolveAccountId(entry, log, err => { + if (err) { + return done(err); + } + if (entry.getStatus() === 'success') { + return this.handleSuccessfullTransition(entry, log, done); + } + + return this.handleFailedTransition(entry, log, done); + }); } } diff --git a/extensions/replication/ReplicationConfigValidator.js b/extensions/replication/ReplicationConfigValidator.js index 8f748c8077..47d1c26636 100644 --- a/extensions/replication/ReplicationConfigValidator.js +++ b/extensions/replication/ReplicationConfigValidator.js @@ -139,6 +139,13 @@ const joiSchema = joi.object({ probeServer: probeServerPerSite, }).optional(), objectSizeMetrics: joi.array().items(joi.number()).default(OBJECT_SIZE_METRICS), + // Clean room: localization of objects whose data still lives on the source + // (isCRR) location. Enabled by setting `toLocation`. + localization: joi.object({ + toLocation: joi.string().required(), + resultsTopic: joi.string() + .default('backbeat-lifecycle-transition-tasks'), + }).optional(), }); /** diff --git a/extensions/replication/ReplicationQueuePopulator.js b/extensions/replication/ReplicationQueuePopulator.js index 22c31ad0bb..e8d9b3234e 100644 --- a/extensions/replication/ReplicationQueuePopulator.js +++ b/extensions/replication/ReplicationQueuePopulator.js @@ -1,18 +1,25 @@ const { isMasterKey } = require('arsenal').versioning; +const { encode } = require('arsenal').versioning.VersionID; const { usersBucket, mpuBucketPrefix } = require('arsenal').constants; const QueuePopulatorExtension = require('../../lib/queuePopulator/QueuePopulatorExtension'); const ObjectQueueEntry = require('../../lib/models/ObjectQueueEntry'); +const ReplicationAPI = require('./ReplicationAPI'); const locationsConfig = require('../../conf/locationConfig.json') || {}; const safeJsonParse = require('../../lib/util/safeJsonParse'); const { traceHeadersFromEntry } = require('arsenal/build/lib/tracing').kafka; +const TRANSITION_ATTEMPT_MD = 'x-amz-meta-scal-s3-transition-attempt'; + class ReplicationQueuePopulator extends QueuePopulatorExtension { constructor(params) { super(params); this.repConfig = params.config; this.metricsHandler = params.metricsHandler; + // Clean room: when set, objects whose data still lives on the source + // (isCRR) location are queued for localization instead of replication. + this.localizationConfig = params.config.localization; } filter(entry) { @@ -73,6 +80,17 @@ class ReplicationQueuePopulator extends QueuePopulatorExtension { if (sanityCheckRes) { return; } + const dataStoreName = queueEntry.getDataStoreName(); + const locationConfig = (dataStoreName && locationsConfig[dataStoreName]) + || {}; + // Clean room: the object data still lives on the source (isCRR) + // location and first needs to be localized. This is unrelated to + // replicationInfo, which tracks replication of a *local* object to + // remote sites, hence the check before any replication condition. + if (locationConfig.isCRR && this.localizationConfig) { + this._publishLocalizationAction(entry, queueEntry, value); + return; + } // Allow a non-versioned object if being replicated from an NFS bucket. // Or if the master key is of a non versioned object if (!this._entryCanBeReplicated(queueEntry)) { @@ -81,11 +99,8 @@ class ReplicationQueuePopulator extends QueuePopulatorExtension { if (queueEntry.getReplicationStatus() !== 'PENDING') { return; } - const dataStoreName = queueEntry.getDataStoreName(); - const isObjectCold = dataStoreName && locationsConfig[dataStoreName] - && locationsConfig[dataStoreName].isCold; // We do not replicate cold objects. - if (isObjectCold) { + if (locationConfig.isCold) { return; } @@ -124,6 +139,119 @@ class ReplicationQueuePopulator extends QueuePopulatorExtension { traceHeaders); } + /** + * Queue a copyLocation action for an object whose data still lives on the + * source (isCRR) location, so the data mover copies it to the local + * location and the transition processor merges the new location back into + * the object metadata. + * + * Duplicates are expected (and harmless): the same object may show up + * several times in the oplog, and the copy is idempotent. + * + * @param {Object} entry - raw metadata log entry + * @param {ObjectQueueEntry} queueEntry - parsed entry + * @param {Object} value - parsed entry metadata + * @return {undefined} + */ + _publishLocalizationAction(entry, queueEntry, value) { + // Clean room buckets are versioned: the master key is repaired by the + // metadata layer once the version has been localized. + if (isMasterKey(queueEntry.getObjectVersionedKey())) { + return; + } + if (queueEntry.getIsDeleteMarker()) { + return; + } + const locations = queueEntry.getLocation(); + if (!locations || locations.length === 0) { + // Empty objects hold no data, there is nothing to localize. Any + // other object without location information is inconsistent. + if (queueEntry.getContentLength() > 0) { + this.log.error( + 'non-empty object without location, skipping localization', + { + method: 'ReplicationQueuePopulator.' + + '_publishLocalizationAction', + ...queueEntry.getLogInfo(), + dataStoreName: queueEntry.getDataStoreName(), + contentLength: queueEntry.getContentLength(), + }); + } + return; + } + + const bucket = queueEntry.getBucket(); + const objectKey = queueEntry.getObjectKey(); + const contentLength = queueEntry.getContentLength(); + const action = ReplicationAPI.createCopyLocationAction({ + bucketName: bucket, + objectKey, + owner: queueEntry.getOwnerId(), + versionId: value.versionId ? encode(value.versionId) : undefined, + eTag: `"${queueEntry.getContentMd5()}"`, + lastModified: queueEntry.getLastModified(), + toLocation: this.localizationConfig.toLocation, + originLabel: 'localization', + fromLocation: queueEntry.getDataStoreName(), + contentLength, + resultsTopic: this.localizationConfig.resultsTopic, + transitionTime: new Date( + entry.overheadFields?.commitTimestamp ?? Date.now() + ).toISOString(), + attempt: this._getTransitionAttempt(queueEntry), + }); + // 'transition' is what the lifecycle transition processor dispatches + // on to pick up the copyLocation result. + action.addContext({ + origin: 'localization', + ruleType: 'transition', + bucketName: bucket, + objectKey, + versionId: value.versionId, + }); + action.setAttribute('source', { + bucket, + objectKey, + storageClass: queueEntry.getDataStoreName(), + }); + + this.metricsHandler.localizationBytes( + entry.logReader.getMetricLabels(), + contentLength + ); + this.metricsHandler.localizationObjects( + entry.logReader.getMetricLabels() + ); + + this.log.trace('publishing object localization entry', + { entry: queueEntry.getLogInfo() }); + this.publish(ReplicationAPI.getDataMoverTopic(), + `${bucket}/${objectKey}`, + action.toKafkaMessage(), + undefined, + traceHeadersFromEntry(value)); + } + + /** + * Number of times the data mover already tried to copy this object. The + * transition processor bumps the counter on failure, which produces a new + * oplog entry and re-triggers the copy. + * @param {ObjectQueueEntry} queueEntry - parsed entry + * @return {Number|undefined} attempt count, if any + */ + _getTransitionAttempt(queueEntry) { + const umd = queueEntry.getUserMetadata(); + if (!umd) { + return undefined; + } + const { error, result } = safeJsonParse(umd); + if (error) { + return undefined; + } + const attempt = Number.parseInt(result[TRANSITION_ATTEMPT_MD], 10); + return Number.isInteger(attempt) ? attempt : undefined; + } + /** * Filter if the entry is considered a valid master key entry. * There is a case where a single null entry looks like a master key and diff --git a/lib/Config.js b/lib/Config.js index 57a9aaa880..4a1d18646f 100644 --- a/lib/Config.js +++ b/lib/Config.js @@ -89,6 +89,9 @@ class Config extends EventEmitter { if (backbeatSupportsTransition && !replicationConfig.dataMoverTopic) { throw new Error('dataMoverTopic is required when lifecycle transitions is supported'); } + if (replicationConfig?.localization && !replicationConfig.dataMoverTopic) { + throw new Error('dataMoverTopic is required when localization is enabled'); + } const destination = parsedConfig.extensions?.replication?.destination; this.bootstrapList = destination?.bootstrapList?.map(endpoint => { diff --git a/lib/queuePopulator/QueuePopulator.js b/lib/queuePopulator/QueuePopulator.js index 8b63e3d055..d2a716bf15 100644 --- a/lib/queuePopulator/QueuePopulator.js +++ b/lib/queuePopulator/QueuePopulator.js @@ -79,12 +79,26 @@ const notificationEvent = ZenkoMetrics.createCounter({ help: 'Total number of oplog events processed by notification extension', }); +const localizationObjectMetrics = ZenkoMetrics.createCounter({ + name: 's3_backbeat_populator_localization_objects_total', + help: 'Total objects queued for clean room localization', + labelNames: metricLabels, +}); + +const localizationByteMetrics = ZenkoMetrics.createCounter({ + name: 's3_backbeat_populator_localization_bytes_total', + help: 'Total number of bytes queued for clean room localization', + labelNames: metricLabels, +}); + /** * Contains methods to incrememt different metrics * @typedef {Object} MetricsHandler * @property {CounterInc} messages - Increments the message metric * @property {CounterInc} objects - Increments the objects metric * @property {CounterInc} bytes - Increments the bytes metric + * @property {CounterInc} localizationObjects - Increments the localized objects metric + * @property {CounterInc} localizationBytes - Increments the localized bytes metric * @property {GaugeSet} logReadOffset - Set the log read offset metric * @property {GaugeSet} logSize - Set the log size metric */ @@ -92,6 +106,8 @@ const metricsHandler = { messages: wrapCounterInc(messageMetrics, {}), objects: wrapCounterInc(objectMetrics, {}), bytes: wrapCounterInc(byteMetrics, {}), + localizationObjects: wrapCounterInc(localizationObjectMetrics, {}), + localizationBytes: wrapCounterInc(localizationByteMetrics, {}), logReadOffset: wrapGaugeSet(logReadOffsetMetric, {}), logSize: wrapGaugeSet(logSizeMetric, {}), logTimestamp: wrapGaugeSet(logTimestamp, {}), diff --git a/tests/unit/lifecycle/LifecycleUpdateTransitionTask.spec.js b/tests/unit/lifecycle/LifecycleUpdateTransitionTask.spec.js index 61748e96e3..bebdc9c572 100644 --- a/tests/unit/lifecycle/LifecycleUpdateTransitionTask.spec.js +++ b/tests/unit/lifecycle/LifecycleUpdateTransitionTask.spec.js @@ -169,4 +169,47 @@ describe('LifecycleUpdateTransitionTask', () => { done(); }); }); + + // clean room localization actions are published by the queue populator, + // which only knows the object owner's canonical id + describe('account id resolution', () => { + it('should not look up the account id when the entry has one', done => { + actionEntry.setAttribute('target.accountId', '000000000042'); + actionEntry.setAttribute('target.owner', 'some-canonical-id'); + task.processActionEntry(actionEntry, err => { + assert.ifError(err); + assert.strictEqual(objectProcessor.accountIdLookups, 0); + done(); + }); + }); + + it('should resolve the account id from the owner canonical id', done => { + objectProcessor.setAccountId('some-canonical-id', '000000000042'); + actionEntry.setAttribute('target.owner', 'some-canonical-id'); + task.processActionEntry(actionEntry, err => { + assert.ifError(err); + assert.strictEqual(objectProcessor.accountIdLookups, 1); + assert.strictEqual( + actionEntry.getAttribute('target.accountId'), + '000000000042'); + // the garbage collection entry must not resolve it again + const receivedGcEntry = gcProducer.getReceivedEntry(); + assert.strictEqual( + receivedGcEntry.getAttribute('target.accountId'), + '000000000042'); + done(); + }); + }); + + it('should fail the entry when the account id cannot be resolved', + done => { + actionEntry.setAttribute('target.owner', 'unknown-canonical-id'); + task.processActionEntry(actionEntry, err => { + assert(err); + assert.strictEqual( + backbeatMetadataProxyClient.getReceivedMd(), null); + done(); + }); + }); + }); }); diff --git a/tests/unit/mocks.js b/tests/unit/mocks.js index 2b9096f88a..99ebb0871d 100644 --- a/tests/unit/mocks.js +++ b/tests/unit/mocks.js @@ -1,4 +1,5 @@ const assert = require('assert'); +const { errors } = require('arsenal'); const { ObjectMD } = require('arsenal').models; class GarbageCollectorProducerMock { @@ -157,6 +158,21 @@ class ProcessorMock { this.coldProducer = coldProducer; this._gcConfig = gcConfig; this.logger = logger; + this.accountIds = {}; + this.accountIdLookups = 0; + } + + setAccountId(ownerId, accountId) { + this.accountIds[ownerId] = accountId; + } + + getAccountId(ownerId, log, cb) { + this.accountIdLookups += 1; + const accountId = this.accountIds[ownerId]; + if (!accountId) { + return process.nextTick(cb, errors.NoSuchEntity); + } + return process.nextTick(cb, null, accountId); } getStateVars() { @@ -170,6 +186,7 @@ class ProcessorMock { getBackbeatClient: () => this.backbeatClient, getBackbeatMetadataProxy: () => this.backbeatMetadataProxy, getS3Client: () => this.s3Client, + getAccountId: this.getAccountId.bind(this), }; } } diff --git a/tests/unit/replication/ReplicationQueuePopulator.spec.js b/tests/unit/replication/ReplicationQueuePopulator.spec.js index cc384b6950..bb9a95d16f 100644 --- a/tests/unit/replication/ReplicationQueuePopulator.spec.js +++ b/tests/unit/replication/ReplicationQueuePopulator.spec.js @@ -1,8 +1,11 @@ const assert = require('assert'); const sinon = require('sinon'); +const { encode } = require('arsenal').versioning.VersionID; + const ReplicationQueuePopulator = require('../../../extensions/replication/ReplicationQueuePopulator'); +const ReplicationAPI = require('../../../extensions/replication/ReplicationAPI'); const fakeLogger = require('../../utils/fakeLogger'); @@ -382,3 +385,224 @@ describe('replication queue populator', () => { assert.deepStrictEqual(rqp.getState(), {}); }); }); + +/** + * Records every published message, whatever the topic, so localization + * entries (data mover topic) can be inspected. + * @class + */ +class RecordingQueuePopulatorMock extends ReplicationQueuePopulator { + constructor(params) { + super(params); + + this.published = []; + } + + publish(topic, key, message) { + this.published.push({ topic, key, message }); + } +} + +describe('replication queue populator: clean room localization', () => { + const CRR_LOCATION = 'location-crr-source'; + const LOCAL_LOCATION = 'us-east-1'; + const RESULTS_TOPIC = 'test-transition-results'; + const VERSION_ID = '98477724999464999999RG001 1.30.12'; + const VERSIONED_KEY = `a-test-key\u0000${VERSION_ID}`; + + let params; + let rqp; + + function makeValue(overrides = {}) { + return JSON.stringify({ + ...kafkaValue, + dataStoreName: CRR_LOCATION, + location: [{ + key: 'some-data-key', + size: 128, + start: 0, + dataStoreName: CRR_LOCATION, + dataStoreETag: '1:d41d8cd98f00b204e9800118ecf8427e', + }], + ...overrides, + }); + } + + function makeEntry(value, key = VERSIONED_KEY) { + return { + type: 'put', + bucket: 'test-bucket-source', + key, + value, + overheadFields: { commitTimestamp: '2024-05-06T10:11:12.000Z' }, + logReader: { getMetricLabels: stubMetricLabels() }, + }; + } + + beforeEach(() => { + params = { + config: { + topic: TOPIC, + localization: { + toLocation: LOCAL_LOCATION, + resultsTopic: RESULTS_TOPIC, + }, + }, + logger: fakeLogger, + metricsHandler: { + bytes: sinon.spy(), + objects: sinon.spy(), + localizationBytes: sinon.spy(), + localizationObjects: sinon.spy(), + }, + }; + rqp = new RecordingQueuePopulatorMock(params); + }); + + it('should publish a copyLocation action for a non-localized object', () => { + rqp._filterKeyOp(makeEntry(makeValue())); + + assert.strictEqual(rqp.published.length, 1); + const [{ topic, key, message }] = rqp.published; + assert.strictEqual(topic, ReplicationAPI.getDataMoverTopic()); + assert.strictEqual(key, 'test-bucket-source/a-test-key'); + + const action = JSON.parse(message); + assert.strictEqual(action.action, 'copyLocation'); + assert.strictEqual(action.toLocation, LOCAL_LOCATION); + assert.strictEqual(action.resultsTopic, RESULTS_TOPIC); + assert.strictEqual(action.contextInfo.ruleType, 'transition'); + assert.strictEqual(action.contextInfo.origin, 'localization'); + assert.deepStrictEqual(action.target, { + owner: kafkaValue['owner-id'], + bucket: 'test-bucket-source', + key: 'a-test-key', + version: encode(VERSION_ID), + eTag: `"${kafkaValue['content-md5']}"`, + lastModified: kafkaValue['last-modified'], + }); + // resolved by the transition processor, not by the populator + assert.strictEqual(action.target.accountId, undefined); + assert.deepStrictEqual(action.source, { + bucket: 'test-bucket-source', + objectKey: 'a-test-key', + storageClass: CRR_LOCATION, + }); + assert.strictEqual(action.metrics.fromLocation, CRR_LOCATION); + assert.strictEqual(action.metrics.contentLength, 128); + assert.strictEqual(action.metrics.transitionTime, + '2024-05-06T10:11:12.000Z'); + }); + + it('should account localized objects and bytes', () => { + rqp._filterKeyOp(makeEntry(makeValue())); + + sinon.assert.calledOnceWithExactly( + params.metricsHandler.localizationBytes, labels, 128); + sinon.assert.calledOnceWithExactly( + params.metricsHandler.localizationObjects, labels); + sinon.assert.notCalled(params.metricsHandler.objects); + }); + + // localization is about where the data lives, forward replication is + // about where it has been copied to: the two are independent. + ['PENDING', 'COMPLETED', 'FAILED'].forEach(status => { + it(`should publish regardless of replication status ${status}`, () => { + const value = makeValue({ + replicationInfo: { ...repInfo, status }, + }); + rqp._filterKeyOp(makeEntry(value)); + + assert.strictEqual(rqp.published.length, 1); + }); + }); + + it('should publish when there is no replication configured', () => { + const value = makeValue({ replicationInfo: null }); + rqp._filterKeyOp(makeEntry(value)); + + assert.strictEqual(rqp.published.length, 1); + }); + + it('should propagate the transition attempt count', () => { + const value = makeValue({ + 'x-amz-meta-scal-s3-transition-attempt': '3', + }); + rqp._filterKeyOp(makeEntry(value)); + + assert.strictEqual(rqp.published.length, 1); + const action = JSON.parse(rqp.published[0].message); + assert.strictEqual(action.target.attempt, 3); + }); + + it('should not set an attempt count for a first copy', () => { + rqp._filterKeyOp(makeEntry(makeValue())); + + const action = JSON.parse(rqp.published[0].message); + assert.strictEqual(action.target.attempt, undefined); + }); + + it('should skip master keys', () => { + rqp._filterKeyOp(makeEntry(makeValue(), 'a-test-key')); + + assert.strictEqual(rqp.published.length, 0); + }); + + it('should skip delete markers', () => { + const value = makeValue({ isDeleteMarker: true }); + rqp._filterKeyOp(makeEntry(value)); + + assert.strictEqual(rqp.published.length, 0); + }); + + it('should skip empty objects', () => { + const value = makeValue({ + 'location': null, + 'content-length': 0, + }); + rqp._filterKeyOp(makeEntry(value)); + + assert.strictEqual(rqp.published.length, 0); + }); + + it('should skip and report non-empty objects without location', () => { + const errorSpy = sinon.spy(rqp.log, 'error'); + const value = makeValue({ location: null }); + rqp._filterKeyOp(makeEntry(value)); + + assert.strictEqual(rqp.published.length, 0); + sinon.assert.calledOnce(errorSpy); + errorSpy.restore(); + }); + + // partial oplog projections (change stream `update` events) may not carry + // the location: they cannot be localized, and behave as before. + it('should not localize entries with no dataStoreName', () => { + const value = makeValue({ dataStoreName: undefined }); + rqp._filterKeyOp(makeEntry(value)); + + sinon.assert.notCalled(params.metricsHandler.localizationObjects); + assert.strictEqual( + rqp.published.filter( + p => p.topic === ReplicationAPI.getDataMoverTopic()).length, + 0); + }); + + it('should not localize objects on a regular location', () => { + const value = makeValue({ dataStoreName: LOCAL_LOCATION }); + rqp._filterKeyOp(makeEntry(value)); + + assert.strictEqual(rqp.published.length, 1); + assert.strictEqual(rqp.published[0].topic, TOPIC); + sinon.assert.notCalled(params.metricsHandler.localizationObjects); + }); + + it('should fall back to replication when localization is disabled', () => { + delete params.config.localization; + rqp = new RecordingQueuePopulatorMock(params); + rqp._filterKeyOp(makeEntry(makeValue())); + + assert.strictEqual(rqp.published.length, 1); + assert.strictEqual(rqp.published[0].topic, TOPIC); + }); +}); From 7139bf2287c49a5f29dd36313a8b1762158da4e7 Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Mon, 24 Aug 2026 09:48:34 +0200 Subject: [PATCH 2/5] Fail account id resolution on a cached vault miss AccountIdCache.isKnown() is also true for cached misses, where get() returns undefined. The localization flow would then carry on with target.accountId unset and surface a confusing "failed to get backbeat client" InternalError instead of the NoSuchEntity that actually caused it, making a bad canonical id hard to diagnose in production. Treat a cached miss as the error it is, and only stamp the resolved account id onto the entry when there is one. Issue: BB-814 --- .../LifecycleObjectProcessor.js | 10 ++- .../tasks/LifecycleUpdateTransitionTask.js | 4 +- ...LifecycleObjectTransitionProcessor.spec.js | 89 +++++++++++++++++++ 3 files changed, 101 insertions(+), 2 deletions(-) diff --git a/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js b/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js index 39a8b8ae03..14e1ee5181 100644 --- a/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js +++ b/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js @@ -2,6 +2,7 @@ const { EventEmitter } = require('events'); const Logger = require('werelogs').Logger; +const { errors } = require('arsenal'); const BackbeatConsumerManager = require('../../../lib/BackbeatConsumerManager'); const ActionQueueEntry = require('../../../lib/models/ActionQueueEntry'); @@ -91,7 +92,14 @@ class LifecycleObjectProcessor extends EventEmitter { return process.nextTick(cb); } - if (this._accountIdCache.isKnown(ownerId)) { + // A cached miss must fail like a fresh lookup would: `isKnown()` is also + // true for misses, and `get()` would then hand back `undefined`. + if (this._accountIdCache.isMiss(ownerId)) { + log.error('canonical id does not exist (cached)', { ownerId }); + return process.nextTick(cb, errors.NoSuchEntity); + } + + if (this._accountIdCache.has(ownerId)) { return process.nextTick(cb, null, this._accountIdCache.get(ownerId)); } diff --git a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js index a3450f48ad..2fe958e045 100644 --- a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js +++ b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js @@ -275,7 +275,9 @@ class LifecycleUpdateTransitionTask extends BackbeatTask { if (err) { return cb(err); } - entry.setAttribute('target.accountId', resolvedAccountId); + if (resolvedAccountId) { + entry.setAttribute('target.accountId', resolvedAccountId); + } return cb(); }); } diff --git a/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js b/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js index aa51388517..94b8c5b417 100644 --- a/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js +++ b/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js @@ -1,5 +1,6 @@ const assert = require('assert'); const sinon = require('sinon'); +const { errors } = require('arsenal'); const config = require('../../config.json'); const BackbeatTask = require('../../../lib/tasks/BackbeatTask'); const LifecycleObjectTransitionProcessor = @@ -125,4 +126,92 @@ describe('LifecycleObjectTransitionProcessor', () => { }); }); }); + + describe('getAccountId', () => { + const ownerId = 'canonical-id-1'; + const accountId = '834789881858'; + let processor; + let log; + + beforeEach(() => { + processor = new LifecycleObjectTransitionProcessor( + config.zookeeper, + config.kafka, + { + ...config.extensions.lifecycle, + transitionProcessor: { + ...config.extensions.lifecycle.transitionProcessor, + auth: { type: 'assumeRole', roleName: 'role' }, + }, + }, + config.s3, + ); + log = { debug: () => {}, error: () => {} }; + }); + + afterEach(() => { + sinon.restore(); + }); + + it('should skip the lookup when auth type is not assume role', done => { + const spy = sinon.spy(objectProcessor.vaultClientWrapper, 'getAccountId'); + objectProcessor.getAccountId(ownerId, log, (err, id) => { + assert.ifError(err); + assert.strictEqual(id, undefined); + assert.strictEqual(spy.callCount, 0); + done(); + }); + }); + + it('should resolve through vault and cache the result', done => { + const stub = sinon.stub(processor.vaultClientWrapper, 'getAccountId') + .yields(null, accountId); + + processor.getAccountId(ownerId, log, (err, id) => { + assert.ifError(err); + assert.strictEqual(id, accountId); + assert.strictEqual(stub.callCount, 1); + + processor.getAccountId(ownerId, log, (err2, id2) => { + assert.ifError(err2); + assert.strictEqual(id2, accountId); + assert.strictEqual(stub.callCount, 1); + done(); + }); + }); + }); + + it('should fail on a cached miss instead of returning no account id', done => { + const stub = sinon.stub(processor.vaultClientWrapper, 'getAccountId') + .yields(errors.NoSuchEntity); + + processor.getAccountId(ownerId, log, err => { + assert(err.NoSuchEntity); + assert.strictEqual(stub.callCount, 1); + + // the miss is cached, but must still surface as an error + processor.getAccountId(ownerId, log, (err2, id2) => { + assert(err2.NoSuchEntity); + assert.strictEqual(id2, undefined); + assert.strictEqual(stub.callCount, 1); + done(); + }); + }); + }); + + it('should propagate other vault errors without caching them', done => { + const stub = sinon.stub(processor.vaultClientWrapper, 'getAccountId') + .yields(errors.InternalError); + + processor.getAccountId(ownerId, log, err => { + assert(err.InternalError); + + processor.getAccountId(ownerId, log, err2 => { + assert(err2.InternalError); + assert.strictEqual(stub.callCount, 2); + done(); + }); + }); + }); + }); }); From 943cbb0753999b55e225ca4c0cb4cc0f6f484b93 Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Mon, 24 Aug 2026 10:20:43 +0200 Subject: [PATCH 3/5] Only set up the vault client where account ids are resolved Initialising the vault client wrapper for every processor using assume role auth broke the expiration processor, which has no vaultAdmin in its config: start() threw on the missing host, and isReady() would have waited forever on credentials nobody was fetching. Only the transition processor receives actions without an account id, so gate both on a vault endpoint actually being configured, and fail the lookup explicitly if one is ever requested without it. Also add the new clean room location to the circuit breaker probe expectations, which enumerate every configured location. Issue: BB-814 --- .../LifecycleObjectProcessor.js | 24 +++++++++++-- .../lifecycle/CircuitBreakerGroup.spec.js | 10 ++++++ ...LifecycleObjectTransitionProcessor.spec.js | 34 +++++++++++++++++++ 3 files changed, 66 insertions(+), 2 deletions(-) diff --git a/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js b/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js index 14e1ee5181..5352e58b68 100644 --- a/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js +++ b/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js @@ -77,6 +77,19 @@ class LifecycleObjectProcessor extends EventEmitter { this.retryWrapper = new BackbeatTask(this._processConfig.retry); } + /** + * Whether this processor can resolve canonical ids through Vault. Only the + * transition processor receives actions published without an account id + * (clean room localization), and only it is configured with a Vault admin + * endpoint - so leave the Vault client alone everywhere else. + * @return {Boolean} true if account id lookups are available + */ + _accountIdLookupEnabled() { + const authConfig = this.getAuthConfig(this._lcConfig); + return authConfig.type === authTypeAssumeRole && + !!(this._processConfig.vaultAdmin || authConfig.vault); + } + /** * Resolve the account id of a canonical id. Actions published by the * lifecycle conductor already carry the account id; those published by the @@ -92,6 +105,12 @@ class LifecycleObjectProcessor extends EventEmitter { return process.nextTick(cb); } + if (!this._accountIdLookupEnabled()) { + log.error('cannot resolve canonical id: no vault endpoint configured'); + return process.nextTick(cb, errors.InternalError.customizeDescription( + 'account id resolution requires a vault endpoint')); + } + // A cached miss must fail like a fresh lookup would: `isKnown()` is also // true for misses, and `get()` would then hand back `undefined`. if (this._accountIdCache.isMiss(ownerId)) { @@ -187,7 +206,7 @@ class LifecycleObjectProcessor extends EventEmitter { start(done) { this.clientManager.initSTSConfig(); this.clientManager.initCredentialsManager(); - if (this.getAuthConfig(this._lcConfig).type === authTypeAssumeRole) { + if (this._accountIdLookupEnabled()) { this.vaultClientWrapper.init(); } this._setupConsumers(done); @@ -292,7 +311,8 @@ class LifecycleObjectProcessor extends EventEmitter { isReady() { return this._consumers && this._consumers.isReady() && - this.vaultClientWrapper.tempCredentialsReady(); + (!this._accountIdLookupEnabled() || + this.vaultClientWrapper.tempCredentialsReady()); } } diff --git a/tests/unit/lifecycle/CircuitBreakerGroup.spec.js b/tests/unit/lifecycle/CircuitBreakerGroup.spec.js index 56e825d455..c8c8737b6f 100644 --- a/tests/unit/lifecycle/CircuitBreakerGroup.spec.js +++ b/tests/unit/lifecycle/CircuitBreakerGroup.spec.js @@ -436,6 +436,11 @@ describe('extractBucketProcessorCircuitBreakerConfigs', () => { '${location}', 'location-dmf-v1', ), + formatProbeConfig( + topicSpecificLocationTemplateProbe, + '${location}', + 'location-crr-source', + ), ], }, global: [], @@ -493,6 +498,11 @@ describe('extractBucketProcessorCircuitBreakerConfigs', () => { '${location}', 'location-dmf-v1', ), + formatProbeConfig( + topicSpecificLocationTemplateProbe, + '${location}', + 'location-crr-source', + ), ], }, global: [], diff --git a/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js b/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js index 94b8c5b417..faa9032839 100644 --- a/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js +++ b/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js @@ -142,6 +142,7 @@ describe('LifecycleObjectTransitionProcessor', () => { transitionProcessor: { ...config.extensions.lifecycle.transitionProcessor, auth: { type: 'assumeRole', roleName: 'role' }, + vaultAdmin: { host: 'localhost', port: 8600 }, }, }, config.s3, @@ -213,5 +214,38 @@ describe('LifecycleObjectTransitionProcessor', () => { }); }); }); + + it('should not touch vault when no vault endpoint is configured', done => { + // assume role auth, but no vaultAdmin and no auth.vault: the + // expiration processor is deployed this way, and must neither + // start a vault client nor be held back by its readiness. + const noVault = new LifecycleObjectTransitionProcessor( + config.zookeeper, + config.kafka, + { + ...config.extensions.lifecycle, + auth: { type: 'assumeRole', roleName: 'role', sts: {} }, + transitionProcessor: { + ...config.extensions.lifecycle.transitionProcessor, + vaultAdmin: undefined, + }, + }, + config.s3, + ); + const spy = sinon.spy(noVault.vaultClientWrapper, 'getAccountId'); + + assert.strictEqual(noVault._accountIdLookupEnabled(), false); + // readiness must not wait on credentials that are never fetched + assert.strictEqual( + noVault.vaultClientWrapper.tempCredentialsReady(), false); + noVault._consumers = { isReady: () => true }; + assert.strictEqual(noVault.isReady(), true); + + noVault.getAccountId(ownerId, log, err => { + assert(err.InternalError); + assert.strictEqual(spy.callCount, 0); + done(); + }); + }); }); }); From 1e1d0083bd3eca9d05d84c32286bd88e08d166e9 Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Mon, 24 Aug 2026 09:51:39 +0200 Subject: [PATCH 4/5] Skip non-localized entries in replication and lifecycle In a clean room, an object's metadata is local but its data still lives on the source cluster: dataStoreName points at an isCRR location. Such a version has nothing local to offer, so the existing workflows must leave it alone until localization has rewritten the metadata. The replication populator already diverted those entries to localization, but only when localization was configured; otherwise they fell through and got replicated, which would copy a location the target cannot read. Skip them in both cases. Nothing is lost: the localization merge produces a new oplog entry with the real location, and replication picks it up then. Lifecycle transitions get the same treatment. Localization is the only valid transition out of an isCRR location and it has its own trigger, so a transition rule matching such a version is simply skipped and the version re-evaluated on a later scan. Issue: BB-816 --- extensions/lifecycle/tasks/LifecycleTask.js | 17 +++- .../replication/ReplicationQueuePopulator.js | 16 ++-- tests/unit/lifecycle/LifecycleTask.spec.js | 84 +++++++++++++++++++ .../ReplicationQueuePopulator.spec.js | 27 +++++- 4 files changed, 133 insertions(+), 11 deletions(-) diff --git a/extensions/lifecycle/tasks/LifecycleTask.js b/extensions/lifecycle/tasks/LifecycleTask.js index 8a83118404..6bf7556f8c 100644 --- a/extensions/lifecycle/tasks/LifecycleTask.js +++ b/extensions/lifecycle/tasks/LifecycleTask.js @@ -31,6 +31,8 @@ const errorTransitionInProgress = errors.InternalError. customizeDescription('transition is currently in progress'); const errorTransitionColdObject = errors.InternalError. customizeDescription('transitioning a cold object is forbidden'); +const errorTransitionNonLocalizedObject = errors.InternalError. + customizeDescription('transitioning a non-localized object is forbidden'); const errorObjectTemporarilyRestored = errors.InternalError. customizeDescription('object temporarily restored'); const errorReplicationInProgress = errors.InternalError. @@ -1270,12 +1272,21 @@ class LifecycleTask extends BackbeatTask { return next(errorReplicationInProgress); } const dataStoreName = objectMD.getDataStoreName(); - const isObjectCold = dataStoreName && locationsConfig[dataStoreName] - && locationsConfig[dataStoreName].isCold; + const locationConfig = (dataStoreName + && locationsConfig[dataStoreName]) || {}; // We do not transition cold objects - if (isObjectCold) { + if (locationConfig.isCold) { return next(errorTransitionColdObject); } + // Clean room: the object data still lives on the source + // (isCRR) location. Localization is the only valid transition + // out of such a location, and it is triggered by its own path + // (the queue populator), not by lifecycle rules. The version + // is simply re-evaluated on a later scan, once localization + // has completed. + if (locationConfig.isCRR) { + return next(errorTransitionNonLocalizedObject); + } // If transition is in progress, do not re-publish entry // to data-mover or cold-archive topic. if (objectMD.getTransitionInProgress()) { diff --git a/extensions/replication/ReplicationQueuePopulator.js b/extensions/replication/ReplicationQueuePopulator.js index e8d9b3234e..17145a86d8 100644 --- a/extensions/replication/ReplicationQueuePopulator.js +++ b/extensions/replication/ReplicationQueuePopulator.js @@ -84,11 +84,17 @@ class ReplicationQueuePopulator extends QueuePopulatorExtension { const locationConfig = (dataStoreName && locationsConfig[dataStoreName]) || {}; // Clean room: the object data still lives on the source (isCRR) - // location and first needs to be localized. This is unrelated to - // replicationInfo, which tracks replication of a *local* object to - // remote sites, hence the check before any replication condition. - if (locationConfig.isCRR && this.localizationConfig) { - this._publishLocalizationAction(entry, queueEntry, value); + // location, there is nothing local to replicate yet, so localize it + // first (when configured) and skip it. Nothing is lost: the + // localization merge rewrites the metadata, which comes back through + // the oplog with the real location and is replicated then. + // This is unrelated to replicationInfo, which tracks replication of a + // *local* object to remote sites, hence the check before any + // replication condition. + if (locationConfig.isCRR) { + if (this.localizationConfig) { + this._publishLocalizationAction(entry, queueEntry, value); + } return; } // Allow a non-versioned object if being replicated from an NFS bucket. diff --git a/tests/unit/lifecycle/LifecycleTask.spec.js b/tests/unit/lifecycle/LifecycleTask.spec.js index 7150bbe147..ae46563b2a 100644 --- a/tests/unit/lifecycle/LifecycleTask.spec.js +++ b/tests/unit/lifecycle/LifecycleTask.spec.js @@ -11,6 +11,7 @@ const LifecycleTask = require( const LifecycleTaskV2 = require( '../../../extensions/lifecycle/tasks/LifecycleTaskV2'); const ActionQueueEntry = require('../../../lib/models/ActionQueueEntry'); +const ReplicationAPI = require('../../../extensions/replication/ReplicationAPI'); const { LifecycleMetrics } = require('../../../extensions/lifecycle/LifecycleMetrics'); const fakeLogger = require('../../utils/fakeLogger'); const { withActiveSpan } = require('../../utils/withActiveSpan'); @@ -2473,6 +2474,89 @@ describe('lifecycle task helper methods', () => { }); }); + describe('_applyTransitionRule', () => { + const CRR_LOCATION = 'location-crr-source'; + const testParams = { + bucket: 'test-bucket', + owner: 'test-owner', + objectKey: 'test-key', + versionId: 'test-version-id', + eTag: '"test-etag"', + lastModified: '2023-01-01T00:00:00.000Z', + site: 'test-site', + accountId: 'test-account-id', + transitionTime: Date.now(), + bucketData: { + target: { + bucket: 'test-bucket', + owner: 'test-owner', + accountId: 'test-account-id', + }, + }, + }; + + let lifecycleTask; + let objectMD; + let sendDataMoverAction; + let putObjectMD; + + function setupObjectMD(dataStoreName) { + objectMD = { + getReplicationStatus: () => 'COMPLETED', + getDataStoreName: () => dataStoreName, + getDataStoreVersionId: () => 'version-123', + getTransitionInProgress: () => false, + getArchive: () => undefined, + getContentLength: () => 1024, + getUserMetadata: () => null, + setTransitionInProgress: sinon.spy(), + setOriginOp: sinon.spy(), + getSerialized: () => '{}', + }; + sinon.stub(lifecycleTask, '_getObjectMD') + .callsFake((params, log, cb) => cb(null, objectMD)); + } + + beforeEach(() => { + lifecycleTask = new LifecycleTask(lp); + lifecycleTask.pausedLocations = new Set(); + lifecycleTask.circuitBreakers = { tripped: () => false }; + lifecycleTask.producer = {}; + lifecycleTask.transitionTasksTopic = 'test-transition-topic'; + sendDataMoverAction = sinon.stub( + ReplicationAPI, 'sendDataMoverAction') + .callsFake((producer, entry, log, cb) => cb()); + putObjectMD = sinon.stub(lifecycleTask, '_putObjectMD') + .callsFake((params, log, cb) => cb()); + }); + + it('should not transition an object still on an isCRR location', + done => { + setupObjectMD(CRR_LOCATION); + + lifecycleTask._applyTransitionRule(testParams, fakeLogger, err => { + assert.strictEqual(err.description, + 'transitioning a non-localized object is forbidden'); + sinon.assert.notCalled(sendDataMoverAction); + sinon.assert.notCalled(objectMD.setTransitionInProgress); + sinon.assert.notCalled(putObjectMD); + done(); + }); + }); + + it('should transition an object on a regular location', done => { + setupObjectMD('us-east-1'); + + lifecycleTask._applyTransitionRule(testParams, fakeLogger, err => { + assert.ifError(err); + sinon.assert.calledOnce(sendDataMoverAction); + sinon.assert.calledOnce(objectMD.setTransitionInProgress); + sinon.assert.calledOnce(putObjectMD); + done(); + }); + }); + }); + describe('_sendObjectAction', () => { it('should emit trigger metrics with the entry location', done => { const lifecycleTask = new LifecycleTask(lp); diff --git a/tests/unit/replication/ReplicationQueuePopulator.spec.js b/tests/unit/replication/ReplicationQueuePopulator.spec.js index bb9a95d16f..baa31e1dc5 100644 --- a/tests/unit/replication/ReplicationQueuePopulator.spec.js +++ b/tests/unit/replication/ReplicationQueuePopulator.spec.js @@ -597,12 +597,33 @@ describe('replication queue populator: clean room localization', () => { sinon.assert.notCalled(params.metricsHandler.localizationObjects); }); - it('should fall back to replication when localization is disabled', () => { + it('should not replicate a non-localized object when localization is ' + + 'disabled', () => { delete params.config.localization; rqp = new RecordingQueuePopulatorMock(params); rqp._filterKeyOp(makeEntry(makeValue())); - assert.strictEqual(rqp.published.length, 1); - assert.strictEqual(rqp.published[0].topic, TOPIC); + assert.strictEqual(rqp.published.length, 0); + }); + + // the data still lives on the source location: there is nothing local to + // replicate, whatever replicationInfo says. + [true, false].forEach(localizationEnabled => { + it('should never replicate a pending non-localized object ' + + `(localization ${localizationEnabled ? 'enabled' : 'disabled'})`, + () => { + if (!localizationEnabled) { + delete params.config.localization; + } + rqp = new RecordingQueuePopulatorMock(params); + const value = makeValue({ + replicationInfo: { ...repInfo, status: 'PENDING' }, + }); + rqp._filterKeyOp(makeEntry(value)); + + assert.strictEqual( + rqp.published.filter(p => p.topic === TOPIC).length, 0); + sinon.assert.notCalled(params.metricsHandler.objects); + }); }); }); From f7e19118d0802d87e987555f5ffe64ca598f7a72 Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Mon, 24 Aug 2026 10:30:48 +0200 Subject: [PATCH 5/5] Restore the original clean room comment The rewording added in the previous commit did not carry any new information, and only made the stacked diff harder to read: the localization trigger comment from BB-814 already explains why the check sits before any replication condition. Issue: BB-816 --- extensions/replication/ReplicationQueuePopulator.js | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/extensions/replication/ReplicationQueuePopulator.js b/extensions/replication/ReplicationQueuePopulator.js index 17145a86d8..9d9d899076 100644 --- a/extensions/replication/ReplicationQueuePopulator.js +++ b/extensions/replication/ReplicationQueuePopulator.js @@ -84,13 +84,9 @@ class ReplicationQueuePopulator extends QueuePopulatorExtension { const locationConfig = (dataStoreName && locationsConfig[dataStoreName]) || {}; // Clean room: the object data still lives on the source (isCRR) - // location, there is nothing local to replicate yet, so localize it - // first (when configured) and skip it. Nothing is lost: the - // localization merge rewrites the metadata, which comes back through - // the oplog with the real location and is replicated then. - // This is unrelated to replicationInfo, which tracks replication of a - // *local* object to remote sites, hence the check before any - // replication condition. + // location and first needs to be localized. This is unrelated to + // replicationInfo, which tracks replication of a *local* object to + // remote sites, hence the check before any replication condition. if (locationConfig.isCRR) { if (this.localizationConfig) { this._publishLocalizationAction(entry, queueEntry, value);