Skip to content

Commit bf0b6d9

Browse files
committed
fix(service-messaging): a device registration is one engine transaction, three attempts on a unique-index race
The takeover's delete of the previous holder and the registering device's write run in one engine transaction where the engine offers one, so a refused write rolls the delete back; the docblock states the window on an engine without one. REGISTER_ATTEMPTS is three, with the cross-node 409 stated. Co-authored-by: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ADHkmRD9KKqvTU3THnute6
1 parent 82e68ce commit bf0b6d9

2 files changed

Lines changed: 80 additions & 15 deletions

File tree

‎packages/services/service-messaging/src/device-registry.integration.test.ts‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,8 @@
2222
* - the vendor derives from the platform, and a `web` registration without
2323
* one is refused `VALIDATION_FAILED` at `vendor`, storing nothing;
2424
* - `listDevices` answers only the caller's devices, newest first;
25+
* - a takeover whose write is refused leaves the previous holder's row in
26+
* place: the delete and the write are one engine transaction;
2527
* - the database refuses a second holder of a token (the unique index), and
2628
* two concurrent registrations of one token end with one row.
2729
*
@@ -210,6 +212,36 @@ describe('push device registry: INotificationService device members over sys_use
210212
expect(mine.every((d) => d.userId === alice)).toBe(true);
211213
});
212214

215+
it('a takeover whose write is refused rolls the previous holder\'s delete back: one unit, not two steps', async () => {
216+
const { ql, time, alice, bob, rows } = await boot();
217+
// The real engine, but the registry's insert is refused once armed, as a
218+
// database refusal of the write would be (after the delete has run).
219+
let refuseInsert = false;
220+
const refusing = new Proxy(ql as any, {
221+
get(target, prop, receiver) {
222+
if (prop === 'insert') {
223+
return async (object: string, ...rest: unknown[]) => {
224+
if (refuseInsert && object === USER_DEVICE_OBJECT) throw new Error('write refused by the database');
225+
return (target.insert as (...a: unknown[]) => unknown).call(target, object, ...rest);
226+
};
227+
}
228+
const value = Reflect.get(target, prop, receiver);
229+
return typeof value === 'function' ? value.bind(target) : value;
230+
},
231+
});
232+
const service = new MessagingService({ logger: silent, getData: () => refusing, now: time.now });
233+
await service.registerDevice(alice, { token: 'tok-unit', platform: 'ios', deviceId: 'alice-phone' });
234+
235+
refuseInsert = true;
236+
const refused = await thrown(service.registerDevice(bob, { token: 'tok-unit', platform: 'ios', deviceId: 'bob-phone' }));
237+
expect(String(refused?.message)).toBe('write refused by the database');
238+
239+
const holders = await rows({ token: 'tok-unit' });
240+
expect(holders).toHaveLength(1);
241+
expect(String(holders[0].user_id)).toBe(alice);
242+
expect(await rows({ user_id: bob })).toHaveLength(0);
243+
});
244+
213245
it('the database refuses a second holder of a token, and concurrent registrations end with one row', async () => {
214246
const { ql, service, alice, bob, rows } = await boot();
215247
await service.registerDevice(alice, { token: 'tok-race', platform: 'ios', deviceId: 'a1' });

‎packages/services/service-messaging/src/device-registry.ts‎

Lines changed: 48 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -32,13 +32,24 @@
3232
* 2. Across processes, `sys_user_device.token` carries a unique index, so a
3333
* second holder can never be stored: a racing writer's insert or update is
3434
* refused by the database.
35-
* 3. That refusal is not the caller's error. The registration runs once more
35+
* 3. That refusal is not the caller's error. The registration runs again
3636
* ({@link REGISTER_ATTEMPTS}): the re-read now sees the winner's row and
3737
* moves the token from it, which is what the losing request asked for.
38+
*
39+
* ## One registration is one unit
40+
*
41+
* A registration that moves a token deletes the previous holder's row and then
42+
* writes its own. Where the engine offers a transaction (`IObjectQLEngine.
43+
* transaction`, the ObjectQL engine always does) the reads, the delete and the
44+
* write run in ONE, so a write the database refuses rolls the delete back with
45+
* it and the previous holder keeps its row. On an engine without one the steps
46+
* run in order with no rollback: a refused write after the delete leaves the
47+
* token held by nobody until the device registers again (the next attempt, or
48+
* the app's next start), which loses a push target but never misdirects a push.
3849
*/
3950

4051
import { randomUUID } from 'node:crypto';
41-
import type { IDataEngine } from '@objectstack/spec/contracts';
52+
import type { IDataEngine, IObjectQLEngine } from '@objectstack/spec/contracts';
4253
import type {
4354
RegisterDeviceRequest,
4455
RegisterDeviceResponse,
@@ -58,10 +69,13 @@ export const DEVICE_SYSTEM_CONTEXT = { isSystem: true } as const;
5869

5970
/**
6071
* How many times a registration runs before a unique-index refusal is let
61-
* through. Two: the first attempt, and the one that re-reads after a racing
62-
* writer won.
72+
* through. Three: the first attempt, the one that re-reads after a racing
73+
* writer on another node won, and one more for a second such race in the same
74+
* window. Past that the refusal surfaces (the dispatcher answers it `409
75+
* UNIQUE_VIOLATION`) and the client's next registration settles it; the
76+
* in-process lock means only writers on OTHER nodes can race at all.
6377
*/
64-
const REGISTER_ATTEMPTS = 2;
78+
const REGISTER_ATTEMPTS = 3;
6579

6680
type Vendor = 'apns' | 'fcm';
6781

@@ -73,8 +87,11 @@ const DEFAULT_VENDOR: Readonly<Record<string, Vendor | undefined>> = {
7387

7488
type DeviceRow = Record<string, unknown> & { id?: unknown };
7589

90+
/** The engine surface the registry uses: every engine's reads and writes, plus its transaction where it offers one. */
91+
export type DeviceRegistryEngine = IDataEngine & Partial<Pick<IObjectQLEngine, 'transaction'>>;
92+
7693
export interface DeviceRegistryOptions {
77-
getData: () => IDataEngine | undefined;
94+
getData: () => DeviceRegistryEngine | undefined;
7895
now?: () => string;
7996
}
8097

@@ -101,7 +118,7 @@ export class DeviceRegistry {
101118
this.now = opts.now ?? (() => new Date().toISOString());
102119
}
103120

104-
private engine(): IDataEngine {
121+
private engine(): DeviceRegistryEngine {
105122
const data = this.opts.getData();
106123
if (!data) throw new Error('[messaging] the device registry has no data engine');
107124
return data;
@@ -119,35 +136,47 @@ export class DeviceRegistry {
119136
return this.withTokenLock(request.token, async () => {
120137
for (let attempt = 1; ; attempt += 1) {
121138
try {
122-
return await this.registerOnce(userId, request, vendor);
139+
return await this.inOneUnit((context) => this.registerOnce(userId, request, vendor, context));
123140
} catch (err) {
124141
if (attempt >= REGISTER_ATTEMPTS || !isUniqueViolationError(err)) throw err;
125142
}
126143
}
127144
});
128145
}
129146

147+
/**
148+
* Run `work` in one engine transaction when the engine offers one, with the
149+
* system context the transaction hands back; else directly, with the
150+
* system context (see the module docblock's "One registration is one unit").
151+
*/
152+
private async inOneUnit<T>(work: (context: Record<string, unknown>) => Promise<T>): Promise<T> {
153+
const data = this.engine();
154+
if (typeof data.transaction !== 'function') return work({ ...DEVICE_SYSTEM_CONTEXT });
155+
return data.transaction((trxCtx) => work(trxCtx as Record<string, unknown>), { ...DEVICE_SYSTEM_CONTEXT });
156+
}
157+
130158
private async registerOnce(
131159
userId: string,
132160
request: RegisterDeviceRequest,
133161
vendor: Vendor,
162+
context: Record<string, unknown>,
134163
): Promise<RegisterDeviceResponse> {
135164
const data = this.engine();
136-
const holder = await this.findOne(data, { token: request.token });
165+
const holder = await this.findOne(data, { token: request.token }, context);
137166

138167
// The row this registration refreshes: the named device, else (no
139168
// deviceId) the caller's own row that already holds this token.
140169
let target: DeviceRow | null = null;
141170
if (request.deviceId) {
142-
target = await this.findOne(data, { user_id: userId, device_id: request.deviceId });
171+
target = await this.findOne(data, { user_id: userId, device_id: request.deviceId }, context);
143172
} else if (holder && String(holder.user_id) === userId) {
144173
target = holder;
145174
}
146175

147176
// The token moves: whoever else holds it — another user's device, or
148177
// this user's stale row under another deviceId — gives it up, by row.
149178
if (holder && (!target || String(holder.id) !== String(target.id))) {
150-
await data.delete(USER_DEVICE_OBJECT, { where: { id: holder.id }, context: DEVICE_SYSTEM_CONTEXT } as never);
179+
await data.delete(USER_DEVICE_OBJECT, { where: { id: holder.id }, context } as never);
151180
}
152181

153182
const at = this.now();
@@ -167,7 +196,7 @@ export class DeviceRegistry {
167196
await data.update(
168197
USER_DEVICE_OBJECT,
169198
values,
170-
{ where: { id: target.id }, context: DEVICE_SYSTEM_CONTEXT } as never,
199+
{ where: { id: target.id }, context } as never,
171200
);
172201
return { deviceId: String(target.device_id), success: true };
173202
}
@@ -176,7 +205,7 @@ export class DeviceRegistry {
176205
await data.insert(
177206
USER_DEVICE_OBJECT,
178207
{ id: randomUUID(), user_id: userId, device_id: deviceId, ...values },
179-
{ context: DEVICE_SYSTEM_CONTEXT },
208+
{ context },
180209
);
181210
return { deviceId, success: true };
182211
}
@@ -220,8 +249,12 @@ export class DeviceRegistry {
220249
);
221250
}
222251

223-
private async findOne(data: IDataEngine, where: Record<string, unknown>): Promise<DeviceRow | null> {
224-
const row = await data.findOne(USER_DEVICE_OBJECT, { where }, { context: DEVICE_SYSTEM_CONTEXT });
252+
private async findOne(
253+
data: IDataEngine,
254+
where: Record<string, unknown>,
255+
context: Record<string, unknown> = DEVICE_SYSTEM_CONTEXT,
256+
): Promise<DeviceRow | null> {
257+
const row = await data.findOne(USER_DEVICE_OBJECT, { where }, { context });
225258
return (row as DeviceRow | null) ?? null;
226259
}
227260

0 commit comments

Comments
 (0)