Cover the receipt refusal and the sweep that wakes a drain
This commit is contained in:
@@ -407,10 +407,9 @@ Grouped by what each one constrains.
|
|||||||
up answering the same question two different ways at admit and at release. For the same reason the
|
up answering the same question two different ways at admit and at release. For the same reason the
|
||||||
retry in `pastDrain()` asks `gate.isUp()` rather than `linkDown()`, which also reads the socket — a
|
retry in `pastDrain()` asks `gate.isUp()` rather than `linkDown()`, which also reads the socket — a
|
||||||
condition that loops on something the gate does not gate on spins against a gate that admits it
|
condition that loops on something the gate does not gate on spins against a gate that admits it
|
||||||
straight back.
|
straight back. `LinkGate.returning` is a copy of `retrying()` taken at teardown, and stays true
|
||||||
`LinkGate.returning` is a copy of `retrying()` taken at teardown, and stays true only because
|
only because nothing stops the reconnect loop without `emitClose()` following it: `drain()` and
|
||||||
nothing stops the reconnect loop without `emitClose()` following it: `drain()` and `end()` are the
|
`end()` are the only callers of `stop()`. A third caller has to shut the gate itself.
|
||||||
only callers of `stop()`. A third caller has to shut the gate itself.
|
|
||||||
|
|
||||||
### Internals and tests
|
### Internals and tests
|
||||||
|
|
||||||
|
|||||||
@@ -352,8 +352,7 @@ A send issued while the link is down waits for the reconnect instead of failing,
|
|||||||
the new link is bound — up to `responseTimeout`, after which it gives up having sent nothing. A
|
the new link is bound — up to `responseTimeout`, after which it gives up having sent nothing. A
|
||||||
request already on the wire is the other case: the SMSC may have taken it and lost only the response,
|
request already on the wire is the other case: the SMSC may have taken it and lost only the response,
|
||||||
so it fails, and `sendSms()` and `sms.sendDlr()` count it in `unanswered`, whether the link dropped
|
so it fails, and `sendSms()` and `sms.sendDlr()` count it in `unanswered`, whether the link dropped
|
||||||
under it, the peer
|
under it, the peer never answered in time, or you aborted it after it went out. Neither applies with `reconnect: false`,
|
||||||
never answered in time, or you aborted it after it went out. Neither applies with `reconnect: false`,
|
|
||||||
where a drop ends the session and every send after it is refused.
|
where a drop ends the session and every send after it is refused.
|
||||||
|
|
||||||
`responseTimeout` bounds the wait for a link and the wait for an answer separately, and a send also
|
`responseTimeout` bounds the wait for a link and the wait for an answer separately, and a send also
|
||||||
|
|||||||
@@ -49,7 +49,7 @@ export class HeldMessages {
|
|||||||
this.sweep();
|
this.sweep();
|
||||||
|
|
||||||
if (this.held.get(key)) {
|
if (this.held.get(key)) {
|
||||||
this.log.warn('heldMessages - replacing a message on a re-used sequence number', { seqNr: key });
|
this.log.warn('heldMessages - replacing a message on a re-used sequence number', { seqNr: Number(key) });
|
||||||
} else if (this.held.full) {
|
} else if (this.held.full) {
|
||||||
this.dropOldest();
|
this.dropOldest();
|
||||||
}
|
}
|
||||||
@@ -92,7 +92,10 @@ export class HeldMessages {
|
|||||||
|
|
||||||
const [seqNr] = oldest;
|
const [seqNr] = oldest;
|
||||||
|
|
||||||
this.log.warn('heldMessages - buffer full, dropping the oldest message', { max: this.max, seqNr });
|
this.log.warn('heldMessages - buffer full, dropping the oldest message', {
|
||||||
|
max: this.max,
|
||||||
|
seqNr: Number(seqNr),
|
||||||
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Drops every message past its deadline. Runs before each hold and on its own timer. */
|
/** Drops every message past its deadline. Runs before each hold and on its own timer. */
|
||||||
|
|||||||
+1
-1
@@ -9,7 +9,7 @@ import { receiptCodes } from './dlr.ts';
|
|||||||
import { smppDate } from './message.ts';
|
import { smppDate } from './message.ts';
|
||||||
import { uuidv7 } from './uuid.ts';
|
import { uuidv7 } from './uuid.ts';
|
||||||
|
|
||||||
/** Both fields hold what the peer took, so a partial failure names what is already receipted. */
|
/** `pduObjs` holds what the peer took, so a partial failure names what is already receipted. */
|
||||||
export type SendDlrResult = {
|
export type SendDlrResult = {
|
||||||
err?: Error;
|
err?: Error;
|
||||||
pduObjs: PduObject[];
|
pduObjs: PduObject[];
|
||||||
|
|||||||
+48
-22
@@ -73,6 +73,19 @@ function delay(ms: number): Promise<void> {
|
|||||||
return new Promise(resolve => { setTimeout(resolve, ms); });
|
return new Promise(resolve => { setTimeout(resolve, ms); });
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function submitPdu(seqNr: number, cmdStatus: ErrorName = 'ESME_ROK'): PduObject {
|
||||||
|
return {
|
||||||
|
cmdId: 0x00000004,
|
||||||
|
cmdLength: 0,
|
||||||
|
cmdName: 'submit_sm',
|
||||||
|
cmdStatus,
|
||||||
|
cmdStatusId: 0,
|
||||||
|
params: { destination_addr: '46709771337', short_message: 'held', source_addr: '46701113311' },
|
||||||
|
seqNr,
|
||||||
|
tlvs: {},
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
/** A cmd_length below the 16-octet header: a stream no framing can recover from. */
|
/** A cmd_length below the 16-octet header: a stream no framing can recover from. */
|
||||||
const unreadablePdu = Buffer.from([0, 0, 0, 4, 0, 0, 0, 4, 0, 0, 0, 0, 0, 0, 0, 1]);
|
const unreadablePdu = Buffer.from([0, 0, 0, 4, 0, 0, 0, 4, 0, 0, 0, 0, 0, 0, 0, 1]);
|
||||||
|
|
||||||
@@ -862,21 +875,8 @@ describe('LinkGate', () => {
|
|||||||
|
|
||||||
// Goal 4: an application that answers nothing must not grow this for the life of the link.
|
// Goal 4: an application that answers nothing must not grow this for the life of the link.
|
||||||
describe('held message bounds', () => {
|
describe('held message bounds', () => {
|
||||||
function heldPdu(seqNr: number): PduObject {
|
|
||||||
return {
|
|
||||||
cmdId: 0x00000004,
|
|
||||||
cmdLength: 0,
|
|
||||||
cmdName: 'submit_sm',
|
|
||||||
cmdStatus: 'ESME_ROK',
|
|
||||||
cmdStatusId: 0,
|
|
||||||
params: { destination_addr: '46709771337', short_message: 'held', source_addr: '46701113311' },
|
|
||||||
seqNr,
|
|
||||||
tlvs: {},
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
function message(seqNr: number): PduObject[] {
|
function message(seqNr: number): PduObject[] {
|
||||||
return [heldPdu(seqNr)];
|
return [submitPdu(seqNr)];
|
||||||
}
|
}
|
||||||
|
|
||||||
test('drops the message held longest rather than holding every one', () => {
|
test('drops the message held longest rather than holding every one', () => {
|
||||||
@@ -885,6 +885,11 @@ describe('held message bounds', () => {
|
|||||||
|
|
||||||
held.hold(oldest);
|
held.hold(oldest);
|
||||||
held.hold(message(2));
|
held.hold(message(2));
|
||||||
|
held.hold(message(2));
|
||||||
|
|
||||||
|
assert.equal(held.size, 2, 'a re-used sequence number replaces rather than evicting');
|
||||||
|
assert.equal(held.has(oldest), true);
|
||||||
|
|
||||||
held.hold(message(3));
|
held.hold(message(3));
|
||||||
|
|
||||||
assert.equal(held.size, 2);
|
assert.equal(held.size, 2);
|
||||||
@@ -908,9 +913,25 @@ describe('held message bounds', () => {
|
|||||||
held.clear();
|
held.clear();
|
||||||
});
|
});
|
||||||
|
|
||||||
// A receipt cannot be resent wholesale without duplicating the segments that landed, so sendDlr()
|
// Without this the drain sits out its whole budget before returning what a sweep already settled.
|
||||||
// names what the peer took and what it may have, the way sendSms() does.
|
test('wakes a waiting drain when the last message expires', async () => {
|
||||||
test('a partial receipt names the segments the peer took and the ones it may have', async t => {
|
let now = 0;
|
||||||
|
const held = new HeldMessages({ log: silentLog, max: 10, now: () => now, timeout: 60 });
|
||||||
|
|
||||||
|
held.hold(message(1));
|
||||||
|
|
||||||
|
const waiting = held.idle(1000, undefined);
|
||||||
|
|
||||||
|
now = 61;
|
||||||
|
held.sweep();
|
||||||
|
|
||||||
|
assert.equal(await waiting, 0);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
describe('sendDlr()', () => {
|
||||||
|
// A receipt cannot be resent wholesale without duplicating the segments that landed.
|
||||||
|
test('names the segments the peer took, refused, and may have taken', async t => {
|
||||||
const session = new Session({ sock: new net.Socket() });
|
const session = new Session({ sock: new net.Socket() });
|
||||||
|
|
||||||
closeAfter(t, session);
|
closeAfter(t, session);
|
||||||
@@ -919,7 +940,7 @@ describe('held message bounds', () => {
|
|||||||
const sms = createSms({
|
const sms = createSms({
|
||||||
from: '46701113311',
|
from: '46701113311',
|
||||||
message: 'three segments',
|
message: 'three segments',
|
||||||
pduObjs: [heldPdu(1), heldPdu(2), heldPdu(3)],
|
pduObjs: [submitPdu(1), submitPdu(2), submitPdu(3)],
|
||||||
session,
|
session,
|
||||||
to: '46709771337',
|
to: '46709771337',
|
||||||
}, {
|
}, {
|
||||||
@@ -927,15 +948,20 @@ describe('held message bounds', () => {
|
|||||||
send: () => {
|
send: () => {
|
||||||
call++;
|
call++;
|
||||||
|
|
||||||
return Promise.resolve(call === 2
|
if (call === 1) return Promise.resolve({ pduObj: submitPdu(1, 'ESME_RX_T_APPN') });
|
||||||
? { err: new UnansweredError(new Error('nothing came back')) }
|
|
||||||
: { pduObj: heldPdu(call) });
|
if (call === 2) {
|
||||||
|
return Promise.resolve({ err: new UnansweredError(new Error('nothing came back')) });
|
||||||
|
}
|
||||||
|
|
||||||
|
return Promise.resolve({ pduObj: submitPdu(3) });
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
const report = await sms.sendDlr('DELIVERED');
|
const report = await sms.sendDlr('DELIVERED');
|
||||||
|
|
||||||
assert.ok(report.err instanceof Error);
|
assert.ok(report.err instanceof Error);
|
||||||
assert.equal(report.pduObjs.length, 2);
|
assert.match(report.err.message, /deliver_sm refused by the peer/);
|
||||||
|
assert.equal(report.pduObjs.length, 1);
|
||||||
assert.equal(report.unanswered, 1);
|
assert.equal(report.unanswered, 1);
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
Reference in New Issue
Block a user