From ae74a263914736a9bc73b57583bae3d4db135e47 Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Tue, 25 Aug 2026 11:20:19 +0200 Subject: [PATCH 1/3] Trigger cold transition from oplog Support "direct-to-cold" transitions: when an object is uploaded with a cold storage class, cloudserver stores the data in the hot location and records the requested cold class in the object metadata, together with the transition-in-progress flag. The lifecycle queue populator now detects these objects from the oplog and publishes the cold archive request, reusing the whole existing transition pipeline. The transition-in-progress flag must not be cleared while requeuing such an object, since the flag (and the cold storage class) is what identifies it as pending a direct transition. Conversely, a bucket lifecycle rule must not transition an object which is already declared as cold. The metadata update completing the transition is stamped with a distinct 's3:LifecycleTransition:Direct' origin op, so that consumers can tell a direct transition from a lifecycle-driven one. Issue: BB-786 --- extensions/gc/tasks/GarbageCollectorTask.js | 6 +- .../lifecycle/LifecycleQueuePopulator.js | 129 ++++++++++ .../tasks/LifecycleColdStatusArchiveTask.js | 6 +- .../LifecycleResetTransitionInProgressTask.js | 17 +- extensions/lifecycle/tasks/LifecycleTask.js | 7 + tests/unit/gc/GarbageCollectorTask.spec.js | 37 +++ .../LifecycleColdStatusArchiveTask.spec.js | 21 ++ .../lifecycle/LifecycleQueuePopulator.spec.js | 224 +++++++++++++++++- ...cycleResetTransitionInProgressTask.spec.js | 39 +++ tests/unit/lifecycle/LifecycleTask.spec.js | 62 +++++ 10 files changed, 542 insertions(+), 6 deletions(-) diff --git a/extensions/gc/tasks/GarbageCollectorTask.js b/extensions/gc/tasks/GarbageCollectorTask.js index dae27c5807..8cb294f87c 100644 --- a/extensions/gc/tasks/GarbageCollectorTask.js +++ b/extensions/gc/tasks/GarbageCollectorTask.js @@ -286,10 +286,14 @@ class GarbageCollectorTask extends BackbeatTask { version, }); + // The object already exposes the location it is transitioned to, so this is a + // direct-to-cold transition. + const isDirectToCold = objMD.getAmzStorageClass() === newLocation; + objMD.setLocation() .setDataStoreName(newLocation) .setAmzStorageClass(newLocation) - .setOriginOp('s3:LifecycleTransition') + .setOriginOp(isDirectToCold ? 's3:LifecycleTransition:Direct' : 's3:LifecycleTransition') .setTransitionInProgress(false) .setUserMetadata({ 'x-amz-meta-scal-s3-transition-attempt': undefined, diff --git a/extensions/lifecycle/LifecycleQueuePopulator.js b/extensions/lifecycle/LifecycleQueuePopulator.js index 756e5c3b75..31d590be79 100644 --- a/extensions/lifecycle/LifecycleQueuePopulator.js +++ b/extensions/lifecycle/LifecycleQueuePopulator.js @@ -20,10 +20,20 @@ const { coldStorageRestoreAdjustTopicPrefix, coldStorageRestoreTopicPrefix, coldStorageGCTopicPrefix, + coldStorageArchiveTopicPrefix, } = config.extensions.lifecycle; const BackbeatProducer = require('../../lib/BackbeatProducer'); const locations = require('../../conf/locationConfig.json') || {}; +// Object creation is the only operation which may declare a cold storage class, and a retry asks +// for a new attempt. Any other originOp comes from backbeat itself, and must not re-trigger. +const transitionOriginOps = [ + 's3:ObjectCreated:Put', + 's3:ObjectCreated:CompleteMultipartUpload', + 's3:ObjectCreated:Copy', + 's3:LifecycleTransition:Retry', +]; + const nSecsPerDay = () => Math.ceil(scaleMsPerDay(config.timeOptions.timeProgressionFactor) / 1000); class LifecycleQueuePopulator extends QueuePopulatorExtension { @@ -100,6 +110,7 @@ class LifecycleQueuePopulator extends QueuePopulatorExtension { next => this._setupProducer(`${coldStorageRestoreAdjustTopicPrefix}${location}`, next), next => this._setupProducer(`${coldStorageRestoreTopicPrefix}${location}`, next), next => this._setupProducer(`${coldStorageGCTopicPrefix}${location}`, next), + next => this._setupProducer(`${coldStorageArchiveTopicPrefix}${location}`, next), ], done); }, cb); } else { @@ -239,6 +250,123 @@ class LifecycleQueuePopulator extends QueuePopulatorExtension { return new Date(date.$date || date); } + _isColdLocation(locationName) { + return !!this.locationConfigs[locationName]?.isCold; + } + + /** + * A transition is "direct" when the object declares a cold storage class while its data still + * sits in a hot location, it has not been archived yet, and cloudserver has flagged it as in + * progress. + * + * @param {Object} md - The object metadata. + * @return {boolean} true if a direct transition is pending for this object. + */ + _isDirectToCold(md) { + return md['x-amz-scal-transition-in-progress'] + && this._isColdLocation(md['x-amz-storage-class']) + && !this._isColdLocation(md.dataStoreName) + && !md.archive?.archiveInfo; + } + + /** + * Handle a "direct-to-cold" transition: cloudserver has written the object data to a hot + * location, but the user requested a cold storage class in the PUT request. Cloudserver + * flags the object as "transition in progress", and it is up to the queue populator to + * trigger the archival, as there is no lifecycle rule (nor lifecycle scan) involved. + * + * The message published here is strictly the same as the one the lifecycle bucket processor + * publishes (c.f. ReplicationAPI.sendDataMoverAction), so that the whole downstream pipeline + * (Sorbet, cold status processor, garbage collector) is reused unchanged. + * + * @param {Object} entry - The record log entry from metadata. + * @return {undefined} + */ + _handleTransitionOp(entry) { + if (!this.vaultClientWrapper) { + return; + } + + if (entry.type !== 'put' || entry.key.startsWith(mpuBucketPrefix)) { + return; + } + + const value = JSON.parse(entry.value); + if (!transitionOriginOps.includes(value.originOp) || !this._isDirectToCold(value)) { + return; + } + + // if entry is a versioned object and is the master entry, skip task as + // the non-master entry will be processed + if (this._isVersionedObject(value) && isMasterKey(entry.key)) { + this.log.trace('skip processing of object master entry'); + return; + } + + const coldLocation = value['x-amz-storage-class']; + const attemptHeader = value['x-amz-meta-scal-s3-transition-attempt']; + const attempt = attemptHeader ? Number.parseInt(attemptHeader, 10) : undefined; + const ownerId = value['owner-id']; + this.vaultClientWrapper.getAccountId(ownerId, (err, accountId) => { + if (err) { + this.log.error('unable to get account', { + method: 'LifecycleQueuePopulator._handleTransitionOp', + ownerId, + err, + }); + return; + } + + this.log.trace( + 'publishing object transition entry', + { bucket: entry.bucket, key: entry.key, version: value.versionId, coldLocation }, + ); + + const topic = `${coldStorageArchiveTopicPrefix}${coldLocation}`; + const key = `${entry.bucket}/${value.key}`; + + let version; + if (value.versionId) { + version = encode(value.versionId); + } + + const transitionTime = this._parseDate( + value['x-amz-scal-transition-time'] || value['last-modified']); + const message = JSON.stringify({ + accountId, + bucketName: entry.bucket, + objectKey: value.key, + objectVersion: version, + requestId: uuid(), + size: value['content-length'], + eTag: `"${value['content-md5']}"`, + try: attempt, + transitionTime: transitionTime.toISOString(), + }); + + const producer = this._producers[topic]; + if (producer) { + LifecycleMetrics.onLifecycleTriggered(this.log, 'queuePopulator', 'archive', + coldLocation, Date.now() - transitionTime.getTime()); + + const kafkaEntry = { key, message }; + producer.send([kafkaEntry], err => { + LifecycleMetrics.onKafkaPublish(this.log, 'ColdStorageArchiveTopic', 'queuePopulator', err, 1); + if (err) { + this.log.error('error publishing object transition request entry', { + error: err, + method: 'LifecycleQueuePopulator._handleTransitionOp', + }); + } + }); + } else { + this.log.error(`producer not available for location ${coldLocation}`, { + method: 'LifecycleQueuePopulator._handleTransitionOp', + }); + } + }); + } + _handleRestoreOp(entry) { if (!this.vaultClientWrapper) { return; @@ -505,6 +633,7 @@ class LifecycleQueuePopulator extends QueuePopulatorExtension { } this._handleRestoreOp(entry); + this._handleTransitionOp(entry); if (this.extConfig.conductor.bucketSource !== 'zookeeper') { this.log.debug('bucket source is not zookeeper, skipping entry', { diff --git a/extensions/lifecycle/tasks/LifecycleColdStatusArchiveTask.js b/extensions/lifecycle/tasks/LifecycleColdStatusArchiveTask.js index 8cea3c62bc..9bf574a3dc 100644 --- a/extensions/lifecycle/tasks/LifecycleColdStatusArchiveTask.js +++ b/extensions/lifecycle/tasks/LifecycleColdStatusArchiveTask.js @@ -112,10 +112,14 @@ class LifecycleColdStatusArchiveTask extends LifecycleUpdateTransitionTask { objectMD.setOriginOp('s3:LifecycleTransition:SetArchive'); if (skipLocationDeletion) { + // The object already exposes the location it is transitioned to, so this + // is a direct-to-cold transition. + const isDirectToCold = objectMD.getAmzStorageClass() === coldLocation; + objectMD.setDataStoreName(coldLocation) .setAmzStorageClass(coldLocation) .setTransitionInProgress(false) - .setOriginOp('s3:LifecycleTransition') + .setOriginOp(isDirectToCold ? 's3:LifecycleTransition:Direct' : 's3:LifecycleTransition') .setUserMetadata({ 'x-amz-meta-scal-s3-transition-attempt': undefined, }); diff --git a/extensions/lifecycle/tasks/LifecycleResetTransitionInProgressTask.js b/extensions/lifecycle/tasks/LifecycleResetTransitionInProgressTask.js index 8768d577fe..2507e47b94 100644 --- a/extensions/lifecycle/tasks/LifecycleResetTransitionInProgressTask.js +++ b/extensions/lifecycle/tasks/LifecycleResetTransitionInProgressTask.js @@ -1,6 +1,7 @@ 'use strict'; const { LifecycleRequeueTask } = require('./LifecycleRequeueTask'); +const locationsConfig = require('../../../conf/locationConfig.json') || {}; class LifecycleResetTransitionInProgressTask extends LifecycleRequeueTask { /** @@ -18,13 +19,27 @@ class LifecycleResetTransitionInProgressTask extends LifecycleRequeueTask { return false; } md.setOriginOp('s3:LifecycleTransition:Retry'); - md.setTransitionInProgress(false); + if (!this._isDirectToCold(md)) { + // Keep the flag as the queue populator keys on it to trigger the next attempt + md.setTransitionInProgress(false); + } md.setUserMetadata({ 'x-amz-meta-scal-s3-transition-attempt': try_, }); return true; } + /** + * Check if object transition was initiated by direct-to-cold request instead of lifecycle rule. + * + * @param {ObjectMD} md - object metadata + * @return {boolean} true if this is a pending direct transition + */ + _isDirectToCold(md) { + return locationsConfig[md.getAmzStorageClass()]?.isCold + && !locationsConfig[md.getDataStoreName()]?.isCold; + } + shouldSkipObject(md, expectedEtag, log) { try { const etag = JSON.parse(expectedEtag); diff --git a/extensions/lifecycle/tasks/LifecycleTask.js b/extensions/lifecycle/tasks/LifecycleTask.js index 8a83118404..0160bbe4b4 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 errorTransitionDeclaredColdObject = errors.InternalError. + customizeDescription('transitioning an object declared as cold is forbidden'); const errorObjectTemporarilyRestored = errors.InternalError. customizeDescription('object temporarily restored'); const errorReplicationInProgress = errors.InternalError. @@ -1276,6 +1278,11 @@ class LifecycleTask extends BackbeatTask { if (isObjectCold) { return next(errorTransitionColdObject); } + // Skip direct-to-cold objects, whose transition is triggered by the queue + // populator and require the transition in progress flag to be set + if (locationsConfig[objectMD.getAmzStorageClass()]?.isCold) { + return next(errorTransitionDeclaredColdObject); + } // If transition is in progress, do not re-publish entry // to data-mover or cold-archive topic. if (objectMD.getTransitionInProgress()) { diff --git a/tests/unit/gc/GarbageCollectorTask.spec.js b/tests/unit/gc/GarbageCollectorTask.spec.js index 3cd4d7fdbc..b39b2de9dc 100644 --- a/tests/unit/gc/GarbageCollectorTask.spec.js +++ b/tests/unit/gc/GarbageCollectorTask.spec.js @@ -99,10 +99,47 @@ describe('GarbageCollectorTask', () => { assert.strictEqual(updatedMD.getDataStoreName(), 'new-location'); assert.strictEqual(updatedMD.getAmzStorageClass(), 'new-location'); assert.strictEqual(updatedMD.getTransitionInProgress(), false); + assert.strictEqual(updatedMD.getOriginOp(), 's3:LifecycleTransition'); done(); }); }); + it('should set the direct transition origin op if the new location was requested', done => { + backbeatClient.batchDeleteResponse = { error: null, res: null }; + + const entry = ActionQueueEntry.create('deleteArchivedSourceData') + .addContext({ + origin: 'lifecycle', + ruleType: 'archive', + bucketName: bucket, + objectKey: key, + versionId: version, + }) + .setAttribute('serviceName', 'lifecycle-transition') + .setAttribute('target.oldLocation', 'old-location') + .setAttribute('target.newLocation', 'new-location') + .setAttribute('target.bucket', bucket) + .setAttribute('target.key', version) + .setAttribute('target.version', key) + .setAttribute('target.accountId', accountId) + .setAttribute('target.owner', owner); + + mdObj.setLocation(loc) + .setDataStoreName('old-location') + .setAmzStorageClass('new-location') + .setTransitionInProgress(true); + backbeatMetadataProxyClient.setMdObj(mdObj); + + gcTask.processActionEntry(entry, err => { + assert.ifError(err); + + const updatedMD = backbeatMetadataProxyClient.mdObj; + assert.strictEqual(updatedMD.getDataStoreName(), 'new-location'); + assert.strictEqual(updatedMD.getTransitionInProgress(), false); + assert.strictEqual(updatedMD.getOriginOp(), 's3:LifecycleTransition:Direct'); + done(); + }); + }); it('should delete archived location info if gc failed with 404', done => { backbeatClient.batchDeleteResponse = { error: { statusCode: 404 }, res: null }; diff --git a/tests/unit/lifecycle/LifecycleColdStatusArchiveTask.spec.js b/tests/unit/lifecycle/LifecycleColdStatusArchiveTask.spec.js index cfc7127ce8..0966cb8f22 100644 --- a/tests/unit/lifecycle/LifecycleColdStatusArchiveTask.spec.js +++ b/tests/unit/lifecycle/LifecycleColdStatusArchiveTask.spec.js @@ -117,6 +117,7 @@ describe('LifecycleColdStatusArchiveTask', () => { assert.strictEqual(gcEntry, null); assert.strictEqual(updatedMD.dataStoreName, 'cold'); assert.strictEqual(updatedMD['x-amz-storage-class'], 'cold'); + assert.strictEqual(updatedMD.originOp, 's3:LifecycleTransition'); assert.deepStrictEqual(updatedMD.archive.archiveInfo, { archiveId: 'da80b6dc-280d-4dce-83b5-d5b40276e321', archiveVersion: 5166759712787974, @@ -144,6 +145,26 @@ describe('LifecycleColdStatusArchiveTask', () => { }); }); + it('should set the direct transition origin op if the cold class was requested', done => { + backbeatClient.batchDeleteResponse = { error: { statusCode: 404 }, res: null }; + + const entry = ColdStorageStatusQueueEntry.createFromKafkaEntry({ value: message }); + mdObj.setLocation() + .setDataStoreName('us-east-1') + .setAmzStorageClass(coldLocation) + .setArchive(null); + backbeatMetadataProxyClient.setMdObj(mdObj); + + archiveTask.processEntry(coldLocation, entry, err => { + assert.ifError(err); + + const updatedMD = backbeatMetadataProxyClient.getReceivedMd(); + assert.strictEqual(updatedMD.dataStoreName, 'cold'); + assert.strictEqual(updatedMD.originOp, 's3:LifecycleTransition:Direct'); + done(); + }); + }); + it('should send kafka entry to delete orphan cold object when source object was deleted', done => { const entry = ColdStorageStatusQueueEntry.createFromKafkaEntry({ value: message }); diff --git a/tests/unit/lifecycle/LifecycleQueuePopulator.spec.js b/tests/unit/lifecycle/LifecycleQueuePopulator.spec.js index 8cd79202c3..38c2e7e68c 100644 --- a/tests/unit/lifecycle/LifecycleQueuePopulator.spec.js +++ b/tests/unit/lifecycle/LifecycleQueuePopulator.spec.js @@ -7,7 +7,8 @@ const config = require('../../../lib/Config'); const { coldStorageRestoreAdjustTopicPrefix, coldStorageRestoreTopicPrefix, - coldStorageGCTopicPrefix + coldStorageGCTopicPrefix, + coldStorageArchiveTopicPrefix } = config.extensions.lifecycle; const LifecycleQueuePopulator = require('../../../extensions/lifecycle/LifecycleQueuePopulator'); @@ -78,6 +79,10 @@ const templateEntry = { 'versionId': '98500086134471999999RG001 0', 'isNFS': true, 'archive': { + archiveInfo: { + archiveId: '04425717-a65c-4e8a-95e1-fa1d902d9d9f', + archiveVersion: 7504504064263669, + }, restoreRequestedAt: Date.now(), restoreRequestedDays: 1, }, @@ -131,16 +136,17 @@ describe('LifecycleQueuePopulator', () => { done(); }); }); - it('should have three producers per cold location', done => { + it('should have four producers per cold location', done => { lcqp.locationConfigs = Object.assign({}, locationConfigs, coldLocationConfigs); lcqp.setupProducers(() => { const producers = Object.keys(lcqp._producers); const coldLocations = Object.keys(coldLocationConfigs); - assert.strictEqual(producers.length, coldLocations.length * 3); + assert.strictEqual(producers.length, coldLocations.length * 4); coldLocations.forEach(loc => { assert(producers.includes(`${coldStorageRestoreAdjustTopicPrefix}${loc}`)); assert(producers.includes(`${coldStorageRestoreTopicPrefix}${loc}`)); assert(producers.includes(`${coldStorageGCTopicPrefix}${loc}`)); + assert(producers.includes(`${coldStorageArchiveTopicPrefix}${loc}`)); }); done(); }); @@ -377,6 +383,212 @@ describe('LifecycleQueuePopulator', () => { }); }); + describe(':_handleTransitionOp', () => { + const accountId = '79a59df900b949e55d96a1e698fbacedfd6e09d98eacf8f8d5218e7cd47ef2be'; + const versionId = '98500086134471999999RG001 0'; + const archiveTopic = `${coldStorageArchiveTopicPrefix}dmf-v1`; + + let lcqp; + let getAccountIdStub; + let kafkaSendStub; + + function getTransitionEntry(overrides) { + const value = Object.assign({ + 'md-model-version': 2, + 'owner-display-name': 'Bart', + 'owner-id': accountId, + 'content-length': 542, + 'content-type': 'text/plain', + 'last-modified': '2017-07-13T02:44:25.519Z', + 'content-md5': '01064f35c238bd2b785e34508c3d27f4', + 'x-amz-storage-class': 'dmf-v1', + 'x-amz-scal-transition-in-progress': true, + 'x-amz-scal-transition-time': '2017-07-13T02:44:20.000Z', + 'key': 'hosts', + 'location': [], + 'isDeleteMarker': false, + 'isNull': false, + versionId, + 'dataStoreName': 'us-east-1', + 'originOp': 's3:ObjectCreated:Put', + }, overrides); + // allow overrides to remove a field by passing `undefined` + Object.keys(value).forEach(k => { + if (value[k] === undefined) { + delete value[k]; + } + }); + return { + type: 'put', + bucket: 'lc-queue-populator-test-bucket', + key: `hosts\x00${versionId}`, + value: JSON.stringify(value), + }; + } + + beforeEach(() => { + lcqp = new LifecycleQueuePopulator(params); + lcqp.locationConfigs = Object.assign({}, coldLocationConfigs, locationConfigs); + getAccountIdStub = sinon.stub().yields(null, accountId); + lcqp.vaultClientWrapper = { + getAccountId: getAccountIdStub, + }; + kafkaSendStub = sinon.stub().yields(); + lcqp._producers[archiveTopic] = { + send: kafkaSendStub, + }; + }); + + afterEach(() => { + sinon.restore(); + }); + + [ + { originOp: 's3:ObjectCreated:Put', ignore: false }, + { originOp: 's3:ObjectCreated:CompleteMultipartUpload', ignore: false }, + { originOp: 's3:ObjectCreated:Copy', ignore: false }, + { originOp: 's3:LifecycleTransition:Retry', ignore: false }, + { originOp: 's3:LifecycleTransition:Start', ignore: true }, + { originOp: 's3:LifecycleTransition:SetArchive', ignore: true }, + { originOp: 's3:LifecycleTransition:Direct', ignore: true }, + { originOp: 's3:LifecycleTransition', ignore: true }, + { originOp: 's3:ObjectRestore:Post', ignore: true }, + ].forEach(({ originOp, ignore }) => { + const outcome = ignore ? 'ignore' : 'consider'; + it(`should ${outcome} ${originOp} event`, () => { + lcqp._handleTransitionOp(getTransitionEntry({ originOp })); + assert.strictEqual(kafkaSendStub.calledOnce, !ignore); + }); + }); + + it('should publish an archive request matching the bucket processor message', () => { + lcqp._handleTransitionOp(getTransitionEntry()); + + assert(kafkaSendStub.calledOnce); + const kafkaEntry = kafkaSendStub.args[0][0][0]; + assert.strictEqual(kafkaEntry.key, 'lc-queue-populator-test-bucket/hosts'); + + const message = JSON.parse(kafkaEntry.message); + assert.deepStrictEqual(message, { + accountId, + bucketName: 'lc-queue-populator-test-bucket', + objectKey: 'hosts', + objectVersion: encode(versionId), + requestId: message.requestId, + size: 542, + eTag: '"01064f35c238bd2b785e34508c3d27f4"', + transitionTime: '2017-07-13T02:44:20.000Z', + }); + assert(message.requestId); + }); + + it('should fall back on last-modified when no transition time is set', () => { + lcqp._handleTransitionOp(getTransitionEntry({ + 'x-amz-scal-transition-time': undefined, + })); + + assert(kafkaSendStub.calledOnce); + const message = JSON.parse(kafkaSendStub.args[0][0][0].message); + assert.strictEqual(message.transitionTime, '2017-07-13T02:44:25.519Z'); + }); + + it('should publish the transition attempt count', () => { + lcqp._handleTransitionOp(getTransitionEntry({ + 'originOp': 's3:LifecycleTransition:Retry', + 'x-amz-meta-scal-s3-transition-attempt': '3', + })); + + assert(kafkaSendStub.calledOnce); + const message = JSON.parse(kafkaSendStub.args[0][0][0].message); + assert.strictEqual(message.try, 3); + }); + + it('should not set objectVersion for a non-versioned object', () => { + const entry = getTransitionEntry({ versionId: undefined }); + entry.key = 'hosts'; + lcqp._handleTransitionOp(entry); + + assert(kafkaSendStub.calledOnce); + const message = JSON.parse(kafkaSendStub.args[0][0][0].message); + assert.strictEqual(message.objectVersion, undefined); + }); + + [ + { + desc: 'transition is not in progress', + overrides: { 'x-amz-scal-transition-in-progress': undefined }, + }, + { + desc: 'the storage class is not cold', + overrides: { 'x-amz-storage-class': 'us-east-2' }, + }, + { + desc: 'the object has no storage class', + overrides: { 'x-amz-storage-class': undefined }, + }, + { + desc: 'the data is already in the cold location', + overrides: { dataStoreName: 'dmf-v1' }, + }, + { + desc: 'the object is already archived', + overrides: { + archive: { + archiveInfo: { + archiveId: '04425717-a65c-4e8a-95e1-fa1d902d9d9f', + archiveVersion: 7504504064263669, + }, + }, + }, + }, + ].forEach(({ desc, overrides }) => { + it(`should not publish when ${desc}`, () => { + lcqp._handleTransitionOp(getTransitionEntry(overrides)); + assert(!getAccountIdStub.called); + assert(!kafkaSendStub.called); + }); + }); + + it('should skip the master key of a versioned object', () => { + const entry = getTransitionEntry(); + entry.key = 'hosts'; + lcqp._handleTransitionOp(entry); + assert(!kafkaSendStub.called); + }); + + it('should skip mpu shadow bucket entries', () => { + const entry = getTransitionEntry(); + entry.key = `mpuShadowBucket${entry.key}`; + lcqp._handleTransitionOp(entry); + assert(!kafkaSendStub.called); + }); + + it('should skip delete operations', () => { + const entry = getTransitionEntry(); + entry.type = 'delete'; + lcqp._handleTransitionOp(entry); + assert(!kafkaSendStub.called); + }); + + it('should do nothing without a vault client', () => { + lcqp.vaultClientWrapper = null; + lcqp._handleTransitionOp(getTransitionEntry()); + assert(!kafkaSendStub.called); + }); + + it('should not publish when the account cannot be resolved', () => { + getAccountIdStub.yields(errors.InternalError); + lcqp._handleTransitionOp(getTransitionEntry()); + assert(!kafkaSendStub.called); + }); + + it('should not throw when no producer is available', () => { + delete lcqp._producers[archiveTopic]; + lcqp._handleTransitionOp(getTransitionEntry()); + assert(getAccountIdStub.calledOnce); + }); + }); + describe(':filter', () => { const bucketMD = { name: 'lc-queue-populator-test-bucket', @@ -410,6 +622,12 @@ describe('LifecycleQueuePopulator', () => { assert(handleDeleteStub.calledOnce); }); + it('should call _handleTransitionOp on put message', () => { + const handleTransitionStub = sinon.stub(lcqp, '_handleTransitionOp').returns(); + lcqp.filter(getKafkaEntry('s3:ObjectCreated:Put')); + assert(handleTransitionStub.calledOnce); + }); + it('should not update zookeeper when bucketSource is mongodb (default)', () => { lcqp.extConfig.conductor.bucketSource = 'mongodb'; const putEntry = { diff --git a/tests/unit/lifecycle/LifecycleResetTransitionInProgressTask.spec.js b/tests/unit/lifecycle/LifecycleResetTransitionInProgressTask.spec.js index 4b466174f0..af90dbb3f3 100644 --- a/tests/unit/lifecycle/LifecycleResetTransitionInProgressTask.spec.js +++ b/tests/unit/lifecycle/LifecycleResetTransitionInProgressTask.spec.js @@ -38,6 +38,18 @@ describe('LifecycleResetTransitionInProgressTask', () => { .setUserMetadata({ 'x-amz-meta-scal-s3-transition-attempt': 11, }); + // "direct" transition: the cold storage class was requested in the PUT request, and the + // data still lies in the hot location + const objectDirectTransitioning = new ObjectMD() + .setContentMd5('etag1') + .setTransitionInProgress(true) + .setAmzStorageClass('location-dmf-v1') + .setDataStoreName('us-east-1'); + const objectDirectTransitioned = new ObjectMD() + .setContentMd5('etag1') + .setTransitionInProgress(true) + .setAmzStorageClass('location-dmf-v1') + .setDataStoreName('location-dmf-v1'); beforeEach(() => { backbeatMetadataProxyClient = new BackbeatMetadataProxyMock(); @@ -90,4 +102,31 @@ describe('LifecycleResetTransitionInProgressTask', () => { done(); }); }); + + it('should keep transition in progress flag for a direct transition', done => { + backbeatMetadataProxyClient.setMdObj(objectDirectTransitioning); + task.processActionEntry(actionEntry, err => { + assert.ifError(err); + + const md = backbeatMetadataProxyClient.mdObj; + assert.ok(md.getTransitionInProgress()); + assert.strictEqual(md.getOriginOp(), 's3:LifecycleTransition:Retry'); + const umd = JSON.parse(md.getUserMetadata()); + assert.strictEqual(umd['x-amz-meta-scal-s3-transition-attempt'], 12); + + done(); + }); + }); + + it('should reset transition in progress flag once the object is in the cold location', done => { + backbeatMetadataProxyClient.setMdObj(objectDirectTransitioned); + task.processActionEntry(actionEntry, err => { + assert.ifError(err); + + const md = backbeatMetadataProxyClient.mdObj; + assert.ok(!md.getTransitionInProgress()); + + done(); + }); + }); }); diff --git a/tests/unit/lifecycle/LifecycleTask.spec.js b/tests/unit/lifecycle/LifecycleTask.spec.js index 7150bbe147..4e9d0c5bc7 100644 --- a/tests/unit/lifecycle/LifecycleTask.spec.js +++ b/tests/unit/lifecycle/LifecycleTask.spec.js @@ -12,6 +12,7 @@ const LifecycleTaskV2 = require( '../../../extensions/lifecycle/tasks/LifecycleTaskV2'); const ActionQueueEntry = require('../../../lib/models/ActionQueueEntry'); const { LifecycleMetrics } = require('../../../extensions/lifecycle/LifecycleMetrics'); +const ReplicationAPI = require('../../../extensions/replication/ReplicationAPI'); const fakeLogger = require('../../utils/fakeLogger'); const { withActiveSpan } = require('../../utils/withActiveSpan'); const { timeOptions } = require('../../functional/lifecycle/configObjects'); @@ -2473,6 +2474,67 @@ describe('lifecycle task helper methods', () => { }); }); + describe('_applyTransitionRule', () => { + const testParams = { + bucket: 'test-bucket', + owner: 'test-owner', + objectKey: 'test-key', + site: 'us-east-2', + transitionTime: Date.now(), + }; + + let lifecycleTask; + + beforeEach(() => { + lifecycleTask = new LifecycleTask(lp); + lifecycleTask.pausedLocations = new Set(); + lifecycleTask.circuitBreakers = { tripped: () => false }; + }); + + afterEach(() => { + sinon.restore(); + }); + + function stubObjectMD(overrides) { + const objectMD = Object.assign({ + getReplicationStatus: () => 'COMPLETED', + getDataStoreName: () => 'us-east-1', + getAmzStorageClass: () => 'us-east-1', + getTransitionInProgress: () => false, + getArchive: () => undefined, + setTransitionInProgress: () => {}, + setOriginOp: () => {}, + getSerialized: () => '{}', + }, overrides); + sinon.stub(lifecycleTask, '_getObjectMD').yields(null, objectMD); + } + + it('should not transition an object declared as cold', done => { + stubObjectMD({ getAmzStorageClass: () => 'location-dmf-v1' }); + const getEntryStub = sinon.stub(lifecycleTask, '_getTransitionActionEntry'); + + lifecycleTask._applyTransitionRule(testParams, fakeLogger, err => { + assert.strictEqual(err.description, + 'transitioning an object declared as cold is forbidden'); + assert(!getEntryStub.called); + done(); + }); + }); + + it('should transition an object with a hot storage class', done => { + stubObjectMD(); + const getEntryStub = sinon.stub(lifecycleTask, '_getTransitionActionEntry').yields(null, {}); + sinon.stub(ReplicationAPI, 'sendDataMoverAction').yields(); + sinon.stub(lifecycleTask, '_putObjectMD').yields(); + + lifecycleTask._applyTransitionRule(testParams, fakeLogger, err => { + assert.ifError(err); + assert(getEntryStub.calledOnce); + done(); + }); + }); + }); + describe('_sendObjectAction', () => { it('should emit trigger metrics with the entry location', done => { const lifecycleTask = new LifecycleTask(lp); From b1ca3b399762054b125612b456241cc171ffaf94 Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Tue, 25 Aug 2026 17:51:45 +0200 Subject: [PATCH 2/3] Dispatch oplog entries from filter() The queue populator parsed each oplog entry several times: once in filter() to look at the bucket, then again in every handler it may call. Each handler also re-checked the originOp it cares about, so the list of interesting operations was spread over the whole file. Decode the entry once in filter() and switch on the originOp there, passing the parsed value down. Handlers now only decide whether the object itself qualifies, which is what the new transition handler needs anyway. Issue: BB-786 --- .../lifecycle/LifecycleQueuePopulator.js | 88 +++++++------- .../lifecycle/LifecycleQueuePopulator.spec.js | 111 ++++++------------ 2 files changed, 82 insertions(+), 117 deletions(-) diff --git a/extensions/lifecycle/LifecycleQueuePopulator.js b/extensions/lifecycle/LifecycleQueuePopulator.js index 31d590be79..c5858bbb4f 100644 --- a/extensions/lifecycle/LifecycleQueuePopulator.js +++ b/extensions/lifecycle/LifecycleQueuePopulator.js @@ -25,15 +25,6 @@ const { const BackbeatProducer = require('../../lib/BackbeatProducer'); const locations = require('../../conf/locationConfig.json') || {}; -// Object creation is the only operation which may declare a cold storage class, and a retry asks -// for a new attempt. Any other originOp comes from backbeat itself, and must not re-trigger. -const transitionOriginOps = [ - 's3:ObjectCreated:Put', - 's3:ObjectCreated:CompleteMultipartUpload', - 's3:ObjectCreated:Copy', - 's3:LifecycleTransition:Retry', -]; - const nSecsPerDay = () => Math.ceil(scaleMsPerDay(config.timeOptions.timeProgressionFactor) / 1000); class LifecycleQueuePopulator extends QueuePopulatorExtension { @@ -280,19 +271,19 @@ class LifecycleQueuePopulator extends QueuePopulatorExtension { * (Sorbet, cold status processor, garbage collector) is reused unchanged. * * @param {Object} entry - The record log entry from metadata. + * @param {Object} [value] - The object metadata, already decoded by the caller. * @return {undefined} */ - _handleTransitionOp(entry) { + _handleTransitionOp(entry, value) { if (!this.vaultClientWrapper) { return; } - if (entry.type !== 'put' || entry.key.startsWith(mpuBucketPrefix)) { + if (entry.key.startsWith(mpuBucketPrefix)) { return; } - const value = JSON.parse(entry.value); - if (!transitionOriginOps.includes(value.originOp) || !this._isDirectToCold(value)) { + if (!this._isDirectToCold(value)) { return; } @@ -367,23 +358,12 @@ class LifecycleQueuePopulator extends QueuePopulatorExtension { }); } - _handleRestoreOp(entry) { + _handleRestoreOp(entry, value) { if (!this.vaultClientWrapper) { return; } - if (entry.type !== 'put' || - entry.key.startsWith(mpuBucketPrefix)) { - return; - } - - const value = JSON.parse(entry.value); - - const operation = value.originOp; - // supporting both 's3:ObjectRestore' and 's3:ObjectRestore:Post' to keep - // compatibility with older cloudserver versions, the switch to 's3:ObjectRestore:Post' - // was made to have the correct event type for bucket notifications - if (!['s3:ObjectRestore', 's3:ObjectRestore:Post', 's3:ObjectRestore:Retry'].includes(operation)) { + if (entry.key.startsWith(mpuBucketPrefix)) { return; } @@ -632,8 +612,42 @@ class LifecycleQueuePopulator extends QueuePopulatorExtension { return undefined; } - this._handleRestoreOp(entry); - this._handleTransitionOp(entry); + // The entry is decoded once here, then dispatched to the single handler which may be + // interested in it: most entries are of no interest to any of them. + const { error, result: value } = safeJsonParse(entry.value); + if (error) { + this.log.error('could not parse log entry', { + method: 'LifecycleQueuePopulator.filter', + bucket: entry.bucket, + key: entry.key, + error, + }); + return undefined; + } + + switch (value.originOp) { + // supporting both 's3:ObjectRestore' and 's3:ObjectRestore:Post' to keep compatibility with + // older cloudserver versions, the switch to 's3:ObjectRestore:Post' was made to have the + // correct event type for bucket notifications + case 's3:ObjectRestore': + case 's3:ObjectRestore:Post': + case 's3:ObjectRestore:Retry': + this._handleRestoreOp(entry, value); + break; + + // Object creation is the only operation which may declare a cold storage class, and a retry + // asks for a new attempt: any other originOp comes from backbeat itself, and must not + // re-trigger a transition. + case 's3:ObjectCreated:Put': + case 's3:ObjectCreated:CompleteMultipartUpload': + case 's3:ObjectCreated:Copy': + case 's3:LifecycleTransition:Retry': + this._handleTransitionOp(entry, value); + break; + + default: + break; + } if (this.extConfig.conductor.bucketSource !== 'zookeeper') { this.log.debug('bucket source is not zookeeper, skipping entry', { @@ -644,15 +658,7 @@ class LifecycleQueuePopulator extends QueuePopulatorExtension { let bucketValue = {}; if (this._isBucketEntryFromBucketd(entry)) { - const parsedEntry = safeJsonParse(entry.value); - if (parsedEntry.error) { - this.log.error('could not parse raft log entry', { - value: entry.value, - error: parsedEntry.error, - }); - return undefined; - } - const parsedAttr = safeJsonParse(parsedEntry.result.attributes); + const parsedAttr = safeJsonParse(value.attributes); if (parsedAttr.error) { this.log.error('could not parse raft log entry attribute', { value: entry.value, @@ -662,13 +668,7 @@ class LifecycleQueuePopulator extends QueuePopulatorExtension { } bucketValue = parsedAttr.result; } else if (this._isBucketEntryFromFileMD(entry)) { - const { error, result } = safeJsonParse(entry.value); - if (error) { - this.log.error('could not parse file md log entry', - { value: entry.value, error }); - return undefined; - } - bucketValue = result; + bucketValue = value; } else { // not a bucket entry return undefined; diff --git a/tests/unit/lifecycle/LifecycleQueuePopulator.spec.js b/tests/unit/lifecycle/LifecycleQueuePopulator.spec.js index 38c2e7e68c..8d71c80a8f 100644 --- a/tests/unit/lifecycle/LifecycleQueuePopulator.spec.js +++ b/tests/unit/lifecycle/LifecycleQueuePopulator.spec.js @@ -155,6 +155,7 @@ describe('LifecycleQueuePopulator', () => { describe(':_handleRestoreOp', () => { let lcqp; + const handleRestoreOp = entry => lcqp._handleRestoreOp(entry, JSON.parse(entry.value)); const getAccountIdStub = sinon.stub().yields(null, '79a59df900b949e55d96a1e698fbacedfd6e09d98eacf8f8d5218e7cd47ef2be'); beforeEach(() => { @@ -167,37 +168,6 @@ describe('LifecycleQueuePopulator', () => { afterEach(() => { sinon.restore(); }); - [ - { - event: 's3:ObjectRestore', - ignore: false, - }, - { - event: 's3:ObjectRestore:Post', - ignore: false, - }, - { - event: 's3:ObjectRestore:Retry', - ignore: false, - }, - { - event: 's3:ObjectCreated:Put', - ignore: true, - }, - ].forEach(params => { - const outcome = params.ignore ? 'ignore' : 'consider'; - it(`should ${outcome} ${params.event} event`, () => { - const getAccountIdStub = sinon.stub().yields(null, - '79a59df900b949e55d96a1e698fbacedfd6e09d98eacf8f8d5218e7cd47ef2be'); - lcqp.vaultClientWrapper = { - getAccountId: getAccountIdStub, - }; - const entry = getKafkaEntry(params.event); - lcqp._handleRestoreOp(entry); - assert.strictEqual(getAccountIdStub.calledOnce, !params.ignore); - }); - }); - describe('restore requests', () => { const kafkaSendStub = sinon.stub().yields(); const kafkaAdjustSendStub = sinon.stub().yields(); @@ -268,7 +238,7 @@ describe('LifecycleQueuePopulator', () => { value: JSON.stringify(objMd), }; - lcqp._handleRestoreOp(entry); + handleRestoreOp(entry); assert(!kafkaAdjustSendStub.calledOnce); assert(kafkaSendStub.calledOnce); @@ -323,7 +293,7 @@ describe('LifecycleQueuePopulator', () => { value: JSON.stringify(objMd), }; - lcqp._handleRestoreOp(entry); + handleRestoreOp(entry); assert(kafkaAdjustSendStub.calledOnce); assert(!kafkaSendStub.calledOnce); @@ -375,7 +345,7 @@ describe('LifecycleQueuePopulator', () => { value: JSON.stringify(objMd), }; - lcqp._handleRestoreOp(entry); + handleRestoreOp(entry); assert(!kafkaAdjustSendStub.calledOnce); assert(!kafkaSendStub.calledOnce); @@ -392,6 +362,8 @@ describe('LifecycleQueuePopulator', () => { let getAccountIdStub; let kafkaSendStub; + const handleTransitionOp = entry => lcqp._handleTransitionOp(entry, JSON.parse(entry.value)); + function getTransitionEntry(overrides) { const value = Object.assign({ 'md-model-version': 2, @@ -443,26 +415,8 @@ describe('LifecycleQueuePopulator', () => { sinon.restore(); }); - [ - { originOp: 's3:ObjectCreated:Put', ignore: false }, - { originOp: 's3:ObjectCreated:CompleteMultipartUpload', ignore: false }, - { originOp: 's3:ObjectCreated:Copy', ignore: false }, - { originOp: 's3:LifecycleTransition:Retry', ignore: false }, - { originOp: 's3:LifecycleTransition:Start', ignore: true }, - { originOp: 's3:LifecycleTransition:SetArchive', ignore: true }, - { originOp: 's3:LifecycleTransition:Direct', ignore: true }, - { originOp: 's3:LifecycleTransition', ignore: true }, - { originOp: 's3:ObjectRestore:Post', ignore: true }, - ].forEach(({ originOp, ignore }) => { - const outcome = ignore ? 'ignore' : 'consider'; - it(`should ${outcome} ${originOp} event`, () => { - lcqp._handleTransitionOp(getTransitionEntry({ originOp })); - assert.strictEqual(kafkaSendStub.calledOnce, !ignore); - }); - }); - it('should publish an archive request matching the bucket processor message', () => { - lcqp._handleTransitionOp(getTransitionEntry()); + handleTransitionOp(getTransitionEntry()); assert(kafkaSendStub.calledOnce); const kafkaEntry = kafkaSendStub.args[0][0][0]; @@ -483,7 +437,7 @@ describe('LifecycleQueuePopulator', () => { }); it('should fall back on last-modified when no transition time is set', () => { - lcqp._handleTransitionOp(getTransitionEntry({ + handleTransitionOp(getTransitionEntry({ 'x-amz-scal-transition-time': undefined, })); @@ -493,7 +447,7 @@ describe('LifecycleQueuePopulator', () => { }); it('should publish the transition attempt count', () => { - lcqp._handleTransitionOp(getTransitionEntry({ + handleTransitionOp(getTransitionEntry({ 'originOp': 's3:LifecycleTransition:Retry', 'x-amz-meta-scal-s3-transition-attempt': '3', })); @@ -506,7 +460,7 @@ describe('LifecycleQueuePopulator', () => { it('should not set objectVersion for a non-versioned object', () => { const entry = getTransitionEntry({ versionId: undefined }); entry.key = 'hosts'; - lcqp._handleTransitionOp(entry); + handleTransitionOp(entry); assert(kafkaSendStub.calledOnce); const message = JSON.parse(kafkaSendStub.args[0][0][0].message); @@ -543,7 +497,7 @@ describe('LifecycleQueuePopulator', () => { }, ].forEach(({ desc, overrides }) => { it(`should not publish when ${desc}`, () => { - lcqp._handleTransitionOp(getTransitionEntry(overrides)); + handleTransitionOp(getTransitionEntry(overrides)); assert(!getAccountIdStub.called); assert(!kafkaSendStub.called); }); @@ -552,39 +506,32 @@ describe('LifecycleQueuePopulator', () => { it('should skip the master key of a versioned object', () => { const entry = getTransitionEntry(); entry.key = 'hosts'; - lcqp._handleTransitionOp(entry); + handleTransitionOp(entry); assert(!kafkaSendStub.called); }); it('should skip mpu shadow bucket entries', () => { const entry = getTransitionEntry(); entry.key = `mpuShadowBucket${entry.key}`; - lcqp._handleTransitionOp(entry); - assert(!kafkaSendStub.called); - }); - - it('should skip delete operations', () => { - const entry = getTransitionEntry(); - entry.type = 'delete'; - lcqp._handleTransitionOp(entry); + handleTransitionOp(entry); assert(!kafkaSendStub.called); }); it('should do nothing without a vault client', () => { lcqp.vaultClientWrapper = null; - lcqp._handleTransitionOp(getTransitionEntry()); + handleTransitionOp(getTransitionEntry()); assert(!kafkaSendStub.called); }); it('should not publish when the account cannot be resolved', () => { getAccountIdStub.yields(errors.InternalError); - lcqp._handleTransitionOp(getTransitionEntry()); + handleTransitionOp(getTransitionEntry()); assert(!kafkaSendStub.called); }); it('should not throw when no producer is available', () => { delete lcqp._producers[archiveTopic]; - lcqp._handleTransitionOp(getTransitionEntry()); + handleTransitionOp(getTransitionEntry()); assert(getAccountIdStub.calledOnce); }); }); @@ -622,10 +569,28 @@ describe('LifecycleQueuePopulator', () => { assert(handleDeleteStub.calledOnce); }); - it('should call _handleTransitionOp on put message', () => { - const handleTransitionStub = sinon.stub(lcqp, '_handleTransitionOp').returns(); - lcqp.filter(getKafkaEntry('s3:ObjectCreated:Put')); - assert(handleTransitionStub.calledOnce); + [ + { originOp: 's3:ObjectRestore', handler: '_handleRestoreOp' }, + { originOp: 's3:ObjectRestore:Post', handler: '_handleRestoreOp' }, + { originOp: 's3:ObjectRestore:Retry', handler: '_handleRestoreOp' }, + { originOp: 's3:ObjectCreated:Put', handler: '_handleTransitionOp' }, + { originOp: 's3:ObjectCreated:CompleteMultipartUpload', handler: '_handleTransitionOp' }, + { originOp: 's3:ObjectCreated:Copy', handler: '_handleTransitionOp' }, + { originOp: 's3:LifecycleTransition:Retry', handler: '_handleTransitionOp' }, + { originOp: 's3:LifecycleTransition:Start', handler: null }, + { originOp: 's3:LifecycleTransition:SetArchive', handler: null }, + { originOp: 's3:LifecycleTransition:Direct', handler: null }, + { originOp: 's3:LifecycleTransition', handler: null }, + ].forEach(({ originOp, handler }) => { + it(`should dispatch ${originOp} to ${handler || 'no handler'}`, () => { + const restoreStub = sinon.stub(lcqp, '_handleRestoreOp').returns(); + const transitionStub = sinon.stub(lcqp, '_handleTransitionOp').returns(); + + lcqp.filter(getKafkaEntry(originOp)); + + assert.strictEqual(restoreStub.calledOnce, handler === '_handleRestoreOp'); + assert.strictEqual(transitionStub.calledOnce, handler === '_handleTransitionOp'); + }); }); it('should not update zookeeper when bucketSource is mongodb (default)', () => { From 63b047fe85caea2fd472bdc3000f755b41821942 Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Wed, 2 Sep 2026 14:30:22 +0200 Subject: [PATCH 3/3] Skip master entries before adjusting restore expiry The restore and transition handlers each repeated the same two guards on the oplog entry: mpu shadow bucket keys, and the master entry of a versioned object, which duplicates the version entry processed on its own. Group them in a single predicate, so both handlers agree on what counts as an object entry worth acting upon. In the restore handler this guard sat below the branch adjusting the restore expiry of an already-restored object: master and version entry both reached it, and the adjust message was published twice for every versioned object. Checking upfront leaves a single publication. The delete handler keeps its own guards: it takes the raw entry before parsing, and skips null versions and delete markers as well. Issue: BB-786 --- .../lifecycle/LifecycleQueuePopulator.js | 44 ++++++++++-------- .../lifecycle/LifecycleQueuePopulator.spec.js | 46 +++++++++++++++++++ 2 files changed, 71 insertions(+), 19 deletions(-) diff --git a/extensions/lifecycle/LifecycleQueuePopulator.js b/extensions/lifecycle/LifecycleQueuePopulator.js index c5858bbb4f..bac52c43c9 100644 --- a/extensions/lifecycle/LifecycleQueuePopulator.js +++ b/extensions/lifecycle/LifecycleQueuePopulator.js @@ -246,9 +246,29 @@ class LifecycleQueuePopulator extends QueuePopulatorExtension { } /** - * A transition is "direct" when the object declares a cold storage class while its data still - * sits in a hot location, it has not been archived yet, and cloudserver has flagged it as in - * progress. + * Whether a put entry designates an object we may act upon: bucket entries carry no object + * key, mpu shadow bucket entries are internal, and the master entry of a versioned object + * duplicates the version entry, which is processed on its own. + * + * @param {Object} entry - The record log entry from metadata. + * @param {Object} value - The object metadata, already decoded by the caller. + * @return {boolean} true if the entry may be dispatched to an object handler. + */ + _isDistinctObjectEntry(entry, value) { + if (!entry.key || entry.key.startsWith(mpuBucketPrefix)) { + return false; + } + + if (this._isVersionedObject(value) && isMasterKey(entry.key)) { + this.log.trace('skip processing of object master entry'); + return false; + } + + return true; + } + + /** + * Check if object transition was initiated by direct-to-cold request instead of lifecycle rule. * * @param {Object} md - The object metadata. * @return {boolean} true if a direct transition is pending for this object. @@ -279,7 +299,7 @@ class LifecycleQueuePopulator extends QueuePopulatorExtension { return; } - if (entry.key.startsWith(mpuBucketPrefix)) { + if (!this._isDistinctObjectEntry(entry, value)) { return; } @@ -287,13 +307,6 @@ class LifecycleQueuePopulator extends QueuePopulatorExtension { return; } - // if entry is a versioned object and is the master entry, skip task as - // the non-master entry will be processed - if (this._isVersionedObject(value) && isMasterKey(entry.key)) { - this.log.trace('skip processing of object master entry'); - return; - } - const coldLocation = value['x-amz-storage-class']; const attemptHeader = value['x-amz-meta-scal-s3-transition-attempt']; const attempt = attemptHeader ? Number.parseInt(attemptHeader, 10) : undefined; @@ -363,7 +376,7 @@ class LifecycleQueuePopulator extends QueuePopulatorExtension { return; } - if (entry.key.startsWith(mpuBucketPrefix)) { + if (!this._isDistinctObjectEntry(entry, value)) { return; } @@ -391,13 +404,6 @@ class LifecycleQueuePopulator extends QueuePopulatorExtension { return; } - // if entry is a versioned object and is the master entry, skip task as - // the non-master entry will be processed - if (this._isVersionedObject(value) && isMasterKey(entry.key)) { - this.log.trace('skip processing of object master entry'); - return; - } - // We would need to provide the object's bucket's account id as part of the kafka entry. // This account id would be used by Sorbet to assume the bucket's account role. // The assumed credentials will be sent and used by TLP server to put object version diff --git a/tests/unit/lifecycle/LifecycleQueuePopulator.spec.js b/tests/unit/lifecycle/LifecycleQueuePopulator.spec.js index 8d71c80a8f..72bf496256 100644 --- a/tests/unit/lifecycle/LifecycleQueuePopulator.spec.js +++ b/tests/unit/lifecycle/LifecycleQueuePopulator.spec.js @@ -350,6 +350,52 @@ describe('LifecycleQueuePopulator', () => { assert(!kafkaAdjustSendStub.calledOnce); assert(!kafkaSendStub.calledOnce); }); + + it('should skip send duration-adjust for the master entry of a versioned object', () => { + const versionId = '98500086134471999999RG001 0'; + const objMd = { + 'md-model-version': 2, + 'owner-display-name': 'Bart', + 'owner-id': '79a59df900b949e55d96a1e698fbacedfd6e09d98eacf8f8d5218e7cd47ef2be', + 'x-amz-storage-class': 'dmf-v1', + 'content-length': 542, + 'content-type': 'text/plain', + 'last-modified': '2017-07-13T02:44:25.515Z', + 'content-md5': '01064f35c238bd2b785e34508c3d27f4', + 'key': 'object', + 'location': [], + 'isDeleteMarker': false, + 'isNull': false, + versionId, + 'archive': { + archiveInfo: { + archiveId: '04425717-a65c-4e8a-95e1-fa1d902d9d9f', + archiveVersion: 7504504064263669 + }, + restoreCompletedAt: '2017-07-13T02:44:25.519Z', + restoreWillExpireAt: '2017-07-15T02:44:25.519Z', + }, + 'dataStoreName': 'dmf-v1', + 'originOp': 's3:ObjectRestore:Post', + }; + // the version entry carries the same update, and is processed on its own + const entry = { + type: 'put', + bucket: 'lc-queue-populator-test-bucket', + key: 'object', + value: JSON.stringify(objMd), + }; + + handleRestoreOp(entry); + + assert(!kafkaAdjustSendStub.called); + assert(!kafkaSendStub.called); + + handleRestoreOp(Object.assign({}, entry, { key: `object\x00${versionId}` })); + + assert(kafkaAdjustSendStub.calledOnce); + assert(!kafkaSendStub.called); + }); }); });