Skip to content

Commit 2837d6a

Browse files
committed
fix(knowledge): check the page deadline after the lease heartbeat
1 parent 76db356 commit 2837d6a

2 files changed

Lines changed: 23 additions & 2 deletions

File tree

‎apps/sim/lib/knowledge/connectors/member-observations.test.ts‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -529,6 +529,27 @@ describe('writeProjectionPages', () => {
529529
}
530530
})
531531

532+
/** The lease heartbeat runs before each page and can itself use up what is left of the budget. */
533+
it('opens no transaction when the heartbeat before a page outlasts the deadline', async () => {
534+
let clock = Date.now()
535+
const now = vi.spyOn(Date, 'now').mockImplementation(() => clock)
536+
try {
537+
const transaction = vi.fn()
538+
const write = vi.fn()
539+
const beforePage = vi.fn(async () => {
540+
clock += 2_000
541+
})
542+
543+
await expect(
544+
writeProjectionPages(['a'], transaction, write, { beforePage, deadlineAt: clock + 1_000 })
545+
).resolves.toEqual({ written: 0, finished: false })
546+
expect(beforePage).toHaveBeenCalledOnce()
547+
expect(transaction).not.toHaveBeenCalled()
548+
} finally {
549+
now.mockRestore()
550+
}
551+
})
552+
532553
it('writes every page when no deadline is given', async () => {
533554
queueTableRows(schemaMock.document, [
534555
{ id: 'a', chunkCount: 200 },

‎apps/sim/lib/knowledge/connectors/member-observations.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -425,9 +425,9 @@ export async function writeProjectionPages(
425425
let pending = [...documentIds]
426426
let written = 0
427427
while (pending.length > 0) {
428+
await options.beforePage?.()
428429
if (options.deadlineAt !== undefined && Date.now() >= options.deadlineAt)
429430
return { written, finished: false }
430-
await options.beforePage?.()
431431
const { rows, rest } = await transaction(async (tx) => {
432432
const { page, rest } = await lockProjectionPage(tx, pending)
433433
return { rows: page.length > 0 ? await write(tx, page) : 0, rest }
@@ -471,8 +471,8 @@ export async function rewriteConnectorDocumentAcls(input: {
471471
let rewritten = 0
472472
let after: ConnectorDocumentCursor | undefined
473473
for (;;) {
474-
if (expired()) return { rewritten, finished: false }
475474
await input.beforePage?.()
475+
if (expired()) return { rewritten, finished: false }
476476
const window = await db
477477
.select({
478478
id: document.id,

0 commit comments

Comments
 (0)