Merge branch 'improvement/BB-592' into q/8.6

Author: bert-e

extensions/replication/tasks/CopyLocationTask.js

@@ -797,8 +797,8 @@ class CopyLocationTask extends BackbeatTask {
     }
 
     /**
-     * Get the source object metadata and ensure the latest object MD5 hash is
-     * the same as in the action entry.
+     * Ensure the latest object MD5 hash is the same as in the action entry,
+     * and that the object has not already been transitioned to the destination
      * @param {ActionQueueEntry} actionEntry - the action entry
      * @param {ObjectMD} objMD - metadata object
 
@@ -816,6 +816,11 @@ class CopyLocationTask extends BackbeatTask {
                     'object contents have changed');
             }
         }
+        if (objMD.getDataStoreName() === actionEntry.getAttribute('toLocation')) {
+            // The object was already transitioned to the destination location
+            return errors.InvalidObjectState.customizeDescription(
+                'object already transitioned');
+        }
         return null;
     }
 
@@ -824,6 +829,11 @@ class CopyLocationTask extends BackbeatTask {
             actionEntry.setError(err);
         }
         log.info('action execution ended', actionEntry.getLogInfo());
+        // skip object if it was already transitioned
+        if (err && (err.InvalidObjectState || err.code === 'InvalidObjectState')) {
+            log.info('object skipped: invalid object state', actionEntry.getLogInfo());
+            return { committable: true };
+        }
         if (!actionEntry.getResultsTopic()) {
             // no result requested, we may commit immediately
             return { committable: true };

package.json

@@ -1,6 +1,6 @@
 {
   "name": "backbeat",
-  "version": "8.6.43",
+  "version": "8.6.44",
   "description": "Asynchronous queue and job manager",
   "main": "index.js",
   "scripts": {

tests/unit/replication/CopyLocationTask.spec.js

@@ -0,0 +1,100 @@
+const assert = require('assert');
+const sinon = require('sinon');
+
+const CopyLocationTask = require('../../../extensions/replication/tasks/CopyLocationTask');
+const ActionQueueEntry = require('../../../lib/models/ActionQueueEntry');
+const { errors } = require('arsenal');
+const { ObjectMD } = require('arsenal').models;
+
+const fakeLogger = require('../../utils/fakeLogger');
+
+describe('CopyLocationTask', () => {
+    describe('_checkObjectState', () => {
+        let task;
+
+        beforeEach(() => {
+            task = new CopyLocationTask({
+                getStateVars: () => ({
+                    site: 'test-site',
+                    mProducer: {
+                        getProducer: () => {},
+                    }
+                }),
+            });
+        });
+
+        it('should return invalidState error object has been changed', () => {
+            const objMd = new ObjectMD();
+            objMd.setContentMd5('1234-9');
+
+            const entry = new ActionQueueEntry({
+                target: {
+                    eTag: '"156781-9"',
+                },
+            });
+
+            const res = task._checkObjectState(entry, objMd);
+            assert(res.InvalidObjectState);
+        });
+
+        it('should return invalidState error when object already transitioned', () => {
+            const objMd = new ObjectMD();
+            objMd.setDataStoreName('test-site');
+
+            const entry = new ActionQueueEntry({
+                target: {},
+                toLocation: 'test-site',
+            });
+
+            const res = task._checkObjectState(entry, objMd);
+            assert(res.InvalidObjectState);
+        });
+
+        it('should not return error if object is valid', () => {
+            const objMd = new ObjectMD();
+            objMd.setDataStoreName('STANDARD');
+            objMd.setContentMd5('1234-9');
+
+            const entry = new ActionQueueEntry({
+                target: {
+                    eTag: '"1234-9"',
+                },
+                toLocation: 'test-site',
+            });
+
+            const res = task._checkObjectState(entry, objMd);
+            assert.equal(res, null);
+        });
+    });
+
+    describe('_publishCopyLocationStatus', () => {
+        let task;
+
+        beforeEach(() => {
+            task = new CopyLocationTask({
+                getStateVars: () => ({
+                    site: 'test-site',
+                    mProducer: {
+                        getProducer: () => {},
+                    }
+                }),
+            });
+        });
+
+        it('should skip object if object state is invalid', () => {
+            const entry = new ActionQueueEntry({
+                target: {
+                    key: 'key',
+                    eTag: '"1234-9"',
+                },
+                toLocation: 'test-site',
+            });
+
+            task.replicationStatusProducer = sinon.stub().yields();
+
+            const res = task._publishCopyLocationStatus(errors.InvalidObjectState, entry, null, fakeLogger);
+            assert.strictEqual(res.committable, true);
+            assert(task.replicationStatusProducer.notCalled);
+        });
+    });
+});