mirror of
https://github.com/bookorbit/bookorbit.git
synced 2026-11-03 00:11:37 -06:00
fix(authors): chunk inserts to prevent call stack overflow on backfill
The drizzle-orm query builder hits a RangeError when generating a single bulk INSERT for tens of thousands of authors during backfill. Chunking the insertions into smaller batches resolves the issue. Closes #452
This commit is contained in:
@@ -98,6 +98,24 @@ describe('AuthorEnrichmentRepository', () => {
|
||||
);
|
||||
});
|
||||
|
||||
it('upsertSchedule chunks large author id lists', async () => {
|
||||
const { db, insertBuilder } = makeDb();
|
||||
insertBuilder.returning.mockResolvedValueOnce([{ authorId: 1 }, { authorId: 2 }]).mockResolvedValueOnce([{ authorId: 3 }]);
|
||||
const repo = new AuthorEnrichmentRepository(db as never);
|
||||
|
||||
const count = await repo.upsertSchedule([1, 2, 3], 'metadata_replace', 2);
|
||||
|
||||
expect(count).toBe(3);
|
||||
expect(db.insert).toHaveBeenCalledTimes(2);
|
||||
expect(insertBuilder.values).toHaveBeenNthCalledWith(1, [
|
||||
expect.objectContaining({ authorId: 1, status: 'queued', reason: 'metadata_replace', attemptCount: 0 }),
|
||||
expect.objectContaining({ authorId: 2, status: 'queued', reason: 'metadata_replace', attemptCount: 0 }),
|
||||
]);
|
||||
expect(insertBuilder.values).toHaveBeenNthCalledWith(2, [
|
||||
expect.objectContaining({ authorId: 3, status: 'queued', reason: 'metadata_replace', attemptCount: 0 }),
|
||||
]);
|
||||
});
|
||||
|
||||
it('fetchDue filters by active statuses and due time', async () => {
|
||||
const { db, selectBuilder } = makeDb();
|
||||
const repo = new AuthorEnrichmentRepository(db as never);
|
||||
@@ -197,7 +215,7 @@ describe('AuthorEnrichmentRepository', () => {
|
||||
|
||||
it('enqueueEligibleLinkedAuthors selects eligible linked ids directly in SQL', async () => {
|
||||
const { db, selectBuilder, insertBuilder } = makeDb();
|
||||
selectBuilder.where.mockResolvedValueOnce([{ authorId: 31 }, { authorId: 32 }]);
|
||||
selectBuilder.limit.mockResolvedValueOnce([{ authorId: 31 }, { authorId: 32 }]);
|
||||
insertBuilder.returning.mockResolvedValueOnce([{ authorId: 31 }, { authorId: 32 }]);
|
||||
const repo = new AuthorEnrichmentRepository(db as never);
|
||||
|
||||
@@ -209,6 +227,34 @@ describe('AuthorEnrichmentRepository', () => {
|
||||
|
||||
expect(queued).toBe(2);
|
||||
expect(db.selectDistinct).toHaveBeenCalled();
|
||||
expect(selectBuilder.orderBy).toHaveBeenCalled();
|
||||
expect(selectBuilder.limit).toHaveBeenCalledWith(1000);
|
||||
});
|
||||
|
||||
it('enqueueEligibleLinkedAuthors walks cursor pages and accumulates queued totals', async () => {
|
||||
const { db, selectBuilder } = makeDb();
|
||||
selectBuilder.limit
|
||||
.mockResolvedValueOnce([{ authorId: 10 }, { authorId: 20 }])
|
||||
.mockResolvedValueOnce([{ authorId: 40 }])
|
||||
.mockResolvedValueOnce([]);
|
||||
const repo = new AuthorEnrichmentRepository(db as never);
|
||||
const upsert = vi.spyOn(repo, 'upsertSchedule').mockResolvedValueOnce(2).mockResolvedValueOnce(1);
|
||||
|
||||
const queued = await repo.enqueueEligibleLinkedAuthors(
|
||||
AUTHOR_ENRICHMENT_REASONS.MANUAL_BACKFILL,
|
||||
{
|
||||
neverEnriched: false,
|
||||
missingBio: true,
|
||||
missingPhoto: false,
|
||||
},
|
||||
2,
|
||||
);
|
||||
|
||||
expect(queued).toBe(3);
|
||||
expect(upsert).toHaveBeenNthCalledWith(1, [10, 20], AUTHOR_ENRICHMENT_REASONS.MANUAL_BACKFILL, 2);
|
||||
expect(upsert).toHaveBeenNthCalledWith(2, [40], AUTHOR_ENRICHMENT_REASONS.MANUAL_BACKFILL, 2);
|
||||
expect(selectBuilder.limit).toHaveBeenCalledTimes(3);
|
||||
expect(selectBuilder.limit).toHaveBeenCalledWith(2);
|
||||
});
|
||||
|
||||
it('countEligibleLinkedAuthors returns SQL count without materializing all ids', async () => {
|
||||
@@ -266,12 +312,28 @@ describe('AuthorEnrichmentRepository', () => {
|
||||
|
||||
it('enqueueAllLinkedAuthors schedules distinct linked author ids', async () => {
|
||||
const { db, selectBuilder, insertBuilder } = makeDb();
|
||||
selectBuilder.from.mockResolvedValueOnce([{ authorId: 11 }, { authorId: 12 }]);
|
||||
selectBuilder.limit.mockResolvedValueOnce([{ authorId: 11 }, { authorId: 12 }]);
|
||||
insertBuilder.returning.mockResolvedValueOnce([{ authorId: 11 }, { authorId: 12 }]);
|
||||
const repo = new AuthorEnrichmentRepository(db as never);
|
||||
|
||||
await expect(repo.enqueueAllLinkedAuthors(AUTHOR_ENRICHMENT_REASONS.MANUAL_BACKFILL_ALL)).resolves.toBe(2);
|
||||
expect(db.selectDistinct).toHaveBeenCalled();
|
||||
expect(selectBuilder.orderBy).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('enqueueAllLinkedAuthors walks cursor pages and accumulates queued totals', async () => {
|
||||
const { db, selectBuilder } = makeDb();
|
||||
selectBuilder.limit
|
||||
.mockResolvedValueOnce([{ authorId: 11 }, { authorId: 12 }])
|
||||
.mockResolvedValueOnce([{ authorId: 15 }])
|
||||
.mockResolvedValueOnce([]);
|
||||
const repo = new AuthorEnrichmentRepository(db as never);
|
||||
const upsert = vi.spyOn(repo, 'upsertSchedule').mockResolvedValueOnce(2).mockResolvedValueOnce(1);
|
||||
|
||||
await expect(repo.enqueueAllLinkedAuthors(AUTHOR_ENRICHMENT_REASONS.MANUAL_BACKFILL_ALL, 2)).resolves.toBe(3);
|
||||
expect(upsert).toHaveBeenNthCalledWith(1, [11, 12], AUTHOR_ENRICHMENT_REASONS.MANUAL_BACKFILL_ALL, 2);
|
||||
expect(upsert).toHaveBeenNthCalledWith(2, [15], AUTHOR_ENRICHMENT_REASONS.MANUAL_BACKFILL_ALL, 2);
|
||||
expect(selectBuilder.limit).toHaveBeenCalledTimes(3);
|
||||
});
|
||||
|
||||
it('markDone clears queue row and updates author enrichment fields', async () => {
|
||||
|
||||
@@ -11,6 +11,8 @@ import { AuthorEnrichmentReason } from './author-enrichment-reasons';
|
||||
|
||||
type Db = NodePgDatabase<typeof schema>;
|
||||
|
||||
const DEFAULT_SCHEDULE_BATCH_SIZE = 1000;
|
||||
|
||||
export const AUTHOR_ENRICHMENT_ACTIVE_STATUSES = ['queued', 'rate_limited'] as const;
|
||||
|
||||
export type AuthorEnrichmentQueueRow = typeof authorEnrichmentQueue.$inferSelect;
|
||||
@@ -19,15 +21,28 @@ export type AuthorEnrichmentQueueRow = typeof authorEnrichmentQueue.$inferSelect
|
||||
export class AuthorEnrichmentRepository {
|
||||
constructor(@Inject(DB) private readonly db: Db) {}
|
||||
|
||||
async upsertSchedule(authorIds: number[], reason: AuthorEnrichmentReason): Promise<number> {
|
||||
async upsertSchedule(authorIds: number[], reason: AuthorEnrichmentReason, batchSize = DEFAULT_SCHEDULE_BATCH_SIZE): Promise<number> {
|
||||
const uniqueAuthorIds = [...new Set(authorIds)].filter((id) => Number.isInteger(id) && id > 0);
|
||||
if (uniqueAuthorIds.length === 0) return 0;
|
||||
|
||||
const resolvedBatchSize = resolveBatchSize(batchSize);
|
||||
if (!resolvedBatchSize) return 0;
|
||||
|
||||
const now = new Date();
|
||||
let totalTouched = 0;
|
||||
for (let offset = 0; offset < uniqueAuthorIds.length; offset += resolvedBatchSize) {
|
||||
const batch = uniqueAuthorIds.slice(offset, offset + resolvedBatchSize);
|
||||
totalTouched += await this.upsertScheduleBatch(batch, reason, now);
|
||||
}
|
||||
|
||||
return totalTouched;
|
||||
}
|
||||
|
||||
private async upsertScheduleBatch(authorIds: number[], reason: AuthorEnrichmentReason, now: Date): Promise<number> {
|
||||
const touched = await this.db
|
||||
.insert(authorEnrichmentQueue)
|
||||
.values(
|
||||
uniqueAuthorIds.map((authorId) => ({
|
||||
authorIds.map((authorId) => ({
|
||||
authorId,
|
||||
status: 'queued',
|
||||
reason,
|
||||
@@ -51,36 +66,95 @@ export class AuthorEnrichmentRepository {
|
||||
return touched.length;
|
||||
}
|
||||
|
||||
async enqueueAllLinkedAuthors(reason: AuthorEnrichmentReason): Promise<number> {
|
||||
const rows = await this.db.selectDistinct({ authorId: bookAuthors.authorId }).from(bookAuthors);
|
||||
const authorIds = rows.map((row) => row.authorId);
|
||||
return this.upsertSchedule(authorIds, reason);
|
||||
async enqueueAllLinkedAuthors(reason: AuthorEnrichmentReason, batchSize = DEFAULT_SCHEDULE_BATCH_SIZE): Promise<number> {
|
||||
const resolvedBatchSize = resolveBatchSize(batchSize);
|
||||
if (!resolvedBatchSize) return 0;
|
||||
|
||||
let cursorAuthorId = 0;
|
||||
let totalQueued = 0;
|
||||
|
||||
for (;;) {
|
||||
const rows = await this.db
|
||||
.selectDistinct({ authorId: bookAuthors.authorId })
|
||||
.from(bookAuthors)
|
||||
.where(sql`${bookAuthors.authorId} > ${cursorAuthorId}`)
|
||||
.orderBy(asc(bookAuthors.authorId))
|
||||
.limit(resolvedBatchSize);
|
||||
|
||||
if (rows.length === 0) break;
|
||||
|
||||
totalQueued += await this.upsertSchedule(
|
||||
rows.map((row) => row.authorId),
|
||||
reason,
|
||||
resolvedBatchSize,
|
||||
);
|
||||
cursorAuthorId = rows[rows.length - 1]!.authorId;
|
||||
}
|
||||
|
||||
return totalQueued;
|
||||
}
|
||||
|
||||
async filterEligibleAuthorIds(authorIds: number[], conditions: AuthorEnrichmentConditions): Promise<number[]> {
|
||||
async filterEligibleAuthorIds(
|
||||
authorIds: number[],
|
||||
conditions: AuthorEnrichmentConditions,
|
||||
batchSize = DEFAULT_SCHEDULE_BATCH_SIZE,
|
||||
): Promise<number[]> {
|
||||
if (authorIds.length === 0) return [];
|
||||
|
||||
const eligibility = this.buildEligibilityPredicate(conditions);
|
||||
if (!eligibility) return [];
|
||||
|
||||
const rows = await this.db
|
||||
.select({ id: authors.id })
|
||||
.from(authors)
|
||||
.where(and(inArray(authors.id, authorIds), eligibility));
|
||||
const resolvedBatchSize = resolveBatchSize(batchSize);
|
||||
if (!resolvedBatchSize) return [];
|
||||
|
||||
return rows.map((r) => r.id);
|
||||
const eligibleIds: number[] = [];
|
||||
const uniqueAuthorIds = [...new Set(authorIds)].filter((id) => Number.isInteger(id) && id > 0);
|
||||
for (let offset = 0; offset < uniqueAuthorIds.length; offset += resolvedBatchSize) {
|
||||
const batch = uniqueAuthorIds.slice(offset, offset + resolvedBatchSize);
|
||||
const rows = await this.db
|
||||
.select({ id: authors.id })
|
||||
.from(authors)
|
||||
.where(and(inArray(authors.id, batch), eligibility));
|
||||
eligibleIds.push(...rows.map((r) => r.id));
|
||||
}
|
||||
|
||||
return eligibleIds;
|
||||
}
|
||||
|
||||
async enqueueEligibleLinkedAuthors(reason: AuthorEnrichmentReason, conditions: AuthorEnrichmentConditions): Promise<number> {
|
||||
async enqueueEligibleLinkedAuthors(
|
||||
reason: AuthorEnrichmentReason,
|
||||
conditions: AuthorEnrichmentConditions,
|
||||
batchSize = DEFAULT_SCHEDULE_BATCH_SIZE,
|
||||
): Promise<number> {
|
||||
const eligibility = this.buildEligibilityPredicate(conditions);
|
||||
if (!eligibility) return 0;
|
||||
const rows = await this.db
|
||||
.selectDistinct({ authorId: bookAuthors.authorId })
|
||||
.from(bookAuthors)
|
||||
.innerJoin(authors, eq(authors.id, bookAuthors.authorId))
|
||||
.where(eligibility);
|
||||
const eligibleIds = rows.map((row) => row.authorId);
|
||||
return this.upsertSchedule(eligibleIds, reason);
|
||||
|
||||
const resolvedBatchSize = resolveBatchSize(batchSize);
|
||||
if (!resolvedBatchSize) return 0;
|
||||
|
||||
let cursorAuthorId = 0;
|
||||
let totalQueued = 0;
|
||||
|
||||
for (;;) {
|
||||
const rows = await this.db
|
||||
.selectDistinct({ authorId: bookAuthors.authorId })
|
||||
.from(bookAuthors)
|
||||
.innerJoin(authors, eq(authors.id, bookAuthors.authorId))
|
||||
.where(and(eligibility, sql`${bookAuthors.authorId} > ${cursorAuthorId}`))
|
||||
.orderBy(asc(bookAuthors.authorId))
|
||||
.limit(resolvedBatchSize);
|
||||
|
||||
if (rows.length === 0) break;
|
||||
|
||||
totalQueued += await this.upsertSchedule(
|
||||
rows.map((row) => row.authorId),
|
||||
reason,
|
||||
resolvedBatchSize,
|
||||
);
|
||||
cursorAuthorId = rows[rows.length - 1]!.authorId;
|
||||
}
|
||||
|
||||
return totalQueued;
|
||||
}
|
||||
|
||||
async countEligibleLinkedAuthors(conditions: AuthorEnrichmentConditions): Promise<number> {
|
||||
@@ -287,3 +361,9 @@ function isUniqueViolation(error: unknown): boolean {
|
||||
const code = Reflect.get(error, 'code');
|
||||
return code === '23505';
|
||||
}
|
||||
|
||||
function resolveBatchSize(batchSize: number): number | null {
|
||||
const resolved = Math.floor(batchSize);
|
||||
if (!Number.isFinite(resolved) || resolved <= 0) return null;
|
||||
return resolved;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user