Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,10 @@

### NEXT

- Node: Fix the `Channel` buffer builder being left unusable when a message serialization fails ([PR #1999](https://github.com/versatica/mediasoup/pull/1999)).
- Which made every subsequent request and notification in the worker fail.
- Also validate `encoding.maxBitrate` in `transport.produce()`. It must be a non negative integer that fits in an `uint64`, and a `bigint` is now also accepted.

### 3.30.0

- Update `ankerl/unordered_dense` Meson subproject to 5.3.2 ([PR #1985](https://github.com/versatica/mediasoup/pull/1985)).
Expand Down
27 changes: 21 additions & 6 deletions node/src/Channel.ts
Original file line number Diff line number Diff line change
Expand Up @@ -191,8 +191,16 @@ export class Channel extends EnhancedEventEmitter {

/**
* flatbuffer builder.
*
* NOTE: It must be retrieved once per message, right before building
* its body.
* It's reset here so a new message always starts with a clean builder,
* which also recovers it if a previous build threw in the middle of the
* serialization (which would otherwise break every future message).
*/
get bufferBuilder(): flatbuffers.Builder {
this.#bufferBuilder.clear();

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't love this solution because it basically means that, in case of corrupted buffer, we leave it that way until the next time we need to use the buffer.


return this.#bufferBuilder;
}

Expand Down Expand Up @@ -243,6 +251,12 @@ export class Channel extends EnhancedEventEmitter {
);
}

// If there is no body then the buffer builder holds nothing we need, so

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is also post-defensive approach that we are supposed to hate.

// reset it in case a previous message left content in it.
if (!bodyType || !bodyOffset) {
this.#bufferBuilder.clear();
}

const handlerIdOffset = this.#bufferBuilder.createString(handlerId ?? '');

let notificationOffset: number;
Expand Down Expand Up @@ -278,9 +292,6 @@ export class Channel extends EnhancedEventEmitter {
// do not point to the builder buffer overriding others info.
const buffer = new Uint8Array(this.#bufferBuilder.asUint8Array());

// Clear the buffer builder so it's reused for the next request.
this.#bufferBuilder.clear();

if (buffer.byteLength > MESSAGE_MAX_LEN) {
logger.error(`notify() | notification too big [event:${Event[event]}]`);

Expand Down Expand Up @@ -318,6 +329,13 @@ export class Channel extends EnhancedEventEmitter {
}

const id = this.#nextId;

// If there is no body then the buffer builder holds nothing we need, so
// reset it in case a previous message left content in it.
if (!bodyType || !bodyOffset) {
this.#bufferBuilder.clear();
}

const handlerIdOffset = this.#bufferBuilder.createString(handlerId ?? '');

let requestOffset: number;
Expand Down Expand Up @@ -355,9 +373,6 @@ export class Channel extends EnhancedEventEmitter {
// do not point to the builder buffer overriding others info.
const buffer = new Uint8Array(this.#bufferBuilder.asUint8Array());

// Clear the buffer builder so it's reused for the next request.
this.#bufferBuilder.clear();

if (buffer.byteLength > MESSAGE_MAX_LEN) {
throw new Error(`request too big [method:${Method[method]}]`);
}
Expand Down
27 changes: 27 additions & 0 deletions node/src/ortc.ts
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,10 @@ const DynamicPayloadTypes = [
99,
];

// Biggest value that fits in an uint64, which is what the worker expects for
// bitrate related fields.
const MaxUint64 = 2n ** 64n - 1n;

// TODO: Remove this if we switch to 'sendrecv' in Dependency-Descriptor header
// extension.
//
Expand Down Expand Up @@ -1310,6 +1314,29 @@ function validateAndNormalizeRtpEncodingParameters(
) {
throw new TypeError('invalid encoding.scalabilityMode');
}

// maxBitrate is optional, it can be a number or a bigint.
const maxBitrate: number | bigint | undefined = encoding.maxBitrate;

if (!maxBitrate) {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is not valid since 0 is a valid value but it enters this condition. It works by accident but this solution is not valid for other similar cases such as transport.setMaxIncomingBitrate(bitrate) where 0 means "no limit".

delete encoding.maxBitrate;
} else if (typeof maxBitrate === 'bigint') {
if (maxBitrate < 0n || maxBitrate > MaxUint64) {
throw new TypeError('invalid encoding.maxBitrate');
}
} else if (typeof maxBitrate === 'number') {
// NOTE: Compare as bigint. MaxUint64 rounds up to 2^64 if converted to a
// number, so an out of range 2^64 would wrongly be accepted.
if (
!Number.isInteger(maxBitrate) ||
maxBitrate < 0 ||
BigInt(maxBitrate) > MaxUint64
) {
throw new TypeError('invalid encoding.maxBitrate');
}
} else {
throw new TypeError('invalid encoding.maxBitrate');
}
}

/**
Expand Down
2 changes: 1 addition & 1 deletion node/src/rtpParametersFbsUtils.ts
Original file line number Diff line number Diff line change
Expand Up @@ -194,7 +194,7 @@ export function serializeRtpEncodingParameters(
}

// Add max bitrate.
if (encoding.maxBitrate !== undefined) {
if (encoding.maxBitrate) {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Again this condition breaks the valid case of value 0.

FbsRtpEncodingParameters.addMaxBitrate(
builder,
BigInt(encoding.maxBitrate)
Expand Down
135 changes: 135 additions & 0 deletions node/src/test/test-Producer.ts
Original file line number Diff line number Diff line change
Expand Up @@ -361,6 +361,141 @@ test('webRtcTransport1.produce() with wrong arguments rejects with TypeError', a
},
})
).rejects.toThrow(TypeError);

// Invalid maxBitrate.
await expect(
ctx.webRtcTransport1!.produce({
kind: 'audio',
rtpParameters: {
codecs: [],
headerExtensions: [],
// @ts-expect-error --- Testing purposes.
encodings: [{ ssrc: 1111, maxBitrate: '2000000' }],
rtcp: { cname: 'qwerty' },
},
})
).rejects.toThrow(TypeError);

// Non integer maxBitrate.
await expect(
ctx.webRtcTransport1!.produce({
kind: 'audio',
rtpParameters: {
codecs: [],
headerExtensions: [],
encodings: [{ ssrc: 1111, maxBitrate: 2000000.5 }],
rtcp: { cname: 'qwerty' },
},
})
).rejects.toThrow(TypeError);

// Negative bigint maxBitrate.
await expect(
ctx.webRtcTransport1!.produce({
kind: 'audio',
rtpParameters: {
codecs: [],
headerExtensions: [],
// @ts-expect-error --- Testing purposes.
encodings: [{ ssrc: 1111, maxBitrate: -2000000n }],
rtcp: { cname: 'qwerty' },
},
})
).rejects.toThrow(TypeError);

// Too big bigint maxBitrate (2^64, so it doesn't fit in an uint64).
await expect(
ctx.webRtcTransport1!.produce({
kind: 'audio',
rtpParameters: {
codecs: [],
headerExtensions: [],
// @ts-expect-error --- Testing purposes.
encodings: [{ ssrc: 1111, maxBitrate: 2n ** 64n }],
rtcp: { cname: 'qwerty' },
},
})
).rejects.toThrow(TypeError);

// Too big number maxBitrate (it doesn't fit in an uint64).
await expect(
ctx.webRtcTransport1!.produce({
kind: 'audio',
rtpParameters: {
codecs: [],
headerExtensions: [],
encodings: [{ ssrc: 1111, maxBitrate: 1e30 }],
rtcp: { cname: 'qwerty' },
},
})
).rejects.toThrow(TypeError);

// 2^64 as a number maxBitrate (it doesn't fit in an uint64 either, and it
// must not pass due to Number(MaxUint64) rounding up to 2^64).
await expect(
ctx.webRtcTransport1!.produce({
kind: 'audio',
rtpParameters: {
codecs: [],
headerExtensions: [],
encodings: [{ ssrc: 1111, maxBitrate: 18446744073709551616 }],
rtcp: { cname: 'qwerty' },
},
})
).rejects.toThrow(TypeError);
}, 2000);

test('webRtcTransport1.produce() with bigint encoding.maxBitrate succeeds', async () => {
await expect(
ctx.webRtcTransport1!.produce({
kind: 'video',
rtpParameters: {
mid: 'VIDEO',
codecs: [
{
mimeType: 'video/VP8',
payloadType: 112,
clockRate: 90000,
},
],
headerExtensions: [],
// @ts-expect-error --- Testing purposes.
encodings: [{ ssrc: 22222222, maxBitrate: 2n ** 64n - 1n }],
rtcp: { cname: 'video-1' },
},
})
).resolves.toBeDefined();
}, 2000);

// Regression test. A non TypeScript client may send null in optional fields.
// null in encoding.maxBitrate used to throw "Cannot convert null to a BigInt"
// in the middle of the Channel buffer builder serialization, leaving it in a
// nested state and hence breaking every subsequent request in the Worker.
test('webRtcTransport1.produce() with null encoding.maxBitrate succeeds and does not break the Channel', async () => {
const videoProducer = await ctx.webRtcTransport1!.produce({
kind: 'video',
rtpParameters: {
mid: 'VIDEO',
codecs: [
{
mimeType: 'video/VP8',
payloadType: 112,
clockRate: 90000,
},
],
headerExtensions: [],
// @ts-expect-error --- Testing purposes.
encodings: [{ ssrc: 22222222, maxBitrate: null }],
rtcp: { cname: 'video-1' },
},
});

expect(videoProducer.rtpParameters.encodings![0]!.maxBitrate).toBeUndefined();

// The Channel buffer builder must still be usable.
await expect(ctx.router!.dump()).resolves.toBeDefined();

videoProducer.close();
}, 2000);

test('webRtcTransport1.produce() with unsupported codecs rejects with UnsupportedError', async () => {
Expand Down
30 changes: 30 additions & 0 deletions node/src/test/test-WebRtcTransport.ts
Original file line number Diff line number Diff line change
Expand Up @@ -688,6 +688,36 @@ test('transport.enableTraceEvent() with wrong arguments rejects with TypeError',
).rejects.toThrow(TypeError);
}, 2000);

// Regression test. A message body serialization that threw in the middle used
// to leave the Channel buffer builder in a nested state, making every
// subsequent request and notification in the Worker fail forever.
test('Channel recovers if a message serialization fails in the middle', async () => {
const webRtcTransport = await ctx.router!.createWebRtcTransport({
listenInfos: [
{ protocol: 'udp', ip: '127.0.0.1', portRange: { min: 2000, max: 3000 } },
],
});

// API not exposed in the interface.
const channel = (webRtcTransport as WebRtcTransportImpl).channelForTesting;

// Simulate a message body serialization that throws in the middle by starting
// an object and never ending it.
channel.bufferBuilder.startObject(1);

// A request without body must still succeed.
await expect(webRtcTransport.dump()).resolves.toBeDefined();

channel.bufferBuilder.startObject(1);

// A request with body must still succeed.
await expect(
webRtcTransport.setMaxIncomingBitrate(100000)
).resolves.toBeUndefined();

webRtcTransport.close();
}, 2000);

test('WebRtcTransport events succeed', async () => {
const webRtcTransport = await ctx.router!.createWebRtcTransport({
listenInfos: [
Expand Down