fix(kobo): replace a stale delivery instead of failing its download forever

When a book's file changed after it was offered to a Kobo, every download
returned 409 because the offer could no longer be rebuilt. A device download
now retires that offer and serves a fresh one from the current file, and the
device bookmark adopts the copy it downloaded. Log the conflict code when a
sync defers a delivery. Also drop a dev-only relocate log from the reader.

Fixes #1816
This commit is contained in:
neon
2026-10-16 23:23:08 +00:00
parent 7f12e8cd6d
commit e070460876
6 changed files with 235 additions and 18 deletions
@@ -94,6 +94,39 @@ export class KoboDeliveryRepository {
});
}
/** Stops a delivery from standing for its book, so the next offer starts a fresh one. */
async retire(userId: number, deviceId: number, id: string): Promise<void> {
await this.db
.update(schema.koboDeviceDeliveries)
.set({ current: false, updatedAt: new Date() })
.where(
and(
eq(schema.koboDeviceDeliveries.userId, userId),
eq(schema.koboDeviceDeliveries.deviceId, deviceId),
eq(schema.koboDeviceDeliveries.id, id),
eq(schema.koboDeviceDeliveries.current, true),
),
);
}
/**
* Points the device's stored bookmark at a delivery it has just downloaded. Without this, a bookmark
* first attributed to an earlier copy keeps every later push from the new copy uncertain.
*/
async adoptForDeviceReadingState(userId: number, deviceId: number, bookId: number, id: string): Promise<void> {
await this.db
.update(schema.koboDeviceReadingStates)
.set({ deliveryId: id, updatedAt: new Date() })
.where(
and(
eq(schema.koboDeviceReadingStates.userId, userId),
eq(schema.koboDeviceReadingStates.deviceId, deviceId),
eq(schema.koboDeviceReadingStates.bookId, bookId),
sql`${schema.koboDeviceReadingStates.deliveryId} is distinct from ${id}`,
),
);
}
transition(userId: number, deviceId: number, id: string, expected: string[], state: string) {
return this.db
.update(schema.koboDeviceDeliveries)
@@ -0,0 +1,151 @@
import { ConflictException, ForbiddenException, NotFoundException } from '@nestjs/common';
import { describe, expect, it, vi } from 'vitest';
import { KoboDeliveryService, deliveryErrorCode, type KoboDeliveryRecord } from './kobo-delivery.service';
function record(overrides: Partial<KoboDeliveryRecord> = {}): KoboDeliveryRecord {
return {
id: 'stale',
userId: 1,
deviceId: 2,
bookId: 3,
sourceFileId: 4,
metadata: {
artifactId: 'artifact',
sourceRevision: 'revision',
requestedVariant: null,
actualVariant: 'original',
format: 'kepub',
sizeBytes: 100,
sha256: 'sha',
fileHash: 'hash',
outcome: 'original',
},
options: { hyphenate: false, converterVersion: 'v1' },
state: 'transfer_observed',
previousId: null,
current: true,
createdAt: new Date(),
updatedAt: new Date(),
...overrides,
};
}
function makeService() {
const repository = {
retire: vi.fn().mockResolvedValue(undefined),
reoffer: vi.fn().mockResolvedValue(undefined),
transition: vi.fn().mockResolvedValue([{}]),
adoptForDeviceReadingState: vi.fn().mockResolvedValue(undefined),
};
const service = new KoboDeliveryService(repository as never, {} as never, {} as never, {} as never, {} as never, {} as never, {} as never);
const offer = vi.spyOn(service, 'offer');
const resolve = vi.spyOn(service, 'resolve');
const artifact = (id: string) => ({ path: `/cache/${id}`, release: vi.fn() }) as never;
return { service, repository, offer, resolve, artifact };
}
describe('KoboDeliveryService.prepareDownload', () => {
it('serves the current offer when its source still rebuilds it', async () => {
const { service, repository, offer, resolve, artifact } = makeService();
const current = record();
offer.mockResolvedValue(current);
resolve.mockResolvedValue(artifact('stale'));
const result = await service.prepareDownload(1, 2, 3, 'stale');
expect(result.record).toBe(current);
expect(repository.retire).not.toHaveBeenCalled();
expect(repository.reoffer).not.toHaveBeenCalled();
});
it.each(['DELIVERY_SOURCE_CHANGED', 'DELIVERY_ARTIFACT_UNAVAILABLE'])(
'replaces an offer whose source changed (%s) with a fresh one of the current file',
async (code) => {
const { service, repository, offer, resolve, artifact } = makeService();
const stale = record();
const fresh = record({ id: 'fresh', state: 'offered' });
offer.mockResolvedValueOnce(stale).mockResolvedValueOnce(fresh);
const freshArtifact = artifact('fresh');
resolve.mockRejectedValueOnce(new ConflictException({ code })).mockResolvedValueOnce(freshArtifact);
const result = await service.prepareDownload(1, 2, 3, 'stale');
expect(result).toEqual({ record: fresh, artifact: freshArtifact });
expect(repository.retire).toHaveBeenCalledWith(1, 2, 'stale');
expect(repository.reoffer).toHaveBeenCalledWith(1, 2, 3);
expect(resolve).toHaveBeenLastCalledWith(1, 2, fresh);
},
);
it('asks the device to sync when the fresh offer changes format instead of sending it under the old one', async () => {
const { service, repository, offer, resolve } = makeService();
offer.mockResolvedValueOnce(record()).mockResolvedValueOnce(record({ id: 'fresh', metadata: { ...record().metadata, format: 'epub' } }));
resolve.mockRejectedValueOnce(new ConflictException({ code: 'DELIVERY_SOURCE_CHANGED' }));
const error = await service.prepareDownload(1, 2, 3).catch((caught: unknown) => caught);
expect(deliveryErrorCode(error)).toBe('KOBO_DELIVERY_FORMAT_CHANGED');
expect(repository.reoffer).toHaveBeenCalledWith(1, 2, 3);
expect(resolve).toHaveBeenCalledTimes(1);
});
it('keeps a pending replacement on its guided flow', async () => {
const { service, repository, offer, resolve } = makeService();
offer.mockResolvedValue(record({ options: { hyphenate: false, converterVersion: 'v1', replacement: true } }));
resolve.mockRejectedValue(new ConflictException({ code: 'DELIVERY_SOURCE_CHANGED' }));
await expect(service.prepareDownload(1, 2, 3)).rejects.toThrow(ConflictException);
expect(repository.retire).not.toHaveBeenCalled();
});
it('does not replace a previous delivery standing in for a cancelled replacement', async () => {
const { service, repository, offer, resolve } = makeService();
offer.mockResolvedValue(record({ current: false }));
resolve.mockRejectedValue(new ConflictException({ code: 'DELIVERY_SOURCE_CHANGED' }));
await expect(service.prepareDownload(1, 2, 3)).rejects.toThrow(ConflictException);
expect(repository.retire).not.toHaveBeenCalled();
});
it.each([new ForbiddenException({ code: 'DELIVERY_SOURCE_OUTSIDE_LIBRARY' }), new ConflictException({ code: 'DELIVERY_SOURCE_UNAVAILABLE' })])(
'rethrows failures that a fresh offer cannot fix: %s',
async (failure) => {
const { service, repository, offer, resolve } = makeService();
offer.mockResolvedValue(record());
resolve.mockRejectedValue(failure);
await expect(service.prepareDownload(1, 2, 3)).rejects.toBe(failure);
expect(repository.retire).not.toHaveBeenCalled();
},
);
it('rejects a delivery id that is no longer current without preparing any bytes', async () => {
const { service, offer, resolve } = makeService();
offer.mockResolvedValue(record({ id: 'fresh' }));
await expect(service.prepareDownload(1, 2, 3, 'stale')).rejects.toThrow(NotFoundException);
expect(resolve).not.toHaveBeenCalled();
});
});
describe('KoboDeliveryService.transferObserved', () => {
it('points the device bookmark at a delivery it just downloaded', async () => {
const { service, repository } = makeService();
await service.transferObserved(1, 2, record({ id: 'fresh', state: 'offered' }));
expect(repository.transition).toHaveBeenCalledWith(1, 2, 'fresh', ['offered', 'transfer_observed'], 'transfer_observed');
expect(repository.adoptForDeviceReadingState).toHaveBeenCalledWith(1, 2, 3, 'fresh');
});
it('leaves the bookmark alone for a replacement or a delivery that is no longer transferable', async () => {
const { service, repository } = makeService();
await service.transferObserved(1, 2, record({ options: { hyphenate: false, converterVersion: 'v1', replacement: true } }));
repository.transition.mockResolvedValueOnce([]);
await service.transferObserved(1, 2, record({ state: 'cancelled' }));
expect(repository.adoptForDeviceReadingState).not.toHaveBeenCalled();
});
});
@@ -1,6 +1,6 @@
import { createHash } from 'crypto';
import { open } from 'fs/promises';
import { ConflictException, Injectable, NotFoundException, Logger } from '@nestjs/common';
import { ConflictException, HttpException, Injectable, NotFoundException, Logger } from '@nestjs/common';
import type { ReadAloudDeliveryPreference } from '@bookorbit/types';
import { DeliveryArtifactService, type DeliveryArtifact } from '../../book/delivery-artifact.service';
import { KoboDeliveryRepository } from '../kobo-delivery.repository';
@@ -14,6 +14,15 @@ import { computeFileHashFromHandle } from '../../../common/utils/file-hash.utils
export type KoboDeliveryRecord = NonNullable<Awaited<ReturnType<KoboDeliveryRepository['findCurrent']>>>;
/** Failures meaning the source can no longer rebuild an offer's bytes, as opposed to the book being unreadable. */
const STALE_DELIVERY_CODES = new Set(['DELIVERY_SOURCE_CHANGED', 'DELIVERY_ARTIFACT_UNAVAILABLE']);
export function deliveryErrorCode(error: unknown): string | null {
if (!(error instanceof HttpException)) return null;
const response = error.getResponse();
return typeof response === 'object' && response !== null && 'code' in response && typeof response.code === 'string' ? response.code : null;
}
@Injectable()
export class KoboDeliveryService {
private readonly logger = new Logger(KoboDeliveryService.name);
@@ -123,6 +132,42 @@ export class KoboDeliveryService {
}
}
/**
* The bytes a device download is served from. A download asks for the book now, so an offer whose source
* can no longer rebuild it, because the file changed after it was offered, is replaced by a fresh offer of
* the current file instead of failing that book on every attempt. Pending replacements keep their guided flow.
*/
async prepareDownload(
userId: number,
deviceId: number,
bookId: number,
deliveryId?: string,
): Promise<{ record: KoboDeliveryRecord; artifact: DeliveryArtifact }> {
const record = await this.offer(userId, deviceId, bookId);
if (deliveryId && record.id !== deliveryId) throw new NotFoundException('Delivery offer is no longer current');
try {
return { record, artifact: await this.resolve(userId, deviceId, record) };
} catch (error) {
const code = deliveryErrorCode(error);
if (!record.current || record.options.replacement || !code || !STALE_DELIVERY_CODES.has(code)) throw error;
const startedAt = Date.now();
await this.repository.retire(userId, deviceId, record.id);
const fresh = await this.offer(userId, deviceId, bookId);
// The device learns the fresh offer's size and URL on its next sync either way.
await this.repository.reoffer(userId, deviceId, bookId);
if (fresh.metadata.format !== record.metadata.format) {
this.logger.warn(
`[kobo.delivery_refresh] [fail] userId=${userId} deviceId=${deviceId} bookId=${bookId} staleId=${record.id} freshId=${fresh.id} reason=${code} durationMs=${Date.now() - startedAt} - fresh offer changes format, the device must sync before downloading`,
);
throw new ConflictException({ code: 'KOBO_DELIVERY_FORMAT_CHANGED' });
}
this.logger.log(
`[kobo.delivery_refresh] [end] userId=${userId} deviceId=${deviceId} bookId=${bookId} staleId=${record.id} freshId=${fresh.id} reason=${code} durationMs=${Date.now() - startedAt} - stale offer replaced from the current file`,
);
return { record: fresh, artifact: await this.resolve(userId, deviceId, fresh) };
}
}
async resolve(userId: number, deviceId: number, record: KoboDeliveryRecord): Promise<DeliveryArtifact> {
await this.repository.assertDevice(userId, deviceId);
await this.access.assertBookAccessible(userId, record.bookId);
@@ -222,6 +267,7 @@ export class KoboDeliveryService {
}
async transferObserved(userId: number, deviceId: number, record: KoboDeliveryRecord) {
await this.repository.transition(userId, deviceId, record.id, ['offered', 'transfer_observed'], 'transfer_observed');
const moved = await this.repository.transition(userId, deviceId, record.id, ['offered', 'transfer_observed'], 'transfer_observed');
if (moved.length && !record.options.replacement) await this.repository.adoptForDeviceReadingState(userId, deviceId, record.bookId, record.id);
}
}
@@ -61,9 +61,7 @@ export class KoboDownloadService implements OnModuleInit {
async streamBook(userId: number, bookId: number, reply: FastifyReply, deviceId?: number, deliveryId?: string) {
if (deviceId && this.deliveries && this.selections) {
const record = await this.deliveries.offer(userId, deviceId, bookId);
if (deliveryId && record.id !== deliveryId) throw new NotFoundException('Delivery offer is no longer current');
const artifact = await this.deliveries.resolve(userId, deviceId, record);
const { record, artifact } = await this.deliveries.prepareDownload(userId, deviceId, bookId, deliveryId);
try {
if (record.sourceFileId !== null) await this.hashRegistration.record(record.sourceFileId, artifact.fileHash);
} catch (error) {
@@ -14,7 +14,7 @@ import { ContentFilterRepository } from '../../user/content-filter.repository';
import { BookQueryBuilder } from '../../book/book-query-builder.service';
import { MetadataExtractionService } from '../../metadata/metadata-extraction.service';
import { SmartScopeService } from '../../smart-scope/smart-scope.service';
import { KoboDeliveryService } from './kobo-delivery.service';
import { KoboDeliveryService, deliveryErrorCode } from './kobo-delivery.service';
import { KoboBookAccessService } from './kobo-book-access.service';
import { KoboBookIdentityService } from './kobo-book-identity.service';
import { normalizeKoboLanguage, selectKoboIsbn } from './kobo-metadata.utils';
@@ -657,7 +657,7 @@ export class KoboSyncService {
}
} catch (error) {
this.logger.warn(
`[kobo.delivery] [fail] userId=${userId} deviceId=${snapshot?.deviceId ?? 'none'} bookId=${row.bookId} errorClass=${error instanceof Error ? error.name : 'Error'} error="${sanitizeLogValue(error instanceof Error ? error.message : String(error))}" - preparation deferred for retry`,
`[kobo.delivery] [fail] userId=${userId} deviceId=${snapshot?.deviceId ?? 'none'} bookId=${row.bookId} code=${deliveryErrorCode(error) ?? 'none'} errorClass=${error instanceof Error ? error.name : 'Error'} error="${sanitizeLogValue(error instanceof Error ? error.message : String(error))}" - preparation deferred for retry`,
);
await this.db
.update(schema.koboSnapshotBooks)