fix(kobo): stop store sync looping when Kobo's cursor is too large to carry

A store cursor over the carry limit was dropped while the device was still
told to continue, so the store restarted at its first page forever. Stop
store paging for that sync instead, and carry Kobo's cursor verbatim in a
compact PX2 token so the limit can rise to 7,000 characters. Tokens in the
old format still decode.

Fixes #1679

server/src/modules/kobo/kobo-sync.controller.test.ts

@@ -280,10 +280,7 @@ describe('KoboSyncController', () => {
       await runSync(reply, 'kobo-native-token');
 
       expect(proxyService.request).not.toHaveBeenCalled();
-      expect(reply.header).toHaveBeenCalledWith(
-        'x-kobo-synctoken',
-        `PX.${Buffer.from(JSON.stringify({ snapshotId: 4, koboSyncToken: 'kobo-native-token' })).toString('base64')}`,
-      );
+      expect(reply.header).toHaveBeenCalledWith('x-kobo-synctoken', 'PX2.4.kobo-native-token');
     });
 
     it('appends store entitlements and stamps Kobo’s cursor into the sync token', async () => {
@@ -298,10 +295,7 @@ describe('KoboSyncController', () => {
         expect.objectContaining({ extraHeaders: { 'x-kobo-synctoken': 'kobo-cursor-1' } }),
       );
       expect(reply.send).toHaveBeenCalledWith([{ Local: 1 }, { Store: 1 }, { Store: 2 }]);
-      expect(reply.header).toHaveBeenCalledWith(
-        'x-kobo-synctoken',
-        `PX.${Buffer.from(JSON.stringify({ snapshotId: 4, koboSyncToken: 'kobo-cursor-2' })).toString('base64')}`,
-      );
+      expect(reply.header).toHaveBeenCalledWith('x-kobo-synctoken', 'PX2.4.kobo-cursor-2');
       expect(historyService.recordSuccess).toHaveBeenCalledWith(
         expect.objectContaining({ counts: { entitlements: 3, hasMore: false, storeEntitlements: 2 } }),
       );
@@ -388,7 +382,7 @@ describe('KoboSyncController', () => {
     });
 
     it('drops an upstream cursor too large to carry and leaves the token without one', async () => {
-      const oversized = 'x'.repeat(4097);
+      const oversized = 'x'.repeat(7001);
       proxyService.request.mockResolvedValue(storeResponse([{ Store: 1 }], { 'x-kobo-synctoken': oversized }));
       const reply = makeReply();
 
@@ -398,15 +392,49 @@ describe('KoboSyncController', () => {
       expect(reply.header).toHaveBeenCalledWith('x-kobo-synctoken', SNAPSHOT_TOKEN);
     });
 
+    it('stops paging when the store says continue with a cursor too large to carry, since the next page would restart the store', async () => {
+      proxyService.request.mockResolvedValue(storeResponse([{ Store: 1 }], { 'x-kobo-sync': 'continue', 'x-kobo-synctoken': 'x'.repeat(7001) }));
+      const reply = makeReply();
+
+      await runSync(reply, SNAPSHOT_TOKEN);
+
+      expect(reply.header).toHaveBeenCalledWith('x-kobo-sync', '');
+      expect(reply.header).toHaveBeenCalledWith('x-kobo-synctoken', SNAPSHOT_TOKEN);
+      expect(reply.send).toHaveBeenCalledWith([{ Local: 1 }, { Store: 1 }]);
+      expect(historyService.recordSuccess).toHaveBeenCalledWith(
+        expect.objectContaining({ counts: { entitlements: 2, hasMore: false, storeEntitlements: 1 } }),
+      );
+    });
+
+    it('keeps paging with a cursor the size of a real continuation token', async () => {
+      const continuation = `eyJ${'a'.repeat(4600)}.${'b'.repeat(40)}`;
+      proxyService.request.mockResolvedValue(storeResponse([{ Store: 1 }], { 'x-kobo-sync': 'continue', 'x-kobo-synctoken': continuation }));
+      const reply = makeReply();
+
+      await runSync(reply, SNAPSHOT_TOKEN);
+
+      expect(reply.header).toHaveBeenCalledWith('x-kobo-sync', 'continue');
+      expect(reply.header).toHaveBeenCalledWith('x-kobo-synctoken', `PX2.4.${continuation}`);
+    });
+
     it('preserves a known Kobo cursor across a failed store call so paging resumes next sync', async () => {
       proxyService.request.mockRejectedValue(new Error('upstream down'));
       const reply = makeReply();
 
       await runSync(reply, `PX.${Buffer.from(JSON.stringify({ snapshotId: 4, koboSyncToken: 'kobo-cursor-1' })).toString('base64')}`);
 
-      expect(reply.header).toHaveBeenCalledWith(
-        'x-kobo-synctoken',
-        `PX.${Buffer.from(JSON.stringify({ snapshotId: 4, koboSyncToken: 'kobo-cursor-1' })).toString('base64')}`,
+      expect(reply.header).toHaveBeenCalledWith('x-kobo-synctoken', 'PX2.4.kobo-cursor-1');
+    });
+
+    it('forwards the Kobo cursor from a compact token upstream', async () => {
+      proxyService.request.mockResolvedValue(storeResponse([]));
+
+      await runSync(makeReply(), 'PX2.4.eyJhbGciOiJub25lIn0.eyJ0Ijo1fQ');
+
+      expect(proxyService.request).toHaveBeenCalledWith(
+        expect.anything(),
+        'token-9',
+        expect.objectContaining({ extraHeaders: { 'x-kobo-synctoken': 'eyJhbGciOiJub25lIn0.eyJ0Ijo1fQ' } }),
       );
     });
   });

server/src/modules/kobo/kobo-sync.controller.ts

@@ -176,15 +176,17 @@ export class KoboSyncController {
 
     // Kobo's cursor rides inside our token, so one too large to carry means the next store sync
     // starts over. Say so rather than letting the store silently resend its whole library.
-    if (store?.syncToken !== undefined && !isUsableKoboSyncToken(store.syncToken)) {
+    const storeCursorCarried = store?.syncToken !== undefined && isUsableKoboSyncToken(store.syncToken);
+    if (store?.syncToken !== undefined && !storeCursorCarried) {
       this.logger.warn(
         `[kobo.store_sync] [fail] userId=${user.id} deviceId=${device.deviceId} errorClass=StoreCursorTooLarge tokenLength=${store.syncToken.length} - upstream cursor dropped, the next store sync starts from the beginning`,
       );
     }
 
-    // Replaying an unchanged cursor returns the same page forever, so keep paging the store only
-    // while its token actually advances. The page just fetched is still delivered either way.
-    const storeCursorAdvanced = store?.syncToken !== undefined && store.syncToken !== incomingKoboSyncToken;
+    // Replaying an unchanged cursor returns the same page forever, and so does a dropped one, which
+    // sends the store back to its first page. Keep paging the store only while the cursor the device
+    // will send back actually advances. The page just fetched is still delivered either way.
+    const storeCursorAdvanced = storeCursorCarried && store?.syncToken !== incomingKoboSyncToken;
     if (store?.hasMore === true && !storeCursorAdvanced) {
       this.logger.warn(
         `[kobo.store_sync] [fail] userId=${user.id} deviceId=${device.deviceId} errorClass=StoreCursorStalled error="continue without an advanced sync token" - store paging stopped to avoid a sync loop`,

server/src/modules/kobo/services/kobo-proxy.service.test.ts

@@ -265,6 +265,10 @@ describe('KoboProxyService', () => {
       expect((await sentHeaders({ 'x-kobo-synctoken': composite }))['x-kobo-synctoken']).toBe('kobo-cursor-9');
     });
 
+    it('substitutes the Kobo cursor for a compact composite token', async () => {
+      expect((await sentHeaders({ 'x-kobo-synctoken': 'PX2.4.eyJ0Ijo5fQ.c2ln' }))['x-kobo-synctoken']).toBe('eyJ0Ijo5fQ.c2ln');
+    });
+
     it('drops a composite token it cannot decode rather than forwarding it', async () => {
       expect(await sentHeaders({ 'x-kobo-synctoken': 'PX.not-base64-json' })).not.toHaveProperty('x-kobo-synctoken');
     });

server/src/modules/kobo/services/kobo-sync-token.test.ts

@@ -12,6 +12,23 @@ describe('kobo sync token', () => {
     expect(decodeSyncToken(encodeSyncToken(7, 'kobo-cursor'))).toEqual({ snapshotId: 7, koboSyncToken: 'kobo-cursor' });
   });
 
+  it('carries Kobo’s cursor verbatim instead of re-encoding it', () => {
+    const cursor = 'eyJJbnRlcm5hbFN5bmNUb2tlbiI6ImV5SjAifQ.eyJ0eXAiOjF9';
+
+    expect(encodeSyncToken(7, cursor)).toBe(`PX2.7.${cursor}`);
+    expect(decodeSyncToken(`PX2.7.${cursor}`)).toEqual({ snapshotId: 7, koboSyncToken: cursor });
+  });
+
+  it('still reads a Kobo cursor from tokens minted in the JSON format', () => {
+    const legacy = `PX.${Buffer.from(JSON.stringify({ snapshotId: 3, koboSyncToken: 'kobo-cursor' })).toString('base64')}`;
+
+    expect(decodeSyncToken(legacy)).toEqual({ snapshotId: 3, koboSyncToken: 'kobo-cursor' });
+  });
+
+  it('keeps the header within a proxy line limit for the largest cursor it carries', () => {
+    expect(encodeSyncToken(2_147_483_647, 'x'.repeat(7000)).length).toBeLessThan(8000);
+  });
+
   it('reads tokens minted before store sync existed', () => {
     const legacy = `PX.${Buffer.from(JSON.stringify({ snapshotId: 3 })).toString('base64')}`;
 
@@ -29,6 +46,8 @@ describe('kobo sync token', () => {
     ['malformed base64 payload', 'PX.!!!not-base64!!!'],
     ['a JSON array payload', `PX.${Buffer.from(JSON.stringify([1, 2])).toString('base64')}`],
     ['a non-object payload', `PX.${Buffer.from(JSON.stringify('nope')).toString('base64')}`],
+    ['a compact token with a non-numeric snapshot', 'PX2.abc.kobo-cursor'],
+    ['a compact token with nothing after the prefix', 'PX2.'],
   ])('never throws on %s', (_label, raw) => {
     expect(() => decodeSyncToken(raw)).not.toThrow();
     expect(decodeSyncToken(raw).snapshotId).toBeUndefined();
@@ -36,13 +55,13 @@ describe('kobo sync token', () => {
 
   it('reports which upstream cursors can be carried, so callers can log the ones that cannot', () => {
     expect(isUsableKoboSyncToken('kobo-cursor')).toBe(true);
-    expect(isUsableKoboSyncToken('x'.repeat(4096))).toBe(true);
-    expect(isUsableKoboSyncToken('x'.repeat(4097))).toBe(false);
+    expect(isUsableKoboSyncToken('x'.repeat(7000))).toBe(true);
+    expect(isUsableKoboSyncToken('x'.repeat(7001))).toBe(false);
     expect(isUsableKoboSyncToken('')).toBe(false);
   });
 
   it('drops an oversized upstream cursor rather than growing the response header without bound', () => {
-    const oversized = 'x'.repeat(4097);
+    const oversized = 'x'.repeat(7001);
 
     expect(decodeSyncToken(oversized)).toEqual({});
     expect(decodeSyncToken(encodeSyncToken(7, oversized))).toEqual({ snapshotId: 7, koboSyncToken: undefined });

server/src/modules/kobo/services/kobo-sync-token.ts

@@ -1,9 +1,14 @@
 const TOKEN_PREFIX = 'PX.';
+// Carries Kobo's cursor verbatim after the snapshot id: `PX2.<snapshotId>.<kobo token>`. Kobo's token
+// already travels in a header, and wrapping it in JSON and base64 again grew it by a third.
+const COMPACT_TOKEN_PREFIX = 'PX2.';
 
 // Kobo's own sync tokens are opaque and we round-trip them through the device inside ours, so a
 // hostile or corrupted upstream value cannot be allowed to grow the response header without bound.
-// Kobo's real tokens sit far under this; anything larger is dropped and store paging restarts.
-const MAX_KOBO_SYNC_TOKEN_LENGTH = 4096;
+// A small store library already produces a 3,300 character cursor and continuation pages carry more,
+// so this leaves room for those while the whole header stays under the common 8 KB proxy line limit.
+// Anything larger is dropped, and store paging for that sync stops instead of restarting.
+const MAX_KOBO_SYNC_TOKEN_LENGTH = 7000;
 
 export type KoboSyncTokenParts = {
   snapshotId?: number;
@@ -20,8 +25,8 @@ export function isUsableKoboSyncToken(koboSyncToken: string): boolean {
 
 export function encodeSyncToken(snapshotId: number, koboSyncToken?: string): string {
   const usableKoboToken = koboSyncToken && isUsableKoboSyncToken(koboSyncToken) ? koboSyncToken : undefined;
-  const payload: KoboSyncTokenParts = usableKoboToken ? { snapshotId, koboSyncToken: usableKoboToken } : { snapshotId };
-  return TOKEN_PREFIX + Buffer.from(JSON.stringify(payload)).toString('base64');
+  if (usableKoboToken) return `${COMPACT_TOKEN_PREFIX}${snapshotId}.${usableKoboToken}`;
+  return TOKEN_PREFIX + Buffer.from(JSON.stringify({ snapshotId })).toString('base64');
 }
 
 /**
@@ -33,6 +38,16 @@ export function encodeSyncToken(snapshotId: number, koboSyncToken?: string): str
 export function decodeSyncToken(raw: string | undefined): KoboSyncTokenParts {
   const value = raw?.trim();
   if (!value) return {};
+  if (value.startsWith(COMPACT_TOKEN_PREFIX)) {
+    const rest = value.slice(COMPACT_TOKEN_PREFIX.length);
+    const separator = rest.indexOf('.');
+    const id = separator === -1 ? rest : rest.slice(0, separator);
+    const koboSyncToken = separator === -1 ? '' : rest.slice(separator + 1);
+    return {
+      snapshotId: /^\d{1,15}$/.test(id) ? Number(id) : undefined,
+      koboSyncToken: isUsableKoboSyncToken(koboSyncToken) ? koboSyncToken : undefined,
+    };
+  }
   if (!value.startsWith(TOKEN_PREFIX)) {
     return isUsableKoboSyncToken(value) ? { koboSyncToken: value } : {};
   }