diff --git a/src/task/kb/IndexFilingSectionsTask.test.ts b/src/task/kb/IndexFilingSectionsTask.test.ts index bab61dde..dd6254b3 100644 --- a/src/task/kb/IndexFilingSectionsTask.test.ts +++ b/src/task/kb/IndexFilingSectionsTask.test.ts @@ -5,15 +5,17 @@ */ import { afterEach, describe, expect, it, vi } from "vitest"; -import { globalServiceRegistry } from "workglow"; +import type { DocumentNode } from "workglow"; +import { Document, globalServiceRegistry, NodeKind } from "workglow"; import { withSqliteDb } from "../../config/testing/withSqliteDb"; -import { resetSecKnowledgeBaseForTesting } from "../../kb/secKnowledgeBase"; +import { getSecKnowledgeBase, resetSecKnowledgeBaseForTesting } from "../../kb/secKnowledgeBase"; import { FILING_DOCUMENT_REPOSITORY_TOKEN, type FilingDocument, } from "../../storage/document/FilingDocumentSchema"; import { FILING_SECTION_REPOSITORY_TOKEN } from "../../storage/document/FilingSectionSchema"; import { IndexFilingSectionsTask } from "./IndexFilingSectionsTask"; +import { kbDocIdFor } from "./selectDocumentsToIndex"; const header = (index: number, filingDate: string): FilingDocument => ({ cik: 320193, @@ -54,30 +56,55 @@ describe("IndexFilingSectionsTask selection", () => { } }; - it("streams the converted filings rather than loading every header first", async () => { + /** Mark a filing as already in the knowledge base. */ + const markIndexed = async (index: number): Promise => { + const kb = await getSecKnowledgeBase(); + const title = `Filing ${index}`; + const root = { kind: NodeKind.DOCUMENT, title, children: [] } as unknown as DocumentNode; + const docId = kbDocIdFor(`0000320193-26-${String(index).padStart(6, "0")}`, "primary.htm"); + await kb.upsertDocument(new Document(root, { title } as never, [], docId)); + }; + + it("picks the work in the database rather than reading headers to sift them", async () => { await seed(5); const repo = globalServiceRegistry.get(FILING_DOCUMENT_REPOSITORY_TOKEN); - const getAll = vi.spyOn(repo, "getAll"); + const reads = (["getAll", "records", "query", "queryPage"] as const).map((method) => + vi.spyOn(repo, method) + ); - const out = await new IndexFilingSectionsTask().run({ limit: 0 }); + const out = await new IndexFilingSectionsTask().run({ cik: 320193, limit: 0 }); - // `getAll()` on the unscoped path is the whole converted corpus in memory - // — hundreds of thousands of rows to take the first few of. - expect(getAll).not.toHaveBeenCalled(); - expect(out).toMatchObject({ indexed: 0, truncated: true }); + // Reading headers to decide is the whole converted corpus crossing the + // process boundary to take the first few of it. + for (const read of reads) expect(read).not.toHaveBeenCalled(); + expect(out.truncated).toBe(true); }); - it("streams a scoped selection by page too", async () => { + it("still reports what is already indexed once nothing is left to do", async () => { + // The count `sec index` prints, and what makes it suggest asking a question + // rather than converting filings there are none of. await seed(3); - const repo = globalServiceRegistry.get(FILING_DOCUMENT_REPOSITORY_TOKEN); - const query = vi.spyOn(repo, "query"); - const queryPage = vi.spyOn(repo, "queryPage"); + for (const index of [0, 1, 2]) await markIndexed(index); - const out = await new IndexFilingSectionsTask().run({ cik: 320193, limit: 0 }); + const out = await new IndexFilingSectionsTask().run({ limit: 5 }); - expect(query).not.toHaveBeenCalled(); - expect(queryPage).toHaveBeenCalled(); - expect(out.truncated).toBe(true); + expect(out).toMatchObject({ indexed: 0, sections: 0, skipped: 3, truncated: false }); + }); + + it("spends its limit on filings that need work, not on ones already indexed", async () => { + // The defect: `limit` counted filings INDEXED, and an already-indexed one + // was skipped without counting toward it — so a small limit on an indexed + // corpus read every row and probed the knowledge base once per row. + await seed(4); + await markIndexed(0); + await markIndexed(1); + const kb = await getSecKnowledgeBase(); + const getDocument = vi.spyOn(kb, "getDocument"); + + const out = await new IndexFilingSectionsTask().run({ limit: 2 }); + + expect(getDocument).not.toHaveBeenCalled(); + expect(out).toMatchObject({ skipped: 2, truncated: false }); }); it("reports no truncation when the scope leaves nothing to index", async () => { diff --git a/src/task/kb/IndexFilingSectionsTask.ts b/src/task/kb/IndexFilingSectionsTask.ts index d8039bc0..e31f4051 100644 --- a/src/task/kb/IndexFilingSectionsTask.ts +++ b/src/task/kb/IndexFilingSectionsTask.ts @@ -5,7 +5,7 @@ */ import { Type } from "typebox"; -import type { DocumentNode, IExecuteContext, PageCursor } from "workglow"; +import type { DocumentNode, IExecuteContext } from "workglow"; import { Document, globalServiceRegistry, @@ -15,17 +15,14 @@ import { TaskAbortedError, } from "workglow"; import { getSecKnowledgeBase } from "../../kb/secKnowledgeBase"; -import { - FILING_DOCUMENT_REPOSITORY_TOKEN, - type FilingDocument, - type FilingDocumentRepositoryStorage, -} from "../../storage/document/FilingDocumentSchema"; +import type { FilingDocument } from "../../storage/document/FilingDocumentSchema"; import { FILING_SECTION_REPOSITORY_TOKEN, type FilingSection, } from "../../storage/document/FilingSectionSchema"; import { accessionWithoutDashes } from "../../util/accession"; import type { TaskPorts } from "../taskPorts"; +import { countAlreadyIndexed, kbDocIdFor, selectDocumentsToIndex } from "./selectDocumentsToIndex"; export interface IndexFilingSectionsTaskInput { /** Only this issuer's filings. */ @@ -63,9 +60,6 @@ export interface IndexFilingSectionsTaskOutput { */ export const DEFAULT_ASK_INDEX_LIMIT = 25; -/** Rows per read while walking the converted filings. */ -const HEADER_PAGE_SIZE = 500; - /** The EDGAR URL a citation points at. */ function filingUrl(cik: number, accession: string, docFile: string): string { return ( @@ -117,32 +111,6 @@ function toDocument(header: FilingDocument, sections: readonly FilingSection[]): } as never); } -/** - * The converted filings matching `criteria`, a page at a time. - * - * Streamed rather than collected: the unscoped case is every converted filing - * in the database, and materializing that array to take the first few of it - * costs the whole corpus in memory before any work starts. - */ -async function* streamHeaders( - repo: FilingDocumentRepositoryStorage, - criteria: Record -): AsyncGenerator { - if (Object.keys(criteria).length === 0) { - yield* repo.records(HEADER_PAGE_SIZE); - return; - } - let cursor: PageCursor | undefined; - for (;;) { - const page = await repo.queryPage(criteria as never, { limit: HEADER_PAGE_SIZE, cursor }); - yield* page.items; - // Both conditions: a cursor can be handed back for a page that concurrent - // deletes have since emptied, and looping on it alone would not terminate. - if (page.nextCursor === undefined || page.items.length === 0) return; - cursor = page.nextCursor; - } -} - /** * Embeds converted filing sections into the knowledge base `sec ask` reads. * @@ -187,43 +155,33 @@ export class IndexFilingSectionsTask extends Task< context: IExecuteContext ): Promise> { const kb = await getSecKnowledgeBase(); - const documentRepo = globalServiceRegistry.get(FILING_DOCUMENT_REPOSITORY_TOKEN); const sectionRepo = globalServiceRegistry.get(FILING_SECTION_REPOSITORY_TOKEN); - const criteria: Record = {}; - if (input.cik !== undefined) criteria.cik = input.cik; - if (input.form !== undefined) criteria.form = input.form; - if (input.accession !== undefined) criteria.accession_number = input.accession; + const scope = { + cik: input.cik, + form: input.form, + since: input.since, + accession: input.accession, + }; - // Only a denominator for the progress line, so an over-count from the - // `since` filter (which no backend expresses as a criterion) is harmless. - const candidates = await documentRepo.count( - Object.keys(criteria).length === 0 ? undefined : (criteria as never) - ); + // One document more than asked for, so "there are more" is something this + // run observed rather than inferred from having filled the limit exactly. const limit = input.limit; - const denominator = Math.max(1, limit === undefined ? candidates : Math.min(limit, candidates)); + const candidates = await selectDocumentsToIndex({ + ...scope, + limit: limit === undefined ? undefined : limit + 1, + force: input.force, + }); + const truncated = limit !== undefined && candidates.length > limit; + const work = truncated ? candidates.slice(0, limit) : candidates; + const skipped = input.force === true ? 0 : await countAlreadyIndexed(scope); + const denominator = Math.max(1, work.length); let indexed = 0; let sectionTotal = 0; - let skipped = 0; - let truncated = false; - for await (const header of streamHeaders(documentRepo, criteria)) { + for (const header of work) { if (context.signal?.aborted) throw new TaskAbortedError(); - // Scope first, then the limit, so a run that stops has genuinely left - // candidates behind rather than rows the scope excludes. - if (input.since !== undefined && (header.filing_date ?? "") < input.since) continue; - if (limit !== undefined && indexed >= limit) { - truncated = true; - break; - } - - // The document id is derived from the filing, so "already indexed" is a - // lookup rather than a second table to keep in step with this one. - const docId = `${header.accession_number}:${header.doc_file}`; - if (input.force !== true && (await kb.getDocument(docId)) !== undefined) { - skipped += 1; - continue; - } + const docId = kbDocIdFor(header.accession_number, header.doc_file); const sections = ( (await sectionRepo.query({ diff --git a/src/task/kb/selectDocumentsToIndex.sqlite.test.ts b/src/task/kb/selectDocumentsToIndex.sqlite.test.ts new file mode 100644 index 00000000..2558827b --- /dev/null +++ b/src/task/kb/selectDocumentsToIndex.sqlite.test.ts @@ -0,0 +1,151 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import type { DocumentNode } from "workglow"; +import { Document, globalServiceRegistry, NodeKind } from "workglow"; +import { withSqliteDb } from "../../config/testing/withSqliteDb"; +import { getSecKnowledgeBase, resetSecKnowledgeBaseForTesting } from "../../kb/secKnowledgeBase"; +import { + FILING_DOCUMENT_REPOSITORY_TOKEN, + type FilingDocument, +} from "../../storage/document/FilingDocumentSchema"; +import { countAlreadyIndexed, kbDocIdFor, selectDocumentsToIndex } from "./selectDocumentsToIndex"; + +const doc = (index: number, over: Partial = {}): FilingDocument => ({ + cik: 320193, + accession_number: `0000320193-26-${String(index).padStart(6, "0")}`, + doc_file: "primary.htm", + doc_type: "10-K", + description: null, + sequence: 1, + is_primary: true, + form: "10-K", + filing_date: "2026-03-01", + title: `Filing ${index}`, + section_count: 1, + char_count: 100, + converter_version: "1", + converted_at: "2026-01-01T00:00:00.000Z", + ...over, +}); + +/** + * `limit` used to bound the number of documents INDEXED, and an already-indexed + * one was skipped without counting toward it. So a small `--limit` on a corpus + * that is already indexed read every row and probed the knowledge base once per + * row before concluding there was nothing to do — and `sec ask` builds the + * index implicitly, so that ran on every question. + * + * These cases are about which rows come back and how many, not about embedding. + */ +describe("selectDocumentsToIndex (sqlite)", () => { + withSqliteDb("select_documents_to_index", [FILING_DOCUMENT_REPOSITORY_TOKEN]); + + beforeEach(async () => { + await resetSecKnowledgeBaseForTesting(); + delete process.env.SEC_EMBEDDING_MODEL; + }); + + afterEach(async () => { + await resetSecKnowledgeBaseForTesting(); + }); + + async function seed(count: number, over: (i: number) => Partial = () => ({})) { + const repo = globalServiceRegistry.get(FILING_DOCUMENT_REPOSITORY_TOKEN); + for (let i = 1; i <= count; i += 1) await repo.put(doc(i, over(i)) as never); + } + + /** Mark a document as already in the knowledge base. */ + async function markIndexed(index: number) { + const kb = await getSecKnowledgeBase(); + const title = `Filing ${index}`; + const root = { kind: NodeKind.DOCUMENT, title, children: [] } as unknown as DocumentNode; + const docId = kbDocIdFor(`0000320193-26-${String(index).padStart(6, "0")}`, "primary.htm"); + await kb.upsertDocument(new Document(root, { title } as never, [], docId)); + } + + it("returns nothing once every document is indexed", async () => { + await seed(3); + for (const i of [1, 2, 3]) await markIndexed(i); + + expect(await selectDocumentsToIndex({ limit: 5 })).toEqual([]); + }); + + it("counts the limit in documents that need work, not documents examined", async () => { + // The defect: two of five are already indexed, so a limit of 2 must return + // the two that are NOT — never fewer because the indexed ones consumed it. + await seed(5); + await markIndexed(1); + await markIndexed(2); + + const picked = await selectDocumentsToIndex({ limit: 2 }); + + expect(picked).toHaveLength(2); + expect(picked.map((d) => d.accession_number)).not.toContain("0000320193-26-000001"); + expect(picked.map((d) => d.accession_number)).not.toContain("0000320193-26-000002"); + }); + + it("re-selects everything under force", async () => { + await seed(3); + for (const i of [1, 2, 3]) await markIndexed(i); + + expect(await selectDocumentsToIndex({ limit: 5, force: true })).toHaveLength(3); + }); + + it("applies `since` in the query rather than after it", async () => { + await seed(4, (i) => ({ filing_date: i <= 2 ? "2020-01-01" : "2026-06-01" })); + + const picked = await selectDocumentsToIndex({ limit: 10, since: "2026-01-01" }); + + expect(picked).toHaveLength(2); + expect(picked.every((d) => (d.filing_date ?? "") >= "2026-01-01")).toBe(true); + }); + + it("narrows by cik, form and accession", async () => { + await seed(3, (i) => ({ form: i === 2 ? "8-K" : "10-K" })); + + expect(await selectDocumentsToIndex({ limit: 10, form: "8-K" })).toHaveLength(1); + expect(await selectDocumentsToIndex({ limit: 10, cik: 999 })).toHaveLength(0); + expect( + await selectDocumentsToIndex({ limit: 10, accession: "0000320193-26-000003" }) + ).toHaveLength(1); + }); + + it("returns newest first, so an interrupted backfill covers what people read", async () => { + await seed(3, (i) => ({ filing_date: `2026-0${i}-01` })); + + const picked = await selectDocumentsToIndex({ limit: 3 }); + + expect(picked.map((d) => d.filing_date)).toEqual(["2026-03-01", "2026-02-01", "2026-01-01"]); + }); + + it("leaves out documents with no sections, which can never be indexed", async () => { + // They embed nothing, so they never enter the knowledge base — selecting + // them spends the limit on work that cannot happen, on every run. + await seed(3, (i) => ({ section_count: i === 2 ? 0 : 1 })); + + const picked = await selectDocumentsToIndex({ limit: 10 }); + + expect(picked.map((d) => d.accession_number)).not.toContain("0000320193-26-000002"); + expect(picked).toHaveLength(2); + }); + + it("counts nothing as already indexed before the index exists", async () => { + // The knowledge-base tables are built by the first command that opens the + // index, and this runs before that — a database with converted filings and + // no index has no table to join against. + await seed(2); + + expect(await selectDocumentsToIndex({ limit: 10 })).toHaveLength(2); + expect(await countAlreadyIndexed({})).toBe(0); + }); + + it("returns nothing for a non-positive limit rather than everything", async () => { + await seed(2); + expect(await selectDocumentsToIndex({ limit: 0 })).toEqual([]); + }); +}); diff --git a/src/task/kb/selectDocumentsToIndex.ts b/src/task/kb/selectDocumentsToIndex.ts new file mode 100644 index 00000000..c36da210 --- /dev/null +++ b/src/task/kb/selectDocumentsToIndex.ts @@ -0,0 +1,248 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { PageCursor } from "workglow"; +import { globalServiceRegistry } from "workglow"; +import { KB_DOCUMENT_TABLE } from "../../kb/secKbTables"; +import { + FILING_DOCUMENT_REPOSITORY_TOKEN, + type FilingDocument, + type FilingDocumentRepositoryStorage, +} from "../../storage/document/FilingDocumentSchema"; +import { getDb } from "../../util/db"; +import { resolveSqlBackend } from "../../util/sqlBackend"; + +export interface SelectDocumentsOptions { + readonly cik?: number | undefined; + readonly form?: string | undefined; + readonly since?: string | undefined; + readonly accession?: string | undefined; + /** Stop after this many documents that still need indexing; all of them when unset. */ + readonly limit?: number | undefined; + /** Re-index documents the knowledge base already holds. */ + readonly force?: boolean | undefined; +} + +/** The `kb_document.doc_id` a converted document is stored under. */ +export function kbDocIdFor(accession: string, docFile: string): string { + return `${accession}:${docFile}`; +} + +function documentRepoIfRegistered(): FilingDocumentRepositoryStorage | undefined { + return globalServiceRegistry.has(FILING_DOCUMENT_REPOSITORY_TOKEN) + ? globalServiceRegistry.get(FILING_DOCUMENT_REPOSITORY_TOKEN) + : undefined; +} + +/** + * Whether there is a table to anti-join against. + * + * The three knowledge-base tables are built lazily by the first command that + * opens the index, and this selector runs before that — so on a database with + * converted filings that has never been indexed they do not exist yet. Nothing + * is indexed in that case, which is the answer, not an error. + */ +function kbDocumentTableExists(): boolean { + return ( + getDb() + .prepare<[string], { name: string }>( + "SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?" + ) + .all(KB_DOCUMENT_TABLE).length > 0 + ); +} + +/** + * Converted documents the knowledge base does not already hold, newest first. + * + * The set difference is an anti-join, so it is raw SQL for the same reason + * `selectFilingsToConvert` is: `ITabularStorage` cannot express one, and the + * alternative is a knowledge-base round trip per document over a table with + * hundreds of thousands of rows. + * + * That alternative is what this replaces, and the round trips were not the + * worst of it. The limit used to count documents actually INDEXED, and an + * already-indexed one was skipped without counting toward it — so `--limit 5` + * against a corpus that was already indexed read every row and probed the + * knowledge base once per row before concluding it had nothing to do. `ask` + * builds the index implicitly, so that ran on every question. + * + * The fast path is SQLite only, and there is no Postgres arm to write: the + * knowledge base lives in the connection `getDb()` owns and + * `getSecKnowledgeBase` refuses every other backend by name, so where this + * query cannot run the index cannot exist either. + */ +export async function selectDocumentsToIndex( + options: SelectDocumentsOptions +): Promise { + if (options.limit !== undefined && options.limit <= 0) return []; + const documentRepo = documentRepoIfRegistered(); + if (documentRepo === undefined) return []; + + if (resolveSqlBackend("read", documentRepo) !== "sqlite") { + return await selectByScan(documentRepo, options); + } + + const antiJoin = options.force !== true && kbDocumentTableExists(); + const params: (string | number)[] = []; + const clauses: string[] = []; + if (options.cik !== undefined) { + clauses.push("d.`cik` = ?"); + params.push(options.cik); + } + if (options.form !== undefined) { + clauses.push("d.`form` = ?"); + params.push(options.form); + } + if (options.accession !== undefined) { + clauses.push("d.`accession_number` = ?"); + params.push(options.accession); + } + // In the query rather than after it. Filtered in the loop, `since` cost a + // full read of everything older than the cutoff to discard it. + if (options.since !== undefined) { + clauses.push("d.`filing_date` >= ?"); + params.push(options.since); + } + // A document with no sections has nothing to embed, so it never enters the + // knowledge base and would be re-selected on every run — spending the limit + // on work that cannot happen. `section_count` is written in the same + // transaction as the section rows, so it is the same answer reading them + // gives. + clauses.push("d.`section_count` > 0"); + if (antiJoin) clauses.push("k.`doc_id` IS NULL"); + // SQLite numbers `?` by position, so the limit binds last because it is + // written last. + if (options.limit !== undefined) params.push(options.limit); + + const join = antiJoin + ? "LEFT JOIN `" + + KB_DOCUMENT_TABLE + + "` k ON k.`doc_id` = d.`accession_number` || ':' || d.`doc_file`" + : ""; + const where = clauses.length === 0 ? "1 = 1" : clauses.join(" AND "); + return getDb() + .prepare<(string | number)[], FilingDocument>( + `SELECT d.* + FROM \`filing_document\` d + ${join} + WHERE ${where} + ORDER BY d.\`filing_date\` DESC, d.\`accession_number\` DESC + ${options.limit === undefined ? "" : "LIMIT ?"}` + ) + .all(...params); +} + +/** + * How many documents in scope the knowledge base already holds. + * + * Its own query rather than a by-product of the selection, because the + * selection stops at the limit and so cannot count what it never reached. One + * COUNT replaces the unfiltered `count` this run used to take for its progress + * denominator, so it costs no extra query — and the denominator it replaces was + * wrong anyway, measuring a run that indexes three filings out of three hundred + * candidates at one percent. + * + * Zero where the anti-join cannot run, which is the same answer + * {@link selectDocumentsToIndex} gives there: no index, nothing already in it. + */ +export async function countAlreadyIndexed( + options: Omit +): Promise { + const documentRepo = documentRepoIfRegistered(); + if (documentRepo === undefined) return 0; + if (resolveSqlBackend("read", documentRepo) !== "sqlite") return 0; + if (!kbDocumentTableExists()) return 0; + + const params: (string | number)[] = []; + const clauses: string[] = []; + if (options.cik !== undefined) { + clauses.push("d.`cik` = ?"); + params.push(options.cik); + } + if (options.form !== undefined) { + clauses.push("d.`form` = ?"); + params.push(options.form); + } + if (options.accession !== undefined) { + clauses.push("d.`accession_number` = ?"); + params.push(options.accession); + } + if (options.since !== undefined) { + clauses.push("d.`filing_date` >= ?"); + params.push(options.since); + } + const where = clauses.length === 0 ? "1 = 1" : clauses.join(" AND "); + const row = getDb() + .prepare<(string | number)[], { n: number }>( + `SELECT COUNT(*) AS n + FROM \`filing_document\` d + JOIN \`${KB_DOCUMENT_TABLE}\` k + ON k.\`doc_id\` = d.\`accession_number\` || ':' || d.\`doc_file\` + WHERE ${where}` + ) + .get(...params); + return row?.n ?? 0; +} + +const HEADER_PAGE_SIZE = 500; + +/** + * The repository path: page the document table, apply the filters, stop at the + * limit. + * + * It consults no knowledge base, because it is reached only where there cannot + * be one to consult. A non-durable document repository is invisible to + * `getDb()`, so opening the index here would read a real database this caller + * never wrote to and report the wrong documents as already indexed; and on + * Postgres the index does not exist at all. Both are better served by naming + * the candidates and letting the caller fail where the knowledge base itself + * refuses. + */ +async function selectByScan( + repo: FilingDocumentRepositoryStorage, + options: SelectDocumentsOptions +): Promise { + const criteria: Record = {}; + if (options.cik !== undefined) criteria.cik = options.cik; + if (options.form !== undefined) criteria.form = options.form; + if (options.accession !== undefined) criteria.accession_number = options.accession; + + const picked: FilingDocument[] = []; + for await (const header of streamHeaders(repo, criteria)) { + if (options.since !== undefined && (header.filing_date ?? "") < options.since) continue; + if (header.section_count <= 0) continue; + picked.push(header); + if (options.limit !== undefined && picked.length >= options.limit) break; + } + return picked; +} + +/** + * The matching documents, a page at a time. + * + * Streamed rather than collected: the unscoped case is every converted filing + * in the database, and materializing that to take the first few of it costs the + * whole corpus in memory before any work starts. + */ +async function* streamHeaders( + repo: FilingDocumentRepositoryStorage, + criteria: Record +): AsyncGenerator { + if (Object.keys(criteria).length === 0) { + yield* repo.records(HEADER_PAGE_SIZE); + return; + } + let cursor: PageCursor | undefined; + for (;;) { + const page = await repo.queryPage(criteria as never, { limit: HEADER_PAGE_SIZE, cursor }); + yield* page.items; + // Both conditions: a cursor can be handed back for a page that concurrent + // deletes have since emptied, and looping on it alone would not terminate. + if (page.nextCursor === undefined || page.items.length === 0) return; + cursor = page.nextCursor; + } +}