ARSN-392 Put metadata to a standalone null master breaks CloudServer CRUD

lib/versioning/Version.ts

@@ -83,6 +83,33 @@ export class Version {
         return `{ "isPHD": true, "versionId": "${versionId}" }`;
     }
 
+    /**
+     * Appends a key-value pair to a JSON object represented as a string. It adds
+     * a comma if the object is not empty (i.e., not just '{}'). It assumes the input
+     * string is formatted as a JSON object.
+     *
+     * @param {string} stringifiedObject The JSON object as a string to which the key-value pair will be appended.
+     * @param {string} key The key to append to the JSON object.
+     * @param {string} value The value associated with the key to append to the JSON object.
+     * @returns {string} The updated JSON object as a string with the new key-value pair appended.
+     * @example
+     * _jsonAppend('{"existingKey":"existingValue"}', 'newKey', 'newValue');
+     * // returns '{"existingKey":"existingValue","newKey":"newValue"}'
+     */
+    static _jsonAppend(stringifiedObject: string, key: string, value: string): string {
+        // stringifiedObject value has the format of '{...}'
+        let index = stringifiedObject.length - 2;
+        while (stringifiedObject.charAt(index) === ' ') {
+            index -= 1;
+        }
+        const needComma = stringifiedObject.charAt(index) !== '{';
+        return (
+            `${stringifiedObject.slice(0, stringifiedObject.length - 1)}` +
+            (needComma ? ',' : '') +
+            `"${key}":"${value}"}`
+        );
+    }
+
     /**
      * Put versionId into an object in the (cheap) way of string manipulation,
      * instead of the more expensive alternative parsing and stringification.
@@ -93,14 +120,32 @@ export class Version {
      */
     static appendVersionId(value: string, versionId: string): string {
         // assuming value has the format of '{...}'
-        let index = value.length - 2;
-        while (value.charAt(index--) === ' ');
-        const comma = value.charAt(index + 1) !== '{';
-        return (
-            `${value.slice(0, value.length - 1)}` + // eslint-disable-line
-            (comma ? ',' : '') +
-            `"versionId":"${versionId}"}`
-        );
+        return Version._jsonAppend(value, 'versionId', versionId);
+    }
+
+    /**
+    * Updates or appends a `nullVersionId` property to a JSON-formatted string.
+    * This function first checks if the `nullVersionId` property already exists within the input string.
+    * If it exists, the function updates the `nullVersionId` with the new value provided.
+    * If it does not exist, the function appends a `nullVersionId` property with the provided value.
+    *
+    * @static
+    * @param {string} value - The JSON-formatted string that may already contain a `nullVersionId` property.
+    * @param {string} nullVersionId - The new value for the `nullVersionId` property to be updated or appended.
+    * @returns {string} The updated JSON-formatted string with the new `nullVersionId` value.
+    */
+    static updateOrAppendNullVersionId(value: string, nullVersionId: string): string {
+        // Check if "nullVersionId" already exists in the string
+        const nullVersionIdPattern = /"nullVersionId":"[^"]*"/;
+        const nullVersionIdExists = nullVersionIdPattern.test(value);
+
+        if (nullVersionIdExists) {
+            // Replace the existing nullVersionId with the new one
+            return value.replace(nullVersionIdPattern, `"nullVersionId":"${nullVersionId}"`);
+        } else {
+            // Append nullVersionId
+            return Version._jsonAppend(value, 'nullVersionId', nullVersionId);
+        }
     }
 
     /**

lib/versioning/VersioningRequestProcessor.ts

@@ -1,6 +1,6 @@
 import errors, { ArsenalError } from '../errors';
 import { Version } from './Version';
-import { generateVersionId as genVID } from './VersionID';
+import { generateVersionId as genVID, getInfVid } from './VersionID';
 import WriteCache from './WriteCache';
 import WriteGatheringManager from './WriteGatheringManager';
 
@@ -380,12 +380,41 @@ export default class VersioningRequestProcessor {
             const versionId = request.options.versionId;
             const versionKey = formatVersionKey(request.key, versionId);
             const ops = [{ key: versionKey, value: request.value }];
-            if (data === undefined ||
-                (Version.from(data).getVersionId() ?? '') >= versionId) {
-                // master does not exist or is not newer than put
-                // version and needs to be updated as well.
-                // Note that older versions have a greater version ID.
-                ops.push({ key: request.key, value: request.value });
+            const masterVersion = data !== undefined &&
+                Version.from(data);
+            if (masterVersion) {
+                const versionIdFromMaster = masterVersion.getVersionId();
+                // master key exists
+                if (versionIdFromMaster === undefined ||
+                    versionIdFromMaster >= versionId) {
+                        // master key is not newer than the put version
+                    let masterVersionId;
+                    let value = request.value;
+                    if (masterVersion.isNullVersion() && versionIdFromMaster) {
+                        // master key is a null version
+                        masterVersionId = versionIdFromMaster;
+                    } else if (versionIdFromMaster === undefined) {
+                        // master key does not have a versionID
+                        // => create one with the "infinite" version ID
+                        masterVersionId = getInfVid(this.replicationGroupId);
+                        masterVersion.setVersionId(masterVersionId);
+                    }
+                    if (masterVersionId) {
+                        // => create a new version key from the master version
+                        const masterVersionKey = formatVersionKey(key, masterVersionId);
+                        value = Version.updateOrAppendNullVersionId(request.value, masterVersionId);
+                        masterVersion.setNullVersion();
+                        ops.push({ key: masterVersionKey,
+                                   value: masterVersion.toString() });
+                    }
+                    // => update the master key, note that older
+                    //    versions have a greater version ID
+                    ops.push({ key, value });
+                }
+                // otherwise, master key is newer so do not update it
+            } else {
+                // master key does not exist: create it
+                ops.push({ key, value: request.value });
             }
             return callback(null, ops, versionId);
         });

tests/unit/versioning/Version.spec.js

@@ -0,0 +1,72 @@
+const { Version } = require('../../../lib/versioning/Version');
+
+describe('Version', () => {
+    describe('_jsonAppend', () => {
+        it('should append key-value pair to an empty object', () => {
+            const result = Version._jsonAppend('{}', 'versionId', '123');
+            expect(result).toBe('{"versionId":"123"}');
+        });
+
+        it('should append key-value pair to an object with existing properties', () => {
+            const result = Version._jsonAppend('{"existingKey":"existingValue"}', 'versionId', '123');
+            expect(result).toBe('{"existingKey":"existingValue","versionId":"123"}');
+        });
+
+        it('should append key-value pair to an object with existing key', () => {
+            const result = Version._jsonAppend('{"versionId":"0"}', 'versionId', '123');
+            expect(result).toBe('{"versionId":"0","versionId":"123"}');
+        });
+    });
+
+    describe('appendVersionId', () => {
+        it('should append versionId to an empty object', () => {
+            const emptyObject = '{}';
+            const versionId = '123';
+            const expected = '{"versionId":"123"}';
+            const result = Version.appendVersionId(emptyObject, versionId);
+            expect(result).toEqual(expected);
+        });
+
+        it('should append versionId to an object with existing properties', () => {
+            const existingObject = '{"key":"value"}';
+            const versionId = '456';
+            const expected = '{"key":"value","versionId":"456"}';
+            const result = Version.appendVersionId(existingObject, versionId);
+            expect(result).toEqual(expected);
+        });
+
+        it('should append versionId to an object with existing versionId', () => {
+            const objectWithVersionId = '{"key":"value","versionId":"old"}';
+            const versionId = 'new';
+            const expected = '{"key":"value","versionId":"old","versionId":"new"}';
+            const result = Version.appendVersionId(objectWithVersionId, versionId);
+            expect(result).toEqual(expected);
+        });
+    });
+
+    describe('updateOrAppendNullVersionId', () => {
+        it('should append nullVersionId when it does not exist', () => {
+            const initialValue = '{"key":"value"}';
+            const nullVersionId = '12345';
+            const expectedValue = '{"key":"value","nullVersionId":"12345"}';
+            const result = Version.updateOrAppendNullVersionId(initialValue, nullVersionId);
+            expect(result).toEqual(expectedValue);
+        });
+
+        it('should update nullVersionId when it exists', () => {
+            const initialValue = '{"key":"value","nullVersionId":"initial"}';
+            const nullVersionId = 'updated12345';
+            const expectedValue = '{"key":"value","nullVersionId":"updated12345"}';
+            const result = Version.updateOrAppendNullVersionId(initialValue, nullVersionId);
+            expect(result).toEqual(expectedValue);
+        });
+
+        it('should handle empty string by appending nullVersionId', () => {
+            const initialValue = '{}';
+            const nullVersionId = 'emptyCase12345';
+            const expectedValue = '{"nullVersionId":"emptyCase12345"}';
+            const result = Version.updateOrAppendNullVersionId(initialValue, nullVersionId);
+            expect(result).toEqual(expectedValue);
+        });
+    });
+});

tests/unit/versioning/VersioningRequestProcessor.spec.js

@@ -219,4 +219,142 @@ describe('test VSP', () => {
         }],
         done);
     });
+
+    it('should be able to put Metadata on top of a standalone null version', done => {
+        const versionId = '00000000000000999999PARIS  ';
+
+        async.waterfall([next => {
+            // simulate the creation of a standalone null version.
+            const request = {
+                db: 'foo',
+                key: 'bar',
+                value: '{"qux":"quz"}',
+                options: {},
+            };
+            vsp.put(request, logger, next);
+        },
+        (res, next) => {
+            // simulate a BackbeatClient.putMetadata
+            const request = {
+                db: 'foo',
+                key: 'bar',
+                value: `{"qux":"quz2","versionId":"${versionId}"}`,
+                options: {
+                    versioning: true,
+                    versionId,
+                },
+            };
+            vsp.put(request, logger, next);
+        },
+        (res, next) => {
+            wgm.list({}, logger, next);
+        },
+        (res, next) => {
+            const expectedListing = [
+                // master version should have the provided version id and a reference of the null version id.
+                {
+                    key: 'bar',
+                    value: `{"qux":"quz2","versionId":"${versionId}","nullVersionId":"99999999999999999999PARIS  "}`,
+                },
+                // the "internal" master version should have the provided version id.
+                {
+                    key: `bar\x00${versionId}`,
+                    value: `{"qux":"quz2","versionId":"${versionId}"}`,
+                },
+                // should create a version that represents the old null master with the infinite version id and
+                // the isNull property set to true.
+                {
+                    key: 'bar\x0099999999999999999999PARIS  ',
+                    value: '{"qux":"quz","versionId":"99999999999999999999PARIS  ","isNull":true}',
+                },
+            ];
+            assert.deepStrictEqual(res, expectedListing);
+            const request = {
+                db: 'foo',
+                key: 'bar',
+            };
+            vsp.get(request, logger, next);
+        },
+        (res, next) => {
+            const expectedGet = {
+                qux: 'quz2',
+                versionId,
+                nullVersionId: '99999999999999999999PARIS  ',
+            };
+            assert.deepStrictEqual(JSON.parse(res), expectedGet);
+            next();
+        }],
+        done);
+    });
+
+    it('should be able to put Metadata on top of a null suspended version', done => {
+        const versionId = '00000000000000999999PARIS  ';
+        let nullVersionId;
+
+        async.waterfall([next => {
+            // simulate the creation of a null suspended version.
+            const request = {
+                db: 'foo',
+                key: 'bar',
+                value: '{"qux":"quz","isNull":true}',
+                options: {
+                    versionId: '',
+                },
+            };
+            vsp.put(request, logger, next);
+        },
+        (res, next) => {
+            nullVersionId = JSON.parse(res).versionId;
+            // simulate a BackbeatClient.putMetadata
+            const request = {
+                db: 'foo',
+                key: 'bar',
+                value: `{"qux":"quz2","versionId":"${versionId}"}`,
+                options: {
+                    versioning: true,
+                    versionId,
+                },
+            };
+            vsp.put(request, logger, next);
+        },
+        (res, next) => {
+            wgm.list({}, logger, next);
+        },
+        (res, next) => {
+            const expectedListing = [
+                // master version should have the provided version id and a reference of the null version id.
+                {
+                    key: 'bar',
+                    value: `{"qux":"quz2","versionId":"${versionId}","nullVersionId":"${nullVersionId}"}`,
+                },
+                // the "internal" master version should have the provided version id.
+                {
+                    key: `bar\x00${versionId}`,
+                    value: `{"qux":"quz2","versionId":"${versionId}"}`,
+                },
+                // should create a version that represents the old null master with the infinite version id and
+                // the isNull property set to true.
+                {
+                    key: `bar\x00${nullVersionId}`,
+                    value: `{"qux":"quz","isNull":true,"versionId":"${nullVersionId}"}`,
+                },
+            ];
+            assert.deepStrictEqual(res, expectedListing);
+            const request = {
+                db: 'foo',
+                key: 'bar',
+            };
+            vsp.get(request, logger, next);
+        },
+        (res, next) => {
+            const expectedGet = {
+                qux: 'quz2',
+                versionId,
+                nullVersionId,
+            };
+            assert.deepStrictEqual(JSON.parse(res), expectedGet);
+            next();
+        }],
+        done);
+    });
 });