chore(deps): bump aedes-persistence from 8.1.3 to 9.0.1

[dependabot]

Bumps aedes-persistence from 8.1.3 to 9.0.0.

Release notes

Sourced from aedes-persistence's releases.

Release 9.0.0

• chore: es5 to es6, drop nodejs 12 (#82) (6132db2)
• chore: update dependencies (#81) (883c249)

Commits

• a374197 Release 9.0.0
• 6132db2 chore: es5 to es6, drop nodejs 12 (#82)
• 883c249 chore: update dependencies (#81)
• See full diff in compare view

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting @dependabot rebase.

---

Dependabot commands and options

You can trigger Dependabot actions by commenting on this PR:

• @dependabot rebase will rebase this PR
• @dependabot recreate will recreate this PR, overwriting any edits that have been made to it
• @dependabot merge will merge this PR after your CI passes on it
• @dependabot squash and merge will squash and merge this PR after your CI passes on it
• @dependabot cancel merge will cancel a previously requested merge and block automerging
• @dependabot reopen will reopen this PR if it is closed
• @dependabot close will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
• @dependabot ignore this major version will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
• @dependabot ignore this minor version will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
• @dependabot ignore this dependency will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)

[robertsLando]

We should drop nodejs 12 here too

cc @seriousme

[robertsLando]

@seriousme Tests are still failing here, any clue?

[robertsLando]

Tests failing

[seriousme]

Issue seems to be triggered by

aedes/test/qos1.js

Lines 156 to 220 in 7c13920

	test('publish QoS 1 and empty offline queue', function (t) {
	t.plan(13)
	const broker = aedes()
	t.teardown(broker.close.bind(broker))
	const publisher = connect(setup(broker), { clean: false })
	const subscriberClient = {
	id: 'abcde'
	}
	const subscriber = connect(setup(broker), { clean: false, clientId: subscriberClient.id })
	const expected = {
	cmd: 'publish',
	topic: 'hello',
	qos: 1,
	dup: false,
	retain: false
	}
	const expectedAck = {
	cmd: 'puback',
	retain: false,
	qos: 0,
	dup: false,
	length: 2,
	messageId: 10
	}
	const sent = {
	cmd: 'publish',
	topic: 'hello',
	payload: 'world',
	qos: 1,
	messageId: 10,
	retain: false,
	dup: false
	}
	const queue = []
	subscribe(t, subscriber, 'hello', 1, function () {
	publisher.outStream.on('data', function (packet) {
	t.same(packet, expectedAck, 'ack packet must patch')
	})
	subscriber.outStream.on('data', function (packet) {
	queue.push(packet)
	delete packet.payload
	delete packet.length
	t.not(packet.messageId, undefined, 'messageId is assigned a value')
	t.not(packet.messageId, 10, 'messageId should be unique')
	expected.messageId = packet.messageId
	t.same(packet, expected, 'publish packet must patch')
	if (queue.length === 2) {
	setImmediate(() => {
	broker.clients[subscriberClient.id].emptyOutgoingQueue(function () {
	for (let i = 0; i < queue.length; i++) {
	broker.persistence.outgoingClearMessageId(subscriberClient, queue[i], function (_, origPacket) {
	t.equal(!!origPacket, false, 'Packet has been removed')
	})
	}
	})
	})
	}
	})
	publisher.inStream.write(sent)
	sent.payload = 'world2world'
	publisher.inStream.write(sent)
	})
	})

I will have a look tonight as last weeks Easter leave is over ;-)

[robertsLando]

No rush, thanks @seriousme :)

[seriousme]

I did some research and found two issues in test/qos1.js

aedes/test/qos1.js

Lines 448 to 511 in 7c13920

	test('resend publish on non-clean reconnect QoS 1', function (t) {
	t.plan(8)
	const broker = aedes()
	t.teardown(broker.close.bind(broker))
	const opts = { clean: false, clientId: 'abcde' }
	let subscriber = connect(setup(broker), opts)
	const subscriberClient = {
	id: opts.clientId
	}
	const expected = {
	cmd: 'publish',
	topic: 'hello',
	payload: Buffer.from('world'),
	qos: 1,
	dup: false,
	length: 14,
	retain: false
	}
	subscribe(t, subscriber, 'hello', 1, function () {
	subscriber.inStream.end()
	const publisher = connect(setup(broker))
	publisher.inStream.write({
	cmd: 'publish',
	topic: 'hello',
	payload: 'world',
	qos: 1,
	messageId: 42
	})
	publisher.inStream.write({
	cmd: 'publish',
	topic: 'hello',
	payload: 'world world',
	qos: 1,
	messageId: 42
	})
	publisher.outStream.once('data', function (packet) {
	t.equal(packet.cmd, 'puback')
	subscriber = connect(setup(broker), opts)
	subscriber.outStream.once('data', function (packet) {
	subscriber.inStream.write({
	cmd: 'puback',
	messageId: packet.messageId
	})
	t.not(packet.messageId, 42, 'messageId must differ')
	delete packet.messageId
	t.same(packet, expected, 'packet must match')
	setImmediate(() => {
	const stream = broker.persistence.outgoingStream(subscriberClient)
	stream.pipe(concat(function (list) {
	t.equal(list.length, 1, 'should remain one item in queue')
	t.same(list[0].payload, Buffer.from('world world'), 'packet must match')
	}))
	})
	})
	})
	})
	})

Here the problem is that 32 messages are being sent (2x highwatermark) out and only 17 come back.
Reducing it to the high watermark makes this test pass.
That would suggest an issue with backpressure. Looking at aedes-persistence there is little that can go wrong:
https://github.com/moscajs/aedes-persistence/blob/master/persistence.js#L232-L234
It just creates a Readable stream from an array. I checked the array size and it is 32 at the moment that the Readable stream is created.

aedes/test/qos1.js

Lines 156 to 220 in 7c13920

	test('publish QoS 1 and empty offline queue', function (t) {
	t.plan(13)
	const broker = aedes()
	t.teardown(broker.close.bind(broker))
	const publisher = connect(setup(broker), { clean: false })
	const subscriberClient = {
	id: 'abcde'
	}
	const subscriber = connect(setup(broker), { clean: false, clientId: subscriberClient.id })
	const expected = {
	cmd: 'publish',
	topic: 'hello',
	qos: 1,
	dup: false,
	retain: false
	}
	const expectedAck = {
	cmd: 'puback',
	retain: false,
	qos: 0,
	dup: false,
	length: 2,
	messageId: 10
	}
	const sent = {
	cmd: 'publish',
	topic: 'hello',
	payload: 'world',
	qos: 1,
	messageId: 10,
	retain: false,
	dup: false
	}
	const queue = []
	subscribe(t, subscriber, 'hello', 1, function () {
	publisher.outStream.on('data', function (packet) {
	t.same(packet, expectedAck, 'ack packet must patch')
	})
	subscriber.outStream.on('data', function (packet) {
	queue.push(packet)
	delete packet.payload
	delete packet.length
	t.not(packet.messageId, undefined, 'messageId is assigned a value')
	t.not(packet.messageId, 10, 'messageId should be unique')
	expected.messageId = packet.messageId
	t.same(packet, expected, 'publish packet must patch')
	if (queue.length === 2) {
	setImmediate(() => {
	broker.clients[subscriberClient.id].emptyOutgoingQueue(function () {
	for (let i = 0; i < queue.length; i++) {
	broker.persistence.outgoingClearMessageId(subscriberClient, queue[i], function (_, origPacket) {
	t.equal(!!origPacket, false, 'Packet has been removed')
	})
	}
	})
	})
	}
	})
	publisher.inStream.write(sent)
	sent.payload = 'world2world'
	publisher.inStream.write(sent)
	})
	})

This seems to be a timing issue where results from outgoingClearMessageId come back in the wrong order in the test script, but I need to spend some more time on sorting this out.

Any suggestions would be gratefully appreciated ;-)

Kind regards,
Hans

[robertsLando]

Here the problem is that 32 messages are being sent (2x highwatermark) out and only 17 come back.
Reducing it to the high watermark makes this test pass.

So the problem seems related to the stream that is not handling backpressure correctly, right? Did you try with other nodejs versions like 16/18 to see if there is any difference?

This seems to be a timing issue where results from outgoingClearMessageId come back in the wrong order in the test script, but I need to spend some more time on sorting this out.

Could it be the order is reversed for some reason?

cc @gnought

[gnought]

A good work, @seriousme
Need to study the changes first in my spare time. @robertsLando

[seriousme]

Well this is where it gets interesting:

test publish QoS 1 and empty offline queue succeeds on NodeJS 16 and 18 without modification !
The problem lies in emptyOutgoingQueue which is basically a Transform on a stream. The Readstream is created with 2 items in the outgoing array, but the transform only handles the first one, hence the second one is never deleted and triggers the test to fail.

The other error is resend many publish on non-clean reconnect QoS 1 (and not the resend publish on non-clean reconnect QoS 1) This test still fails on NodeJS 16 and 18.

I tried to replicate the failure outside Mosca but I cannot get it to fail:

const { Readable, Transform, pipeline } = require("stream")
const outgoing = []
const total = 32


// busy loop such that it cannot be optimized out
function doWork() {
    let proof = 0
    for (let i = 0; i < 1000000; i++) {
        proof += Math.random()
    }
    return proof
}

for (let sent = 0; sent < total; sent++) {
    outgoing.push({
        cmd: 'publish',
        topic: 'hello',
        payload: 'message-' + sent,
        qos: 1,
        messageId: sent
    })
}

function filter(packet, enc, next) {
    setImmediate(()=> {
        console.log(packet,doWork())
        next()
    })
}

const readable = new Readable.from(outgoing)
const done = () => console.log("done")

function through(transform) {
    return new Transform({
        objectMode: true,
        transform
    })
}

pipeline(readable, through(filter), done)

Works like a charm, even on Node 14..
I also tried without the doWork loop and without the setImmediate, but all variants work.

Any suggestions ?

[seriousme]

Btw: I also ran the above code using Tap, just to rule out that it was a Tap failure, and that did not fail either
So the mystery continues

[seriousme]

Ok I found it :-)

It is indeed a race condition !
The Readable.from(outgoing) in aedes-persistence reads directly on the outgoing array that outgoingClearMessageId modifies.

This can be observered if we change the filter in the code above to:

function filter(packet, enc, next) {
    setImmediate(() => {
        console.log(packet, doWork())
        const toRemove = outgoing.findIndex(p => p.messageId === packet.messageId)
        if (toRemove > 0) {
            outgoing.splice(toRemove, 1)
        }
        next()
    })
}

we only see 17 messages , numbers 0-16 ;-)
If we remove the setImmediate it all works fine, until we remove the doWork() and then we see only the odd message numbers (0,1,3,5 ... 31)

The solution is trivial and was present in 8.1.3 although I did not notice this before :-(
We just need to change:

const readable = new Readable.from(outgoing)

to:

const readable = new Readable.from([].concat(outgoing))

As this will do a shallow clone of the array and the Readable stream can continue without interference by the splice in the filter.

I will do a PR on aedes-persistence to fix this , and add a test as well ;-)

Sorry for any inconvenience,
Hans

[robertsLando]

test publish QoS 1 and empty offline queue succeeds on NodeJS 16 and 18 without modification !

I'm quite sure @mcollina knows why this happens as he is implied in Nodejs development

About the second test failing there must be something weird going on on this function:

https://github.com/moscajs/aedes/blob/main/lib/handlers/connect.js#L224

It's using persistence outgoingStream that is actually returning a stream from an array of objects

In the previous imlementation it was a bit different as it was actually copy the array before passing it to from2 to create the stream

[robertsLando]

Ok I found it :-)

Sorry I have just read your answer, yeah the problem is the concat, without that there could be some race condition as you are working in the same array instead of creating a copy of it :)

[dependabot]

A newer version of aedes-persistence exists, but since this PR has been edited by someone other than Dependabot I haven't updated it. You'll get a PR for the updated version as normal once this PR is merged.

[robertsLando]

Let's see how test goes now

[robertsLando]

@seriousme interesting thing now seems there is a test failing on mac os nodejs 18 :(

[seriousme]

@seriousme interesting thing now seems there is a test failing on mac os nodejs 18 :(

Thats 1 out of 9 variants, which makes it a bit tricky to debug. Esp. since the module itself is platform independent as it doesn't do any I/O ;-)

Can we rule out any github runner deficiency ? E.g. can you rerun the failed github job ?

[seriousme]

Did a rerun on my own fork using Node 18 on MacOs latest and that succeeded.
So I'm tempted to say that this is a github runner issue :-(

Kind regards,
Hans
ps. on my local chromebook the non-blocking test (with 500 clients in parallel) often fails so the test might require more resources than a simple chromebook can provide ;-)