fix(kobo): stop endless sync retries and cut round trips

Report success when Kobo rejects a delete for a tag it no longer has, so the device stops retrying
one it can never complete. Skip the full-library reconcile while a device is still draining pages,
and page fifty books at a time so a large re-announce costs a few round trips instead of hundreds.

Ref #1005
This commit is contained in:
neon
2026-08-14 02:26:46 +00:00
parent fddb41fdb9
commit 2b8e1d6a79
6 changed files with 221 additions and 44 deletions
@@ -1,4 +1,4 @@
import { HttpStatus } from '@nestjs/common'; import { BadRequestException, HttpStatus } from '@nestjs/common';
import { KoboSyncController } from './kobo-sync.controller'; import { KoboSyncController } from './kobo-sync.controller';
@@ -26,6 +26,7 @@ describe('KoboSyncController', () => {
const proxyService = { const proxyService = {
forward: vi.fn(), forward: vi.fn(),
request: vi.fn(), request: vi.fn(),
sendUpstream: vi.fn(),
}; };
const bookIdentityService = { const bookIdentityService = {
resolveBookIdByEntitlementId: vi.fn(), resolveBookIdByEntitlementId: vi.fn(),
@@ -427,16 +428,69 @@ describe('KoboSyncController', () => {
expect(proxyService.forward).not.toHaveBeenCalled(); expect(proxyService.forward).not.toHaveBeenCalled();
}); });
it('proxies delete-items requests for non-BookOrbit Kobo tags', async () => { it('relays the upstream answer for delete-items requests on non-BookOrbit Kobo tags', async () => {
const req = { method: 'POST', url: '/api/v1/kobo/token/v1/library/tags/kobo-tag/items/delete' }; const req = { method: 'POST', url: '/api/v1/kobo/token/v1/library/tags/kobo-tag/items/delete' };
const reply = makeReply(); const reply = makeReply();
const upstream = { status: HttpStatus.OK, headers: {}, body: Buffer.from('[]') };
proxyService.request.mockResolvedValue(upstream);
await controller.deleteTagItems('kobo-tag', { deviceToken: 'token-1' } as never, req as never, reply as never); await controller.deleteTagItems('kobo-tag', { deviceToken: 'token-1', deviceId: 7 } as never, req as never, reply as never);
expect(proxyService.forward).toHaveBeenCalledWith(req, reply, 'token-1'); expect(proxyService.request).toHaveBeenCalledWith(req, 'token-1');
expect(proxyService.sendUpstream).toHaveBeenCalledWith(reply, upstream);
expect(reply.send).not.toHaveBeenCalled(); expect(reply.send).not.toHaveBeenCalled();
}); });
it.each([
['delete-items', HttpStatus.BAD_REQUEST],
['delete-items', HttpStatus.NOT_FOUND],
['delete-items', HttpStatus.GONE],
])('reports %s success when Kobo answers %s for a tag it does not know', async (_label, status) => {
const req = { method: 'POST', url: '/api/v1/kobo/token/v1/library/tags/BL-MS-2/items/delete' };
const reply = makeReply();
proxyService.request.mockResolvedValue({ status, headers: {}, body: Buffer.from('') });
await controller.deleteTagItems('BL-MS-2', { deviceToken: 'token-1', deviceId: 7 } as never, req as never, reply as never);
expect(reply.status).toHaveBeenCalledWith(HttpStatus.OK);
expect(reply.send).toHaveBeenCalledWith({ RequestResult: 'Success' });
expect(proxyService.sendUpstream).not.toHaveBeenCalled();
});
it('relays upstream failures that do not mean the tag is already gone', async () => {
const req = { method: 'POST', url: '/api/v1/kobo/token/v1/library/tags/BL-MS-2/items/delete' };
const reply = makeReply();
const upstream = { status: HttpStatus.UNAUTHORIZED, headers: {}, body: Buffer.from('') };
proxyService.request.mockResolvedValue(upstream);
await controller.deleteTagItems('BL-MS-2', { deviceToken: 'token-1', deviceId: 7 } as never, req as never, reply as never);
expect(proxyService.sendUpstream).toHaveBeenCalledWith(reply, upstream);
expect(reply.send).not.toHaveBeenCalledWith({ RequestResult: 'Success' });
});
it('lets a rejected proxy path stay a client error instead of an upstream fault', async () => {
const req = { method: 'DELETE', url: '/api/v1/kobo/token/v1/library/tags/kobo-tag' };
const reply = makeReply();
proxyService.request.mockRejectedValue(new BadRequestException('Invalid proxy path'));
await expect(controller.deleteTag('kobo-tag', { deviceToken: 'token-1', deviceId: 7 } as never, req as never, reply as never)).rejects.toThrow(
BadRequestException,
);
expect(reply.send).not.toHaveBeenCalled();
});
it('answers 502 when the upstream tag delete cannot be reached', async () => {
const req = { method: 'DELETE', url: '/api/v1/kobo/token/v1/library/tags/kobo-tag' };
const reply = makeReply();
proxyService.request.mockRejectedValue(new Error('socket hang up'));
await controller.deleteTag('kobo-tag', { deviceToken: 'token-1', deviceId: 7 } as never, req as never, reply as never);
expect(reply.status).toHaveBeenCalledWith(HttpStatus.BAD_GATEWAY);
expect(reply.send).not.toHaveBeenCalledWith({ RequestResult: 'Success' });
});
it('acknowledges add-items requests for synced BookOrbit collection tags', async () => { it('acknowledges add-items requests for synced BookOrbit collection tags', async () => {
const req = { method: 'POST', url: '/api/v1/kobo/token/v1/library/tags/col-1/items' }; const req = { method: 'POST', url: '/api/v1/kobo/token/v1/library/tags/col-1/items' };
const reply = makeReply(); const reply = makeReply();
@@ -469,11 +523,23 @@ describe('KoboSyncController', () => {
expect(proxyService.forward).not.toHaveBeenCalled(); expect(proxyService.forward).not.toHaveBeenCalled();
}); });
it('proxies delete-tag requests for non-BookOrbit Kobo tags', async () => { it('reports success when Kobo no longer has the tag the device is deleting', async () => {
const req = { method: 'DELETE', url: '/api/v1/kobo/token/v1/library/tags/kobo-tag' }; const req = { method: 'DELETE', url: '/api/v1/kobo/token/v1/library/tags/a413bef9' };
const reply = makeReply();
proxyService.request.mockResolvedValue({ status: HttpStatus.NOT_FOUND, headers: {}, body: Buffer.from('') });
await controller.deleteTag('a413bef9', { deviceToken: 'token-1', deviceId: 7 } as never, req as never, reply as never);
expect(reply.status).toHaveBeenCalledWith(HttpStatus.OK);
expect(reply.send).toHaveBeenCalledWith({ RequestResult: 'Success' });
expect(proxyService.sendUpstream).not.toHaveBeenCalled();
});
it('never turns a rejected add-items request into a success', async () => {
const req = { method: 'POST', url: '/api/v1/kobo/token/v1/library/tags/BL-MS-2/items' };
const reply = makeReply(); const reply = makeReply();
await controller.deleteTag('kobo-tag', { deviceToken: 'token-1' } as never, req as never, reply as never); await controller.addTagItems('BL-MS-2', { deviceToken: 'token-1', deviceId: 7 } as never, req as never, reply as never);
expect(proxyService.forward).toHaveBeenCalledWith(req, reply, 'token-1'); expect(proxyService.forward).toHaveBeenCalledWith(req, reply, 'token-1');
expect(reply.send).not.toHaveBeenCalled(); expect(reply.send).not.toHaveBeenCalled();
@@ -5,6 +5,7 @@ import {
Get, Get,
Header, Header,
HttpCode, HttpCode,
HttpException,
HttpStatus, HttpStatus,
Logger, Logger,
Param, Param,
@@ -28,6 +29,7 @@ import { KoboTokenGuard } from './guards/kobo-token.guard';
import { KoboSettingsService } from './services/kobo-settings.service'; import { KoboSettingsService } from './services/kobo-settings.service';
import { KoboSyncService } from './services/kobo-sync.service'; import { KoboSyncService } from './services/kobo-sync.service';
import { KoboReadingStateService } from './services/kobo-reading-state.service'; import { KoboReadingStateService } from './services/kobo-reading-state.service';
import type { KoboProxyResponse } from './services/kobo-proxy.service';
import { KoboProxyService } from './services/kobo-proxy.service'; import { KoboProxyService } from './services/kobo-proxy.service';
import { KOBO_STORE_RESOURCES } from './kobo-store-resources'; import { KOBO_STORE_RESOURCES } from './kobo-store-resources';
import { KoboBookIdentityService } from './services/kobo-book-identity.service'; import { KoboBookIdentityService } from './services/kobo-book-identity.service';
@@ -37,6 +39,12 @@ import { decodeSyncToken, isUsableKoboSyncToken, withKoboSyncToken } from './ser
const STORE_SYNC_TIMEOUT_MS = 8_000; const STORE_SYNC_TIMEOUT_MS = 8_000;
const KOBO_SYNC_TOKEN_HEADER = 'x-kobo-synctoken'; const KOBO_SYNC_TOKEN_HEADER = 'x-kobo-synctoken';
// Kobo answers a delete for a tag it does not know with a client error, and the device keeps the
// operation queued until it succeeds, so relaying that error strands a tag that no longer exists
// anywhere in a delete it can never complete. Server faults and auth failures still reach the
// device, because those mean "try again later" rather than "already gone".
const TAG_ALREADY_GONE_STATUSES: number[] = [HttpStatus.BAD_REQUEST, HttpStatus.NOT_FOUND, HttpStatus.GONE];
type StoreSyncResult = { type StoreSyncResult = {
entitlements: unknown[]; entitlements: unknown[];
hasMore: boolean; hasMore: boolean;
@@ -278,7 +286,7 @@ export class KoboSyncController {
@Req() req: FastifyRequest, @Req() req: FastifyRequest,
@Res() reply: FastifyReply, @Res() reply: FastifyReply,
) { ) {
if (!isBookOrbitTag(tagId)) return this.proxyService.forward(req, reply, device.deviceToken); if (!isBookOrbitTag(tagId)) return this.forwardTagDelete(req, reply, device, tagId);
reply.status(HttpStatus.OK).send({ RequestResult: 'Success' }); reply.status(HttpStatus.OK).send({ RequestResult: 'Success' });
} }
@@ -292,10 +300,43 @@ export class KoboSyncController {
@Delete('v1/library/tags/:tagId') @Delete('v1/library/tags/:tagId')
@HttpCode(HttpStatus.OK) @HttpCode(HttpStatus.OK)
async deleteTag(@Param('tagId') tagId: string, @KoboDevice() device: KoboDeviceContext, @Req() req: FastifyRequest, @Res() reply: FastifyReply) { async deleteTag(@Param('tagId') tagId: string, @KoboDevice() device: KoboDeviceContext, @Req() req: FastifyRequest, @Res() reply: FastifyReply) {
if (!isBookOrbitTag(tagId)) return this.proxyService.forward(req, reply, device.deviceToken); if (!isBookOrbitTag(tagId)) return this.forwardTagDelete(req, reply, device, tagId);
reply.status(HttpStatus.OK).send({ RequestResult: 'Success' }); reply.status(HttpStatus.OK).send({ RequestResult: 'Success' });
} }
/**
* Forwards a delete for a tag BookOrbit does not own, reporting success when Kobo says the tag is
* not there. Only deletes get this treatment: a failed add must stay a failure, because reporting
* success would drop the device's pending change instead of retrying it.
*/
private async forwardTagDelete(req: FastifyRequest, reply: FastifyReply, device: KoboDeviceContext, tagId: string): Promise<void> {
let response: KoboProxyResponse;
try {
response = await this.proxyService.request(req, device.deviceToken);
} catch (error: unknown) {
// A path that does not resolve to Kobo is rejected before the request leaves, and that stays
// the client error the shared proxy raises rather than becoming an upstream fault.
if (error instanceof HttpException) throw error;
const errorClass = error instanceof Error ? error.name : 'UnknownError';
const errorMessage = sanitizeLogValue(error instanceof Error ? error.message : 'unknown error');
this.logger.warn(
`[kobo.tag_delete] [fail] deviceId=${device.deviceId} errorClass=${errorClass} error="${errorMessage}" - upstream tag delete failed`,
);
reply.status(HttpStatus.BAD_GATEWAY).send({ message: 'Upstream Kobo API unavailable' });
return;
}
if (TAG_ALREADY_GONE_STATUSES.includes(response.status)) {
this.logger.warn(
`[kobo.tag_delete] [end] deviceId=${device.deviceId} tagId="${sanitizeLogValue(tagId)}" status=${response.status} - upstream does not know this tag, reporting success so the device stops retrying`,
);
reply.status(HttpStatus.OK).send({ RequestResult: 'Success' });
return;
}
this.proxyService.sendUpstream(reply, response);
}
@Get('v1/library/:bookId/metadata') @Get('v1/library/:bookId/metadata')
async getBookMetadata( async getBookMetadata(
@Param('bookId') bookId: string, @Param('bookId') bookId: string,
@@ -76,19 +76,22 @@ export class KoboProxyService {
return { status: upstream.status, headers: responseHeaders, body: Buffer.from(await upstream.arrayBuffer()) }; return { status: upstream.status, headers: responseHeaders, body: Buffer.from(await upstream.arrayBuffer()) };
} }
/** Relays an upstream response to the device, dropping the headers that must not be forwarded. */
sendUpstream(reply: FastifyReply, response: KoboProxyResponse): void {
reply.status(response.status);
for (const [key, value] of Object.entries(response.headers)) {
if (!HOP_BY_HOP_HEADERS.includes(key)) {
reply.header(key, value);
}
}
reply.send(response.body);
}
async forward(req: FastifyRequest, reply: FastifyReply, deviceToken: string) { async forward(req: FastifyRequest, reply: FastifyReply, deviceToken: string) {
const targetUrl = this.resolveTargetUrl(req, deviceToken); const targetUrl = this.resolveTargetUrl(req, deviceToken);
try { try {
const { status, headers, body } = await this.request(req, deviceToken); this.sendUpstream(reply, await this.request(req, deviceToken));
reply.status(status);
for (const [key, value] of Object.entries(headers)) {
if (!HOP_BY_HOP_HEADERS.includes(key)) {
reply.header(key, value);
}
}
reply.send(body);
} catch (err) { } catch (err) {
this.logger.warn(`Proxy failed for ${targetUrl}: ${(err as Error).message}`); this.logger.warn(`Proxy failed for ${targetUrl}: ${(err as Error).message}`);
reply.status(502).send({ message: 'Upstream Kobo API unavailable' }); reply.status(502).send({ message: 'Upstream Kobo API unavailable' });
@@ -238,8 +238,8 @@ describe('KoboSyncService', () => {
expect(createSpy).toHaveBeenNthCalledWith(2, 5, 202, eligible, new Set()); expect(createSpy).toHaveBeenNthCalledWith(2, 5, 202, eligible, new Set());
expect((service as any).fetchEligibleSnapshotRows).toHaveBeenNthCalledWith(1, 5, false, expect.any(Map)); expect((service as any).fetchEligibleSnapshotRows).toHaveBeenNthCalledWith(1, 5, false, expect.any(Map));
expect((service as any).fetchEligibleSnapshotRows).toHaveBeenNthCalledWith(2, 5, true, expect.any(Map)); expect((service as any).fetchEligibleSnapshotRows).toHaveBeenNthCalledWith(2, 5, true, expect.any(Map));
expect(pageSpy).toHaveBeenNthCalledWith(1, 5, 11, 'device-a-token', 'https://reader.example.com', new Set([1]), expect.any(Map)); expect(pageSpy).toHaveBeenNthCalledWith(1, 5, 11, 'device-a-token', 'https://reader.example.com', expect.any(Map), expect.any(Function));
expect(pageSpy).toHaveBeenNthCalledWith(2, 5, 22, 'device-b-token', 'https://reader.example.com', new Set([1]), expect.any(Map)); expect(pageSpy).toHaveBeenNthCalledWith(2, 5, 22, 'device-b-token', 'https://reader.example.com', expect.any(Map), expect.any(Function));
}); });
it('reconciles only the caller device snapshot when it already exists', async () => { it('reconciles only the caller device snapshot when it already exists', async () => {
@@ -254,7 +254,38 @@ describe('KoboSyncService', () => {
await service.getDelta(5, 101, 'device-a-token', 'https://reader.example.com'); await service.getDelta(5, 101, 'device-a-token', 'https://reader.example.com');
expect(reconcileSpy).toHaveBeenCalledWith(9, eligible); expect(reconcileSpy).toHaveBeenCalledWith(9, eligible);
expect(pageSpy).toHaveBeenCalledWith(5, 9, 'device-a-token', 'https://reader.example.com', new Set([1]), expect.any(Map)); expect(pageSpy).toHaveBeenCalledWith(5, 9, 'device-a-token', 'https://reader.example.com', expect.any(Map), expect.any(Function));
await expect(pageSpy.mock.calls[0]![5]()).resolves.toEqual(new Set([1]));
});
it('serves the next page without reconciling while the device still has pending rows', async () => {
const db = makeDb({ select: [[{ bookId: 4 }]] });
const service = makeService(db);
vi.spyOn(service as any, 'findDeviceSnapshot').mockResolvedValue({ id: 9, userId: 5, deviceId: 101 });
const eligibleSpy = vi.spyOn(service as any, 'fetchEligibleSnapshotRows').mockResolvedValue([]);
const reconcileSpy = vi.spyOn(service as any, 'reconcileSnapshot').mockResolvedValue(undefined);
const pageSpy = vi.spyOn(service as any, 'getPageFromSnapshot').mockResolvedValue({ entitlements: [], hasMore: true, syncToken: 'PX.token' });
await service.getDelta(5, 101, 'device-a-token', 'https://reader.example.com');
expect(reconcileSpy).not.toHaveBeenCalled();
expect(eligibleSpy).not.toHaveBeenCalled();
expect(pageSpy).toHaveBeenCalledWith(5, 9, 'device-a-token', 'https://reader.example.com', expect.any(Map), expect.any(Function));
});
it('still resolves eligible ids for tag delivery when a pending page drains before it is served', async () => {
const db = makeDb({ select: [[{ bookId: 4 }]] });
const service = makeService(db);
vi.spyOn(service as any, 'findDeviceSnapshot').mockResolvedValue({ id: 9, userId: 5, deviceId: 101 });
const eligibleSpy = vi
.spyOn(service as any, 'fetchEligibleSnapshotRows')
.mockResolvedValue([{ bookId: 7, fileHash: 'h', deliveryHash: 'd', metadataHash: 'm', needsLegacyNumericRemoval: false }]);
const pageSpy = vi.spyOn(service as any, 'getPageFromSnapshot').mockResolvedValue({ entitlements: [], hasMore: false, syncToken: 'PX.token' });
await service.getDelta(5, 101, 'device-a-token', 'https://reader.example.com');
await expect(pageSpy.mock.calls[0]![5]()).resolves.toEqual(new Set([7]));
expect(eligibleSpy).toHaveBeenCalledWith(5, true, expect.any(Map));
}); });
it('getBookMetadata returns empty array when book is not eligible', async () => { it('getBookMetadata returns empty array when book is not eligible', async () => {
@@ -464,7 +495,7 @@ describe('KoboSyncService', () => {
const service = makeService(db); const service = makeService(db);
vi.spyOn(service as any, 'buildTagItems').mockResolvedValue([{ ChangedTag: {} }]); vi.spyOn(service as any, 'buildTagItems').mockResolvedValue([{ ChangedTag: {} }]);
const result = await (service as any).getPageFromSnapshot(7, 1, 'tok', 'https://base', new Set([1]), new Map()); const result = await (service as any).getPageFromSnapshot(7, 1, 'tok', 'https://base', new Map(), () => Promise.resolve(new Set([1])));
expect(result).toEqual({ expect(result).toEqual({
entitlements: [{ ChangedTag: {} }], entitlements: [{ ChangedTag: {} }],
@@ -491,7 +522,7 @@ describe('KoboSyncService', () => {
); );
readingStateService.getRawState.mockResolvedValue(null); readingStateService.getRawState.mockResolvedValue(null);
const result = await (service as any).getPageFromSnapshot(7, 22, 'tok', 'https://base', new Set([1, 2, 3]), new Map()); const result = await (service as any).getPageFromSnapshot(7, 22, 'tok', 'https://base', new Map(), () => Promise.resolve(new Set([1, 2, 3])));
expect(result.hasMore).toBe(false); expect(result.hasMore).toBe(false);
expect(result.entitlements).toHaveLength(3); expect(result.entitlements).toHaveLength(3);
@@ -510,7 +541,7 @@ describe('KoboSyncService', () => {
vi.spyOn(service as any, 'fetchEligibleBooksByIds').mockResolvedValue(new Map([[3, makeBook(3)]])); vi.spyOn(service as any, 'fetchEligibleBooksByIds').mockResolvedValue(new Map([[3, makeBook(3)]]));
readingStateService.getRawState.mockResolvedValue({ EntitlementId: 'entitlement-3', CurrentBookmark: { ProgressPercent: 61 } }); readingStateService.getRawState.mockResolvedValue({ EntitlementId: 'entitlement-3', CurrentBookmark: { ProgressPercent: 61 } });
const result = await (service as any).getPageFromSnapshot(7, 22, 'tok', 'https://base', new Set([3]), new Map()); const result = await (service as any).getPageFromSnapshot(7, 22, 'tok', 'https://base', new Map(), () => Promise.resolve(new Set([3])));
expect(result.entitlements).toHaveLength(2); expect(result.entitlements).toHaveLength(2);
expect(result.entitlements[0]).toHaveProperty('ChangedProductMetadata'); expect(result.entitlements[0]).toHaveProperty('ChangedProductMetadata');
@@ -530,7 +561,7 @@ describe('KoboSyncService', () => {
bookIdentityService.findByBookIds.mockResolvedValue(new Map([[3, makeIdentity(3, false)]])); bookIdentityService.findByBookIds.mockResolvedValue(new Map([[3, makeIdentity(3, false)]]));
readingStateService.getRawState.mockResolvedValue({ EntitlementId: 'entitlement-3', CurrentBookmark: { ProgressPercent: 61 } }); readingStateService.getRawState.mockResolvedValue({ EntitlementId: 'entitlement-3', CurrentBookmark: { ProgressPercent: 61 } });
const result = await (service as any).getPageFromSnapshot(7, 22, 'tok', 'https://base', new Set([3]), new Map()); const result = await (service as any).getPageFromSnapshot(7, 22, 'tok', 'https://base', new Map(), () => Promise.resolve(new Set([3])));
expect(result.entitlements).toHaveLength(2); expect(result.entitlements).toHaveLength(2);
expect(result.entitlements[0]).toHaveProperty('ChangedEntitlement'); expect(result.entitlements[0]).toHaveProperty('ChangedEntitlement');
@@ -550,7 +581,7 @@ describe('KoboSyncService', () => {
vi.spyOn(service as any, 'fetchEligibleBooksByIds').mockResolvedValue(new Map()); vi.spyOn(service as any, 'fetchEligibleBooksByIds').mockResolvedValue(new Map());
bookIdentityService.findByBookIds.mockResolvedValue(new Map([[3, makeIdentity(3, true)]])); bookIdentityService.findByBookIds.mockResolvedValue(new Map([[3, makeIdentity(3, true)]]));
const result = await (service as any).getPageFromSnapshot(7, 22, 'tok', 'https://base', new Set(), new Map()); const result = await (service as any).getPageFromSnapshot(7, 22, 'tok', 'https://base', new Map(), () => Promise.resolve(new Set()));
const removedIds = result.entitlements.map( const removedIds = result.entitlements.map(
(entry: Record<string, { BookEntitlement: { Id: string } }>) => entry.ChangedEntitlement.BookEntitlement.Id, (entry: Record<string, { BookEntitlement: { Id: string } }>) => entry.ChangedEntitlement.BookEntitlement.Id,
); );
@@ -18,7 +18,10 @@ import { encodeSyncToken } from './kobo-sync-token';
type Db = NodePgDatabase<typeof schema>; type Db = NodePgDatabase<typeof schema>;
const SYNC_PAGE_SIZE = 5; // A device pauses between sync rounds, so wall-clock time for a long delta tracks the number of
// round trips rather than the work inside each one: at five books a page, re-announcing a few
// hundred books cost users tens of minutes of waiting on a mostly idle server.
export const SYNC_PAGE_SIZE = 50;
const SNAPSHOT_RECONCILE_BATCH_SIZE = 5000; const SNAPSHOT_RECONCILE_BATCH_SIZE = 5000;
const SNAPSHOT_CREATE_BATCH_SIZE = 5000; const SNAPSHOT_CREATE_BATCH_SIZE = 5000;
const METADATA_SERIALIZER_VERSION = 2; const METADATA_SERIALIZER_VERSION = 2;
@@ -67,6 +70,8 @@ type SmartScopeMatch = { name: string; bookIds: number[]; where: SQL | undefined
// caching across requests (this service is a singleton). // caching across requests (this service is a singleton).
type SmartScopeMatchCache = Map<number, Promise<Map<number, SmartScopeMatch>>>; type SmartScopeMatchCache = Map<number, Promise<Map<number, SmartScopeMatch>>>;
type EligibleIdsResolver = () => Promise<Set<number>>;
export interface KoboBookEntry { export interface KoboBookEntry {
bookId: number; bookId: number;
koboEntitlementId: string; koboEntitlementId: string;
@@ -111,9 +116,19 @@ export class KoboSyncService {
baseUrl: string, baseUrl: string,
): Promise<{ entitlements: unknown[]; hasMore: boolean; syncToken: string }> { ): Promise<{ entitlements: unknown[]; hasMore: boolean; syncToken: string }> {
let snapshot = await this.findDeviceSnapshot(userId, deviceId); let snapshot = await this.findDeviceSnapshot(userId, deviceId);
const smartScopeMatchCache: SmartScopeMatchCache = new Map();
// Reconciling costs a pass over the whole library, so a device still working through its
// pending rows skips it: those rows already say what it is owed, and whatever changed while
// it paged is picked up by the reconcile that opens its next sync.
if (snapshot && (await this.hasPendingSnapshotBooks(snapshot.id))) {
return this.getPageFromSnapshot(userId, snapshot.id, deviceToken, baseUrl, smartScopeMatchCache, () =>
this.fetchEligibleBookIds(userId, true, smartScopeMatchCache),
);
}
const hadDeviceSnapshot = snapshot ? true : await this.hasDeviceSnapshot(userId); const hadDeviceSnapshot = snapshot ? true : await this.hasDeviceSnapshot(userId);
const legacyNumericRemovalBookIds = snapshot ? new Set<number>() : await this.getLegacyNumericRemovalBookIds(userId, deviceId); const legacyNumericRemovalBookIds = snapshot ? new Set<number>() : await this.getLegacyNumericRemovalBookIds(userId, deviceId);
const smartScopeMatchCache: SmartScopeMatchCache = new Map();
const eligibleSnapshotRows = await this.fetchEligibleSnapshotRows(userId, hadDeviceSnapshot, smartScopeMatchCache); const eligibleSnapshotRows = await this.fetchEligibleSnapshotRows(userId, hadDeviceSnapshot, smartScopeMatchCache);
if (!snapshot) { if (!snapshot) {
@@ -128,14 +143,8 @@ export class KoboSyncService {
await this.reconcileSnapshot(snapshot.id, eligibleSnapshotRows); await this.reconcileSnapshot(snapshot.id, eligibleSnapshotRows);
} }
return this.getPageFromSnapshot( const eligibleIds = new Set(eligibleSnapshotRows.map((row) => row.bookId));
userId, return this.getPageFromSnapshot(userId, snapshot.id, deviceToken, baseUrl, smartScopeMatchCache, () => Promise.resolve(eligibleIds));
snapshot.id,
deviceToken,
baseUrl,
new Set(eligibleSnapshotRows.map((row) => row.bookId)),
smartScopeMatchCache,
);
} }
async getBookMetadata(userId: number, bookId: number, deviceToken: string, baseUrl: string): Promise<unknown[]> { async getBookMetadata(userId: number, bookId: number, deviceToken: string, baseUrl: string): Promise<unknown[]> {
@@ -172,6 +181,24 @@ export class KoboSyncService {
}); });
} }
private async hasPendingSnapshotBooks(snapshotId: number): Promise<boolean> {
const [row] = await this.db
.select({ bookId: schema.koboSnapshotBooks.bookId })
.from(schema.koboSnapshotBooks)
.where(and(eq(schema.koboSnapshotBooks.snapshotId, snapshotId), eq(schema.koboSnapshotBooks.synced, false)))
.limit(1);
return Boolean(row);
}
private async fetchEligibleBookIds(
userId: number,
needsLegacyNumericRemovalForNewMappings: boolean,
smartScopeMatchCache: SmartScopeMatchCache,
): Promise<Set<number>> {
const rows = await this.fetchEligibleSnapshotRows(userId, needsLegacyNumericRemovalForNewMappings, smartScopeMatchCache);
return new Set(rows.map((row) => row.bookId));
}
private async hasDeviceSnapshot(userId: number): Promise<boolean> { private async hasDeviceSnapshot(userId: number): Promise<boolean> {
const snapshot = await this.db.query.koboLibrarySnapshots.findFirst({ const snapshot = await this.db.query.koboLibrarySnapshots.findFirst({
where: eq(schema.koboLibrarySnapshots.userId, userId), where: eq(schema.koboLibrarySnapshots.userId, userId),
@@ -475,13 +502,18 @@ export class KoboSyncService {
}); });
} }
/**
* Serves the next page of a snapshot. Eligibility is resolved lazily because only the final page
* needs the full set, and reaching for it on every page is what makes a long delta O(library) per
* request.
*/
private async getPageFromSnapshot( private async getPageFromSnapshot(
userId: number, userId: number,
snapshotId: number, snapshotId: number,
deviceToken: string, deviceToken: string,
baseUrl: string, baseUrl: string,
eligibleIds: Set<number>,
smartScopeMatchCache: SmartScopeMatchCache, smartScopeMatchCache: SmartScopeMatchCache,
resolveEligibleIds: EligibleIdsResolver,
): Promise<{ entitlements: unknown[]; hasMore: boolean; syncToken: string }> { ): Promise<{ entitlements: unknown[]; hasMore: boolean; syncToken: string }> {
const syncToken = encodeSyncToken(snapshotId); const syncToken = encodeSyncToken(snapshotId);
@@ -496,7 +528,7 @@ export class KoboSyncService {
const page = pending.slice(0, SYNC_PAGE_SIZE); const page = pending.slice(0, SYNC_PAGE_SIZE);
if (page.length === 0) { if (page.length === 0) {
const tagItems = await this.buildTagItems(userId, eligibleIds, smartScopeMatchCache); const tagItems = await this.buildTagItems(userId, await resolveEligibleIds(), smartScopeMatchCache);
return { entitlements: tagItems, hasMore: false, syncToken }; return { entitlements: tagItems, hasMore: false, syncToken };
} }
+10 -6
View File
@@ -6,6 +6,7 @@ import { and, eq, inArray } from 'drizzle-orm';
import { Client } from 'pg'; import { Client } from 'pg';
import * as schema from '../src/db/schema'; import * as schema from '../src/db/schema';
import { SYNC_PAGE_SIZE } from '../src/modules/kobo/services/kobo-sync.service';
import { waitForCondition } from './e2e/app-harness'; import { waitForCondition } from './e2e/app-harness';
import { createEpubFixture } from './e2e/reader-state-isolation/reader-state-isolation-fixture-builder'; import { createEpubFixture } from './e2e/reader-state-isolation/reader-state-isolation-fixture-builder';
import { import {
@@ -25,6 +26,9 @@ type NewEntitlement = { BookEntitlement: { Id: string }; ReadingState: Record<st
const MIGRATIONS_DIRECTORY = join(dirname(fileURLToPath(import.meta.url)), '../src/db/migrations'); const MIGRATIONS_DIRECTORY = join(dirname(fileURLToPath(import.meta.url)), '../src/db/migrations');
// One book past a full page, so the suite keeps exercising a page boundary whatever SYNC_PAGE_SIZE is.
const LIBRARY_BOOK_COUNT = SYNC_PAGE_SIZE + 1;
function escapeIdentifier(identifier: string): string { function escapeIdentifier(identifier: string): string {
return `"${identifier.replace(/"/g, '""')}"`; return `"${identifier.replace(/"/g, '""')}"`;
} }
@@ -240,7 +244,7 @@ describe('Kobo multi-device library sync (e2e)', { timeout: 180_000 }, () => {
library = await createLibraryWithFolder(ctx, { name: `kobo-multi-device-${randomUUID()}` }); library = await createLibraryWithFolder(ctx, { name: `kobo-multi-device-${randomUUID()}` });
const paths = await Promise.all( const paths = await Promise.all(
Array.from({ length: 6 }, (_, index) => Array.from({ length: LIBRARY_BOOK_COUNT }, (_, index) =>
createEpubFixture(library.folderPath, `kobo-multi-device-${index + 1}.epub`, { createEpubFixture(library.folderPath, `kobo-multi-device-${index + 1}.epub`, {
title: `Kobo Multi Device ${index + 1}`, title: `Kobo Multi Device ${index + 1}`,
uid: `urn:uuid:${randomUUID()}`, uid: `urn:uuid:${randomUUID()}`,
@@ -291,7 +295,7 @@ describe('Kobo multi-device library sync (e2e)', { timeout: 180_000 }, () => {
const first = await sync(echoDevice); const first = await sync(echoDevice);
const firstEntitlements = newEntitlements(first.entries); const firstEntitlements = newEntitlements(first.entries);
expect(first.hasMore).toBe(true); expect(first.hasMore).toBe(true);
expect(firstEntitlements).toHaveLength(5); expect(firstEntitlements).toHaveLength(SYNC_PAGE_SIZE);
for (const entitlement of firstEntitlements) { for (const entitlement of firstEntitlements) {
await putReadingState(echoDevice, entitlement.BookEntitlement.Id, entitlement.ReadingState); await putReadingState(echoDevice, entitlement.BookEntitlement.Id, entitlement.ReadingState);
@@ -323,8 +327,8 @@ describe('Kobo multi-device library sync (e2e)', { timeout: 180_000 }, () => {
const bFirst = await sync(deviceB); const bFirst = await sync(deviceB);
expect(aFirst.hasMore).toBe(true); expect(aFirst.hasMore).toBe(true);
expect(bFirst.hasMore).toBe(true); expect(bFirst.hasMore).toBe(true);
expect(entitlementIds(aFirst.entries)).toHaveLength(5); expect(entitlementIds(aFirst.entries)).toHaveLength(SYNC_PAGE_SIZE);
expect(entitlementIds(bFirst.entries)).toHaveLength(5); expect(entitlementIds(bFirst.entries)).toHaveLength(SYNC_PAGE_SIZE);
const aSecond = await sync(deviceA, aFirst.syncToken); const aSecond = await sync(deviceA, aFirst.syncToken);
expect(aSecond.hasMore).toBe(false); expect(aSecond.hasMore).toBe(false);
@@ -373,7 +377,7 @@ describe('Kobo multi-device library sync (e2e)', { timeout: 180_000 }, () => {
snapshots.map((snapshot) => snapshot.id), snapshots.map((snapshot) => snapshot.id),
), ),
); );
expect(rows).toHaveLength(12); expect(rows).toHaveLength(LIBRARY_BOOK_COUNT * 2);
expect(rows.every((row) => row.synced)).toBe(true); expect(rows.every((row) => row.synced)).toBe(true);
}); });
@@ -482,7 +486,7 @@ describe('Kobo multi-device library sync (e2e)', { timeout: 180_000 }, () => {
.map((snapshot) => snapshot.id), .map((snapshot) => snapshot.id),
), ),
); );
expect(aAndBRows).toHaveLength(12); expect(aAndBRows).toHaveLength(LIBRARY_BOOK_COUNT * 2);
expect(aAndBRows.every((row) => row.synced)).toBe(true); expect(aAndBRows.every((row) => row.synced)).toBe(true);
const [targetIdentity] = await ctx.db const [targetIdentity] = await ctx.db