From b5a067f688ccab03f10bba3177f507f1cb964c36 Mon Sep 17 00:00:00 2001 From: Thomas Flament Date: Tue, 18 Aug 2026 16:55:42 +0200 Subject: [PATCH 1/2] Move the QueueEntry spec next to the model it tests createFromKafkaEntry coverage lived under tests/unit/replication/, while tests/unit/lib/models/ mirrors the source layout and already holds the sibling model specs. Pure relocation: only the two relative require paths change. Issue: BB-847 --- tests/unit/{replication => lib/models}/QueueEntry.spec.js | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) rename tests/unit/{replication => lib/models}/QueueEntry.spec.js (96%) diff --git a/tests/unit/replication/QueueEntry.spec.js b/tests/unit/lib/models/QueueEntry.spec.js similarity index 96% rename from tests/unit/replication/QueueEntry.spec.js rename to tests/unit/lib/models/QueueEntry.spec.js index 601c658f6..30ad73962 100644 --- a/tests/unit/replication/QueueEntry.spec.js +++ b/tests/unit/lib/models/QueueEntry.spec.js @@ -2,9 +2,8 @@ const assert = require('assert'); -const QueueEntry = - require('../../../lib/models/QueueEntry'); -const { replicationEntry } = require('../../utils/kafkaEntries'); +const QueueEntry = require('../../../../lib/models/QueueEntry'); +const { replicationEntry } = require('../../../utils/kafkaEntries'); describe('QueueEntry helper class', () => { describe('built from Kafka queue entry', () => { From f8b8eb2cd58b80be4bfab090d57582ab512005b7 Mon Sep 17 00:00:00 2001 From: Thomas Flament Date: Tue, 18 Aug 2026 17:06:41 +0200 Subject: [PATCH 2/2] Accept object-valued value and overheadFields in kafka entries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit createFromKafkaEntry parsed the kafka message body and then parsed the nested value (and overheadFields) a second time, requiring producers to send them as JSON text. The D/R metadata stream is built by a MongoDB change-stream pipeline, which cannot serialise a document to a string — there is no $toJson and $convert rejects documents — so it can only emit these fields as objects. Decode them only when they are strings and pass objects through. This is a superset of the existing format: producers that stringify, S3C ingestion and the oplog reader, take the identical path. A field that is neither a string nor an object now yields the usual malformed-entry error naming the offending type. Previously JSON.parse coerced it, so an entry whose value was a number reached ObjectQueueEntry and passed checkSanity, since only bucket and key are checked. Issue: BB-847 --- lib/models/QueueEntry.js | 24 ++++++++- .../unit/lib/models/ObjectQueueEntry.spec.js | 17 +++++++ tests/unit/lib/models/QueueEntry.spec.js | 50 +++++++++++++++++++ 3 files changed, 89 insertions(+), 2 deletions(-) diff --git a/lib/models/QueueEntry.js b/lib/models/QueueEntry.js index 516416e3b..3d0049f51 100644 --- a/lib/models/QueueEntry.js +++ b/lib/models/QueueEntry.js @@ -5,6 +5,26 @@ const BucketQueueEntry = require('./BucketQueueEntry'); const BucketMdQueueEntry = require('./BucketMdQueueEntry'); const DeleteOpQueueEntry = require('./DeleteOpQueueEntry'); +/** + * Decode a nested field of a kafka entry, which producers may send either + * as a JSON string or as an already-decoded object. + * + * @param {string|object} field - JSON string or decoded object + * @return {object} the decoded field + * @throws {Error} if the field is neither a JSON string nor an object; the + * caller turns this into a "malformed JSON in kafka entry" error + */ +function decodeField(field) { + if (typeof field === 'string') { + return JSON.parse(field); + } + if (typeof field === 'object') { + return field; + } + throw new Error( + `expected a JSON string or an object, got ${typeof field}`); +} + class QueueEntry { /** @@ -27,14 +47,14 @@ class QueueEntry { } let entry; if (record.type === 'del') { - const overheadFields = record.overheadFields && JSON.parse(record.overheadFields); + const overheadFields = record.overheadFields && decodeField(record.overheadFields); entry = new DeleteOpQueueEntry(record.bucket, record.key, overheadFields); } else if (record.bucket === usersBucket) { // BucketQueueEntry class just handles puts of keys // to usersBucket entry = new BucketQueueEntry(record.key, record.value); } else if (record.value) { - const metadataVal = JSON.parse(record.value); + const metadataVal = decodeField(record.value); if (metadataVal.mdBucketModelVersion) { // it's bucket metadata entry = new BucketMdQueueEntry(record.key, metadataVal); diff --git a/tests/unit/lib/models/ObjectQueueEntry.spec.js b/tests/unit/lib/models/ObjectQueueEntry.spec.js index d9cfb96dc..5a682524a 100644 --- a/tests/unit/lib/models/ObjectQueueEntry.spec.js +++ b/tests/unit/lib/models/ObjectQueueEntry.spec.js @@ -384,5 +384,22 @@ describe('ObjectQueueEntry', () => { assert.strictEqual(entry.getKey(), 'key'); assert.deepStrictEqual(entry.getOverheadField('foo'), 'bar'); }); + + it('should create a DeleteOpQueueEntry with object overhead fields', + () => { + const entry = QueueEntry.createFromKafkaEntry({ + value: JSON.stringify({ + type: 'del', + bucket: 'bucket', + key: 'key', + overheadFields: { foo: 'bar' }, + }), + }); + + assert(entry instanceof DeleteOpQueueEntry); + assert.strictEqual(entry.getBucket(), 'bucket'); + assert.strictEqual(entry.getKey(), 'key'); + assert.deepStrictEqual(entry.getOverheadField('foo'), 'bar'); + }); }); }); diff --git a/tests/unit/lib/models/QueueEntry.spec.js b/tests/unit/lib/models/QueueEntry.spec.js index 30ad73962..f430bc973 100644 --- a/tests/unit/lib/models/QueueEntry.spec.js +++ b/tests/unit/lib/models/QueueEntry.spec.js @@ -71,4 +71,54 @@ describe('QueueEntry helper class', () => { assert.strictEqual(completed1.getReplicationStatus(), 'COMPLETED'); }); }); + + describe('nested field decoding', () => { + function withDecodedValue(kafkaEntry) { + const record = JSON.parse(kafkaEntry.value); + record.value = JSON.parse(record.value); + return { ...kafkaEntry, value: JSON.stringify(record) }; + } + + it('should accept an object as the entry value', () => { + const fromString = QueueEntry.createFromKafkaEntry(replicationEntry); + const fromObject = QueueEntry.createFromKafkaEntry( + withDecodedValue(replicationEntry)); + + assert.strictEqual(fromObject.error, undefined); + assert.deepStrictEqual(fromObject.getValue(), fromString.getValue()); + assert.strictEqual(fromObject.getBucket(), fromString.getBucket()); + assert.strictEqual(fromObject.getObjectKey(), + fromString.getObjectKey()); + }); + + it('should report a malformed value string', () => { + const entry = QueueEntry.createFromKafkaEntry({ + value: JSON.stringify({ + type: 'put', + bucket: 'bucket', + key: 'key', + value: 'not json', + }), + }); + + assert.strictEqual(entry.error.message, + 'malformed JSON in kafka entry'); + }); + + it('should reject a value that is neither a string nor an object', + () => { + const entry = QueueEntry.createFromKafkaEntry({ + value: JSON.stringify({ + type: 'put', + bucket: 'bucket', + key: 'key', + value: 42, + }), + }); + + assert.strictEqual(entry.error.message, + 'malformed JSON in kafka entry'); + assert.match(entry.error.description, /got number/); + }); + }); });