Cap the heap unanswered messages hold, and detach them from the chunk they arrived in #28
+2
-2
@@ -41,8 +41,8 @@
|
|||||||
tuned low**: it now holds several times fewer segments, and an incomplete message evicted over
|
tuned low**: it now holds several times fewer segments, and an incomplete message evicted over
|
||||||
the cap is lost, since its segments were already answered.
|
the cap is lost, since its segments were already answered.
|
||||||
- Unanswered `sms` messages are capped at 64 MiB per session by the `maxOctets` charge, beside the
|
- Unanswered `sms` messages are capped at 64 MiB per session by the `maxOctets` charge, beside the
|
||||||
1000-message cap: the oldest is dropped with a warning, as over the count. A peer faster than an
|
1000-message cap: the oldest is dropped with a warning, as over the count. The library used to
|
||||||
application answering asynchronously could hold gigabytes per session.
|
hold up to 1000 messages of any size for an application that answered none of them.
|
||||||
- `server()` refuses a `maxOctets` below 1 or not a whole number, `Infinity` included, like its
|
- `server()` refuses a `maxOctets` below 1 or not a whole number, `Infinity` included, like its
|
||||||
other limits. `server({ maxOctets: 0 })` used to start and then refuse every multipart message.
|
other limits. `server({ maxOctets: 0 })` used to start and then refuse every multipart message.
|
||||||
- `callback_num`, `callback_num_atag`, `callback_num_pres_ind`, `broadcast_area_identifier` and
|
- `callback_num`, `callback_num_atag`, `callback_num_pres_ind`, `broadcast_area_identifier` and
|
||||||
|
|||||||
@@ -128,7 +128,7 @@ export const defaults = {
|
|||||||
heldMessageTimeout: 300_000,
|
heldMessageTimeout: 300_000,
|
||||||
maxDlrMerges: 1000,
|
maxDlrMerges: 1000,
|
||||||
maxHeldMessages: 1000,
|
maxHeldMessages: 1000,
|
||||||
maxHeldOctets: defaultMaxOctets,
|
maxHeldOctets: 64 * 1024 * 1024,
|
||||||
maxOutstanding: 10,
|
maxOutstanding: 10,
|
||||||
maxReassembly: 1000,
|
maxReassembly: 1000,
|
||||||
reassemblyTimeout: 300_000,
|
reassemblyTimeout: 300_000,
|
||||||
|
|||||||
@@ -29,7 +29,6 @@ import { client } from '../src/client.ts';
|
|||||||
import { closeAfter, closeListenerAfter } from './teardown.ts';
|
import { closeAfter, closeListenerAfter } from './teardown.ts';
|
||||||
import { concatOf } from '../src/concat.ts';
|
import { concatOf } from '../src/concat.ts';
|
||||||
import { consts } from '../src/defs/constants.ts';
|
import { consts } from '../src/defs/constants.ts';
|
||||||
import { detach } from '../src/retained-pdu.ts';
|
|
||||||
import { errors } from '../src/defs/errors.ts';
|
import { errors } from '../src/defs/errors.ts';
|
||||||
import { paramNumber, paramText } from '../src/defs/types.ts';
|
import { paramNumber, paramText } from '../src/defs/types.ts';
|
||||||
import { server } from '../src/server.ts';
|
import { server } from '../src/server.ts';
|
||||||
@@ -1504,7 +1503,8 @@ describe('held message bounds', () => {
|
|||||||
|
|
||||||
// submitPdu() holds 1026 octets by the maxOctets charge: its object, and the three text fields.
|
// submitPdu() holds 1026 octets by the maxOctets charge: its object, and the three text fields.
|
||||||
test('drops the message held longest once the octets held pass the cap', () => {
|
test('drops the message held longest once the octets held pass the cap', () => {
|
||||||
const held = new HeldMessages({ log: silentLog, max: 10, maxOctets: 2100, timeout: 10_000 });
|
let now = 0;
|
||||||
|
const held = new HeldMessages({ log: silentLog, max: 10, maxOctets: 2100, now: () => now, timeout: 10_000 });
|
||||||
const oldest = message(1);
|
const oldest = message(1);
|
||||||
|
|
||||||
held.hold(oldest);
|
held.hold(oldest);
|
||||||
@@ -1517,6 +1517,26 @@ describe('held message bounds', () => {
|
|||||||
assert.equal(held.size, 2);
|
assert.equal(held.size, 2);
|
||||||
assert.equal(held.has(oldest), false);
|
assert.equal(held.has(oldest), false);
|
||||||
|
|
||||||
|
// A message that leaves any other way gives its octets back, so two still fit afterwards.
|
||||||
|
const answered = message(4);
|
||||||
|
|
||||||
|
held.hold(answered);
|
||||||
|
held.release(answered);
|
||||||
|
held.hold(message(5));
|
||||||
|
assert.equal(held.size, 2, 'after a release');
|
||||||
|
|
||||||
|
now = 20_000;
|
||||||
|
held.sweep();
|
||||||
|
now = 0;
|
||||||
|
held.hold(message(6));
|
||||||
|
held.hold(message(7));
|
||||||
|
assert.equal(held.size, 2, 'after a sweep');
|
||||||
|
|
||||||
|
held.clear();
|
||||||
|
held.hold(message(8));
|
||||||
|
held.hold(message(9));
|
||||||
|
assert.equal(held.size, 2, 'after a clear');
|
||||||
|
|
||||||
held.clear();
|
held.clear();
|
||||||
});
|
});
|
||||||
|
|
||||||
@@ -1534,13 +1554,30 @@ describe('held message bounds', () => {
|
|||||||
held.clear();
|
held.clear();
|
||||||
});
|
});
|
||||||
|
|
||||||
test('holds a message detached from the chunk it was read from', () => {
|
test('holds a message detached from the chunk it was read from', async t => {
|
||||||
|
const session = new Session({ sock: new net.Socket() });
|
||||||
|
|
||||||
|
closeAfter(t, session);
|
||||||
|
session.boundAs = 'transceiver';
|
||||||
|
|
||||||
|
const incoming = new IncomingRequests({
|
||||||
|
dlrMerger: new DlrMerger({ log: silentLog, max: 10, timeout: 10_000 }),
|
||||||
|
log: silentLog,
|
||||||
|
sendPastDrain: () => Promise.resolve({ err: new Error('never sent') }),
|
||||||
|
session,
|
||||||
|
});
|
||||||
const chunk = Buffer.alloc(64 * 1024);
|
const chunk = Buffer.alloc(64 * 1024);
|
||||||
const carried = submitPdu(1);
|
const carried = submitPdu(1);
|
||||||
const retained = detach({ ...carried, params: { ...carried.params, short_message: chunk.subarray(16, 20) } });
|
let received: Sms | undefined;
|
||||||
|
|
||||||
assert.ok(Buffer.isBuffer(retained.params.short_message));
|
session.on('sms', sms => { received = sms; });
|
||||||
assert.notEqual(retained.params.short_message.buffer, chunk.buffer);
|
await incoming.handle({ ...carried, params: { ...carried.params, short_message: chunk.subarray(16, 20) } });
|
||||||
|
|
||||||
|
const retained = received?.pduObjs[0]?.params.short_message;
|
||||||
|
|
||||||
|
assert.ok(Buffer.isBuffer(retained));
|
||||||
|
assert.notEqual(retained.buffer, chunk.buffer);
|
||||||
|
incoming.clear();
|
||||||
});
|
});
|
||||||
|
|
||||||
test('gives up on a message the application never answers', () => {
|
test('gives up on a message the application never answers', () => {
|
||||||
|
|||||||
@@ -6,6 +6,17 @@ hard rules first — they constrain every item below.
|
|||||||
This is a working file that sets its own rules. The documentation conventions in AGENTS.md do not
|
This is a working file that sets its own rules. The documentation conventions in AGENTS.md do not
|
||||||
govern it, and nothing here is a source anything else may cite.
|
govern it, and nothing here is a source anything else may cite.
|
||||||
|
|
||||||
|
## Security
|
||||||
|
|
||||||
|
- [ ] **Bound what a peer can make the application hold, not only the library.** An `sms` the
|
||||||
|
application is still answering pins its PDUs through `Sms.pduObjs` and its `sendResp`/`sendDlr`
|
||||||
|
closures, so the held-message caps free nothing while it works, and nothing slows the peer: a
|
||||||
|
peer faster than an application answering asynchronously (a DB write per message) grows the
|
||||||
|
heap without bound. Each eviction by the caps also stops the drain waiting for a message still
|
||||||
|
being answered, so `close()` can cut it off and the peer re-sends it. Flow control is a wire
|
||||||
|
change under goal 4 and the maintainer's call: answer `ESME_RTHROTTLED`, or stop reading the
|
||||||
|
socket, once the held count or octets are at the cap. From the stability review of #28.
|
||||||
|
|
||||||
## Status
|
## Status
|
||||||
|
|
||||||
The rewrite is **feature complete and green**: the suite, lint and typecheck are clean on Node 18
|
The rewrite is **feature complete and green**: the suite, lint and typecheck are clean on Node 18
|
||||||
|
|||||||
Reference in New Issue
Block a user