Skip to content

Fix cancel heartbeat delay timer on abort #527

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
parfeon merged 3 commits into from
Oct 14, 2026
Merged

Fix cancel heartbeat delay timer on abort #527

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c5e2a42
fix(presence): cancel heartbeat delay timer on abort
parfeon Oct 13, 2026
21405e1
refactor(presence): unsubscribe abort listener after heartbeat delay
parfeon Oct 13, 2026
3e7adb8
PubNub SDK v13.1.2 release.
pubnub-release-bot Oct 14, 2026
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
31 changes: 22 additions & 9 deletions dist/web/pubnub.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9203,8 +9203,11 @@
}
})));
this.on(wait.type, asyncHandler((_1, abortSignal_1, _a) => __awaiter(this, [_1, abortSignal_1, _a], void 0, function* (_, abortSignal, { heartbeatDelay }) {
const { cancelDelay, promise } = heartbeatDelay();
abortSignal.throwIfAborted();
yield heartbeatDelay();
const unsubscribeAbortListener = abortSignal.subscribe(() => cancelDelay());
yield promise;
unsubscribeAbortListener();
abortSignal.throwIfAborted();
return engine.transition(timesUp());
})));
Expand Down Expand Up @@ -18607,13 +18610,24 @@
}));
this.makeUnsubscribe(parameters, () => { });
},
heartbeatDelay: () => new Promise((resolve, reject) => {
heartbeatInterval = this._configuration.getHeartbeatInterval();
if (!heartbeatInterval)
reject(new PubNubError('Heartbeat interval has been reset.'));
else
setTimeout(resolve, heartbeatInterval * 1000);
}),
heartbeatDelay: () => {
let cancelDelay = () => { };
const promise = new Promise((resolve, reject) => {
heartbeatInterval = this._configuration.getHeartbeatInterval();
if (!heartbeatInterval)
reject(new PubNubError('Heartbeat interval has been reset.'));
else {
const timeoutId = setTimeout(() => {
resolve();
}, heartbeatInterval * 1000);
cancelDelay = () => {
clearTimeout(timeoutId);
reject(new PubNubError('Heartbeat interval has been cancelled.'));
};
}
});
return { cancelDelay, promise };
},
emitStatus: (status) => this.emitStatus(status),
config: this._configuration,
presenceState: this.presenceState,
Expand All @@ -18639,7 +18653,6 @@
}));
return this.subscribeReceiveMessages(parameters);
},
delay: (amount) => new Promise((resolve) => setTimeout(resolve, amount)),
join: (parameters) => {
var _a, _b;
this.logger.trace('EventEngine', () => ({
Expand Down
2 changes: 1 addition & 1 deletion dist/web/pubnub.min.js
View file
Open in desktop

Large diffs are not rendered by default.

26 changes: 18 additions & 8 deletions lib/core/pubnub-common.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -244,13 +244,24 @@ class PubNubCore {
}));
this.makeUnsubscribe(parameters, () => { });
},
heartbeatDelay: () => new Promise((resolve, reject) => {
heartbeatInterval = this._configuration.getHeartbeatInterval();
if (!heartbeatInterval)
reject(new pubnub_error_1.PubNubError('Heartbeat interval has been reset.'));
else
setTimeout(resolve, heartbeatInterval * 1000);
}),
heartbeatDelay: () => {
let cancelDelay = () => { };
const promise = new Promise((resolve, reject) => {
heartbeatInterval = this._configuration.getHeartbeatInterval();
if (!heartbeatInterval)
reject(new pubnub_error_1.PubNubError('Heartbeat interval has been reset.'));
else {
const timeoutId = setTimeout(() => {
resolve();
}, heartbeatInterval * 1000);
cancelDelay = () => {
clearTimeout(timeoutId);
reject(new pubnub_error_1.PubNubError('Heartbeat interval has been cancelled.'));
};
}
});
return { cancelDelay, promise };
},
emitStatus: (status) => this.emitStatus(status),
config: this._configuration,
presenceState: this.presenceState,
Expand All @@ -276,7 +287,6 @@ class PubNubCore {
}));
return this.subscribeReceiveMessages(parameters);
},
delay: (amount) => new Promise((resolve) => setTimeout(resolve, amount)),
join: (parameters) => {
var _a, _b;
this.logger.trace('EventEngine', () => ({
Expand Down
5 changes: 4 additions & 1 deletion lib/event-engine/presence/dispatcher.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,8 +93,11 @@ class PresenceEventEngineDispatcher extends core_1.Dispatcher {
}
})));
this.on(effects.wait.type, (0, core_1.asyncHandler)((_1, abortSignal_1, _a) => __awaiter(this, [_1, abortSignal_1, _a], void 0, function* (_, abortSignal, { heartbeatDelay }) {
const { cancelDelay, promise } = heartbeatDelay();
abortSignal.throwIfAborted();
yield heartbeatDelay();
const unsubscribeAbortListener = abortSignal.subscribe(() => cancelDelay());
yield promise;
unsubscribeAbortListener();
abortSignal.throwIfAborted();
return engine.transition(events.timesUp());
})));
Expand Down
21 changes: 16 additions & 5 deletions src/core/pubnub-common.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -495,12 +495,24 @@ export class PubNubCore<

this.makeUnsubscribe(parameters, () => {});
},
heartbeatDelay: () =>
new Promise((resolve, reject) => {
heartbeatDelay: () => {
let cancelDelay = () => {};
const promise = new Promise<void>((resolve, reject) => {
heartbeatInterval = this._configuration.getHeartbeatInterval();
if (!heartbeatInterval) reject(new PubNubError('Heartbeat interval has been reset.'));
else setTimeout(resolve, heartbeatInterval * 1000);
}),
else {
const timeoutId = setTimeout(() => {
resolve();
}, heartbeatInterval * 1000);

cancelDelay = () => {
clearTimeout(timeoutId);
reject(new PubNubError('Heartbeat interval has been cancelled.'));
};
}
});
return { cancelDelay, promise };
},
emitStatus: (status) => this.emitStatus(status),
config: this._configuration,
presenceState: this.presenceState,
Expand Down Expand Up @@ -529,7 +541,6 @@ export class PubNubCore<

return this.subscribeReceiveMessages(parameters);
},
delay: (amount) => new Promise((resolve) => setTimeout(resolve, amount)),
join: (parameters) => {
this.logger.trace('EventEngine', () => ({
messageType: 'object',
Expand Down
2 changes: 0 additions & 2 deletions src/event-engine/dispatcher.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,6 @@ export type Dependencies = {
presenceState: Record<string, Payload>;
config: PrivateClientConfiguration;

delay: (milliseconds: number) => Promise<void>;

emitMessages: (
cursor: Subscription.SubscriptionCursor,
events: Subscription.SubscriptionResponse['messages'],
Expand Down
8 changes: 6 additions & 2 deletions src/event-engine/presence/dispatcher.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ export type Dependencies = {
callback?: ResultCallback<Presence.PresenceHeartbeatResponse>,
) => Promise<Presence.PresenceHeartbeatResponse | void>;
leave: (parameters: Presence.PresenceLeaveParameters) => void;
heartbeatDelay: () => Promise<void>;
heartbeatDelay: () => { cancelDelay: () => void; promise: Promise<void> };

config: PrivateClientConfiguration;
presenceState: Record<string, Payload>;
Expand Down Expand Up @@ -85,9 +85,13 @@ export class PresenceEventEngineDispatcher extends Dispatcher<effects.Effects, D
this.on(
effects.wait.type,
asyncHandler(async (_, abortSignal, { heartbeatDelay }) => {
const { cancelDelay, promise } = heartbeatDelay();
abortSignal.throwIfAborted();
await heartbeatDelay();
const unsubscribeAbortListener = abortSignal.subscribe(() => cancelDelay());

await promise;

unsubscribeAbortListener();
abortSignal.throwIfAborted();

return engine.transition(events.timesUp());
Expand Down
96 changes: 96 additions & 0 deletions test/unit/event_engine_presence.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -514,4 +514,100 @@ describe('EventEngine', () => {
expect(presenceEngine?.channels).to.deep.equal([]);
expect(presenceEngine?.groups).to.deep.equal([]);
});

describe('heartbeat delay timer', () => {
const sleep = (ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms));

it('should cancel the pending heartbeat delay timer when destroyed', async () => {
utils
.createNock()
.get('/v2/presence/sub-key/demo/channel/test/heartbeat')
.query(true)
.reply(200, '{"message":"OK", "service":"Presence"}', { 'content-type': 'text/javascript' })
.persist();

// Unique delay identifies the heartbeat delay timer among other timers.
const heartbeatDelay = 7000;
pubnub.setHeartbeatInterval(heartbeatDelay / 1000);

const { setTimeout: scheduleTimeout, clearTimeout: cancelTimeout } = global;
const delayTimers: unknown[] = [];
const cancelledTimers = new Set<unknown>();
global.setTimeout = ((callback: () => void, delay?: number, ...args: unknown[]) => {
const timer = scheduleTimeout(callback, delay, ...args);
if (delay === heartbeatDelay) delayTimers.push(timer);
return timer;
}) as typeof setTimeout;
global.clearTimeout = ((timer: unknown) => {
cancelledTimers.add(timer);
cancelTimeout(timer as NodeJS.Timeout);
}) as typeof clearTimeout;

try {
// @ts-expect-error Intentional access to the private interface.
pubnub.join({ channels: ['test'] });

await forState('HEARTBEAT_COOLDOWN', 1000);
expect(delayTimers).to.have.lengthOf(1);

pubnub.destroy(true);
// Cancellation is dispatched asynchronously (`AbortSignal` notifies listeners on the next tick).
await sleep(50);

expect(cancelledTimers.has(delayTimers[0])).to.equal(true);
} finally {
global.setTimeout = scheduleTimeout;
global.clearTimeout = cancelTimeout;
}
});

it('should not send heartbeats after destroy', async () => {
let heartbeats = 0;
utils
.createNock()
.get('/v2/presence/sub-key/demo/channel/test/heartbeat')
.query(true)
.reply(() => {
heartbeats++;
return [200, '{"message":"OK", "service":"Presence"}', { 'content-type': 'text/javascript' }];
})
.persist();

// @ts-expect-error Intentional access to the private interface.
pubnub.join({ channels: ['test'] });

await forState('HEARTBEAT_COOLDOWN', 1000);
pubnub.destroy(true);
const heartbeatsAtDestroy = heartbeats;

await sleep(1500);

expect(heartbeats).to.equal(heartbeatsAtDestroy);
});

it('should stop the heartbeat loop when heartbeat interval is reset to zero', async () => {
let heartbeats = 0;
utils
.createNock()
.get('/v2/presence/sub-key/demo/channel/test/heartbeat')
.query(true)
.reply(() => {
heartbeats++;
return [200, '{"message":"OK", "service":"Presence"}', { 'content-type': 'text/javascript' }];
})
.persist();

// @ts-expect-error Intentional access to the private interface.
pubnub.join({ channels: ['test'] });

await forState('HEARTBEAT_COOLDOWN', 1000);
pubnub.setHeartbeatInterval(0);
const heartbeatsAtReset = heartbeats;

// At most one heartbeat already scheduled with the previous interval is expected after the reset.
await sleep(2500);

expect(heartbeats - heartbeatsAtReset).to.be.at.most(1);
});
});
});
Loading